fix(host_agent): maintenance reads the swarm through the loaded Docker plugin, not by importing its models again (#5768)

Importing plugins.docker.models by path defines its tables a second time,
because the plugin manager loaded them as _steward_plugin_docker.models.
Inside the app that raises InvalidRequestError, not ImportError. The cadence
integration test hit it, and #5767's _swarm (used when a job is planned)
carried the same import. Both now go through maintenance.docker_models().

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

plugins/host_agent/cadence.py

@@ -172,11 +172,9 @@ async def _back(session, host, job, reg, now: datetime) -> str | None:
     node = None
     services: list = []
     if job.node_id:
-        try:
-            from plugins.docker.models import DockerSwarmNode, DockerSwarmService
-        except ImportError:
-            DockerSwarmNode = DockerSwarmService = None
-        if DockerSwarmNode is not None:
+        models = maintenance.docker_models()
+        if models is not None:
+            DockerSwarmNode, DockerSwarmService = models.DockerSwarmNode, models.DockerSwarmService
             node = (await session.execute(select(DockerSwarmNode).where(
                 DockerSwarmNode.node_id == job.node_id)
                 .order_by(DockerSwarmNode.updated_at.desc()).limit(1))).scalar_one_or_none()

plugins/host_agent/maintenance.py

@@ -378,14 +378,29 @@ def summary(job) -> str:
     return f"{head}: {', '.join(parts)}" if parts else head
 
 
+def docker_models():
+    """The loaded Docker plugin's models module, or None when it isn't loaded.
+    Reached through the plugin manager: importing plugins.docker.models by its
+    path would define its tables a second time, and the app would refuse it."""
+    import importlib
+    from steward.core.plugin_manager import get_loaded_plugin_modules
+    loaded = get_loaded_plugin_modules().get("docker")
+    if loaded is None:
+        return None
+    try:
+        return importlib.import_module(f"{loaded.__name__}.models")
+    except ImportError:
+        return None
+
+
 async def _swarm(session) -> tuple:
     """({node hostname: node id}, {manager host id: its node id}) from what the
     Docker plugin last stored; empty when it isn't installed. A manager is a
     host whose agent reported the swarm's nodes."""
-    try:
-        from plugins.docker.models import DockerSwarmNode
-    except ImportError:
+    models = docker_models()
+    if models is None:
         return {}, {}
+    DockerSwarmNode = models.DockerSwarmNode
     from sqlalchemy import select
     from steward.models.hosts import Host
     rows = (await session.execute(select(DockerSwarmNode))).scalars().all()