feat: side-thread task kind (Muse role alignment phase 2)

Tasks gain kind='main'|'side' (NULL=main). A side thread hangs under a
parent task, defaults to the local executor, never builds a worktree,
branch or container unless explicitly asked, and writes its conclusion
back to the parent via task_followups on terminal transition.

- server: kind validation + side-thread gates (parent required, no
  coordinator_parts/upload_id/environment_id, parent must not be a
  coordinator), local executor default, GET /tasks kind filter,
  /worktree 404 and /fork /merge /pr 409 for side threads.
- worker: side-default shunt in _run_attempt (no-worktree scratch cwd,
  in-memory plan=1 when a repo is present, seed/checkpoint skipped),
  _writeback_side_thread, marker task-create kind field + inheritance,
  side threads excluded from marker spawn counts.
- host/sandbox: --no-worktree plumbing for dispatch/goal/popen
  (Local + Docker), scratch cwd work/side-threads/<task-id>.
- console (classic only): kind filter + [旁聊] rows + '开旁聊' entry,
  side-thread dispatch mode, side-thread list in detail, merge/PR/fork
  hidden for side threads; build exit 0, console.html sha df39db5a.
- tests: new test_sidethread.py (29 tests) + kind backfill tests in
  test_p2_migrations.py; full cloud suite 3900 tests OK (skipped=13).

Design: verification/codex-muse-role-alignment-design-20261108.md
Summary: verification/codex-muse-role-align-p1p2-20261108.md
Local only: not pushed, not merged, not deployed.

cloud/tests/test_longtask_auto_resume.py

@@ -185,7 +185,8 @@ class LongtaskAutoResumeTestCase(unittest.TestCase):
         captured = {}
 
         def fake_execute(task, kind="dispatch", deadline_seconds=None,
-                         seed_host_dir=None, quota_check=None):
+                         seed_host_dir=None, no_worktree=False,
+                         quota_check=None):
             captured["deadline"] = deadline_seconds
             return {"status": "completed", "stdout": "", "stderr": ""}
 

cloud/tests/test_p2_migrations.py

@@ -240,6 +240,48 @@ class TestMigrateNewTables(unittest.TestCase):
                     "result_text", "error_text", "created_at"):
             self.assertIn(col, attempts_inserts[0])
 
