"""旁聊任务类型（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()