Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions hc_agent/__init__.py
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
"""Health check agent package."""
1 change: 1 addition & 0 deletions hc_agent/cli/__init__.py
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
"""CLI commands for the health check agent."""
69 changes: 69 additions & 0 deletions hc_agent/cli/cmd_doctor.py
Original file line number Diff line number Diff line change
@@ -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())

116 changes: 116 additions & 0 deletions hc_agent/doctor.py
Original file line number Diff line number Diff line change
@@ -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)