diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index f3367e20fe..debd3093a1 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -204,6 +204,19 @@ jobs:
             packages/eval/harbor
           npm --workspace @maka/eval run test:egress-proxy:live
 
+      - name: Install Harbor lifecycle test dependencies
+        if: contains(steps.plan.outputs.standard_workspaces, 'packages/eval')
+        run: |
+          python3 -m venv "$RUNNER_TEMP/maka-harbor-lifecycle"
+          "$RUNNER_TEMP/maka-harbor-lifecycle/bin/python" -m pip install --disable-pip-version-check 'harbor==0.20.0'
+
+      - name: Run real Harbor lifecycle tests
+        if: contains(steps.plan.outputs.standard_workspaces, 'packages/eval')
+        env:
+          MAKA_EVAL_HARBOR_LIFECYCLE_TEST: '1'
+        run: |
+          "$RUNNER_TEMP/maka-harbor-lifecycle/bin/python" packages/eval/harbor/test_harbor_trial_lifecycle.py
+
       - name: Run Runtime Host tests
         if: steps.plan.outputs.runtime_host == 'true'
         run: npm --workspace @maka/runtime-host run test:dist
diff --git a/packages/cli/src/__tests__/runtime-host-operator-command.test.ts b/packages/cli/src/__tests__/runtime-host-operator-command.test.ts
index 477757f534..f0053e9dfe 100644
--- a/packages/cli/src/__tests__/runtime-host-operator-command.test.ts
+++ b/packages/cli/src/__tests__/runtime-host-operator-command.test.ts
@@ -209,6 +209,7 @@ describe('Runtime Host operator commands', () => {
         'access.credential.rotation.prepare',
         'access.credential.rotation.revoke',
         'host.upgrade.prepare',
+        'hosted.execution.admit',
         'hosted.execution.cancel',
         'hosted.execution.start',
       ],
diff --git a/packages/eval/harbor/relay_agent.py b/packages/eval/harbor/relay_agent.py
index 903ba96ac4..a7dcacdbc7 100644
--- a/packages/eval/harbor/relay_agent.py
+++ b/packages/eval/harbor/relay_agent.py
@@ -91,6 +91,7 @@ def __init__(
         relay_port: int,
         relay_token: str,
         teardown_timeout_ms: int,
+        framework_timeout_ms: int | None = None,
         **kwargs: Any,
     ):
         super().__init__(*args, **kwargs)
@@ -100,6 +101,14 @@ def __init__(
         if not isinstance(teardown_timeout_ms, int) or teardown_timeout_ms <= 0:
             raise RuntimeError("Maka Eval teardown timeout is invalid")
         self._teardown_timeout = teardown_timeout_ms / 1000
+        if framework_timeout_ms is None:
+            framework_timeout_ms = teardown_timeout_ms
+        if not isinstance(framework_timeout_ms, int) or framework_timeout_ms <= 0:
+            raise RuntimeError("Maka Eval framework timeout is invalid")
+        self._framework_timeout = min(
+            framework_timeout_ms / 1000,
+            self._teardown_timeout,
+        )
 
     @staticmethod
     def name() -> str:
@@ -116,6 +125,7 @@ async def run(self, instruction: str, environment: Any, context: Any) -> None:
         execution: asyncio.Task[Any] | None = None
         decision: asyncio.Task[dict[str, Any]] | None = None
         request: dict[str, Any] | None = None
+        cwd = ""
         execution_reported = False
         scope_path = f"/logs/agent/.maka-eval-{self._token}.pid"
         environment_path = f"/tmp/maka-eval-{self._token}.env"
@@ -160,8 +170,9 @@ async def run(self, instruction: str, environment: Any, context: Any) -> None:
             # waiting on those processes here — `environment.exec` has already
             # returned — so tearing them down would not unblock anything; it
             # would only edit the thing about to be measured, and edit it for
-            # some subjects and not others. Cancellation still quiesces, because
-            # there the subject has not stopped and the trial is being abandoned.
+            # some subjects and not others. Host abort still quiesces, because
+            # that trial is abandoned. Framework timeout does not: the verifier
+            # still scores what the subject left.
             if not await _send(
                 writer,
                 {
@@ -181,60 +192,25 @@ async def run(self, instruction: str, environment: Any, context: Any) -> None:
                 "verify",
             )
         except asyncio.CancelledError:
+            # A cancelled task would otherwise fail every await in this
+            # handler. Uncancel so cleanup can run; a second cancel or any
+            # cleanup error destroys the environment before we re-raise.
+            current = asyncio.current_task()
+            if current is not None and hasattr(current, "uncancel"):
+                current.uncancel()
+            framework_timeout_handled = False
             if request is not None and execution is not None:
-                execution_terminal = execution.done() and not execution.cancelled()
-                terminal_projection = None
-                if execution_terminal:
-                    terminal_result = execution.result()
-                    terminal_projection = _project_result(terminal_result, request)
-                # A subject that already exited has nothing left to settle, and
-                # tearing its scope down here would remove what the verifier is
-                # about to score -- the same environment edit this relay stopped
-                # making on the ordinary path. Only a subject still running is
-                # brought to a stop.
-                if terminal_projection is not None:
-                    result = terminal_result
-                else:
-                    result = await _settle_or_destroy(
-                        environment, cwd, scope_path, execution, self._teardown_timeout
-                    )
-                if result is not None:
-                    await _persist_subject_outputs(environment, result)
-                if (
-                    result is not None
-                    and not execution_reported
-                    and (execution_terminal or not _host_teardown_requested)
-                ):
-                    stdout, diagnostic = terminal_projection or _project_result(result, request)
-                    with contextlib.suppress(Exception):
-                        await _send(
-                            writer,
-                            {
-                                "token": self._token,
-                                "kind": "executed",
-                                "termination": "exited" if execution_terminal else "framework_timeout",
-                                "exitCode": result.return_code if execution_terminal else 124,
-                                "stdout": stdout,
-                                "diagnostic": diagnostic,
-                            },
-                        )
-                elif not execution_reported and not _host_teardown_requested:
-                    with contextlib.suppress(Exception):
-                        await _send(
-                            writer,
-                            {
-                                "token": self._token,
-                                "kind": "executed",
-                                "termination": "framework_timeout",
-                                "exitCode": 124,
-                                "stdout": "",
-                                "diagnostic": (
-                                    _carrier_diagnostic("result-frame-missing", b"")
-                                    if request.get("captureStdout", True)
-                                    else {"category": "none"}
-                                ),
-                            },
-                        )
+                framework_timeout_handled = await self._cleanup_cancelled_execution(
+                    environment,
+                    cwd,
+                    scope_path,
+                    execution,
+                    request,
+                    writer,
+                    execution_reported,
+                )
+            if framework_timeout_handled and not _host_teardown_requested:
+                return
             raise
         except RelayTransportClosed:
             if request is not None and execution is not None:
@@ -272,6 +248,102 @@ 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,
+    ) -> bool:
+        cleanup_timeout = (
+            self._teardown_timeout
+            if _host_teardown_requested
+            else self._framework_timeout
+        )
+        deadline = asyncio.get_running_loop().time() + cleanup_timeout
+        try:
+            return await self._finalize_cancelled_execution(
+                environment,
+                cwd,
+                scope_path,
+                execution,
+                request,
+                writer,
+                execution_reported,
+                cleanup_timeout,
+            )
+        except BaseException:
+            remaining = deadline - asyncio.get_running_loop().time()
+            with contextlib.suppress(Exception):
+                if remaining > 0:
+                    await asyncio.wait_for(
+                        _settle_or_destroy(
+                            environment, cwd, scope_path, execution, remaining
+                        ),
+                        timeout=remaining,
+                    )
+            raise
+
+    async def _finalize_cancelled_execution(
+        self,
+        environment: Any,
+        cwd: str,
+        scope_path: str,
+        execution: asyncio.Task[Any],
+        request: dict[str, Any],
+        writer: Any,
+        execution_reported: bool,
+        cleanup_timeout: float,
+    ) -> bool:
+        execution_terminal = execution.done() and not execution.cancelled()
+        terminal_projection = None
+        if execution_terminal:
+            terminal_result = execution.result()
+            terminal_projection = _project_result(terminal_result, request)
+        # A subject that already exited has nothing left to settle, and
+        # tearing its scope down here would remove what the verifier is
+        # about to score -- the same environment edit this relay stopped
+        # making on the ordinary path. A still-running subject is
+        # stopped so the execution call can return. Host abort then
+        # quiesces the leftover group; framework timeout does not,
+        # because the verifier still runs.
+        if terminal_projection is not None:
+            result = terminal_result
+        elif _host_teardown_requested:
+            result = await _settle_or_destroy(
+                environment, cwd, scope_path, execution, cleanup_timeout
+            )
+        else:
+            result = await _stop_subject_for_timeout(
+                environment, cwd, scope_path, execution, cleanup_timeout
+            )
+        if result is not None:
+            await _persist_subject_outputs(environment, result)
+        if (
+            result is not None
+            and not execution_reported
+            and (execution_terminal or not _host_teardown_requested)
+        ):
+            stdout, diagnostic = terminal_projection or _project_result(result, request)
+            try:
+                execution_reported = await _send(
+                    writer,
+                    {
+                        "token": self._token,
+                        "kind": "executed",
+                        "termination": "exited" if execution_terminal else "framework_timeout",
+                        "exitCode": result.return_code if execution_terminal else 124,
+                        "stdout": stdout,
+                        "diagnostic": diagnostic,
+                    },
+                )
+            except Exception:
+                execution_reported = False
+        return execution_reported and not _host_teardown_requested
+
 
 async def _prepare_command(
     environment: Any,
@@ -321,7 +393,10 @@ async def _prepare_command(
         if secret_path is not None:
             secret_path.unlink(missing_ok=True)
     subject = shlex.join([request["command"], *request["args"]])
-    output_redirect = "" if capture_stdout else " >/dev/null"
+    # An exit-code subject has no structured stdout to preserve. Detach both
+    # streams so a task-owned background process cannot keep Harbor's
+    # `docker compose exec` output pipe open after the subject leader exits.
+    output_redirect = "" if capture_stdout else " >/dev/null 2>&1"
     scope_error = shlex.quote(f"{SCOPE_ERROR_PREFIX} {result_token}\\n")
     inner = (
         "umask 077; "
@@ -568,7 +643,7 @@ async def _settle(environment: Any, cwd: str, scope_path: str, execution: Any) -
     else:
         result = None
         for signal, timeout in (("TERM", 20), ("KILL", 10)):
-            await _signal(environment, cwd, scope_path, signal)
+            await _signal_group(environment, cwd, scope_path, signal)
             try:
                 result = await asyncio.wait_for(asyncio.shield(execution), timeout=timeout)
                 break
@@ -584,6 +659,59 @@ async def _settle(environment: Any, cwd: str, scope_path: str, execution: Any) -
     return result
 
 
+async def _stop_subject_for_timeout(
+    environment: Any,
+    cwd: str,
+    scope_path: str,
+    execution: Any,
+    timeout: float,
+) -> Any:
+    # The verifier still scores this trial. Stop only the subject so
+    # `environment.exec` can return; do not hunt descendants or delete the
+    # environment. Those leftovers are what the task asked the subject to leave.
+    if execution.cancelled():
+        raise RuntimeError("Maka Eval subject execution was cancelled")
+    if execution.done():
+        return execution.result()
+    loop = asyncio.get_running_loop()
+    deadline = loop.time() + timeout
+    destroy_reserve = min(20.0, timeout * 0.2)
+    stop_deadline = deadline - destroy_reserve
+    remaining = stop_deadline - loop.time()
+    if remaining > 0:
+        # Escalate inside one environment call. Docker-backed environments pay
+        # a material control-plane round trip for every exec; separate TERM and
+        # KILL calls can consume a short framework budget even when KILL works.
+        # This command still targets only the session leader, so task-owned
+        # background processes remain available to the verifier.
+        await _stop_leader(
+            environment,
+            cwd,
+            scope_path,
+            grace_sec=min(0.1, max(0.001, remaining * 0.1)),
+            timeout_sec=remaining,
+        )
+        remaining = stop_deadline - loop.time()
+        if remaining > 0:
+            try:
+                return await asyncio.wait_for(
+                    asyncio.shield(execution), timeout=remaining
+                )
+            except asyncio.CancelledError:
+                if execution.cancelled():
+                    raise RuntimeError("Maka Eval subject execution was cancelled") from None
+                raise
+            except (TimeoutError, asyncio.TimeoutError):
+                pass
+    if execution.done() and not execution.cancelled():
+        return execution.result()
+    # A verifier cannot measure a stable environment while the subject may
+    # still be mutating it. Fail closed instead of publishing a scoreable
+    # framework_timeout frame without positive leader-exit evidence.
+    await _destroy_environment(environment, execution, deadline, loop)
+    raise RuntimeError("Maka Eval could not confirm subject exit after framework timeout")
+
+
 async def _settle_or_destroy(
     environment: Any,
     cwd: str,
@@ -600,21 +728,30 @@ async def _settle_or_destroy(
             timeout=max(0.001, deadline - loop.time() - stop_reserve),
         )
     except Exception:
-        try:
-            remaining = max(0.001, deadline - loop.time())
-            await asyncio.wait_for(environment.stop(delete=True), timeout=remaining)
-        except Exception:
-            pass
-        finally:
-            if not execution.done():
-                execution.cancel()
-            with contextlib.suppress(asyncio.CancelledError, Exception):
-                remaining = max(0.001, deadline - loop.time())
-                await asyncio.wait_for(execution, timeout=remaining)
+        await _destroy_environment(environment, execution, deadline, loop)
         return None
 
 
-async def _signal(environment: Any, cwd: str, scope_path: str, signal: str) -> None:
+async def _destroy_environment(
+    environment: Any,
+    execution: Any,
+    deadline: float,
+    loop: asyncio.AbstractEventLoop,
+) -> None:
+    try:
+        remaining = max(0.001, deadline - loop.time())
+        await asyncio.wait_for(environment.stop(delete=True), timeout=remaining)
+    except Exception:
+        pass
+    finally:
+        if not execution.done():
+            execution.cancel()
+        with contextlib.suppress(asyncio.CancelledError, Exception):
+            remaining = max(0.001, deadline - loop.time())
+            await asyncio.wait_for(execution, timeout=remaining)
+
+
+async def _signal_group(environment: Any, cwd: str, scope_path: str, signal: str) -> None:
     command = (
         f"pgid=$(cat {shlex.quote(scope_path)} 2>/dev/null) || exit 0; "
         "case $pgid in ''|0|*[!0-9]*) exit 0;; esac; "
@@ -628,11 +765,37 @@ async def _signal(environment: Any, cwd: str, scope_path: str, signal: str) -> N
         )
 
 
+async def _stop_leader(
+    environment: Any,
+    cwd: str,
+    scope_path: str,
+    grace_sec: float,
+    timeout_sec: float = 5,
+) -> bool:
+    command = (
+        f"pgid=$(cat {shlex.quote(scope_path)} 2>/dev/null) || exit 1; "
+        "case $pgid in ''|0|*[!0-9]*) exit 1;; esac; "
+        "kill -TERM -- \"$pgid\" 2>/dev/null || exit 1; "
+        f"sleep {grace_sec:.6f}; "
+        "kill -0 -- \"$pgid\" 2>/dev/null || exit 0; "
+        "kill -KILL -- \"$pgid\" 2>/dev/null"
+    )
+    try:
+        result = await environment.exec(
+            command,
+            cwd=cwd,
+            timeout_sec=max(0.001, timeout_sec),
+        )
+        return result.return_code == 0
+    except Exception:
+        return False
+
+
 async def _quiesce_scope(environment: Any, cwd: str, scope_path: str) -> None:
     if not await _scope_active(environment, cwd, scope_path):
         return
     for signal, timeout in (("TERM", 10), ("KILL", 5)):
-        await _signal(environment, cwd, scope_path, signal)
+        await _signal_group(environment, cwd, scope_path, signal)
         deadline = asyncio.get_running_loop().time() + timeout
         while asyncio.get_running_loop().time() < deadline:
             if not await _scope_active(environment, cwd, scope_path):
diff --git a/packages/eval/harbor/run_trial.py b/packages/eval/harbor/run_trial.py
index 9419ef67a1..efebdc4308 100644
--- a/packages/eval/harbor/run_trial.py
+++ b/packages/eval/harbor/run_trial.py
@@ -24,6 +24,7 @@
 import importlib
 import importlib.metadata
 import json
+import math
 import os
 import signal
 import sys
@@ -162,10 +163,43 @@ def apply_subject_egress_policy(task: object) -> None:
     agent.allowed_hosts = [allowed_host]
 
 
+def apply_framework_timeout_budget(config: object, task: object) -> None:
+    task_config = task.config
+    defaults = (
+        [
+            step.agent.timeout_sec
+            if step.agent.timeout_sec is not None
+            else task_config.agent.timeout_sec
+            for step in task_config.steps
+        ]
+        if task_config.steps
+        else [task_config.agent.timeout_sec]
+    )
+    multiplier = (
+        config.agent_timeout_multiplier
+        if config.agent_timeout_multiplier is not None
+        else config.timeout_multiplier
+    )
+    timeouts = []
+    for default_timeout in defaults:
+        base_timeout = config.agent.override_timeout_sec or default_timeout
+        if base_timeout is None:
+            continue
+        maximum = config.agent.max_timeout_sec or float("inf")
+        timeouts.append(min(base_timeout, maximum) * multiplier)
+    if not timeouts:
+        return
+    timeout_ms = math.floor(min(timeouts) * 1000)
+    if timeout_ms <= 0:
+        raise RuntimeError("Harbor agent timeout is shorter than one millisecond")
+    config.agent.kwargs["framework_timeout_ms"] = timeout_ms
+
+
 async def create_harbor_trial(trial_type: type, config: object) -> object:
     trial_type._resolve_agent_skills(config)
     task, task_download_result = await trial_type._load_task(config)
     apply_subject_egress_policy(task)
+    apply_framework_timeout_budget(config, task)
     if task.has_steps:
         from harbor.trial.multi_step import MultiStepTrial
 
diff --git a/packages/eval/harbor/test_harbor_trial_lifecycle.py b/packages/eval/harbor/test_harbor_trial_lifecycle.py
new file mode 100644
index 0000000000..6dd66350ea
--- /dev/null
+++ b/packages/eval/harbor/test_harbor_trial_lifecycle.py
@@ -0,0 +1,376 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+from __future__ import annotations
+
+import asyncio
+import importlib.metadata
+import json
+import os
+import shutil
+import subprocess
+import sys
+import tempfile
+import unittest
+import uuid
+from pathlib import Path
+from typing import Any
+
+
+ENABLED = os.environ.get("MAKA_EVAL_HARBOR_LIFECYCLE_TEST") == "1"
+HARBOR_VERSION = "0.20.0"
+RELAY_TOKEN = "0" * 32
+RESULT_TOKEN = "1" * 32
+ROOT = Path(__file__).resolve().parent
+
+
+@unittest.skipUnless(ENABLED, "real Harbor lifecycle test is opt-in")
+class HarborTrialLifecycleTest(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls) -> None:
+        if shutil.which("docker") is None:
+            raise AssertionError("Docker CLI is required for the real Harbor lifecycle test")
+        if importlib.metadata.version("harbor") != HARBOR_VERSION:
+            raise AssertionError(f"Harbor {HARBOR_VERSION} is required")
+        _docker("info", "--format", "{{.ServerVersion}}")
+        _docker("compose", "version")
+
+        if str(ROOT) not in sys.path:
+            sys.path.insert(0, str(ROOT))
+        import eval_framework
+
+        eval_framework.install("harbor")
+
+    def setUp(self) -> None:
+        import relay_agent
+
+        relay_agent._host_teardown_requested = False
+
+    def test_framework_timeout_preserves_background_for_verifier(self) -> None:
+        asyncio.run(self._framework_timeout_preserves_background_for_verifier())
+
+    def test_host_abort_destroys_background_environment(self) -> None:
+        asyncio.run(self._host_abort_destroys_background_environment())
+
+    async def _framework_timeout_preserves_background_for_verifier(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            task_dir = _write_task(root)
+            exchange = RelayExchange(wait_for_verification=True)
+            async with exchange.serve() as port:
+                trial = await _create_trial(
+                    task_dir,
+                    root / "trials",
+                    _trial_name("framework-timeout"),
+                    port,
+                    timeout_sec=1.0,
+                )
+                result = await asyncio.wait_for(trial.run(), timeout=120)
+
+            self.assertIsNone(result.exception_info)
+            self.assertIsNotNone(result.verifier_result)
+            assert result.verifier_result is not None
+            self.assertEqual(result.verifier_result.rewards, {"reward": 1.0})
+            self.assertIsNotNone(exchange.executed)
+            assert exchange.executed is not None
+            self.assertEqual(exchange.executed["termination"], "framework_timeout")
+            self.assertEqual(exchange.executed["exitCode"], 124)
+            self.assertIsNone(exchange.error)
+            self.assertEqual(_project_containers(trial.config.trial_name), [])
+
+    async def _host_abort_destroys_background_environment(self) -> None:
+        from relay_agent import request_host_teardown
+
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            task_dir = _write_task(root)
+            trial_name = _trial_name("host-abort")
+            exchange = RelayExchange(wait_for_verification=False)
+            trial_task: asyncio.Task[Any] | None = None
+            container_id: str | None = None
+            async with exchange.serve() as port:
+                trial = await _create_trial(
+                    task_dir,
+                    root / "trials",
+                    trial_name,
+                    port,
+                    timeout_sec=60.0,
+                )
+                try:
+                    trial_task = asyncio.create_task(trial.run())
+                    await asyncio.wait_for(exchange.execute_sent.wait(), timeout=60)
+                    container_id = await _wait_for_subject(trial_name)
+
+                    request_host_teardown()
+                    trial_task.cancel()
+                    with self.assertRaises(asyncio.CancelledError):
+                        await asyncio.wait_for(trial_task, timeout=30)
+                    await _wait_for_container_removal(container_id)
+                finally:
+                    if trial_task is not None and not trial_task.done():
+                        request_host_teardown()
+                        trial_task.cancel()
+                        try:
+                            await asyncio.wait_for(trial_task, timeout=30)
+                        except (asyncio.CancelledError, TimeoutError):
+                            pass
+                    _remove_project_containers(trial_name)
+
+            self.assertIsNone(exchange.executed)
+            self.assertIsNone(exchange.error)
+            self.assertEqual(_project_containers(trial_name), [])
+
+
+class RelayExchange:
+    def __init__(self, *, wait_for_verification: bool):
+        self.wait_for_verification = wait_for_verification
+        self.execute_sent = asyncio.Event()
+        self.executed: dict[str, Any] | None = None
+        self.error: BaseException | None = None
+        self._connections = 0
+        self._server: asyncio.Server | None = None
+
+    class _Serving:
+        def __init__(self, exchange: RelayExchange):
+            self.exchange = exchange
+
+        async def __aenter__(self) -> int:
+            server = await asyncio.start_server(self.exchange._handle, "127.0.0.1", 0)
+            self.exchange._server = server
+            socket = server.sockets[0]
+            return int(socket.getsockname()[1])
+
+        async def __aexit__(self, exc_type: Any, exc: Any, traceback: Any) -> None:
+            assert self.exchange._server is not None
+            self.exchange._server.close()
+            await self.exchange._server.wait_closed()
+
+    def serve(self) -> RelayExchange._Serving:
+        return self._Serving(self)
+
+    async def _handle(self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None:
+        self._connections += 1
+        try:
+            if self._connections != 1:
+                raise AssertionError("Harbor trial opened more than one relay connection")
+            ready = await _read_message(reader)
+            if ready.get("token") != RELAY_TOKEN or ready.get("kind") != "ready":
+                raise AssertionError(f"invalid relay ready frame: {ready!r}")
+            await _write_message(
+                writer,
+                {
+                    "token": RELAY_TOKEN,
+                    "kind": "execute",
+                    "command": "/bin/sh",
+                    "args": ["-c", _subject_script()],
+                    "credentials": {},
+                    "environment": {},
+                    "resultToken": RESULT_TOKEN,
+                    "captureStdout": False,
+                },
+            )
+            self.execute_sent.set()
+            if not self.wait_for_verification:
+                await reader.read()
+                return
+
+            executed = await _read_message(reader)
+            if executed.get("token") != RELAY_TOKEN or executed.get("kind") != "executed":
+                raise AssertionError(f"invalid relay execution frame: {executed!r}")
+            self.executed = executed
+            await _write_message(writer, {"token": RELAY_TOKEN, "kind": "verify"})
+            await reader.read()
+        except BaseException as error:
+            self.error = error
+        finally:
+            writer.close()
+            await writer.wait_closed()
+
+
+async def _create_trial(
+    task_dir: Path,
+    trials_dir: Path,
+    trial_name: str,
+    relay_port: int,
+    *,
+    timeout_sec: float,
+) -> Any:
+    from harbor.models.trial.config import TrialConfig
+    from harbor.trial.trial import Trial
+
+    config = TrialConfig.model_validate(
+        {
+            "task": {"path": str(task_dir)},
+            "trial_name": trial_name,
+            "trials_dir": str(trials_dir),
+            "agent": {
+                "import_path": "relay_agent:RelayAgent",
+                "override_timeout_sec": timeout_sec,
+                "kwargs": {
+                    "relay_host": "127.0.0.1",
+                    "relay_port": relay_port,
+                    "relay_token": RELAY_TOKEN,
+                    "teardown_timeout_ms": 5000,
+                    "framework_timeout_ms": int(timeout_sec * 1000),
+                },
+            },
+            "environment": {"type": "docker", "delete": True},
+            "verifier": {"override_timeout_sec": 10},
+        }
+    )
+    return await Trial.create(config)
+
+
+def _write_task(root: Path) -> Path:
+    task = root / "task"
+    (task / "environment").mkdir(parents=True)
+    (task / "tests").mkdir()
+    (task / "instruction.md").write_text("Keep a background service running.\n")
+    (task / "task.toml").write_text(
+        '\n'.join(
+            [
+                'version = "1.0"',
+                "",
+                "[agent]",
+                "timeout_sec = 60.0",
+                "",
+                "[verifier]",
+                "timeout_sec = 10.0",
+                "",
+                "[environment]",
+                "build_timeout_sec = 60.0",
+                "",
+            ]
+        )
+    )
+    (task / "environment" / "Dockerfile").write_text(
+        "FROM ubuntu:24.04\n"
+        "RUN command -v setsid >/dev/null\n"
+        "WORKDIR /app\n"
+    )
+    verifier = task / "tests" / "test.sh"
+    verifier.write_text(
+        "#!/bin/sh\n"
+        "reward=0\n"
+        "if test -s /tmp/maka-background.pid && "
+        "kill -0 \"$(cat /tmp/maka-background.pid)\" 2>/dev/null; then\n"
+        "  before=$(wc -c < /tmp/maka-background.heartbeat 2>/dev/null || printf 0)\n"
+        "  sleep 0.4\n"
+        "  after=$(wc -c < /tmp/maka-background.heartbeat 2>/dev/null || printf 0)\n"
+        "  if test \"$after\" -gt \"$before\"; then reward=1; fi\n"
+        "fi\n"
+        "printf '%s\\n' \"$reward\" > /logs/verifier/reward.txt\n"
+    )
+    verifier.chmod(0o755)
+    return task
+
+
+def _subject_script() -> str:
+    return (
+        "trap 'exit 0' TERM; "
+        "(trap '' TERM; while :; do printf x >> /tmp/maka-background.heartbeat; "
+        "sleep 0.1; done) & "
+        "child=$!; printf '%s\\n' \"$child\" > /tmp/maka-background.pid; "
+        "while :; do sleep 1; done"
+    )
+
+
+async def _read_message(reader: asyncio.StreamReader) -> dict[str, Any]:
+    raw = await asyncio.wait_for(reader.readline(), timeout=30)
+    if not raw:
+        raise AssertionError("relay connection closed before the expected frame")
+    value = json.loads(raw)
+    if not isinstance(value, dict):
+        raise AssertionError(f"relay frame is not an object: {value!r}")
+    return value
+
+
+async def _write_message(writer: asyncio.StreamWriter, value: dict[str, Any]) -> None:
+    writer.write(json.dumps(value, separators=(",", ":")).encode() + b"\n")
+    await writer.drain()
+
+
+async def _wait_for_subject(trial_name: str) -> str:
+    deadline = asyncio.get_running_loop().time() + 60
+    while asyncio.get_running_loop().time() < deadline:
+        containers = _project_containers(trial_name)
+        if len(containers) == 1:
+            container_id = containers[0]
+            ready = subprocess.run(
+                ["docker", "exec", container_id, "sh", "-c", "test -s /tmp/maka-background.pid"],
+                check=False,
+                capture_output=True,
+                text=True,
+            )
+            if ready.returncode == 0:
+                return container_id
+        await asyncio.sleep(0.1)
+    raise AssertionError(f"subject did not start in Harbor project {trial_name!r}")
+
+
+async def _wait_for_container_removal(container_id: str) -> None:
+    deadline = asyncio.get_running_loop().time() + 20
+    while asyncio.get_running_loop().time() < deadline:
+        inspected = subprocess.run(
+            ["docker", "inspect", container_id],
+            check=False,
+            capture_output=True,
+            text=True,
+        )
+        if inspected.returncode != 0:
+            return
+        await asyncio.sleep(0.1)
+    raise AssertionError(f"Harbor container {container_id} survived Host abort")
+
+
+def _project_containers(trial_name: str) -> list[str]:
+    project = f"{trial_name}__env"
+    result = _docker(
+        "ps",
+        "--all",
+        "--quiet",
+        "--filter",
+        f"label=com.docker.compose.project={project}",
+        "--filter",
+        "label=com.docker.compose.service=main",
+    )
+    return [line for line in result.splitlines() if line]
+
+
+def _remove_project_containers(trial_name: str) -> None:
+    containers = _project_containers(trial_name)
+    if containers:
+        _docker("rm", "--force", *containers)
+
+
+def _docker(*args: str) -> str:
+    completed = subprocess.run(
+        ["docker", *args],
+        check=True,
+        capture_output=True,
+        text=True,
+        timeout=30,
+    )
+    return completed.stdout.strip()
+
+
+def _trial_name(prefix: str) -> str:
+    return f"maka-{prefix}-{uuid.uuid4().hex[:12]}"
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/packages/eval/harbor/test_relay_lifecycle.py b/packages/eval/harbor/test_relay_lifecycle.py
index fe1e8d1fd6..7730e2f388 100644
--- a/packages/eval/harbor/test_relay_lifecycle.py
+++ b/packages/eval/harbor/test_relay_lifecycle.py
@@ -23,6 +23,8 @@
 import json
 import os
 import shutil
+import signal
+import socket
 import subprocess
 import sys
 import tempfile
@@ -98,6 +100,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
@@ -159,6 +172,113 @@ async def stop(self, delete=False):
         self.stopped = delete
 
 
+class TimeoutScopeEnvironment(SimultaneousEnvironment):
+    """A still-running subject whose descendants must survive framework timeout."""
+
+    def __init__(self):
+        super().__init__()
+        self.commands = []
+        self.stopped = False
+
+    async def stop(self, delete=False):
+        self.stopped = delete
+
+    async def exec(self, command, cwd=None, timeout_sec=None):
+        self.commands.append(command)
+        if _is_teardown(command):
+            raise AssertionError("framework timeout must not signal the process group")
+        if _is_leader_stop(command):
+            self.release.set()
+            return SimpleNamespace(return_code=0, stdout="", stderr="")
+        return await super().exec(command, cwd=cwd, timeout_sec=timeout_sec)
+
+
+class FrameworkBudgetEnvironment(TimeoutScopeEnvironment):
+    def __init__(self):
+        super().__init__()
+        self.leader_timeouts = []
+
+    async def exec(self, command, cwd=None, timeout_sec=None):
+        if _is_leader_stop(command):
+            self.commands.append(command)
+            self.leader_timeouts.append(timeout_sec)
+            self.release.set()
+            return SimpleNamespace(return_code=0, stdout="", stderr="")
+        return await super().exec(command, cwd=cwd, timeout_sec=timeout_sec)
+
+
+class IgnoringLeaderEnvironment(SimultaneousEnvironment):
+    """Delivers leader TERM/KILL but keeps the subject running."""
+
+    def __init__(self):
+        super().__init__()
+        self.commands = []
+        self.signal_timeouts = []
+        self.stopped = False
+
+    async def stop(self, delete=False):
+        self.stopped = delete
+
+    async def exec(self, command, cwd=None, timeout_sec=None):
+        self.commands.append(command)
+        if _is_leader_stop(command):
+            self.signal_timeouts.append(timeout_sec)
+        if _is_teardown(command):
+            raise AssertionError("framework timeout must not signal the process group")
+        if _is_leader_stop(command):
+            return SimpleNamespace(return_code=0, stdout="", stderr="")
+        return await super().exec(command, cwd=cwd, timeout_sec=timeout_sec)
+
+
+class DelayedLeaderControlEnvironment(TimeoutScopeEnvironment):
+    """A Docker-like control path with signal and result-observation latency."""
+
+    def __init__(self):
+        super().__init__()
+        self.signal_timeouts = []
+
+    async def exec(self, command, cwd=None, timeout_sec=None):
+        if command.startswith("setsid"):
+            self.started.set()
+            await self.release.wait()
+            await asyncio.sleep(0.05)
+            return SimpleNamespace(return_code=0, stdout="", stderr="")
+        if _is_leader_stop(command):
+            self.commands.append(command)
+            self.signal_timeouts.append(timeout_sec)
+            await asyncio.sleep(0.05)
+            if "kill -KILL" in command:
+                self.release.set()
+            return SimpleNamespace(return_code=0, stdout="", stderr="")
+        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__()
+        self.leader_stop_started = asyncio.Event()
+
+    async def exec(self, command, cwd=None, timeout_sec=None):
+        if _is_leader_stop(command):
+            self.leader_stop_started.set()
+            await asyncio.sleep(0.05)
+        return await super().exec(command, cwd=cwd, timeout_sec=timeout_sec)
+
+
 class LiveScopeEnvironment(SimultaneousEnvironment):
     """A subject whose process group outlives it, as a task's own service does."""
 
@@ -177,6 +297,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:
@@ -194,9 +327,17 @@ async def exec(self, command, cwd=None, timeout_sec=None):
 
 
 def _is_teardown(command: str) -> bool:
-    # `kill -0` is how the relay asks whether the scope is still there; only
-    # TERM and KILL end it.
-    return "kill -TERM" in command or "kill -KILL" in command
+    # Host abort emits `kill -- "-$pgid"`. Framework timeout emits
+    # `kill -- "$pgid"`, which must not count as teardown.
+    return ('"-$pgid"' in command) and ("kill -TERM" in command or "kill -KILL" in command)
+
+
+def _is_leader_stop(command: str) -> bool:
+    return (
+        ("kill -TERM" in command or "kill -KILL" in command)
+        and '"$pgid"' in command
+        and '"-$pgid"' not in command
+    )
 
 
 class RecordingEnvironment:
@@ -292,6 +433,136 @@ async def trial(*_args):
 
 
 class RelayLifecycleTest(unittest.IsolatedAsyncioTestCase):
+    async def test_exit_code_subject_detaches_both_output_streams(self):
+        relay = load_relay()
+        command = await relay._prepare_command(
+            ExplodingSubjectEnvironment(),
+            {
+                "command": "/bin/sh",
+                "args": ["-c", "run-subject"],
+                "credentials": {},
+                "resultToken": "0" * 32,
+                "captureStdout": False,
+            },
+            "detach-output",
+            "/tmp/maka-eval-detach-output.pid",
+        )
+
+        self.assertIn(">/dev/null 2>&1", command)
+
+    async def test_framework_timeout_uses_its_own_budget_and_completes_the_relay(self):
+        relay = load_relay()
+        environment = FrameworkBudgetEnvironment()
+        token = f"framework-deadline-{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=5_000,
+            framework_timeout_ms=50,
+        )
+        running = asyncio.create_task(
+            asyncio.wait_for(agent.run("solve", environment, None), timeout=0.05)
+        )
+        reader, writer = await connected
+        try:
+            await reader.readline()
+            writer.write((json.dumps({
+                "token": token, "kind": "execute", "command": "/bin/true", "args": [],
+                "credentials": {}, "resultToken": "0" * 32,
+            }) + "\n").encode())
+            await writer.drain()
+            executed = json.loads(await asyncio.wait_for(reader.readline(), 0.5))
+            self.assertEqual(executed["termination"], "framework_timeout")
+            await asyncio.wait_for(running, timeout=0.5)
+            self.assertNotEqual(environment.leader_timeouts, [])
+            self.assertTrue(
+                all(timeout is not None and timeout <= 0.05 for timeout in environment.leader_timeouts)
+            )
+            self.assertFalse(environment.stopped)
+        finally:
+            writer.close()
+            server.close()
+            await server.wait_closed()
+
+    async def test_leader_stop_exec_timeout_stays_inside_the_teardown_budget(self):
+        relay = load_relay()
+        environment = IgnoringLeaderEnvironment()
+        token = f"framework-budget-{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=50,
+        )
+        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()
+            await environment.started.wait()
+            running.cancel()
+            with self.assertRaisesRegex(RuntimeError, "could not confirm subject exit"):
+                await asyncio.wait_for(running, timeout=0.5)
+            self.assertNotEqual(environment.signal_timeouts, [])
+            self.assertTrue(all(timeout is not None and timeout <= 0.05 for timeout in environment.signal_timeouts))
+        finally:
+            writer.close()
+            server.close()
+            await server.wait_closed()
+
+    async def test_short_framework_timeout_escalates_in_one_control_round_trip(self):
+        relay = load_relay()
+        environment = DelayedLeaderControlEnvironment()
+        execution = asyncio.create_task(environment.exec("setsid subject"))
+        await environment.started.wait()
+
+        result = await relay._stop_subject_for_timeout(
+            environment,
+            "/workspace",
+            "/tmp/maka-eval-scope.pid",
+            execution,
+            0.2,
+        )
+
+        self.assertEqual(result.return_code, 0)
+        self.assertEqual(
+            [
+                signal
+                for signal in ("TERM", "KILL")
+                if any(f"kill -{signal}" in command for command in environment.commands)
+            ],
+            ["TERM", "KILL"],
+        )
+        self.assertEqual(len(environment.signal_timeouts), 1)
+        self.assertTrue(all(0 < timeout <= 0.16 for timeout in environment.signal_timeouts))
+        self.assertFalse(environment.stopped)
+
+    def test_scope_predicates_distinguish_group_teardown_from_leader_stop(self):
+        group = 'kill -TERM -- "-$pgid"'
+        leader = 'kill -TERM -- "$pgid"'
+        self.assertTrue(_is_teardown(group))
+        self.assertFalse(_is_leader_stop(group))
+        self.assertFalse(_is_teardown(leader))
+        self.assertTrue(_is_leader_stop(leader))
+
     def test_host_teardown_exits_successfully_after_trial_unwinds(self):
         with tempfile.TemporaryDirectory() as directory:
             marker = Path(directory) / "cleanup-attempted"
@@ -521,17 +792,16 @@ async def accept(reader, writer):
             executed = __import__("json").loads(await asyncio.wait_for(reader.readline(), 0.5))
             self.assertEqual(executed["termination"], "framework_timeout")
             self.assertEqual(executed["exitCode"], 124)
-            with self.assertRaises(asyncio.CancelledError):
-                await running
+            await running
         finally:
             writer.close()
             server.close()
             await server.wait_closed()
 
-    async def test_framework_timeout_survives_destroy_fallback(self):
+    async def test_unconfirmed_framework_timeout_is_not_reported_as_scoreable(self):
         relay = load_relay()
         environment = FrameworkTimeoutEnvironment()
-        token = f"framework-destroy-{os.getpid()}"
+        token = f"framework-timeout-{os.getpid()}"
         connected = asyncio.get_running_loop().create_future()
 
         async def accept(reader, writer):
@@ -545,6 +815,201 @@ async def accept(reader, writer):
         )
         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()
+            await environment.started.wait()
+            running.cancel()
+            self.assertEqual(await asyncio.wait_for(reader.readline(), 0.5), b"")
+            # SimultaneousEnvironment answers every `pgid=` command with 3, so
+            # `_stop_leader` reports a vanished leader. The completion race
+            # never delivers, so the relay must fail closed rather than score.
+            self.assertTrue(environment.stopped)
+            with self.assertRaisesRegex(RuntimeError, "could not confirm subject exit"):
+                await running
+        finally:
+            writer.close()
+            server.close()
+            await server.wait_closed()
+
+    async def test_ignored_leader_signals_fail_closed_without_a_scoreable_timeout(self):
+        relay = load_relay()
+        environment = IgnoringLeaderEnvironment()
+        token = f"framework-ignore-{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=50,
+        )
+        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()
+            await environment.started.wait()
+            running.cancel()
+            self.assertEqual(await asyncio.wait_for(reader.readline(), 0.5), b"")
+            self.assertTrue(any(_is_leader_stop(command) for command in environment.commands))
+            self.assertEqual(
+                [command for command in environment.commands if _is_teardown(command)],
+                [],
+            )
+            self.assertTrue(environment.stopped)
+            with self.assertRaisesRegex(RuntimeError, "could not confirm subject exit"):
+                await running
+        finally:
+            writer.close()
+            server.close()
+            await server.wait_closed()
+
+    async def test_second_cancel_during_timeout_cleanup_still_destroys(self):
+        relay = load_relay()
+        environment = SlowLeaderStopEnvironment()
+        token = f"framework-recancel-{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=200,
+        )
+        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()
+            await environment.started.wait()
+            running.cancel()
+            await environment.leader_stop_started.wait()
+            running.cancel()
+            with self.assertRaises(asyncio.CancelledError):
+                await asyncio.wait_for(running, timeout=1)
+            self.assertTrue(environment.stopped)
+        finally:
+            writer.close()
+            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()
+        token = f"framework-leader-{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({
@@ -557,10 +1022,48 @@ async def accept(reader, writer):
             executed = __import__("json").loads(await asyncio.wait_for(reader.readline(), 0.5))
             self.assertEqual(executed["termination"], "framework_timeout")
             self.assertEqual(executed["exitCode"], 124)
-            self.assertEqual(executed["diagnostic"]["category"], "result-frame-missing")
-            self.assertTrue(environment.stopped)
+            self.assertTrue(any(_is_leader_stop(command) for command in environment.commands))
+            self.assertEqual(
+                [command for command in environment.commands if _is_teardown(command)],
+                [],
+            )
+            self.assertFalse(environment.stopped)
+            await running
+        finally:
+            writer.close()
+            server.close()
+            await server.wait_closed()
+
+    async def test_host_teardown_of_a_running_subject_still_destroys(self):
+        relay = load_relay()
+        environment = FrameworkTimeoutEnvironment()
+        token = f"host-destroy-{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=50,
+        )
+        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()
+            await environment.started.wait()
+            relay.request_host_teardown()
+            running.cancel()
             with self.assertRaises(asyncio.CancelledError):
-                await running
+                await asyncio.wait_for(running, timeout=0.5)
+            self.assertTrue(environment.stopped)
         finally:
             writer.close()
             server.close()
@@ -649,6 +1152,70 @@ async def test_settle_kills_descendants_before_verification_boundary(self):
             finally:
                 Path(scope_path).unlink(missing_ok=True)
 
+    @unittest.skipUnless(shutil.which("setsid"), "requires GNU setsid")
+    async def test_framework_timeout_preserves_a_live_background_service(self):
+        relay = load_relay()
+        environment = LocalEnvironment()
+        token = f"timeout-service-{os.getpid()}"
+        scope_path = f"/tmp/maka-eval-{token}.pid"
+        service_pid = None
+        execution = None
+        with tempfile.TemporaryDirectory() as directory:
+            pid_path = Path(directory) / "service.pid"
+            port_path = Path(directory) / "service.port"
+            server = (
+                "import os,socket;"
+                "server=socket.socket();"
+                "server.setsockopt(socket.SOL_SOCKET,socket.SO_REUSEADDR,1);"
+                "server.bind(('127.0.0.1',0));server.listen(1);"
+                f"open({str(pid_path)!r},'w').write(str(os.getpid()));"
+                f"open({str(port_path)!r},'w').write(str(server.getsockname()[1]));"
+                "connection,_=server.accept();connection.sendall(b'alive');connection.close()"
+            )
+            subject = (
+                "import subprocess,sys,time;"
+                f"subprocess.Popen([sys.executable,'-c',{server!r}],"
+                "stdin=subprocess.DEVNULL,stdout=subprocess.DEVNULL,stderr=subprocess.DEVNULL);"
+                "time.sleep(60)"
+            )
+            request = {
+                "command": sys.executable,
+                "args": ["-c", subject],
+                "credentials": {},
+                "resultToken": "0" * 32,
+            }
+            try:
+                command = await relay._prepare_command(
+                    environment, request, token, scope_path
+                )
+                execution = asyncio.create_task(
+                    environment.exec(command, cwd=directory)
+                )
+                deadline = time.monotonic() + 2
+                while not (pid_path.exists() and port_path.exists()):
+                    if time.monotonic() >= deadline:
+                        self.fail("background service did not start")
+                    await asyncio.sleep(0.01)
+                service_pid = int(pid_path.read_text())
+                result = await relay._stop_subject_for_timeout(
+                    environment, directory, scope_path, execution, 2
+                )
+                self.assertIsNotNone(result)
+                with socket.create_connection(
+                    ("127.0.0.1", int(port_path.read_text())), timeout=1
+                ) as connection:
+                    self.assertEqual(connection.recv(5), b"alive")
+            finally:
+                if execution is not None and not execution.done():
+                    with contextlib.suppress(Exception):
+                        await relay._settle(
+                            environment, directory, scope_path, execution
+                        )
+                Path(scope_path).unlink(missing_ok=True)
+                if service_pid is not None:
+                    with contextlib.suppress(ProcessLookupError):
+                        os.kill(service_pid, signal.SIGTERM)
+
     async def test_an_exited_subject_is_left_alone_whatever_it_reported(self):
         # The verifier scores the environment the task was left in, so a subject
         # that stopped on its own keeps whatever it started — a service the task
diff --git a/packages/eval/harbor/test_run_trial_policy.py b/packages/eval/harbor/test_run_trial_policy.py
index 13e7d39814..de85577cee 100644
--- a/packages/eval/harbor/test_run_trial_policy.py
+++ b/packages/eval/harbor/test_run_trial_policy.py
@@ -32,6 +32,31 @@
 
 
 class RunTrialPolicyTest(unittest.TestCase):
+    def test_framework_cleanup_budget_uses_the_shortest_resolved_agent_timeout(self) -> None:
+        config = SimpleNamespace(
+            timeout_multiplier=2,
+            agent_timeout_multiplier=None,
+            agent=SimpleNamespace(
+                override_timeout_sec=None,
+                max_timeout_sec=4,
+                kwargs={"relay_token": "token"},
+            ),
+        )
+        task = SimpleNamespace(
+            config=SimpleNamespace(
+                agent=SimpleNamespace(timeout_sec=3),
+                steps=[
+                    SimpleNamespace(agent=SimpleNamespace(timeout_sec=None)),
+                    SimpleNamespace(agent=SimpleNamespace(timeout_sec=1.5)),
+                ],
+            )
+        )
+
+        MODULE.apply_framework_timeout_budget(config, task)
+
+        self.assertEqual(config.agent.kwargs["framework_timeout_ms"], 3000)
+        self.assertEqual(config.agent.kwargs["relay_token"], "token")
+
     def test_forces_only_the_subject_phase_through_the_cell_proxy(self) -> None:
         agent = SimpleNamespace(network_mode=None, allowed_hosts=None)
         task = SimpleNamespace(config=SimpleNamespace(agent=agent))
diff --git a/packages/eval/package.json b/packages/eval/package.json
index f23d7fce60..e66c4aca47 100644
--- a/packages/eval/package.json
+++ b/packages/eval/package.json
@@ -34,7 +34,7 @@
     "clean": "node ../../scripts/clean-paths.mjs dist tsconfig.tsbuildinfo",
     "build": "tsc -p tsconfig.json",
     "typecheck": "tsc -p tsconfig.json --noEmit",
-    "test:dist": "node --test \"dist/**/*.test.js\" && python3 harbor/test_eval_framework.py && python3 harbor/test_relay_contract.py && python3 harbor/test_relay_lifecycle.py && python3 harbor/test_egress_filter.py && python3 harbor/test_run_trial_policy.py && python3 harbor/test_relay_artifacts.py && python3 harbor/test_cell_egress_namespace.py",
+    "test:dist": "node --test \"dist/**/*.test.js\" && python3 harbor/test_eval_framework.py && python3 harbor/test_relay_contract.py && python3 harbor/test_relay_lifecycle.py && python3 harbor/test_harbor_trial_lifecycle.py && python3 harbor/test_egress_filter.py && python3 harbor/test_run_trial_policy.py && python3 harbor/test_relay_artifacts.py && python3 harbor/test_cell_egress_namespace.py",
     "test:egress-proxy:live": "python3 harbor/test_egress_filter_live.py"
   },
   "dependencies": {
diff --git a/packages/eval/src/__tests__/lifecycle-boundaries.test.ts b/packages/eval/src/__tests__/lifecycle-boundaries.test.ts
index c039dabe11..889d00baec 100644
--- a/packages/eval/src/__tests__/lifecycle-boundaries.test.ts
+++ b/packages/eval/src/__tests__/lifecycle-boundaries.test.ts
@@ -560,12 +560,11 @@ test('Maka forwards the configured Runtime Host settlement budget', async () =>
   );
 });
 
