fix(ansible): write DB inventory as static YAML, not dynamic --list JSON

generate_inventory() emits Ansible's dynamic --list shape (all.hosts is a
LIST, vars under _meta) — valid only as an executable inventory script's
stdout. We were writing it to a static file and passing -i, so Ansible's yaml
plugin rejected it ("Invalid 'hosts' entry for 'all' group, requires a
dictionary, found ...list...") and fell back to implicit localhost → "no hosts
matched". Affected every steward:* scope run; surfaced on the first real
provision.

- New inventory_to_yaml(inv): convert the --list dict → a valid static YAML
  inventory (all.hosts dict keyed by host, groups under all.children, group
  vars preserved, injected per-host vars like steward_token retained).
- Wire it into runner.trigger_run, host_agent deploy + provision.
- executor writes the file as inventory.yml so the yaml plugin's extension
  check reliably claims it.
- generate_inventory unchanged (still the --list dict); conversion happens at
  write time. Unit tests added.

Scribe issue #885.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

plugins/host_agent/routes.py

@@ -707,9 +707,9 @@ async def deploy_via_ansible():
     Uses the core "ansible.run_playbook" capability (no hard import of the
     runner) and injects a freshly-minted per-host token as an inventory hostvar.
     """
-    import json as _json
     from steward.core.capabilities import has_capability, invoke_capability
-    from steward.ansible.inventory_gen import fetch_scope_targets, generate_inventory
+    from steward.ansible.inventory_gen import (
+        fetch_scope_targets, generate_inventory, inventory_to_yaml)
     from steward.ansible.sources import BUILTIN_SOURCE_NAME
 
     if not has_capability("ansible.run_playbook"):
@@ -742,7 +742,7 @@ async def deploy_via_ansible():
             hv = inv["_meta"]["hostvars"].setdefault(name, {})
             hv["steward_url"] = url
             hv["steward_token"] = tok
-        inventory_content = _json.dumps(inv)
+        inventory_content = inventory_to_yaml(inv)
         await db.commit()
 
     actor_role = UserRole(session.get("user_role", "viewer"))
@@ -771,9 +771,9 @@ async def provision_via_ansible():
     The bootstrap password is passed to the runner as a connection override and
     is NOT persisted on the run row.
     """
-    import json as _json
     from steward.core.capabilities import has_capability, invoke_capability
-    from steward.ansible.inventory_gen import fetch_scope_targets, generate_inventory
+    from steward.ansible.inventory_gen import (
+        fetch_scope_targets, generate_inventory, inventory_to_yaml)
     from steward.ansible.sources import BUILTIN_SOURCE_NAME
 
     if not has_capability("ansible.run_playbook"):
@@ -823,7 +823,7 @@ async def provision_via_ansible():
             # Ansible would shlex-split on whitespace.
             hv["steward_pubkey"] = pubkey
             hv["steward_user"] = steward_user
-        inventory_content = _json.dumps(inv)
+        inventory_content = inventory_to_yaml(inv)
         await db.commit()
 
     actor_role = UserRole(session.get("user_role", "viewer"))

steward/ansible/executor.py

@@ -375,7 +375,9 @@ async def start_run(
         os.chmod(tmpdir, 0o700)
         try:
             if inventory_content:
-                inv_target = os.path.join(tmpdir, "inventory")
+                # .yml so Ansible's yaml inventory plugin reliably claims it
+                # (the plugin's verify_file checks the extension).
+                inv_target = os.path.join(tmpdir, "inventory.yml")
                 with open(inv_target, "w", encoding="utf-8") as invf:
                     invf.write(inventory_content)
             else:

steward/ansible/inventory_gen.py

@@ -37,6 +37,33 @@ def generate_inventory(targets: list) -> dict:
     return inv
 
 
+def inventory_to_yaml(inv: dict) -> str:
+    """Convert the ``--list`` dict from generate_inventory into a STATIC YAML
+    inventory string.
+
+    The ``--list`` shape (``all.hosts`` is a *list*, vars under ``_meta``) is only
+    valid as the stdout of an executable dynamic-inventory script. When written to
+    a plain file and passed via ``-i``, Ansible's yaml plugin requires
+    ``all.hosts`` to be a *dict* keyed by hostname — otherwise it fails to parse
+    and silently falls back to implicit localhost. This emits that valid form.
+    """
+    import yaml
+    hostvars = (inv.get("_meta") or {}).get("hostvars") or {}
+    all_hosts = (inv.get("all") or {}).get("hosts") or []
+    root: dict = {"hosts": {h: (hostvars.get(h) or {}) for h in all_hosts}}
+    children: dict = {}
+    for key, val in inv.items():
+        if key in ("all", "_meta") or not isinstance(val, dict):
+            continue
+        grp: dict = {"hosts": {h: {} for h in (val.get("hosts") or [])}}
+        if val.get("vars"):
+            grp["vars"] = val["vars"]
+        children[key] = grp
+    if children:
+        root["children"] = children
+    return yaml.safe_dump({"all": root}, default_flow_style=False, sort_keys=True)
+
+
 async def fetch_scope_targets(db, scope: str) -> list:
     """Query AnsibleTarget objects for a given inventory scope string.
 

steward/ansible/runner.py

@@ -7,12 +7,13 @@ manual, alert-triggered, and scheduled runs all take the exact same path.
 from __future__ import annotations
 
 import asyncio
-import json
 import uuid
 from datetime import datetime, timezone
 
 from steward.ansible import executor, sources as src_module
-from steward.ansible.inventory_gen import fetch_scope_targets, generate_inventory
+from steward.ansible.inventory_gen import (
+    fetch_scope_targets, generate_inventory, inventory_to_yaml,
+)
 from steward.models.ansible import AnsibleRun, AnsibleRunStatus
 
 
@@ -53,7 +54,7 @@ async def trigger_run(
     elif inventory_scope.startswith("steward:"):
         async with app.db_sessionmaker() as db:
             targets = await fetch_scope_targets(db, inventory_scope)
-        inventory_content = json.dumps(generate_inventory(targets))
+        inventory_content = inventory_to_yaml(generate_inventory(targets))
     elif inventory_scope.startswith("repo:"):
         parts = inventory_scope.split(":", 2)
         inventory_path = parts[2] if len(parts) == 3 else ""

tests/core/test_inventory_gen.py

@@ -3,7 +3,10 @@
 Uses SimpleNamespace duck-typed objects so no DB or SQLAlchemy is required.
 """
 from types import SimpleNamespace
-from steward.ansible.inventory_gen import generate_inventory
+
+import yaml
+
+from steward.ansible.inventory_gen import generate_inventory, inventory_to_yaml
 
 
 def _group(name, ansible_vars=None):
@@ -91,3 +94,37 @@ def test_target_in_multiple_groups():
     assert "web-prod-01" in result["prod"]["hosts"]
     assert result["_meta"]["hostvars"]["web-prod-01"]["role"] == "web"
     assert result["_meta"]["hostvars"]["web-prod-01"]["env"] == "production"
+
+
+# ── inventory_to_yaml: STATIC inventory (hosts is a dict, not a list) ──────────
+
+def test_yaml_hosts_is_dict_with_hostvars():
+    target = _target("web1", "192.168.1.10", {"ansible_user": "ubuntu"})
+    parsed = yaml.safe_load(inventory_to_yaml(generate_inventory([target])))
+    # The bug was a LIST here; a static YAML inventory requires a dict keyed by host.
+    assert isinstance(parsed["all"]["hosts"], dict)
+    assert parsed["all"]["hosts"]["web1"]["ansible_host"] == "192.168.1.10"
+    assert parsed["all"]["hosts"]["web1"]["ansible_user"] == "ubuntu"
+
+
+def test_yaml_groups_under_children():
+    grp = _group("db", {"db_port": 5432})
+    t1 = _target("db1", "10.0.0.1", groups=[grp])
+    t2 = _target("db2", "10.0.0.2", groups=[grp])
+    parsed = yaml.safe_load(inventory_to_yaml(generate_inventory([t1, t2])))
+    assert set(parsed["all"]["hosts"]) == {"db1", "db2"}
+    assert set(parsed["all"]["children"]["db"]["hosts"]) == {"db1", "db2"}
+    assert parsed["all"]["children"]["db"]["vars"]["db_port"] == 5432
+
+
+def test_yaml_injected_hostvars_survive():
+    """Extra per-host vars injected after generate_inventory (e.g. steward_token)."""
+    inv = generate_inventory([_target("h1", "1.2.3.4")])
+    inv["_meta"]["hostvars"]["h1"]["steward_token"] = "tok with space"
+    parsed = yaml.safe_load(inventory_to_yaml(inv))
+    assert parsed["all"]["hosts"]["h1"]["steward_token"] == "tok with space"
+
+
+def test_yaml_empty_targets():
+    parsed = yaml.safe_load(inventory_to_yaml(generate_inventory([])))
+    assert parsed == {"all": {"hosts": {}}}