fix(doctor): report Kanban worker tools as runtime-gated

This commit is contained in:
suncokret12 2026-05-01 12:31:14 +02:00 committed by Teknium
parent f0b95cc93d
commit eda326df16
2 changed files with 66 additions and 5 deletions

View file

@ -107,15 +107,35 @@ def _honcho_is_configured_for_doctor() -> bool:
return False
def _is_kanban_worker_env_gate(item: dict) -> bool:
"""Return True when Kanban is unavailable only because this is not a worker process."""
if item.get("name") != "kanban":
return False
if os.environ.get("HERMES_KANBAN_TASK"):
return False
tools = item.get("tools") or []
return bool(tools) and all(str(tool).startswith("kanban_") for tool in tools)
def _doctor_tool_availability_detail(toolset: str) -> str:
"""Optional explanatory suffix for toolsets whose doctor status needs context."""
if toolset == "kanban" and not os.environ.get("HERMES_KANBAN_TASK"):
return "(runtime-gated; loaded only for dispatcher-spawned workers)"
return ""
def _apply_doctor_tool_availability_overrides(available: list[str], unavailable: list[dict]) -> tuple[list[str], list[dict]]:
"""Adjust runtime-gated tool availability for doctor diagnostics."""
if not _honcho_is_configured_for_doctor():
return available, unavailable
updated_available = list(available)
updated_unavailable = []
for item in unavailable:
if item.get("name") == "honcho":
name = item.get("name")
if _is_kanban_worker_env_gate(item):
if "kanban" not in updated_available:
updated_available.append("kanban")
continue
if name == "honcho" and _honcho_is_configured_for_doctor():
if "honcho" not in updated_available:
updated_available.append("honcho")
continue
@ -1278,7 +1298,7 @@ def run_doctor(args):
for tid in available:
info = TOOLSET_REQUIREMENTS.get(tid, {})
check_ok(info.get("name", tid))
check_ok(info.get("name", tid), _doctor_tool_availability_detail(tid))
for item in unavailable:
env_vars = item.get("missing_vars") or item.get("env_vars") or []

View file

@ -126,6 +126,47 @@ class TestDoctorToolAvailabilityOverrides:
assert available == []
assert unavailable == [honcho_entry]
def test_marks_kanban_available_only_when_missing_worker_env_gate(self, monkeypatch):
monkeypatch.setattr(doctor, "_honcho_is_configured_for_doctor", lambda: False)
monkeypatch.delenv("HERMES_KANBAN_TASK", raising=False)
available, unavailable = doctor._apply_doctor_tool_availability_overrides(
[],
[{"name": "kanban", "env_vars": [], "tools": ["kanban_show"]}],
)
assert available == ["kanban"]
assert unavailable == []
def test_leaves_kanban_unavailable_when_worker_env_is_set(self, monkeypatch):
monkeypatch.setenv("HERMES_KANBAN_TASK", "probe")
kanban_entry = {"name": "kanban", "env_vars": [], "tools": ["kanban_show"]}
available, unavailable = doctor._apply_doctor_tool_availability_overrides(
[],
[kanban_entry],
)
assert available == []
assert unavailable == [kanban_entry]
def test_leaves_non_worker_kanban_failure_unavailable(self, monkeypatch):
monkeypatch.delenv("HERMES_KANBAN_TASK", raising=False)
kanban_entry = {"name": "kanban", "env_vars": [], "tools": ["kanban_show", "not_a_kanban_tool"]}
available, unavailable = doctor._apply_doctor_tool_availability_overrides(
[],
[kanban_entry],
)
assert available == []
assert unavailable == [kanban_entry]
def test_kanban_doctor_detail_explains_worker_gate(self, monkeypatch):
monkeypatch.delenv("HERMES_KANBAN_TASK", raising=False)
assert doctor._doctor_tool_availability_detail("kanban") == "(runtime-gated; loaded only for dispatcher-spawned workers)"
class TestHonchoDoctorConfigDetection:
def test_reports_configured_when_enabled_with_api_key(self, monkeypatch):