feat(automation): add ansible ping execution

PROGRESS.md

@@ -2894,6 +2894,7 @@ Notification data hygiene is complete. Suggested next steps:
 ### Phase 5 — Infrastructure Tooling
 - [x] Custom script runner — run arbitrary scripts on hosts/groups with streaming output
 - [x] Service management — start/stop/restart/status with live service autocomplete
+- [x] Ansible automation MVP — feature-gated SSH private-key credential profiles plus host/group `ansible_ping` task runs via the optional `ansible-api` container
 - [x] Interactive terminal — WebSocket PTY terminal with persistent panel, tabs, per-user auth
 - [x] Chart zoom and smart bucketing — click-drag zoom, adaptive time_bucket, reusable chart components
 - [ ] Jenkins plugin bundler (port existing)

apps/ansible-api/server.py

@@ -3,6 +3,9 @@
 import json
 import os
 import subprocess
+import tempfile
+import time
+from dataclasses import dataclass
 from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 from typing import Any
 
@@ -36,7 +39,8 @@ def capabilities_payload() -> dict[str, Any]:
         "capabilities": {
             "health": True,
             "openapi": True,
-            "jobExecution": False,
+            "jobExecution": True,
+            "ansiblePing": True,
             "playbookExecution": False,
             "inventoryManagement": False,
             "credentialProfiles": False,
@@ -73,10 +77,169 @@ def openapi_payload() -> dict[str, Any]:
                     "responses": {"200": {"description": "OpenAPI JSON"}},
                 },
             },
+            "/api/v1/runs/ansible-ping": {
+                "post": {
+                    "summary": "Run the Ansible ping module against supplied SSH inventory",
+                    "responses": {
+                        "200": {"description": "Per-host Ansible ping result"},
+                        "400": {"description": "Invalid request"},
+                    },
+                },
+            },
         },
     }
 
 
