feat(docker): schema for lifecycle events + swarm topology

Adds the milestone-77 storage that doesn't fit on the per-container row:

  * docker_events — lifecycle (start/stop/die/oom/health_change), to be
    derived by diffing consecutive host snapshots; host-scoped, indexed for
    timeline lookups (host_id, container_name, at) and retention pruning (at).
  * docker_swarm_services / docker_swarm_nodes — manager-reported Swarm
    topology (desired-vs-running replicas, node role/availability/status).

Migration docker_004 extends the docker branch (down_revision docker_003);
purely additive, no DROP+recreate. event/mode/role are plain strings (no
CHECK whitelist), matching how docker_containers models status. Integration
guard asserts the three new host-scoped tables exist.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_016Jg27rgypiW2efULXJDtMC

plugins/docker/migrations/versions/docker_004_events_swarm.py

@@ -0,0 +1,77 @@
+"""Docker lifecycle events + swarm topology tables
+
+Adds the storage for milestone-77 monitoring depth that doesn't fit on the
+per-container row:
+
+  * docker_events — lifecycle (start/stop/die/oom/health_change) derived by
+    diffing consecutive host snapshots; host-scoped, indexed for both timeline
+    lookups (host_id, container_name, at) and retention pruning (at).
+  * docker_swarm_services / docker_swarm_nodes — manager-reported Swarm
+    topology (desired-vs-running replicas, node role/availability/status).
+
+All new tables — purely additive, so no DROP+recreate. `event`/`mode`/`role`
+etc. are plain strings (no CHECK whitelist), matching how docker_containers
+models `status`; keeps the value sets open without a migration per addition
+(so rule 36 doesn't apply here).
+
+Revision ID: docker_004_events_swarm
+Revises: docker_003_container_enrichment
+Create Date: 2026-06-19
+"""
+from typing import Sequence, Union
+from alembic import op
+import sqlalchemy as sa
+
+revision: str = "docker_004_events_swarm"
+down_revision: Union[str, None] = "docker_003_container_enrichment"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "docker_events",
+        sa.Column("id", sa.String(36), primary_key=True),
+        sa.Column("host_id", sa.String(36),
+                  sa.ForeignKey("hosts.id", ondelete="CASCADE"), nullable=False),
+        sa.Column("container_name", sa.String(255), nullable=False),
+        sa.Column("event", sa.String(16), nullable=False),
+        sa.Column("detail", sa.String(255), nullable=True),
+        sa.Column("at", sa.DateTime(timezone=True), nullable=False),
+    )
+    op.create_index("ix_docker_events_host_container_time", "docker_events",
+                    ["host_id", "container_name", "at"])
+    op.create_index("ix_docker_events_at", "docker_events", ["at"])
+
+    op.create_table(
+        "docker_swarm_services",
+        sa.Column("host_id", sa.String(36),
+                  sa.ForeignKey("hosts.id", ondelete="CASCADE"), primary_key=True),
+        sa.Column("service_name", sa.String(255), primary_key=True),
+        sa.Column("mode", sa.String(16), nullable=False, server_default="replicated"),
+        sa.Column("desired", sa.Integer, nullable=False, server_default="0"),
+        sa.Column("running", sa.Integer, nullable=False, server_default="0"),
+        sa.Column("image", sa.String(512), nullable=True),
+        sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False),
+    )
+
+    op.create_table(
+        "docker_swarm_nodes",
+        sa.Column("host_id", sa.String(36),
+                  sa.ForeignKey("hosts.id", ondelete="CASCADE"), primary_key=True),
+        sa.Column("node_id", sa.String(64), primary_key=True),
+        sa.Column("hostname", sa.String(255), nullable=False, server_default=""),
+        sa.Column("role", sa.String(16), nullable=False, server_default="worker"),
+        sa.Column("availability", sa.String(16), nullable=False, server_default="active"),
+        sa.Column("status", sa.String(16), nullable=False, server_default="unknown"),
+        sa.Column("leader", sa.Boolean, nullable=False, server_default=sa.false()),
+        sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False),
+    )
+
+
+def downgrade() -> None:
+    op.drop_table("docker_swarm_nodes")
+    op.drop_table("docker_swarm_services")
+    op.drop_index("ix_docker_events_at", table_name="docker_events")
+    op.drop_index("ix_docker_events_host_container_time", table_name="docker_events")
+    op.drop_table("docker_events")

plugins/docker/models.py

@@ -85,3 +85,89 @@ class DockerMetric(Base):
         Index("ix_docker_metrics_host_container_time",
               "host_id", "container_name", "scraped_at"),
     )
