fix(plugins): show load-failure reasons on the page the banner links to

The admin banner counts every plugin load failure, but settings/plugins.html
only rendered failure reasons inside the plugin card macro -- which iterates
discovered plugins. A plugin that failed BECAUSE it has no directory was
therefore counted in the banner and shown nowhere on the page the banner links
to. That dead end is what the operator hit: "1 plugin failed to load" on the
dashboard, nothing wrong on the details page, and no way to clean it up.

Folds the reason into the "Configured but not installed" rows rather than
adding a competing section, and drives that list from the UNION of stored
plugin.* settings and recorded load failures instead of stored settings alone.
A failure with no stored row -- a plugin enabled by a DEFAULTS entry, which is
exactly what plugin.http was -- would otherwise still be counted and still be
invisible.

Rows carry `removable`: only a plugin with a real stored row can be cleaned up,
so a default-declared one shows the reason and an explanation instead of a
Remove button that would delete nothing while the default reasserts it. Rows
with a reason read as an error (red, "Failed to load"); rows that merely have
leftover config stay a warning. A discovered plugin that failed is excluded --
its own card already shows the reason.

Replaces find_orphaned_plugin_names outright rather than adding a second
overlapping helper; it had one caller.

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

.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
+          # 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
+            MOVING="dev"
+          elif [ "${{ github.ref }}" = "refs/heads/main" ]; then
+            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 }}"
-          if [ "${{ github.ref }}" = "refs/heads/dev" ]; then
-            docker tag "$IMAGE:${{ github.sha }}" "$IMAGE:dev"
-            docker push "$IMAGE:dev"
-          elif [ "${{ github.ref }}" = "refs/heads/main" ]; then
-            docker tag "$IMAGE:${{ github.sha }}" "$IMAGE:latest"
-            docker push "$IMAGE:latest"
           fi
       - name: Prune dangling layers
         if: always()

ci-requirements.md

@@ -46,7 +46,7 @@ git.fabledsword.com/bvandeusen/ci-python:3.14
 
 - Steward has **no frontend and no Redis/Celery** (no external workers), so there
   is no frontend-build lane and the only service container is Postgres.
-- Integration uses Forgejo Actions `services:` + a socket-discovered bridge IP
+- Integration uses Fabled-Git Actions `services:` + a socket-discovered bridge IP
   because `act_runner` (swarm-runner v0.6+) puts services on the default bridge
   with no embedded DNS. FabledCurator's `ci.yml` is the canonical example of the
   pattern; Steward's is the same shape minus Redis and minus sharding.

config.example.yaml

@@ -13,6 +13,14 @@
 database:
   url: "postgresql+asyncpg://steward:password@localhost/steward"
 
+  # Optional: seconds to wait at startup for the database to become reachable
+  # before giving up (default: 60). Steward retries with backoff rather than
+  # crashing when the DB is merely slow to arrive — replaying WAL after an
+  # unclean shutdown, or container DNS not yet resolving after a host reboot.
+  # Raise it if your database is routinely slower than this to accept
+  # connections. Env var: STEWARD_DB_CONNECT_TIMEOUT
+  # connect_timeout: 60
+
 # Optional: override the auto-generated secret key.
 # If not set, a key is auto-generated on first run and saved to /data/secret.key.
 # secret_key: "change-me-to-a-random-string"

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/app.py

@@ -3,9 +3,12 @@ from __future__ import annotations
 import asyncio
 import logging
 from pathlib import Path
-from quart import Quart, render_template
+from quart import Quart, render_template, request
 from .config import load_bootstrap
