Host-level sandbox tasks: LocalExecutor + bwrap workspace-write

Docker outer layer and bwrap inner layer are mutually exclusive
on this host (fresh proc mounts fail with kernel EPERM inside
Docker's nested pidns, but work on the host under a userns
profile). So tasks that need an inner sandbox run on the host:
- Worker --claim-executor-only: strict executor claim (only
  executor == own type, never NULL). The host worker MUST use
  it, or it would steal NULL-executor tasks meant for Docker
  workers (those tasks would lose the Docker outer layer).
  Default behavior unchanged (NULL claimable by any worker).
- host-config-host.json: sandbox workspace-write, state
  work/host-state-host.json (distinct from Docker tasks'
  host-state.json), same relay home and test repo as Docker.
- deploy/apparmor/codex-host: docker-codex base + userns,
  mount, pivot_root for bwrap; still denies raw network,
  ptrace, dangerous capabilities, /proc/sys writes.
- deploy/host-worker/codex-host-worker.service: aa-exec -p
  codex-host, --executor local --claim-executor-only,
  capacity 2, proxy env 127.0.0.1:8888 so host tasks' relay/git
  egress also goes through the Tinyproxy allowlist (host direct
  egress would bypass it).
POST /tasks already accepts executor (local/docker/cloud), so
no API change was needed for routing.
Tests: 4 host-sandbox-routing tests (strict claim incl. CAS,
default still takes NULL, config sandbox/state, profile
allows bwrap without deny mount). Full suites pass.

cloud/tests/test_host_sandbox_routing.py

@@ -0,0 +1,99 @@
+"""宿主级沙箱路由单测：严格 executor 认领 + 配置/profile 文件。
+
+只用 SQLite 测试库与假 runner，不跑真实 codex/Docker。
+覆盖：claim_executor_only 开启时 _ready_tasks 只返回
+executor == 'local' 的任务（NULL 与 docker 都排除），且
+_claim 对 NULL 任务返回 False、对 local 任务返回 True；
+host-config-host.json 的 sandbox 确为 workspace-write；
+codex-host profile 含 userns 与 mount，且不含 deny mount。
+"""
+
+import json
+import os
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+
+try:
+    from cloud.server import init_db
+    from cloud.worker import Worker
+except ImportError:
+    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
+    from cloud.server import init_db
+    from cloud.worker import Worker
+
+import sqlite3
+
+
+def _seed(db_path):
+    init_db(db_path)
+    conn = sqlite3.connect(db_path)
+    conn.execute(
+        "INSERT INTO projects (id, name) VALUES ('p1', 'p1')")
+    conn.execute(
+        "INSERT INTO tasks (id, project_id, prompt, status, executor)"
+        " VALUES ('null-task', 'p1', 'hi', 'queued', NULL)")
+    conn.execute(
+        "INSERT INTO tasks (id, project_id, prompt, status, executor)"
+        " VALUES ('docker-task', 'p1', 'hi', 'queued', 'docker')")
+    conn.execute(
+        "INSERT INTO tasks (id, project_id, prompt, status, executor)"
+        " VALUES ('local-task', 'p1', 'hi', 'queued', 'local')")
+    conn.commit()
+    conn.close()
+
+
+class StrictClaimTestCase(unittest.TestCase):
+    def test_strict_worker_only_takes_local(self):
+        tmp = tempfile.TemporaryDirectory()
+        db_path = os.path.join(tmp.name, "control.db")
+        _seed(db_path)
+        worker = Worker(db_path, "/fake/host.py",
+                        env={"XCM_API_KEY": "test-placeholder"},
+                        claim_executor_only=True)
+        ready = worker._ready_tasks()
+        self.assertEqual([t["id"] for t in ready], ["local-task"])
+        # CAS 层同样严格：NULL 任务抢不到，local 任务抢得到。
+        self.assertFalse(worker._claim("null-task", executor="local"))
+        self.assertTrue(worker._claim("local-task", executor="local"))
+        tmp.cleanup()
+
+    def test_default_worker_still_takes_null(self):
+        tmp = tempfile.TemporaryDirectory()
+        db_path = os.path.join(tmp.name, "control.db")
+        _seed(db_path)
+        worker = Worker(db_path, "/fake/host.py",
+                        env={"XCM_API_KEY": "test-placeholder"})
+        ready = worker._ready_tasks()
+        ids = [t["id"] for t in ready]
+        self.assertIn("null-task", ids)
+        self.assertIn("local-task", ids)
+        self.assertNotIn("docker-task", ids)
+        tmp.cleanup()
+
+
+class HostConfigProfileTestCase(unittest.TestCase):
+    def test_host_config_uses_workspace_write(self):
+        repo_root = Path(__file__).resolve().parents[2]
+        cfg = json.loads((repo_root / "host-config-host.json").read_text())
+        self.assertEqual(cfg["sandbox"], "workspace-write")
+        self.assertEqual(cfg["state"], "work/host-state-host.json")
+        self.assertEqual(cfg["codex_home"],
+                         "/var/lib/codex-app-server/codex-home")
+
+    def test_codex_host_profile_allows_bwrap(self):
+        repo_root = Path(__file__).resolve().parents[2]
+        text = (repo_root / "deploy" / "apparmor" / "codex-host"
+                ).read_text()
+        self.assertIn("userns,", text)
+        # bwrap 需要 mount 与 pivot_root：profile 必须允许，
+        # 不能像 docker-codex 那样 deny。
+        self.assertIn("mount,", text)
+        self.assertIn("pivot_root,", text)
+        self.assertNotIn("deny mount,", text)
+        self.assertNotIn("deny umount,", text)
+
+
+if __name__ == "__main__":
+    unittest.main()