docs(host-agent): document container-log collection [M79 step 6]

Design-spec updates for m79: the docker_logs_enabled / docker_log_exclude config
keys, the collect_docker_logs collector + the drop-logs-on-backoff note in the
agent internals, and a wire-format design point on docker_logs riding in the same
push (with server-side toggle/exclude enforcement + the per-container ring).

The end-to-end chain (store → rotate → query + ingest enforcement) is already
covered by the integration tests added across steps 2–5, so no capstone test is
added. AGENT_VERSION bump (1.7.0) landed in step 1.

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

.forgejo/workflows/ci.yml

@@ -14,6 +14,17 @@ on:
   # pull_request intentionally absent — push on [dev, main] already fires CI for
   # every dev commit and dev→main PRs. Single-operator repo, no fork PRs.
 
+# Serialize CI runs per ref so the publish lane never shares the runner's docker
+# daemon with another run. Two publishes racing on one daemon evict each other's
+# freshly-built image mid-push, breaking the post-build tag/push (issue #1093).
+# cancel-in-progress:false is deliberate — rule 46 requires EVERY dev/main push to
+# publish its own immutable :<sha> image, so a superseded run must still run to
+# completion (never cancelled) to emit its SHA. (Forgejo honors `concurrency:` —
+# CI-runner's build workflows use it.)
+concurrency:
+  group: ci-${{ github.ref }}
+  cancel-in-progress: false
+
 jobs:
   # Fast-fail lint lane. ruff is pre-installed in the ci-python image, so this
   # runs with NO dependency install and surfaces lint bounces in seconds.
@@ -115,14 +126,25 @@ jobs:
         run: |
           set -euxo pipefail
           IMAGE=git.fabledsword.com/bvandeusen/steward
-          docker build -t "$IMAGE:${{ github.sha }}" .
-          docker push "$IMAGE:${{ github.sha }}"
+          # The moving tag for this ref: dev→:dev, main→:latest (rule 46). main IS
+          # the production line, so :latest tracks main's tip; there is no :main.
+          MOVING=""
           if [ "${{ github.ref }}" = "refs/heads/dev" ]; then
-            docker tag "$IMAGE:${{ github.sha }}" "$IMAGE:dev"
-            docker push "$IMAGE:dev"
+            MOVING="dev"
           elif [ "${{ github.ref }}" = "refs/heads/main" ]; then
-            docker tag "$IMAGE:${{ github.sha }}" "$IMAGE:latest"
-            docker push "$IMAGE:latest"
+            MOVING="latest"
+          fi
+          # Tag BOTH targets at build time (mirrors CI-runner's build workflows):
+          # one `docker build -t :<sha> -t :<moving>` points both tags at the image
+          # atomically, so we never run a post-push `docker tag` of an image a
+          # concurrent run could have evicted from the shared daemon (issue #1093).
+          if [ -n "$MOVING" ]; then
+            docker build -t "$IMAGE:${{ github.sha }}" -t "$IMAGE:$MOVING" .
+            docker push "$IMAGE:${{ github.sha }}"
+            docker push "$IMAGE:$MOVING"
+          else
+            docker build -t "$IMAGE:${{ github.sha }}" .
+            docker push "$IMAGE:${{ github.sha }}"
           fi
       - name: Prune dangling layers
         if: always()

docs/plugins/host-agent-design.md

@@ -74,8 +74,15 @@ token = a1b2c3d4...
 interval_seconds = 30
 hostname = myhost              # optional; defaults to uname -n
 mounts = /, /mnt/data          # optional; defaults to all real mounts (excluding tmpfs/devtmpfs/etc.)
