diff --git a/plugins/host_agent/routes.py b/plugins/host_agent/routes.py
index ed8ef20..07e672b 100644
--- a/plugins/host_agent/routes.py
+++ b/plugins/host_agent/routes.py
@@ -8,7 +8,7 @@ from typing import Any
 from pathlib import Path
 
 import secrets
-from quart import Blueprint, current_app, jsonify, request, Response, render_template, redirect, url_for
+from quart import Blueprint, current_app, jsonify, request, Response, render_template, redirect, url_for, session
 from steward.auth.middleware import require_role
 from steward.models.users import UserRole
 from sqlalchemy import select, func, or_
@@ -535,12 +535,22 @@ def _new_token_pair() -> tuple[str, str]:
 @host_agent_bp.get("/settings/")
 @require_role(UserRole.admin)
 async def settings_list():
-    async with current_app.db_sessionmaker() as session:
-        regs = (await session.execute(select(HostAgentRegistration))).scalars().all()
+    from steward.core.capabilities import has_capability
+    from steward.models.ansible_inventory import AnsibleTarget, AnsibleGroup
+
+    ansible_available = has_capability("ansible.run_playbook")
+    async with current_app.db_sessionmaker() as db:
+        regs = (await db.execute(select(HostAgentRegistration))).scalars().all()
         hosts_by_id = {
-            h.id: h for h in (await session.execute(
+            h.id: h for h in (await db.execute(
                 select(Host).where(Host.id.in_([r.host_id for r in regs])))).scalars().all()
         } if regs else {}
+        targets, groups = [], []
+        if ansible_available:
+            targets = (await db.execute(
+                select(AnsibleTarget).order_by(AnsibleTarget.name))).scalars().all()
+            groups = (await db.execute(
+                select(AnsibleGroup).order_by(AnsibleGroup.name))).scalars().all()
 
     new_token = request.args.get("new_token")
     new_host_id = request.args.get("host_id")
@@ -555,6 +565,9 @@ async def settings_list():
         ],
         new_token=new_token,
         install_url=install_url,
+        ansible_available=ansible_available,
+        deploy_targets=targets,
+        deploy_groups=groups,
     )
 
 
@@ -617,3 +630,98 @@ async def delete_registration(host_id: str):
             await session.delete(reg)
             await session.commit()
     return redirect(url_for("host_agent.settings_list"))
