feat(ansible): in-app playbook authoring/editor

Adds create/edit/delete of playbooks from the UI (admin only), so a homelab
user without a git workflow can author automation in-app. A new always-present
writable local source "steward-local" (/data/ansible/playbooks, env-overridable,
created on first save) is editable alongside operator local-dir sources; the
bundled and git sources stay read-only (git is GitOps, clobbered on pull).

sources.py: write_playbook / delete_playbook (traversal-guarded, .yml/.yaml
only) + validate_playbook_yaml (YAML + play-list check) + is_editable_source.
routes.py: /playbooks/new, /edit, /save, /delete (admin). Browse gains a
"New playbook" button and per-playbook + view-page Edit/Delete for editable
sources. Plain textarea editor with save-time YAML validation. Unit tests for
write/delete/guard/validate.

Task #579 — completes milestone #37 (Ansible automation).

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

steward/ansible/routes.py

@@ -56,6 +56,7 @@ async def browse():
             "source": source,
             "playbooks": playbooks,
             "inventories": inventories,
+            "editable": src_module.is_editable_source(source),
         })
 
     async with current_app.db_sessionmaker() as db:
@@ -90,6 +91,7 @@ async def view_playbook(source_name: str, playbook_path: str):
         view_source=source_name,
         view_path=playbook_path,
         view_contents=contents,
+        view_editable=src_module.is_editable_source(source),
     )
 
 
@@ -354,3 +356,102 @@ async def run_schedule_now(schedule_id: str):
     if run:
         return redirect(url_for("ansible.run_detail", run_id=run.id))
     return redirect(url_for("ansible.schedules"))
+
+
+# ── In-app playbook authoring/editor (admin only) ─────────────────────────────
+
+def _editable_sources() -> list[dict]:
+    return [s for s in _get_sources() if src_module.is_editable_source(s)]
+
+
+def _editable_source(name: str) -> dict | None:
+    return next((s for s in _editable_sources() if s["name"] == name), None)
+
+
+@ansible_bp.get("/playbooks/new")
+@require_role(UserRole.admin)
+async def new_playbook():
+    sources = _editable_sources()
+    return await render_template(
+        "ansible/playbook_editor.html",
+        editable_sources=sources, editing=False,
+        source_name=(sources[0]["name"] if sources else ""),
+        playbook_path="", content=_NEW_PLAYBOOK_TEMPLATE, error=None,
+    )
+
+
+@ansible_bp.get("/playbooks/edit/<source_name>/<path:playbook_path>")
+@require_role(UserRole.admin)
+async def edit_playbook(source_name: str, playbook_path: str):
+    source = _editable_source(source_name)
+    if source is None:
+        return "Source is not editable", 404
+    content = src_module.read_playbook(source["path"], playbook_path)
+    if content is None:
+        return "Playbook not found", 404
+    return await render_template(
+        "ansible/playbook_editor.html",
+        editable_sources=_editable_sources(), editing=True,
+        source_name=source_name, playbook_path=playbook_path,
+        content=content, error=None,
+    )
+
+
+@ansible_bp.post("/playbooks/save")
+@require_role(UserRole.admin)
+async def save_playbook():
+    form = await request.form
+    source_name = (form.get("source_name", "") or "").strip()
+    playbook_path = (form.get("playbook_path", "") or "").strip()
+    content = form.get("content", "") or ""
+    editing = (form.get("editing", "") == "1")
+
+    source = _editable_source(source_name)
+    err = None
+    if source is None:
+        err = "Source is not editable"
+    elif not playbook_path:
+        err = "Filename is required"
+    else:
+        ok, verr = src_module.validate_playbook_yaml(content)
+        if not ok:
+            err = verr
+    if err is None:
+        ok, werr = src_module.write_playbook(source["path"], playbook_path, content)
+        if not ok:
+            err = werr
+
+    if err is not None:
+        return await render_template(
+            "ansible/playbook_editor.html",
+            editable_sources=_editable_sources(), editing=editing,
+            source_name=source_name, playbook_path=playbook_path,
+            content=content, error=err,
+        ), 400
+
+    return redirect(url_for(
+        "ansible.view_playbook", source_name=source_name, playbook_path=playbook_path))
+
+
+@ansible_bp.post("/playbooks/delete")
+@require_role(UserRole.admin)
+async def delete_playbook():
+    form = await request.form
+    source = _editable_source((form.get("source_name", "") or "").strip())
+    if source is None:
+        return "Source is not editable", 400
+    ok, err = src_module.delete_playbook(source["path"], (form.get("playbook_path", "") or "").strip())
+    if not ok:
+        return err, 400
+    return redirect(url_for("ansible.browse"))
+
+
+_NEW_PLAYBOOK_TEMPLATE = """\
+---
+- name: My playbook
+  hosts: all
+  gather_facts: false
+  tasks:
+    - name: Ping
+      ansible.builtin.ping:
+"""