+docker_logs_enabled = true     # optional; on by default — per-host container-log kill-switch
+docker_log_exclude = watchtower, noisy-svc   # optional; container names whose logs this host skips collecting
 ```
 
+Container-log collection is on by default (no config needed). The two keys above
+are the per-host opt-outs; the operator-facing global toggle + exclude live in
+Settings and are enforced server-side (the push model has no channel to command
+an agent), so they don't require touching a host's conf.
+
 ### Agent internals (function list, not classes)
 
 - `read_config(path)` — parses the conf file into a dict.
@@ -85,10 +92,17 @@ mounts = /, /mnt/data          # optional; defaults to all real mounts (excludin
 - `collect_load()` — reads `/proc/loadavg`, returns `[1m, 5m, 15m]`.
 - `collect_uptime()` — reads `/proc/uptime`, returns seconds since boot (int).
 - `collect_metadata()` — `os.uname()` for kernel + arch, `/etc/os-release` for distro. Called once at startup and cached.
+- `collect_docker_logs(socket, containers, state, exclude)` — (m79) per running
+  container, fetches new log lines over the Docker socket with an incremental
+  `since` cursor kept per container in `state` (a container's first interval
+  seeds from a short tail). Demuxes the multiplexed stream, parses each line's
+  RFC3339 timestamp, caps the whole batch at a byte limit (a marker line records
+  a truncation; the deferred lines come next interval). Folded into the sample as
+  `docker_logs`; omitted when empty.
 - `build_payload()` — assembles a snapshot from all collectors into one dict.
 - `post_payload(url, token, payloads)` — POSTs a list of samples, returns success/failure.
 - `RingBuffer(maxlen=20)` — tiny FIFO wrapper, drops oldest when full.
-- `main_loop()` — the 30s loop: collect → try POST → on failure push to buffer + backoff → on success flush buffer.
+- `main_loop()` — the 30s loop: collect → try POST → on failure push to buffer + backoff → on success flush buffer. Container logs are stripped from a sample before it's buffered (metrics survive an outage; stale logs are dropped).
 
 **Target: ~300 lines total including docstrings.** More than that is a smell that the agent is over-scoping.
 
@@ -255,6 +269,7 @@ Content-Type: application/json
 - **`metadata` is sent on every POST**, not just on change. Server-side diff detects actual changes and only writes on change. Cost per POST is one dict — negligible. Benefit: server can cleanly detect agent restarts.
 - **Raw bytes, not percentages, for memory and storage.** Percentages are derived server-side. Changing the "what counts as used" math doesn't require re-releasing the agent.
 - **CPU is the one exception** — reported as a percentage because it's inherently a derivative (delta over time), not a snapshot. The agent must sample twice to compute it.
+- **Container logs (m79) ride in the same push** as `docker_logs`: a list of `{container, stream, ts, line}` records — incremental since the previous interval. This pushes the *same direction* as metrics, so batched log history needs no inbound channel (only sub-second live-follow would). The server ingests them into `docker_logs`, enforces the global toggle + exclude list on ingest, and bounds storage with a per-container size+age ring. `docker_logs` is omitted when there's nothing new.
 
 ### Server expansion into `PluginMetric` rows
 

plugins/docker/ingest.py

@@ -92,6 +92,55 @@ 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. The global
+    toggle + exclude list (Settings, no restart) are enforced here: this is the
+    authoritative drop point, since the push model has no channel to tell an
+    agent to stop collecting.
+    """
+    from steward.core.settings import get_setting
+    from .models import DockerLog
+
+    if not await get_setting(session, "docker.logs.enabled"):
+        return
+    exclude = set(await get_setting(session, "docker.logs.exclude") or ())
+    for row in _log_rows(log_batches, host.id):
+        if row["container_name"] in exclude:
+            continue
+        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 +250,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 +261,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 +272,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/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.
 

plugins/docker/retention.py

@@ -38,6 +38,8 @@ async def run_docker_retention(
     events_days: int,
     metrics_raw_days: int,
     metrics_rollup_days: int,
+    logs_retention_days: int = 3,
+    logs_max_bytes_per_container: int = 5_000_000,
     now: datetime | None = None,
 ) -> dict:
     """Roll up + prune Docker time-series. Returns a counts dict for logging.
@@ -47,18 +49,22 @@ async def run_docker_retention(
        re-run is idempotent, then delete those raw rows.
     2. Prune rolled-up rows older than the rollup window.
     3. Prune docker_events older than the events window.
+    4. Prune docker_logs with a per-container size+age ring (m79): drop lines
+       older than the age window, then keep only the newest ~cap bytes per
+       (host, container).
     """
     from datetime import timezone
 
     from sqlalchemy import delete, func, select
     from sqlalchemy.dialects.postgresql import insert as pg_insert
 
-    from .models import DockerEvent, DockerMetric, DockerMetricHourly
+    from .models import DockerEvent, DockerLog, DockerMetric, DockerMetricHourly
 
     if now is None:
         now = datetime.now(timezone.utc)
 
     rolled = rolled_rows = events_pruned = rollup_pruned = 0
+    logs_age_pruned = logs_size_pruned = 0
 
     # ── 1. Roll up raw metrics older than the raw window into hourly buckets ──
     raw_cutoff = _rollup_cutoff(now, metrics_raw_days)
@@ -120,9 +126,35 @@ async def run_docker_retention(
     )
     events_pruned = res.rowcount or 0
 
+    # ── 4. Container-log ring: age cutoff, then per-container byte cap (m79) ──
+    logs_cutoff = now - timedelta(days=logs_retention_days)
+    res = await session.execute(
+        delete(DockerLog).where(DockerLog.ts < logs_cutoff)
+    )
+    logs_age_pruned = res.rowcount or 0
+
+    # Size ring: per (host, container), sum line bytes newest-first; delete a row
+    # once its strictly-newer siblings already fill the cap. Using the EXCLUSIVE
+    # prefix (running total minus this row) means the newest row always survives,
+    # so a single line larger than the cap is never wiped out.
+    running = func.sum(func.length(DockerLog.line)).over(
+        partition_by=[DockerLog.host_id, DockerLog.container_name],
+        order_by=[DockerLog.ts.desc(), DockerLog.id.desc()],
+    )
+    prefix_excl = (running - func.length(DockerLog.line)).label("prefix_excl")
+    ranked = select(DockerLog.id, prefix_excl).subquery()
+    over_cap = select(ranked.c.id).where(
+        ranked.c.prefix_excl >= logs_max_bytes_per_container)
+    res = await session.execute(
+        delete(DockerLog).where(DockerLog.id.in_(over_cap))
+    )
+    logs_size_pruned = res.rowcount or 0
+
     return {
         "buckets_rolled": rolled,
         "raw_rows_rolled": rolled_rows,
         "rollup_pruned": rollup_pruned,
         "events_pruned": events_pruned,
+        "logs_age_pruned": logs_age_pruned,
+        "logs_size_pruned": logs_size_pruned,
     }

plugins/docker/routes.py

@@ -3,7 +3,10 @@ from __future__ import annotations
 import json
 from datetime import datetime, timedelta, timezone
 
-from quart import Blueprint, current_app, render_template, request
+from quart import (
+    Blueprint, current_app, jsonify, redirect, render_template, request,
+    session, url_for,
+)
 from sqlalchemy import func, select
 
 from steward.auth.middleware import require_role
@@ -13,13 +16,31 @@ from steward.core.time_range import parse_range, DEFAULT_RANGE
 from .dedup import dedup_by_container_id
 from .swarm_view import build_swarm_services
 from .models import (
-    DockerContainer, DockerDiskUsage, DockerEvent, DockerImage, DockerMetric,
-    DockerSwarmNode, DockerSwarmService,
+    DockerContainer, DockerDiskUsage, DockerEvent, DockerImage, DockerLog,
+    DockerMetric, DockerSwarmNode, DockerSwarmService,
 )
 
 docker_bp = Blueprint("docker", __name__, template_folder="templates")
 
 
+def _error(status: int, code: str, detail: str | None = None):
+    body: dict = {"ok": False, "error": code}
+    if detail:
+        body["detail"] = detail
+    return jsonify(body), status
+
+
+def _prune_extra_vars(prune_target: str) -> dict:
+    """Extra-vars for the bundled docker_prune playbook. "Prune unused images"
+    means ALL unused (docker image prune -a), not just dangling; the full system
+    prune stays conservative (-f, no -a) per the operator's choice (m78). Pure
+    helper so the mapping is unit-tested without the request/DB stack."""
+    extra_vars: dict = {"prune_target": prune_target}
+    if prune_target == "images":
+        extra_vars["prune_all_images"] = True
+    return extra_vars
+
+
 def _human_bytes(n: int | None) -> str:
     """Compact binary size string (e.g. '1.4 GiB', '512 MiB', '0 B')."""
     if n is None:
@@ -499,9 +520,13 @@ async def swarm():
 async def disk():
     """Image/disk usage page: reclaimable space, per-image sizes, stopped count.
 
-    Read-only — prune actions are deferred to the cleanup-actions milestone, so
-    this surfaces the numbers and notes where reclaim lives.
+    Admins can reclaim space per host via the prune buttons, which fire the
+    audited bundled prune playbook through Ansible (m78) — the collection agent
+    itself stays read-only.
     """
+    from steward.core.capabilities import has_capability
+    from steward.models.ansible_inventory import AnsibleTarget
+
     async with current_app.db_sessionmaker() as db:
         summaries = list((await db.execute(select(DockerDiskUsage))).scalars())
         images = list((await db.execute(
@@ -516,6 +541,16 @@ async def disk():
         for hid in stopped_rows:
             stopped_by_host[hid] = stopped_by_host.get(hid, 0) + 1
         hosts = await _host_map(db, {s.host_id for s in summaries})
+        # Which hosts have a linked Ansible target — gates the prune buttons
+        # (a target is required to route the playbook run to that host).
+        host_ids = {s.host_id for s in summaries}
+        target_by_host: dict[str, str] = {}
+        if host_ids:
+            for hid, tid in (await db.execute(
+                select(AnsibleTarget.host_id, AnsibleTarget.id)
+                .where(AnsibleTarget.host_id.in_(host_ids))
+            )).all():
+                target_by_host[hid] = tid
 
     images_by_host: dict[str, list] = {}
     for im in images:
@@ -542,11 +577,60 @@ async def disk():
             },
             "stopped": stopped_by_host.get(s.host_id, 0),
             "images": images_by_host.get(s.host_id, []),
+            "has_target": s.host_id in target_by_host,
         }
         for s in summaries
     ]
     host_groups.sort(key=lambda g: g["host_name"].lower())
-    return await render_template("docker/disk.html", host_groups=host_groups)
+    return await render_template(
+        "docker/disk.html", host_groups=host_groups,
+        ansible_available=has_capability("ansible.run_playbook"),
+    )
+
+
+@docker_bp.post("/disk/<host_id>/prune")
+@require_role(UserRole.admin)
+async def disk_prune(host_id: str):
+    """Reclaim Docker disk on a host by firing the bundled prune playbook via
+    Ansible (audited, admin-gated) — the collection agent stays read-only (m78).
+    `target` selects the scope: containers | images | system.
+    """
+    from steward.core.capabilities import has_capability, invoke_capability
+    from steward.ansible.sources import BUILTIN_SOURCE_NAME
+    from steward.models.ansible_inventory import AnsibleTarget
+
+    if not has_capability("ansible.run_playbook"):
+        return _error(400, "ansible_unavailable", "Ansible is not available")
+
+    form = await request.form
+    prune_target = (form.get("target", "") or "").strip()
+    if prune_target not in ("containers", "images", "system"):
+        return _error(400, "bad_target", "Unknown prune target")
+
+    async with current_app.db_sessionmaker() as db:
+        target = (await db.execute(
+            select(AnsibleTarget).where(AnsibleTarget.host_id == host_id)
+        )).scalar_one_or_none()
+    if target is None:
+        return _error(400, "no_target",
+                      "Link an Ansible target to this host before pruning")
+
+    # extra_vars_map outranks the playbook's own `vars:` defaults.
+    extra_vars = _prune_extra_vars(prune_target)
+
+    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="maintenance/docker_prune.yml",
+        inventory_scope=f"steward:target:{target.id}",
+        params={"extra_vars_map": extra_vars},
+        triggered_by=session.get("user_id"),
+    )
+    if err:
+        return _error(400, "prune_failed", err)
+    return redirect(url_for("ansible.run_detail", run_id=run.id))
 
 
 @docker_bp.get("/container/<host_id>/<name>/history")
@@ -563,3 +647,56 @@ async def container_history(host_id: str, name: str):
         have_data=len(cpu_hist) >= 2,
         range_key=range_key,
     )
+
+
+# Newest lines returned per fetch — the retained window is bounded by the ring,
+# but a container can still hold thousands of lines; cap what one fetch renders.
+_LOG_TAIL_DEFAULT = 500
+
+
+async def _query_logs(db, host_id: str, name: str, stream: str, query: str,
+                      limit: int) -> list:
+    """The recent retained lines for one container, newest-first, optionally
+    filtered by stream and a case-insensitive substring.
+
+    Newest-first is deliberate: the viewer replaces the list on each poll, which
+    resets scroll to the top — so the latest lines stay visible without any
+    scroll handling, and older lines are a scroll away.
+    """
+    stmt = (select(DockerLog.ts, DockerLog.stream, DockerLog.line)
+            .where(DockerLog.host_id == host_id)
+            .where(DockerLog.container_name == name))
+    if stream in ("stdout", "stderr"):
+        stmt = stmt.where(DockerLog.stream == stream)
+    if query:
+        stmt = stmt.where(DockerLog.line.ilike(f"%{query}%"))
+    stmt = stmt.order_by(DockerLog.ts.desc(), DockerLog.id.desc()).limit(limit)
+    return (await db.execute(stmt)).all()
+
+
+@docker_bp.get("/container/<host_id>/<name>/logs")
+@require_role(UserRole.viewer)
+async def container_logs(host_id: str, name: str):
+    """Full log-viewer page for one container (lines load via an HTMX fragment)."""
+    async with current_app.db_sessionmaker() as db:
+        host = await db.get(Host, host_id)
+    return await render_template(
+        "docker/container_logs.html", host=host, host_id=host_id, name=name,
+    )
+
+
+@docker_bp.get("/container/<host_id>/<name>/logs/lines")
+@require_role(UserRole.viewer)
+async def container_logs_lines(host_id: str, name: str):
+    """HTMX fragment: the recent retained log lines, stream + text filtered.
+    Polled every few seconds by the viewer for near-live follow."""
+    stream = request.args.get("stream", "all")
+    query = (request.args.get("q") or "").strip()
+    async with current_app.db_sessionmaker() as db:
+        rows = await _query_logs(db, host_id, name, stream, query, _LOG_TAIL_DEFAULT)
+    return await render_template(
+        "docker/_container_logs_lines.html",
+        lines=[{"ts": r.ts, "stream": r.stream, "line": r.line} for r in rows],
+        name=name, tail=_LOG_TAIL_DEFAULT,
+        filtered=bool(query) or stream in ("stdout", "stderr"),
+    )

plugins/docker/templates/docker/_container_logs_lines.html

@@ -0,0 +1,22 @@
+{# docker/_container_logs_lines.html — log-line fragment, HTMX-polled (m79).
+   {{ l.line }} is auto-escaped by Jinja, so log content can't inject markup. #}
+{% if lines %}
+{% for l in lines %}
+<div style="display:flex;gap:0.6rem;white-space:pre-wrap;word-break:break-word;">
+  <span style="color:var(--text-dim);flex-shrink:0;" title="{{ l.ts }}">{{ l.ts.strftime("%m-%d %H:%M:%S") }}</span>
+  <span style="flex-shrink:0;width:3.2rem;color:{% if l.stream == 'stderr' %}var(--red){% else %}var(--text-muted){% endif %};">{{ l.stream }}</span>
+  <span style="flex:1;min-width:0;{% if l.stream == 'stderr' %}color:var(--text);{% endif %}">{{ l.line }}</span>
+</div>
+{% endfor %}
+{% else %}
+<div style="color:var(--text-muted);padding:1rem 0.25rem;">
+  {% if filtered %}
+  No lines match the current filter.
+  {% else %}
+  No logs collected yet for <code>{{ name }}</code>. Lines appear within a few
+  seconds of the container writing to stdout/stderr — unless it's on the log
+  exclude list or log collection is turned off in
+  <a href="/settings/thresholds/">Settings → Thresholds &amp; Retention</a>.
+  {% endif %}
+</div>
+{% endif %}

plugins/docker/templates/docker/container_detail.html

@@ -26,6 +26,7 @@
 <div style="font-size:0.82rem;color:var(--text-muted);margin-bottom:1.5rem;">
   {{ container.status }}{% if uptime %} · up {{ uptime }}{% endif %}
   {% if host %} · on <a href="/hosts/{{ host.id }}" style="color:var(--text-muted);">{{ host.name }}</a>{% endif %}
+  · <a href="/plugins/docker/container/{{ host_id }}/{{ name }}/logs">View logs</a>
 </div>
 
 {# ── Facts grid ──────────────────────────────────────────────────────────── #}

plugins/docker/templates/docker/container_logs.html

@@ -0,0 +1,51 @@
+{# docker/container_logs.html — per-container log viewer (m79) #}
+{% extends "base.html" %}
+{% from "_macros.html" import crumbs %}
+{% block title %}Logs — {{ name }} — Docker — Steward{% endblock %}
+{% block breadcrumb %}{{ crumbs([
+  ("Docker", "/plugins/docker/"),
+  (name, "/plugins/docker/container/" ~ host_id ~ "/" ~ name),
+  ("Logs", "")]) }}{% endblock %}
+{% block content %}
+
+<div style="display:flex;align-items:baseline;gap:0.6rem;margin-bottom:0.35rem;flex-wrap:wrap;">
+  <h1 class="page-title" style="margin-bottom:0;">{{ name }}</h1>
+  <span style="font-size:0.9rem;color:var(--text-muted);">logs</span>
+</div>
+<div style="font-size:0.82rem;color:var(--text-muted);margin-bottom:1rem;">
+  Recent lines (newest first) collected by the host agent{% if host %} on
+  <a href="/hosts/{{ host.id }}" style="color:var(--text-muted);">{{ host.name }}</a>{% endif %} —
+  updated every few seconds.
+  <a href="/plugins/docker/container/{{ host_id }}/{{ name }}">← back to container</a>
+</div>
+
+{# ── Controls: stream filter + text search (drive the fragment via HTMX) ───── #}
+<form id="log-controls" onsubmit="return false;"
+      style="display:flex;gap:0.6rem;align-items:center;flex-wrap:wrap;margin-bottom:0.6rem;">
+  <label style="font-size:0.8rem;color:var(--text-muted);display:flex;align-items:center;gap:0.35rem;">
+    Stream
+    <select name="stream" style="padding:0.25rem 0.4rem;">
+      <option value="all">all</option>
+      <option value="stdout">stdout</option>
+      <option value="stderr">stderr</option>
+    </select>
+  </label>
+  <input type="search" name="q" placeholder="Filter lines…" autocomplete="off"
+         aria-label="Filter log lines"
+         style="flex:1;min-width:180px;padding:0.3rem 0.5rem;">
+</form>
+
+<div class="card-flush">
+  <div id="log-lines"
+       hx-get="/plugins/docker/container/{{ host_id }}/{{ name }}/logs/lines"
+       hx-trigger="load, every 5s, change from:#log-controls, keyup changed delay:400ms from:#log-controls"
+       hx-include="#log-controls"
+       hx-swap="innerHTML"
+       role="log" aria-live="polite" tabindex="0"
+       style="max-height:70vh;overflow:auto;padding:0.5rem 0.75rem;
+              font-family:ui-monospace,monospace;font-size:0.8rem;line-height:1.5;">
+    <div style="color:var(--text-muted);">Loading…</div>
+  </div>
+</div>
+
+{% endblock %}

plugins/docker/templates/docker/disk.html

@@ -7,8 +7,8 @@
 
 <h1 class="page-title" style="margin-bottom:0.4rem;">Image &amp; disk usage</h1>
 <p style="font-size:0.82rem;color:var(--text-muted);margin-bottom:1.5rem;">
-  Reclaimable = space held by images no container references. Cleanup actions
-  (prune) arrive in a later release — these are read-only figures for now.
+  Reclaimable = space held by images no container references. Admins can prune
+  per host below; each action runs an audited Ansible playbook on that host.
 </p>
 
 {% if host_groups %}
@@ -47,6 +47,43 @@
     </div>
   </div>
 
+  {# ── Cleanup actions (admin only; audited Ansible prune run) ───────────── #}
+  {% if session.user_role == 'admin' %}
+  <div style="margin-bottom:1rem;">
+    {% if ansible_available and g.has_target %}
+    <div style="display:flex;flex-wrap:wrap;gap:0.5rem;align-items:center;">
+      <form method="post" action="/plugins/docker/disk/{{ g.host_id }}/prune" style="margin:0;"
+            data-msg="Remove all STOPPED containers on {{ g.host_name|e }}? This cannot be undone."
+            onsubmit="return confirm(this.dataset.msg);">
+        <input type="hidden" name="target" value="containers">
+        <button type="submit" class="btn btn-sm">Prune stopped containers</button>
+      </form>
+      <form method="post" action="/plugins/docker/disk/{{ g.host_id }}/prune" style="margin:0;"
+            data-msg="Remove ALL unused images on {{ g.host_name|e }} (docker image prune -a)? Any image not used by a container is deleted."
+            onsubmit="return confirm(this.dataset.msg);">
+        <input type="hidden" name="target" value="images">
+        <button type="submit" class="btn btn-sm">Prune unused images</button>
+      </form>
+      <form method="post" action="/plugins/docker/disk/{{ g.host_id }}/prune" style="margin:0;"
+            data-msg="Run a full system prune on {{ g.host_name|e }}? Removes stopped containers, unused networks, dangling images and build cache."
+            onsubmit="return confirm(this.dataset.msg);">
+        <input type="hidden" name="target" value="system">
+        <button type="submit" class="btn btn-sm btn-danger">System prune…</button>
+      </form>
+    </div>
+    <div style="font-size:0.72rem;color:var(--text-muted);margin-top:0.4rem;">
+      Reclaimed space appears on the next agent sample.
+    </div>
+    {% elif not ansible_available %}
+    <div style="font-size:0.74rem;color:var(--text-muted);">Cleanup needs the Ansible runner (currently unavailable).</div>
+    {% else %}
+    <div style="font-size:0.74rem;color:var(--text-muted);">
+      Link an <a href="/hosts/{{ g.host_id }}">Ansible target</a> to this host to enable prune actions.
+    </div>
+    {% endif %}
+  </div>
+  {% endif %}
+
   {# ── Per-image table ──────────────────────────────────────────────────── #}
   <div class="card-flush">
     <table class="table">

plugins/host_agent/agent.py

@@ -20,7 +20,7 @@ from collections import deque
 from concurrent.futures import ThreadPoolExecutor
 from datetime import datetime, timezone
 
-AGENT_VERSION = "1.6.0"
+AGENT_VERSION = "1.7.0"
 
 # Default path to the local Docker Engine socket. Overridable via the
 # `docker_socket` config key; collection is silently skipped if it's absent or
@@ -34,7 +34,10 @@ class ConfigError(Exception):
 
 REQUIRED_KEYS = ("url", "token")
 INT_KEYS = ("interval_seconds",)
-LIST_KEYS = ("mounts",)
+LIST_KEYS = ("mounts", "docker_log_exclude")
+# Truthy strings for bool keys; anything else (incl. empty) is False.
+BOOL_KEYS = ("docker_logs_enabled",)
+_BOOL_TRUE = ("1", "true", "yes", "on")
 
 
 def read_config(path: str) -> dict:
@@ -58,6 +61,8 @@ def read_config(path: str) -> dict:
                         raise ConfigError(f"{path}:{lineno}: {key} must be int")
                 elif key in LIST_KEYS:
                     cfg[key] = [v.strip() for v in value.split(",") if v.strip()]
+                elif key in BOOL_KEYS:
+                    cfg[key] = value.lower() in _BOOL_TRUE
                 else:
                     cfg[key] = value
     except FileNotFoundError:
@@ -69,6 +74,10 @@ def read_config(path: str) -> dict:
 
     cfg.setdefault("interval_seconds", 30)
     cfg.setdefault("docker_socket", DEFAULT_DOCKER_SOCKET)
+    # Container-log collection is on by default (operator preference); a host can
+    # opt out with `docker_logs_enabled = false` or thin it with docker_log_exclude.
+    cfg.setdefault("docker_logs_enabled", True)
+    cfg.setdefault("docker_log_exclude", [])
     return cfg
 
 
@@ -403,6 +412,15 @@ def _rates(cur: dict, prev: dict, dt: float) -> dict:
 
 DOCKER_API_TIMEOUT = 5.0
 
+# Container-log collection (m79). Logs ride in the same push as metrics, so one
+# interval's batch is capped to keep a chatty container from bloating a POST (and
+# the backoff ring): once the cap is hit a marker line is emitted and collection
+# stops — the deferred lines come on the next interval. On a container's first
+# interval we seed from a short tail, then switch to an incremental since-cursor
+# kept per container in the agent's rate-state.
+DOCKER_LOG_BATCH_MAX_BYTES = 262144   # 256 KiB of log text per push, all containers
+DOCKER_LOG_FIRST_TAIL = 50            # lines seeded on a container's first interval
+
 
 def _dechunk(body: bytes) -> bytes:
     """Decode an HTTP/1.1 chunked-transfer body into the raw payload."""
@@ -650,6 +668,198 @@ def collect_docker(socket_path: str) -> list:
         return list(ex.map(lambda c: _collect_one_container(socket_path, c), containers))
 
 
+# ─── container logs (m79) ─────────────────────────────────────────────────────
+#
+# The logs endpoint returns a raw multiplexed byte stream, not JSON, so it needs
+# a sibling of _docker_request that hands back bytes. Non-TTY containers frame
+# stdout/stderr with an 8-byte header (stream byte + 4-byte big-endian length);
+# TTY containers emit unframed bytes. We request timestamps=1, so every line is
+# prefixed with an RFC3339Nano timestamp we parse for the since-cursor.
+
+_LOG_STREAM_NAMES = {0: "stdout", 1: "stdout", 2: "stderr"}
+
+
+def _docker_request_raw(socket_path: str, path: str,
+                        timeout: float = DOCKER_API_TIMEOUT) -> bytes:
+    """GET `path` from the Docker API over the Unix socket; return the raw body.
+
+    Sibling of `_docker_request` for non-JSON endpoints (container logs). Reuses
+    the connect / send / de-chunk scaffolding; raises OSError on any transport
+    problem or non-2xx status so callers can silent-skip. A 200 with an empty
+    body (no new lines since the cursor) returns b"".
+    """
+    sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+    sock.settimeout(timeout)
+    try:
+        sock.connect(socket_path)
+        req = (
+            "GET " + path + " HTTP/1.1\r\n"
+            "Host: docker\r\n"
+            "Accept: application/octet-stream\r\n"
+            "Connection: close\r\n"
+            "\r\n"
+        )
+        sock.sendall(req.encode("ascii"))
+        chunks = []
+        while True:
+            buf = sock.recv(65536)
+            if not buf:
+                break
+            chunks.append(buf)
+    finally:
+        sock.close()
+
+    raw = b"".join(chunks)
+    head, _, body = raw.partition(b"\r\n\r\n")
+    header_text = head.decode("latin-1")
+    status_line = header_text.split("\r\n", 1)[0]
+    parts = status_line.split(None, 2)
+    status = int(parts[1]) if len(parts) >= 2 and parts[1].isdigit() else 0
+    if not (200 <= status < 300):
+        raise OSError(f"docker API {path} returned {status}")
+    if "transfer-encoding: chunked" in header_text.lower():
+        body = _dechunk(body)
+    return body
+
+
+def _demux_docker_logs(raw: bytes) -> list:
+    """Split a Docker logs stream into [(stream_name, payload_bytes), …].
+
+    Non-TTY containers multiplex with an 8-byte frame header (stream byte in
+    {0,1,2} + 4-byte big-endian length). The stream is treated as framed only if
+    the whole buffer parses as clean back-to-back frames; anything else (a TTY
+    container's raw stream, or a malformed header) falls back to a single stdout
+    blob so no bytes are lost.
+    """
+    if not raw:
+        return []
+    frames = []
+    i, n = 0, len(raw)
+    while i + 8 <= n:
+        stream = raw[i]
+        if stream > 2 or raw[i + 1] or raw[i + 2] or raw[i + 3]:
+            break  # not a valid frame header → not framed
+        size = int.from_bytes(raw[i + 4:i + 8], "big")
+        if i + 8 + size > n:
+            break  # frame overruns the buffer → not framed
+        frames.append((_LOG_STREAM_NAMES.get(stream, "stdout"),
+                       raw[i + 8:i + 8 + size]))
+        i += 8 + size
+    if frames and i == n:
+        return frames
+    return [("stdout", raw)]
+
+
+def _parse_log_ts(token: str):
+    """Parse a Docker RFC3339Nano timestamp token into a datetime, or None.
+
+    Normalises `Z` → `+00:00` and trims Docker's nanosecond precision to the
+    microseconds datetime.fromisoformat accepts (stdlib, Python 3.8+ safe).
+    """
+    t = token.strip()
+    if not t:
+        return None
+    if t.endswith("Z"):
+        t = t[:-1] + "+00:00"
+    m = re.match(r"^(\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2})(\.\d+)?(.*)$", t)
+    if m:
+        frac = m.group(2) or ""
+        if len(frac) > 7:               # "." + up to 6 fractional digits
+            frac = frac[:7]
+        t = m.group(1) + frac + m.group(3)
+    try:
+        return datetime.fromisoformat(t)
+    except ValueError:
+        return None
+
+
+def _parse_container_logs(raw: bytes) -> list:
+    """Demux + line-split a logs stream → [(dt|None, stream, line), …] in order.
+
+    Each line is timestamp-prefixed (we request timestamps=1); an unparseable
+    prefix yields dt=None with the full raw line kept.
+    """
+    out = []
+    for stream, payload in _demux_docker_logs(raw):
+        text = payload.decode("utf-8", "replace")
+        for raw_line in text.split("\n"):
+            if not raw_line:
+                continue
+            token, _, msg = raw_line.partition(" ")
+            dt = _parse_log_ts(token)
+            if dt is not None:
+                out.append((dt, stream, msg))
+            else:
+                out.append((None, stream, raw_line))
+    return out
+
+
+def collect_docker_logs(socket_path: str, containers: list, state: dict,
+                        exclude=None,
+                        max_bytes: int = DOCKER_LOG_BATCH_MAX_BYTES) -> list:
+    """New container-log lines since the last interval, for running containers.
+
+    Keeps a per-container since-cursor in `state["docker_log_cursors"]`
+    (name → unix ts) so each interval fetches only what's new; a container's
+    first interval seeds from a short tail. The whole batch is capped at
+    `max_bytes` of log text — once exceeded a truncation marker is appended and
+    collection stops (logs must never balloon a push). `exclude` container names
+    are skipped entirely (a per-host bandwidth opt-out). Best-effort: a
+    per-container transport error just contributes nothing this interval.
+    Returns [{"container", "stream", "ts", "line"}, …].
+    """
+    exclude_set = set(exclude or ())
+    cursors = state.setdefault("docker_log_cursors", {})
+    running = {c.get("name") for c in (containers or [])
+               if c.get("name") and c.get("status") == "running"}
+    # Forget cursors for containers no longer running so state can't grow without
+    # bound over the agent's lifetime.
+    for gone in [k for k in cursors if k not in running]:
+        del cursors[gone]
+
+    out: list = []
+    total = 0
+    truncated = False
+    for c in (containers or []):
+        name = c.get("name")
+        if not name or name not in running or name in exclude_set:
+            continue
+        prev = cursors.get(name)
+        base = f"/containers/{name}/logs?stdout=1&stderr=1&timestamps=1"
+        # since accepts a fractional Unix ts; we still dedup the boundary line.
+        path = (f"{base}&tail={DOCKER_LOG_FIRST_TAIL}" if prev is None
+                else f"{base}&since={prev:.9f}")
+        try:
+            raw = _docker_request_raw(socket_path, path)
+        except (OSError, ValueError):
+            continue
+        newest = prev
+        for dt, stream, line in _parse_container_logs(raw):
+            ts_unix = dt.timestamp() if dt is not None else None
+            if prev is not None and ts_unix is not None and ts_unix <= prev:
+                continue  # boundary line already sent last interval
+            if total >= max_bytes:
+                truncated = True
+                break
+            out.append({"container": name, "stream": stream,
+                        "ts": dt.isoformat() if dt is not None else None,
+                        "line": line})
+            total += len(line)
+            if ts_unix is not None and (newest is None or ts_unix > newest):
+                newest = ts_unix
+        if newest is not None:
+            cursors[name] = newest
+        if truncated:
+            break
+
+    if truncated:
+        out.append({"container": "_steward", "stream": "stderr",
+                    "ts": datetime.now(timezone.utc).isoformat(),
+                    "line": (f"[steward] log batch truncated at {max_bytes} bytes; "
+                             "remaining lines deferred to the next interval")})
+    return out
+
+
 # ─── swarm (manager-only) ────────────────────────────────────────────────────
 
 
@@ -852,13 +1062,17 @@ class RingBuffer:
 
 
 def build_sample(mounts: list[str], state: dict,
-                 docker_socket: str = DEFAULT_DOCKER_SOCKET) -> dict:
+                 docker_socket: str = DEFAULT_DOCKER_SOCKET, *,
+                 docker_logs_enabled: bool = True,
+                 docker_log_exclude=None) -> dict:
     """Collect one full sample. Partial samples allowed if a collector fails.
 
     `state` carries the previous network/disk counters + monotonic timestamp so
-    throughput rates can be derived from deltas; it is mutated in place.
-    `docker_socket` is probed best-effort — the `docker` key is omitted entirely
-    when no containers are found, so non-Docker hosts add nothing to the payload.
+    throughput rates can be derived from deltas (and the per-container log
+    since-cursors); it is mutated in place. `docker_socket` is probed
+    best-effort — the `docker` key is omitted entirely when no containers are
+    found, so non-Docker hosts add nothing to the payload. `docker_logs_enabled`
+    / `docker_log_exclude` gate incremental container-log collection.
     """
     sample: dict = {"ts": datetime.now(timezone.utc).isoformat()}
     try:
@@ -940,6 +1154,18 @@ def build_sample(mounts: list[str], state: dict,
         if disk is not None:
             sample["docker_disk"] = disk
 
+    # Container logs (m79): incremental per-container tail folded into the same
+    # push as metrics. Skipped when there are no containers or logging is off;
+    # the key is omitted when nothing new arrived this interval.
+    if docker and docker_logs_enabled:
+        try:
+            logs = collect_docker_logs(docker_socket, docker, state,
+                                       exclude=docker_log_exclude)
+        except Exception:
+            logs = []
+        if logs:
+            sample["docker_logs"] = logs
+
     return sample
 
 
@@ -958,6 +1184,18 @@ def build_payload(samples: list[dict], hostname: str, metadata: dict) -> dict:
 BACKOFF_CAP = 300
 
 
+def _drop_logs(sample: dict) -> dict:
+    """Strip container logs from a sample before it's buffered for retry.
+
+    Logs are the one payload we never carry across a backoff — they'd bloat the
+    ring buffer and go stale — while metrics are kept so an outage doesn't lose
+    them. The since-cursor already advanced when the logs were collected, so the
+    dropped lines are simply not re-sent (accepted loss during an outage).
+    """
+    sample.pop("docker_logs", None)
+    return sample
+
+
 def next_backoff(current: int) -> int:
     if current <= 0:
         return 30
@@ -1022,6 +1260,8 @@ def main_loop(conf_path: str) -> int:
     hostname = cfg.get("hostname") or socket.gethostname()
     mounts = cfg.get("mounts") or default_mounts()
     docker_socket = cfg.get("docker_socket") or DEFAULT_DOCKER_SOCKET
+    docker_logs_enabled = cfg.get("docker_logs_enabled", True)
+    docker_log_exclude = cfg.get("docker_log_exclude") or []
     buffer = RingBuffer(maxlen=20)
     backoff = 0
     # Carries previous net/disk counters + monotonic ts for rate computation.
@@ -1036,13 +1276,17 @@ def main_loop(conf_path: str) -> int:
                 cfg = read_config(conf_path)
                 mounts = cfg.get("mounts") or default_mounts()
                 docker_socket = cfg.get("docker_socket") or DEFAULT_DOCKER_SOCKET
+                docker_logs_enabled = cfg.get("docker_logs_enabled", True)
+                docker_log_exclude = cfg.get("docker_log_exclude") or []
                 metadata = collect_metadata()  # refresh host_ip/distro on reload
                 _log("INFO", "config reloaded")
             except ConfigError as e:
                 _log("ERROR", f"reload failed: {e}")
             _reload_requested = False
 
-        sample = build_sample(mounts, rate_state, docker_socket)
+        sample = build_sample(mounts, rate_state, docker_socket,
+                              docker_logs_enabled=docker_logs_enabled,
+                              docker_log_exclude=docker_log_exclude)
         buffered = buffer.drain()
         payload = build_payload(
             samples=buffered + [sample],
@@ -1061,12 +1305,12 @@ def main_loop(conf_path: str) -> int:
                 _log("ERROR", "server rejected payload (400) — dropping sample")
             elif status == 401:
                 _log("ERROR", "token rejected (401) — check config + UI")
-                buffer.push(sample)
+                buffer.push(_drop_logs(sample))
             else:
                 _log("WARN", f"POST failed (status={status}); buffering")
                 for s in buffered:
-                    buffer.push(s)
-                buffer.push(sample)
+                    buffer.push(_drop_logs(s))
+                buffer.push(_drop_logs(sample))
             backoff = next_backoff(backoff)
             sleep_for = backoff
 

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(

plugins/host_agent/templates/install.sh.j2

@@ -45,6 +45,10 @@ cat > "$CONF_FILE" <<EOF
 url = $STEWARD_URL
 token = $AGENT_TOKEN
 interval_seconds = 30
+# Container logs are collected by default. To opt this host out entirely:
+#   docker_logs_enabled = false
+# To skip specific noisy containers (comma-separated names):
+#   docker_log_exclude = watchtower, some-chatty-service
 EOF
 chown "root:$AGENT_USER" "$CONF_FILE"
 chmod 0640 "$CONF_FILE"

steward/ansible/bundled/maintenance/docker_prune.yml

@@ -1,29 +1,64 @@
 ---
-# description: Reclaim disk on Docker / Swarm nodes by pruning unused images, containers, networks and build cache.
+# description: Reclaim disk on Docker / Swarm nodes — prune stopped containers, unused images, or a full system prune.
 # steward:category: maintenance
 # steward:confirm: true
 # Reclaim disk on Docker / Docker Swarm nodes by removing unused data.
-# Safe by default: prunes dangling images, stopped containers, unused networks
-# and build cache. Set extra-vars to widen scope:
-#   prune_all_images=true   also remove ALL unused images (not just dangling)
-#   prune_volumes=true      also remove unused named volumes (data loss risk)
-- name: Docker system prune
+# `prune_target` selects the scope (default `system` preserves the original
+# behavior for existing manual/scheduled callers that don't set it):
+#   prune_target=containers   remove stopped containers only (docker container prune)
+#   prune_target=images       remove unused images (docker image prune;
+#                             + prune_all_images=true → -a, i.e. ALL unused, not just dangling)
+#   prune_target=system       docker system prune (dangling images, stopped
+#                             containers, unused networks, build cache). Widen with:
+#                               prune_all_images=true   also ALL unused images
+#                               prune_volumes=true       also unused named volumes (data loss risk)
+- name: Docker prune
   hosts: all
   gather_facts: false
   become: true
   vars:
+    prune_target: system
     prune_all_images: false
     prune_volumes: false
   tasks:
+    - name: Validate prune_target
+      ansible.builtin.assert:
+        that: prune_target in ['containers', 'images', 'system']
+        fail_msg: "prune_target must be one of: containers, images, system (got '{{ prune_target }}')"
+        quiet: true
+
+    # ── Stopped containers only ───────────────────────────────────────────────
+    - name: Prune stopped containers
+      ansible.builtin.command:
+        argv: ['docker', 'container', 'prune', '-f']
+      register: container_prune
+      changed_when: "'Total reclaimed space: 0B' not in container_prune.stdout"
+      when: prune_target == 'containers'
+
+    # ── Unused images (dangling, or all unused with -a) ───────────────────────
+    - name: Prune unused images
+      ansible.builtin.command:
+        argv: >-
+          {{ ['docker', 'image', 'prune', '-f']
+             + (['-a'] if prune_all_images | bool else []) }}
+      register: image_prune
+      changed_when: "'Total reclaimed space: 0B' not in image_prune.stdout"
+      when: prune_target == 'images'
+
+    # ── Full system prune (default) ───────────────────────────────────────────
     - name: Run docker system prune
       ansible.builtin.command:
         argv: >-
           {{ ['docker', 'system', 'prune', '-f']
              + (['-a'] if prune_all_images | bool else [])
              + (['--volumes'] if prune_volumes | bool else []) }}
-      register: prune_result
-      changed_when: "'Total reclaimed space: 0B' not in prune_result.stdout"
+      register: system_prune
+      changed_when: "'Total reclaimed space: 0B' not in system_prune.stdout"
+      when: prune_target == 'system'
 
     - name: Report reclaimed space
       ansible.builtin.debug:
-        msg: "{{ prune_result.stdout_lines | select | list }}"
+        msg: >-
+          {{ ((container_prune.stdout_lines | default([]))
+              + (image_prune.stdout_lines | default([]))
+              + (system_prune.stdout_lines | default([]))) | select | list }}

steward/core/cleanup.py

@@ -76,10 +76,14 @@ async def _run_docker_retention(session, now: datetime) -> None:
         raw_days = int(await get_setting(session, "docker.retention.metrics_raw_days") or 7)
         rollup_days = int(await get_setting(session, "docker.retention.metrics_rollup_days") or 90)
         events_days = int(await get_setting(session, "docker.retention.events_days") or 30)
+        logs_days = int(await get_setting(session, "docker.logs.retention_days") or 3)
+        logs_cap = int(
+            await get_setting(session, "docker.logs.max_bytes_per_container") or 5_000_000)
         counts = await invoke_capability(
             "docker.run_retention", UserRole.viewer, session,
             events_days=events_days, metrics_raw_days=raw_days,
-            metrics_rollup_days=rollup_days, now=now,
+            metrics_rollup_days=rollup_days, logs_retention_days=logs_days,
+            logs_max_bytes_per_container=logs_cap, now=now,
         )
     if counts and any(counts.values()):
         logger.info("Docker retention: %s", counts)

steward/core/settings.py

@@ -84,6 +84,14 @@ DEFAULTS: dict[str, Any] = {
     "docker.retention.metrics_raw_days": 7,
     "docker.retention.metrics_rollup_days": 90,
     "docker.retention.events_days": 30,
+    # Container logs (m79): on by default for every container (operator
+    # preference). `exclude` names containers the server drops on ingest; the
+    # per-container ring bounds storage (rotate oldest past whichever of ~age or
+    # ~bytes hits first — a chatty container just keeps a shorter window).
+    "docker.logs.enabled": True,
+    "docker.logs.exclude": [],
+    "docker.logs.retention_days": 3,
+    "docker.logs.max_bytes_per_container": 5_000_000,
     # Host/plugin metrics retention (plugin_metrics): keep a short raw window at
     # the agent's ~30s cadence, then roll up to hourly averages kept much longer.
     "metrics.retention.raw_days": 7,

steward/settings/routes.py

@@ -133,6 +133,7 @@ _RETENTION_FIELDS = [
     ("docker_metrics_raw_days", "docker.retention.metrics_raw_days"),
     ("docker_metrics_rollup_days", "docker.retention.metrics_rollup_days"),
     ("docker_events_days", "docker.retention.events_days"),
+    ("docker_logs_retention_days", "docker.logs.retention_days"),
     ("metrics_raw_days", "metrics.retention.raw_days"),
     ("metrics_rollup_days", "metrics.retention.rollup_days"),
 ]
@@ -170,6 +171,20 @@ async def save_thresholds():
                 except (TypeError, ValueError):
                     continue
                 await set_setting(db, key, val)
+            # Container-log controls (m79). Checkbox: present ⇒ on (this is a
+            # full-page form, so absence is a genuine "off"). Exclude: comma-split
+            # names. Size: entered in MB, stored as bytes.
+            await set_setting(db, "docker.logs.enabled", "docker_logs_enabled" in form)
+            exclude = [n.strip() for n in form.get("docker_logs_exclude", "").split(",")
+                       if n.strip()]
+            await set_setting(db, "docker.logs.exclude", exclude)
+            max_mb = form.get("docker_logs_max_mb", "")
+            if max_mb != "":
+                try:
+                    await set_setting(db, "docker.logs.max_bytes_per_container",
+                                      max(1, int(max_mb)) * 1_000_000)
+                except (TypeError, ValueError):
+                    pass
     await _reload_app_config()
     await log_audit(current_app, session.get("user_id"), session.get("username", ""),
                     "settings.saved", detail={"section": "thresholds"})

steward/templates/settings/thresholds.html

@@ -70,6 +70,54 @@
           "Keep container start/stop/die/health history this long.") }}
 </div>
 
+<div class="card" style="max-width:640px;margin-top:1rem;">
+  <h2 class="section-title" style="margin-bottom:0.5rem;">Container logs</h2>
+  <p style="font-size:0.82rem;color:var(--text-muted);margin-bottom:1.25rem;">
+    The host agent tails each container's logs and pushes them here, viewable per
+    container. On by default for every container; storage is bounded by a
+    per-container ring (oldest lines rotate out once the age or size cap is hit,
+    whichever comes first). Applied by the hourly cleanup and on ingest.
+  </p>
+
+  <div class="form-group" style="margin-bottom:1.1rem;">
+    <label style="display:flex;align-items:center;gap:0.5rem;cursor:pointer;">
+      <input type="checkbox" name="docker_logs_enabled" style="width:auto;"
+             {% if settings["docker.logs.enabled"] %}checked{% endif %}>
+      Collect container logs
+    </label>
+    <div style="font-size:0.78rem;color:var(--text-muted);margin-top:0.3rem;">
+      Global kill-switch. When off, pushed log lines are dropped and no new logs are stored.
+    </div>
+  </div>
+
+  <div class="form-group" style="margin-bottom:1.1rem;">
+    <label>Exclude containers</label>
+    <div style="margin-top:0.25rem;">
+      <input type="text" name="docker_logs_exclude"
+             value="{{ settings['docker.logs.exclude'] | join(', ') }}"
+             placeholder="watchtower, some-chatty-service" style="width:100%;max-width:420px;">
+    </div>
+    <div style="font-size:0.78rem;color:var(--text-muted);margin-top:0.3rem;">
+      Comma-separated container names whose logs are dropped on ingest (e.g. known-noisy ones).
+    </div>
+  </div>
+
+  {{ days("Log retention", "docker_logs_retention_days", "docker.logs.retention_days",
+          "Keep each container's log lines at most this long before rotating them out.") }}
+
+  <div class="form-group" style="margin-bottom:0.25rem;">
+    <label>Max size per container <span style="color:var(--text-muted);font-size:0.8rem;">(MB)</span></label>
+    <div style="margin-top:0.25rem;">
+      <input type="number" name="docker_logs_max_mb" min="1" step="1"
+             value="{{ (settings['docker.logs.max_bytes_per_container'] // 1000000) or 1 }}"
+             style="max-width:110px;">
+    </div>
+    <div style="font-size:0.78rem;color:var(--text-muted);margin-top:0.3rem;">
+      Newest lines are kept up to this size per container; older lines rotate out first.
+    </div>
+  </div>
+</div>
+
 <div class="card" style="max-width:640px;margin-top:1rem;">
   <h2 class="section-title" style="margin-bottom:0.5rem;">Host metrics retention</h2>
   <p style="font-size:0.82rem;color:var(--text-muted);margin-bottom:1.25rem;">

tests/core/test_builtin_source.py

@@ -1,11 +1,21 @@
 """The bundled first-party playbook source is always present and discoverable."""
+from pathlib import Path
+
+import yaml
+
 from steward.ansible.sources import (
     BUILTIN_SOURCE_NAME,
+    discover_playbook_meta,
     discover_playbooks,
     get_sources,
 )
 
 
+def _bundled_content(rel_path: str) -> str:
+    builtin = get_sources({"sources": []})[0]
+    return (Path(builtin["path"]) / rel_path).read_text()
+
+
 def test_builtin_source_is_first_and_local():
     sources = get_sources({"sources": []})
     assert sources[0]["name"] == BUILTIN_SOURCE_NAME
@@ -17,3 +27,33 @@ def test_bundled_playbooks_are_discoverable():
     playbooks = discover_playbooks(builtin["path"])
     assert "maintenance/docker_prune.yml" in playbooks
     assert "host_agent/install.yml" in playbooks
+
+
+def test_docker_prune_playbook_parses_and_keeps_meta():
+    """The prune playbook stays valid YAML and keeps its self-describing meta
+    (a destructive maintenance run that must prompt for confirmation)."""
+    content = _bundled_content("maintenance/docker_prune.yml")
+    plays = yaml.safe_load(content)
+    assert isinstance(plays, list) and plays  # at least one play
+
+    meta = discover_playbook_meta(content)
+    assert meta["confirm"] is True
+    assert meta["category"] == "maintenance"
+    assert meta["description"]
+
+
+def test_docker_prune_supports_all_three_targets():
+    """M78 drives the three prune buttons via a single `prune_target` var;
+    `system` is the default so pre-M78 callers (no var set) are unchanged."""
+    content = _bundled_content("maintenance/docker_prune.yml")
+    play = yaml.safe_load(content)[0]
+    assert play["vars"]["prune_target"] == "system"
+
+    # Every target the routes/UI can send has a matching guarded task.
+    guards = {
+        task.get("when")
+        for task in play["tasks"]
+        if isinstance(task.get("when"), str) and "prune_target" in task["when"]
+    }
+    for target in ("containers", "images", "system"):
+        assert f"prune_target == '{target}'" in guards

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
@@ -150,6 +172,97 @@ 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_persist_logs_respects_toggle_and_exclude(app):
+    """Server-side controls (Settings): an excluded container's lines are dropped
+    while others persist; the global toggle off stores nothing new."""
+    from types import SimpleNamespace
+    from sqlalchemy import text
+    from steward.models.hosts import Host
+    from steward.core.settings import set_setting
+
+    persist = _persist_fn(app)
+    now = datetime.now(timezone.utc)
+    batch = [(now, [
+        {"container": "web", "stream": "stdout", "ts": None, "line": "keep-me"},
+        {"container": "noisy", "stream": "stdout", "ts": None, "line": "drop-me"},
+    ])]
+
+    hid = str(uuid.uuid4())
+    # Stand-in host: the logs-only persist path reads only host.id, so a plain
+    # object sidesteps ORM attribute-expiry across the commit boundaries below.
+    host = SimpleNamespace(id=hid, name="loghost3")
+
+    async def _go():
+        async with app.db_sessionmaker() as s:
+            # Back-to-back begin blocks with NO read in between — a SELECT between
+            # them would autobegin a txn and make the next begin() collide.
+            async with s.begin():
+                await s.execute(text("DELETE FROM docker_logs"))
+                await set_setting(s, "docker.logs.enabled", True)
+                await set_setting(s, "docker.logs.exclude", ["noisy"])
+                s.add(Host(id=hid, name="loghost3", address="10.7.7.10"))
+                await s.flush()   # host row before its log rows (FK order)
+                await persist(s, host, [], None, None, batch)   # web kept, noisy excluded
+            async with s.begin():          # global kill-switch off → nothing stored
+                await set_setting(s, "docker.logs.enabled", False)
+                await persist(s, host, [], None, None, batch)
+            async with s.begin():          # restore defaults for other tests
+                await set_setting(s, "docker.logs.enabled", True)
+                await set_setting(s, "docker.logs.exclude", [])
+            # Reads only after the last begin block (they autobegin, but nothing
+            # opens a begin() after them). Toggle-off added nothing, so the state
+            # here still reflects the exclude phase.
+            kept = {r[0] for r in (await s.execute(text(
+                "SELECT DISTINCT container_name FROM docker_logs WHERE host_id=:h"),
+                {"h": hid})).all()}
+            after_off = (await s.execute(text(
+                "SELECT COUNT(*) FROM docker_logs WHERE host_id=:h"), {"h": hid})).scalar()
+            return kept, after_off
+
+    kept, after_off = asyncio.run(_go())
+    assert kept == {"web"}      # excluded 'noisy' dropped on ingest
+    assert after_off == 1       # toggle off added nothing → still just the one 'web' line
+
+
 @_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
@@ -451,6 +564,67 @@ def test_retention_rollup_and_prune(app):
     assert counts["events_pruned"] == 1 and counts["rollup_pruned"] == 1
 
 
+@_NEEDS_DB
+def test_retention_logs_ring(app):
+    """docker_logs ring (m79): lines past the age window are pruned; within it,
+    each container keeps only the newest ~cap bytes; containers are independent."""
+    from datetime import timedelta
+    from sqlalchemy import text
+    from steward.models.hosts import Host
+
+    run_retention = _retention_fn(app)
+    now = datetime(2026, 6, 19, 12, 0, 0, tzinfo=timezone.utc)
+    line40 = "x" * 40   # 40 chars/line; cap=100 keeps 3 lines (excl-prefix 0/40/80)
+
+    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="loghost2", address="10.7.7.8")
+                s.add(h)
+                await s.flush()
+                hid = h.id
+
+                def ins(cn, ts, line):
+                    return s.execute(text(
+                        "INSERT INTO docker_logs "
+                        "(id, host_id, container_name, ts, stream, line) "
+                        "VALUES (:id,:h,:cn,:ts,'stdout',:line)"),
+                        {"id": str(uuid.uuid4()), "h": hid, "cn": cn,
+                         "ts": ts, "line": line})
+
+                # Past the 3-day age window → age-pruned.
+                await ins("old", now - timedelta(days=10), "z")
+                # 5 recent 'web' lines (40 bytes each = 200 > cap 100) → keep 3.
+                for i in range(1, 6):
+                    await ins("web", now - timedelta(minutes=6 - i), line40)
+                # 2 recent 'db' lines (80 bytes ≤ cap) → both survive (isolation).
+                await ins("db", now - timedelta(minutes=2), line40)
+                await ins("db", now - timedelta(minutes=1), line40)
+
+            async with s.begin():
+                counts = await run_retention(
+                    s, events_days=30, metrics_raw_days=7, metrics_rollup_days=90,
+                    logs_retention_days=3, logs_max_bytes_per_container=100, now=now,
+                )
+
+            def n(cn):
+                return s.execute(text(
+                    "SELECT COUNT(*) FROM docker_logs WHERE host_id=:h AND container_name=:cn"),
+                    {"h": hid, "cn": cn})
+            web = (await n("web")).scalar()
+            db = (await n("db")).scalar()
+            old = (await n("old")).scalar()
+            return counts, web, db, old
+
+    counts, web, db, old = asyncio.run(_go())
+    assert old == 0                       # age window
+    assert web == 3                       # newest ~100 bytes kept, oldest 2 dropped
+    assert db == 2                        # under cap → untouched (per-container)
+    assert counts["logs_age_pruned"] == 1
+    assert counts["logs_size_pruned"] == 2
+
+
 def test_widget_dedup_collapses_cross_manager_duplicates():
     """The same swarm task is reported by every manager (identical container_id);
     the dashboard widget must count it once. Older agents send no container_id,

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")) == []

tests/plugins/docker/test_routes.py

@@ -30,6 +30,28 @@ def test_routes_module_exposes_new_views():
     assert r.docker_bp.name == "docker"
 
 
+def test_disk_prune_view_defined():
+    # M78 admin-gated prune action.
+    assert callable(r.disk_prune)
+
+
+def test_container_log_views_defined():
+    # M79 per-container log viewer + its HTMX-polled line fragment.
+    assert callable(r.container_logs)
+    assert callable(r.container_logs_lines)
+
+
+def test_prune_extra_vars_mapping():
+    """The prune buttons drive one playbook via prune_target; only 'images'
+    widens to ALL unused images (docker image prune -a), system stays -f."""
+    assert r._prune_extra_vars("containers") == {"prune_target": "containers"}
+    assert r._prune_extra_vars("images") == {
+        "prune_target": "images", "prune_all_images": True,
+    }
+    # System prune stays conservative — no prune_all_images key.
+    assert r._prune_extra_vars("system") == {"prune_target": "system"}
+
+
 def test_human_bytes_formats_binary_units():
     assert r._human_bytes(None) == "—"
     assert r._human_bytes(0) == "0 B"

tests/plugins/host_agent/test_agent_docker.py

@@ -401,3 +401,190 @@ def test_read_config_honours_explicit_docker_socket(tmp_path):
         "docker_socket = /run/user/1000/docker.sock\n")
     cfg = a.read_config(str(cfg_file))
     assert cfg["docker_socket"] == "/run/user/1000/docker.sock"
+
+
+# ── container logs (m79) ──────────────────────────────────────────────────────
+
+def _frame(stream: int, payload: bytes) -> bytes:
+    """Build one Docker multiplexed-log frame (8-byte header + payload)."""
+    return bytes([stream, 0, 0, 0]) + len(payload).to_bytes(4, "big") + payload
+
+
+def test_demux_docker_logs_framed():
+    raw = _frame(1, b"out line\n") + _frame(2, b"err line\n")
+    assert a._demux_docker_logs(raw) == [
+        ("stdout", b"out line\n"), ("stderr", b"err line\n")]
+
+
+def test_demux_docker_logs_tty_fallback():
+    # A TTY container's stream isn't framed — the leading byte ('2') is not a
+    # valid stream id, so the whole blob is treated as one stdout payload.
+    raw = b"2023-11-14T12:00:00.000000000Z hi\n"
+    assert a._demux_docker_logs(raw) == [("stdout", raw)]
+
+
+def test_demux_docker_logs_empty():
+    assert a._demux_docker_logs(b"") == []
+
+
+def test_parse_log_ts_handles_nanoseconds_and_z():
+    dt = a._parse_log_ts("2023-11-14T12:00:00.123456789Z")
+    assert dt is not None
+    assert dt.year == 2023 and dt.microsecond == 123456   # ns trimmed to µs
+    assert dt.utcoffset().total_seconds() == 0
+    assert a._parse_log_ts("not-a-time") is None
+    assert a._parse_log_ts("") is None
+
+
+def test_parse_container_logs_splits_streams_and_ts():
+    raw = (_frame(1, b"2023-11-14T12:00:00.000000001Z hello world\n")
+           + _frame(2, b"2023-11-14T12:00:01.000000000Z oops\n"))
+    parsed = a._parse_container_logs(raw)
+    assert len(parsed) == 2
+    dt0, s0, l0 = parsed[0]
+    dt1, s1, l1 = parsed[1]
+    assert (s0, l0) == ("stdout", "hello world")
+    assert (s1, l1) == ("stderr", "oops")
+    assert dt1 > dt0
+
+
+def test_parse_container_logs_keeps_unparseable_line():
+    # No timestamp prefix (e.g. a partial write) → dt is None, full line kept.
+    parsed = a._parse_container_logs(_frame(1, b"no-timestamp-here\n"))
+    assert parsed == [(None, "stdout", "no-timestamp-here")]
+
+
+def test_collect_docker_logs_since_cursor_and_dedup(monkeypatch):
+    containers = [{"name": "web", "status": "running"}]
+    calls = []
+    t1 = "2023-11-14T12:00:00.000000000Z"
+    t2 = "2023-11-14T12:00:01.000000000Z"
+    t3 = "2023-11-14T12:00:02.000000000Z"
+
+    def fake_raw(socket_path, path, timeout=a.DOCKER_API_TIMEOUT):
+        calls.append(path)
+        if "tail=" in path:            # first interval seeds from a tail
+            return _frame(1, f"{t1} a\n".encode()) + _frame(1, f"{t2} b\n".encode())
+        # second interval: daemon re-returns the boundary line (t2) + a new one
+        return _frame(1, f"{t2} b\n".encode()) + _frame(1, f"{t3} c\n".encode())
+
+    monkeypatch.setattr(a, "_docker_request_raw", fake_raw)
+    state: dict = {}
+    first = a.collect_docker_logs("/sock", containers, state)
+    assert [r["line"] for r in first] == ["a", "b"]
+    assert "tail=" in calls[0]
+
+    second = a.collect_docker_logs("/sock", containers, state)
+    # boundary line b (t2) deduped against the cursor; only c (t3) is new
+    assert [r["line"] for r in second] == ["c"]
+    assert "since=" in calls[1]
+
+
+def test_collect_docker_logs_excludes_and_skips_non_running(monkeypatch):
+    containers = [
+        {"name": "web", "status": "running"},
+        {"name": "noisy", "status": "running"},
+        {"name": "db", "status": "exited"},
+    ]
+    seen = []
+
+    def fake_raw(socket_path, path, timeout=a.DOCKER_API_TIMEOUT):
+        seen.append(path)
+        return _frame(1, b"2023-11-14T12:00:00.000000000Z x\n")
+
+    monkeypatch.setattr(a, "_docker_request_raw", fake_raw)
+    out = a.collect_docker_logs("/sock", containers, {}, exclude=["noisy"])
+    # only web is fetched: noisy is excluded, db isn't running
+    assert seen and all("/containers/web/" in p for p in seen)
+    assert {r["container"] for r in out} == {"web"}
+
+
+def test_collect_docker_logs_byte_cap_truncates(monkeypatch):
+    containers = [{"name": "web", "status": "running"}]
+    blob = b"".join(
+        _frame(1, f"2023-11-14T12:00:{i:02d}.000000000Z {'x' * 20}\n".encode())
+        for i in range(10))
+    monkeypatch.setattr(a, "_docker_request_raw",
+                        lambda s, p, timeout=a.DOCKER_API_TIMEOUT: blob)
+    out = a.collect_docker_logs("/sock", containers, {}, max_bytes=50)
+    # Once the cap is crossed a single marker line is appended and we stop.
+    assert out[-1]["container"] == "_steward"
+    assert "truncated" in out[-1]["line"]
+    real = [r for r in out if r["container"] == "web"]
+    assert 0 < len(real) < 10
+
+
+def test_collect_docker_logs_forgets_gone_container_cursors(monkeypatch):
+    monkeypatch.setattr(
+        a, "_docker_request_raw",
+        lambda s, p, timeout=a.DOCKER_API_TIMEOUT:
+        _frame(1, b"2023-11-14T12:00:00.000000000Z x\n"))
+    state: dict = {}
+    a.collect_docker_logs("/sock", [{"name": "web", "status": "running"}], state)
+    assert "web" in state["docker_log_cursors"]
+    # web is gone next interval → its cursor is pruned so state can't grow forever
+    a.collect_docker_logs("/sock", [{"name": "db", "status": "running"}], state)
+    assert "web" not in state["docker_log_cursors"]
+    assert "db" in state["docker_log_cursors"]
+
+
+def test_build_sample_includes_logs_when_present(monkeypatch):
+    monkeypatch.setattr(a, "collect_docker",
+                        lambda _s: [{"name": "web", "status": "running"}])
+    monkeypatch.setattr(a, "collect_swarm", lambda _s: None)
+    monkeypatch.setattr(a, "collect_disk_usage", lambda _s: None)
+    monkeypatch.setattr(
+        a, "collect_docker_logs",
+        lambda *args, **kw: [{"container": "web", "stream": "stdout",
+                              "ts": "t", "line": "hello"}])
+    sample = a.build_sample(["/"], {}, "/var/run/docker.sock")
+    assert sample["docker_logs"][0]["line"] == "hello"
+
+
+def test_build_sample_omits_logs_when_disabled(monkeypatch):
+    monkeypatch.setattr(a, "collect_docker",
+                        lambda _s: [{"name": "web", "status": "running"}])
+    monkeypatch.setattr(a, "collect_swarm", lambda _s: None)
+    monkeypatch.setattr(a, "collect_disk_usage", lambda _s: None)
+    called = {"logs": False}
+
+    def _logs(*args, **kw):
+        called["logs"] = True
+        return [{"container": "web", "line": "x"}]
+
+    monkeypatch.setattr(a, "collect_docker_logs", _logs)
+    sample = a.build_sample(["/"], {}, "/var/run/docker.sock",
+                            docker_logs_enabled=False)
+    assert "docker_logs" not in sample
+    assert called["logs"] is False   # collection skipped entirely, not just dropped
+
+
+def test_drop_logs_strips_only_logs():
+    sample = {"ts": "t", "cpu_pct": 5.0,
+              "docker": [{"name": "web"}], "docker_logs": [{"line": "x"}]}
+    out = a._drop_logs(sample)
+    assert "docker_logs" not in out
+    assert out["docker"] == [{"name": "web"}] and out["cpu_pct"] == 5.0   # metrics kept
+    assert a._drop_logs(out) is out                                       # idempotent
+
+
+def test_read_config_docker_logs_default_on(tmp_path):
+    p = tmp_path / "agent.conf"
+    p.write_text("url = x\ntoken = y\n")
+    cfg = a.read_config(str(p))
+    assert cfg["docker_logs_enabled"] is True
+    assert cfg["docker_log_exclude"] == []
+
+
+def test_read_config_disables_docker_logs(tmp_path):
+    p = tmp_path / "agent.conf"
+    p.write_text("url = x\ntoken = y\ndocker_logs_enabled = false\n")
+    cfg = a.read_config(str(p))
+    assert cfg["docker_logs_enabled"] is False
+
+
+def test_read_config_parses_docker_log_exclude(tmp_path):
+    p = tmp_path / "agent.conf"
+    p.write_text("url = x\ntoken = y\ndocker_log_exclude = watchtower, foo\n")
+    cfg = a.read_config(str(p))
+    assert cfg["docker_log_exclude"] == ["watchtower", "foo"]