feat(docker): persist pushed container logs into docker_logs [M79 step 3]

Ingest side of container logs. The host_agent ingest route extracts
sample["docker_logs"] as time-series batches (append every line, each carrying
its own Docker ts with recorded_at as fallback) and hands them to the docker
capability alongside the existing container/swarm/disk data — still under the
begin_nested SAVEPOINT so a logs failure can't sink host metrics.

- ingest.py: _persist_logs + pure _log_rows(batches, host_id) that shapes/filters
  records (drops the _steward truncation marker, malformed + lineless records,
  normalises unknown stream → stdout); persist_host_docker gains a logs= param
- routes.py: accumulate docker_log_batches, add to the guard + capability call
- unit test for _log_rows shaping/filtering; integration push→store→query test

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

plugins/docker/ingest.py

@@ -92,6 +92,46 @@ def _derive_events(old_state: dict, new_containers: list) -> list:
     return events
 
 
+def _log_rows(log_batches, host_id: str):
+    """Pure: flatten the agent's log batches → docker_logs row kwargs (no DB).
+
+    `log_batches` is a list of (recorded_at, records) where each record is
+    {"container", "stream", "ts", "line"}. A line's own Docker timestamp wins;
+    recorded_at is the fallback when the agent couldn't parse one. Drops the
+    agent's advisory truncation marker (container "_steward" — it signals a
+    deferral, not a real container), malformed records, and lineless records;
+    normalises an unknown stream to stdout. Unit-testable in isolation.
+    """
+    for recorded_at, records in log_batches:
+        if not isinstance(records, list):
+            continue
+        for rec in records:
+            if not isinstance(rec, dict):
+                continue
+            name = rec.get("container")
+            line = rec.get("line")
+            if not name or name == "_steward" or line is None:
+                continue
+            ts = _parse_started_at(rec.get("ts")) or recorded_at
+            stream = rec.get("stream")
+            if stream not in ("stdout", "stderr"):
+                stream = "stdout"
+            yield {"host_id": host_id, "container_name": str(name)[:255],
+                   "ts": ts, "stream": stream, "line": str(line)}
+
+
+async def _persist_logs(session, host, log_batches) -> None:
+    """Append pushed container log lines (one row per line) for this host.
+
+    Time-series / append-only — the per-container size+age ring (retention)
+    bounds growth, so a chatty container just keeps a shorter window.
+    """
+    from .models import DockerLog
+
+    for row in _log_rows(log_batches, host.id):
+        session.add(DockerLog(**row))
+
+
 async def _persist_swarm(session, host, swarm: dict) -> None:
     """Upsert this manager's swarm topology; drop rows no longer reported.
 
@@ -201,7 +241,8 @@ async def _persist_disk(session, host, disk: dict) -> None:
     await session.execute(stale)
 
 
-async def persist_host_docker(session, host, snapshots, swarm=None, disk=None) -> None:
+async def persist_host_docker(session, host, snapshots, swarm=None, disk=None,
+                              logs=None) -> None:
     """Upsert containers + time-series + lifecycle events + swarm for one host.
 
     `snapshots` is a list of (recorded_at: datetime, containers: list[dict]) —
@@ -211,7 +252,9 @@ async def persist_host_docker(session, host, snapshots, swarm=None, disk=None) -
     alert pipeline, and lifecycle-event derivation. `swarm` is the newest
     sample's swarm object (or None off managers) — persisted when present.
     `disk` is the newest sample's /system/df summary (or None on Docker-less
