feat(docker): docker_logs table + docker_009 migration [M79 step 2]

Dedicated host-scoped table for pushed container log lines (one row per line),
chained after docker_008. host_id FK CASCADE; twin indexes — (host_id,
container_name, ts) for the viewer, ts alone for the age-cutoff prune — mirroring
docker_events. Integration schema-shape test asserts columns + both indexes.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01CAGR73DUowdVFVvYzLXC5C

plugins/docker/migrations/versions/docker_009_container_logs.py

@@ -0,0 +1,48 @@
+"""Docker container logs table
+
+Adds docker_logs — one row per container log line, pushed by the host agent and
+folded into its metrics push. Time-series, host-scoped (container names are only
+unique within a host). Bounded by a per-container size+age ring in retention, so
+a chatty container keeps a shorter window rather than growing without limit.
+Additive create_table + twin indexes (viewer lookup by (host, container, ts);
+age-cutoff prune by ts).
+
+Revision ID: docker_009_container_logs
+Revises: docker_008_bigint_mem
+Create Date: 2026-07-19
+"""
+from typing import Sequence, Union
+from alembic import op
+import sqlalchemy as sa
+
+revision: str = "docker_009_container_logs"
+down_revision: Union[str, None] = "docker_008_bigint_mem"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "docker_logs",
+        sa.Column("id", sa.String(length=36), nullable=False),
+        sa.Column("host_id", sa.String(length=36), nullable=False),
+        sa.Column("container_name", sa.String(length=255), nullable=False),
+        sa.Column("ts", sa.DateTime(timezone=True), nullable=False),
+        sa.Column("stream", sa.String(length=8), nullable=False, server_default="stdout"),
+        sa.Column("line", sa.Text(), nullable=False, server_default=""),
+        sa.ForeignKeyConstraint(["host_id"], ["hosts.id"], ondelete="CASCADE"),
+        sa.PrimaryKeyConstraint("id"),
+    )
+    op.create_index("ix_docker_logs_host_id", "docker_logs", ["host_id"])
+    op.create_index("ix_docker_logs_container_name", "docker_logs", ["container_name"])
+    op.create_index("ix_docker_logs_host_container_time",
+                    "docker_logs", ["host_id", "container_name", "ts"])
+    op.create_index("ix_docker_logs_ts", "docker_logs", ["ts"])
+
+
+def downgrade() -> None:
+    op.drop_index("ix_docker_logs_ts", table_name="docker_logs")
+    op.drop_index("ix_docker_logs_host_container_time", table_name="docker_logs")
+    op.drop_index("ix_docker_logs_container_name", table_name="docker_logs")
+    op.drop_index("ix_docker_logs_host_id", table_name="docker_logs")
+    op.drop_table("docker_logs")

plugins/docker/models.py

@@ -163,6 +163,43 @@ class DockerEvent(Base):
     )
 
 
+class DockerLog(Base):
+    """Container log lines pushed by the host agent — one row per line.
+
+    Time-series, scoped to the reporting host (container names are only unique
+    within a host, same identity as docker_metrics). The agent tails each running
+    container incrementally and folds new lines into its metrics push; `ts` is the
+    line's own Docker timestamp. Bounded by a per-container size+age ring
+    (retention), so a chatty container just keeps a shorter window rather than
+    growing without limit.
+    """
+    __tablename__ = "docker_logs"
+
+    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, index=True
+    )
+    container_name: Mapped[str] = mapped_column(String(255), nullable=False, index=True)
+    ts: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), nullable=False,
+        default=lambda: datetime.now(timezone.utc),
+    )
+    stream: Mapped[str] = mapped_column(String(8), nullable=False, default="stdout")
+    # stdout | stderr
+    line: Mapped[str] = mapped_column(Text, nullable=False, default="")
+
+    # Viewer filters on (host_id, container_name) and sorts by time; the ring
+    # prune walks the same key newest-first. A second index on `ts` alone serves
+    # the age-cutoff delete. Twin-index idiom, mirroring docker_events.
+    __table_args__ = (
+        Index("ix_docker_logs_host_container_time",
+              "host_id", "container_name", "ts"),
+        Index("ix_docker_logs_ts", "ts"),
+    )
+
+
 class DockerSwarmService(Base):
     """A Swarm service as seen by a manager host: desired vs running replicas.
 

tests/integration/test_docker.py

@@ -75,6 +75,28 @@ def test_events_and_swarm_tables_exist(app):
     asyncio.run(_go())
 
 
+@_NEEDS_DB
+def test_docker_logs_table_shape(app):
+    """docker_009 created docker_logs: host-scoped, with the twin indexes the
+    viewer (host, container, ts) and the age-cutoff prune (ts) rely on."""
+    from sqlalchemy import text
+
+    async def _go():
+        async with app.db_sessionmaker() as s:
+            cols = {r[0] for r in (await s.execute(text(
+                "SELECT column_name FROM information_schema.columns "
+                "WHERE table_name = 'docker_logs'"))).all()}
+            assert cols, "docker_logs missing entirely"
+            assert {"id", "host_id", "container_name", "ts", "stream", "line"} <= cols
+            idx = {r[0] for r in (await s.execute(text(
+                "SELECT indexname FROM pg_indexes "
+                "WHERE tablename = 'docker_logs'"))).all()}
+            assert "ix_docker_logs_host_container_time" in idx
+            assert "ix_docker_logs_ts" in idx
+
+    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