+@dataclass
+class CommandResult:
+    returncode: int
+    stdout: str
+    stderr: str
+    elapsedMs: int
+
+
+def _private_key_is_valid(value: str) -> bool:
+    stripped = value.strip()
+    return stripped.startswith("-----BEGIN ") and " PRIVATE KEY-----" in stripped and "-----END " in stripped
+
+
+def validate_ansible_ping_request(payload: dict[str, Any]) -> dict[str, Any]:
+    hosts = payload.get("hosts")
+    credential = payload.get("credential")
+    if not isinstance(hosts, list) or len(hosts) == 0:
+        raise ValueError("hosts must be a non-empty list")
+    if len(hosts) > 100:
+        raise ValueError("hosts must contain 100 entries or fewer")
+    if not isinstance(credential, dict):
+        raise ValueError("credential is required")
+
+    username = credential.get("username")
+    private_key = credential.get("privateKey")
+    if not isinstance(username, str) or not username.strip():
+        raise ValueError("credential.username is required")
+    if not isinstance(private_key, str) or not _private_key_is_valid(private_key):
+        raise ValueError("credential.privateKey must be an SSH private key")
+
+    cleaned_hosts: list[dict[str, Any]] = []
+    for host in hosts:
+        if not isinstance(host, dict):
+            raise ValueError("each host must be an object")
+        host_id = host.get("id")
+        name = host.get("name")
+        address = host.get("address")
+        port = host.get("port", 22)
+        if not isinstance(host_id, str) or not host_id:
+            raise ValueError("host.id is required")
+        if not isinstance(name, str) or not name:
+            raise ValueError("host.name is required")
+        if not isinstance(address, str) or not address:
+            raise ValueError("host.address is required")
+        if not isinstance(port, int) or port < 1 or port > 65535:
+            raise ValueError("host.port must be between 1 and 65535")
+        cleaned_hosts.append({"id": host_id, "name": name, "address": address, "port": port})
+
+    return {
+        "credential": {"username": username.strip(), "privateKey": private_key},
+        "hosts": cleaned_hosts,
+    }
+
+
+def inventory_text(hosts: list[dict[str, Any]], username: str) -> str:
+    lines = ["[ct_ops_targets]"]
+    for host in hosts:
+        lines.append(
+            f"{host['id']} ansible_host={host['address']} ansible_user={username} ansible_port={host['port']}"
+        )
+    return "\n".join(lines) + "\n"
+
+
+def run_ansible_ping_command(payload: dict[str, Any]) -> CommandResult:
+    start = time.monotonic()
+    credential = payload["credential"]
+    hosts = payload["hosts"]
+    with tempfile.TemporaryDirectory(prefix="ctops-ansible-") as tmpdir:
+        key_path = os.path.join(tmpdir, "id_key")
+        inventory_path = os.path.join(tmpdir, "inventory.ini")
+        with open(key_path, "w", encoding="utf-8") as key_file:
+            key_file.write(credential["privateKey"].strip() + "\n")
+        os.chmod(key_path, 0o600)
+        with open(inventory_path, "w", encoding="utf-8") as inventory_file:
+            inventory_file.write(inventory_text(hosts, credential["username"]))
+
+        env = {
+            **os.environ,
+            "ANSIBLE_HOST_KEY_CHECKING": os.environ.get("ANSIBLE_HOST_KEY_CHECKING", "False"),
+            "ANSIBLE_STDOUT_CALLBACK": "json",
+        }
+        proc = subprocess.run(
+            [
+                "ansible",
+                "ct_ops_targets",
+                "-i",
+                inventory_path,
+                "--private-key",
+                key_path,
+                "-m",
+                "ping",
+            ],
+            text=True,
+            capture_output=True,
+            timeout=45,
+            env=env,
+        )
+
+    return CommandResult(
+        returncode=proc.returncode,
+        stdout=proc.stdout,
+        stderr=proc.stderr,
+        elapsedMs=int((time.monotonic() - start) * 1000),
+    )
+
+
+def _host_succeeded_from_json(stdout: str, host_id: str) -> bool | None:
+    try:
+        data = json.loads(stdout)
+    except Exception:
+        return None
+    plays = data.get("plays")
+    if not isinstance(plays, list):
+        return None
+    for play in plays:
+        for task in play.get("tasks", []) if isinstance(play, dict) else []:
+            hosts = task.get("hosts") if isinstance(task, dict) else None
+            if not isinstance(hosts, dict) or host_id not in hosts:
+                continue
+            result = hosts[host_id]
+            if isinstance(result, dict):
+                return bool(result.get("ping") == "pong" or result.get("ok", 0)) and not bool(result.get("failed", False))
+    return None
+
+
+def run_ansible_ping(payload: dict[str, Any]) -> dict[str, Any]:
+    request = validate_ansible_ping_request(payload)
+    command = run_ansible_ping_command(request)
+    results = []
+
+    for host in request["hosts"]:
+        succeeded = _host_succeeded_from_json(command.stdout, host["id"])
+        if succeeded is None:
+            succeeded = command.returncode == 0
+        results.append({
+            "id": host["id"],
+            "name": host["name"],
+            "status": "success" if succeeded else "failed",
+            "exitCode": 0 if succeeded else (command.returncode or 1),
+            "stdout": command.stdout,
+            "stderr": command.stderr,
+        })
+
+    return {
+        "ok": all(host["status"] == "success" for host in results),
+        "elapsedMs": command.elapsedMs,
+        "hosts": results,
+    }
+
+
 class Handler(BaseHTTPRequestHandler):
     def do_GET(self) -> None:
         if self.path == "/healthz":
@@ -91,6 +254,25 @@ def do_GET(self) -> None:
 
         self.write_json({"error": "not_found"}, status=404)
 