-    hosts) — persisted when present.
+    hosts) — persisted when present. `logs` is a list of (recorded_at, records)
+    log batches (m79) — appended one row per line when present, independent of
+    whether this sample also carried container metrics.
     """
     from steward.core.alerts import record_metric
     from .models import DockerContainer, DockerEvent, DockerMetric
@@ -220,6 +263,8 @@ async def persist_host_docker(session, host, snapshots, swarm=None, disk=None) -
         await _persist_swarm(session, host, swarm)
     if disk is not None:
         await _persist_disk(session, host, disk)
+    if logs:
+        await _persist_logs(session, host, logs)
 
     if not snapshots:
         return

plugins/host_agent/routes.py

@@ -202,6 +202,7 @@ async def ingest():
         accepted = 0
         latest_ts: datetime | None = None
         docker_snapshots: list[tuple[datetime, list]] = []
+        docker_log_batches: list[tuple[datetime, list]] = []
         latest_swarm: dict | None = None
         latest_swarm_ts: datetime | None = None
         latest_disk: dict | None = None
@@ -217,6 +218,11 @@ async def ingest():
             docker = sample.get("docker")
             if isinstance(docker, list) and docker:
                 docker_snapshots.append((recorded_at, docker))
+            # Container logs are time-series (append every line, not newest-only);
+            # each record carries its own Docker ts, recorded_at is the fallback.
+            docker_logs = sample.get("docker_logs")
+            if isinstance(docker_logs, list) and docker_logs:
+                docker_log_batches.append((recorded_at, docker_logs))
             # Swarm is current-state, not time-series — keep only the newest
             # sample's topology (a manager re-reports it every interval).
             swarm = sample.get("swarm")
@@ -240,7 +246,8 @@ async def ingest():
         # (opportunistic synergy via the capability registry — no hard import,
         # no-op when docker is disabled). A failure here must never sink the
         # whole ingest, so the metrics above still land.
-        if docker_snapshots or latest_swarm is not None or latest_disk is not None:
+        if (docker_snapshots or latest_swarm is not None
+                or latest_disk is not None or docker_log_batches):
             from steward.core.capabilities import has_capability, invoke_capability
             if has_capability("docker.persist_host_samples"):
                 try:
@@ -250,7 +257,7 @@ async def ingest():
                         await invoke_capability(
                             "docker.persist_host_samples", UserRole.admin,
                             session, host, docker_snapshots, latest_swarm,
-                            latest_disk,
+                            latest_disk, docker_log_batches,
                         )
                 except Exception:
                     current_app.logger.exception(

tests/integration/test_docker.py

@@ -172,6 +172,44 @@ def test_persist_scopes_containers_by_host(app):
     assert enrich == ("healthy", 2, "web", 1000)   # enrichment round-trips
 
 
+@_NEEDS_DB
+def test_persist_logs_stores_lines(app):
+    """Pushed container logs land in docker_logs, host-scoped, one row per line,
+    ordered by their own ts; the agent's _steward truncation marker is dropped."""
+    from sqlalchemy import text
+    from steward.models.hosts import Host
+
+    persist = _persist_fn(app)
+    now = datetime.now(timezone.utc)
+    log_batches = [(now, [
+        {"container": "web", "stream": "stdout",
+         "ts": "2026-01-01T00:00:01+00:00", "line": "started"},
+        {"container": "web", "stream": "stderr",
+         "ts": "2026-01-01T00:00:02+00:00", "line": "warn: x"},
+        {"container": "_steward", "stream": "stderr", "ts": None, "line": "truncated"},
+    ])]
+
+    async def _go():
+        async with app.db_sessionmaker() as s:
+            async with s.begin():
+                await s.execute(text("DELETE FROM docker_logs"))
+                h = Host(id=str(uuid.uuid4()), name="loghost", address="10.0.0.9")
+                s.add(h)
+                await s.flush()
+                hid = h.id
+                # snapshots empty; logs passed as the 6th positional arg.
+                await persist(s, h, [], None, None, log_batches)
+            rows = (await s.execute(text(
+                "SELECT container_name, stream, line FROM docker_logs "
+                "WHERE host_id = :h ORDER BY ts"), {"h": hid})).all()
+            return [tuple(r) for r in rows]
+
+    assert asyncio.run(_go()) == [
+        ("web", "stdout", "started"),
+        ("web", "stderr", "warn: x"),
+    ]
+
+
 @_NEEDS_DB
 def test_large_memory_values_persist_as_bigint(app):
     """A container using >2^31 bytes of RAM must persist. Regression: mem_usage_bytes

tests/plugins/docker/test_ingest_events.py

@@ -65,3 +65,35 @@ def test_no_event_when_unchanged():
     old = {"web": {"status": "running", "health": "healthy", "oom_killed": False, "exit_code": None}}
     events = _derive_events(old, [_c("web", "running", health="healthy")])
     assert events == []
+
+
+# ── container-log row shaping (m79, pure) ─────────────────────────────────────
+
+def test_log_rows_shapes_and_filters():
+    from datetime import datetime, timezone
+    from plugins.docker.ingest import _log_rows
+
+    rec_at = datetime(2026, 1, 1, tzinfo=timezone.utc)
+    batches = [(rec_at, [
+        {"container": "web", "stream": "stdout",
+         "ts": "2026-01-01T00:00:01+00:00", "line": "hello"},
+        {"container": "web", "stream": "stderr", "ts": None, "line": "no-ts"},
+        {"container": "_steward", "stream": "stderr",
+         "ts": None, "line": "truncated"},          # advisory marker → dropped
+        {"container": "web", "stream": "weird", "ts": None, "line": "bad-stream"},
+        {"container": "", "line": "x"},              # no name → dropped
+        {"container": "web", "ts": None, "line": None},   # no line → dropped
+        "not-a-dict",                                # malformed → dropped
+    ])]
+    rows = list(_log_rows(batches, "host-1"))
+    assert [r["line"] for r in rows] == ["hello", "no-ts", "bad-stream"]
+    assert rows[0]["host_id"] == "host-1" and rows[0]["container_name"] == "web"
+    assert rows[1]["ts"] == rec_at            # None ts → recorded_at fallback
+    assert rows[2]["stream"] == "stdout"      # unknown stream normalised
+
+
+def test_log_rows_skips_non_list_records():
+    from datetime import datetime, timezone
+    from plugins.docker.ingest import _log_rows
+    rec_at = datetime(2026, 1, 1, tzinfo=timezone.utc)
+    assert list(_log_rows([(rec_at, None)], "h")) == []