steward/ansible/sources.py

@@ -14,6 +14,12 @@ INVENTORY_NAMES = {"hosts", "inventory", "inventory.yml", "inventory.ini"}
 # inside the app (maintenance tasks, host-agent install). Not operator-editable.
 BUILTIN_SOURCE_NAME = "steward-builtin"
 
+# Always-present, WRITABLE local source where the in-app editor saves playbooks,
+# so a homelab user with no git workflow can author automation. Lives on the
+# persistent /data volume; created lazily on first save.
+LOCAL_SOURCE_NAME = "steward-local"
+USER_PLAYBOOK_DIR = os.environ.get("STEWARD_PLAYBOOK_DIR", "/data/ansible/playbooks")
+
 
 def _builtin_source() -> dict:
     return {
@@ -27,6 +33,24 @@ def _builtin_source() -> dict:
     }
 
 
+def _local_source() -> dict:
+    return {
+        "name": LOCAL_SOURCE_NAME,
+        "type": "local",
+        "path": USER_PLAYBOOK_DIR,
+        "url": None,
+        "branch": "main",
+        "pull_interval_seconds": 0,
+        "http_token": "",
+    }
+
+
+def is_editable_source(source: dict) -> bool:
+    """Editable in the in-app editor: local dir sources except the read-only
+    bundled one. Git sources are GitOps (clobbered on pull) so not editable."""
+    return source.get("type") == "local" and source.get("name") != BUILTIN_SOURCE_NAME
+
+
 def get_sources(ansible_cfg: dict) -> list[dict]:
     """Return resolved source list from ansible config section.
 
@@ -49,8 +73,8 @@ def get_sources(ansible_cfg: dict) -> list[dict]:
     """
     sources = ansible_cfg.get("sources", [])
     cache_dir = ansible_cfg.get("cache_dir", "/var/cache/steward/ansible")
-    # Always expose the bundled first-party playbooks as a source.
-    result = [_builtin_source()]
+    # Always expose the bundled (read-only) + local (writable) first-party sources.
+    result = [_builtin_source(), _local_source()]
     for src in sources:
         src_type = src.get("type", "local")
         if src_type == "git":
@@ -119,6 +143,65 @@ def read_playbook(source_path: str, relative_path: str) -> str | None:
     return target.read_text(errors="replace")
 
 
+def _resolve_writable(source_path: str, relative_path: str) -> tuple[Path | None, str | None]:
+    """Resolve relative_path under source_path, guarding traversal + extension.
+
+    Returns (target_path, None) on success or (None, error). The root need not
+    exist yet (steward-local is created on first save)."""
+    rel = relative_path.strip().lstrip("/")
+    if not rel:
+        return None, "Filename is required"
+    if not rel.endswith((".yml", ".yaml")):
+        return None, "Playbook filename must end in .yml or .yaml"
+    root = Path(source_path).resolve()
+    target = (root / rel).resolve()
+    try:
+        target.relative_to(root)
+    except ValueError:
+        return None, "Invalid path"
+    return target, None
+
+
+def write_playbook(source_path: str, relative_path: str, content: str) -> tuple[bool, str | None]:
+    """Write a playbook into a source, creating parent dirs. Traversal-guarded."""
+    target, err = _resolve_writable(source_path, relative_path)
+    if err:
+        return False, err
+    try:
+        target.parent.mkdir(parents=True, exist_ok=True)
+        target.write_text(content, encoding="utf-8")
+    except OSError as exc:
+        return False, f"Could not write file: {exc}"
+    return True, None
+
+
+def delete_playbook(source_path: str, relative_path: str) -> tuple[bool, str | None]:
+    """Delete a playbook from a source. Traversal-guarded; no-op if absent."""
+    target, err = _resolve_writable(source_path, relative_path)
+    if err:
+        return False, err
+    try:
+        if target.exists():
+            target.unlink()
+    except OSError as exc:
+        return False, f"Could not delete file: {exc}"
+    return True, None
+
+
+def validate_playbook_yaml(content: str) -> tuple[bool, str | None]:
+    """Cheap save-time validation: parses as YAML and is a non-empty play list."""
+    import yaml
+    try:
+        data = yaml.safe_load(content)
+    except yaml.YAMLError as exc:
+        return False, f"YAML error: {exc}"
+    if data is None:
+        return False, "Playbook is empty"
+    if not isinstance(data, list):
+        return False, "A playbook must be a list of plays (top-level YAML list)"
+    return True, None
+
+
 async def git_pull(source: dict) -> None:
     """Clone the git repo if absent; pull if already present.