diff --git a/cloud/server.py b/cloud/server.py index 1082662..0f4f4e3 100644 --- a/cloud/server.py +++ b/cloud/server.py @@ -315,13 +315,18 @@ def _host_py_path(): return os.path.join(os.path.dirname(here), "host.py") -def _run_host_cli(argv, codex_home, timeout=60): +def _run_host_cli(argv, codex_home, timeout=60, model_provider=None): """调 host.py CLI 子命令,返回 (ok, result_or_error)。 ok 为 True 时 result 是 stdout 解析出的 JSON 对象;ok 为 False 时 result 是可直接放进 502 error 字段的错误文本 (CLI 自带的 error 原样透传,解析失败/缺文件/超时则给 明确的服务端错误,绝不把 stderr 或原始 key 混进响应)。 + model_provider 是 host.py 顶层参数,必须放在子命令之前: + app-server 按进程配置的 provider 过滤 thread/list,不带 + 注入时 sub2api 下建的 thread 在列表里根本看不到(见 + _list_task_threads 的 provider 重试),只在解析处传, + 绝不由调用方把 provider 拼进 argv(子命令解析器不认)。 """ host_py = _host_py_path() if not host_py or not os.path.isfile(host_py): @@ -329,6 +334,8 @@ def _run_host_cli(argv, codex_home, timeout=60): cmd = [sys.executable, host_py] if codex_home: cmd += ["--codex-home", codex_home] + if model_provider: + cmd += ["--model-provider", model_provider] cmd += list(argv) try: proc = subprocess.run(cmd, capture_output=True, text=True, @@ -376,6 +383,34 @@ def _tenant_codex_home(server, project_id): return os.path.join(_tenants_root_for(server), safe, "codex-home") +def _local_worker_codex_home(): + """本地 worker 的共享 codex-home(兜底解析源)。 + + 本地 worker 按 host-config-host.json 跑,codex_home 是该 + 配置里的共享目录,不在租户 home 里;服务端主解析先查 + 租户 home,空时才查这里——只读该配置文件的 codex_home + 一个键,文件不存在、解析失败或键为空时回 None(调用方 + 跳过该候选,不因此 502)。相对路径按配置文件所在目录 + 解析(与 host.py 的 _resolve_path 同规则)。 + """ + try: + cfg_path = os.path.join( + os.path.dirname(_host_py_path()), "host-config-host.json") + with open(cfg_path, encoding="utf-8") as f: + cfg = json.load(f) + home = cfg.get("codex_home") if isinstance(cfg, dict) else None + if not home or not isinstance(home, str): + return None + home = home.strip() + if not home: + return None + if not os.path.isabs(home): + home = os.path.join(os.path.dirname(cfg_path), home) + return home + except Exception: + return None + + def _tenant_work_dir(server, project_id): """项目对应的租户工作目录:tenants_root/<安全项目名>/work, skills/list 的 cwds 过滤用它(不传 cwd 时 app-server 会 @@ -4209,36 +4244,74 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): def _list_task_threads(self, task): """对任务调 thread-list:返回 (threads, codex_home, cwd, - err),err 为 (status, msg) 或 None。cwd 过滤用任务 - worktree(见 _thread_worktree_for_task);codex-home - 用租户 home(见 _tenant_codex_home)。""" + err, provider),err 为 (status, msg) 或 None。cwd 过滤 + 用任务 worktree(见 _thread_worktree_for_task)。 + + 候选按序穷尽,首个非空即停(thread 操作不是热路径, + 多次 CLI 调用可接受): + - home:租户 home(见 _tenant_codex_home)优先,本地 + worker 共享 home(见 _local_worker_codex_home)兜底 + ——本地派发至今仍用共享 home,旧 thread 全在那里, + 只查租户 home 永远解析不到; + - provider:缺省(配置 provider)优先,sub2api 次之 + ——app-server 按进程 provider 过滤列表,不带注入时 + sub2api 下建的 thread 在列表里不存在;未配 + SUB2API_BASE_URL 时跳过次选(host.py 会 loud fail, + 不能把配置缺失变成 502)。 + 任一候选成功返回空列表时不再继续(空就是空);全部 + 候选都报错(host.py 本身不可用)时回首个错误 502。""" codex_home = _tenant_codex_home(self.server, task["project_id"]) if codex_home is None: return None, None, None, ( - 502, "cannot resolve tenant codex-home") + 502, "cannot resolve tenant codex-home"), None cwd = _thread_worktree_for_task(self.server, task) - ok, result = _run_host_cli(["thread-list", "--cwd", cwd], - codex_home) - if not ok: - return None, None, None, (502, result) - threads = _extract_thread_list(result) - if threads is None: - return None, None, None, ( - 502, "host.py thread-list output has no thread list") - return threads, codex_home, cwd, None + homes = [codex_home] + shared = _local_worker_codex_home() + if shared and shared != codex_home: + homes.append(shared) + providers = [None] + if os.environ.get("SUB2API_BASE_URL", "").strip(): + providers.append("sub2api") + first_error = None + any_success = False + for home in homes: + for provider in providers: + ok, result = _run_host_cli( + ["thread-list", "--cwd", cwd], home, + model_provider=provider) + if not ok: + if first_error is None: + first_error = result + continue + threads = _extract_thread_list(result) + if threads is None: + if first_error is None: + first_error = ("host.py thread-list output has" + " no thread list") + continue + any_success = True + if threads: + return threads, home, cwd, None, provider + if not any_success and first_error is not None: + return None, None, None, (502, first_error), None + return [], codex_home, cwd, None, None def _resolve_task_thread(self, task): """单线程端点的解析:thread-list 后按 _pick_latest_thread 取最新一个非 archived。返回 (thread, codex_home, cwd, - err);零个 thread 时 err 为 (404, "thread not found")。""" - threads, codex_home, cwd, err = self._list_task_threads(task) + err, provider);零个 thread 时 err 为 (404, "thread not + found")。provider 必须原样带给后续的 thread-read / + rename / archive CLI——同一 thread 在另一 provider + 进程里同样看不到,丢了它改操作会 502。""" + threads, codex_home, cwd, err, provider = \ + self._list_task_threads(task) if err is not None: - return None, None, None, err + return None, None, None, err, None thread = _pick_latest_thread(threads) if thread is None: - return None, None, None, (404, "thread not found") - return thread, codex_home, cwd, None + return None, None, None, (404, "thread not found"), None + return thread, codex_home, cwd, None, provider def _task_threads(self, user_id, task_id): """GET /tasks/{id}/threads:thread/list 原样返回列表。 @@ -4253,7 +4326,7 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): if guard is not None: self._send_json(guard[0], {"error": guard[1]}) return - threads, _home, cwd, err = self._list_task_threads(task) + threads, _home, cwd, err, _provider = self._list_task_threads(task) if err is not None: self._send_json(err[0], {"error": err[1]}) return @@ -4273,7 +4346,8 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): if guard is not None: self._send_json(guard[0], {"error": guard[1]}) return - thread, codex_home, cwd, err = self._resolve_task_thread(task) + thread, codex_home, cwd, err, provider = \ + self._resolve_task_thread(task) if err is not None: self._send_json(err[0], {"error": err[1]}) return @@ -4281,7 +4355,8 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): include_turns = (query.get("include_turns") or ["0"])[0] if include_turns == "1": argv.append("--include-turns") - ok, result = _run_host_cli(argv, codex_home) + ok, result = _run_host_cli(argv, codex_home, + model_provider=provider) if not ok: self._send_json(502, {"error": result}) return @@ -4312,13 +4387,14 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): "error": "thread ops require a terminal task", "status": task["status"]}) return - thread, codex_home, _cwd, err = self._resolve_task_thread(task) + thread, codex_home, _cwd, err, provider = \ + self._resolve_task_thread(task) if err is not None: self._send_json(err[0], {"error": err[1]}) return ok, result = _run_host_cli( list(argv) + ["--thread-id", thread["id"]], - codex_home, timeout=timeout) + codex_home, timeout=timeout, model_provider=provider) if not ok: self._send_json(502, {"error": result}) return @@ -4567,14 +4643,15 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): if guard is not None: note = guard[1] else: - resolved, _home, _cwd, rerr = self._resolve_task_thread(task) + resolved, _home, _cwd, rerr, provider = \ + self._resolve_task_thread(task) if rerr is not None: note = rerr[1] else: thread_id = resolved.get("id") ok, result = _run_host_cli( ["thread-read", "--thread-id", thread_id], - _tenant_codex_home(self.server, task["project_id"])) + _home, model_provider=provider) if not ok: note = result else: diff --git a/cloud/tests/test_thread_resolve.py b/cloud/tests/test_thread_resolve.py new file mode 100644 index 0000000..a322348 --- /dev/null +++ b/cloud/tests/test_thread_resolve.py @@ -0,0 +1,220 @@ +"""线程解析兜底单测:provider 过滤 + 共享 home 回退。 + +背景(生产实测 2026-10-28):app-server 按进程 provider +过滤 thread/list,不带 sub2api 注入的进程看不到 sub2api +下建的 thread;本地 worker 又用共享 home 而非租户 home。 +旧解析只查租户 home + 缺省 provider,本地任务一律 +404 thread not found。本批修法:home(租户→共享)× +provider(缺省→sub2api)按序穷尽,首个非空即停,并把 +provider 原样带给后续的 thread-read/rename/archive CLI。 + +不跑真实模型、不调真实 host.py:_run_host_cli 与 +_local_worker_codex_home 均以 stub 替换。 +""" + +import json +import os +import sys +import tempfile +import unittest +from pathlib import Path +from unittest import mock + +try: + import cloud.server as server_mod + from cloud.server import ( + CloudControlPlaneHandler, + _local_worker_codex_home, + _run_host_cli, + ) +except ImportError: + sys.path.insert(0, str(Path(__file__).resolve().parents[2])) + import cloud.server as server_mod + from cloud.server import ( + CloudControlPlaneHandler, + _local_worker_codex_home, + _run_host_cli, + ) + + +def _handler(tmp): + h = CloudControlPlaneHandler.__new__(CloudControlPlaneHandler) + srv = mock.Mock() + srv.tenants_root = os.path.join(tmp, "tenants") + srv.host_worktrees_dir = os.path.join(tmp, "worktrees") + h.server = srv + return h + + +def _list_result(threads): + return {"data": threads} + + +class RunHostCliTestCase(unittest.TestCase): + def test_model_provider_goes_before_subcommand(self): + seen = {} + + class Proc: + returncode = 0 + stdout = json.dumps({"data": []}) + + def fake_run(cmd, **kwargs): + seen["cmd"] = cmd + return Proc() + + with mock.patch.object(server_mod.subprocess, "run", fake_run): + ok, _ = _run_host_cli(["thread-list", "--cwd", "/wt"], + "/home/x", model_provider="sub2api") + self.assertTrue(ok) + cmd = seen["cmd"] + self.assertLess(cmd.index("--model-provider"), + cmd.index("thread-list")) + self.assertEqual(cmd[cmd.index("--model-provider") + 1], + "sub2api") + + def test_no_provider_arg_when_none(self): + seen = {} + + class Proc: + returncode = 0 + stdout = json.dumps({"data": []}) + + def fake_run(cmd, **kwargs): + seen["cmd"] = cmd + return Proc() + + with mock.patch.object(server_mod.subprocess, "run", fake_run): + ok, _ = _run_host_cli(["thread-list"], "/home/x") + self.assertTrue(ok) + self.assertNotIn("--model-provider", seen["cmd"]) + + +class LocalWorkerHomeTestCase(unittest.TestCase): + def test_reads_host_config_next_to_host_py(self): + with tempfile.TemporaryDirectory() as tmp: + with open(os.path.join(tmp, "host-config-host.json"), + "w", encoding="utf-8") as f: + json.dump({"codex_home": "/srv/shared-home"}, f) + with mock.patch.object( + server_mod, "_host_py_path", + return_value=os.path.join(tmp, "host.py")): + self.assertEqual(_local_worker_codex_home(), + "/srv/shared-home") + + def test_relative_home_resolves_against_config_dir(self): + with tempfile.TemporaryDirectory() as tmp: + with open(os.path.join(tmp, "host-config-host.json"), + "w", encoding="utf-8") as f: + json.dump({"codex_home": "work/home"}, f) + with mock.patch.object( + server_mod, "_host_py_path", + return_value=os.path.join(tmp, "host.py")): + self.assertEqual(_local_worker_codex_home(), + os.path.join(tmp, "work/home")) + + def test_missing_config_returns_none(self): + with tempfile.TemporaryDirectory() as tmp: + with mock.patch.object( + server_mod, "_host_py_path", + return_value=os.path.join(tmp, "host.py")): + self.assertIsNone(_local_worker_codex_home()) + + +class ListFallbackTestCase(unittest.TestCase): + def _run(self, calls, env_sub2api=True): + """calls: {(home, provider): threads-or-error}。 + + home 用标记串:'tenant' / 'shared',stub 按 _run_host_cli + 收到的 codex_home 参数映射。 + """ + def fake_cli(argv, home, model_provider=None, **kwargs): + key = ("tenant" if "tenants" in home else "shared", + model_provider) + calls.append(key) + val = self.script.get(key) + if isinstance(val, Exception): + return False, str(val) + return True, _list_result(val if val is not None else []) + + return fake_cli + + def test_shared_home_plus_sub2api_finds_thread(self): + thread = {"id": "t-1", "updatedAt": 100, "cwd": "/wt"} + self.script = { + ("tenant", None): [], + ("tenant", "sub2api"): [], + ("shared", None): [], + ("shared", "sub2api"): [thread], + } + calls = [] + with tempfile.TemporaryDirectory() as tmp: + h = _handler(tmp) + task = {"id": "task-1", "project_id": "proj-1"} + env = {"SUB2API_BASE_URL": "https://example.invalid"} + with mock.patch.dict(os.environ, env), \ + mock.patch.object(server_mod, "_run_host_cli", + self._run(calls)), \ + mock.patch.object( + server_mod, "_local_worker_codex_home", + return_value=os.path.join(tmp, "shared")): + threads, home, _cwd, err, provider = \ + h._list_task_threads(task) + self.assertIsNone(err) + self.assertEqual(threads, [thread]) + self.assertEqual(provider, "sub2api") + self.assertIn("shared", home) + self.assertEqual(calls[-1], ("shared", "sub2api")) + + def test_tenant_hit_stops_early(self): + thread = {"id": "t-9", "updatedAt": 5} + self.script = {("tenant", None): [thread]} + calls = [] + with tempfile.TemporaryDirectory() as tmp: + h = _handler(tmp) + task = {"id": "task-1", "project_id": "proj-1"} + with mock.patch.dict(os.environ, {}, clear=False), \ + mock.patch.object(server_mod, "_run_host_cli", + self._run(calls)), \ + mock.patch.object( + server_mod, "_local_worker_codex_home", + return_value=os.path.join(tmp, "shared")): + threads, _home, _cwd, err, provider = \ + h._list_task_threads(task) + self.assertIsNone(err) + self.assertEqual(threads, [thread]) + self.assertIsNone(provider) + self.assertEqual(calls, [("tenant", None)]) + + def test_all_empty_resolves_to_404_not_502(self): + self.script = {} + with tempfile.TemporaryDirectory() as tmp: + h = _handler(tmp) + task = {"id": "task-1", "project_id": "proj-1"} + with mock.patch.dict( + os.environ, + {"SUB2API_BASE_URL": "https://example.invalid"}), \ + mock.patch.object(server_mod, "_run_host_cli", + self._run([])), \ + mock.patch.object( + server_mod, "_local_worker_codex_home", + return_value=os.path.join(tmp, "shared")): + _t, _h, _c, err, _p = h._resolve_task_thread(task) + self.assertEqual(err, (404, "thread not found")) + + def test_host_py_failure_stays_502(self): + self.script = {} + with tempfile.TemporaryDirectory() as tmp: + h = _handler(tmp) + task = {"id": "task-1", "project_id": "proj-1"} + with mock.patch.object( + server_mod, "_run_host_cli", + return_value=(False, "host.py not found")), \ + mock.patch.object( + server_mod, "_local_worker_codex_home", + return_value=None): + _t, _h, _c, err, _p = h._resolve_task_thread(task) + self.assertEqual(err[0], 502) + + +if __name__ == "__main__": + unittest.main() diff --git a/docs/thread-resolve-fix.md b/docs/thread-resolve-fix.md new file mode 100644 index 0000000..b97e82d --- /dev/null +++ b/docs/thread-resolve-fix.md @@ -0,0 +1,63 @@ +# 线程解析修复:provider 过滤 + 共享 home 回退 + +## 故障(生产实测 2026-10-28) + +本地任务的 thread rename/archive/unarchive 稳定 404 +`thread not found`,thread/read 回 200 但 thread 为 null。 +而同一任务的 worktree、bindings、快照、心跳、移交全部正常。 + +## 定位(证据链,全部在生产主机上实测) + +1. host-state 里有该任务的 thread_id,cwd 与服务端推导的 + worktree 路径一字不差;rollout 文件的 session_meta cwd + 也是同一路径;state_5.sqlite 的 threads 表里有该行 + (cwd 正确、archived=0、最新)。 +2. 不带 provider 注入的 app-server 进程调 thread/list: + 全量翻页(cursor 可用)共 77 条,探针 thread 不在其中; + 带 cwd 过滤时回空;页内已知 thread 的 cwd 过滤正常回 1 条。 +3. 区分点是 provider:列出的 77 条全是 xcmapi,探针两条 + 是 sub2api;带 `--model-provider sub2api` 注入的进程调 + 同一 cwd 过滤,立刻回 1 条(正是探针 thread),无过滤 + 首页即含它。 +4. 第二个错位:服务端一直查租户 home + (tenants/<项目>/codex-home),本地 worker 按 + host-config-host.json 用共享 home + (/var/lib/codex-app-server/codex-home),旧 thread 全在后者。 + +结论:app-server 按进程 provider 过滤 thread/list,且 +服务端只查租户 home——两个错位叠加,本地任务解析必 404。 +不是分页 cap(翻页后仍无)、不是 cwd 记错、不是 home 丢数据。 + +## 修法(只改服务端,不动派发) + +`cloud/server.py`: + +- `_run_host_cli` 新增 `model_provider` 参数,拼在子命令 + 之前(host.py 顶层参数,拼进 argv 会被子命令解析器拒绝)。 +- `_list_task_threads` 按序穷尽,首个非空即停: + home:租户 home → 本地 worker 共享 home + (`_local_worker_codex_home`,只读 host-config-host.json + 的 codex_home 一键,缺失/解析失败时跳过); + provider:缺省 → sub2api(仅当 server 进程配了 + SUB2API_BASE_URL,否则 host.py loud fail,不能变 502)。 + 任一候选成功回空即停(空就是空);全部报错才 502 首错。 +- `_resolve_task_thread` 返回 provider,后续的 thread-read / + rename / archive / unarchive 原样带 provider——同一 thread + 在另一 provider 进程里同样看不到,丢了它改操作会 502。 +- `_task_thread_read`(#103 的富化 read)此前解析后又用 + 租户 home 调 thread-read,已改成解析出的 home + provider。 + +不改派发的原因:本地 worker 改租户 home 会丢共享 home +里的 skills 等内容,有回归风险;且服务端兜底已覆盖全部 +现存 thread。租户隔离(本地任务共用一个 home)是另案, +不在本批修。 + +## 未覆盖 + +- Docker 任务仍 409(thread 在容器内,服务端查不到是设计)。 +- `_task_threads` 列表端点同样走 `_list_task_threads`,兜底 + 对它同样生效;但它不翻页,只返回解析到的该任务 thread。 +- 真 PR 创建仍未实弹(与本批无关)。 +- 新测试 9 个:`cloud/tests/test_thread_resolve.py` + (CLI 参数位置、共享 home 读取、回退顺序、空→404、 + host.py 失败→502)。