-from .database import init_db
+from .database import (
+    init_db, ensure_database_reachable, DatabaseUnavailable,
+    DB_CONNECT_TIMEOUT_SECONDS,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -23,6 +26,7 @@ def create_app(
         bootstrap = {
             "database_url": "postgresql+asyncpg://test/test",
             "secret_key": "test-secret-key",
+            "db_connect_timeout": DB_CONNECT_TIMEOUT_SECONDS,
             "plugin_dirs": ["plugins"],
             "plugin_install_dir": "plugins",
         }
@@ -42,6 +46,19 @@ def create_app(
         from unittest.mock import MagicMock
         app.db_sessionmaker = MagicMock()
 
+    # ── 2b. Block until the database is actually reachable ────────────────────
+    # Everything from here down touches the DB synchronously (migrations,
+    # secret re-encryption, settings load). Gate all of it behind one readiness
+    # check so a DB that is merely slow to come up — WAL recovery after an
+    # unclean shutdown, or Docker DNS not yet serving the `db` name after a host
+    # reboot — is waited out instead of crashing the container on a traceback.
+    if not testing:
+        from .database import wait_for_database
+        wait_for_database(
+            app.config["DATABASE_URL"],
+            timeout_seconds=bootstrap["db_connect_timeout"],
+        )
+
     # ── 3. Core migrations only (creates app_settings table) ──────────────────
     if not testing:
         from .core.migration_runner import run_core_migrations
@@ -231,6 +248,29 @@ def create_app(
     async def health():
         return {"status": "ok"}
 
+    # ── 11b. Database availability gate ────────────────────────────────────────
+    # Every page in Steward reads the database, so a database that has gone away
+    # under a running app turns each request into an opaque 500 (this is how a
+    # login attempt surfaced a bare gaierror). Acquire a connection up front,
+    # retrying briefly to ride out a restart, and answer honestly if it stays
+    # down rather than failing deep inside a handler with a generic error.
+    if not testing:
+        @app.before_request
+        async def _database_gate():
+            # /health is a liveness probe for the container itself — it must
+            # stay answerable while the database is down, or a restart loop
+            # gets triggered by a dependency outage. Static files need no DB.
+            if request.endpoint in ("health", "static"):
+                return None
+            try:
+                await ensure_database_reachable(app._db_engine)
+            except DatabaseUnavailable as exc:
+                logger.error("Database unreachable while serving %s: %s",
+                             request.path, exc)
+                return await render_template(
+                    "errors/database_unavailable.html"), 503
+            return None
+
     # ── 12. Error handlers ─────────────────────────────────────────────────────
     @app.errorhandler(404)
     async def not_found(_):

steward/config.py

@@ -49,6 +49,8 @@ def load_bootstrap(config_path: Path | str | None = None) -> dict[str, Any]:
 
     secret_key = _resolve_secret_key(raw)
 
+    db_connect_timeout = _resolve_db_connect_timeout(raw)
+
     # Plugin discovery spans two roots (see load_plugins / migration_runner):
     #   • bundled  — first-party plugins shipped inside the image at repo-root
     #     `plugins/`; they version atomically with core and are read-only at runtime.
@@ -68,12 +70,47 @@ def load_bootstrap(config_path: Path | str | None = None) -> dict[str, Any]:
     return {
         "database_url": database_url,
         "secret_key": secret_key,
+        "db_connect_timeout": db_connect_timeout,
         "plugin_dirs": plugin_dirs,
         # Installs/downloads target the external (writable, persistent) dir.
         "plugin_install_dir": external_plugin_dir or bundled_plugin_dir,
     }
 
 
+def _resolve_db_connect_timeout(raw: dict) -> float:
+    """How long to wait for the database at startup, in seconds.
+
+    Bootstrap-only by necessity: this governs reaching the DB, so it cannot
+    itself be read from the DB like the rest of Steward's settings.
+
+    A deployment whose database is slower to come up than the default (a large
+    cluster replaying WAL, a remote DB behind a link that takes a while) can
+    raise it rather than crash-looping the container.
+    """
+    from .database import DB_CONNECT_TIMEOUT_SECONDS
+
+    value = _env("DB_CONNECT_TIMEOUT") or raw.get("database", {}).get("connect_timeout")
+    if value is None or value == "":
+        return DB_CONNECT_TIMEOUT_SECONDS
+    try:
+        parsed = float(value)
+    except (TypeError, ValueError):
+        logger.warning(
+            "Invalid database connect timeout %r — using default %.0fs",
+            value, DB_CONNECT_TIMEOUT_SECONDS,
+        )
+        return DB_CONNECT_TIMEOUT_SECONDS
+    if parsed <= 0:
+        # 0/negative would mean "never wait", which is the broken behaviour this
+        # setting exists to fix — treat it as a mistake, not as an opt-out.
+        logger.warning(
+            "Database connect timeout %.0fs is not positive — using default %.0fs",
+            parsed, DB_CONNECT_TIMEOUT_SECONDS,
+        )
+        return DB_CONNECT_TIMEOUT_SECONDS
+    return parsed
+
+
 def _resolve_secret_key(raw: dict) -> str:
     """Resolve secret_key: env var → file → auto-generate.
 

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/plugin_manager.py

@@ -87,7 +87,8 @@ def _import_plugin(name: str, plugin_path: Path):
     """Load a plugin module by file path, avoiding sys.modules stdlib collisions.
 
     Using importlib.import_module(name) fails for plugins whose names shadow
-    Python stdlib modules (e.g. the 'http' plugin vs stdlib's 'http' package).
+    Python stdlib modules (e.g. a plugin named 'json' or 'http' vs the stdlib
+    package of the same name).
     This helper loads from the filesystem path directly and registers the module
     under a namespaced key so relative imports within the plugin still work.
     """

steward/core/settings.py

@@ -18,6 +18,7 @@ from __future__ import annotations
 import asyncio
 import json
 import logging
+from collections.abc import Iterable, Mapping
 from datetime import datetime, timezone
 from typing import Any
 
@@ -84,6 +85,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,
@@ -98,7 +107,6 @@ DEFAULTS: dict[str, Any] = {
     # Per-plugin yaml config defaults are merged on top at load time.
     "plugin.docker": {"enabled": True},
     "plugin.host_agent": {"enabled": True},
-    "plugin.http": {"enabled": True},
     "plugin.snmp": {"enabled": True},
     # OIDC single-sign-on
     "oidc.enabled": False,
@@ -221,6 +229,82 @@ async def set_setting(session: AsyncSession, key: str, value: Any) -> None:
         _undecryptable_secrets.discard(key)
 
 
+async def delete_setting(session: AsyncSession, key: str) -> bool:
+    """Remove a stored setting row. Returns True if a row was actually deleted.
+
+    Call in a transaction. Deleting a key that also has a DEFAULT reverts it to
+    that default rather than unsetting it — so this only meaningfully *removes*
+    a setting when no default declares it.
+    """
+    result = await session.execute(
+        select(AppSetting).where(AppSetting.key == key)
+    )
+    row = result.scalar_one_or_none()
+    if row is None:
+        return False
+    await session.delete(row)
+    return True
+
+
+async def get_stored_plugin_names(session: AsyncSession) -> list[str]:
+    """Plugin names that have a real stored row, ignoring DEFAULTS.
+
+    Deliberately NOT derived from get_all_settings: that merges DEFAULTS in, and
+    a plugin.* key present only as a default is a code-level declaration rather
+    than operator data. Offering to "remove" such a key would be a lie — the
+    default would simply reassert it on the next load (exactly the bug behind
+    the phantom `plugin.http` entry). Only stored rows can actually be cleaned up.
+
+    Returns names only, never values: plugin config can hold credentials, and
+    nothing that lists orphans needs to read them.
+    """
+    prefix = "plugin."
+    result = await session.execute(
+        select(AppSetting.key).where(AppSetting.key.like(f"{prefix}%"))
+    )
+    return sorted(
+        key[len(prefix):] for key in result.scalars() if key != prefix
+    )
+
+
+def find_orphaned_plugins(
+    stored_names: Iterable[str],
+    installed_names: Iterable[str],
+    failures: Mapping[str, str] | None = None,
+) -> list[dict[str, Any]]:
+    """Plugins that are configured or failed to load, but are not installed.
+
+    Pure so it can be tested without a DB or a filesystem. "Not installed" is
+    intentionally not treated as "safe to delete" — an external plugin under
+    /data/plugins can be missing merely because the volume isn't mounted or an
+    install failed, so the caller surfaces these for an explicit operator
+    decision instead of removing them automatically.
+
+    Driven by the UNION of stored settings and load failures, not by stored
+    settings alone. `load_plugins` records a failure for any enabled plugin
+    whose directory it cannot find, and the admin banner counts those — so a
+    failure with no stored row (a plugin enabled by a DEFAULTS entry) would
+    otherwise be counted in the banner and shown nowhere, which is exactly the
+    dead end this section exists to close.
+
+    Each row carries `removable`: only a plugin with a real stored row can
+    actually be cleaned up. Offering Remove for a default-declared plugin would
+    be a lie — deleting nothing, while the default reasserts it on next load.
+    """
+    installed = set(installed_names)
+    failures = dict(failures or {})
+    stored = {name for name in stored_names if name}
+    candidates = (stored | set(failures)) - installed
+    return [
+        {
+            "name": name,
+            "reason": failures.get(name),
+            "removable": name in stored,
+        }
+        for name in sorted(candidates)
+    ]
+
+
 async def get_all_settings(session: AsyncSession) -> dict[str, Any]:
     """Return flat key→value dict with defaults filled in for missing keys."""
     result = await session.execute(select(AppSetting))

steward/database.py

@@ -1,10 +1,49 @@
 from __future__ import annotations
+import asyncio
+import logging
+import time
 from typing import TYPE_CHECKING
-from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
+from sqlalchemy import text
+from sqlalchemy.exc import SQLAlchemyError
+from sqlalchemy.ext.asyncio import (
+    create_async_engine, async_sessionmaker, AsyncEngine, AsyncSession,
+)
+from sqlalchemy.pool import NullPool
 
 if TYPE_CHECKING:
     from quart import Quart
 
+logger = logging.getLogger(__name__)
+
+# Recycle pooled connections after 30 minutes. Nothing in Postgres closes an
+# idle connection by default, but the path to it is not always durable: NAT and
+# conntrack tables drop idle flows, and a Docker network rebuild silently
+# strands existing sockets. Capping connection age means a stranded connection
+# is retired on a timer instead of surfacing as a failed query later.
+POOL_RECYCLE_SECONDS = 1800
+
+# Total budget for the database to become reachable at startup, and the backoff
+# ceiling between attempts. 60s comfortably covers Postgres WAL recovery after
+# an unclean shutdown (observed ~7s) plus Docker DNS coming up on a host reboot.
+DB_CONNECT_TIMEOUT_SECONDS = 60.0
+_BACKOFF_INITIAL_SECONDS = 0.5
+_BACKOFF_MAX_SECONDS = 5.0
+
+# Per-REQUEST retry schedule (seconds between attempts), total ~1.75s across 4
+# tries. Far shorter than the startup budget on purpose: nobody is watching a
+# container boot, but somebody is watching this page load. Long enough to ride
+# out a pool blip or a fast reconnect, short enough that a genuinely-down
+# database gets an honest answer instead of a spinner.
+_REQUEST_RETRY_DELAYS = (0.25, 0.5, 1.0)
+
+# Postgres SQLSTATEs that will never resolve by waiting — retrying these just
+# delays a clear error behind a full timeout budget.
+_FATAL_SQLSTATES = {
+    "28P01",  # invalid_password
+    "28000",  # invalid_authorization_specification
+    "3D000",  # invalid_catalog_name — database does not exist
+}
+
 
 def init_db(app: "Quart") -> None:
     """Create async engine and attach db_sessionmaker to app.
@@ -13,8 +52,162 @@ def init_db(app: "Quart") -> None:
     Does not create tables — Alembic handles migrations.
     """
     db_url: str = app.config["DATABASE_URL"]
-    engine = create_async_engine(db_url, echo=False)
+    engine = create_async_engine(
+        db_url,
+        echo=False,
+        # Check a pooled connection is still alive before handing it out. When
+        # the database restarts, every connection already in the pool is dead
+        # at the socket level; without this, SQLAlchemy only discovers that by
+        # failing a real query, so the first operation after a DB restart
+        # errors out on whatever triggered it (a request, a scheduled poll).
+        # The check is a cheap round-trip and it makes a DB restart invisible.
+        pool_pre_ping=True,
+        pool_recycle=POOL_RECYCLE_SECONDS,
+    )
     app.db_sessionmaker: async_sessionmaker[AsyncSession] = async_sessionmaker(
         engine, expire_on_commit=False
     )
     app._db_engine = engine
+
+
+def _fatal_sqlstate(exc: BaseException) -> str | None:
+    """Return the SQLSTATE if this error chain carries a non-retryable one."""
+    seen: set[int] = set()
+    cur: BaseException | None = exc
+    while cur is not None and id(cur) not in seen:
+        seen.add(id(cur))
+        sqlstate = getattr(cur, "sqlstate", None)
+        if sqlstate in _FATAL_SQLSTATES:
+            return sqlstate
+        cur = cur.__cause__ or cur.__context__
+    return None
+
+
+def _describe(exc: BaseException) -> str:
+    """Innermost cause, which is the part that says what actually went wrong.
+
+    SQLAlchemy wraps DBAPI errors several layers deep; the outer message is
+    boilerplate, so surface the root for the waiting-for-database log line.
+    """
+    cur: BaseException = exc
+    seen: set[int] = {id(cur)}
+    while True:
+        nxt = cur.__cause__ or cur.__context__
+        if nxt is None or id(nxt) in seen:
+            return f"{type(cur).__name__}: {cur}"
+        seen.add(id(nxt))
+        cur = nxt
+
+
+def wait_for_database(
+    db_url: str,
+    timeout_seconds: float = DB_CONNECT_TIMEOUT_SECONDS,
+) -> None:
+    """Block until the database accepts a connection, or raise after the budget.
+
+    Steward otherwise assumes the database is both resolvable and *ready* the
+    first time it asks, which is false in two ordinary situations:
+
+      • Host reboot / full stack restart — the app container can try to resolve
+        the `db` service name before Docker's embedded DNS has the record,
+        giving `gaierror -2 Name or service not known`.
+      • Unclean shutdown — Postgres is listening but still replaying WAL, and
+        refuses connections with "the database system is not yet accepting
+        connections" until recovery reaches a consistent state.
+
+    compose's `depends_on: service_healthy` covers ordering on a clean `up`,
+    but not either of the above. Both are transient and self-healing, so retry
+    rather than dumping a traceback and dying. Credential and missing-database
+    errors are NOT transient and fail immediately.
+    """
+    deadline = time.monotonic() + timeout_seconds
+    delay = _BACKOFF_INITIAL_SECONDS
+    attempt = 0
+    last_exc: BaseException | None = None
+
+    while True:
+        attempt += 1
+        try:
+            asyncio.run(_probe(db_url))
+        except Exception as exc:  # broad by design — classified just below
+            sqlstate = _fatal_sqlstate(exc)
+            if sqlstate is not None:
+                # Wrong password / missing database: waiting cannot fix it.
+                raise
+            last_exc = exc
+            remaining = deadline - time.monotonic()
+            if remaining <= 0:
+                break
+            logger.warning(
+                "Database not ready (attempt %d, %.0fs budget left): %s — "
+                "retrying in %.1fs",
+                attempt, remaining, _describe(exc), min(delay, remaining),
+            )
+            time.sleep(min(delay, remaining))
+            delay = min(delay * 2, _BACKOFF_MAX_SECONDS)
+        else:
+            if attempt > 1:
+                logger.info("Database ready after %d attempt(s)", attempt)
+            return
+
+    detail = f" Last error: {_describe(last_exc)}" if last_exc is not None else ""
+    raise RuntimeError(
+        f"Database did not become available within {timeout_seconds:.0f}s "
+        f"({attempt} attempts).{detail}"
+    ) from last_exc
+
+
+class DatabaseUnavailable(Exception):
+    """The database could not be reached while serving a request."""
+
+
+async def ensure_database_reachable(engine: AsyncEngine) -> None:
+    """Check out one pooled connection, retrying briefly, or raise.
+
+    Startup readiness (wait_for_database) does not help once the app is already
+    serving: if the database goes away underneath a running Steward, the next
+    request needs a connection, the pooled ones are dead, and establishing a new
+    one fails — which is how a login attempt turned into an opaque 500.
+
+    pool_pre_ping already makes recovery automatic *once the database is back*.
+    What it cannot do is wait: while the container is genuinely down, its
+    replacement connect fails too. So retry briefly here to ride out a restart,
+    then give up and let the caller render an honest "database unavailable"
+    page rather than a generic error.
+
+    The budget is deliberately short — a person is waiting on this request, and
+    a page that hangs for half a minute is worse than one that says plainly
+    what is wrong and offers a retry.
+    """
+    last_exc: BaseException | None = None
+    for attempt, delay in enumerate(_REQUEST_RETRY_DELAYS + (None,)):
+        try:
+            async with engine.connect():
+                if attempt:
+                    logger.info(
+                        "Database reachable again after %d retry attempt(s)", attempt)
+                return
+        except (SQLAlchemyError, OSError) as exc:
+            # Bad credentials / missing database will never resolve by waiting,
+            # and they are not what this guard is for — let them surface.
+            if _fatal_sqlstate(exc) is not None:
+                raise
+            last_exc = exc
+            if delay is not None:
+                await asyncio.sleep(delay)
+
+    raise DatabaseUnavailable(_describe(last_exc)) from last_exc
+
+
+async def _probe(db_url: str) -> None:
+    """Open one throwaway connection and round-trip a trivial query.
+
+    Uses its own engine with NullPool: this runs before the app engine exists,
+    and a probe connection must never be left in a pool for real work to reuse.
+    """
+    engine = create_async_engine(db_url, echo=False, poolclass=NullPool)
+    try:
+        async with engine.connect() as conn:
+            await conn.execute(text("SELECT 1"))
+    finally:
+        await engine.dispose()

steward/migrations/versions/0025_drop_http_plugin_setting.py

@@ -0,0 +1,40 @@
+"""Drop the stored setting for the removed http plugin
+
+The standalone `http` plugin was dissolved into the unified Monitor entity
+(0022_unify_monitors), but its `plugin.http` settings key outlived it: the key
+stayed in core DEFAULTS, so the operator saw a plugin that no longer exists
+reported as "enabled" with no way to clear it — deleting the row alone did
+nothing, because the default reasserted it on the next settings load.
+
+The DEFAULTS entry is removed in the same change; this migration clears any row
+an operator's database still carries so the two agree. Per family rule 22, the
+removed subsystem takes its setting row with it.
+
+Note: `http` remains a valid MONITOR TYPE (icmp/tcp/dns/http). This touches only
+the plugin-enablement key, never monitor data.
+
+Revision ID: 0025_drop_http_plugin_setting
+Revises: 0024_plugin_metrics_hourly
+Create Date: 2026-08-13
+"""
+from typing import Sequence, Union
+from alembic import op
+import sqlalchemy as sa
+
+revision: str = "0025_drop_http_plugin_setting"
+down_revision: Union[str, None] = "0024_plugin_metrics_hourly"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.execute(
+        sa.text("DELETE FROM app_settings WHERE key = 'plugin.http'")
+    )
+
+
+def downgrade() -> None:
+    # Deliberately empty. Re-inserting `plugin.http` would recreate the exact
+    # phantom this migration exists to remove, and the plugin it configured no
+    # longer exists to read it.
+    pass

steward/settings/routes.py

@@ -8,7 +8,8 @@ from steward.auth.middleware import require_role
 from steward.core.audit import log_audit
 from steward.models.users import UserRole
 from steward.core.settings import (
-    get_all_settings, set_setting,
+    get_all_settings, set_setting, delete_setting,
+    get_stored_plugin_names, find_orphaned_plugins,
     to_smtp_cfg, to_webhook_cfg, to_ansible_cfg, to_plugins_cfg,
     to_oidc_cfg, to_ldap_cfg, to_thresholds_cfg,
 )
@@ -133,6 +134,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 +172,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"})
@@ -572,7 +588,7 @@ def _build_plugin_cfg_from_form(plugin: dict, form) -> dict:
 # of a host), not discrete vendor integrations. Presentation-only split — they
 # still load through the normal plugin mechanism. A plugin.yaml may set
 # kind: capability|integration to override.
-CAPABILITY_PLUGINS = {"host_agent", "http", "snmp", "docker"}
+CAPABILITY_PLUGINS = {"host_agent", "snmp", "docker"}
 
 
 @settings_bp.get("/plugins/")
@@ -580,22 +596,65 @@ CAPABILITY_PLUGINS = {"host_agent", "http", "snmp", "docker"}
 async def plugins():
     async with current_app.db_sessionmaker() as db:
         settings = await get_all_settings(db)
+        stored_plugin_names = await get_stored_plugin_names(db)
     discovered = _discover_plugins()
     _merge_plugin_config(discovered, to_plugins_cfg(settings))
     for p in discovered:
         p["_kind"] = p.get("kind") or (
             "capability" if p["_dir"] in CAPABILITY_PLUGINS else "integration")
     repos = _get_plugin_repos(settings)
+    # Config left behind by plugins that are no longer installed. Surfaced for an
+    # explicit operator decision rather than cleaned up automatically: an external
+    # plugin can be missing because /data/plugins isn't mounted or an install
+    # failed, and silently dropping its row would destroy stored credentials.
+    #
+    # Load failures are folded in so every plugin counted by the admin banner has
+    # a row here. A plugin that failed BECAUSE it has no directory is not in
+    # `discovered`, so without this it would be counted in the banner and shown
+    # nowhere on the page that banner links to.
+    from steward.core.plugin_manager import get_plugin_failures
+    orphans = find_orphaned_plugins(
+        stored_plugin_names,
+        [p["_dir"] for p in discovered],
+        get_plugin_failures(),
+    )
     return await render_template(
         "settings/plugins.html",
         capabilities=[p for p in discovered if p["_kind"] == "capability"],
         integrations=[p for p in discovered if p["_kind"] != "capability"],
         discovered_plugins=discovered,
+        orphaned_plugins=orphans,
         repos=repos,
         settings=settings,
     )
 
 
+@settings_bp.post("/plugins/orphans/<name>/remove/")
+@require_role(UserRole.admin)
+async def plugin_orphan_remove(name: str):
+    """Delete the stored settings row for a plugin that is no longer installed.
+
+    Refuses if the plugin IS installed — that would silently wipe a live
+    plugin's config, and disabling it is what the operator wants there instead.
+    """
+    installed = {p["_dir"] for p in _discover_plugins()}
+    if name in installed:
+        return redirect(url_for("settings.plugins"))
+
+    async with current_app.db_sessionmaker() as db:
+        async with db.begin():
+            removed = await delete_setting(db, f"plugin.{name}")
+
+    if removed:
+        await _reload_app_config()
+        from steward.core.plugin_index import clear_catalog_cache
+        clear_catalog_cache()
+        await log_audit(
+            current_app, session.get("user_id"), session.get("username", ""),
+            "plugin.settings_removed", entity_type="plugin", entity_id=name)
+    return redirect(url_for("settings.plugins"))
+
+
 # ── Per-plugin detail (settings) ──────────────────────────────────────────────
 
 @settings_bp.get("/plugins/<name>/")

steward/templates/errors/database_unavailable.html

@@ -0,0 +1,16 @@
+{% extends "base.html" %}
+{% block title %}Database unavailable — Steward{% endblock %}
+{% block content %}
+<div style="text-align:center; padding: 4rem 2rem; font-family: var(--font-serif);">
+  <h1 style="color: var(--yellow); font-size: 2.5rem; margin-bottom: 0.5rem;">503</h1>
+  <p style="color: var(--text); font-size: 1.2rem; margin-bottom: 0.25rem;">The records are out of reach.</p>
+  <p style="color: var(--text-dim); margin-bottom: 0.5rem; max-width: 34rem; margin-left:auto; margin-right:auto;">
+    Steward is running, but cannot reach its database. Nothing has been lost —
+    this page will work again as soon as the database is back.
+  </p>
+  <p style="color: var(--text-dim); font-size: 0.85rem; margin-bottom: 1.5rem;">
+    Steward already retried for a moment before showing this.
+  </p>
+  <a href="{{ request.path }}" class="btn">Try again</a>
+</div>
+{% endblock %}

steward/templates/settings/plugins.html

@@ -60,7 +60,7 @@
   </form>
 </div>
 <p style="color:var(--text-muted);font-size:0.82rem;margin:0 0 0.75rem;max-width:720px;">
-  Built-in ways to monitor your hosts (agent metrics, HTTP/uptime, SNMP, Docker). These are
+  Built-in ways to monitor your hosts (agent metrics, SNMP, Docker). These are
   facets of a host — you'll see their data in the <a href="/hosts/">Hosts</a> and
   <a href="/status">Status</a> sections, not as separate areas. On by default.
 </p>
@@ -88,6 +88,63 @@
 </div>
 {% endif %}
 
+{# ── Orphaned settings ─────────────────────────────────────────────────────── #}
+{# Hidden entirely when there is nothing to clean up — an empty "Configured but
+   not installed" panel would read as a problem rather than a clean state. #}
+{% if orphaned_plugins %}
+<div style="margin-bottom:1.5rem;max-width:720px;">
+  <div class="section-title" style="margin-bottom:0.4rem;">Configured but not installed</div>
+  <p style="color:var(--text-muted);font-size:0.82rem;margin:0 0 0.75rem;">
+    Plugins Steward has settings for, or tried to load, but cannot find on disk — including
+    anything counted by the "failed to load" banner. This usually means the plugin was
+    removed, but it also happens when an external plugin directory isn't mounted or an
+    install failed part-way, so nothing is deleted automatically. Removing an entry
+    discards that plugin's saved configuration, including any credentials.
+  </p>
+  <div style="display:grid;gap:0.6rem;">
+    {% for orphan in orphaned_plugins %}
+    <div class="card" style="padding:0.85rem 1rem;display:flex;align-items:center;gap:0.75rem;">
+      <span style="width:8px;height:8px;border-radius:50%;flex-shrink:0;
+                   background:{{ 'var(--red)' if orphan.reason else 'var(--yellow)' }};"
+            title="{{ orphan.reason or 'Configured but not installed' }}"></span>
+      <div style="flex:1;min-width:0;">
+        <div style="display:flex;align-items:baseline;gap:0.5rem;flex-wrap:wrap;">
+          <span style="font-weight:600;font-size:0.9rem;color:var(--text);">{{ orphan.name }}</span>
+          {% if orphan.reason %}
+          <span style="font-size:0.72rem;padding:0.1em 0.45em;border-radius:3px;
+                       background:color-mix(in srgb,var(--red) 15%,var(--bg));color:var(--red);">Failed to load</span>
+          {% else %}
+          <span style="font-size:0.72rem;padding:0.1em 0.45em;border-radius:3px;
+                       background:var(--yellow-dim);color:var(--yellow);">Not installed</span>
+          {% endif %}
+        </div>
+        {% if orphan.reason %}
+        <div style="font-size:0.77rem;color:var(--red);margin-top:0.1rem;
+                    white-space:nowrap;overflow:hidden;text-overflow:ellipsis;"
+             title="{{ orphan.reason }}">{{ orphan.reason }}</div>
+        {% endif %}
+        <div style="font-size:0.8rem;color:var(--text-muted);margin-top:0.1rem;">
+          {% if orphan.removable %}
+            Settings key <code>plugin.{{ orphan.name }}</code> has no matching plugin.
+          {% else %}
+            Enabled by a built-in default with no stored settings to remove — this is a
+            packaging bug, not leftover configuration. Please report it.
+          {% endif %}
+        </div>
+      </div>
+      {% if orphan.removable %}
+      <form method="post" action="/settings/plugins/orphans/{{ orphan.name }}/remove/"
+            onsubmit="return confirm('Remove stored settings for &quot;{{ orphan.name }}&quot;? Its saved configuration, including any credentials, will be discarded. This cannot be undone.');"
+            style="margin:0;flex-shrink:0;">
+        <button type="submit" class="btn btn-danger btn-sm" style="font-size:0.78rem;">Remove</button>
+      </form>
+      {% endif %}
+    </div>
+    {% endfor %}
+  </div>
+</div>
+{% endif %}
+
 {# ── Plugin Repositories ───────────────────────────────────────────────────── #}
 <div style="margin-bottom:2rem;max-width:720px;">
   <div class="section-title" style="margin-bottom:0.75rem;">Plugin Repositories</div>

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/core/test_db_readiness.py

@@ -0,0 +1,307 @@
+"""DB connection resilience: pool, startup readiness, and request-time outage.
+
+Covers the three recovery gaps that let a database restart take the app down:
+the engine handing out stale pooled connections (#2626), startup assuming the
+DB is reachable and ready the instant it asks (#2627), and a request finding
+the database gone and failing with an opaque 500 (#2635).
+"""
+import types
+import pytest
+
+from steward import database
+from steward.database import (
+    POOL_RECYCLE_SECONDS,
+    _describe,
+    _fatal_sqlstate,
+    init_db,
+    wait_for_database,
+)
+
+
+class _PGError(Exception):
+    """Stand-in for an asyncpg error, which carries a SQLSTATE attribute."""
+
+    def __init__(self, message: str, sqlstate: str | None = None):
+        super().__init__(message)
+        self.sqlstate = sqlstate
+
+
+@pytest.fixture
+def no_sleep(monkeypatch):
+    """Drive the retry budget off a fake clock instead of wall time.
+
+    sleep() records its duration and advances the clock by exactly that much,
+    so backoff and timeout behaviour are deterministic rather than dependent on
+    how fast the test machine happens to run.
+    """
+    slept: list[float] = []
+    clock = {"now": 0.0}
+
+    def fake_sleep(seconds: float) -> None:
+        slept.append(seconds)
+        clock["now"] += seconds
+
+    monkeypatch.setattr(database.time, "sleep", fake_sleep)
+    monkeypatch.setattr(database.time, "monotonic", lambda: clock["now"])
+    return slept
+
+
+def _probe_raising(*errors, then_succeed: bool = True):
+    """Async probe stub raising the given errors in order.
+
+    Once the list is exhausted it succeeds, unless then_succeed is False — in
+    which case it keeps raising the last error forever (a DB that never comes
+    back).
+    """
+    calls = {"n": 0}
+
+    async def probe(db_url):
+        i = calls["n"]
+        calls["n"] += 1
+        if i < len(errors):
+            raise errors[i]
+        if not then_succeed:
+            raise errors[-1]
+
+    probe.calls = calls
+    return probe
+
+
+# ── engine pool configuration (issue #2626) ──────────────────────────────────
+
+
+def test_engine_enables_pre_ping_and_recycle():
+    app = types.SimpleNamespace(
+        config={"DATABASE_URL": "postgresql+asyncpg://u:p@localhost/db"}
+    )
+    init_db(app)
+
+    # _pre_ping / _recycle are SQLAlchemy pool internals; there is no public
+    # accessor, and these are exactly the settings a DB restart depends on.
+    pool = app._db_engine.sync_engine.pool
+    assert pool._pre_ping is True
+    assert pool._recycle == POOL_RECYCLE_SECONDS
+
+
+def test_pool_recycle_is_positive():
+    # A non-positive recycle disables age-based retirement entirely.
+    assert POOL_RECYCLE_SECONDS > 0
+
+
+# ── retry classification ─────────────────────────────────────────────────────
+
+
+def test_fatal_sqlstate_detects_bad_password():
+    exc = _PGError("password authentication failed", sqlstate="28P01")
+    assert _fatal_sqlstate(exc) == "28P01"
+
+
+def test_fatal_sqlstate_finds_sqlstate_through_cause_chain():
+    inner = _PGError("database does not exist", sqlstate="3D000")
+    outer = RuntimeError("wrapped by sqlalchemy")
+    outer.__cause__ = inner
+    assert _fatal_sqlstate(outer) == "3D000"
+
+
+def test_transient_errors_are_not_fatal():
+    assert _fatal_sqlstate(OSError(-2, "Name or service not known")) is None
+    # "the database system is not yet accepting connections" — resolves on its own.
+    assert _fatal_sqlstate(_PGError("not yet accepting", sqlstate="57P03")) is None
+
+
+def test_describe_unwraps_to_innermost_cause():
+    inner = OSError("Name or service not known")
+    outer = RuntimeError("sqlalchemy boilerplate")
+    outer.__cause__ = inner
+    assert "Name or service not known" in _describe(outer)
+
+
+def test_describe_survives_self_referential_cause():
+    exc = RuntimeError("loop")
+    exc.__cause__ = exc
+    assert "loop" in _describe(exc)
+
+
+# ── wait_for_database (issue #2627) ──────────────────────────────────────────
+
+
+def test_returns_immediately_when_db_is_up(monkeypatch, no_sleep):
+    probe = _probe_raising()
+    monkeypatch.setattr(database, "_probe", probe)
+
+    wait_for_database("postgresql+asyncpg://u:p@db/steward", timeout_seconds=10)
+
+    assert probe.calls["n"] == 1
+    assert no_sleep == []
+
+
+def test_retries_dns_failure_then_succeeds(monkeypatch, no_sleep):
+    # The reported failure: gaierror -2 while Docker DNS is not yet serving `db`.
+    probe = _probe_raising(
+        OSError(-2, "Name or service not known"),
+        OSError(-2, "Name or service not known"),
+    )
+    monkeypatch.setattr(database, "_probe", probe)
+
+    wait_for_database("postgresql+asyncpg://u:p@db/steward", timeout_seconds=30)
+
+    assert probe.calls["n"] == 3
+    assert len(no_sleep) == 2
+
+
+def test_retries_while_postgres_is_still_recovering(monkeypatch, no_sleep):
+    probe = _probe_raising(
+        _PGError("the database system is not yet accepting connections", "57P03"),
+    )
+    monkeypatch.setattr(database, "_probe", probe)
+
+    wait_for_database("postgresql+asyncpg://u:p@db/steward", timeout_seconds=30)
+
+    assert probe.calls["n"] == 2
+
+
+def test_backoff_grows_between_attempts(monkeypatch, no_sleep):
+    probe = _probe_raising(*[OSError("refused")] * 4)
+    monkeypatch.setattr(database, "_probe", probe)
+
+    wait_for_database("postgresql+asyncpg://u:p@db/steward", timeout_seconds=60)
+
+    assert no_sleep == sorted(no_sleep), "delays should be non-decreasing"
+    assert no_sleep[-1] > no_sleep[0], "backoff should grow, not stay flat"
+
+
+def test_backoff_is_capped(monkeypatch, no_sleep):
+    probe = _probe_raising(*[OSError("refused")] * 12)
+    monkeypatch.setattr(database, "_probe", probe)
+
+    wait_for_database("postgresql+asyncpg://u:p@db/steward", timeout_seconds=600)
+
+    assert max(no_sleep) <= database._BACKOFF_MAX_SECONDS
+
+
+def test_bad_credentials_fail_immediately_without_retrying(monkeypatch, no_sleep):
+    probe = _probe_raising(_PGError("password authentication failed", "28P01"))
+    monkeypatch.setattr(database, "_probe", probe)
+
+    with pytest.raises(_PGError):
+        wait_for_database("postgresql+asyncpg://u:bad@db/steward", timeout_seconds=30)
+
+    # Waiting cannot fix a wrong password — don't burn the whole budget on it.
+    assert probe.calls["n"] == 1
+    assert no_sleep == []
+
+
+def test_missing_database_fails_immediately(monkeypatch, no_sleep):
+    probe = _probe_raising(_PGError("database does not exist", "3D000"))
+    monkeypatch.setattr(database, "_probe", probe)
+
+    with pytest.raises(_PGError):
+        wait_for_database("postgresql+asyncpg://u:p@db/nope", timeout_seconds=30)
+
+    assert probe.calls["n"] == 1
+
+
+def test_gives_up_after_budget_with_actionable_message(monkeypatch, no_sleep):
+    probe = _probe_raising(
+        OSError("Name or service not known"), then_succeed=False
+    )
+    monkeypatch.setattr(database, "_probe", probe)
+
+    with pytest.raises(RuntimeError) as excinfo:
+        wait_for_database("postgresql+asyncpg://u:p@db/steward", timeout_seconds=2)
+
+    message = str(excinfo.value)
+    assert "did not become available" in message
+    # The operator needs the underlying cause, not just "timed out".
+    assert "Name or service not known" in message
+
+
+# ── ensure_database_reachable (issue #2635) ──────────────────────────────────
+
+
+class _FakeEngine:
+    """Minimal stand-in for AsyncEngine.connect() as an async context manager."""
+
+    def __init__(self, *errors, then_succeed: bool = True):
+        self.errors = list(errors)
+        self.then_succeed = then_succeed
+        self.calls = 0
+
+    def connect(self):
+        engine = self
+
+        class _Ctx:
+            async def __aenter__(self):
+                index = engine.calls
+                engine.calls += 1
+                if index < len(engine.errors):
+                    raise engine.errors[index]
+                if not engine.then_succeed:
+                    raise engine.errors[-1]
+                return object()
+
+            async def __aexit__(self, *exc_info):
+                return False
+
+        return _Ctx()
+
+
+@pytest.fixture
+def no_async_sleep(monkeypatch):
+    """Record awaited retry delays without spending them."""
+    slept: list[float] = []
+
+    async def fake_sleep(seconds: float) -> None:
+        slept.append(seconds)
+
+    monkeypatch.setattr(database.asyncio, "sleep", fake_sleep)
+    return slept
+
+
+async def test_reachable_on_first_try_costs_no_retries(no_async_sleep):
+    engine = _FakeEngine()
+    await database.ensure_database_reachable(engine)
+    assert engine.calls == 1
+    assert no_async_sleep == []
+
+
+async def test_rides_out_a_brief_outage(no_async_sleep):
+    # The reported shape: DNS gone while the db container restarts.
+    engine = _FakeEngine(OSError(-2, "Name or service not known"))
+    await database.ensure_database_reachable(engine)
+    assert engine.calls == 2
+    assert no_async_sleep == [0.25]
+
+
+async def test_uses_the_full_retry_schedule_before_giving_up(no_async_sleep):
+    engine = _FakeEngine(OSError("refused"), then_succeed=False)
+    with pytest.raises(database.DatabaseUnavailable):
+        await database.ensure_database_reachable(engine)
+
+    # One attempt per delay, plus a final attempt after the last wait.
+    assert engine.calls == len(database._REQUEST_RETRY_DELAYS) + 1
+    assert no_async_sleep == list(database._REQUEST_RETRY_DELAYS)
+
+
+async def test_request_budget_stays_short(no_async_sleep):
+    # A person is waiting on this; guard against the schedule growing into a hang.
+    assert sum(database._REQUEST_RETRY_DELAYS) <= 3.0
+
+
+async def test_unavailable_error_carries_the_underlying_cause(no_async_sleep):
+    engine = _FakeEngine(
+        OSError("Name or service not known"), then_succeed=False)
+    with pytest.raises(database.DatabaseUnavailable) as excinfo:
+        await database.ensure_database_reachable(engine)
+    assert "Name or service not known" in str(excinfo.value)
+
+
+async def test_bad_credentials_are_not_masked_as_unavailable(no_async_sleep):
+    # Retrying a wrong password would be pointless, and reporting it as
+    # "database unavailable" would send the operator chasing the wrong problem.
+    engine = _FakeEngine(
+        _PGError("password authentication failed", "28P01"), then_succeed=False)
+    with pytest.raises(_PGError):
+        await database.ensure_database_reachable(engine)
+    assert engine.calls == 1
+    assert no_async_sleep == []

tests/core/test_plugin_settings_hygiene.py

@@ -0,0 +1,135 @@
+"""Guards against plugin settings outliving the plugin they configure.
+
+The `http` plugin was folded into the unified Monitor entity, but its
+`plugin.http` DEFAULTS entry survived — so the operator saw a plugin that no
+longer exists reported as enabled, with no way to clear it (deleting the row
+did nothing; the default reasserted it). These tests make that class of drift a
+CI failure instead of a support question.
+"""
+from pathlib import Path
+
+from steward.core.settings import DEFAULTS, find_orphaned_plugins
+from steward.settings.routes import CAPABILITY_PLUGINS
+
+BUNDLED_PLUGINS_DIR = Path(__file__).resolve().parents[2] / "plugins"
+
+
+def _bundled_plugin_names() -> set[str]:
+    """Plugin dirs shipped inside the image, identified by their plugin.yaml."""
+    return {
+        entry.name
+        for entry in BUNDLED_PLUGINS_DIR.iterdir()
+        if entry.is_dir() and (entry / "plugin.yaml").exists()
+    }
+
+
+def _default_plugin_names() -> set[str]:
+    prefix = "plugin."
+    return {k[len(prefix):] for k in DEFAULTS if k.startswith(prefix)}
+
+
+def test_bundled_plugins_dir_is_discoverable():
+    # Guards the test itself: a wrong path would make everything below vacuous.
+    assert _bundled_plugin_names(), f"no bundled plugins found under {BUNDLED_PLUGINS_DIR}"
+
+
+def test_every_default_plugin_key_has_a_real_bundled_plugin():
+    """A plugin.* default naming a non-existent plugin is always a bug.
+
+    Bundled plugins ship in the image and version atomically with core, so
+    unlike external plugins they cannot be transiently missing — there is no
+    benign reason for this to fail.
+    """
+    missing = _default_plugin_names() - _bundled_plugin_names()
+    assert not missing, (
+        f"DEFAULTS declares plugin(s) with no bundled directory: {sorted(missing)}. "
+        f"If the plugin was removed, delete its plugin.<name> key from DEFAULTS "
+        f"and add a migration dropping the stored row."
+    )
+
+
+def test_http_plugin_default_is_gone():
+    # Explicit regression: this exact key is what the operator hit.
+    assert "plugin.http" not in DEFAULTS
+
+
+def test_capability_plugins_all_exist():
+    """CAPABILITY_PLUGINS classifies discovered plugins; stale names are dead weight."""
+    missing = CAPABILITY_PLUGINS - _bundled_plugin_names()
+    assert not missing, f"CAPABILITY_PLUGINS names non-existent plugin(s): {sorted(missing)}"
+
+
+# ── orphan detection ─────────────────────────────────────────────────────────
+
+
+def test_orphans_are_stored_names_with_no_installed_plugin():
+    rows = find_orphaned_plugins(["docker", "traefik", "ancient"], ["docker", "traefik"])
+    assert [r["name"] for r in rows] == ["ancient"]
+    assert rows[0]["removable"] is True
+    assert rows[0]["reason"] is None
+
+
+def test_no_orphans_when_everything_is_installed():
+    assert find_orphaned_plugins(["docker", "snmp"], ["docker", "snmp"]) == []
+
+
+def test_installed_plugin_without_stored_settings_is_not_an_orphan():
+    # Never configured is not the same as left behind.
+    assert find_orphaned_plugins([], ["docker"]) == []
+
+
+def test_orphans_are_sorted_and_skip_empty_names():
+    rows = find_orphaned_plugins(["zeta", "", "alpha"], [])
+    assert [r["name"] for r in rows] == ["alpha", "zeta"]
+
+
+def test_accepts_arbitrary_iterables():
+    # Callers pass a generator of discovered dirs, not a list.
+    rows = find_orphaned_plugins((n for n in ["gone"]), (n for n in ["here"]))
+    assert [r["name"] for r in rows] == ["gone"]
+
+
+# ── load failures folded in (issue #2638) ────────────────────────────────────
+
+
+def test_failure_reason_is_attached_to_its_row():
+    rows = find_orphaned_plugins(
+        ["ancient"], [], {"ancient": "Plugin directory not found in: ['/app/plugins']"})
+    assert rows[0]["reason"] == "Plugin directory not found in: ['/app/plugins']"
+
+
+def test_failure_with_no_stored_row_still_gets_a_row():
+    """The banner counts it, so the page must show it — that was the dead end.
+
+    A plugin enabled by a DEFAULTS entry has no stored row, so a stored-only
+    view would leave the banner pointing at a page with nothing on it.
+    """
+    rows = find_orphaned_plugins([], [], {"http": "Plugin directory not found"})
+    assert [r["name"] for r in rows] == ["http"]
+    assert rows[0]["reason"] == "Plugin directory not found"
+
+
+def test_failure_with_no_stored_row_is_not_removable():
+    # Offering Remove would delete nothing while the default reasserts it.
+    rows = find_orphaned_plugins([], [], {"http": "Plugin directory not found"})
+    assert rows[0]["removable"] is False
+
+
+def test_failure_of_an_installed_plugin_is_not_listed_here():
+    # A discovered plugin that failed already shows the reason on its own card.
+    assert find_orphaned_plugins(
+        ["docker"], ["docker"], {"docker": "boom"}) == []
+
+
+def test_stored_orphan_and_failure_are_not_duplicated():
+    rows = find_orphaned_plugins(["gone"], [], {"gone": "Plugin directory not found"})
+    assert len(rows) == 1
+    assert rows[0]["removable"] is True and rows[0]["reason"]
+
+
+def test_every_undiscovered_failure_is_represented():
+    # The invariant that keeps the banner count and this list in agreement.
+    failures = {"a": "x", "b": "y", "docker": "z"}
+    rows = find_orphaned_plugins([], ["docker"], failures)
+    listed = {r["name"] for r in rows}
+    assert listed == {"a", "b"}

tests/core/test_settings_default_plugins.py

@@ -9,7 +9,9 @@ and a DEFAULTS-merged-with-stored dict, mirroring get_all_settings' merge.
 from steward.core import settings as settings_module
 from steward.core.settings import DEFAULTS, to_plugins_cfg
 
-DEFAULT_ON = {"docker", "host_agent", "http", "snmp"}
+# NB: no "http" — that plugin was dissolved into the unified Monitor entity, and
+# its lingering default is what test_plugin_settings_hygiene.py now guards against.
+DEFAULT_ON = {"docker", "host_agent", "snmp"}
 VENDOR_OPT_IN = {"traefik", "unifi"}
 
 
@@ -32,7 +34,7 @@ def test_stored_choice_overrides_default():
     cfg = to_plugins_cfg(merged)
     assert cfg["docker"]["enabled"] is False
     # untouched defaults remain enabled
-    assert cfg["http"]["enabled"] is True
+    assert cfg["snmp"]["enabled"] is True
 
 
 def test_defaults_use_plugin_dot_namespace():

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"]

tests/test_config.py

@@ -65,3 +65,50 @@ def test_secret_key_unpersistable_raises_instead_of_ephemeral(tmp_path, monkeypa
     monkeypatch.delenv("STEWARD_SECRET_KEY", raising=False)
     with pytest.raises(RuntimeError, match="could not persist"):
         _resolve_secret_key({})
+
+
+# ── database connect timeout (bootstrap-only: it governs reaching the DB) ────
+
+
+def test_db_connect_timeout_defaults_when_unset(tmp_path, monkeypatch):
+    from steward.database import DB_CONNECT_TIMEOUT_SECONDS
+    cfg_file = tmp_path / "config.yaml"
+    cfg_file.write_text("database:\n  url: x\nsecret_key: s\n")
+    monkeypatch.delenv("STEWARD_DB_CONNECT_TIMEOUT", raising=False)
+    cfg = load_bootstrap(cfg_file)
+    assert cfg["db_connect_timeout"] == DB_CONNECT_TIMEOUT_SECONDS
+
+
+def test_db_connect_timeout_from_yaml(tmp_path, monkeypatch):
+    cfg_file = tmp_path / "config.yaml"
+    cfg_file.write_text(
+        "database:\n  url: x\n  connect_timeout: 120\nsecret_key: s\n")
+    monkeypatch.delenv("STEWARD_DB_CONNECT_TIMEOUT", raising=False)
+    assert load_bootstrap(cfg_file)["db_connect_timeout"] == 120.0
+
+
+def test_db_connect_timeout_env_overrides_yaml(tmp_path, monkeypatch):
+    cfg_file = tmp_path / "config.yaml"
+    cfg_file.write_text(
+        "database:\n  url: x\n  connect_timeout: 120\nsecret_key: s\n")
+    monkeypatch.setenv("STEWARD_DB_CONNECT_TIMEOUT", "5")
+    assert load_bootstrap(cfg_file)["db_connect_timeout"] == 5.0
+
+
+def test_db_connect_timeout_garbage_falls_back_to_default(tmp_path, monkeypatch):
+    from steward.database import DB_CONNECT_TIMEOUT_SECONDS
+    cfg_file = tmp_path / "config.yaml"
+    cfg_file.write_text("database:\n  url: x\nsecret_key: s\n")
+    monkeypatch.setenv("STEWARD_DB_CONNECT_TIMEOUT", "not-a-number")
+    cfg = load_bootstrap(cfg_file)
+    assert cfg["db_connect_timeout"] == DB_CONNECT_TIMEOUT_SECONDS
+
+
+def test_db_connect_timeout_non_positive_falls_back_to_default(tmp_path, monkeypatch):
+    # 0 would mean "never wait", reinstating the crash this setting exists to fix.
+    from steward.database import DB_CONNECT_TIMEOUT_SECONDS
+    cfg_file = tmp_path / "config.yaml"
+    cfg_file.write_text("database:\n  url: x\nsecret_key: s\n")
+    monkeypatch.setenv("STEWARD_DB_CONNECT_TIMEOUT", "0")
+    cfg = load_bootstrap(cfg_file)
+    assert cfg["db_connect_timeout"] == DB_CONNECT_TIMEOUT_SECONDS