+    def do_POST(self) -> None:
+        if self.path == "/api/v1/runs/ansible-ping":
+            try:
+                length = int(self.headers.get("content-length", "0"))
+                body = self.rfile.read(min(length, 1_000_000))
+                payload = json.loads(body.decode("utf-8"))
+                if not isinstance(payload, dict):
+                    raise ValueError("request body must be an object")
+                self.write_json(run_ansible_ping(payload))
+            except ValueError as err:
+                self.write_json({"error": str(err)}, status=400)
+            except subprocess.TimeoutExpired:
+                self.write_json({"error": "ansible ping timed out"}, status=504)
+            except Exception:
+                self.write_json({"error": "ansible ping failed"}, status=500)
+            return
+
+        self.write_json({"error": "not_found"}, status=404)
+
     def log_message(self, format: str, *args: Any) -> None:
         return
 

apps/ansible-api/tests/test_contract.py

@@ -5,6 +5,14 @@
 import server
 
 
+def private_key_block() -> str:
+    return "\n".join([
+        f"{'-' * 5}BEGIN OPENSSH PRIVATE KEY{'-' * 5}",
+        "fixture-key-body",
+        f"{'-' * 5}END OPENSSH PRIVATE KEY{'-' * 5}",
+    ])
+
+
 class AnsibleApiContractTests(unittest.TestCase):
     @mock.patch("server.ansible_version", return_value="2.19.0")
     def test_health_payload_reports_ansible_provider(self, _version):
@@ -18,14 +26,49 @@ def test_capabilities_mark_job_execution_unimplemented(self):
         capabilities = server.capabilities_payload()
         self.assertEqual(capabilities["provider"], "ansible")
         self.assertTrue(capabilities["capabilities"]["health"])
-        self.assertFalse(capabilities["capabilities"]["jobExecution"])
+        self.assertTrue(capabilities["capabilities"]["jobExecution"])
+        self.assertTrue(capabilities["capabilities"]["ansiblePing"])
         self.assertFalse(capabilities["capabilities"]["playbookExecution"])
 
     def test_openapi_payload_is_json_serialisable(self):
         payload = server.openapi_payload()
         self.assertEqual(payload["openapi"], "3.1.0")
+        self.assertIn("/api/v1/runs/ansible-ping", payload["paths"])
         json.dumps(payload)
 
+    def test_validate_ansible_ping_request_rejects_missing_hosts(self):
+        with self.assertRaisesRegex(ValueError, "hosts"):
+            server.validate_ansible_ping_request({
+                "credential": {
+                    "username": "deploy",
+                    "privateKey": private_key_block(),
+                },
+                "hosts": [],
+            })
+
+    @mock.patch("server.run_ansible_ping_command")
+    def test_run_ansible_ping_shapes_per_host_results(self, run_command):
+        run_command.return_value = server.CommandResult(
+            returncode=0,
+            stdout='{"plays":[{"tasks":[{"hosts":{"host-1":{"ok":1,"failed":0}}}]}]}',
+            stderr="",
+            elapsedMs=25,
+        )
+
+        payload = server.run_ansible_ping({
+            "credential": {
+                "username": "deploy",
+                "privateKey": private_key_block(),
+            },
+            "hosts": [{"id": "host-1", "name": "host-1", "address": "10.0.0.10", "port": 22}],
+        })
+
+        self.assertEqual(payload["ok"], True)
+        self.assertEqual(payload["hosts"][0]["id"], "host-1")
+        self.assertEqual(payload["hosts"][0]["status"], "success")
+        self.assertEqual(payload["hosts"][0]["exitCode"], 0)
+        self.assertEqual(payload["elapsedMs"], 25)
+
 
 if __name__ == "__main__":
     unittest.main()

apps/docs/docs/features/feature-flags.md

@@ -40,3 +40,5 @@ CT-Ops does not start Docker containers from the web app. After enabling or disa
 ```
 
 The start script reads the database after migrations and starts the optional `ansible-api` Compose profile only when both the flag and provider are enabled.
+
+When the API is healthy, administrators can save encrypted SSH private-key credential profiles and run an Ansible ping task from host or host-group task views. CT-Ops stores task state and redacted output in its task history; the Ansible container only executes the requested ping operation.