"""线程等待与移交操作单测：cloud/wait_ops 纯逻辑 + server 端点。 不跑真实模型、不联网、不调真实 gh：thread-read 用 stub host.py（与 test_thread_ops.py 同款契约），pr checks 用 注入的 fake gh runner，worktree 状态用临时目录。覆盖鉴权、 权限、校验、状态机（移交转移、终态不可动）、超时语义与 诚实降级（异机 worktree / Docker 任务的 thread-read 仍回 200，只 thread 为空）。 """ import http.client import json import os import sqlite3 import sys import tempfile import threading import unittest from pathlib import Path from types import SimpleNamespace try: from cloud.auth import AuthStore from cloud.server import create_server from cloud.wait_ops import (build_pr_diagnostics, create_handoff, decode_wait_cursor, detect_significant_change, encode_wait_cursor, interruptible_sleep, latest_handoff, parse_pr_checks, poll_for_first_change, poll_task_statuses, report_handoff, worktree_status) except ImportError: sys.path.insert(0, str(Path(__file__).resolve().parents[2])) from cloud.auth import AuthStore from cloud.server import create_server from cloud.wait_ops import (build_pr_diagnostics, create_handoff, decode_wait_cursor, detect_significant_change, encode_wait_cursor, interruptible_sleep, latest_handoff, parse_pr_checks, poll_for_first_change, poll_task_statuses, report_handoff, worktree_status) STUB = r'''import json import os import sys def main(): argv = sys.argv[1:] cmd = None rest = [] i = 0 while i < len(argv): a = argv[i] if a == "--codex-home": i += 2 continue if a.startswith("--codex-home="): i += 1 continue if cmd is None: cmd = a i += 1 continue rest.append(a) i += 1 if cmd == "thread-list": print(json.dumps({"data": [ {"id": "t-old", "updatedAt": 100, "createdAt": 90}, {"id": "t-new", "updatedAt": 200, "createdAt": 190}, ]})) elif cmd == "thread-read": print(json.dumps({"thread": {"id": "t-new", "updatedAt": 200}})) else: print(json.dumps({"error": "unknown subcommand"})) sys.exit(1) main() ''' class WaitOpsPureTestCase(unittest.TestCase): def test_parse_pr_checks_tabular_and_human(self): checks, all_pass, failed, pending = parse_pr_checks( "build\tpass\ntest\tfail\nlint\tpending\n") self.assertEqual([c["state"] for c in checks], ["pass", "fail", "pending"]) self.assertFalse(all_pass) self.assertEqual(failed, ["test"]) self.assertEqual(pending, ["lint"]) checks, all_pass, _, _ = parse_pr_checks( "\u2713 build pass\n\u2713 test pass\n") self.assertTrue(all_pass) checks, all_pass, _, _ = parse_pr_checks("no checks reported") self.assertEqual(checks, []) self.assertFalse(all_pass) def test_worktree_status_matrix(self): self.assertEqual( worktree_status("running", "/tmp/x", False, False)["status"], "creating") self.assertEqual( worktree_status("completed", "/tmp/x", False, False)["status"], "missing") self.assertEqual( worktree_status("running", "/tmp/x", True, True)["status"], "ready") self.assertEqual( worktree_status("running", "/tmp/x", True, False)["status"], "present_no_git") remote = worktree_status("running", "/tmp/x", False, False, remote_host="host-b") self.assertEqual(remote["status"], "remote") self.assertIsNone(remote["exists"]) def test_handoff_transitions(self): conn = sqlite3.connect(":memory:") conn.row_factory = sqlite3.Row h = create_handoff(conn, "t1", "p1", "host-b", "u1", source_host="host-a") self.assertEqual(h["status"], "pending") with self.assertRaises(ValueError): create_handoff(conn, "t1", "p1", "host-c", "u1") ok, h = report_handoff(conn, h["id"], "running", progress=40) self.assertTrue(ok) self.assertEqual((h["status"], h["progress"]), ("running", 40)) ok, h = report_handoff(conn, h["id"], "completed") self.assertTrue(ok) ok, err = report_handoff(conn, h["id"], "running") self.assertFalse(ok) self.assertIn("cannot move", err) ok, err = report_handoff(conn, h["id"], "bogus") self.assertFalse(ok) self.assertEqual(latest_handoff(conn, "t1")["status"], "completed") conn.close() def test_wait_cursor_and_first_change(self): base = {"a": {"status": "running", "pending_approvals": 0, "pending_input": 0, "pending_followups": 0}, "b": {"status": "running", "pending_approvals": 0, "pending_input": 0, "pending_followups": 0}} cursor = encode_wait_cursor(base) self.assertEqual(decode_wait_cursor(cursor), base) with self.assertRaises(ValueError): decode_wait_cursor("not-a-cursor!!") cur = dict(base) cur["b"] = {"status": "running", "pending_approvals": 0, "pending_input": 1, "pending_followups": 0} found = detect_significant_change(base, cur, ["a", "b"]) self.assertEqual(found[0], "b") self.assertEqual(found[1], "needs_input") cur2 = dict(base) cur2["a"] = {"status": "failed", "pending_approvals": 0, "pending_input": 0, "pending_followups": 0} found = detect_significant_change(base, cur2, ["a", "b"]) self.assertEqual(found[1], "terminal") # 基线里没有的任务不误报；无游标时已终态立即算变化。 found = detect_significant_change( base, {"c": {"status": "completed", "pending_approvals": 0, "pending_input": 0, "pending_followups": 0}}, ["c"]) self.assertIsNone(found) found = detect_significant_change( None, {"c": {"status": "completed", "pending_approvals": 0, "pending_input": 0, "pending_followups": 0}}, ["c"]) self.assertEqual(found[1], "terminal") changed, summaries, _w, timed_out = poll_for_first_change( lambda: {"a": {"status": "completed", "pending_approvals": 0, "pending_input": 0, "pending_followups": 0}}, ["a"], baseline=None, timeout_seconds=5) self.assertFalse(timed_out) self.assertEqual(changed["kind"], "terminal") def test_build_pr_diagnostics_json_and_check_runs(self): def run(cmd): if "--json" in cmd: return (0, json.dumps([ {"name": "build", "bucket": "pass", "workflow": "CI", "link": "https://example.invalid/ci/1", "description": "ok"}, {"name": "test", "bucket": "fail", "workflow": "CI", "link": "https://example.invalid/ci/2", "description": "1 failed"}]), "") if "view" in cmd: return (0, "abc123\n", "") if any("commits" in c for c in cmd): return (0, json.dumps({"check_runs": [ {"id": 99, "name": "test", "conclusion": "failure"}]}), "") if any("check-runs" in c for c in cmd): return (0, json.dumps({ "id": 99, "conclusion": "failure", "output": { "title": "Test failed", "summary": "1 test failed", "text": "FAILED test_x: assert failed", "annotations": [{ "path": "src/a.py", "start_line": 12, "annotation_level": "failure", "message": "assert failed"}]}}), "") return (1, "", "unknown cmd") diag = build_pr_diagnostics(run, 42, owner_repo="o/r", head_sha="abc123") self.assertFalse(diag["all_pass"]) self.assertEqual(diag["failed"], ["test"]) self.assertEqual(diag["checks"][1]["workflow"], "CI") self.assertTrue(diag["check_runs"]["available"]) run_detail = diag["check_runs"]["runs"][0] self.assertIn("FAILED test_x", run_detail["log_text"]) self.assertEqual(run_detail["annotations"][0]["path"], "src/a.py") self.assertEqual(run_detail["annotations"][0]["start_line"], 12) # 制表符回落：--json 失败时仍能解析，且无仓库时不查 API。 diag2 = build_pr_diagnostics( lambda cmd: (1, "", "boom") if "--json" in cmd else (0, "build\tpass\ntest\tfail\n", ""), 7) self.assertEqual(diag2["failed"], ["test"]) self.assertFalse(diag2["check_runs"]["available"]) def test_poll_and_sleep(self): calls = {"n": 0} def fetch(): calls["n"] += 1 return {"a": "running"} if calls["n"] < 3 else {"a": "completed"} ok, statuses, waited, timed_out = poll_task_statuses( fetch, timeout_seconds=5) self.assertTrue(ok) self.assertFalse(timed_out) self.assertEqual(statuses, {"a": "completed"}) ok, _, _, timed_out = poll_task_statuses( lambda: {"a": "running"}, timeout_seconds=0) self.assertFalse(ok) self.assertTrue(timed_out) ok, _, _, _ = poll_task_statuses(lambda: {}, timeout_seconds=5) self.assertTrue(ok) slept, interrupted = interruptible_sleep(0) self.assertEqual((slept, interrupted), (0.0, False)) class ThreadWaitServerTestCase(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.reader_id, self.reader_key = self.auth.create_user("reader") self.outsider_id, self.outsider_key = self.auth.create_user( "outsider") self.stub_path = os.path.join(self.tmp.name, "stub_host.py") with open(self.stub_path, "w", encoding="utf-8") as f: f.write(STUB) self._old_host_py = os.environ.get("CODEX_HOST_PY") os.environ["CODEX_HOST_PY"] = self.stub_path self.server = create_server(self.db_path, "127.0.0.1", 0) self.server.tenants_root = os.path.join(self.tmp.name, "tenants") self.server.host_worktrees_dir = os.path.join( self.tmp.name, "worktrees") 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, self.project) self.auth.set_member(self.project["id"], self.owner_id, self.reader_id, "viewer") status, self.project2 = self._req( "POST", "/projects", {"name": "other"}, self.owner_key) self.assertEqual(status, 201, self.project2) def tearDown(self): self.server.shutdown() self.server.server_close() self.thread.join(timeout=5) if self._old_host_py is None: os.environ.pop("CODEX_HOST_PY", None) else: os.environ["CODEX_HOST_PY"] = self._old_host_py 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, project_id=None, **extra): body = {"project_id": project_id or self.project["id"], "prompt": "做事"} body.update(extra) status, task = self._req("POST", "/tasks", body, self.owner_key) self.assertEqual(status, 201, task) return task def _set_task(self, task_id, **fields): conn = sqlite3.connect(self.db_path) try: for key, value in fields.items(): conn.execute( "UPDATE tasks SET %s = ? WHERE id = ?" % key, (value, task_id)) conn.commit() finally: conn.close() def test_thread_read_summary_and_docker_downgrade(self): # 默认执行器是 docker（线程守卫会挡），读摘要的 thread # 部分必须用显式 local 任务测；docker 任务测降级分支。 task = self._create_task(executor="local") self._set_task(task["id"], status="running") status, body = self._req( "GET", "/tasks/%s/thread/read" % task["id"], None, self.reader_key) self.assertEqual(status, 200, body) self.assertEqual(body["status"], "running") self.assertEqual(body["state"], "running") self.assertEqual(body["thread_id"], "t-new") self.assertEqual(body["thread"]["id"], "t-new") self.assertEqual(body["progress"]["trace_entries"], 0) self.assertIsNone(body["result"]["text"]) self.assertEqual(body["changed_files"], []) self.assertEqual(body["commands"], []) self.assertEqual(body["recent_activity"], []) self.assertFalse(body["attention"]["blocked"]) self.assertFalse(body["attention"]["needs_instruction"]) self.assertIsNone(body["test_results"]) docker = self._create_task() status, body = self._req( "GET", "/tasks/%s/thread/read" % docker["id"], None, self.owner_key) self.assertEqual(status, 200, body) self.assertIsNone(body["thread"]) self.assertIn("docker", body["note"]) status, _ = self._req( "GET", "/tasks/%s/thread/read" % task["id"], None, self.outsider_key) self.assertEqual(status, 403) status, _ = self._req( "GET", "/tasks/nope/thread/read", None, self.owner_key) self.assertEqual(status, 404) def test_thread_message_followup_semantics(self): task = self._create_task() self._set_task(task["id"], status="running") status, body = self._req( "POST", "/tasks/%s/thread/message" % task["id"], {"body": "下一步改 README"}, self.owner_key) self.assertEqual(status, 201, body) self.assertEqual(body["delivery"], "queued") status, body = self._req( "POST", "/tasks/%s/thread/message" % task["id"], {"body": "立刻改", "mode": "direct"}, self.owner_key) self.assertEqual(status, 201, body) self.assertEqual(body["delivery"], "direct") status, body = self._req( "POST", "/tasks/%s/thread/message" % task["id"], {"body": "坏模式", "mode": "bogus"}, self.owner_key) self.assertEqual(status, 400, body) # 失败任务发修复指令：复活（不是 409），状态回 queued。 done = self._create_task() self._set_task(done["id"], status="failed", error_text="compile error") status, body = self._req( "POST", "/tasks/%s/thread/message" % done["id"], {"body": "继续修复编译错误并重跑测试"}, self.owner_key) self.assertEqual(status, 201, body) self.assertEqual(body["delivery"], "revived") status, task_after = self._req( "GET", "/tasks/%s" % done["id"], None, self.owner_key) self.assertEqual(task_after["status"], "queued") self.assertIsNone(task_after["error_text"]) # 已合并任务不能复活（worktree 已删），应 409。 merged = self._create_task() self._set_task(merged["id"], status="merged") status, body = self._req( "POST", "/tasks/%s/thread/message" % merged["id"], {"body": "再改一下"}, self.owner_key) self.assertEqual(status, 409, body) def test_handoff_lifecycle_and_cross_project(self): task = self._create_task() status, body = self._req( "GET", "/tasks/%s/handoff" % task["id"], None, self.owner_key) self.assertEqual(status, 200, body) self.assertIsNone(body["handoff"]) status, h = self._req( "POST", "/tasks/%s/handoff" % task["id"], {"target_host": "host-b"}, self.owner_key) self.assertEqual(status, 201, h) self.assertEqual(h["status"], "pending") status, body = self._req( "POST", "/tasks/%s/handoff" % task["id"], {"target_host": "host-c"}, self.owner_key) self.assertEqual(status, 409, body) status, body = self._req( "POST", "/tasks/%s/handoff/%s/report" % (task["id"], h["id"]), {"status": "running", "progress": 40}, self.owner_key) self.assertEqual(status, 200, body) self.assertEqual((body["status"], body["progress"]), ("running", 40)) status, body = self._req( "GET", "/tasks/%s/handoff" % task["id"], None, self.owner_key) self.assertEqual(body["handoff"]["status"], "running") other = self._create_task(self.project2["id"]) status, body = self._req( "POST", "/tasks/%s/handoff/%s/report" % (other["id"], h["id"]), {"status": "completed"}, self.owner_key) self.assertEqual(status, 404, body) status, body = self._req( "POST", "/tasks/%s/handoff/%s/report" % (task["id"], h["id"]), {"status": "completed"}, self.owner_key) self.assertEqual(status, 200, body) status, body = self._req( "POST", "/tasks/%s/handoff/%s/report" % (task["id"], h["id"]), {"status": "running"}, self.owner_key) self.assertEqual(status, 409, body) def test_worktree_status_derived(self): task = self._create_task() status, body = self._req( "GET", "/tasks/%s/worktree" % task["id"], None, self.owner_key) self.assertEqual(status, 200, body) self.assertEqual(body["status"], "creating") import re safe = re.sub(r"[^A-Za-z0-9_.-]", "-", task["id"]) path = os.path.join(self.server.host_worktrees_dir, safe) os.makedirs(os.path.join(path, ".git"), exist_ok=True) status, body = self._req( "GET", "/tasks/%s/worktree" % task["id"], None, self.owner_key) self.assertEqual(body["status"], "ready") self.assertTrue(body["exists"]) self._set_task(task["id"], worktree_host="other-host") status, body = self._req( "GET", "/tasks/%s/worktree" % task["id"], None, self.owner_key) self.assertEqual(body["status"], "remote") self.assertIsNone(body["exists"]) def test_pr_checks_and_no_pr(self): task = self._create_task() status, body = self._req( "GET", "/tasks/%s/pr/checks" % task["id"], None, self.owner_key) self.assertEqual(status, 409, body) self._set_task(task["id"], pr_number=42, pr_url="https://example.invalid/pr/42") # 制表符回落 + 无仓库时不查 check-runs API。 self.server.pr_gh_runner = lambda cmd, cwd: SimpleNamespace( returncode=0, stdout="build\tpass\ntest\tfail\n", stderr="") status, body = self._req( "GET", "/tasks/%s/pr/checks" % task["id"], None, self.owner_key) self.assertEqual(status, 200, body) self.assertFalse(body["all_pass"]) self.assertEqual(body["failed"], ["test"]) self.assertIn("test", body["diagnostics"]) self.assertFalse(body["check_runs"]["available"]) # 有仓库时：JSON checks + check-runs 日志与行号注释。 status, repo_proj = self._req( "POST", "/projects", {"name": "repo-proj", "repo_url": "https://github.com/openqj/demo.git"}, self.owner_key) self.assertEqual(status, 201, repo_proj) repo_task = self._create_task(repo_proj["id"]) self._set_task(repo_task["id"], pr_number=43, pr_url="https://example.invalid/pr/43") def _gh(cmd, cwd): if "--json" in cmd: return SimpleNamespace(returncode=0, stdout=json.dumps([ {"name": "build", "bucket": "pass", "workflow": "CI", "link": "https://example.invalid/ci/1", "description": "ok"}, {"name": "test", "bucket": "fail", "workflow": "CI", "link": "https://example.invalid/ci/2", "description": "1 failed"}]), stderr="") if "view" in cmd: return SimpleNamespace(returncode=0, stdout="abc123\n", stderr="") if any("commits" in c for c in cmd): return SimpleNamespace(returncode=0, stdout=json.dumps( {"check_runs": [{"id": 99, "name": "test", "conclusion": "failure"}]}), stderr="") if any("check-runs" in c for c in cmd): return SimpleNamespace(returncode=0, stdout=json.dumps({ "id": 99, "conclusion": "failure", "output": { "title": "Test failed", "summary": "1 test failed", "text": "FAILED test_x: assert failed", "annotations": [{ "path": "src/a.py", "start_line": 12, "annotation_level": "failure", "message": "assert failed"}]}}), stderr="") return SimpleNamespace(returncode=1, stdout="", stderr="boom") self.server.pr_gh_runner = _gh status, body = self._req( "GET", "/tasks/%s/pr/checks" % repo_task["id"], None, self.owner_key) self.assertEqual(status, 200, body) self.assertEqual(body["checks"][1]["workflow"], "CI") self.assertIn("example.invalid/ci/2", body["checks"][1]["link"]) self.assertTrue(body["check_runs"]["available"]) run_detail = body["check_runs"]["runs"][0] self.assertIn("FAILED test_x", run_detail["log_text"]) self.assertEqual(run_detail["annotations"][0]["path"], "src/a.py") self.assertEqual(run_detail["annotations"][0]["start_line"], 12) self._set_task(task["id"], worktree_host="other-host") status, body = self._req( "GET", "/tasks/%s/pr/checks" % task["id"], None, self.owner_key) self.assertEqual(status, 409, body) def test_sleep_and_wait(self): task = self._create_task() status, body = self._req( "POST", "/tasks/%s/sleep" % task["id"], {"seconds": 0}, self.owner_key) self.assertEqual(status, 200, body) self.assertEqual(body["slept_seconds"], 0) status, body = self._req( "POST", "/tasks/%s/sleep" % task["id"], {"seconds": 61}, self.owner_key) self.assertEqual(status, 400, body) parent = self._create_task() child = self._create_task() self._set_task(child["id"], status="completed", parent_task_id=parent["id"]) # 无游标：已终态的子任务立即作为首变返回，不等满超时。 status, body = self._req( "POST", "/tasks/%s/wait" % parent["id"], {"timeout_seconds": 30}, self.owner_key) self.assertEqual(status, 200, body) self.assertTrue(body["all_terminal"]) self.assertFalse(body["timed_out"]) self.assertEqual(body["children"][0]["task_id"], child["id"]) self.assertEqual(body["changed_task"]["task_id"], child["id"]) self.assertEqual(body["changed_task"]["kind"], "terminal") cursor = body["cursor"] # 同一游标再等：已终态不再重复报，超时回进度 + 新游标。 status, body = self._req( "POST", "/tasks/%s/wait" % parent["id"], {"timeout_seconds": 0, "after_cursor": cursor}, self.owner_key) self.assertEqual(status, 200, body) self.assertTrue(body["timed_out"]) self.assertIsNone(body["changed_task"]) # 非法游标 400；超 8 个 400；运行中 + 超时 0 回超时。 status, body = self._req( "POST", "/tasks/%s/wait" % parent["id"], {"timeout_seconds": 0, "after_cursor": "bogus!!"}, self.owner_key) self.assertEqual(status, 400, body) running = self._create_task() status, body = self._req( "POST", "/tasks/%s/wait" % parent["id"], {"task_ids": [running["id"]], "timeout_seconds": 0}, self.owner_key) self.assertEqual(status, 200, body) self.assertTrue(body["timed_out"]) self.assertFalse(body["all_terminal"]) status, body = self._req( "POST", "/tasks/%s/wait" % parent["id"], {"task_ids": ["x"] * 9, "timeout_seconds": 0}, self.owner_key) self.assertEqual(status, 400, body) status, body = self._req( "POST", "/tasks/%s/wait" % parent["id"], {"timeout_seconds": 121}, self.owner_key) self.assertEqual(status, 400, body) def test_wait_cursor_approval_trigger(self): # 基线游标（运行中、无待审批）-> 新建 elicitation 审批 # -> 再等同一游标，立即报 needs_input（不是等满超时）。 parent = self._create_task() target = self._create_task() status, body = self._req( "POST", "/tasks/%s/wait" % parent["id"], {"task_ids": [target["id"]], "timeout_seconds": 0}, self.owner_key) self.assertEqual(status, 200, body) self.assertTrue(body["timed_out"]) cursor = body["cursor"] status, appr = self._req( "POST", "/tasks/%s/approvals" % target["id"], {"kind": "elicitation", "summary": "请确认是否继续"}, self.owner_key) self.assertEqual(status, 201, appr) status, body = self._req( "POST", "/tasks/%s/wait" % parent["id"], {"task_ids": [target["id"]], "timeout_seconds": 30, "after_cursor": cursor}, self.owner_key) self.assertEqual(status, 200, body) self.assertFalse(body["timed_out"]) self.assertEqual(body["changed_task"]["kind"], "needs_input") self.assertEqual(body["changed_task"]["task_id"], target["id"]) def test_automation_update_and_get(self): status, auto = self._req( "POST", "/projects/%s/automations" % self.project["id"], {"name": "daily", "prompt": "做每日检查"}, self.owner_key) self.assertEqual(status, 201, auto) status, body = self._req( "PATCH", "/projects/%s/automations/%s" % (self.project["id"], auto["id"]), {"interval_seconds": 7200}, self.owner_key) self.assertEqual(status, 200, body) self.assertEqual(body["interval_seconds"], 7200) status, body = self._req( "GET", "/projects/%s/automations/%s" % (self.project["id"], auto["id"]), None, self.owner_key) self.assertEqual(status, 200, body) self.assertEqual(body["runs"], []) status, body = self._req( "PATCH", "/projects/%s/automations/%s" % (self.project["id"], auto["id"]), {}, self.owner_key) self.assertEqual(status, 400, body) status, body = self._req( "PATCH", "/projects/%s/automations/%s" % (self.project2["id"], auto["id"]), {"interval_seconds": 3600}, self.owner_key) self.assertEqual(status, 404, body) status, body = self._req( "POST", "/projects/%s/automations/%s/cancel" % (self.project["id"], auto["id"]), {}, self.owner_key) self.assertEqual(status, 200, body) status, body = self._req( "PATCH", "/projects/%s/automations/%s" % (self.project["id"], auto["id"]), {"interval_seconds": 3600}, self.owner_key) self.assertEqual(status, 409, body) if __name__ == "__main__": unittest.main()