From d36c95b2af384faa8f0a06a07e06163cd5cdbcab Mon Sep 17 00:00:00 2001 From: binrogithub Date: Sun, 8 Feb 2026 17:58:52 +0300 Subject: [PATCH] Add doctor selection CLI and module --- hc_agent/__init__.py | 1 + hc_agent/cli/__init__.py | 1 + hc_agent/cli/cmd_doctor.py | 69 ++++++++++++++++++++++ hc_agent/doctor.py | 116 +++++++++++++++++++++++++++++++++++++ 4 files changed, 187 insertions(+) create mode 100644 hc_agent/__init__.py create mode 100644 hc_agent/cli/__init__.py create mode 100644 hc_agent/cli/cmd_doctor.py create mode 100644 hc_agent/doctor.py diff --git a/hc_agent/__init__.py b/hc_agent/__init__.py new file mode 100644 index 0000000000..720041397d --- /dev/null +++ b/hc_agent/__init__.py @@ -0,0 +1 @@ +"""Health check agent package.""" diff --git a/hc_agent/cli/__init__.py b/hc_agent/cli/__init__.py new file mode 100644 index 0000000000..5016d266c0 --- /dev/null +++ b/hc_agent/cli/__init__.py @@ -0,0 +1 @@ +"""CLI commands for the health check agent.""" diff --git a/hc_agent/cli/cmd_doctor.py b/hc_agent/cli/cmd_doctor.py new file mode 100644 index 0000000000..36971d9c6e --- /dev/null +++ b/hc_agent/cli/cmd_doctor.py @@ -0,0 +1,69 @@ +"""CLI entry point for the doctor selection logic.""" + +from __future__ import annotations + +import argparse +import json +import sys +from pathlib import Path +from typing import Any, Iterable, Mapping + +from hc_agent import doctor + + +def _load_resources(path: Path | None) -> Iterable[Mapping[str, Any]]: + if path is None: + payload = sys.stdin.read().strip() + if not payload: + return [] + return json.loads(payload) + return json.loads(path.read_text()) + + +def _list_resources(path: Path | None) -> Iterable[Mapping[str, Any]]: + return _load_resources(path) + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + description="Run doctor selection using read-only listing data.", + ) + parser.add_argument( + "--resources-file", + type=Path, + help="Path to JSON list of resources (defaults to stdin).", + ) + parser.add_argument( + "--explicit", + help="Explicit resource id or name to prioritize.", + ) + parser.add_argument( + "--default-tag", + default="default", + help="Tag name used to indicate default resources.", + ) + return parser + + +def main(argv: list[str] | None = None) -> int: + parser = build_parser() + args = parser.parse_args(argv) + + result = doctor.run_doctor( + lambda: _list_resources(args.resources_file), + explicit_input=args.explicit, + default_tag=args.default_tag, + ) + + output = { + "ctx_patch": result.ctx_patch, + "summary": result.summary, + } + json.dump(output, sys.stdout, indent=2, sort_keys=True) + sys.stdout.write("\n") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) + diff --git a/hc_agent/doctor.py b/hc_agent/doctor.py new file mode 100644 index 0000000000..1572a016af --- /dev/null +++ b/hc_agent/doctor.py @@ -0,0 +1,116 @@ +"""Doctor module for deterministic resource selection with read-only listings.""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any, Callable, Iterable, Mapping, Sequence + + +@dataclass(frozen=True) +class DoctorResult: + """Outcome from a doctor selection run.""" + + ctx_patch: dict[str, Any] + summary: str + + +def _normalize_tags(tags: Any) -> dict[str, Any]: + if tags is None: + return {} + if isinstance(tags, Mapping): + return dict(tags) + if isinstance(tags, Sequence) and not isinstance(tags, (str, bytes)): + return {str(tag): True for tag in tags} + return {str(tags): True} + + +def _is_tagged_default(tags: Any, default_tag: str) -> bool: + normalized = _normalize_tags(tags) + if default_tag in normalized: + value = normalized[default_tag] + if isinstance(value, str): + return value.strip().lower() in {"true", "yes", "1", "default"} + return bool(value) + return False + + +def _stable_sort_key(resource: Mapping[str, Any]) -> tuple[str, str]: + name = str(resource.get("name", "")) + resource_id = str(resource.get("id", "")) + return (name, resource_id) + + +def _select_candidate(resources: Iterable[Mapping[str, Any]]) -> Mapping[str, Any] | None: + sorted_resources = sorted(resources, key=_stable_sort_key) + return sorted_resources[0] if sorted_resources else None + + +def run_doctor( + list_resources: Callable[[], Iterable[Mapping[str, Any]]], + *, + explicit_input: str | None = None, + default_tag: str = "default", +) -> DoctorResult: + """Run doctor selection using read-only list_resources calls. + + Selection precedence: + 1. explicit_input (id or name match) + 2. resources tagged with default_tag + 3. stable tie-breaker (name, id) + """ + + resources = list(list_resources()) + summary_parts = [f"resources={len(resources)}"] + selected: Mapping[str, Any] | None = None + reason = "" + + if explicit_input: + matches = [ + resource + for resource in resources + if str(resource.get("id")) == explicit_input + or str(resource.get("name")) == explicit_input + ] + selected = _select_candidate(matches) + if selected: + reason = "explicit" + else: + summary_parts.append("explicit_input_not_found=true") + + if selected is None: + tagged = [ + resource + for resource in resources + if _is_tagged_default(resource.get("tags"), default_tag) + ] + selected = _select_candidate(tagged) + if selected: + reason = "tagged_default" + + if selected is None: + selected = _select_candidate(resources) + if selected: + reason = "stable_tiebreak" + + ctx_patch = { + "doctor": { + "selected": { + "id": None if selected is None else selected.get("id"), + "name": None if selected is None else selected.get("name"), + "reason": reason or None, + }, + "default_tag": default_tag, + } + } + + if selected is None: + summary_parts.append("selected=none") + else: + summary_parts.append( + f"selected={selected.get('name', selected.get('id', 'unknown'))}" + ) + summary_parts.append(f"reason={reason}") + + summary = "; ".join(summary_parts) + return DoctorResult(ctx_patch=ctx_patch, summary=summary) +