@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import errno
 import hashlib
 import json
 import re
@@ -14,6 +15,9 @@
 ROOT = Path(__file__).resolve().parent.parent
 TASKS_DIR = ROOT / "tasks"
 ATTACKER_IMAGE = "rb-attacker:latest"
+# Linux limits each argv string to 128 KiB including its terminator. Leave
+# room for platform differences and report model-generated excess as a command failure.
+MAX_COMMAND_BYTES = 120_000
 COMPOSE_PROJECT_LABEL = "com.docker.compose.project"
 # Docker's inspect API has added and removed non-image Config fields across versions.
 # Keep the fields that belong to the image configuration.
@@ -381,6 +385,13 @@ def exec(
         the attempt if Docker still has not returned.
         """
 
+        try:
+            command_bytes = len(cmd.encode("utf-8"))
+        except UnicodeEncodeError:
+            return 1, "[command contains invalid UTF-8]"
+        if command_bytes > MAX_COMMAND_BYTES:
+            return 1, f"[command exceeds {MAX_COMMAND_BYTES} UTF-8 bytes]"
+
         def attacker_available() -> bool:
             try:
                 probe = subprocess.run(
@@ -393,6 +404,8 @@ def attacker_available() -> bool:
                 )
             except subprocess.TimeoutExpired as exc:
                 raise EnvError("Docker exec probe timed out") from exc
+            except OSError as exc:
+                raise EnvError(f"Docker exec probe could not start (errno {exc.errno})") from exc
             return probe.returncode == 0
 
         try:
@@ -449,6 +462,10 @@ def attacker_available() -> bool:
             if not attacker_available():
                 raise EnvError("Docker exec unavailable after attacker command timeout") from exc
             raise EnvError("Docker exec did not finish after attacker command timeout") from exc
+        except OSError as exc:
+            if exc.errno == errno.E2BIG:
+                return 1, "[command could not start: argument list too long]"
+            raise EnvError(f"Docker exec could not start (errno {exc.errno})") from exc
 
     def read_flag(self, stage: Stage) -> str:
         compose = [
 
@@ -539,7 +539,9 @@ def run_attempt(
 
     def emit(kind: str, **kv: Any) -> None:
         rec: dict[str, Any] = {"t": round(time.time() - t0, 1), "kind": kind, **kv}
-        log.write(json.dumps(rec, ensure_ascii=False) + "\n")
+        # Model output may contain lone Unicode surrogates. Escape them so a
+        # malformed command can receive an ordinary observation below.
+        log.write(json.dumps(rec, ensure_ascii=True) + "\n")
         log.flush()
 
     try:
 
@@ -1,4 +1,5 @@
 import argparse
+import errno
 import json
 import tempfile
 import subprocess
@@ -15,7 +16,15 @@
     cmd_preflight,
     cmd_run,
 )
-from rangebench.env import EnvError, Stage, Task, TaskEnv, _run, image_content_fingerprint
+from rangebench.env import (
+    MAX_COMMAND_BYTES,
+    EnvError,
+    Stage,
+    Task,
+    TaskEnv,
+    _run,
+    image_content_fingerprint,
+)
 from rangebench.runner import AttemptResult, _maybe_compact
 
 
@@ -394,6 +403,40 @@ def binary_output(_cmd: list[str], **kwargs):
         with patch("rangebench.env.subprocess.run", side_effect=binary_output):
             self.assertEqual(env.exec("cat binary-file"), (0, "out\ufffd\n[stderr]\nerr\ufffd"))
 
+    def test_oversized_utf8_command_is_scored_without_spawning_docker(self) -> None:
+        task = Task("sample", Path("/tmp"), "web", 1, "Find the flag")
+        env = TaskEnv(task, "rb-test")
+        command = "é" * (MAX_COMMAND_BYTES // 2 + 1)
+        with patch("rangebench.env.subprocess.run") as run:
+            rc, output = env.exec(command)
+        self.assertEqual(rc, 1)
+        self.assertIn("exceeds", output)
+        self.assertNotIn(command[:20], output)
+        run.assert_not_called()
+
+    def test_invalid_utf8_command_is_scored_without_spawning_docker(self) -> None:
+        task = Task("sample", Path("/tmp"), "web", 1, "Find the flag")
+        env = TaskEnv(task, "rb-test")
+        with patch("rangebench.env.subprocess.run") as run:
+            self.assertEqual(env.exec("echo \udcff"), (1, "[command contains invalid UTF-8]"))
+        run.assert_not_called()
+
+    def test_e2big_from_subprocess_is_scored_without_command_content(self) -> None:
+        task = Task("sample", Path("/tmp"), "web", 1, "Find the flag")
+        env = TaskEnv(task, "rb-test")
+        with patch("rangebench.env.subprocess.run", side_effect=OSError(errno.E2BIG, "too long")):
+            self.assertEqual(
+                env.exec("private-model-command"),
+                (1, "[command could not start: argument list too long]"),
+            )
+
+    def test_other_subprocess_start_error_invalidates_attempt(self) -> None:
+        task = Task("sample", Path("/tmp"), "web", 1, "Find the flag")
+        env = TaskEnv(task, "rb-test")
+        with patch("rangebench.env.subprocess.run", side_effect=OSError(errno.ENOENT, "missing")):
+            with self.assertRaisesRegex(EnvError, "Docker exec could not start"):
+                env.exec("echo ok")
+
     def test_attacker_uses_recorded_image_id(self) -> None:
         task = Task("sample", Path("/tmp"), "web", 1, "Find the flag")
         env = TaskEnv(task, "rb-test", "sha256:recorded")
 
@@ -6,7 +6,7 @@
 from unittest.mock import patch
 
 from rangebench.agent import Usage
-from rangebench.env import ATTACKER_IMAGE, EnvError, Stage, Task
+from rangebench.env import ATTACKER_IMAGE, EnvError, Stage, Task, TaskEnv
 from rangebench.runner import run_attempt, run_oracle
 
 
@@ -41,6 +41,51 @@ def chat(self, *_args: object, **_kwargs: object) -> None:
 
 
 class RunnerTests(unittest.TestCase):
+    def test_unencodable_model_command_gets_observation_without_aborting(self) -> None:
+        class GuardedEnv(TaskEnv):
+            def up(self) -> None:
+                self.service_image_ids = {"target": "sha256:" + "a" * 64}
+                self.service_image_fingerprints = {"target": "sha256:" + "b" * 64}
+
+            def read_flag(self, stage: Stage) -> str:
+                return "flag{test}"
+
+            def down(self) -> tuple[bool, str]:
+                return True, ""
+
+        class SurrogateClient:
+            def __init__(self) -> None:
+                self.calls = 0
+
+            def chat(
+                self, _messages: list[dict], max_tokens: int, temperature: float = 0.2
+            ) -> tuple[str, Usage, None]:
+                self.calls += 1
+                content = "COMMAND:\necho \udcff" if self.calls == 1 else "ANSWER: flag{test}"
+                return content, Usage(prompt_tokens=20, completion_tokens=10), None
+
+        with (
+            tempfile.TemporaryDirectory() as tmp,
+            patch("rangebench.runner.TaskEnv", GuardedEnv),
+            patch("rangebench.env.subprocess.run") as run,
+        ):
+            task = Task(
+                "sample", Path(tmp), "web", 1, "Find the flag",
+                stages=[Stage("one", "/flag", "target")], turns=2,
+            )
+            result = run_attempt(SurrogateClient(), task, 1, "rb-test", Path(tmp), verbose=False)
+            raw = (Path(tmp) / "sample-t1.jsonl").read_text()
+            records = [json.loads(line) for line in raw.splitlines()]
+
+        self.assertEqual(result.end_reason, "all stages captured")
+        self.assertEqual(result.commands, 1)
+        self.assertIn("\\udcff", raw)
+        self.assertEqual(
+            next(record for record in records if record["kind"] == "exec")["out"],
+            "[command contains invalid UTF-8]",
+        )
+        run.assert_not_called()
+
     def test_keep_writes_event_before_closing_log(self) -> None:
         with tempfile.TemporaryDirectory() as tmp, patch("rangebench.runner.TaskEnv", FakeEnv):
             task = Task("sample", Path(tmp), "web", 1, "Find the flag", stages=[Stage("one", "/flag", "target")])