-// The relay tears the subject's process group down unless the wrapper exits
-// zero, so every wrapper has to project the same status the same way — an arm
-// whose failures exit zero would keep its background services through the
-// verifier while the others lose theirs. This pins both halves of the Maka
-// side: what the shim projects, and that the adapter reads the frame rather
-// than re-deciding from the code it just projected.
+// The wrapper's exit code is a projection of the status in its result frame,
+// carried for anything that can read only the exit code. An arm that exited
+// zero on failure would disagree with the frame. This pins both halves of the
+// Maka side: what the shim projects, and that the adapter reads the frame
+// rather than re-deciding from the code it just projected.
 test('the Maka shim projects only a completed subject as a zero exit', async () => {
   const shim = new URL('../harbor-maka-subject.js', import.meta.url);
   for (const [projection, expectedExit, expectedStatus] of [
@@ -588,7 +587,10 @@ test('the Maka shim projects only a completed subject as a zero exit', async ()
       const client = join(root, 'client.mjs');
       await writeFile(
         client,
-        `export async function runHostedExecution() { return ${JSON.stringify(frame)}; }\n`,
+        `export async function runHostedExecution(input) {
+  if (input.abortPolicy !== 'preserve_environment') throw new Error('missing Eval abort policy');
+  return ${JSON.stringify(frame)};
+}\n`,
       );
       const { exitCode, stdout } = await execFileAsync(
         process.execPath,
diff --git a/packages/eval/src/harbor-maka-subject.ts b/packages/eval/src/harbor-maka-subject.ts
index fbd82b2b2c..b98f940553 100644
--- a/packages/eval/src/harbor-maka-subject.ts
+++ b/packages/eval/src/harbor-maka-subject.ts
@@ -73,6 +73,7 @@ try {
     baseUrl: payload.baseUrl,
     execution: payload.execution,
     signal: abort.signal,
+    abortPolicy: 'preserve_environment',
     hostSettlementTimeoutMs: payload.hostSettlementTimeoutMs,
   });
 } finally {
diff --git a/packages/runtime-host/src/__tests__/execution-composition.test.ts b/packages/runtime-host/src/__tests__/execution-composition.test.ts
index 67180218e0..255718fed8 100644
--- a/packages/runtime-host/src/__tests__/execution-composition.test.ts
+++ b/packages/runtime-host/src/__tests__/execution-composition.test.ts
@@ -349,16 +349,23 @@ test('hosted execution settles while its tracked environment resource remains ve
     try {
       await composition.recover();
       const executionId = '00000000-0000-4000-8000-000000000111';
-      const execution = composition.handlers['hosted.execution.start'](
-        {
-          executionId,
-          session: {
-            workspace: { kind: 'host_path', path: root },
-            modelTarget: { kind: 'default' },
-            name: 'Hosted environment test',
-          },
-          content: { text: 'leave the environment ready for verification' },
+      const executionInput = {
+        executionId,
+        session: {
+          workspace: { kind: 'host_path' as const, path: root },
+          modelTarget: { kind: 'default' as const },
+          name: 'Hosted environment test',
         },
+        content: { text: 'leave the environment ready for verification' },
+      };
+      const admitted = await composition.handlers['hosted.execution.admit'](
+        executionInput,
+        operationContext,
+      );
+      assert.equal(admitted.ok, true);
+      if (!admitted.ok) return;
+      const execution = composition.handlers['hosted.execution.start'](
+        { execution: executionInput, admissionToken: admitted.result.admissionToken },
         operationContext,
       );
       let settled = false;
diff --git a/packages/runtime-host/src/__tests__/execution-model-composition.test.ts b/packages/runtime-host/src/__tests__/execution-model-composition.test.ts
index 424eaa35a6..539f9928f5 100644
--- a/packages/runtime-host/src/__tests__/execution-model-composition.test.ts
+++ b/packages/runtime-host/src/__tests__/execution-model-composition.test.ts
@@ -888,24 +888,28 @@ test('hosted execution freezes the headless coding provider wire contract', asyn
     );
     await composition.recover();
     const executionId = '00000000-0000-4000-8000-000000000777';
-    const outcome = await composition.handlers['hosted.execution.start'](
-      {
-        executionId,
-        session: {
-          workspace: { kind: 'host_path', path: root },
-          modelTarget: {
-            kind: 'explicit',
-            connectionSlug: 'profile-deepseek',
-            model: 'deepseek-v4-flash',
-          },
-          permissionMode: 'bypass',
-          collaborationMode: 'agent',
-          orchestrationMode: 'default',
-          toolProfile: 'headless-coding-v1',
+    const executionInput = {
+      executionId,
+      session: {
+        workspace: { kind: 'host_path' as const, path: root },
+        modelTarget: {
+          kind: 'explicit' as const,
+          connectionSlug: 'profile-deepseek',
+          model: 'deepseek-v4-flash',
         },
-        content: { text: 'Complete the benchmark task.' },
-        maxSteps: 100_000,
+        permissionMode: 'bypass' as const,
+        collaborationMode: 'agent' as const,
+        orchestrationMode: 'default' as const,
+        toolProfile: 'headless-coding-v1' as const,
       },
+      content: { text: 'Complete the benchmark task.' },
+      maxSteps: 100_000,
+    };
+    const admitted = await composition.handlers['hosted.execution.admit'](executionInput, context);
+    assert.equal(admitted.ok, true);
+    if (!admitted.ok) return;
+    const outcome = await composition.handlers['hosted.execution.start'](
+      { execution: executionInput, admissionToken: admitted.result.admissionToken },
       context,
     );
     assert.equal(outcome.ok, true);
diff --git a/packages/runtime-host/src/__tests__/hosted-execution-client.test.ts b/packages/runtime-host/src/__tests__/hosted-execution-client.test.ts
index 596cfab0ab..ff767b8e3f 100644
--- a/packages/runtime-host/src/__tests__/hosted-execution-client.test.ts
+++ b/packages/runtime-host/src/__tests__/hosted-execution-client.test.ts
@@ -19,13 +19,15 @@
 
 import assert from 'node:assert/strict';
 import test from 'node:test';
+import { RuntimeHostRequestInterruptedError } from '../client/connection.js';
 import { runHostedExecutionWithDependencies } from '../client/hosted-execution.js';
 
 test('diagnostics disconnect after settlement preserves the canonical result', async () => {
   for (const status of ['completed', 'failed'] as const) {
     const projection = settled(status);
     const connected = ownedHost({
-      request: async () => projection,
+      request: async (operation: string) =>
+        operation === 'hosted.execution.admit' ? admission() : projection,
       queryHostDiagnostics: async () => {
         throw new Error('connection closed');
       },
@@ -44,6 +46,7 @@ test('abort observed with a completed response does not replace the result', asy
   const projection = settled('completed');
   const connected = ownedHost({
     request: async (operation: string) => {
+      if (operation === 'hosted.execution.admit') return admission();
       if (operation === 'hosted.execution.start') {
         abort.abort();
         return projection;
@@ -110,6 +113,135 @@ 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 admitted = deferred();
+  const closed = deferred();
+  const events: string[] = [];
+  const connected = ownedHost({
+    request: async (operation: string, requestInput: unknown) => {
+      events.push(operation);
+      if (operation === 'hosted.execution.admit') {
+        admitted.resolve();
+        return admission();
+      }
+      if (operation !== 'hosted.execution.start') {
+        throw new Error(`Unexpected operation ${operation}`);
+      }
+      assert.deepEqual(requestInput, {
+        execution: input().execution,
+        admissionToken: ADMISSION_TOKEN,
+      });
+      await closed.promise;
+      throw dispatchedInterrupt();
+    },
+  });
+  connected.connection.close = async () => {
+    if (events.includes('close')) return;
+    events.push('close');
+    closed.resolve();
+  };
+  connected.host.releaseToEnvironment = () => {
+    events.push('release');
+  };
+  connected.host.settle = async () => {
+    assert.fail('a detached Host must not settle');
+  };
+
+  const execution = runHostedExecutionWithDependencies(
+    { ...input(abort.signal), abortPolicy: 'preserve_environment' },
+    { connectOwnedRuntimeHost: async () => connected as never },
+  );
+  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.admit',
+    'close',
+    'hosted.execution.start',
+    'release',
+  ]);
+});
+
+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.admit') {
+        throw new Error(`Unexpected operation ${operation}`);
+      }
+      started.resolve();
+      await closed.promise;
+      throw dispatchedInterrupt();
+    },
+  });
+  connected.connection.close = async () => {
+    if (events.includes('close')) return;
+    events.push('close');
+    closed.resolve();
+  };
+  connected.host.releaseToEnvironment = () => {
+    events.push('release');
+  };
+  connected.host.settle = async () => {
+    events.push('settle');
+    return true;
+  };
+
+  const execution = runHostedExecutionWithDependencies(
+    { ...input(abort.signal), abortPolicy: 'preserve_environment' },
+    { connectOwnedRuntimeHost: async () => connected as never },
+  );
+  await started.promise;
+  abort.abort();
+  const result = await execution;
+
+  assert.equal(result.kind, 'indeterminate');
+  assert.equal(result.failureReason, 'Hosted execution was cancelled');
+  assert.deepEqual(events, ['hosted.execution.admit', 'close', 'settle']);
+});
+
+test('environment-preserving abort before start does not claim execution continues', async () => {
+  const abort = new AbortController();
+  const events: string[] = [];
+  const connected = ownedHost({
+    request: async (operation: string) => {
+      events.push(operation);
+      throw new Error(`Unexpected operation ${operation}`);
+    },
+  });
+  connected.host.releaseToEnvironment = () => {
+    events.push('release');
+  };
+  connected.host.settle = async () => {
+    events.push('settle');
+    return true;
+  };
+
+  const result = await runHostedExecutionWithDependencies(
+    { ...input(abort.signal), abortPolicy: 'preserve_environment' },
+    {
+      connectOwnedRuntimeHost: async () => {
+        abort.abort();
+        return connected as never;
+      },
+    },
+  );
+
+  assert.equal(result.kind, 'indeterminate');
+  assert.equal(result.failureReason, 'Hosted execution was cancelled');
+  assert.equal(events.includes('hosted.execution.start'), false);
+  assert.equal(events.includes('release'), false);
+  assert.equal(events.includes('settle'), true);
+});
+
 test('explicit target mutation settles the first Host before execution reconnects', async () => {
   const projection = settled('completed');
   const events: string[] = [];
@@ -117,6 +249,7 @@ test('explicit target mutation settles the first Host before execution reconnect
   const second = ownedHost({
     request: async (operation: string) => {
       events.push(`second:${operation}`);
+      if (operation === 'hosted.execution.admit') return admission();
       assert.equal(operation, 'hosted.execution.start');
       return projection;
     },
@@ -147,6 +280,7 @@ test('explicit target mutation settles the first Host before execution reconnect
     'first:close',
     'first:settle',
     'connect:2',
+    'second:hosted.execution.admit',
     'second:hosted.execution.start',
     'second:close',
     'second:release',
@@ -160,6 +294,7 @@ test('explicit target already admitted executes without reconnecting', async ()
     request: async (operation: string) => {
       events.push(operation);
       if (operation === 'connection.catalog.query') return catalogPage(true);
+      if (operation === 'hosted.execution.admit') return admission();
       if (operation === 'hosted.execution.start') return projection;
       throw new Error(`Unexpected operation ${operation}`);
     },
@@ -181,6 +316,7 @@ test('explicit target already admitted executes without reconnecting', async ()
   assert.deepEqual(events, [
     'connection.catalog.query',
     'connection.catalog.query',
+    'hosted.execution.admit',
     'hosted.execution.start',
     'release',
   ]);
@@ -223,6 +359,11 @@ 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 admission() {
+  return { executionId: ID, admissionToken: ADMISSION_TOKEN };
+}
 
 function input(signal?: AbortSignal) {
   return {
@@ -358,3 +499,20 @@ function ownedHost(connection: Record<string, unknown>, clean = false) {
     },
   };
 }
+
+function dispatchedInterrupt() {
+  return new RuntimeHostRequestInterruptedError(
+    'hosted.execution.admit',
+    'command',
+    'dispatched',
+    'connection_lost',
+  );
+}
+
+function deferred() {
+  let resolve!: () => void;
+  const promise = new Promise<void>((accept) => {
+    resolve = accept;
+  });
+  return { promise, resolve };
+}
diff --git a/packages/runtime-host/src/__tests__/hosted-execution-coordinator.test.ts b/packages/runtime-host/src/__tests__/hosted-execution-coordinator.test.ts
index ed448e41dc..968a8bfb97 100644
--- a/packages/runtime-host/src/__tests__/hosted-execution-coordinator.test.ts
+++ b/packages/runtime-host/src/__tests__/hosted-execution-coordinator.test.ts
@@ -35,7 +35,10 @@ test('cancel before start prevents the hosted execution from running', async ()
     { executionId: ID },
     context(),
   );
-  const started = await coordinator.handlers['hosted.execution.start'](input(), context());
+  const started = await coordinator.handlers['hosted.execution.start'](
+    { execution: input(), admissionToken: TOKEN },
+    context(),
+  );
 
   assert.equal(cancelled.ok, true);
   assert.equal(started.ok, true);
@@ -43,6 +46,35 @@ 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'](
+    { execution: input(), admissionToken: admitted.result.admissionToken },
+    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(
@@ -51,14 +83,169 @@ test('cancelling a settled subject reclaims its verification environment', async
       drains += 1;
     },
   );
-  await coordinator.handlers['hosted.execution.start'](input(), context());
+  const admitted = await coordinator.handlers['hosted.execution.admit'](input(), context());
+  assert.equal(admitted.ok, true);
+  if (!admitted.ok) return;
+  await coordinator.handlers['hosted.execution.start'](
+    { execution: input(), admissionToken: admitted.result.admissionToken },
+    context(),
+  );
 
   await coordinator.handlers['hosted.execution.cancel']({ executionId: ID }, context());
 
   assert.equal(drains, 1);
 });
 
+test('start without server admission is rejected without running', async () => {
+  let runs = 0;
+  const coordinator = new HostHostedExecutionCoordinator(
+    async (execution) => {
+      runs += 1;
+      return settled(execution.executionId, 'completed');
+    },
+    () => {},
+  );
+
+  const started = await coordinator.handlers['hosted.execution.start'](
+    { execution: input(), admissionToken: TOKEN },
+    context(),
+  );
+
+  assert.equal(started.ok, false);
+  if (!started.ok) assert.equal(started.error.code, 'invalid_request');
+  assert.equal(runs, 0);
+});
+
+test('wrong admission token cannot claim an admitted execution', async () => {
+  let runs = 0;
+  const coordinator = new HostHostedExecutionCoordinator(
+    async (execution) => {
+      runs += 1;
+      return settled(execution.executionId, 'completed');
+    },
+    () => {},
+  );
+  const admitted = await coordinator.handlers['hosted.execution.admit'](input(), context());
+  assert.equal(admitted.ok, true);
+  if (admitted.ok) assert.notEqual(admitted.result.admissionToken, TOKEN);
+
+  const started = await coordinator.handlers['hosted.execution.start'](
+    { execution: input(), admissionToken: TOKEN },
+    context(),
+  );
+
+  assert.equal(started.ok, false);
+  if (!started.ok) assert.equal(started.error.code, 'operation_conflict');
+  assert.equal(runs, 1);
+});
+
+test('fast settlement remains cached and cannot execute twice', async () => {
+  let runs = 0;
+  const coordinator = new HostHostedExecutionCoordinator(
+    async (execution) => {
+      runs += 1;
+      return settled(execution.executionId, 'completed');
+    },
+    () => {},
+  );
+  const admitted = await coordinator.handlers['hosted.execution.admit'](input(), context());
+  assert.equal(admitted.ok, true);
+  if (!admitted.ok) return;
+  await new Promise<void>((resolve) => setImmediate(resolve));
+
+  const startInput = { execution: input(), admissionToken: admitted.result.admissionToken };
+  const first = await coordinator.handlers['hosted.execution.start'](startInput, context());
+  const replay = await coordinator.handlers['hosted.execution.start'](startInput, context());
+
+  assert.equal(first.ok, true);
+  assert.deepEqual(replay, first);
+  assert.equal(runs, 1);
+});
+
+test('one-shot Host rejects every different execution identity and retains one receipt', async () => {
+  let runs = 0;
+  let aborts = 0;
+  const coordinator = new HostHostedExecutionCoordinator(
+    async (execution, signal) => {
+      runs += 1;
+      signal.addEventListener('abort', () => {
+        aborts += 1;
+      });
+      return settled(execution.executionId, 'completed');
+    },
+    () => {},
+  );
+  const admitted = await coordinator.handlers['hosted.execution.admit'](input(), context());
+  assert.equal(admitted.ok, true);
+
+  const outcomes = await Promise.all(
+    Array.from({ length: 200 }, (_, index) =>
+      coordinator.handlers['hosted.execution.admit'](
+        input(`00000000-0000-4000-8001-${String(index).padStart(12, '0')}`),
+        context(),
+      ),
+    ),
+  );
+
+  assert.equal(
+    outcomes.every((outcome) => !outcome.ok),
+    true,
+  );
+  for (const outcome of outcomes) {
+    if (!outcome.ok) assert.equal(outcome.error.code, 'operation_conflict');
+  }
+  assert.equal(runs, 1);
+  assert.equal(aborts, 0);
+  coordinator.beginDrain();
+  assert.equal(aborts, 1);
+});
+
+test('cancelling first binds the one-shot Host to that execution identity', async () => {
+  const coordinator = new HostHostedExecutionCoordinator(
+    async (execution) => settled(execution.executionId, 'completed'),
+    () => {},
+  );
+  const cancelled = await coordinator.handlers['hosted.execution.cancel'](
+    { executionId: ID },
+    context(),
+  );
+  assert.equal(cancelled.ok, true);
+
+  const different = await coordinator.handlers['hosted.execution.admit'](
+    input('00000000-0000-4000-8000-000000000002'),
+    context(),
+  );
+  assert.equal(different.ok, false);
+  if (!different.ok) assert.equal(different.error.code, 'operation_conflict');
+});
+
+test('admission authority cannot cross connection or Host epoch', async () => {
+  const coordinator = new HostHostedExecutionCoordinator(
+    async (execution) => settled(execution.executionId, 'completed'),
+    () => {},
+  );
+  const owner = context();
+  const admitted = await coordinator.handlers['hosted.execution.admit'](input(), owner);
+  assert.equal(admitted.ok, true);
+  if (!admitted.ok) return;
+  const startInput = { execution: input(), admissionToken: admitted.result.admissionToken };
+
+  for (const foreign of [
+    context({ connectionId: 'other-connection' }),
+    context({ hostEpoch: 'other-epoch' }),
+  ]) {
+    const started = await coordinator.handlers['hosted.execution.start'](startInput, foreign);
+    assert.equal(started.ok, false);
+    if (!started.ok) assert.equal(started.error.code, 'operation_conflict');
+
+    const readmitted = await coordinator.handlers['hosted.execution.admit'](input(), foreign);
+    assert.equal(readmitted.ok, false);
+    if (!readmitted.ok) assert.equal(readmitted.error.code, 'operation_conflict');
+  }
+});
+
 const ID = '00000000-0000-4000-8000-000000000001';
+const TOKEN = '00000000-0000-4000-8000-0000000000ff';
 
 function settled(executionId: string, status: 'completed' | 'failed') {
   return {
@@ -78,9 +265,9 @@ function settled(executionId: string, status: 'completed' | 'failed') {
   };
 }
 
-function input() {
+function input(executionId = ID) {
   return {
-    executionId: ID,
+    executionId,
     session: {
       workspace: { kind: 'host_path' as const, path: '/workspace' },
       modelTarget: { kind: 'explicit' as const, connectionSlug: 'env-openai', model: 'model' },
@@ -89,11 +276,19 @@ function input() {
   };
 }
 
-function context() {
+function context(overrides: { hostEpoch?: string; connectionId?: string } = {}) {
   return {
-    hostEpoch: 'host-epoch',
-    connectionId: 'hosted-execution',
+    hostEpoch: overrides.hostEpoch ?? 'host-epoch',
+    connectionId: overrides.connectionId ?? 'hosted-execution',
     principal: 'runtime_host' as const,
     acquireResidency: () => ({ release() {} }),
   };
 }
+
+function deferred() {
+  let resolve!: () => void;
+  const promise = new Promise<void>((accept) => {
+    resolve = accept;
+  });
+  return { promise, resolve };
+}
diff --git a/packages/runtime-host/src/__tests__/hosted-execution-tool-profile.test.ts b/packages/runtime-host/src/__tests__/hosted-execution-tool-profile.test.ts
index 3e3bf291bc..0e8e76bc0e 100644
--- a/packages/runtime-host/src/__tests__/hosted-execution-tool-profile.test.ts
+++ b/packages/runtime-host/src/__tests__/hosted-execution-tool-profile.test.ts
@@ -21,7 +21,10 @@ import assert from 'node:assert/strict';
 import test from 'node:test';
 import type { MakaTool } from '@maka/runtime/tool-runtime';
 import { z } from 'zod';
-import { decodeHostedExecutionStartInput } from '../protocol/index.js';
+import {
+  decodeHostedExecutionAdmittedStartInput,
+  decodeHostedExecutionStartInput,
+} from '../protocol/index.js';
 import {
   hostedExecutionRunProfile,
   projectHostedExecutionTools,
@@ -48,6 +51,27 @@ test('hosted execution tool profiles are durable Session creation inputs', () =>
   );
 });
 
+test('hosted execution start requires the server admission token', () => {
+  const execution = decodeHostedExecutionStartInput({
+    executionId: '00000000-0000-4000-8000-000000000001',
+    session: {
+      workspace: { kind: 'host_path', path: '/workspace' },
+      modelTarget: { kind: 'explicit', connectionSlug: 'provider', model: 'model' },
+    },
+    content: { text: 'solve' },
+  });
+  const decoded = decodeHostedExecutionAdmittedStartInput({
+    execution,
+    admissionToken: '00000000-0000-4000-8000-000000000002',
+  });
+  assert.deepEqual(decoded.execution, execution);
+  assert.equal(decoded.admissionToken, '00000000-0000-4000-8000-000000000002');
+  assert.throws(
+    () => decodeHostedExecutionAdmittedStartInput({ execution, admissionToken: undefined }),
+    /admissionToken/u,
+  );
+});
+
 test('the headless coding profile freezes prompt, tools, memory, and foreground Bash', async () => {
   const profile = hostedExecutionRunProfile('headless-coding-v1');
   assert.ok(profile);
diff --git a/packages/runtime-host/src/__tests__/protocol.test.ts b/packages/runtime-host/src/__tests__/protocol.test.ts
index 7bcbdbc7ae..90545a5fc4 100644
--- a/packages/runtime-host/src/__tests__/protocol.test.ts
+++ b/packages/runtime-host/src/__tests__/protocol.test.ts
@@ -97,6 +97,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(
       () =>
diff --git a/packages/runtime-host/src/client/hosted-execution.ts b/packages/runtime-host/src/client/hosted-execution.ts
index fbce39b1f5..1124c7cab0 100644
--- a/packages/runtime-host/src/client/hosted-execution.ts
+++ b/packages/runtime-host/src/client/hosted-execution.ts
@@ -25,7 +25,7 @@ import {
   type HostedExecutionStartInput,
 } from '../protocol/index.js';
 import { connectOwnedRuntimeHost } from './connect-or-spawn.js';
-import type { RuntimeHostConnection } from './connection.js';
+import { type RuntimeHostConnection } from './connection.js';
 import { configureHostedExecutionTarget } from './hosted-execution-target.js';
 
 export interface RunHostedExecutionInput {
@@ -33,6 +33,7 @@ export interface RunHostedExecutionInput {
   readonly execution: HostedExecutionStartInput;
   readonly baseUrl?: string;
   readonly signal?: AbortSignal;
+  readonly abortPolicy?: 'cancel' | 'preserve_environment';
   readonly hostSettlementTimeoutMs?: number;
 }
 
@@ -76,6 +77,7 @@ export async function runHostedExecutionWithDependencies(
     { kind: 'connected' }
   > = initial;
   let projection: HostedExecutionProjection;
+  let detached = false;
   try {
     input.signal?.throwIfAborted();
     const target = input.execution.session.modelTarget;
@@ -114,7 +116,15 @@ export async function runHostedExecutionWithDependencies(
         connected = reconnected;
       }
     }
-    projection = await executeHostedExecution(connected.connection, input.execution, input.signal);
+    const execution = await executeHostedExecution(
+      connected.connection,
+      connected.host,
+      input.execution,
+      input.signal,
+      input.abortPolicy ?? 'cancel',
+    );
+    projection = execution.projection;
+    detached = execution.detached;
   } catch {
     projection = input.signal?.aborted
       ? indeterminate(input.execution.executionId, 'Hosted execution was cancelled')
@@ -126,6 +136,7 @@ export async function runHostedExecutionWithDependencies(
     await connected.connection.close().catch(() => undefined);
   }
 
+  if (detached) return projection;
   if (preservesHostedExecutionEnvironment(projection)) {
     connected.host.releaseToEnvironment();
     return projection;
@@ -137,21 +148,72 @@ export async function runHostedExecutionWithDependencies(
 }
 
 async function executeHostedExecution(
-  connection: Pick<RuntimeHostConnection, 'request'>,
+  connection: Pick<RuntimeHostConnection, 'request' | 'close'>,
+  host: { releaseToEnvironment(): void },
   execution: HostedExecutionStartInput,
   signal: AbortSignal | undefined,
-): Promise<HostedExecutionProjection> {
-  const cancel = () => {
+  abortPolicy: NonNullable<RunHostedExecutionInput['abortPolicy']>,
+): Promise<{ readonly projection: HostedExecutionProjection; readonly detached: boolean }> {
+  let closeForAbort: Promise<void> | undefined;
+  const onAbort = () => {
+    if (abortPolicy === 'preserve_environment') {
+      closeForAbort = connection.close().catch(() => undefined);
+      return;
+    }
     void connection
       .request('hosted.execution.cancel', { executionId: execution.executionId })
       .catch(() => undefined);
   };
-  signal?.addEventListener('abort', cancel, { once: true });
-  if (signal?.aborted) cancel();
+  signal?.addEventListener('abort', onAbort, { once: true });
+  if (signal?.aborted) {
+    signal.removeEventListener('abort', onAbort);
+    return {
+      projection: indeterminate(execution.executionId, 'Hosted execution was cancelled'),
+      detached: false,
+    };
+  }
+  let admissionToken: string | undefined;
   try {
-    return await connection.request('hosted.execution.start', execution);
+    const admission = await connection.request('hosted.execution.admit', execution);
+    admissionToken = admission.admissionToken;
+    const projection = await connection.request('hosted.execution.start', {
+      execution,
+      admissionToken,
+    });
+    if (signal?.aborted && abortPolicy === 'preserve_environment' && admissionToken) {
+      host.releaseToEnvironment();
+      return {
+        projection: preservesHostedExecutionEnvironment(projection)
+          ? projection
+          : indeterminate(
+              execution.executionId,
+              'Hosted execution continues for environment verification',
+            ),
+        detached: true,
+      };
+    }
+    return { projection, detached: false };
+  } catch (error) {
+    if (signal?.aborted && abortPolicy === 'preserve_environment') {
+      if (admissionToken) {
+        host.releaseToEnvironment();
+        return {
+          projection: indeterminate(
+            execution.executionId,
+            'Hosted execution continues for environment verification',
+          ),
+          detached: true,
+        };
+      }
+      return {
+        projection: indeterminate(execution.executionId, 'Hosted execution was cancelled'),
+        detached: false,
+      };
+    }
+    throw error;
   } finally {
-    signal?.removeEventListener('abort', cancel);
+    signal?.removeEventListener('abort', onAbort);
+    await closeForAbort;
   }
 }
 
diff --git a/packages/runtime-host/src/protocol/hosted-execution.ts b/packages/runtime-host/src/protocol/hosted-execution.ts
index f1fee49dea..a0aef92bf4 100644
--- a/packages/runtime-host/src/protocol/hosted-execution.ts
+++ b/packages/runtime-host/src/protocol/hosted-execution.ts
@@ -50,6 +50,16 @@ export interface HostedExecutionReferenceInput {
   readonly executionId: string;
 }
 
+export interface HostedExecutionAdmissionAck {
+  readonly executionId: string;
+  readonly admissionToken: string;
+}
+
+export interface HostedExecutionAdmittedStartInput {
+  readonly execution: HostedExecutionStartInput;
+  readonly admissionToken: string;
+}
+
 export interface HostedExecutionUsage {
   readonly inputTokens: number;
   readonly outputTokens: number;
@@ -84,9 +94,9 @@ export function preservesHostedExecutionEnvironment(
 }
 
 export const HOSTED_EXECUTION_OPERATION_SPECS = {
-  'hosted.execution.start': defineOperation<
+  'hosted.execution.admit': defineOperation<
     HostedExecutionStartInput,
-    HostedExecutionProjection,
+    HostedExecutionAdmissionAck,
     (typeof ERRORS)[number]
   >({
     mode: 'command',
@@ -94,6 +104,18 @@ export const HOSTED_EXECUTION_OPERATION_SPECS = {
     errors: ERRORS,
     usesHostPaths: () => true,
     decodeInput: decodeHostedExecutionStartInput,
+    decodeOutput: decodeHostedExecutionAdmissionAck,
+  }),
+  'hosted.execution.start': defineOperation<
+    HostedExecutionAdmittedStartInput,
+    HostedExecutionProjection,
+    (typeof ERRORS)[number]
+  >({
+    mode: 'command',
+    availability: 'ready',
+    errors: ERRORS,
+    usesHostPaths: () => true,
+    decodeInput: decodeHostedExecutionAdmittedStartInput,
     decodeOutput: decodeHostedExecutionProjection,
   }),
   'hosted.execution.cancel': defineOperation<
@@ -131,11 +153,35 @@ export function decodeHostedExecutionStartInput(value: unknown): HostedExecution
   };
 }
 
+export function decodeHostedExecutionAdmittedStartInput(
+  value: unknown,
+): HostedExecutionAdmittedStartInput {
+  const input = requireExactRecord(value, 'Admitted Hosted execution start input', [
+    'execution',
+    'admissionToken',
+  ]);
+  return {
+    execution: decodeHostedExecutionStartInput(input.execution),
+    admissionToken: requireEntityId(input.admissionToken, 'admissionToken'),
+  };
+}
+
 export function decodeHostedExecutionReferenceInput(value: unknown): HostedExecutionReferenceInput {
   const input = requireExactRecord(value, 'Hosted execution reference', ['executionId']);
   return { executionId: requireEntityId(input.executionId, 'executionId') };
 }
 
+export function decodeHostedExecutionAdmissionAck(value: unknown): HostedExecutionAdmissionAck {
+  const ack = requireExactRecord(value, 'Hosted execution admission', [
+    'executionId',
+    'admissionToken',
+  ]);
+  return {
+    executionId: requireEntityId(ack.executionId, 'executionId'),
+    admissionToken: requireEntityId(ack.admissionToken, 'admissionToken'),
+  };
+}
+
 export function decodeHostedExecutionProjection(value: unknown): HostedExecutionProjection {
   const result = requireRecord(value, 'Hosted execution projection');
   const executionId = requireEntityId(result.executionId, 'executionId');
diff --git a/packages/runtime-host/src/protocol/index.ts b/packages/runtime-host/src/protocol/index.ts
index 0edebdf8df..525b68fb3c 100644
--- a/packages/runtime-host/src/protocol/index.ts
+++ b/packages/runtime-host/src/protocol/index.ts
@@ -91,7 +91,10 @@ export const RUNTIME_HOST_REGISTRATION_SCHEMA_VERSION = 1 as const;
 export const RUNTIME_HOST_PROTOCOL_VERSION = 0 as const;
 // Increment when the same protocol version no longer guarantees safe Client-Host
 // interoperability. Mismatches are rejected before domain commands are admitted.
-export const RUNTIME_HOST_COMPATIBILITY_EPOCH = 46 as const;
+export const RUNTIME_HOST_COMPATIBILITY_EPOCH = 47 as const;
+// 47: Hosted execution requires a server-issued admission token on start and
+// binds it to one Host epoch and connection. Older peers cannot enforce
+// at-most-once execution or prove admission before preserve-detach cleanup.
 // 46: Queued message content can be edited in place (queue.entry.update).
 // 45: Connection onboarding inputs require `baseUrl` and `connectionId`, and
 // results can carry the `base_url_not_configured` / `connection_not_found`
diff --git a/packages/runtime-host/src/server/hosted-execution-coordinator.ts b/packages/runtime-host/src/server/hosted-execution-coordinator.ts
index e8307b0813..abd1faf91d 100644
--- a/packages/runtime-host/src/server/hosted-execution-coordinator.ts
+++ b/packages/runtime-host/src/server/hosted-execution-coordinator.ts
@@ -17,31 +17,42 @@
  * under the License.
  */
 
+import { randomUUID } from 'node:crypto';
 import { isDeepStrictEqual } from 'node:util';
 import { preservesHostedExecutionEnvironment } from '../protocol/index.js';
 import type {
+  HostedExecutionAdmittedStartInput,
   HostedExecutionProjection,
   HostedExecutionReferenceInput,
   HostedExecutionStartInput,
   OperationOutcome,
 } from '../protocol/index.js';
-import type { HostedExecutionOperationHandlerMap } from './operation-dispatcher.js';
+import type {
+  ConnectionContext,
+  HostedExecutionOperationHandlerMap,
+} from './operation-dispatcher.js';
+
+interface HostedExecutionRecord {
+  readonly input: HostedExecutionStartInput;
+  readonly authority: {
+    readonly hostEpoch: string;
+    readonly connectionId: string;
+  };
+  readonly abort: AbortController;
+  readonly admissionToken: string;
+  readonly task: Promise<HostedExecutionProjection>;
+}
 
 export class HostHostedExecutionCoordinator {
   readonly handlers: HostedExecutionOperationHandlerMap = {
-    'hosted.execution.start': (input) => this.#start(input),
+    'hosted.execution.admit': (input, context) => this.#admit(input, context),
+    'hosted.execution.start': (input, context) => this.#start(input, context),
     'hosted.execution.cancel': (input) => this.#cancel(input),
   };
 
-  readonly #executions = new Map<
-    string,
-    {
-      readonly input: HostedExecutionStartInput;
-      readonly abort: AbortController;
-      readonly task: Promise<HostedExecutionProjection>;
-    }
-  >();
+  readonly #executions = new Map<string, HostedExecutionRecord>();
   readonly #cancelled = new Set<string>();
+  #executionId: string | undefined;
   #accepting = true;
 
   constructor(
@@ -62,41 +73,117 @@ export class HostHostedExecutionCoordinator {
     await Promise.all([...this.#executions.values()].map(({ task }) => task));
   }
 
-  async #start(
+  async #admit(
     input: HostedExecutionStartInput,
-  ): Promise<OperationOutcome<'hosted.execution.start'>> {
+    context: ConnectionContext,
+  ): Promise<OperationOutcome<'hosted.execution.admit'>> {
+    if (this.#executionId !== undefined && this.#executionId !== input.executionId) {
+      return conflict();
+    }
     if (this.#cancelled.has(input.executionId)) {
-      this.requestDrain();
       return {
-        ok: true,
-        result: indeterminate(input.executionId, 'Hosted execution was cancelled before admission'),
+        ok: false,
+        error: {
+          code: 'invalid_request',
+          message: 'Hosted execution was cancelled before admission',
+        },
       };
     }
     const existing = this.#executions.get(input.executionId);
     if (existing) {
-      if (!isDeepStrictEqual(existing.input, input)) return conflict();
-      return { ok: true, result: structuredClone(await existing.task) };
+      if (
+        !isDeepStrictEqual(existing.input, input) ||
+        !sameAuthority(existing.authority, context)
+      ) {
+        return conflict();
+      }
+      return {
+        ok: true,
+        result: { executionId: input.executionId, admissionToken: existing.admissionToken },
+      };
     }
     if (!this.#accepting) {
-      return { ok: false, error: { code: 'host_draining', message: 'Runtime Host is draining' } };
+      return {
+        ok: false,
+        error: { code: 'host_draining', message: 'Runtime Host is draining' },
+      };
     }
+
+    this.#executionId = input.executionId;
+    const execution = this.#createExecution(input, context);
+    return {
+      ok: true,
+      result: { executionId: input.executionId, admissionToken: execution.admissionToken },
+    };
+  }
+
+  async #start(
+    input: HostedExecutionAdmittedStartInput,
+    context: ConnectionContext,
+  ): Promise<OperationOutcome<'hosted.execution.start'>> {
+    if (this.#executionId !== undefined && this.#executionId !== input.execution.executionId) {
+      return conflict();
+    }
+    if (this.#cancelled.has(input.execution.executionId)) {
+      this.requestDrain();
+      return {
+        ok: true,
+        result: indeterminate(
+          input.execution.executionId,
+          'Hosted execution was cancelled before admission',
+        ),
+      };
+    }
+    const existing = this.#executions.get(input.execution.executionId);
+    if (!existing) {
+      return {
+        ok: false,
+        error: {
+          code: 'invalid_request',
+          message: 'Hosted execution was not admitted',
+        },
+      };
+    }
+    if (
+      existing.admissionToken !== input.admissionToken ||
+      !isDeepStrictEqual(existing.input, input.execution) ||
+      !sameAuthority(existing.authority, context)
+    ) {
+      return conflict();
+    }
+    return { ok: true, result: structuredClone(await existing.task) };
+  }
+
+  #createExecution(
+    input: HostedExecutionStartInput,
+    context: ConnectionContext,
+  ): HostedExecutionRecord {
     const abort = new AbortController();
+    const admissionToken = randomUUID();
     const task = this.run(input, abort.signal)
       .catch(() => indeterminate(input.executionId, 'Runtime Host could not settle execution'))
       .then((result) => {
         if (!preservesHostedExecutionEnvironment(result)) this.requestDrain();
         return result;
-      })
-      .finally(() => {
-        this.#executions.delete(input.executionId);
       });
-    this.#executions.set(input.executionId, { input: structuredClone(input), abort, task });
-    return { ok: true, result: structuredClone(await task) };
+    const execution = {
+      input: structuredClone(input),
+      authority: { hostEpoch: context.hostEpoch, connectionId: context.connectionId },
+      abort,
+      admissionToken,
+      task,
+    };
+    this.#executions.set(input.executionId, execution);
+    return execution;
   }
 
   async #cancel(
     input: HostedExecutionReferenceInput,
   ): Promise<OperationOutcome<'hosted.execution.cancel'>> {
+    if (this.#executionId !== undefined && this.#executionId !== input.executionId) {
+      return conflict();
+    }
+    this.#executionId = input.executionId;
     this.#cancelled.add(input.executionId);
     const execution = this.#executions.get(input.executionId);
     if (!execution) {
@@ -107,15 +194,28 @@ export class HostHostedExecutionCoordinator {
       };
     }
     execution.abort.abort();
-    return { ok: true, result: structuredClone(await execution.task) };
+    const result = await execution.task;
+    if (preservesHostedExecutionEnvironment(result)) this.requestDrain();
+    return { ok: true, result: structuredClone(result) };
   }
 }
 
-function conflict(): OperationOutcome<'hosted.execution.start'> {
+function sameAuthority(
+  authority: HostedExecutionRecord['authority'],
+  context: ConnectionContext,
+): boolean {
+  return (
+    authority.hostEpoch === context.hostEpoch && authority.connectionId === context.connectionId
+  );
+}
+
+function conflict<
+  K extends 'hosted.execution.admit' | 'hosted.execution.start' | 'hosted.execution.cancel',
+>(): OperationOutcome<K> {
   return {
     ok: false,
     error: { code: 'operation_conflict', message: 'Hosted execution identity is already in use' },
-  };
+  } as OperationOutcome<K>;
 }
 
 function indeterminate(executionId: string, failureReason: string): HostedExecutionProjection {
diff --git a/scripts/ci-test-plan.test.mjs b/scripts/ci-test-plan.test.mjs
index 5324bebe6d..29d03138fa 100644
--- a/scripts/ci-test-plan.test.mjs
+++ b/scripts/ci-test-plan.test.mjs
@@ -317,6 +317,22 @@ test('core CI validates affected installed CLI packages on its existing runner',
   assert.match(workflow, /run: npm run release:cli:smoke/u);
 });
 
+test('Eval changes run the pinned real Harbor lifecycle contract', () => {
+  const workflow = readWorkflow('ci.yml');
+  const install = workflow.indexOf('      - name: Install Harbor lifecycle test dependencies\n');
+  const run = workflow.indexOf('      - name: Run real Harbor lifecycle tests\n');
+
+  assert.ok(install >= 0);
+  assert.ok(run > install);
+  for (const start of [install, run]) {
+    const step = workflow.slice(start, workflow.indexOf('\n      - ', start + 1));
+    assert.match(step, /contains\(steps\.plan\.outputs\.standard_workspaces, 'packages\/eval'\)/u);
+  }
+  assert.match(workflow.slice(install, run), /'harbor==0\.20\.0'/u);
+  assert.match(workflow.slice(run), /MAKA_EVAL_HARBOR_LIFECYCLE_TEST: '1'/u);
+  assert.match(workflow.slice(run), /packages\/eval\/harbor\/test_harbor_trial_lifecycle\.py/u);
+});
+
 test('release contracts run against built CLI outputs', () => {
   const workflow = readWorkflow('ci.yml');
   const buildIndex = workflow.indexOf('      - name: Build\n');