test(code-execution): cover cell-ship eviction and fail-closed setup

The kernel eviction on a failed cell ship and the raise in _execute_checked
had no test teeth: removing either left the suite green. Extend the existing
shared-host lockdown test (no new test functions) so a failed cell ship must
raise and empty the registry, and a failed dir setup must spawn nothing and
ship nothing.

Also reuse the loop's parsed sandbox.env ship/token in the per-call lockdown
test instead of re-parsing it, and fix a comment that still described the
removed pipe-vs-heredoc stdin branch.

tests/tools/test_code_execution.py

@@ -254,6 +254,7 @@ class TestRemoteSharedHostLockdown(unittest.TestCase):
         # The kernel path runs first and fails open here (no PID), so both the
         # kernel.env and sandbox.env ships are recorded. Neither token may
         # appear in any executed command.
+        ships, tokens = {}, {}
         for env_name in ("kernel.env", "sandbox.env"):
             env_ship = next((c for c in ship_mock.call_args_list
                              if c.args[1].endswith(env_name)), None)
@@ -264,6 +265,7 @@ class TestRemoteSharedHostLockdown(unittest.TestCase):
             self.assertTrue(token)
             self.assertFalse(any(token in c for c in commands),
                              f"{env_name} token appeared in a remote command line")
+            ships[env_name], tokens[env_name] = env_ship, token
         run_cmd = next(c for c in commands if "python3 script.py" in c)
         self.assertNotIn("HERMES_RPC_TOKEN=", run_cmd)
         if sys.platform == "win32":
@@ -275,8 +277,8 @@ class TestRemoteSharedHostLockdown(unittest.TestCase):
         import tempfile
         mkdir_cmd = next(c for c in commands
                          if "mkdir -p" in c and "hermes_exec_" in c)
-        sandbox = next(c.args[1] for c in ship_mock.call_args_list
-                       if c.args[1].endswith("sandbox.env")).rsplit("/", 1)[0]
+        env_ship, token = ships["sandbox.env"], tokens["sandbox.env"]
+        sandbox = env_ship.args[1].rsplit("/", 1)[0]
         root = tempfile.mkdtemp()
         self.addCleanup(shutil.rmtree, root, True)
         local = root + sandbox
@@ -285,11 +287,6 @@ class TestRemoteSharedHostLockdown(unittest.TestCase):
         self.assertEqual(sh(mkdir_cmd).returncode, 0)
         for d in (local, f"{local}/rpc"):
             self.assertEqual(os.stat(d).st_mode & 0o777, 0o700, d)
-        env_ship = next(c for c in ship_mock.call_args_list
-                        if c.args[1].endswith("sandbox.env"))
-        token = next(l for l in env_ship.args[2].splitlines()
-                     if l.startswith("HERMES_RPC_TOKEN="))
-        token = token.split("=", 1)[1].strip("'\"")
         with open(f"{local}/sandbox.env", "w") as fh:
             fh.write(env_ship.args[2])
         with open(f"{local}/script.py", "w") as fh:

tests/tools/test_code_kernel_remote.py

@@ -376,9 +376,8 @@ class TestSharedHostLockdown(RemoteKernelBase):
             self.assertEqual(r.returncode, 0, r.stderr)
             for d in (local, f"{local}/cells", f"{local}/rpc"):
                 self.assertEqual(os.stat(d).st_mode & 0o777, 0o700, d)
-        # Ships write owner-only; on a pipe-capable backend the payload rides
-        # stdin, so the base64 (which decodes to the token for kernel.env)
-        # never enters argv either.
+        # Ships write owner-only and carry the base64 as stdin_data, so it
+        # (which decodes to the token for kernel.env) never enters argv.
         ship_cmds = [c for c in env.commands if "base64 -d" in c]
         self.assertTrue(any("kernel.env" in c for c in ship_cmds))
         import base64
@@ -386,6 +385,19 @@ class TestSharedHostLockdown(RemoteKernelBase):
                         if p and "kernel.env" in c)
         env_content = base64.b64decode(env_ship).decode()
         self.assertIn(f"HERMES_RPC_TOKEN={kernel.rpc_token}", env_content)
+        # Fail closed on a failed cell ship: the checked write raises and the
+        # kernel is evicted, so the next call cannot reuse a kernel whose
+        # state silently missed this cell.
+        env.handlers.insert(0, ("cell_req_", lambda c: {"output": "ENOSPC", "returncode": 1}))
+        with self.assertRaises(RuntimeError):
+            _run(env, timeout=1)
+        self.assertEqual(len(_REMOTE_KERNELS), 0)
+        # Fail closed on a failed dir setup: nothing (env file, runner) is
+        # shipped into a dir that may still be missing or permissive.
+        env = ScriptedEnv([("mkdir -p", lambda c: {"output": "EACCES", "returncode": 1})]
+                          + _spawn_ok_handlers([_cell()]))
+        self.assertIsNone(_run(env))
+        self.assertFalse(any("base64 -d" in c for c in env.commands), env.commands)
 
 
 if __name__ == "__main__":