add target OS info command

Author: marcofortina

docs/reference/cli.md

@@ -104,6 +104,20 @@ automax runs show <run-id> --server web01
 automax runs show <run-id> --json
 ```
 
+
+## OS information
+
+Detect operating-system release facts for inventory targets before capability
+planning or dependency installation:
+
+```bash
+automax os info --inventory inventory.yaml
+automax os info --inventory inventory.yaml --limit web --format=json
+```
+
+The output includes `ID`, `ID_LIKE`, pretty release name, version, codename,
+normalized family (`debian`, `rhel` or `unknown`) and package manager.
+
 ## Inventory inspection
 
 Show the inventory targets selected by a specific job:

src/automax/cli/cli.py

@@ -174,6 +174,47 @@ def _echo_capability_install_payload(payload: Dict[str, Any], output_format: str
                 click.echo(f"    {line}")
 
 
+def _echo_os_info_payload(payload: Dict[str, Any], output_format: str) -> None:
+    if output_format == "json":
+        click.echo(json.dumps(payload, indent=2, sort_keys=True))
+        return
+    click.echo(f"Inventory: {payload['inventory']}")
+    click.echo(f"Targets: {payload['target_count']}")
+    for target in payload["targets"]:
+        os_info = target["os"]
+        pretty = os_info.get("pretty_name") or os_info.get("name") or os_info.get("id") or "unknown"
+        version = os_info.get("version_id") or os_info.get("version") or "-"
+        codename = os_info.get("version_codename") or "-"
+        id_like = ",".join(os_info.get("id_like") or []) or "-"
+        user = target.get("user") or "-"
+        click.echo(
+            f"Target {target['target']} {target['host']}:{target['port']} "
+            f"user={user} os={pretty} family={os_info.get('family', 'unknown')} "
+            f"id={os_info.get('id', 'unknown')} version={version} codename={codename} "
+            f"id_like={id_like} pkg={os_info.get('package_manager', 'unknown')}"
+        )
+
+
+def _os_info_payload(
+    *,
+    inventory_path: str,
+    vars_path: str | None,
+    secrets_path: str | None,
+    cli_vars: tuple[str, ...],
+    limit: tuple[str, ...],
+    exclude: tuple[str, ...],
+    plugin_path: tuple[str, ...],
+) -> Dict[str, Any]:
+    return _engine(plugin_path).os_info_inventory(
+        inventory_path=inventory_path,
+        vars_path=vars_path,
+        secrets_path=secrets_path,
+        limit=_split_selectors(limit),
+        exclude=_split_selectors(exclude),
+        cli_vars=_parse_vars(cli_vars),
+    )
+
+
 @click.group(context_settings={"help_option_names": ["-h", "--help"]})
 @click.version_option(version=__version__, prog_name="Automax")
 def cli() -> None:
@@ -965,6 +1006,53 @@ def check_secrets(
 
 
 
+@cli.group()
+def os() -> None:
+    """Inspect target operating-system facts."""
+
+
+@os.command("info")
+@click.option("--inventory", "inventory_path", required=True, type=click.Path(exists=True), help="External inventory YAML path.")
+@click.option("--vars", "vars_path", type=click.Path(exists=True), help="External variables YAML path.")
+@click.option("--secrets", "secrets_path", type=click.Path(exists=True), help="External secrets YAML path.")
+@click.option("--var", "cli_vars", multiple=True, help="Override variable, format KEY=VALUE.")
+@click.option("--limit", multiple=True, help="Limit targets. Accepts server, group or group:name.")
+@click.option("--exclude", multiple=True, help="Exclude targets. Accepts server, group or group:name.")
+@click.option("--plugin-path", multiple=True, help="External plugin file or directory.")
+@click.option(
+    "--format",
+    "output_format",
+    type=click.Choice(["text", "json"]),
+    default="text",
+    show_default=True,
+    help="Output format.",
+)
+def os_info(
+    inventory_path: str,
+    vars_path: str | None,
+    secrets_path: str | None,
+    cli_vars: tuple[str, ...],
+    limit: tuple[str, ...],
+    exclude: tuple[str, ...],
+    plugin_path: tuple[str, ...],
+    output_format: str,
+) -> None:
+    """Detect OS release facts for selected inventory targets."""
+    try:
+        payload = _os_info_payload(
+            inventory_path=inventory_path,
+            vars_path=vars_path,
+            secrets_path=secrets_path,
+            cli_vars=cli_vars,
+            limit=limit,
+            exclude=exclude,
+            plugin_path=plugin_path,
+        )
+    except (AutomaxError, ValueError, RuntimeError) as exc:
+        raise click.ClickException(str(exc)) from exc
+    _echo_os_info_payload(payload, output_format)
+
+
 @cli.group()
 def capabilities() -> None:
     """Inspect job-scoped remote capability requirements."""

src/automax/core/engine.py

@@ -787,6 +787,48 @@ def manual_commands_job(
             "nodes": rows,
         }
 
+    def os_info_inventory(
+        self,
+        *,
+        inventory_path: str,
+        vars_path: str | None = None,
+        secrets_path: str | None = None,
+        limit: Iterable[str] = (),
+        exclude: Iterable[str] = (),
+        cli_vars: Optional[Dict[str, Any]] = None,
+    ) -> Dict[str, Any]:
+        """Detect and report operating-system facts for inventory targets."""
+        vars_document = load_yaml_file(vars_path, required=False) if vars_path else {}
+        secrets_document = load_yaml_file(secrets_path, required=False) if secrets_path else {}
+        secrets = self.secret_manager.resolve_all(
+            secrets_document,
+            base_dir=self._path_parent(secrets_path),
+        )
+        variables = self._merge_variables(vars_document, cli_vars or {})
+        context = {"vars": variables, "secrets": secrets}
+        inventory_document = load_inventory_document(inventory_path, context)
+        inventory = Inventory(inventory_document, context)
+        targets = inventory.select("all", limit=list(limit), exclude=list(exclude))
+        os_by_target = self._detect_os_for_targets(targets, secrets)
+        rows = []
+        for target in targets:
+            os_info = os_by_target[target.name]
+            rows.append(
+                {
+                    "target": target.name,
+                    "host": target.host,
+                    "port": target.port,
+                    "user": target.user,
+                    "os": self._os_to_mapping(os_info),
+                }
+            )
+        return {
+            "mode": "os-info",
+            "inventory": str(Path(inventory_path).expanduser()),
+            "target_count": len(rows),
+            "targets": rows,
+        }
+
     def capability_requirements_job(
         self,
         *,
@@ -1233,8 +1275,13 @@ def _plan_requires_capability_preflight(
 
     def _detect_os_for_plan(self, plan: List[Dict[str, Any]], secrets: Dict[str, Any]) -> Dict[str, TargetOS]:
         """Detect target OS once per target before capability preflight or install."""
+        targets = {item["target"].name: item["target"] for item in plan}.values()
+        return self._detect_os_for_targets(targets, secrets)
+
+    def _detect_os_for_targets(self, targets: Iterable[Target], secrets: Dict[str, Any]) -> Dict[str, TargetOS]:
+        """Detect operating-system facts once for each target."""
         detected: Dict[str, TargetOS] = {}
-        for target in {item["target"].name: item["target"] for item in plan}.values():
+        for target in targets:
             with self.ssh_manager.connect(target) as client:
                 _stdin, stdout, stderr = client.exec_command(DETECT_OS_COMMAND)
                 rc = stdout.channel.recv_exit_status()
@@ -1251,7 +1298,11 @@ def _os_to_mapping(os_info: TargetOS) -> Dict[str, Any]:
         return {
             "id": os_info.id,
             "id_like": list(os_info.id_like),
+            "name": os_info.name,
+            "pretty_name": os_info.pretty_name,
+            "version": os_info.version,
             "version_id": os_info.version_id,
+            "version_codename": os_info.version_codename,
             "family": os_info.family,
             "package_manager": os_info.package_manager,
         }

src/automax/core/os_detect.py

@@ -14,7 +14,11 @@ class TargetOS:
 
     id: str = "unknown"
     id_like: tuple[str, ...] = ()
+    name: str = ""
+    pretty_name: str = ""
+    version: str = ""
     version_id: str = ""
+    version_codename: str = ""
     family: str = "unknown"
     package_manager: str = "unknown"
 
@@ -52,7 +56,11 @@ def parse_os_release(text: str) -> TargetOS:
     return TargetOS(
         id=os_id,
         id_like=id_like,
+        name=values.get("NAME", ""),
+        pretty_name=values.get("PRETTY_NAME", ""),
+        version=values.get("VERSION", ""),
         version_id=values.get("VERSION_ID", ""),
+        version_codename=values.get("VERSION_CODENAME", values.get("UBUNTU_CODENAME", "")),
         family=family,
         package_manager=package_manager,
     )

tests/test_next_engine.py

@@ -4656,3 +4656,71 @@ def fake_install(*, target, os_family, packages, sudo_password):
     assert installs == [("node", "debian", ["acl", "zip"], None)]
     assert payload["targets"][0]["missing_tools"] == ["setfacl", "zip"]
     assert payload["targets"][0]["packages"] == ["acl", "zip"]
+
+
+def test_os_info_inventory_reports_release_details(tmp_path: Path, monkeypatch):
+    from automax.core.os_detect import TargetOS
+
+    inventory = write(
+        tmp_path / "inventory.yaml",
+        """
+servers:
+  node:
+    host: 127.0.0.1
+    ssh:
+      user: ubuntu
+""",
+    )
+    engine = AutomaxEngine()
+    monkeypatch.setattr(
+        engine,
+        "_detect_os_for_targets",
+        lambda targets, secrets: {
+            "node": TargetOS(
+                id="ubuntu",
+                id_like=("debian",),
+                name="Ubuntu",
+                pretty_name="Ubuntu 24.04.2 LTS",
+                version="24.04.2 LTS (Noble Numbat)",
+                version_id="24.04",
+                version_codename="noble",
+                family="debian",
+                package_manager="apt",
+            )
+        },
+    )
+
+    payload = engine.os_info_inventory(inventory_path=str(inventory))
+
+    assert payload["mode"] == "os-info"
+    assert payload["target_count"] == 1
+    target = payload["targets"][0]
+    assert target["target"] == "node"
+    assert target["user"] == "ubuntu"
+    assert target["os"]["pretty_name"] == "Ubuntu 24.04.2 LTS"
+    assert target["os"]["version_codename"] == "noble"
+    assert target["os"]["family"] == "debian"
+
+
+def test_os_info_cli_json_output(tmp_path: Path, monkeypatch):
+    from automax.core.os_detect import TargetOS
+
+    inventory = write(tmp_path / "inventory.yaml", "servers:\n  node:\n    host: 127.0.0.1\n")
+    monkeypatch.setattr(
+        AutomaxEngine,
+        "_detect_os_for_targets",
+        lambda self, targets, secrets: {
+            "node": TargetOS(id="debian", id_like=(), pretty_name="Debian GNU/Linux 12", version_id="12", family="debian", package_manager="apt")
+        },
+    )
+
+    result = CliRunner().invoke(
+        cli,
+        ["os", "info", "--inventory", str(inventory), "--format", "json"],
+    )
+
+    assert result.exit_code == 0, result.output
+    payload = json.loads(result.output)
+    assert payload["mode"] == "os-info"
+    assert payload["targets"][0]["os"]["id"] == "debian"
+    assert payload["targets"][0]["os"]["pretty_name"] == "Debian GNU/Linux 12"