+
+
+# ── Deploy via Ansible (plugin↔core synergy via the capability registry) ──────
+
+async def _ensure_host_for_target(db, target) -> "Host":
+    """Find or create the Host that an AnsibleTarget should report under."""
+    if getattr(target, "host_id", None):
+        h = await db.get(Host, target.host_id)
+        if h:
+            return h
+    h = (await db.execute(
+        select(Host).where(Host.name == target.name))).scalar_one_or_none()
+    if h is None:
+        h = Host(name=target.name, address=getattr(target, "address", "") or "")
+        db.add(h)
+        await db.flush()
+    return h
+
+
+async def _mint_registration_token(db, host) -> str:
+    """Create or rotate the host's agent registration; return the raw token.
+
+    Tokens are stored hashed (unrecoverable), so deploying always mints a fresh
+    token — the run installs the agent with it. Existing agents get rotated.
+    """
+    reg = (await db.execute(select(HostAgentRegistration).where(
+        HostAgentRegistration.host_id == host.id))).scalar_one_or_none()
+    raw, hashed = _new_token_pair()
+    if reg is None:
+        db.add(HostAgentRegistration(host_id=host.id, token_hash=hashed))
+    else:
+        reg.token_hash = hashed
+        reg.token_created_at = datetime.now(timezone.utc)
+    return raw
+
+
+@host_agent_bp.post("/deploy")
+@require_role(UserRole.admin)
+async def deploy_via_ansible():
+    """Install/update the agent on Ansible inventory targets via the bundled
+    install playbook — the 'curl | sh becomes a button' synergy.
+
+    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.sources import BUILTIN_SOURCE_NAME
+
+    if not has_capability("ansible.run_playbook"):
+        return _error(400, "ansible_unavailable", "Ansible is not available")
+
+    form = await request.form
+    scope = (form.get("inventory_scope", "") or "").strip()
+    try:
+        interval = max(5, int(form.get("agent_interval", "30")))
+    except (TypeError, ValueError):
+        interval = 30
+    if not (scope.startswith("steward:target:")
+            or scope.startswith("steward:group:")
+            or scope == "steward:all"):
+        return _error(400, "bad_scope", "Choose a target or group")
+
+    url = public_base_url(request)
+    async with current_app.db_sessionmaker() as db:
+        targets = await fetch_scope_targets(db, scope)
+        if not targets:
+            return _error(400, "no_targets", "No Ansible targets in that scope")
+        tokens: dict[str, str] = {}
+        async with db.begin():
+            for t in targets:
+                host = await _ensure_host_for_target(db, t)
+                tokens[t.name] = await _mint_registration_token(db, host)
+        inv = generate_inventory(targets)
+        for name, tok in tokens.items():
+            hv = inv["_meta"]["hostvars"].setdefault(name, {})
+            hv["steward_url"] = url
+            hv["steward_token"] = tok
+        inventory_content = _json.dumps(inv)
+
+    actor_role = UserRole(session.get("user_role", "viewer"))
+    run, _source, err = await invoke_capability(
+        "ansible.run_playbook", actor_role,
+        current_app._get_current_object(),  # type: ignore[attr-defined]
+        source_name=BUILTIN_SOURCE_NAME,
+        playbook_path="host_agent/install.yml",
+        inventory_content=inventory_content,
+        inventory_scope=scope,
+        params={"extra_vars": [f"agent_interval={interval}"]},
+        triggered_by=session.get("user_id"),
+    )
+    if err:
+        return _error(400, "deploy_failed", err)
+    return redirect(url_for("ansible.run_detail", run_id=run.id))
diff --git a/plugins/host_agent/templates/settings_list.html b/plugins/host_agent/templates/settings_list.html
index 49684f7..dc52e79 100644
--- a/plugins/host_agent/templates/settings_list.html
+++ b/plugins/host_agent/templates/settings_list.html
@@ -28,6 +28,37 @@
   </form>
 </div>
 
+{% if ansible_available %}
+<div class="card">
+  <h3 style="margin-bottom:0.4rem;">Deploy via Ansible</h3>
+  <p style="font-size:0.82rem;color:var(--text-muted);margin-bottom:0.75rem;">
+    Install (or update) the agent on Ansible inventory hosts in one run — no per-host <code>curl | sh</code>.
+    A fresh token is minted per host and injected into the run; existing agents are rotated to the new token.
+  </p>
+  {% if not (deploy_targets or deploy_groups) %}
+  <p style="font-size:0.85rem;color:var(--text-dim);">
+    No Ansible inventory targets yet. Add some under <a href="/ansible/browse">Ansible → Browse</a>.
+  </p>
+  {% else %}
+  <form method="post" action="/plugins/host_agent/deploy"
+        style="display:flex;gap:0.75rem;align-items:flex-end;flex-wrap:wrap;">
+    <div class="form-group" style="margin-bottom:0;">
+      <label>Target</label>
+      <select name="inventory_scope" required>
+        {% for g in deploy_groups %}<option value="steward:group:{{ g.id }}">Group: {{ g.name }}</option>{% endfor %}
+        {% for t in deploy_targets %}<option value="steward:target:{{ t.id }}">Target: {{ t.name }}</option>{% endfor %}
+      </select>
+    </div>
+    <div class="form-group" style="margin-bottom:0;">
+      <label>Report interval (s)</label>
+      <input type="number" name="agent_interval" value="30" min="5" style="width:7rem;">
+    </div>
+    <button type="submit" class="btn">Deploy</button>
+  </form>
+  {% endif %}
+</div>
+{% endif %}
+
 <div class="card-flush">
   <table class="table">
     <thead>
diff --git a/steward/ansible/runner.py b/steward/ansible/runner.py
index 1aada6b..a8ea6cd 100644
--- a/steward/ansible/runner.py
+++ b/steward/ansible/runner.py
@@ -24,10 +24,14 @@ async def trigger_run(
     inventory_scope: str = "steward:all",
     params: dict | None = None,
     triggered_by: str | None = None,
+    inventory_content: str | None = None,
 ):
     """Resolve inventory for the scope, persist an AnsibleRun, and launch it.
 
     triggered_by=None marks a system/automated run (alerts, schedules).
