diff --git a/host.py b/host.py index d58a2f4..0d8b7f3 100644 --- a/host.py +++ b/host.py @@ -50,6 +50,7 @@ import os import queue import re import shlex +import shutil import subprocess import sys import threading @@ -201,6 +202,22 @@ def default_codex_bin(): paired = os.path.join(HERE, "bin", "codex") if os.path.exists(paired): return paired + # part 5(RETEST4 新发现 1):单元内既无 CODEX_BIN 又无 + # HERE/bin/codex 时,旧代码返回裸 "codex",run_health_check + # 的 os.path.exists("codex") 以相对路径判恒为 False, + # health 的 ok 恒 False(单元内 which codex 明明是 + # /usr/local/bin/codex)。回落到 shutil.which("codex"): + # PATH 里找得到即返回绝对路径,health 与 Server 都直接 + # 受益;找不到时仍返回裸 "codex",行为与旧代码一致。 + # 不选 (a) CODEX_BIN 进 FORWARD_ENV_ALLOWLIST:要执行器 + # 与 chv-cell 两边接线、还得在客机环境里定一个单元内路径, + # 改动面大,且只修执行器驱动的调用;不选 (b) 执行器显式 + # 加 --codex-bin:把客机镜像路径硬编码进执行器,且直接 + # 跑 host.py 的 health 同样修不到。单点修 host.py 改动 + # 最小、覆盖全部调用方。 + found = shutil.which("codex") + if found: + return found return "codex" diff --git a/tests/test_host.py b/tests/test_host.py index 3e4e16c..fa6e279 100644 --- a/tests/test_host.py +++ b/tests/test_host.py @@ -6,6 +6,7 @@ import sys import tempfile import unittest from pathlib import Path +from unittest import mock REPO_ROOT = Path(__file__).resolve().parents[1] @@ -558,6 +559,98 @@ class HealthRestoreRecoverTest(unittest.TestCase): self.assertTrue(report["checks"]["server_ok"]) +class DefaultCodexBinTest(unittest.TestCase): + """part 5 修复 1(RETEST4 新发现 1):default_codex_bin() + 回落到 shutil.which("codex")。 + + 模拟单元内环境:PATH 含 codex 所在目录、cwd 不含 codex、 + 无 CODEX_BIN、无 HERE/bin/codex。旧代码此时返回裸 "codex", + run_health_check 的 os.path.exists("codex") 恒 False, + health 的 ok 恒 False。""" + + class FakeServer: + def initialize(self): + return {} + + def list_threads(self): + return {"data": []} + + def close(self): + pass + + def _fake_bin_dir(self, tmp): + bindir = os.path.join(tmp, "bin") + os.makedirs(bindir) + codex = os.path.join(bindir, "codex") + with open(codex, "w") as fh: + fh.write("#!/bin/sh\nexit 0\n") + os.chmod(codex, 0o755) + return bindir, codex + + def test_which_fallback_returns_absolute_path(self): + # 仓库根下没有配对的 bin/codex(修的是无配对的情形)。 + self.assertFalse( + os.path.exists(os.path.join(host.HERE, "bin", "codex"))) + with tempfile.TemporaryDirectory() as tmp: + bindir, codex = self._fake_bin_dir(tmp) + workdir = os.path.join(tmp, "work") + os.makedirs(workdir) + old_cwd = os.getcwd() + try: + os.chdir(workdir) + env = {"PATH": bindir} + with mock.patch.dict(os.environ, env, clear=True): + self.assertEqual(host.default_codex_bin(), codex) + finally: + os.chdir(old_cwd) + + def test_codex_bin_env_wins_over_path(self): + with tempfile.TemporaryDirectory() as tmp: + bindir, _codex = self._fake_bin_dir(tmp) + env = {"PATH": bindir, "CODEX_BIN": "/custom/codex"} + with mock.patch.dict(os.environ, env, clear=True): + self.assertEqual(host.default_codex_bin(), + "/custom/codex") + + def test_no_codex_anywhere_returns_bare_name(self): + with tempfile.TemporaryDirectory() as tmp: + empty = os.path.join(tmp, "empty") + os.makedirs(empty) + with mock.patch.dict(os.environ, {"PATH": empty}, + clear=True): + self.assertEqual(host.default_codex_bin(), "codex") + + def test_health_ok_with_which_resolved_bin(self): + # 验收口径:不加任何前缀的单元内 health 应 rc=0 且 ok=true。 + # 这里钉 host 侧:codex_bin_exists 为真且 report ok 为真。 + with tempfile.TemporaryDirectory() as tmp: + bindir, _codex = self._fake_bin_dir(tmp) + workdir = os.path.join(tmp, "work") + os.makedirs(workdir) + path = os.path.join(tmp, "host-state.json") + with open(path, "w") as f: + json.dump({"tasks": {}, "automations": [], + "version": 1}, f) + with open(path + ".bak", "w") as f: + json.dump({"tasks": {}, "automations": [], + "version": 1}, f) + old_cwd = os.getcwd() + try: + os.chdir(workdir) + with mock.patch.dict(os.environ, {"PATH": bindir}, + clear=True): + codex_bin = host.default_codex_bin() + self.assertTrue(os.path.isabs(codex_bin)) + report = host.run_health_check( + path, {}, codex_bin, tmp, "deny", + server_factory=lambda: self.FakeServer()) + finally: + os.chdir(old_cwd) + self.assertTrue(report["checks"]["codex_bin_exists"], + report) + self.assertTrue(report["ok"], report) + + class GithubPRTest(unittest.TestCase): class FakeGh: def __init__(self, checks_output="", pr_url=""):