fix(computer_use): surface CLI --version when health_report version lies

This commit is contained in:
monerostar 2026-07-25 15:13:02 -06:00 committed by Teknium
parent c6db7b0f4f
commit b9215f5bc9
2 changed files with 197 additions and 11 deletions

View file

@ -79,6 +79,24 @@ def _degraded_report() -> dict:
}
@pytest.fixture(autouse=True)
def _default_cli_version_matches_report(monkeypatch):
"""Existing tests mock only the MCP Popen handshake. ``subprocess.run``
(used for ``--version``) goes through Popen too, so without this the
mock breaks version probing. Default to a CLI version that matches
``_ok_report`` / ``_degraded_report`` (0.5.8); identity tests override.
"""
from tools.computer_use import doctor
monkeypatch.setattr(
doctor,
"_read_cli_version",
lambda binary, timeout=5.0: "cua-driver 0.5.8",
)
# ── exit codes ─────────────────────────────────────────────────────────────
@ -286,11 +304,14 @@ class TestJsonOutput:
patch("subprocess.Popen", return_value=proc), \
patch("sys.stdout", new_callable=StringIO) as out:
doctor.run_doctor(json_output=True)
# Verify the captured text round-trips through json.loads and matches
# the input report (the contract: --json passes the structured payload
# through unchanged so downstream tooling can consume it directly).
# Verify the captured text round-trips through json.loads. Upstream
# health_report keys are preserved; Hermes adds hermes_identity.
parsed = json.loads(out.getvalue())
assert parsed == _ok_report()
report = _ok_report()
for key, value in report.items():
assert parsed[key] == value
assert "hermes_identity" in parsed
assert parsed["hermes_identity"]["resolved_binary"]
# ── HERMES_CUA_DRIVER_CMD resolution ───────────────────────────────────────
@ -533,7 +554,11 @@ class TestHealthReportFallback:
assert code == 0
fallback_mock.assert_not_called()
parsed = json.loads(out.getvalue())
assert parsed == _ok_report()
# Upstream health_report keys pass through unchanged; Hermes adds
# only the additive hermes_identity envelope.
for key, value in _ok_report().items():
assert parsed[key] == value
assert "hermes_identity" in parsed
assert "fallback" not in parsed
def test_extract_raises_health_report_unavailable_on_isError(self):
@ -581,3 +606,68 @@ class TestHealthReportFallback:
parsed = json.loads(out.getvalue())
assert parsed["overall"] == "degraded"
assert parsed.get("fallback") is True
# ── binary identity (CLI --version vs health_report) ───────────────────────
class TestDoctorVersionIdentity:
def test_header_prefers_cli_version_on_mismatch(self):
"""Windows has been observed reporting 0.8.3 via health_report while
the resolved binary is 0.12.6 doctor must surface the real version."""
from tools.computer_use import doctor
proc = _fake_proc_with_responses(
{"jsonrpc": "2.0", "id": 1, "result": {}},
{"jsonrpc": "2.0", "id": 2, "result": {"structuredContent": _ok_report()}},
)
# _ok_report claims 0.5.8; CLI says 0.12.6
with patch("shutil.which", return_value="/fake/cua-driver"), \
patch("subprocess.Popen", return_value=proc), \
patch.object(doctor, "_read_cli_version", return_value="cua-driver 0.12.6"), \
patch("sys.stdout", new_callable=StringIO) as out:
code = doctor.run_doctor()
assert code == 0
text = out.getvalue()
assert "0.12.6" in text
assert "version mismatch" in text.lower()
assert "0.5.8" in text # health_report value still shown
def test_json_includes_hermes_identity(self):
from tools.computer_use import doctor
proc = _fake_proc_with_responses(
{"jsonrpc": "2.0", "id": 1, "result": {}},
{"jsonrpc": "2.0", "id": 2, "result": {"structuredContent": _ok_report()}},
)
with patch("shutil.which", return_value="/fake/cua-driver"), \
patch("subprocess.Popen", return_value=proc), \
patch.object(doctor, "_read_cli_version", return_value="cua-driver 0.12.6"), \
patch("sys.stdout", new_callable=StringIO) as out:
code = doctor.run_doctor(json_output=True)
assert code == 0
payload = json.loads(out.getvalue())
assert payload["overall"] == "ok"
assert "hermes_identity" in payload
ident = payload["hermes_identity"]
assert ident["version_mismatch"] is True
assert "0.12.6" in (ident.get("cli_version") or "")
assert ident.get("health_report_driver_version") == "0.5.8"
assert ident.get("resolved_binary")
def test_matching_versions_no_mismatch_flag(self):
from tools.computer_use import doctor
proc = _fake_proc_with_responses(
{"jsonrpc": "2.0", "id": 1, "result": {}},
{"jsonrpc": "2.0", "id": 2, "result": {"structuredContent": _ok_report()}},
)
with patch("shutil.which", return_value="/fake/cua-driver"), \
patch("subprocess.Popen", return_value=proc), \
patch.object(doctor, "_read_cli_version", return_value="cua-driver 0.5.8"), \
patch("sys.stdout", new_callable=StringIO) as out:
code = doctor.run_doctor(json_output=True)
assert code == 0
payload = json.loads(out.getvalue())
assert payload["hermes_identity"]["version_mismatch"] is False

