part 5 修复 1:单元内 health 恒 false——default_codex_bin() 回落 shutil.which

RETEST4 新发现 1:单元内无 CODEX_BIN、无 HERE/bin/codex 时
default_codex_bin() 返回裸 "codex",run_health_check 的
os.path.exists("codex") 以相对路径判恒为 False,health 的 ok
恒 False(execpath4、p2_7v4、P1 resident_health、P1B 两 call
全因此失败;定界探针证实单元内 which codex 就是
/usr/local/bin/codex,加 CODEX_BIN 前缀即 rc=0 且 ok=true)。

三选一取 (c) 回落 shutil.which("codex"):
- (a) CODEX_BIN 进 FORWARD_ENV_ALLOWLIST:要执行器与 chv-cell
  两边接线、还得在客机环境里定一个单元内路径,改动面大,且
  只修执行器驱动的调用,直接跑 host.py 的 health 修不到。
- (b) _guest_host_command 显式加 --codex-bin:把客机镜像路径
  硬编码进执行器,同样只修执行器驱动的调用。
- (c) 单点修 host.py:health、Server、automations 全部调用方
  直接受益;PATH 里找得到即返回绝对路径,找不到仍返回裸
  "codex"(行为与旧代码一致,零回归面);不硬编码任何路径。

新测试 tests/test_host.py::DefaultCodexBinTest(4 项):模拟
单元内环境(PATH 含 codex、cwd 不含 codex、无 CODEX_BIN)
断言返回绝对路径、CODEX_BIN 优先、处处无 codex 时仍返回裸
"codex"、以及该解析下的 run_health_check ok=true。
This commit is contained in:
host committed 2026-11-07 12:06:20 +00:00
1 parent c4661b64db
commit 39084ce2bf
2 files changed
+110

No files matched your search

+17
View File
@@ -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"
+93
View File
@@ -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=""):