+    If inventory_content is provided, it is used verbatim and scope resolution
+    is skipped (the caller built a bespoke inventory — e.g. host_agent deploy
+    injecting per-host tokens); inventory_scope is still recorded for display.
     Returns (run, source, error): on success error is None; on failure run is
     None and error is a short human-readable reason.
     """
@@ -36,9 +40,10 @@ async def trigger_run(
     if source is None:
         return None, None, "Source not found"
 
-    inventory_content: str | None = None
     inventory_path: str | None = None
-    if inventory_scope.startswith("steward:"):
+    if inventory_content is not None:
+        pass  # caller-supplied inventory wins; scope kept only for display
+    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))
diff --git a/steward/app.py b/steward/app.py
index d5bcf06..a97bd86 100644
--- a/steward/app.py
+++ b/steward/app.py
@@ -123,6 +123,19 @@ def create_app(
     register_status_source(ping_status_source)
     register_status_source(dns_status_source)
 
+    # Publish the Ansible "run a playbook" capability so plugins (e.g. host_agent
+    # auto-deploy) can drive runs without importing the runner. Ansible is core,
+    # so this is always available; consumers still gate on has_capability().
+    from .core.capabilities import register_capability
+    from .ansible.runner import trigger_run
+    from .models.users import UserRole as _UserRole
+    register_capability(
+        "ansible.run_playbook", trigger_run,
+        label="Run Ansible playbook",
+        description="Launch an Ansible playbook run (manual, alert, schedule, or plugin-driven).",
+        required_role=_UserRole.operator,
+    )
+
     # ── 8. Build task registry ─────────────────────────────────────────────────
     app._task_registry = []
 
diff --git a/steward/auth/middleware.py b/steward/auth/middleware.py
index d57bd34..6aca001 100644
--- a/steward/auth/middleware.py
+++ b/steward/auth/middleware.py
@@ -6,6 +6,11 @@ from steward.models.users import UserRole
 _ROLE_ORDER = [UserRole.viewer, UserRole.operator, UserRole.admin]
 
 
+def role_meets(user_role: UserRole, minimum_role: UserRole) -> bool:
+    """True when user_role is at least minimum_role in the viewer<operator<admin order."""
+    return _ROLE_ORDER.index(user_role) >= _ROLE_ORDER.index(minimum_role)
+
+
 def require_role(minimum_role: UserRole):
     """Decorator: requires authenticated user with at least minimum_role.
     Also allows access for validated share-token requests (viewer level only).