View file

@ -98,6 +98,57 @@ def _is_valid_health_report(payload: Any) -> bool:
return True
def _read_cli_version(binary: str, *, timeout: float = 5.0) -> Optional[str]:
"""Return ``cua-driver --version`` stdout (stripped), or None on failure.
health_report's ``driver_version`` / binary_version check can disagree
with the actual binary (observed on Windows: health_report claims
0.8.3 while ``--version`` and the on-disk release are 0.12.6). Doctor
surfaces both so operators are not misled when debugging session
issues against a "wrong" version string.
"""
try:
completed = subprocess.run(
[binary, "--version"],
capture_output=True,
text=True,
encoding="utf-8",
errors="replace",
timeout=timeout,
env=_sanitized_cua_env(),
)
except (OSError, subprocess.TimeoutExpired, ValueError, TypeError):
return None
text = (completed.stdout or completed.stderr or "").strip()
if not text:
return None
# First non-empty line only — keep the banner compact.
return text.splitlines()[0].strip()
def _normalize_version_token(text: str) -> str:
"""Pull a dotted version-ish token out of a free-form version string."""
if not text:
return ""
m = re.search(r"(\d+\.\d+(?:\.\d+)?(?:[-+][\w.]+)?)", text)
return m.group(1) if m else text.strip().lower()
def _build_identity(binary: str, report: Dict[str, Any]) -> Dict[str, Any]:
"""Hermes-side identity block comparing resolved binary vs health_report."""
cli = _read_cli_version(binary) or ""
report_v = str(report.get("driver_version") or "")
cli_tok = _normalize_version_token(cli)
report_tok = _normalize_version_token(report_v)
mismatch = bool(cli_tok and report_tok and cli_tok != report_tok)
return {
"resolved_binary": binary,
"cli_version": cli or None,
"health_report_driver_version": report_v or None,
"version_mismatch": mismatch,
}
def _extract_health_report_from_result(result: Dict[str, Any]) -> Dict[str, Any]:
"""Pull a schema_version=1 report out of an MCP tools/call result.
@ -649,13 +700,28 @@ def _drive_health_report_or_fallback(
)
def _print_text_report(report: Dict[str, Any], color: bool) -> None:
def _print_text_report(
report: Dict[str, Any],
color: bool,
*,
identity: Optional[Dict[str, Any]] = None,
) -> None:
"""Render the report in the same style as `cua-driver call health_report`
would (one line per check + a summary footer)."""
would (one line per check + a summary footer).
When *identity* is provided (resolved binary + ``--version``), the header
prefers the CLI version if health_report's ``driver_version`` disagrees,
and a short identity block is printed under the header.
"""
schema = report.get("schema_version", "?")
platform = report.get("platform", "?")
driver_v = report.get("driver_version", "?")
report_v = report.get("driver_version", "?")
overall = report.get("overall", "?")
identity = identity or {}
cli_v = identity.get("cli_version") or ""
mismatch = bool(identity.get("version_mismatch"))
# Prefer the binary's own --version when health_report is wrong/stale.
header_v = cli_v or report_v
header_glyph = _OVERALL_GLYPH.get(overall, "")
@ -673,9 +739,32 @@ def _print_text_report(report: Dict[str, Any], color: bool) -> None:
col_for = ""
print(
f"{header_glyph} cua-driver {driver_v} on {platform}"
f"{header_glyph} cua-driver {header_v} on {platform}"
f"{col_for}{overall}{col_reset}"
)
if identity.get("resolved_binary"):
print(f" {col_dim}binary: {identity['resolved_binary']}{col_reset}")
if cli_v and report_v and str(report_v) not in str(cli_v) and str(cli_v) not in str(report_v):
# Only annotate when the free-form strings clearly differ.
print(
f" {col_dim}--version: {cli_v}{col_reset}"
)
print(
f" {col_dim}health_report.driver_version: {report_v}{col_reset}"
)
elif cli_v and not mismatch:
# Still show the resolved path; version already matches header.
pass
if mismatch:
warn = col_yellow if color else ""
print(
f" {warn}⚠️ version mismatch: health_report says {report_v!r} "
f"but binary --version is {cli_v!r}{col_reset}"
)
print(
f" {col_dim}→ trust --version / packages/current for debugging; "
f"health_report's binary_version check can lag on Windows{col_reset}"
)
for check in report.get("checks", []):
name = check.get("name", "?")
@ -749,13 +838,20 @@ def run_doctor(
print(f"cua-driver health_report failed: {e}", file=sys.stderr)
return 2
identity = _build_identity(binary, report)
if json_output:
json.dump(report, sys.stdout, indent=2, sort_keys=True)
# Additive envelope: preserve the upstream health_report keys and
# attach Hermes identity under hermes_identity so existing parsers
# that only read overall/checks keep working.
payload = dict(report)
payload["hermes_identity"] = identity
json.dump(payload, sys.stdout, indent=2, sort_keys=True)
sys.stdout.write("\n")
else:
if color is None:
color = sys.stdout.isatty()
_print_text_report(report, color=bool(color))
_print_text_report(report, color=bool(color), identity=identity)
overall = report.get("overall")
if overall in ("degraded", "failed"):