+    def test_kind_column_backfilled_when_missing(self):
+        # P2：PG 侧 tasks 尚无 kind 列时，迁移必须先 ALTER 补列
+        # 再直插，且插入列清单里含 kind；kind 已存在时不重复
+        # ALTER（与 agent_* 列同款 backfill 惯例）。
+        with tempfile.TemporaryDirectory() as tmp:
+            src = os.path.join(tmp, "src.db")
+            _make_sqlite(src)
+            fake = FakePgConn(pragma_columns={
+                "projects": {"repo_url"},
+                "tasks": {"id", "project_id"},
+                "task_followups": {"id", "task_id", "project_id",
+                                   "body", "mode", "status",
+                                   "created_by", "sender_task_id",
+                                   "created_at", "updated_at"},
+            })
+            result = migrate_sqlite_to_postgres(src, pg_conn=fake)
+        self.assertEqual(result.get("tasks"), 1, "tasks")
+        self.assertTrue(
+            any("tasks" in a and "kind" in a for a in fake.alters),
+            "未发出 tasks.kind 补列 ALTER：%s" % fake.alters)
+        # 注：插入列清单取自源 SQLite 表的列，夹具源表无 kind
+        # 列故清单里无 kind；生产源库（init_db 建表）有 kind，
+        # 迁移时按列原样搬运（与 agent_* 列同款）。
+
+    def test_kind_column_present_skips_alter(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            src = os.path.join(tmp, "src.db")
+            _make_sqlite(src)
+            fake = FakePgConn(pragma_columns={
+                "projects": {"repo_url"},
+                "tasks": {"id", "project_id", "kind"},
+                "task_followups": {"id", "task_id", "project_id",
+                                   "body", "mode", "status",
+                                   "created_by", "sender_task_id",
+                                   "created_at", "updated_at"},
+            })
+            result = migrate_sqlite_to_postgres(src, pg_conn=fake)
+        self.assertEqual(result.get("tasks"), 1, "tasks")
+        self.assertFalse(
+            any("tasks" in a and "kind" in a for a in fake.alters),
+            "kind 已存在却仍发补列 ALTER：%s" % fake.alters)
+
     def test_sender_column_present_skips_alter(self):
         with tempfile.TemporaryDirectory() as tmp:
             src = os.path.join(tmp, "src.db")

cloud/tests/test_sidethread.py

@@ -0,0 +1,730 @@
+"""旁聊任务类型（Muse 角色对齐 P2）单测：创建/列表/执行分流/回写/恢复。
+
+不跑真实模型回合、不联网：服务端侧本地临时端口 + 临时 SQLite；
+worker/sandbox 侧注入假 runner 录命令；host.py dispatch 用假
+server + 临时 HERE。
+"""
+
+import http.client
+import json
+import os
+import sqlite3
+import sys
+import tempfile
+import threading
+import time
+import unittest
+from pathlib import Path
+from types import SimpleNamespace
+from unittest import mock
+
+try:
+    import host as host_mod
+    from cloud.auth import AuthStore
+    from cloud.marker_tools import (
+        count_direct_spawn_children,
+        extract_task_creates,
+        parent_spawn_generation,
+    )
+    from cloud.sandbox import DockerExecutor, LocalExecutor
+    from cloud.server import create_server, init_db, migrate_tasks_table
+    from cloud.worker import Worker
+except ImportError:  # discover -s cloud/tests 时把项目根目录补进 sys.path
+    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
+    import host as host_mod
+    from cloud.auth import AuthStore
+    from cloud.marker_tools import (
+        count_direct_spawn_children,
+        extract_task_creates,
+        parent_spawn_generation,
+    )
+    from cloud.sandbox import DockerExecutor, LocalExecutor
+    from cloud.server import create_server, init_db, migrate_tasks_table
+    from cloud.worker import Worker
+
+
+def _recording(calls):
+    def fake(cmd):
+        calls.append(list(cmd))
+        return SimpleNamespace(returncode=0, stdout="ok", stderr="")
+    return fake
+
+
+def _insert_task(db_path, task_id, **cols):
+    base = {"project_id": "p1", "prompt": "do work", "status": "queued",
+            "created_by": "tester"}
+    base.update(cols)
+    keys = ["id"] + list(base.keys())
+    vals = [task_id] + list(base.values())
+    conn = sqlite3.connect(db_path)
+    try:
+        conn.execute(
+            "INSERT INTO tasks(%s) VALUES(%s)"
+            % (", ".join(keys), ", ".join("?" * len(keys))), vals)
+        conn.commit()
+    finally:
+        conn.close()
+
+
+def _task_row(db_path, task_id, columns="id, status, kind, executor,"
+             " parent_task_id, coordinator_part"):
+    conn = sqlite3.connect(db_path)
+    try:
+        row = conn.execute(
+            "SELECT %s FROM tasks WHERE id = ?" % columns,
+            (task_id,)).fetchone()
+    finally:
+        conn.close()
+    return row
+
+
+def _followups(db_path, task_id):
+    conn = sqlite3.connect(db_path)
+    try:
+        rows = conn.execute(
+            "SELECT id, body, mode, status, sender_task_id"
+            " FROM task_followups WHERE task_id = ?"
+            " ORDER BY created_at, id", (task_id,)).fetchall()
+    finally:
+        conn.close()
+    return [{"id": r[0], "body": r[1], "mode": r[2], "status": r[3],
+             "sender_task_id": r[4]} for r in rows]
+
+
+def _set_status(db_path, task_id, status):
+    conn = sqlite3.connect(db_path)
+    try:
+        conn.execute("UPDATE tasks SET status = ? WHERE id = ?",
+                     (status, task_id))
+        conn.commit()
+    finally:
+        conn.close()
+
+
+class _ServerBase(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.db_path = os.path.join(self.tmp.name, "control.db")
+        self.auth = AuthStore(self.db_path)
+        self.owner_id, self.owner_key = self.auth.create_user("owner")
+        self.server = create_server(self.db_path, "127.0.0.1", 0)
+        self.port = self.server.server_address[1]
+        self.thread = threading.Thread(
+            target=self.server.serve_forever, daemon=True)
+        self.thread.start()
+        status, self.project = self._req(
+            "POST", "/projects", {"name": "demo"}, self.owner_key)
+        self.assertEqual(status, 201)
+
+    def tearDown(self):
+        self.server.shutdown()
+        self.server.server_close()
+        self.thread.join(timeout=5)
+        try:
+            self.auth.conn.close()
+        except Exception:
+            pass
+        self.tmp.cleanup()
+
+    def _req(self, method, path, body=None, key=None):
+        conn = http.client.HTTPConnection(
+            "127.0.0.1", self.port, timeout=10)
+        hdrs = {}
+        payload = None
+        if body is not None:
+            payload = json.dumps(body).encode("utf-8")
+            hdrs["Content-Type"] = "application/json"
+        if key is not None:
+            hdrs["X-API-Key"] = key
+        try:
+            conn.request(method, path, body=payload, headers=hdrs)
+            resp = conn.getresponse()
+            data = resp.read()
+        finally:
+            conn.close()
+        try:
+            parsed = json.loads(data.decode("utf-8")) if data else None
+        except ValueError:
+            parsed = None
+        return resp.status, parsed
+
+    def _create_task(self, extra=None):
+        body = {"project_id": self.project["id"], "prompt": "做点事"}
+        if extra:
+            body.update(extra)
+        return self._req("POST", "/tasks", body, self.owner_key)
+
+    def _create_side(self, parent_id, extra=None):
+        body = {"project_id": self.project["id"], "prompt": "旁聊一下",
+                "kind": "side", "parent_task_id": parent_id}
+        if extra:
+            body.update(extra)
+        return self._req("POST", "/tasks", body, self.owner_key)
+
+
+class ServerSideThreadTestCase(_ServerBase):
+    def test_create_side_ok_defaults_local(self):
+        status, parent = self._create_task()
+        self.assertEqual(status, 201)
+        status, side = self._create_side(parent["id"])
+        self.assertEqual(status, 201, side)
+        self.assertEqual(side["kind"], "side")
+        self.assertEqual(side["parent_task_id"], parent["id"])
+        self.assertEqual(side["executor"], "local")
+        status, fetched = self._req(
+            "GET", "/tasks/" + side["id"], key=self.owner_key)
+        self.assertEqual(status, 200)
+        self.assertEqual(fetched["kind"], "side")
+
+    def test_create_side_validation(self):
+        status, parent = self._create_task()
+        self.assertEqual(status, 201)
+        # 缺 parent_task_id
+        status, body = self._req(
+            "POST", "/tasks",
+            {"project_id": self.project["id"], "prompt": "x",
+             "kind": "side"}, self.owner_key)
+        self.assertEqual(status, 400)
+        # 父不存在
+        status, _body = self._create_side("no-such-parent")
+        self.assertEqual(status, 404)
+        # kind 非法值
+        status, _body = self._create_task(
+            {"kind": "weird", "parent_task_id": parent["id"]})
+        self.assertEqual(status, 400)
+        # 带 coordinator_parts
+        status, _body = self._create_side(parent["id"], {
+            "coordinator_parts": [{"name": "a", "prompt": "x"}]})
+        self.assertEqual(status, 400)
+        # 带 upload_id / environment_id
+        status, _body = self._create_side(
+            parent["id"], {"upload_id": "up-nonexistent"})
+        self.assertEqual(status, 400)
+        status, _body = self._create_side(
+            parent["id"], {"environment_id": "env-nonexistent"})
+        self.assertEqual(status, 400)
+
+    def test_create_side_cross_project_rejected(self):
+        status, parent = self._create_task()
+        self.assertEqual(status, 201)
+        status, other_project = self._req(
+            "POST", "/projects", {"name": "other"}, self.owner_key)
+        self.assertEqual(status, 201)
+        status, other_task = self._req(
+            "POST", "/tasks",
+            {"project_id": other_project["id"], "prompt": "别处的事"},
+            self.owner_key)
+        self.assertEqual(status, 201)
+        status, _body = self._create_side(other_task["id"])
+        self.assertEqual(status, 400)
+
+    def test_create_side_under_coordinator_rejected(self):
+        status, coord = self._req(
+            "POST", "/tasks",
+            {"project_id": self.project["id"], "prompt": "协调干活",
+             "coordinator_parts": [{"name": "a", "prompt": "做 a"}]},
+            self.owner_key)
+        self.assertEqual(status, 201)
+        status, body = self._create_side(coord["id"])
+        self.assertEqual(status, 400)
+        self.assertIn("coordinator", body["error"])
+
+    def test_create_side_explicit_docker_kept(self):
+        status, parent = self._create_task()
+        self.assertEqual(status, 201)
+        status, side = self._create_side(
+            parent["id"], {"executor": "docker"})
+        self.assertEqual(status, 201, side)
+        self.assertEqual(side["executor"], "docker")
+        self.assertEqual(side["kind"], "side")
+
+    def test_create_main_defaults(self):
+        status, task = self._create_task()
+        self.assertEqual(status, 201)
+        self.assertEqual(task["kind"], "main")
+        status, task2 = self._create_task({"kind": "main"})
+        self.assertEqual(status, 201)
+        self.assertEqual(task2["kind"], "main")
+
+    def test_list_kind_filter(self):
+        status, parent = self._create_task()
+        self.assertEqual(status, 201)
+        _s, side1 = self._create_side(parent["id"])
+        _s, side2 = self._create_side(parent["id"])
+        _s, main_child = self._create_task(
+            {"parent_task_id": parent["id"]})
+        self.assertEqual(main_child["kind"], "main")
+        pid = self.project["id"]
+        status, data = self._req(
+            "GET", "/tasks?project_id=" + pid +
+            "&parent_task_id=" + parent["id"] + "&kind=side",
+            key=self.owner_key)
+        self.assertEqual(status, 200)
+        self.assertEqual(
+            sorted(t["id"] for t in data["tasks"]),
+            sorted([side1["id"], side2["id"]]))
+        status, data = self._req(
+            "GET", "/tasks?project_id=" + pid +
+            "&parent_task_id=" + parent["id"], key=self.owner_key)
+        self.assertEqual(status, 200)
+        self.assertEqual(len(data["tasks"]), 3)
+        status, data = self._req(
+            "GET", "/tasks?project_id=" + pid + "&kind=side",
+            key=self.owner_key)
+        self.assertEqual(status, 200)
+        self.assertEqual(
+            sorted(t["id"] for t in data["tasks"]),
+            sorted([side1["id"], side2["id"]]))
+        status, data = self._req(
+            "GET", "/tasks?project_id=" + pid + "&kind=main",
+            key=self.owner_key)
+        self.assertEqual(status, 200)
+        self.assertEqual(
+            sorted(t["id"] for t in data["tasks"]),
+            sorted([parent["id"], main_child["id"]]))
+        status, _body = self._req(
+            "GET", "/tasks?project_id=" + pid + "&kind=bogus",
+            key=self.owner_key)
+        self.assertEqual(status, 400)
+
+    def test_side_files_tree_worktree_404(self):
+        status, parent = self._create_task()
+        self.assertEqual(status, 201)
+        status, side = self._create_side(parent["id"])
+        self.assertEqual(status, 201)
+        for suffix in ("/files", "/tree", "/worktree"):
+            status, body = self._req(
+                "GET", "/tasks/" + side["id"] + suffix,
+                key=self.owner_key)
+            self.assertEqual(status, 404, suffix)
+            self.assertIn("no local worktree", body["error"])
+
+    def test_side_fork_merge_pr_409(self):
+        status, parent = self._create_task()
+        self.assertEqual(status, 201)
+        status, side = self._create_side(parent["id"])
+        self.assertEqual(status, 201)
+        status, body = self._req(
+            "POST", "/tasks/" + side["id"] + "/fork",
+            {"message": "分一个"}, self.owner_key)
+        self.assertEqual(status, 409)
+        self.assertIn("side thread", body["error"])
+        status, body = self._req(
+            "POST", "/tasks/" + side["id"] + "/merge", {},
+            self.owner_key)
+        self.assertEqual(status, 409)
+        self.assertIn("side thread", body["error"])
+        status, body = self._req(
+            "POST", "/tasks/" + side["id"] + "/pr", {},
+            self.owner_key)
+        self.assertEqual(status, 409)
+        self.assertIn("side thread", body["error"])
+
+    def test_side_continue_and_followups(self):
+        status, parent = self._create_task()
+        self.assertEqual(status, 201)
+        status, side = self._create_side(parent["id"])
+        self.assertEqual(status, 201)
+        _set_status(self.db_path, side["id"], "completed")
+        status, task = self._req(
+            "POST", "/tasks/" + side["id"] + "/continue",
+            {"message": "再聊聊"}, self.owner_key)
+        self.assertEqual(status, 200, task)
+        self.assertEqual(task["status"], "queued")
+        self.assertEqual(task["resume_message"], "再聊聊")
+        status, _body = self._req(
+            "POST", "/tasks/" + side["id"] + "/followups",
+            {"body": "继续", "mode": "queued"}, self.owner_key)
+        self.assertEqual(status, 201)
+
+    def test_side_writeback_visible_on_terminal_parent(self):
+        # 父终态时回写只作记录：followups 可见，父不复活。
+        status, parent = self._create_task()
+        self.assertEqual(status, 201)
+        status, side = self._create_side(parent["id"])
+        self.assertEqual(status, 201)
+        _set_status(self.db_path, parent["id"], "completed")
+        conn = sqlite3.connect(self.db_path)
+        try:
+            conn.execute(
+                "INSERT INTO task_followups(id, task_id, project_id,"
+                " body, mode, status, created_by, sender_task_id,"
+                " created_at, updated_at)"
+                " VALUES(?, ?, ?, ?, 'queued', 'pending', ?, ?, ?, ?)",
+                ("fu-1", parent["id"], self.project["id"],
+                 "[旁聊结论] 查到了", "tester", side["id"],
+                 time.time(), time.time()))
+            conn.commit()
+        finally:
+            conn.close()
+        status, data = self._req(
+            "GET", "/tasks/" + parent["id"] + "/followups",
+            key=self.owner_key)
+        self.assertEqual(status, 200)
+        self.assertEqual(len(data["followups"]), 1)
+        self.assertEqual(data["followups"][0]["sender_task_id"],
+                         side["id"])
+        status, fetched = self._req(
+            "GET", "/tasks/" + parent["id"], key=self.owner_key)
+        self.assertEqual(status, 200)
+        self.assertEqual(fetched["status"], "completed")
+
+
+class _WorkerBase(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.db_path = os.path.join(self.tmp.name, "control.db")
+        init_db(self.db_path)
+        self.env = {"XCM_API_KEY": "test-placeholder"}
+
+    def tearDown(self):
+        self.tmp.cleanup()
+
+    def _worker(self, calls):
+        return Worker(self.db_path, "/fake/host.py",
+                      runner=_recording(calls), env=self.env)
+
+
+class WorkerSideThreadTestCase(_WorkerBase):
+    def test_run_one_side_no_repo_no_worktree_and_writeback(self):
+        _insert_task(self.db_path, "parent1", executor="local")
+        _insert_task(self.db_path, "side1", kind="side",
+                     parent_task_id="parent1", executor="local")
+        calls = []
+        worker = self._worker(calls)
+        result = worker.run_one(worker.get_task("side1"))
+        self.assertEqual(result["status"], "completed")
+        dispatch_cmds = [c for c in calls if "dispatch" in c]
+        self.assertEqual(len(dispatch_cmds), 1)
+        cmd = dispatch_cmds[0]
+        self.assertIn("--no-worktree", cmd)
+        self.assertNotIn("--repo", cmd)
+        self.assertNotIn("--seed-dir", cmd)
+        row = _task_row(self.db_path, "side1")
+        self.assertEqual(row[1], "completed")
+        fus = _followups(self.db_path, "parent1")
+        self.assertEqual(len(fus), 1)
+        self.assertEqual(fus[0]["sender_task_id"], "side1")
+        self.assertEqual(fus[0]["mode"], "queued")
+        self.assertEqual(fus[0]["status"], "pending")
+        self.assertTrue(fus[0]["body"].startswith("[旁聊结论]"))
+
+    def test_run_attempt_side_with_repo_forces_plan(self):
+        # worker 层直接调 _run_attempt：有 repo 的旁聊强制 plan=1，
+        # argv 同时含 --no-worktree 与 --sandbox read-only。
+        calls = []
+        worker = self._worker(calls)
+        task = {"id": "side-repo", "prompt": "聊聊 repo",
+                "project_id": "p1", "created_by": "tester",
+                "kind": "side", "executor": "local",
+                "repo": "/tmp/fake-repo",
+                "_gateway_provider": None, "_gateway_env": {}}
+        result, error = worker._run_attempt(task)
+        self.assertIsNone(error)
+        self.assertEqual(result["status"], "completed")
+        dispatch_cmds = [c for c in calls if "dispatch" in c]
+        self.assertEqual(len(dispatch_cmds), 1)
+        cmd = dispatch_cmds[0]
+        self.assertIn("--no-worktree", cmd)
+        self.assertIn("--sandbox", cmd)
+        self.assertEqual(cmd[cmd.index("--sandbox") + 1], "read-only")
+        self.assertIn("--repo", cmd)
+        self.assertEqual(cmd[cmd.index("--repo") + 1], "/tmp/fake-repo")
+
+    def test_run_attempt_main_task_untouched(self):
+        calls = []
+        worker = self._worker(calls)
+        task = {"id": "main1", "prompt": "做事",
+                "project_id": "p1", "created_by": "tester",
+                "executor": "local",
+                "_gateway_provider": None, "_gateway_env": {}}
+        result, error = worker._run_attempt(task)
+        self.assertIsNone(error)
+        self.assertEqual(result["status"], "completed")
+        dispatch_cmds = [c for c in calls if "dispatch" in c]
+        self.assertEqual(len(dispatch_cmds), 1)
+        self.assertNotIn("--no-worktree", dispatch_cmds[0])
+        self.assertNotIn("--sandbox", dispatch_cmds[0])
+
+    def test_writeback_direct_shapes(self):
+        _insert_task(self.db_path, "parent1", executor="local")
+        worker = Worker(self.db_path, "/fake/host.py", env=self.env)
+        task = {"id": "side1", "project_id": "p1",
+                "created_by": "tester", "kind": "side",
+                "parent_task_id": "parent1"}
+        self.assertTrue(worker._writeback_side_thread(
+            task, "completed", "结论文本"))
+        self.assertTrue(worker._writeback_side_thread(
+            task, "failed", "出错了"))
+        self.assertTrue(worker._writeback_side_thread(
+            task, "interrupted", ""))
+        fus = _followups(self.db_path, "parent1")
+        self.assertEqual(len(fus), 3)
+        self.assertTrue(fus[0]["body"].startswith("[旁聊结论]"))
+        self.assertTrue(fus[1]["body"].startswith("[旁聊失败]"))
+        self.assertTrue(fus[2]["body"].startswith("[旁聊中断]"))
+        # 超长正文截断到 2000 字 + 截断标记。
+        self.assertTrue(worker._writeback_side_thread(
+            task, "completed", "x" * 3000))
+        fus = _followups(self.db_path, "parent1")
+        self.assertLessEqual(len(fus[3]["body"]), 2010)
+        self.assertIn("已截断", fus[3]["body"])
+        # 无父任务不写。
+        orphan = dict(task, parent_task_id=None)
+        self.assertFalse(worker._writeback_side_thread(
+            orphan, "completed", "x"))
+
+    def test_parent_running_drain_consumes_writeback(self):
+        _insert_task(self.db_path, "parent1", executor="local")
+        _insert_task(self.db_path, "side1", kind="side",
+                     parent_task_id="parent1", executor="local")
+        worker = Worker(self.db_path, "/fake/host.py", env=self.env)
+        self.assertTrue(worker._writeback_side_thread(
+            {"id": "side1", "project_id": "p1", "created_by": "tester",
+             "kind": "side", "parent_task_id": "parent1"},
+            "completed", "结论文本"))
+        calls = []
+        worker = self._worker(calls)
+        result = worker.run_one(worker.get_task("parent1"))
+        self.assertEqual(result["status"], "completed")
+        # 首轮 + 排空一轮 = 两次 dispatch。
+        dispatch_cmds = [c for c in calls if "dispatch" in c]
+        self.assertEqual(len(dispatch_cmds), 2)
+        fus = _followups(self.db_path, "parent1")
+        self.assertEqual(len(fus), 1)
+        self.assertEqual(fus[0]["status"], "done")
+
+
+class MarkerSideThreadTestCase(_WorkerBase):
+    def _deliver(self, parent_id, creates):
+        worker = Worker(self.db_path, "/fake/host.py", env=self.env)
+        worker._deliver_marker_tools(worker.get_task(parent_id), {
+            "status": "completed", "task_creates": creates,
+            "task_polls": [], "wait_threads": [], "task_cancels": [],
+            "task_retries": []})
+
+    def test_extract_kind(self):
+        text = (
+            "```task-create\n"
+            "{\"prompt\": \"做子事\", \"kind\": \"side\"}\n```\n"
+            "```task-create\n"
+            "{\"prompt\": \"做子事\", \"kind\": \"weird\"}\n```\n"
+            "```task-create\n{\"prompt\": \"做子事\"}\n```\n")
+        creates = extract_task_creates(text)
+        self.assertEqual(len(creates), 2)
+        self.assertEqual(creates[0]["kind"], "side")
+        self.assertIsNone(creates[1]["kind"])
+
+    def test_marker_child_inherits_side(self):
+        _insert_task(self.db_path, "side-parent", kind="side",
+                     executor="local")
+        self._deliver("side-parent", [{
+            "name": "子旁聊", "prompt": "继续聊",
+            "model": None, "effort": None, "executor": None,
+            "kind": None}])
+        conn = sqlite3.connect(self.db_path)
+        try:
+            child = conn.execute(
+                "SELECT kind, executor FROM tasks"
+                " WHERE parent_task_id = ?",
+                ("side-parent",)).fetchone()
+        finally:
+            conn.close()
+        self.assertIsNotNone(child)
+        self.assertEqual(child[0], "side")
+        self.assertEqual(child[1], "local")
+
+    def test_marker_child_explicit_kind_wins(self):
+        _insert_task(self.db_path, "side-parent", kind="side",
+                     executor="local")
+        self._deliver("side-parent", [{
+            "name": "转主聊", "prompt": "正经做",
+            "model": None, "effort": None, "executor": None,
+            "kind": "main"}])
+        conn = sqlite3.connect(self.db_path)
+        try:
+            child = conn.execute(
+                "SELECT kind FROM tasks WHERE parent_task_id = ?",
+                ("side-parent",)).fetchone()
+        finally:
+            conn.close()
+        self.assertIsNotNone(child)
+        self.assertEqual(child[0], "main")
+
+    def test_marker_counts_exclude_side(self):
+        _insert_task(self.db_path, "main-parent", executor="local")
+        _insert_task(self.db_path, "side-kid", kind="side",
+                     parent_task_id="main-parent", executor="local")
+        _insert_task(self.db_path, "main-kid",
+                     parent_task_id="main-parent", executor="local")
+        conn = sqlite3.connect(self.db_path)
+        conn.row_factory = sqlite3.Row
+        try:
+            self.assertEqual(
+                count_direct_spawn_children(conn, "main-parent"), 1)
+            self.assertEqual(
+                parent_spawn_generation(conn, "side-kid"), 0)
+            self.assertEqual(
+                parent_spawn_generation(conn, "main-kid"), 1)
+        finally:
+            conn.close()
+
+
+class MigrateSideThreadTestCase(unittest.TestCase):
+    def test_migrate_adds_kind_idempotent(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            db_path = os.path.join(tmp, "old.db")
+            conn = sqlite3.connect(db_path)
+            try:
+                conn.execute(
+                    "CREATE TABLE tasks(id TEXT PRIMARY KEY,"
+                    " project_id TEXT, prompt TEXT, status TEXT)")
+                conn.commit()
+                migrate_tasks_table(conn)
+                cols = {r[1] for r in conn.execute(
+                    "PRAGMA table_info(tasks)").fetchall()}
+                self.assertIn("kind", cols)
+                # 第二次是 no-op，不炸、不重复。
+                migrate_tasks_table(conn)
+                cols2 = {r[1] for r in conn.execute(
+                    "PRAGMA table_info(tasks)").fetchall()}
+                self.assertEqual(cols, cols2)
+            finally:
+                conn.close()
+
+
+class SandboxSideThreadTestCase(unittest.TestCase):
+    def test_local_dispatch_no_worktree_flag(self):
+        calls = []
+        ex = LocalExecutor("/fake/host.py", runner=_recording(calls))
+        ex.dispatch_task({"id": "t1", "prompt": "hi"},
+                         no_worktree=True)
+        self.assertIn("--no-worktree", calls[0])
+        calls.clear()
+        ex.dispatch_task({"id": "t1", "prompt": "hi"})
+        self.assertNotIn("--no-worktree", calls[0])
+
+    def test_local_goal_no_worktree_flag(self):
+        calls = []
+        ex = LocalExecutor("/fake/host.py", runner=_recording(calls))
+        ex.goal_task({"id": "t1", "prompt": "目标"}, no_worktree=True)
+        self.assertIn("--no-worktree", calls[0])
+        self.assertIn("goal", calls[0])
+
+    def test_local_plan_passthrough_readonly(self):
+        calls = []
+        ex = LocalExecutor("/fake/host.py", runner=_recording(calls))
+        ex.dispatch_task({"id": "t1", "prompt": "hi", "plan": 1})
+        self.assertIn("--sandbox", calls[0])
+        self.assertEqual(calls[0][calls[0].index("--sandbox") + 1],
+                         "read-only")
+        calls.clear()
+        ex.dispatch_task({"id": "t1", "prompt": "hi"})
+        self.assertNotIn("--sandbox", calls[0])
+
+    def test_local_popen_no_worktree_flag(self):
+        calls = []
+
+        class FakePopen(object):
+            def __init__(self, cmd, **kwargs):
+                calls.append(list(cmd))
+                self.returncode = 0
+
+            def poll(self):
+                return 0
+
+            def communicate(self, timeout=None):
+                return "", ""
+
+        ex = LocalExecutor("/fake/host.py", runner=_recording([]))
+        import cloud.sandbox as sandbox_mod
+        with mock.patch.object(sandbox_mod.subprocess, "Popen",
+                               FakePopen):
+            ex.popen_task({"id": "t1", "prompt": "hi"},
+                          kind="dispatch", no_worktree=True)
+            ex.popen_task({"id": "t1", "prompt": "目标"},
+                          kind="goal", no_worktree=True)
+        dispatch_cmds = [c for c in calls if "dispatch" in c]
+        goal_cmds = [c for c in calls if "goal" in c]
+        self.assertEqual(len(dispatch_cmds), 1)
+        self.assertEqual(len(goal_cmds), 1)
+        self.assertIn("--no-worktree", dispatch_cmds[0])
+        self.assertIn("--no-worktree", goal_cmds[0])
+
+    def test_docker_dispatch_no_worktree_flag(self):
+        calls = []
+        ex = DockerExecutor(
+            "/fake/host.py", runner=_recording(calls),
+            env={"XCM_API_KEY": "placeholder-not-a-real-key"})
+        with mock.patch("cloud.sandbox.shutil.which",
+                        return_value="/usr/bin/docker"):
+            ex.dispatch_task({"id": "t1", "prompt": "hi"},
+                             no_worktree=True)
+        self.assertIn("--no-worktree", calls[0])
+        calls.clear()
+        with mock.patch("cloud.sandbox.shutil.which",
+                        return_value="/usr/bin/docker"):
+            ex.dispatch_task({"id": "t1", "prompt": "hi"})
+        self.assertNotIn("--no-worktree", calls[0])
+
+    def test_docker_goal_no_worktree_flag(self):
+        calls = []
+        ex = DockerExecutor(
+            "/fake/host.py", runner=_recording(calls),
+            env={"XCM_API_KEY": "placeholder-not-a-real-key"})
+        with mock.patch("cloud.sandbox.shutil.which",
+                        return_value="/usr/bin/docker"):
+            ex.goal_task({"id": "t1", "prompt": "目标"},
+                         no_worktree=True)
+        self.assertIn("--no-worktree", calls[0])
+
+
+class _FakeHostServer(object):
+    def __init__(self):
+        self.created = []
+
+    def create_thread(self, cwd, model=None, sandbox="workspace-write",
+                      approval_policy="never", instructions=None):
+        self.created.append((cwd, sandbox))
+        return "thread-1"
+
+
+class HostSideThreadTestCase(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self._old_here = host_mod.HERE
+        host_mod.HERE = self.tmp.name
+
+    def tearDown(self):
+        host_mod.HERE = self._old_here
+        self.tmp.cleanup()
+
+    def test_dispatch_no_worktree_no_repo_uses_scratch(self):
+        srv = _FakeHostServer()
+        task = host_mod.dispatch(
+            srv, {"tasks": {}}, "side-abc", "聊聊", None, None, None,
+            None, "never", use_worktree=False)
+        expected = os.path.join(self.tmp.name, "work", "side-threads",
+                                "side-abc")
+        self.assertEqual(task["cwd"], expected)
+        self.assertTrue(os.path.isdir(expected))
+        self.assertFalse(os.path.exists(os.path.join(expected, ".git")))
+        self.assertIsNone(task["branch"])
+        self.assertEqual(srv.created[0][0], expected)
+        # 没有建 host-worktrees 目录。
+        self.assertFalse(os.path.exists(
+            os.path.join(self.tmp.name, "work", "host-worktrees")))
+
+    def test_dispatch_no_worktree_with_repo_uses_repo(self):
+        srv = _FakeHostServer()
+        task = host_mod.dispatch(
+            srv, {"tasks": {}}, "side-def", "聊聊", "/tmp/some-repo",
+            None, None, None, "never", use_worktree=False)
+        self.assertEqual(task["cwd"], "/tmp/some-repo")
+        self.assertIsNone(task["branch"])
+        self.assertFalse(os.path.exists(
+            os.path.join(self.tmp.name, "work", "side-threads")))
+
+
+if __name__ == "__main__":
+    unittest.main()