diff --git a/steward/core/capabilities.py b/steward/core/capabilities.py
new file mode 100644
index 0000000..df62b11
--- /dev/null
+++ b/steward/core/capabilities.py
@@ -0,0 +1,95 @@
+# steward/core/capabilities.py
+"""Plugin/core capability registry — opportunistic, decoupled synergy.
+
+A capability is a named, permission-gated action that one part of the system
+(core module or plugin) publishes and another can discover + invoke WITHOUT a
+hard import. The publisher registers a callable under a string key; a consumer
+checks `has_capability(key)` (graceful degradation — the synergy is a bonus,
+never a requirement) and calls `invoke_capability(key, actor_role, ...)`.
+
+First consumer: the host_agent plugin invokes "ansible.run_playbook" to deploy
+its agent via Ansible instead of importing the Ansible runner directly.
+
+Security: every capability declares a required_role; invoke_capability enforces
+the caller's role meets it, so a low-privilege context can't drive a privileged
+action in another module.
+"""
+from __future__ import annotations
+
+import inspect
+import logging
+from dataclasses import dataclass
+from typing import Callable
+
+from steward.auth.middleware import role_meets
+from steward.models.users import UserRole
+
+logger = logging.getLogger(__name__)
+
+
+class CapabilityUnavailable(Exception):
+    """Raised when an unknown capability key is invoked."""
+
+
+class CapabilityForbidden(Exception):
+    """Raised when the actor's role is below the capability's required_role."""
+
+
+@dataclass
+class Capability:
+    key: str
+    fn: Callable
+    label: str
+    description: str
+    required_role: UserRole
+
+
+_CAPABILITIES: dict[str, Capability] = {}
+
+
+def register_capability(
+    key: str,
+    fn: Callable,
+    *,
+    label: str,
+    description: str = "",
+    required_role: UserRole = UserRole.admin,
+) -> None:
+    """Publish a capability. Last registration for a key wins (idempotent re-register)."""
+    _CAPABILITIES[key] = Capability(key, fn, label, description, required_role)
+
+
+def has_capability(key: str) -> bool:
+    return key in _CAPABILITIES
+
+
+def get_capability(key: str) -> Capability | None:
+    return _CAPABILITIES.get(key)
+
+
+def list_capabilities() -> list[Capability]:
+    return list(_CAPABILITIES.values())
+
+
+def clear_capabilities() -> None:
+    """Reset the registry (tests)."""
+    _CAPABILITIES.clear()
+
+
+async def invoke_capability(key: str, actor_role: UserRole, /, *args, **kwargs):
+    """Invoke a registered capability after a role check. Awaits async callables.
+
+    Raises CapabilityUnavailable if the key isn't registered, CapabilityForbidden
+    if actor_role is insufficient.
+    """
+    cap = _CAPABILITIES.get(key)
+    if cap is None:
+        raise CapabilityUnavailable(key)
+    if not role_meets(actor_role, cap.required_role):
+        raise CapabilityForbidden(
+            f"capability {key!r} requires role {cap.required_role.value}"
+        )
+    result = cap.fn(*args, **kwargs)
+    if inspect.isawaitable(result):
+        result = await result
+    return result
diff --git a/tests/core/test_capabilities.py b/tests/core/test_capabilities.py
new file mode 100644
index 0000000..33e2677
--- /dev/null
+++ b/tests/core/test_capabilities.py
@@ -0,0 +1,57 @@
+"""Unit tests for the plugin/core capability registry."""
+import asyncio
+
+import pytest
+
+from steward.auth.middleware import role_meets
+from steward.core.capabilities import (
+    CapabilityForbidden,
+    CapabilityUnavailable,
+    clear_capabilities,
+    get_capability,
+    has_capability,
+    invoke_capability,
+    register_capability,
+)
+from steward.models.users import UserRole
+
+
+def setup_function():
+    clear_capabilities()
+
+
+def test_register_has_get_defaults_to_admin():
+    register_capability("x.do", lambda: "ok", label="Do X")
+    assert has_capability("x.do")
+    assert get_capability("x.do").label == "Do X"
+    assert get_capability("x.do").required_role == UserRole.admin
+
+
+def test_invoke_sync_callable():
+    register_capability("x.sum", lambda a, b: a + b, label="Sum", required_role=UserRole.viewer)
+    assert asyncio.run(invoke_capability("x.sum", UserRole.viewer, 2, 3)) == 5
+
+
+def test_invoke_async_callable():
+    async def double(v):
+        return v * 2
+
+    register_capability("x.double", double, label="D", required_role=UserRole.viewer)
+    assert asyncio.run(invoke_capability("x.double", UserRole.operator, 4)) == 8
+
+
+def test_invoke_unknown_capability_raises():
+    with pytest.raises(CapabilityUnavailable):
+        asyncio.run(invoke_capability("nope", UserRole.admin))
+
+
+def test_invoke_forbidden_when_role_too_low():
+    register_capability("x.admin", lambda: 1, label="A", required_role=UserRole.admin)
+    with pytest.raises(CapabilityForbidden):
+        asyncio.run(invoke_capability("x.admin", UserRole.operator))
+
+
+def test_role_meets_ordering():
+    assert role_meets(UserRole.admin, UserRole.operator) is True
+    assert role_meets(UserRole.operator, UserRole.operator) is True
+    assert role_meets(UserRole.viewer, UserRole.operator) is False