+
+
+class DockerEvent(Base):
+    """Lifecycle events derived by diffing consecutive host snapshots.
+
+    The agent only reports current state, so Steward synthesises lifecycle by
+    comparing each new snapshot against the stored DockerContainer state for the
+    host: a container that appears → `start`; one that vanishes → `stop`; a
+    transition to a stopped status with a non-zero exit → `die`; an OOM-kill
+    flip → `oom`; a health status change → `health_change`. Scoped to the
+    reporting host (names are only unique within a host). `event` is a plain
+    string (no CHECK whitelist), matching how `status` is modelled on
+    DockerContainer — keeps the value set open without a migration per addition.
+    """
+    __tablename__ = "docker_events"
+
+    id: Mapped[str] = mapped_column(
+        String(36), primary_key=True, default=lambda: str(uuid.uuid4())
+    )
+    host_id: Mapped[str] = mapped_column(
+        String(36), ForeignKey("hosts.id", ondelete="CASCADE"), nullable=False
+    )
+    container_name: Mapped[str] = mapped_column(String(255), nullable=False)
+    event: Mapped[str] = mapped_column(String(16), nullable=False)
+    # start | stop | die | oom | health_change
+    detail: Mapped[str | None] = mapped_column(String(255), nullable=True)
+    # e.g. "exit 137", "healthy→unhealthy", image on start
+    at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), nullable=False,
+        default=lambda: datetime.now(timezone.utc),
+    )
+
+    # Timeline lookups filter on (host_id, container_name) and sort by time;
+    # retention prunes by `at` alone — index both access paths.
+    __table_args__ = (
+        Index("ix_docker_events_host_container_time",
+              "host_id", "container_name", "at"),
+        Index("ix_docker_events_at", "at"),
+    )
+
+
+class DockerSwarmService(Base):
+    """A Swarm service as seen by a manager host: desired vs running replicas.
+
+    Manager-scoped — only a host that is a Swarm manager reports these, so a
+    single-manager cluster yields one row set keyed by that host. Service names
+    are unique within a swarm; we still key by (host_id, service_name) so two
+    managers reporting independently don't collide.
+    """
+    __tablename__ = "docker_swarm_services"
+
+    host_id: Mapped[str] = mapped_column(
+        String(36), ForeignKey("hosts.id", ondelete="CASCADE"), primary_key=True
+    )
+    service_name: Mapped[str] = mapped_column(String(255), primary_key=True)
+    mode: Mapped[str] = mapped_column(String(16), nullable=False, default="replicated")
+    # replicated | global
+    desired: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
+    running: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
+    image: Mapped[str | None] = mapped_column(String(512), nullable=True)
+    updated_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), nullable=False,
+        default=lambda: datetime.now(timezone.utc),
+    )
+
+
+class DockerSwarmNode(Base):
+    """A Swarm node as seen by a manager host: role / availability / status."""
+    __tablename__ = "docker_swarm_nodes"
+
+    host_id: Mapped[str] = mapped_column(
+        String(36), ForeignKey("hosts.id", ondelete="CASCADE"), primary_key=True
+    )
+    node_id: Mapped[str] = mapped_column(String(64), primary_key=True)
+    hostname: Mapped[str] = mapped_column(String(255), nullable=False, default="")
+    role: Mapped[str] = mapped_column(String(16), nullable=False, default="worker")
+    # manager | worker
+    availability: Mapped[str] = mapped_column(String(16), nullable=False, default="active")
+    # active | pause | drain
+    status: Mapped[str] = mapped_column(String(16), nullable=False, default="unknown")
+    # ready | down | unknown
+    leader: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
+    updated_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), nullable=False,
+        default=lambda: datetime.now(timezone.utc),
+    )

tests/integration/test_docker.py

@@ -53,6 +53,28 @@ def test_docker_tables_are_host_scoped(app):
     asyncio.run(_go())
 
 
+@_NEEDS_DB
+def test_events_and_swarm_tables_exist(app):
+    """docker_004 created the lifecycle + swarm topology tables with host_id FKs."""
+    from sqlalchemy import text
+
+    async def _go():
+        async with app.db_sessionmaker() as s:
+            for tbl in ("docker_events", "docker_swarm_services", "docker_swarm_nodes"):
+                cols = {r[0] for r in (await s.execute(text(
+                    "SELECT column_name FROM information_schema.columns "
+                    "WHERE table_name = :t"), {"t": tbl})).all()}
+                assert cols, f"{tbl} missing entirely"
+                assert "host_id" in cols, f"{tbl} missing host_id"
+            # docker_events carries the lifecycle columns the ingest derivation writes.
+            ev_cols = {r[0] for r in (await s.execute(text(
+                "SELECT column_name FROM information_schema.columns "
+                "WHERE table_name = 'docker_events'"))).all()}
+            assert {"event", "container_name", "at", "detail"} <= ev_cols
+
+    asyncio.run(_go())
+
+
 def _persist_fn(app):
     """Resolve persist_host_docker via the registered capability if the docker
     plugin is loaded, else import it directly (the import is safe only when the