From 98003f3458aee2af5bdcbbd77fc098233b1b7e2f Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Mon, 13 Apr 2026 23:46:37 -0400 Subject: [PATCH] docs(plugins): host_agent implementation plan 15-task TDD plan covering plugin scaffolding, agent collectors, ingest route, install script, settings UI, widgets, detail page, scheduler, and integration test. Co-Authored-By: Claude Opus 4.6 --- docs/plugins/host-agent-plan.md | 2641 +++++++++++++++++++++++++++++++ 1 file changed, 2641 insertions(+) create mode 100644 docs/plugins/host-agent-plan.md diff --git a/docs/plugins/host-agent-plan.md b/docs/plugins/host-agent-plan.md new file mode 100644 index 0000000..fbafab8 --- /dev/null +++ b/docs/plugins/host-agent-plan.md @@ -0,0 +1,2641 @@ +# Host Agent Plugin Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Ship a `host_agent` Roundtable plugin plus a stdlib-only Python push agent that reports CPU/memory/storage/load/uptime from remote Linux hosts to Roundtable. + +**Architecture:** Plugin lives under `plugins/host_agent/` (mirrors `plugins/http/`). It owns a private `host_agent_registrations` table, writes time-series data into the core `PluginMetric` bus, serves its own agent source at `GET /plugins/host_agent/agent.py`, and renders a per-host install script at `GET /plugins/host_agent/install.sh`. The agent is a single ~300-line Python script with a 30s collect→POST loop, in-memory ring buffer, exponential backoff, and systemd unit installed by a curl one-liner. + +**Tech Stack:** Quart Blueprint, SQLAlchemy (async), Alembic (plugin migration dir), Jinja2 for `install.sh.j2`, pytest + pytest_asyncio. Agent: Python 3.8+ stdlib only (`urllib.request`, `json`, `os`, `shutil`, `time`, `socket`, `signal`, `secrets` not needed — token is server-side). + +**Spec:** `docs/plugins/host-agent-design.md` + +--- + +## File Structure + +**New files (plugin):** + +- `plugins/host_agent/__init__.py` — plugin entry: `setup(app)`, `get_scheduled_tasks()`, `get_blueprint()`. +- `plugins/host_agent/plugin.yaml` — metadata + default config. +- `plugins/host_agent/models.py` — `HostAgentRegistration` SQLAlchemy model. +- `plugins/host_agent/routes.py` — Quart blueprint: ingest, install.sh, agent.py, detail, settings, widgets. +- `plugins/host_agent/scheduler.py` — stale-agent marker task. +- `plugins/host_agent/agent.py` — the Python agent script, served to targets. +- `plugins/host_agent/migrations/__init__.py` +- `plugins/host_agent/migrations/env.py` — alembic env (copy of http plugin's, but imports `roundtable.models.base`). +- `plugins/host_agent/migrations/versions/__init__.py` +- `plugins/host_agent/migrations/versions/host_agent_001_initial.py` — creates `host_agent_registrations`. +- `plugins/host_agent/templates/install.sh.j2` — Jinja install script template. +- `plugins/host_agent/templates/settings_list.html` — plugin settings page. +- `plugins/host_agent/templates/detail.html` — per-host detail page. +- `plugins/host_agent/templates/widget_table.html` — fleet table widget partial. +- `plugins/host_agent/templates/widget_history.html` — history chart widget partial. + +**New files (tests):** + +- `tests/plugins/host_agent/__init__.py` +- `tests/plugins/host_agent/conftest.py` — plugin-local fixtures (fake /proc dir, registered host helper). +- `tests/plugins/host_agent/test_agent_collectors.py` +- `tests/plugins/host_agent/test_agent_config.py` +- `tests/plugins/host_agent/test_agent_ring_buffer.py` +- `tests/plugins/host_agent/test_agent_build_payload.py` +- `tests/plugins/host_agent/test_ingest_route.py` +- `tests/plugins/host_agent/test_install_route.py` +- `tests/plugins/host_agent/test_settings_routes.py` +- `tests/plugins/host_agent/test_scheduler.py` +- `tests/plugins/host_agent/test_migration.py` + +**Modified files (core, small edits):** + +- `roundtable/alerts/routes.py` — add `"host_agent"` entry to `METRIC_CATALOG`. +- `roundtable/core/widgets.py` — add `host_resources` and `host_resource_history` entries. +- `docs/plugins/index.yaml.example` — add catalog entry. + +--- + +## Task 1: Plugin scaffolding + migration + +**Files:** +- Create: `plugins/host_agent/__init__.py` +- Create: `plugins/host_agent/plugin.yaml` +- Create: `plugins/host_agent/models.py` +- Create: `plugins/host_agent/routes.py` +- Create: `plugins/host_agent/migrations/__init__.py` +- Create: `plugins/host_agent/migrations/env.py` +- Create: `plugins/host_agent/migrations/versions/__init__.py` +- Create: `plugins/host_agent/migrations/versions/host_agent_001_initial.py` +- Create: `tests/plugins/host_agent/__init__.py` +- Create: `tests/plugins/host_agent/test_migration.py` + +- [ ] **Step 1: Write the migration smoke test** + +```python +# tests/plugins/host_agent/test_migration.py +"""Smoke test: plugin migration upgrades and downgrades cleanly.""" +import pytest +from sqlalchemy import inspect, text +from sqlalchemy.ext.asyncio import create_async_engine + +pytestmark = pytest.mark.asyncio + + +async def test_host_agent_migration_creates_table(app): + # The app fixture runs all plugin migrations on startup. + from roundtable.core.db import get_engine + engine = get_engine() + async with engine.connect() as conn: + def _check(sync_conn): + insp = inspect(sync_conn) + assert "host_agent_registrations" in insp.get_table_names() + cols = {c["name"] for c in insp.get_columns("host_agent_registrations")} + assert {"id", "host_id", "token_hash", "last_seen_at", + "agent_version", "kernel", "distro", "arch", + "created_at", "updated_at"} <= cols + await conn.run_sync(_check) +``` + +- [ ] **Step 2: Run it — expect failure** + +Run: `pytest tests/plugins/host_agent/test_migration.py -v` +Expected: FAIL (plugin not loaded, table missing). + +- [ ] **Step 3: Create `plugins/host_agent/plugin.yaml`** + +```yaml +# plugins/host_agent/plugin.yaml +name: host_agent +version: "1.0.0" +description: "Remote Linux host resource monitoring via a lightweight Python push agent (CPU, memory, storage, load, uptime)" +author: "Roundtable" +license: "MIT" +min_app_version: "0.1.0" +repository_url: "https://git.fabledsword.com/bvandeusen/Roundtable-plugins" +homepage: "https://git.fabledsword.com/bvandeusen/Roundtable-plugins/src/branch/main/host_agent" +tags: + - host + - monitoring + - cpu + - memory + - storage + +config: + stale_after_seconds: 180 + default_interval_seconds: 30 +``` + +- [ ] **Step 4: Create `plugins/host_agent/__init__.py`** + +```python +# plugins/host_agent/__init__.py +from __future__ import annotations +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from quart import Quart + +_app: "Quart | None" = None + + +def setup(app: "Quart") -> None: + global _app + _app = app + + +def get_scheduled_tasks() -> list: + from .scheduler import make_stale_task + return [make_stale_task(_app)] + + +def get_blueprint(): + from .routes import host_agent_bp + return host_agent_bp +``` + +- [ ] **Step 5: Create `plugins/host_agent/models.py`** + +```python +# plugins/host_agent/models.py +from __future__ import annotations +import uuid +from datetime import datetime, timezone +from sqlalchemy import Column, String, DateTime, ForeignKey +from roundtable.models.base import Base + + +def _uuid() -> str: + return str(uuid.uuid4()) + + +def _utcnow() -> datetime: + return datetime.now(timezone.utc) + + +class HostAgentRegistration(Base): + __tablename__ = "host_agent_registrations" + + id = Column(String(36), primary_key=True, default=_uuid) + host_id = Column(String(36), ForeignKey("hosts.id", ondelete="CASCADE"), + unique=True, nullable=False) + token_hash = Column(String(64), nullable=False, unique=True, index=True) + token_created_at = Column(DateTime(timezone=True), nullable=False, default=_utcnow) + agent_version = Column(String(32), nullable=True) + kernel = Column(String(128), nullable=True) + distro = Column(String(128), nullable=True) + arch = Column(String(32), nullable=True) + last_seen_at = Column(DateTime(timezone=True), nullable=True) + created_at = Column(DateTime(timezone=True), nullable=False, default=_utcnow) + updated_at = Column(DateTime(timezone=True), nullable=False, + default=_utcnow, onupdate=_utcnow) +``` + +- [ ] **Step 6: Create the migration file** + +```python +# plugins/host_agent/migrations/versions/host_agent_001_initial.py +"""host_agent plugin initial tables + +Revision ID: host_agent_001_initial +Revises: (none — branch via depends_on) +""" +from typing import Sequence, Union +from alembic import op +import sqlalchemy as sa + +revision: str = "host_agent_001_initial" +down_revision: Union[str, None] = None +branch_labels: Union[str, Sequence[str], None] = "host_agent" +depends_on: Union[str, Sequence[str], None] = ("0004_app_settings",) + + +def upgrade() -> None: + op.create_table( + "host_agent_registrations", + sa.Column("id", sa.String(36), primary_key=True), + sa.Column("host_id", sa.String(36), + sa.ForeignKey("hosts.id", ondelete="CASCADE"), + nullable=False, unique=True), + sa.Column("token_hash", sa.String(64), nullable=False, unique=True, index=True), + sa.Column("token_created_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("agent_version", sa.String(32), nullable=True), + sa.Column("kernel", sa.String(128), nullable=True), + sa.Column("distro", sa.String(128), nullable=True), + sa.Column("arch", sa.String(32), nullable=True), + sa.Column("last_seen_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False), + ) + + +def downgrade() -> None: + op.drop_table("host_agent_registrations") +``` + +- [ ] **Step 7: Create `plugins/host_agent/migrations/env.py`** + +```python +# plugins/host_agent/migrations/env.py +"""Alembic env.py for the host_agent plugin.""" +import asyncio +import os +import sys +from pathlib import Path +from logging.config import fileConfig + +from sqlalchemy import pool +from sqlalchemy.engine import Connection +from sqlalchemy.ext.asyncio import async_engine_from_config +from alembic import context + +sys.path.insert(0, str(Path(__file__).parent.parent.parent.parent)) + +from roundtable.models.base import Base +import roundtable.models # noqa: F401 +from plugins.host_agent.models import HostAgentRegistration # noqa: F401 + +config = context.config +if config.config_file_name is not None: + fileConfig(config.config_file_name) + +target_metadata = Base.metadata + + +def _get_url() -> str: + import yaml + cfg_path = os.environ.get("ROUNDTABLE_CONFIG", "config.yaml") + try: + with open(cfg_path) as f: + cfg = yaml.safe_load(f) or {} + url = cfg.get("database", {}).get("url", "") + except FileNotFoundError: + url = "" + return os.environ.get("ROUNDTABLE_DATABASE__URL", url) + + +def run_migrations_offline() -> None: + context.configure( + url=_get_url(), target_metadata=target_metadata, + literal_binds=True, dialect_opts={"paramstyle": "named"}, + ) + with context.begin_transaction(): + context.run_migrations() + + +def do_run_migrations(connection: Connection) -> None: + context.configure(connection=connection, target_metadata=target_metadata) + with context.begin_transaction(): + context.run_migrations() + + +async def run_async_migrations() -> None: + cfg = config.get_section(config.config_ini_section, {}) + cfg["sqlalchemy.url"] = _get_url() + connectable = async_engine_from_config(cfg, prefix="sqlalchemy.", poolclass=pool.NullPool) + async with connectable.connect() as connection: + await connection.run_sync(do_run_migrations) + await connectable.dispose() + + +def run_migrations_online() -> None: + asyncio.run(run_async_migrations()) + + +if context.is_offline_mode(): + run_migrations_offline() +else: + run_migrations_online() +``` + +- [ ] **Step 8: Create empty `__init__.py` files** + +```python +# plugins/host_agent/migrations/__init__.py +``` + +```python +# plugins/host_agent/migrations/versions/__init__.py +``` + +```python +# tests/plugins/host_agent/__init__.py +``` + +- [ ] **Step 9: Create a minimal `routes.py` stub so the loader can import** + +```python +# plugins/host_agent/routes.py +from __future__ import annotations +from quart import Blueprint + +host_agent_bp = Blueprint("host_agent", __name__, template_folder="templates") +``` + +- [ ] **Step 10: Create a minimal `scheduler.py` stub** + +```python +# plugins/host_agent/scheduler.py +from __future__ import annotations + + +def make_stale_task(app): + async def _task(): + return None + return {"name": "host_agent.mark_stale", "interval_seconds": 60, "func": _task} +``` + +- [ ] **Step 11: Enable the plugin in test config** + +Check `tests/conftest.py` for the `config_file` fixture. Add the plugin to the fixture's test config: + +```python +# tests/conftest.py — extend the config_file fixture body +f.write_text(textwrap.dedent("""\ + secret_key: test-secret-key-32-chars-minimum!! + database: + url: postgresql+asyncpg://test:test@localhost/test + plugins: + host_agent: + enabled: true +""")) +``` + +(If the existing fixture already has a `plugins:` key, append `host_agent` under it instead. Read the file before editing.) + +- [ ] **Step 12: Run the migration test** + +Run: `pytest tests/plugins/host_agent/test_migration.py -v` +Expected: PASS — the plugin loader picks up the new plugin, runs the migration, and the table exists. + +- [ ] **Step 13: Commit** + +```bash +git add plugins/host_agent tests/plugins/host_agent tests/conftest.py +git commit -m "feat(host_agent): plugin scaffolding and initial migration" +``` + +--- + +## Task 2: Agent config parser + +**Files:** +- Create: `plugins/host_agent/agent.py` (partial — config parser only) +- Create: `tests/plugins/host_agent/test_agent_config.py` + +- [ ] **Step 1: Write the failing test** + +```python +# tests/plugins/host_agent/test_agent_config.py +"""Unit tests for the agent's homegrown config parser.""" +import pytest +from plugins.host_agent.agent import read_config, ConfigError + + +def test_parses_flat_key_value(tmp_path): + p = tmp_path / "agent.conf" + p.write_text( + "url = https://roundtable.example\n" + "token = abc123\n" + "interval_seconds = 45\n" + ) + cfg = read_config(str(p)) + assert cfg["url"] == "https://roundtable.example" + assert cfg["token"] == "abc123" + assert cfg["interval_seconds"] == 45 + + +def test_ignores_blank_lines_and_comments(tmp_path): + p = tmp_path / "agent.conf" + p.write_text( + "# top comment\n" + "\n" + "url = https://x\n" + " # indented comment\n" + "token = t\n" + ) + cfg = read_config(str(p)) + assert cfg == {"url": "https://x", "token": "t"} + + +def test_parses_mounts_as_list(tmp_path): + p = tmp_path / "agent.conf" + p.write_text("url = x\ntoken = y\nmounts = /, /mnt/data, /srv\n") + cfg = read_config(str(p)) + assert cfg["mounts"] == ["/", "/mnt/data", "/srv"] + + +def test_missing_required_fields_raises(tmp_path): + p = tmp_path / "agent.conf" + p.write_text("url = https://x\n") + with pytest.raises(ConfigError, match="token"): + read_config(str(p)) + + +def test_malformed_line_raises(tmp_path): + p = tmp_path / "agent.conf" + p.write_text("url https://x\ntoken = y\n") + with pytest.raises(ConfigError): + read_config(str(p)) +``` + +- [ ] **Step 2: Run — expect import error** + +Run: `pytest tests/plugins/host_agent/test_agent_config.py -v` +Expected: FAIL (`read_config` does not exist). + +- [ ] **Step 3: Implement the parser** + +```python +# plugins/host_agent/agent.py +"""Roundtable host agent — pushes resource metrics to a Roundtable instance. + +Python 3.8+ stdlib only. Target ~300 lines. Served to targets at +GET /plugins/host_agent/agent.py. +""" +from __future__ import annotations + +import json +import os +import shutil +import signal +import socket +import sys +import time +import urllib.error +import urllib.request +from collections import deque +from datetime import datetime, timezone + +AGENT_VERSION = "1.0.0" + + +class ConfigError(Exception): + pass + + +REQUIRED_KEYS = ("url", "token") +INT_KEYS = ("interval_seconds",) +LIST_KEYS = ("mounts",) + + +def read_config(path: str) -> dict: + """Parse a flat `key = value` config file.""" + cfg: dict = {} + try: + with open(path, "r", encoding="utf-8") as f: + for lineno, raw in enumerate(f, 1): + line = raw.strip() + if not line or line.startswith("#"): + continue + if "=" not in line: + raise ConfigError(f"{path}:{lineno}: expected 'key = value'") + key, _, value = line.partition("=") + key = key.strip() + value = value.strip() + if key in INT_KEYS: + try: + cfg[key] = int(value) + except ValueError: + 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()] + else: + cfg[key] = value + except FileNotFoundError: + raise ConfigError(f"{path}: not found") + + missing = [k for k in REQUIRED_KEYS if k not in cfg] + if missing: + raise ConfigError(f"{path}: missing required key(s): {', '.join(missing)}") + + cfg.setdefault("interval_seconds", 30) + return cfg +``` + +- [ ] **Step 4: Run — expect pass** + +Run: `pytest tests/plugins/host_agent/test_agent_config.py -v` +Expected: PASS (5 tests). + +- [ ] **Step 5: Commit** + +```bash +git add plugins/host_agent/agent.py tests/plugins/host_agent/test_agent_config.py +git commit -m "feat(host_agent): agent config parser" +``` + +--- + +## Task 3: Agent collectors + +**Files:** +- Modify: `plugins/host_agent/agent.py` — append collector functions. +- Create: `tests/plugins/host_agent/test_agent_collectors.py` + +- [ ] **Step 1: Write failing tests** + +```python +# tests/plugins/host_agent/test_agent_collectors.py +"""Unit tests for per-resource collectors, driven by fixture /proc files.""" +from unittest.mock import patch +import pytest +from plugins.host_agent import agent as a + + +PROC_STAT_SAMPLE_1 = "cpu 100 0 50 850 0 0 0 0 0 0\n" +PROC_STAT_SAMPLE_2 = "cpu 200 0 100 1700 0 0 0 0 0 0\n" + +PROC_MEMINFO = ( + "MemTotal: 16000000 kB\n" + "MemFree: 2000000 kB\n" + "MemAvailable: 6000000 kB\n" + "SwapTotal: 4000000 kB\n" + "SwapFree: 3500000 kB\n" +) + +PROC_LOADAVG = "0.42 0.55 0.61 1/234 5678\n" + +PROC_UPTIME = "482934.12 1000000.00\n" + +OS_RELEASE = 'PRETTY_NAME="Ubuntu 24.04 LTS"\nNAME="Ubuntu"\n' + + +def test_collect_memory_uses_available(tmp_path): + p = tmp_path / "meminfo" + p.write_text(PROC_MEMINFO) + with patch.object(a, "MEMINFO_PATH", str(p)): + mem = a.collect_memory() + # used = total - available + assert mem["total_bytes"] == 16000000 * 1024 + assert mem["available_bytes"] == 6000000 * 1024 + assert mem["used_bytes"] == (16000000 - 6000000) * 1024 + assert mem["swap_used_bytes"] == (4000000 - 3500000) * 1024 + + +def test_collect_load(tmp_path): + p = tmp_path / "loadavg" + p.write_text(PROC_LOADAVG) + with patch.object(a, "LOADAVG_PATH", str(p)): + load = a.collect_load() + assert load == {"1m": 0.42, "5m": 0.55, "15m": 0.61} + + +def test_collect_uptime(tmp_path): + p = tmp_path / "uptime" + p.write_text(PROC_UPTIME) + with patch.object(a, "UPTIME_PATH", str(p)): + assert a.collect_uptime() == 482934 + + +def test_collect_cpu_reads_twice(tmp_path): + p = tmp_path / "stat" + reads = iter([PROC_STAT_SAMPLE_1, PROC_STAT_SAMPLE_2]) + + def _read(_path): + return next(reads) + + with patch.object(a, "STAT_PATH", str(p)), \ + patch.object(a, "_read_file", _read), \ + patch.object(a, "time") as mock_time: + mock_time.sleep = lambda _s: None + pct = a.collect_cpu(sample_window=0.0) + # totals: 1000 → 2000 (delta 1000). idle: 850 → 1700 (delta 850). + # busy = delta_total - delta_idle = 150. pct = 15.0. + assert pct == pytest.approx(15.0, abs=0.1) + + +def test_collect_storage(tmp_path): + fake = {"/": (1000, 400, 600), "/mnt/data": (2000, 500, 1500)} + + def _fake_usage(path): + total, used, free = fake[path] + class R: pass + r = R() + r.total = total + r.used = used + r.free = free + return r + + with patch.object(a.shutil, "disk_usage", side_effect=_fake_usage): + out = a.collect_storage(["/", "/mnt/data"]) + assert out == [ + {"mount": "/", "total_bytes": 1000, "used_bytes": 400}, + {"mount": "/mnt/data", "total_bytes": 2000, "used_bytes": 500}, + ] + + +def test_collect_metadata(tmp_path): + p = tmp_path / "os-release" + p.write_text(OS_RELEASE) + + class Uname: + release = "6.8.0-45-generic" + machine = "x86_64" + + with patch.object(a, "OS_RELEASE_PATH", str(p)), \ + patch.object(a.os, "uname", return_value=Uname()): + meta = a.collect_metadata() + assert meta["kernel"] == "6.8.0-45-generic" + assert meta["arch"] == "x86_64" + assert "Ubuntu 24.04" in meta["distro"] +``` + +- [ ] **Step 2: Run — expect failure** + +Run: `pytest tests/plugins/host_agent/test_agent_collectors.py -v` +Expected: FAIL (collectors not defined). + +- [ ] **Step 3: Append collectors to `plugins/host_agent/agent.py`** + +```python +# ─── collectors ────────────────────────────────────────────────────────────── + +STAT_PATH = "/proc/stat" +MEMINFO_PATH = "/proc/meminfo" +LOADAVG_PATH = "/proc/loadavg" +UPTIME_PATH = "/proc/uptime" +OS_RELEASE_PATH = "/etc/os-release" + + +def _read_file(path: str) -> str: + with open(path, "r", encoding="utf-8") as f: + return f.read() + + +def _parse_cpu_line(stat_text: str) -> tuple[int, int]: + """Return (total_jiffies, idle_jiffies) for the aggregate cpu line.""" + for line in stat_text.splitlines(): + if line.startswith("cpu "): + parts = line.split() + fields = [int(x) for x in parts[1:]] + idle = fields[3] + (fields[4] if len(fields) > 4 else 0) # idle + iowait + total = sum(fields) + return total, idle + raise RuntimeError("no aggregate cpu line in /proc/stat") + + +def collect_cpu(sample_window: float = 0.2) -> float: + """Return CPU utilization % over sample_window seconds.""" + t1, i1 = _parse_cpu_line(_read_file(STAT_PATH)) + time.sleep(sample_window) + t2, i2 = _parse_cpu_line(_read_file(STAT_PATH)) + dt = t2 - t1 + di = i2 - i1 + if dt <= 0: + return 0.0 + return round(100.0 * (dt - di) / dt, 2) + + +def collect_memory() -> dict: + info: dict[str, int] = {} + for line in _read_file(MEMINFO_PATH).splitlines(): + if ":" not in line: + continue + key, _, rest = line.partition(":") + parts = rest.strip().split() + if not parts: + continue + try: + value_kb = int(parts[0]) + except ValueError: + continue + info[key] = value_kb * 1024 # bytes + total = info.get("MemTotal", 0) + available = info.get("MemAvailable", 0) + swap_total = info.get("SwapTotal", 0) + swap_free = info.get("SwapFree", 0) + return { + "total_bytes": total, + "used_bytes": max(total - available, 0), + "available_bytes": available, + "swap_used_bytes": max(swap_total - swap_free, 0), + } + + +def collect_storage(mounts: list[str]) -> list[dict]: + out: list[dict] = [] + for m in mounts: + try: + u = shutil.disk_usage(m) + except (FileNotFoundError, PermissionError): + continue + out.append({"mount": m, "total_bytes": u.total, "used_bytes": u.used}) + return out + + +def collect_load() -> dict: + parts = _read_file(LOADAVG_PATH).split() + return {"1m": float(parts[0]), "5m": float(parts[1]), "15m": float(parts[2])} + + +def collect_uptime() -> int: + return int(float(_read_file(UPTIME_PATH).split()[0])) + + +def collect_metadata() -> dict: + u = os.uname() + distro = "unknown" + try: + for line in _read_file(OS_RELEASE_PATH).splitlines(): + if line.startswith("PRETTY_NAME="): + distro = line.split("=", 1)[1].strip().strip('"') + break + except (OSError, FileNotFoundError): + pass + return {"kernel": u.release, "distro": distro, "arch": u.machine} + + +def default_mounts() -> list[str]: + """Return real mount points from /proc/mounts, skipping pseudo filesystems.""" + skip_types = {"tmpfs", "devtmpfs", "proc", "sysfs", "cgroup", "cgroup2", + "overlay", "squashfs", "ramfs", "devpts", "mqueue", "pstore", + "securityfs", "debugfs", "tracefs", "hugetlbfs", "fusectl", + "configfs", "bpf", "autofs", "efivarfs", "binfmt_misc"} + mounts: list[str] = [] + try: + with open("/proc/mounts", "r") as f: + for line in f: + parts = line.split() + if len(parts) < 3: + continue + if parts[2] in skip_types: + continue + mounts.append(parts[1]) + except OSError: + return ["/"] + return mounts or ["/"] +``` + +- [ ] **Step 4: Run — expect pass** + +Run: `pytest tests/plugins/host_agent/test_agent_collectors.py -v` +Expected: PASS (6 tests). + +- [ ] **Step 5: Commit** + +```bash +git add plugins/host_agent/agent.py tests/plugins/host_agent/test_agent_collectors.py +git commit -m "feat(host_agent): agent resource collectors" +``` + +--- + +## Task 4: Agent payload builder + ring buffer + +**Files:** +- Modify: `plugins/host_agent/agent.py` — add `RingBuffer`, `build_payload()`. +- Create: `tests/plugins/host_agent/test_agent_ring_buffer.py` +- Create: `tests/plugins/host_agent/test_agent_build_payload.py` + +- [ ] **Step 1: Write failing ring buffer test** + +```python +# tests/plugins/host_agent/test_agent_ring_buffer.py +from plugins.host_agent.agent import RingBuffer + + +def test_ring_buffer_preserves_order(): + rb = RingBuffer(maxlen=3) + rb.push(1); rb.push(2); rb.push(3) + assert list(rb.drain()) == [1, 2, 3] + assert len(rb) == 0 + + +def test_ring_buffer_drops_oldest_when_full(): + rb = RingBuffer(maxlen=3) + for v in (1, 2, 3, 4, 5): + rb.push(v) + assert list(rb.drain()) == [3, 4, 5] + + +def test_ring_buffer_drain_is_atomic(): + rb = RingBuffer(maxlen=5) + rb.push("a"); rb.push("b") + out = list(rb.drain()) + rb.push("c") + assert out == ["a", "b"] + assert list(rb.drain()) == ["c"] +``` + +- [ ] **Step 2: Write failing build_payload test** + +```python +# tests/plugins/host_agent/test_agent_build_payload.py +from unittest.mock import patch +from plugins.host_agent import agent as a + + +def test_build_payload_shape(): + fake_sample = { + "ts": "2026-04-14T00:00:00+00:00", + "cpu_pct": 12.5, + "mem": {"total_bytes": 100, "used_bytes": 40, + "available_bytes": 60, "swap_used_bytes": 0}, + "load": {"1m": 0.1, "5m": 0.2, "15m": 0.3}, + "uptime_secs": 1234, + "storage": [{"mount": "/", "total_bytes": 1000, "used_bytes": 400}], + } + metadata = {"kernel": "6.8", "distro": "Ubuntu", "arch": "x86_64"} + + with patch.object(a, "collect_cpu", return_value=12.5), \ + patch.object(a, "collect_memory", return_value=fake_sample["mem"]), \ + patch.object(a, "collect_load", return_value=fake_sample["load"]), \ + patch.object(a, "collect_uptime", return_value=1234), \ + patch.object(a, "collect_storage", return_value=fake_sample["storage"]): + sample = a.build_sample(mounts=["/"]) + + payload = a.build_payload( + samples=[sample], + hostname="myhost", + metadata=metadata, + ) + assert payload["agent_version"] == a.AGENT_VERSION + assert payload["hostname"] == "myhost" + assert payload["metadata"] == metadata + assert len(payload["samples"]) == 1 + s = payload["samples"][0] + assert s["cpu_pct"] == 12.5 + assert s["mem"]["used_bytes"] == 40 + assert s["load"]["1m"] == 0.1 + assert s["storage"][0]["mount"] == "/" + assert "ts" in s +``` + +- [ ] **Step 3: Run — expect failures** + +Run: `pytest tests/plugins/host_agent/test_agent_ring_buffer.py tests/plugins/host_agent/test_agent_build_payload.py -v` +Expected: FAIL. + +- [ ] **Step 4: Append to `plugins/host_agent/agent.py`** + +```python +# ─── ring buffer ───────────────────────────────────────────────────────────── + + +class RingBuffer: + def __init__(self, maxlen: int = 20) -> None: + self._dq: deque = deque(maxlen=maxlen) + + def push(self, item) -> None: + self._dq.append(item) + + def drain(self) -> list: + out = list(self._dq) + self._dq.clear() + return out + + def __len__(self) -> int: + return len(self._dq) + + +# ─── payload ───────────────────────────────────────────────────────────────── + + +def build_sample(mounts: list[str]) -> dict: + """Collect one full sample. Partial samples allowed if a collector fails.""" + sample: dict = {"ts": datetime.now(timezone.utc).isoformat()} + try: + sample["cpu_pct"] = collect_cpu() + except Exception: + sample["cpu_pct"] = None + try: + sample["mem"] = collect_memory() + except Exception: + sample["mem"] = None + try: + sample["load"] = collect_load() + except Exception: + sample["load"] = None + try: + sample["uptime_secs"] = collect_uptime() + except Exception: + sample["uptime_secs"] = None + try: + sample["storage"] = collect_storage(mounts) + except Exception: + sample["storage"] = [] + return sample + + +def build_payload(samples: list[dict], hostname: str, metadata: dict) -> dict: + return { + "agent_version": AGENT_VERSION, + "hostname": hostname, + "metadata": metadata, + "samples": samples, + } +``` + +- [ ] **Step 5: Run — expect pass** + +Run: `pytest tests/plugins/host_agent/test_agent_ring_buffer.py tests/plugins/host_agent/test_agent_build_payload.py -v` +Expected: PASS. + +- [ ] **Step 6: Commit** + +```bash +git add plugins/host_agent/agent.py tests/plugins/host_agent/test_agent_ring_buffer.py tests/plugins/host_agent/test_agent_build_payload.py +git commit -m "feat(host_agent): agent ring buffer and payload builder" +``` + +--- + +## Task 5: Agent POST + backoff + main loop + +**Files:** +- Modify: `plugins/host_agent/agent.py` — add `post_payload`, backoff, `main_loop`, `__main__` guard. + +The main loop does not get a direct unit test — it's thin glue over already-tested pieces and a real sleep loop. We test `post_payload` and the backoff schedule in isolation instead. + +- [ ] **Step 1: Add test for post_payload and backoff schedule** + +```python +# Append to tests/plugins/host_agent/test_agent_build_payload.py +import urllib.error +from unittest.mock import MagicMock, patch + + +def test_post_payload_success(): + from plugins.host_agent import agent as a + captured = {} + + class FakeResp: + status = 200 + def __enter__(self): return self + def __exit__(self, *a): return False + def read(self): return b'{"ok":true}' + + def fake_urlopen(req, timeout=10): + captured["url"] = req.full_url + captured["headers"] = dict(req.header_items()) + captured["body"] = req.data + return FakeResp() + + with patch.object(a.urllib.request, "urlopen", side_effect=fake_urlopen): + ok, status = a.post_payload("https://x", "tok", {"hello": "world"}) + assert ok is True + assert status == 200 + assert captured["url"] == "https://x/plugins/host_agent/ingest" + assert captured["headers"]["Authorization"] == "Bearer tok" + + +def test_post_payload_http_error_returns_status(): + from plugins.host_agent import agent as a + + def raise_401(req, timeout=10): + raise urllib.error.HTTPError(req.full_url, 401, "unauthorized", {}, None) + + with patch.object(a.urllib.request, "urlopen", side_effect=raise_401): + ok, status = a.post_payload("https://x", "tok", {}) + assert ok is False + assert status == 401 + + +def test_post_payload_network_error_returns_none_status(): + from plugins.host_agent import agent as a + + def raise_url(req, timeout=10): + raise urllib.error.URLError("conn refused") + + with patch.object(a.urllib.request, "urlopen", side_effect=raise_url): + ok, status = a.post_payload("https://x", "tok", {}) + assert ok is False + assert status is None + + +def test_backoff_schedule(): + from plugins.host_agent.agent import next_backoff + assert next_backoff(0) == 30 + assert next_backoff(30) == 60 + assert next_backoff(60) == 120 + assert next_backoff(120) == 240 + assert next_backoff(240) == 300 # capped + assert next_backoff(300) == 300 +``` + +- [ ] **Step 2: Run — expect failure** + +Run: `pytest tests/plugins/host_agent/test_agent_build_payload.py -v` +Expected: FAIL on the new tests (`post_payload`, `next_backoff` missing). + +- [ ] **Step 3: Append to `plugins/host_agent/agent.py`** + +```python +# ─── POST + backoff ────────────────────────────────────────────────────────── + + +BACKOFF_CAP = 300 + + +def next_backoff(current: int) -> int: + if current <= 0: + return 30 + return min(current * 2, BACKOFF_CAP) + + +def post_payload(url: str, token: str, payload: dict) -> tuple[bool, int | None]: + body = json.dumps(payload).encode("utf-8") + req = urllib.request.Request( + url.rstrip("/") + "/plugins/host_agent/ingest", + data=body, + headers={ + "Content-Type": "application/json", + "Authorization": f"Bearer {token}", + "User-Agent": f"roundtable-host-agent/{AGENT_VERSION}", + }, + method="POST", + ) + try: + with urllib.request.urlopen(req, timeout=10) as resp: + return (200 <= resp.status < 300, resp.status) + except urllib.error.HTTPError as e: + return (False, e.code) + except (urllib.error.URLError, TimeoutError, socket.timeout, OSError): + return (False, None) + + +# ─── main loop ─────────────────────────────────────────────────────────────── + + +_reload_requested = False +_shutdown_requested = False + + +def _handle_hup(_signum, _frame): + global _reload_requested + _reload_requested = True + + +def _handle_term(_signum, _frame): + global _shutdown_requested + _shutdown_requested = True + + +def _log(level: str, msg: str) -> None: + sys.stderr.write(f"[{level}] {msg}\n") + sys.stderr.flush() + + +def main_loop(conf_path: str) -> int: + global _reload_requested, _shutdown_requested + signal.signal(signal.SIGHUP, _handle_hup) + signal.signal(signal.SIGTERM, _handle_term) + + try: + cfg = read_config(conf_path) + except ConfigError as e: + _log("ERROR", str(e)) + return 2 + + metadata = collect_metadata() + hostname = cfg.get("hostname") or socket.gethostname() + mounts = cfg.get("mounts") or default_mounts() + buffer = RingBuffer(maxlen=20) + backoff = 0 + + _log("INFO", f"roundtable-host-agent {AGENT_VERSION} starting " + f"(url={cfg['url']}, interval={cfg['interval_seconds']}s)") + + while not _shutdown_requested: + if _reload_requested: + try: + cfg = read_config(conf_path) + mounts = cfg.get("mounts") or default_mounts() + _log("INFO", "config reloaded") + except ConfigError as e: + _log("ERROR", f"reload failed: {e}") + _reload_requested = False + + sample = build_sample(mounts) + buffered = buffer.drain() + payload = build_payload( + samples=buffered + [sample], + hostname=hostname, + metadata=metadata, + ) + ok, status = post_payload(cfg["url"], cfg["token"], payload) + + if ok: + if buffered: + _log("INFO", f"flushed {len(buffered)} buffered samples") + backoff = 0 + sleep_for = cfg["interval_seconds"] + else: + if status == 400: + _log("ERROR", "server rejected payload (400) — dropping sample") + # Don't re-buffer — broken payload will just be rejected again. + elif status == 401: + _log("ERROR", "token rejected (401) — check config + UI") + buffer.push(sample) + else: + _log("WARN", f"POST failed (status={status}); buffering") + # Re-buffer what we drained plus the new sample. + for s in buffered: + buffer.push(s) + buffer.push(sample) + backoff = next_backoff(backoff) + sleep_for = backoff + + # Interruptible sleep + slept = 0.0 + while slept < sleep_for and not _shutdown_requested and not _reload_requested: + time.sleep(min(1.0, sleep_for - slept)) + slept += 1.0 + + _log("INFO", "SIGTERM — flushing and exiting") + # Best-effort final flush. + final = buffer.drain() + if final: + post_payload(cfg["url"], cfg["token"], + build_payload(final, hostname, metadata)) + return 0 + + +if __name__ == "__main__": + conf = os.environ.get("ROUNDTABLE_AGENT_CONFIG", "/etc/roundtable-agent.conf") + sys.exit(main_loop(conf)) +``` + +- [ ] **Step 4: Run — expect pass** + +Run: `pytest tests/plugins/host_agent -v` +Expected: PASS across all agent unit tests. + +- [ ] **Step 5: Line-count sanity check** + +Run: `wc -l plugins/host_agent/agent.py` +Expected: under ~350 lines. If significantly over, review for over-scoping. + +- [ ] **Step 6: Commit** + +```bash +git add plugins/host_agent/agent.py tests/plugins/host_agent/test_agent_build_payload.py +git commit -m "feat(host_agent): agent POST, backoff, and main loop" +``` + +--- + +## Task 6: Ingest route (happy path) + +**Files:** +- Modify: `plugins/host_agent/routes.py` +- Create: `tests/plugins/host_agent/conftest.py` +- Create: `tests/plugins/host_agent/test_ingest_route.py` + +- [ ] **Step 1: Write plugin-local test fixtures** + +```python +# tests/plugins/host_agent/conftest.py +"""Fixtures for host_agent plugin tests.""" +import hashlib +import pytest_asyncio + +from roundtable.models.hosts import Host +from roundtable.core.db import get_session +from plugins.host_agent.models import HostAgentRegistration + + +@pytest_asyncio.fixture +async def registered_host(app): + """Create a Host + HostAgentRegistration with a known raw token.""" + raw_token = "test-token-abcdef" + token_hash = hashlib.sha256(raw_token.encode()).hexdigest() + + async with get_session() as session: + host = Host(name="testhost", address="10.0.0.1") + session.add(host) + await session.flush() + reg = HostAgentRegistration( + host_id=host.id, + token_hash=token_hash, + ) + session.add(reg) + await session.commit() + return {"host": host, "registration": reg, "token": raw_token} +``` + +> **Note:** Verify `Host.__init__` accepts `name=` and `address=` kwargs before relying on this. If the real Host model requires more fields (e.g., `probe_type`), pass them here. Read `roundtable/models/hosts.py` first to confirm. + +- [ ] **Step 2: Write failing ingest test** + +```python +# tests/plugins/host_agent/test_ingest_route.py +import json +from datetime import datetime, timezone +import pytest +from sqlalchemy import select + +from roundtable.models.metrics import PluginMetric +from roundtable.core.db import get_session +from plugins.host_agent.models import HostAgentRegistration + +pytestmark = pytest.mark.asyncio + + +def _sample() -> dict: + return { + "ts": datetime.now(timezone.utc).isoformat(), + "cpu_pct": 12.5, + "mem": { + "total_bytes": 16_000_000_000, + "used_bytes": 8_000_000_000, + "available_bytes": 8_000_000_000, + "swap_used_bytes": 0, + }, + "load": {"1m": 0.1, "5m": 0.2, "15m": 0.3}, + "uptime_secs": 1234, + "storage": [ + {"mount": "/", "total_bytes": 1000, "used_bytes": 400}, + {"mount": "/mnt/data", "total_bytes": 2000, "used_bytes": 1800}, + ], + } + + +def _payload() -> dict: + return { + "agent_version": "1.0.0", + "hostname": "testhost", + "metadata": {"kernel": "6.8", "distro": "Ubuntu 24.04", "arch": "x86_64"}, + "samples": [_sample()], + } + + +async def test_ingest_happy_path_writes_metrics(client, registered_host): + resp = await client.post( + "/plugins/host_agent/ingest", + headers={"Authorization": f"Bearer {registered_host['token']}", + "Content-Type": "application/json"}, + data=json.dumps(_payload()), + ) + assert resp.status_code == 200 + body = await resp.get_json() + assert body["ok"] is True + assert body["samples_accepted"] == 1 + + async with get_session() as session: + rows = (await session.execute( + select(PluginMetric).where(PluginMetric.source_module == "host_agent") + )).scalars().all() + metric_names = {r.metric_name for r in rows} + assert "cpu_pct" in metric_names + assert "mem_used_pct" in metric_names + assert "mem_available_bytes" in metric_names + assert "disk_used_pct_worst" in metric_names + assert "load_1m" in metric_names + + # Per-mount rows use ":" resource_name. + per_mount = [r for r in rows if r.metric_name == "disk_used_pct"] + assert {r.resource_name for r in per_mount} == {"testhost:/", "testhost:/mnt/data"} + # Worst mount is /mnt/data at 90%. + worst = [r for r in rows if r.metric_name == "disk_used_pct_worst"] + assert len(worst) == 1 + assert worst[0].value == pytest.approx(90.0, abs=0.1) + + +async def test_ingest_updates_registration(client, registered_host): + resp = await client.post( + "/plugins/host_agent/ingest", + headers={"Authorization": f"Bearer {registered_host['token']}"}, + data=json.dumps(_payload()), + ) + assert resp.status_code == 200 + + async with get_session() as session: + reg = (await session.execute( + select(HostAgentRegistration).where( + HostAgentRegistration.host_id == registered_host["host"].id) + )).scalar_one() + assert reg.last_seen_at is not None + assert reg.agent_version == "1.0.0" + assert reg.kernel == "6.8" + assert reg.distro == "Ubuntu 24.04" + assert reg.arch == "x86_64" +``` + +- [ ] **Step 3: Run — expect failure** + +Run: `pytest tests/plugins/host_agent/test_ingest_route.py -v` +Expected: FAIL (route returns 404). + +- [ ] **Step 4: Implement ingest route** + +Replace `plugins/host_agent/routes.py`: + +```python +# plugins/host_agent/routes.py +from __future__ import annotations + +import hashlib +import json +from datetime import datetime, timezone +from typing import Any + +from quart import Blueprint, jsonify, request +from sqlalchemy import select + +from roundtable.core.db import get_session +from roundtable.models.hosts import Host +from roundtable.models.metrics import PluginMetric +from .models import HostAgentRegistration + +host_agent_bp = Blueprint("host_agent", __name__, template_folder="templates") + +SOURCE_MODULE = "host_agent" + + +def _hash_token(raw: str) -> str: + return hashlib.sha256(raw.encode("utf-8")).hexdigest() + + +def _parse_ts(ts: str) -> datetime: + # Python's fromisoformat in 3.11+ handles trailing Z; be lenient anyway. + if ts.endswith("Z"): + ts = ts[:-1] + "+00:00" + return datetime.fromisoformat(ts) + + +def _expand_sample_to_metrics( + sample: dict, host_name: str, recorded_at: datetime +) -> list[PluginMetric]: + rows: list[PluginMetric] = [] + + def row(metric: str, resource: str, value: float) -> None: + rows.append(PluginMetric( + source_module=SOURCE_MODULE, + resource_name=resource, + metric_name=metric, + value=float(value), + recorded_at=recorded_at, + )) + + if sample.get("cpu_pct") is not None: + row("cpu_pct", host_name, sample["cpu_pct"]) + + mem = sample.get("mem") or {} + if mem.get("total_bytes"): + total = mem["total_bytes"] + available = mem.get("available_bytes", 0) + used_pct = 100.0 * (total - available) / total if total else 0.0 + row("mem_used_pct", host_name, used_pct) + row("mem_available_bytes", host_name, available) + row("swap_used_bytes", host_name, mem.get("swap_used_bytes", 0)) + + load = sample.get("load") or {} + for key in ("1m", "5m", "15m"): + if key in load: + row(f"load_{key}", host_name, load[key]) + + if sample.get("uptime_secs") is not None: + row("uptime_secs", host_name, sample["uptime_secs"]) + + worst_pct = 0.0 + for disk in sample.get("storage") or []: + total = disk.get("total_bytes", 0) + used = disk.get("used_bytes", 0) + pct = 100.0 * used / total if total else 0.0 + mount_res = f"{host_name}:{disk['mount']}" + row("disk_used_pct", mount_res, pct) + row("disk_used_bytes", mount_res, used) + row("disk_total_bytes", mount_res, total) + if pct > worst_pct: + worst_pct = pct + if sample.get("storage"): + row("disk_used_pct_worst", host_name, worst_pct) + + return rows + + +def _error(status: int, code: str, detail: str | None = None) -> tuple[Any, int]: + body: dict = {"ok": False, "error": code} + if detail: + body["detail"] = detail + return jsonify(body), status + + +@host_agent_bp.post("/ingest") +async def ingest(): + auth = request.headers.get("Authorization", "") + if not auth.startswith("Bearer "): + return _error(401, "invalid_token") + raw = auth[len("Bearer "):].strip() + if not raw: + return _error(401, "invalid_token") + + try: + payload = await request.get_json(force=True) + except Exception: + return _error(400, "malformed_payload", "invalid JSON") + if not isinstance(payload, dict) or "samples" not in payload: + return _error(400, "malformed_payload", "missing 'samples'") + + samples = payload.get("samples") or [] + if not isinstance(samples, list) or not samples: + return _error(400, "malformed_payload", "'samples' must be a non-empty list") + + token_hash = _hash_token(raw) + async with get_session() as session: + reg = (await session.execute( + select(HostAgentRegistration).where( + HostAgentRegistration.token_hash == token_hash) + )).scalar_one_or_none() + if reg is None: + return _error(401, "invalid_token") + host = (await session.execute( + select(Host).where(Host.id == reg.host_id) + )).scalar_one_or_none() + if host is None: + return _error(401, "invalid_token") + + accepted = 0 + latest_ts: datetime | None = None + for sample in samples: + try: + recorded_at = _parse_ts(sample["ts"]) + except (KeyError, ValueError): + continue + metrics = _expand_sample_to_metrics(sample, host.name, recorded_at) + for m in metrics: + session.add(m) + accepted += 1 + if latest_ts is None or recorded_at > latest_ts: + latest_ts = recorded_at + + if accepted == 0: + await session.rollback() + return _error(400, "malformed_payload", "no valid samples") + + md = payload.get("metadata") or {} + changed = False + if latest_ts and (reg.last_seen_at is None or latest_ts > reg.last_seen_at): + reg.last_seen_at = latest_ts + changed = True + version = payload.get("agent_version") + if version and reg.agent_version != version: + reg.agent_version = version + changed = True + for field in ("kernel", "distro", "arch"): + if field in md and getattr(reg, field) != md[field]: + setattr(reg, field, md[field]) + changed = True + if changed: + reg.updated_at = datetime.now(timezone.utc) + + # Clock-skew warning — accept but log. + if latest_ts: + skew = abs((datetime.now(timezone.utc) - latest_ts).total_seconds()) + if skew > 300: + from quart import current_app + current_app.logger.warning( + "host_agent ingest: clock skew %.0fs for host=%s", skew, host.name) + + await session.commit() + + return jsonify({"ok": True, "samples_accepted": accepted}), 200 +``` + +- [ ] **Step 5: Run — expect pass** + +Run: `pytest tests/plugins/host_agent/test_ingest_route.py -v` +Expected: PASS. + +- [ ] **Step 6: Commit** + +```bash +git add plugins/host_agent/routes.py tests/plugins/host_agent/conftest.py tests/plugins/host_agent/test_ingest_route.py +git commit -m "feat(host_agent): ingest endpoint with token auth and metric expansion" +``` + +--- + +## Task 7: Ingest error paths (token/malformed) + +**Files:** +- Modify: `tests/plugins/host_agent/test_ingest_route.py` + +- [ ] **Step 1: Add failing tests for error paths** + +```python +# Append to tests/plugins/host_agent/test_ingest_route.py + +async def test_ingest_rejects_missing_auth(client): + resp = await client.post( + "/plugins/host_agent/ingest", + data=json.dumps(_payload()), + headers={"Content-Type": "application/json"}, + ) + assert resp.status_code == 401 + body = await resp.get_json() + assert body["error"] == "invalid_token" + + +async def test_ingest_rejects_unknown_token(client): + resp = await client.post( + "/plugins/host_agent/ingest", + headers={"Authorization": "Bearer no-such-token"}, + data=json.dumps(_payload()), + ) + assert resp.status_code == 401 + + +async def test_ingest_rejects_malformed_body(client, registered_host): + resp = await client.post( + "/plugins/host_agent/ingest", + headers={"Authorization": f"Bearer {registered_host['token']}"}, + data="not json", + ) + assert resp.status_code == 400 + + +async def test_ingest_rejects_missing_samples(client, registered_host): + resp = await client.post( + "/plugins/host_agent/ingest", + headers={"Authorization": f"Bearer {registered_host['token']}"}, + data=json.dumps({"agent_version": "1.0.0"}), + ) + assert resp.status_code == 400 + + +async def test_ingest_partial_sample_without_cpu(client, registered_host): + payload = _payload() + payload["samples"][0]["cpu_pct"] = None + resp = await client.post( + "/plugins/host_agent/ingest", + headers={"Authorization": f"Bearer {registered_host['token']}"}, + data=json.dumps(payload), + ) + assert resp.status_code == 200 + body = await resp.get_json() + assert body["samples_accepted"] == 1 +``` + +- [ ] **Step 2: Run** + +Run: `pytest tests/plugins/host_agent/test_ingest_route.py -v` +Expected: PASS (error handling is already implemented in Task 6). + +- [ ] **Step 3: Commit** + +```bash +git add tests/plugins/host_agent/test_ingest_route.py +git commit -m "test(host_agent): ingest error path coverage" +``` + +--- + +## Task 8: install.sh and agent.py serving routes + +**Files:** +- Create: `plugins/host_agent/templates/install.sh.j2` +- Modify: `plugins/host_agent/routes.py` +- Create: `tests/plugins/host_agent/test_install_route.py` + +- [ ] **Step 1: Write failing tests** + +```python +# tests/plugins/host_agent/test_install_route.py +import pytest + +pytestmark = pytest.mark.asyncio + + +async def test_install_sh_requires_valid_token(client): + resp = await client.get("/plugins/host_agent/install.sh?token=bogus") + assert resp.status_code == 401 + + +async def test_install_sh_renders_with_token(client, registered_host): + resp = await client.get( + f"/plugins/host_agent/install.sh?token={registered_host['token']}") + assert resp.status_code == 200 + assert resp.content_type.startswith("text/plain") + text = (await resp.get_data()).decode() + assert "roundtable-agent" in text + assert registered_host["token"] in text + assert "systemctl enable --now" in text + assert "NoNewPrivileges=yes" in text + assert "--uninstall" in text + + +async def test_agent_py_serves_script(client): + resp = await client.get("/plugins/host_agent/agent.py") + assert resp.status_code == 200 + assert resp.content_type.startswith("text/x-python") + text = (await resp.get_data()).decode() + assert "def main_loop" in text + assert "AGENT_VERSION" in text +``` + +- [ ] **Step 2: Run — expect failure** + +Run: `pytest tests/plugins/host_agent/test_install_route.py -v` +Expected: FAIL (routes missing). + +- [ ] **Step 3: Create `plugins/host_agent/templates/install.sh.j2`** + +Paste the full install script from the spec (§ Install script → Rendered script structure, `docs/plugins/host-agent-design.md:327-409`). Substitute the Jinja variables `{{ url }}`, `{{ token }}`, `{{ agent_version }}`, `{{ host_name }}`, `{{ host_address }}`. Do NOT edit the script structure — the spec is the source of truth. + +- [ ] **Step 4: Add routes to `plugins/host_agent/routes.py`** + +Append imports: + +```python +from pathlib import Path +from quart import render_template, Response, current_app +``` + +Append routes: + +```python +AGENT_SOURCE_PATH = Path(__file__).parent / "agent.py" + + +@host_agent_bp.get("/install.sh") +async def install_script(): + token = request.args.get("token", "") + if not token: + return _error(401, "invalid_token") + async with get_session() as session: + reg = (await session.execute( + select(HostAgentRegistration).where( + HostAgentRegistration.token_hash == _hash_token(token)) + )).scalar_one_or_none() + if reg is None: + return _error(401, "invalid_token") + host = (await session.execute( + select(Host).where(Host.id == reg.host_id) + )).scalar_one_or_none() + if host is None: + return _error(401, "invalid_token") + + # Build the public URL the agent should POST to. + url = request.host_url.rstrip("/") + + rendered = await render_template( + "install.sh.j2", + url=url, + token=token, + agent_version=_agent_version(), + host_name=host.name, + host_address=host.address, + ) + return Response(rendered, mimetype="text/plain") + + +@host_agent_bp.get("/agent.py") +async def agent_source(): + try: + body = AGENT_SOURCE_PATH.read_text(encoding="utf-8") + except OSError: + return _error(500, "agent_missing") + return Response(body, mimetype="text/x-python") + + +def _agent_version() -> str: + # Lightweight parse of agent.py for AGENT_VERSION. Avoids importing the + # agent module (which has signal-handler and main-loop side effects). + try: + for line in AGENT_SOURCE_PATH.read_text().splitlines(): + if line.startswith("AGENT_VERSION"): + return line.split("=", 1)[1].strip().strip('"').strip("'") + except OSError: + pass + return "unknown" +``` + +- [ ] **Step 5: Run — expect pass** + +Run: `pytest tests/plugins/host_agent/test_install_route.py -v` +Expected: PASS. + +- [ ] **Step 6: Shellcheck the rendered install script** + +Run: +```bash +python -c " +from jinja2 import Environment, FileSystemLoader +env = Environment(loader=FileSystemLoader('plugins/host_agent/templates')) +print(env.get_template('install.sh.j2').render( + url='https://r.example', token='tok', agent_version='1.0.0', + host_name='test', host_address='10.0.0.1')) +" > /tmp/install.sh +sh -n /tmp/install.sh +command -v shellcheck >/dev/null && shellcheck /tmp/install.sh || echo "shellcheck not installed (optional)" +``` +Expected: `sh -n` clean. shellcheck warnings optional. + +- [ ] **Step 7: Commit** + +```bash +git add plugins/host_agent/routes.py plugins/host_agent/templates/install.sh.j2 tests/plugins/host_agent/test_install_route.py +git commit -m "feat(host_agent): install.sh and agent.py serving routes" +``` + +--- + +## Task 9: Settings routes (add, rotate, delete) + +**Files:** +- Modify: `plugins/host_agent/routes.py` +- Create: `plugins/host_agent/templates/settings_list.html` +- Create: `tests/plugins/host_agent/test_settings_routes.py` + +Auth note: the existing Roundtable admin decorator pattern needs to be used here. Read `roundtable/settings/routes.py` (or similar) to find the decorator name — likely something like `@require_admin` or a session check. Use whatever the rest of the codebase uses. **Do not invent a new auth mechanism.** + +- [ ] **Step 1: Find the admin auth decorator** + +Run: `grep -rn "require_admin\|@admin_required\|def admin" roundtable/settings/ roundtable/core/auth.py 2>/dev/null | head -20` +Note the decorator name and import path for use in Step 3. + +- [ ] **Step 2: Write failing test** + +```python +# tests/plugins/host_agent/test_settings_routes.py +import pytest +from sqlalchemy import select + +from roundtable.core.db import get_session +from roundtable.models.hosts import Host +from plugins.host_agent.models import HostAgentRegistration + +pytestmark = pytest.mark.asyncio + + +async def test_add_host_creates_registration_and_returns_install_url(admin_client): + resp = await admin_client.post( + "/plugins/host_agent/settings/add-host", + form={"name": "newhost", "address": "10.0.0.99"}, + ) + assert resp.status_code in (200, 302) + + async with get_session() as session: + host = (await session.execute( + select(Host).where(Host.name == "newhost"))).scalar_one() + reg = (await session.execute( + select(HostAgentRegistration).where( + HostAgentRegistration.host_id == host.id))).scalar_one() + assert reg.token_hash # hash persisted + assert len(reg.token_hash) == 64 # sha256 hex + + +async def test_rotate_token_changes_hash(admin_client, registered_host): + old_hash = registered_host["registration"].token_hash + resp = await admin_client.post( + f"/plugins/host_agent/settings/{registered_host['host'].id}/rotate-token") + assert resp.status_code in (200, 302) + + async with get_session() as session: + reg = (await session.execute( + select(HostAgentRegistration).where( + HostAgentRegistration.host_id == registered_host["host"].id) + )).scalar_one() + assert reg.token_hash != old_hash + + +async def test_delete_registration(admin_client, registered_host): + resp = await admin_client.post( + f"/plugins/host_agent/settings/{registered_host['host'].id}/delete") + assert resp.status_code in (200, 302) + + async with get_session() as session: + reg = (await session.execute( + select(HostAgentRegistration).where( + HostAgentRegistration.host_id == registered_host["host"].id) + )).scalar_one_or_none() + assert reg is None +``` + +> **The `admin_client` fixture**: Check `tests/conftest.py` for an existing admin-authenticated test client. If none exists, add one that logs in as an admin session before returning the client. This is an existing codebase pattern — don't reinvent it. Use whatever other plugin tests use for authenticated routes (e.g., grep `admin_client\|auth.*client` under `tests/`). + +- [ ] **Step 3: Run — expect failure** + +Run: `pytest tests/plugins/host_agent/test_settings_routes.py -v` + +- [ ] **Step 4: Implement routes** + +Append to `plugins/host_agent/routes.py`: + +```python +import secrets +from quart import redirect, url_for + +# TODO: replace _require_admin with the project-wide decorator found in Step 1. +# Placeholder below mirrors the shape; swap for real admin auth. +from roundtable.core.auth import require_admin # adjust import to actual path + + +def _new_token_pair() -> tuple[str, str]: + raw = secrets.token_urlsafe(32) + return raw, _hash_token(raw) + + +@host_agent_bp.post("/settings/add-host") +@require_admin +async def add_host(): + form = await request.form + name = (form.get("name") or "").strip() + address = (form.get("address") or "").strip() + if not name: + return _error(400, "missing_name") + + async with get_session() as session: + host = (await session.execute( + select(Host).where(Host.name == name))).scalar_one_or_none() + if host is None: + host = Host(name=name, address=address) + session.add(host) + await session.flush() + existing = (await session.execute( + select(HostAgentRegistration).where( + HostAgentRegistration.host_id == host.id))).scalar_one_or_none() + if existing is not None: + await session.rollback() + return _error(400, "already_registered") + + raw, hashed = _new_token_pair() + reg = HostAgentRegistration(host_id=host.id, token_hash=hashed) + session.add(reg) + await session.commit() + + # Store raw token in flash / query for one-time display on the settings page. + return redirect(url_for("host_agent.settings_list") + f"?new_token={raw}&host_id={host.id}") + + +@host_agent_bp.post("/settings//rotate-token") +@require_admin +async def rotate_token(host_id: str): + async with get_session() as session: + reg = (await session.execute( + select(HostAgentRegistration).where( + HostAgentRegistration.host_id == host_id))).scalar_one_or_none() + if reg is None: + return _error(404, "not_found") + raw, hashed = _new_token_pair() + reg.token_hash = hashed + reg.token_created_at = datetime.now(timezone.utc) + await session.commit() + return redirect(url_for("host_agent.settings_list") + f"?new_token={raw}&host_id={host_id}") + + +@host_agent_bp.post("/settings//delete") +@require_admin +async def delete_registration(host_id: str): + async with get_session() as session: + reg = (await session.execute( + select(HostAgentRegistration).where( + HostAgentRegistration.host_id == host_id))).scalar_one_or_none() + if reg is not None: + await session.delete(reg) + await session.commit() + return redirect(url_for("host_agent.settings_list")) + + +@host_agent_bp.get("/settings/") +@require_admin +async def settings_list(): + async with get_session() as session: + regs = (await session.execute(select(HostAgentRegistration))).scalars().all() + hosts_by_id = { + h.id: h for h in (await session.execute( + select(Host).where(Host.id.in_([r.host_id for r in regs])))).scalars().all() + } if regs else {} + + new_token = request.args.get("new_token") + new_host_id = request.args.get("host_id") + install_url = None + if new_token and new_host_id: + install_url = f"{request.host_url.rstrip('/')}/plugins/host_agent/install.sh?token={new_token}" + + return await render_template( + "settings_list.html", + registrations=[ + {"reg": r, "host": hosts_by_id.get(r.host_id)} for r in regs + ], + new_token=new_token, + install_url=install_url, + ) +``` + +- [ ] **Step 5: Create `plugins/host_agent/templates/settings_list.html`** + +```html +{% extends "base.html" %} +{% block title %}Host Agent — Roundtable{% endblock %} +{% block content %} +

Host Agent — Registered Hosts

+ +{% if new_token %} +
+

New token — copy this install command now

+

+ This is the only time the raw token is shown. Rotate if you lose it. +

+ 
curl -sSL '{{ install_url }}' | sudo sh
+
+{% endif %} + +
+
+
+ + +
+
+ + +
+ +
+
+ +
+ + + + + + + + + {% for item in registrations %} + + + + + + + + {% else %} + + {% endfor %} + +
HostAgent versionDistroLast seenActions
{{ item.host.name if item.host else item.reg.host_id }}{{ item.reg.agent_version or "—" }}{{ item.reg.distro or "—" }}{{ item.reg.last_seen_at.strftime("%Y-%m-%d %H:%M:%S") if item.reg.last_seen_at else "never" }} +
+ +
+
+ +
+
No hosts registered. Add one above.
+
+{% endblock %} +``` + +- [ ] **Step 6: Run tests — iterate on auth decorator if needed** + +Run: `pytest tests/plugins/host_agent/test_settings_routes.py -v` +Expected: PASS. If admin auth fails, the actual import path from Step 1 is wrong — fix the `from roundtable.core.auth import require_admin` line. + +- [ ] **Step 7: Commit** + +```bash +git add plugins/host_agent/routes.py plugins/host_agent/templates/settings_list.html tests/plugins/host_agent/test_settings_routes.py +git commit -m "feat(host_agent): plugin settings page — add, rotate, delete" +``` + +--- + +## Task 10: Widgets + METRIC_CATALOG registration + +**Files:** +- Modify: `plugins/host_agent/routes.py` — add `/widget` and `/widget/history` partials. +- Create: `plugins/host_agent/templates/widget_table.html` +- Create: `plugins/host_agent/templates/widget_history.html` +- Modify: `roundtable/core/widgets.py` +- Modify: `roundtable/alerts/routes.py` + +- [ ] **Step 1: Add widget partial routes to `plugins/host_agent/routes.py`** + +```python +@host_agent_bp.get("/widget") +async def widget_table(): + """Fleet-glance table: one row per monitored host, latest metrics.""" + from sqlalchemy import func + async with get_session() as session: + regs = (await session.execute(select(HostAgentRegistration))).scalars().all() + host_ids = [r.host_id for r in regs] + hosts = { + h.id: h for h in (await session.execute( + select(Host).where(Host.id.in_(host_ids)) + )).scalars().all() + } if host_ids else {} + + # Latest value per (resource, metric) for host_agent source. + subq = ( + select( + PluginMetric.resource_name, + PluginMetric.metric_name, + func.max(PluginMetric.recorded_at).label("max_ts"), + ) + .where(PluginMetric.source_module == SOURCE_MODULE) + .group_by(PluginMetric.resource_name, PluginMetric.metric_name) + ).subquery() + latest_rows = (await session.execute( + select(PluginMetric).join( + subq, + (PluginMetric.resource_name == subq.c.resource_name) & + (PluginMetric.metric_name == subq.c.metric_name) & + (PluginMetric.recorded_at == subq.c.max_ts), + ).where(PluginMetric.source_module == SOURCE_MODULE) + )).scalars().all() + + # Group by resource_name (host name); ignore per-mount entries for the table. + latest: dict[str, dict[str, float]] = {} + for row in latest_rows: + if ":" in row.resource_name: + continue + latest.setdefault(row.resource_name, {})[row.metric_name] = row.value + + rows = [] + for reg in regs: + host = hosts.get(reg.host_id) + if host is None: + continue + m = latest.get(host.name, {}) + rows.append({ + "host": host, + "reg": reg, + "cpu_pct": m.get("cpu_pct"), + "mem_used_pct": m.get("mem_used_pct"), + "disk_worst": m.get("disk_used_pct_worst"), + "load_1m": m.get("load_1m"), + }) + + return await render_template("widget_table.html", rows=rows) + + +@host_agent_bp.get("/widget/history") +async def widget_history(): + host_id = request.args.get("host_id", "") + hours = int(request.args.get("hours", "6")) + from datetime import timedelta + cutoff = datetime.now(timezone.utc) - timedelta(hours=hours) + + async with get_session() as session: + host = (await session.execute( + select(Host).where(Host.id == host_id))).scalar_one_or_none() + if host is None: + return _error(404, "not_found") + points = (await session.execute( + select(PluginMetric).where( + PluginMetric.source_module == SOURCE_MODULE, + PluginMetric.resource_name == host.name, + PluginMetric.metric_name.in_( + ("cpu_pct", "mem_used_pct", "disk_used_pct_worst")), + PluginMetric.recorded_at >= cutoff, + ).order_by(PluginMetric.recorded_at) + )).scalars().all() + + series: dict[str, list[dict]] = {"cpu_pct": [], "mem_used_pct": [], "disk_used_pct_worst": []} + for p in points: + series[p.metric_name].append({"t": p.recorded_at.isoformat(), "v": p.value}) + + return await render_template("widget_history.html", host=host, series=series, hours=hours) +``` + +- [ ] **Step 2: Create `plugins/host_agent/templates/widget_table.html`** + +```html + + + + + + {% for r in rows %} + + + + + + + + + {% else %} + + {% endfor %} + +
HostCPU %Mem %Disk % (worst)Load 1mLast seen
{{ r.host.name }}{{ "%.1f"|format(r.cpu_pct) if r.cpu_pct is not none else "—" }}{{ "%.1f"|format(r.mem_used_pct) if r.mem_used_pct is not none else "—" }}{{ "%.1f"|format(r.disk_worst) if r.disk_worst is not none else "—" }}{{ "%.2f"|format(r.load_1m) if r.load_1m is not none else "—" }}{{ r.reg.last_seen_at.strftime("%H:%M:%S") if r.reg.last_seen_at else "never" }}
No hosts with agent data yet.
+``` + +- [ ] **Step 3: Create `plugins/host_agent/templates/widget_history.html`** + +```html +
+

{{ host.name }} — last {{ hours }}h

+ + +
+``` + +- [ ] **Step 4: Register widgets in `roundtable/core/widgets.py`** + +Add to `WIDGET_REGISTRY` (alphabetically by existing convention, or at the end — check the file first): + +```python + "host_resources": { + "key": "host_resources", + "label": "Host Agent — Resources", + "description": "Fleet view: CPU, memory, disk, and load per monitored host", + "hx_url": "/plugins/host_agent/widget", + "detail_url": "/plugins/host_agent/settings/", + "plugin": "host_agent", + "poll": True, + "params": [], + }, + "host_resource_history": { + "key": "host_resource_history", + "label": "Host Agent — History", + "description": "CPU/memory/disk history for one host", + "hx_url": "/plugins/host_agent/widget/history", + "detail_url": "/plugins/host_agent/settings/", + "plugin": "host_agent", + "poll": True, + "params": [ + { + "key": "hours", + "label": "Time range", + "type": "select", + "default": 6, + "options": [ + {"value": 1, "label": "Last 1 hour"}, + {"value": 6, "label": "Last 6 hours"}, + {"value": 24, "label": "Last 24 hours"}, + ], + }, + ], + }, +``` + +- [ ] **Step 5: Register `host_agent` in `METRIC_CATALOG`** + +Read `roundtable/alerts/routes.py`. Locate the `METRIC_CATALOG` dict. Add: + +```python + "host_agent": [ + "cpu_pct", "mem_used_pct", "mem_available_bytes", "swap_used_bytes", + "disk_used_pct_worst", "load_1m", "load_5m", "load_15m", "uptime_secs", + ], +``` + +- [ ] **Step 6: Smoke-test the widget partial via the running app** + +Run: `pytest tests/plugins/host_agent/test_ingest_route.py -v` (existing tests should still pass — confirms no import breakage.) + +Then manually verify via dev server or add a quick route-reachable test: + +```python +# Append to tests/plugins/host_agent/test_ingest_route.py +async def test_widget_table_renders(client, registered_host): + # Ingest something first + resp = await client.post( + "/plugins/host_agent/ingest", + headers={"Authorization": f"Bearer {registered_host['token']}"}, + data=json.dumps(_payload()), + ) + assert resp.status_code == 200 + + resp = await client.get("/plugins/host_agent/widget") + assert resp.status_code == 200 + text = (await resp.get_data()).decode() + assert "testhost" in text +``` + +Run: `pytest tests/plugins/host_agent/test_ingest_route.py::test_widget_table_renders -v` +Expected: PASS. + +- [ ] **Step 7: Commit** + +```bash +git add plugins/host_agent/routes.py plugins/host_agent/templates/widget_table.html plugins/host_agent/templates/widget_history.html roundtable/core/widgets.py roundtable/alerts/routes.py tests/plugins/host_agent/test_ingest_route.py +git commit -m "feat(host_agent): dashboard widgets and alert metric catalog entry" +``` + +--- + +## Task 11: Per-host detail page + +**Files:** +- Create: `plugins/host_agent/templates/detail.html` +- Modify: `plugins/host_agent/routes.py` + +- [ ] **Step 1: Add route** + +```python +@host_agent_bp.get("//") +async def host_detail(host_id: str): + from datetime import timedelta + cutoff = datetime.now(timezone.utc) - timedelta(hours=6) + async with get_session() as session: + host = (await session.execute( + select(Host).where(Host.id == host_id))).scalar_one_or_none() + if host is None: + return _error(404, "not_found") + reg = (await session.execute( + select(HostAgentRegistration).where( + HostAgentRegistration.host_id == host_id))).scalar_one_or_none() + + latest_rows = (await session.execute( + select(PluginMetric).where( + PluginMetric.source_module == SOURCE_MODULE, + PluginMetric.resource_name == host.name, + PluginMetric.recorded_at >= cutoff, + ).order_by(PluginMetric.recorded_at.desc()) + )).scalars().all() + per_mount = (await session.execute( + select(PluginMetric).where( + PluginMetric.source_module == SOURCE_MODULE, + PluginMetric.resource_name.like(f"{host.name}:%"), + PluginMetric.metric_name == "disk_used_pct", + PluginMetric.recorded_at >= cutoff, + ).order_by(PluginMetric.recorded_at.desc()) + )).scalars().all() + + # Latest value per metric name + current: dict[str, float] = {} + for row in latest_rows: + current.setdefault(row.metric_name, row.value) + mounts: dict[str, float] = {} + for row in per_mount: + mount = row.resource_name.split(":", 1)[1] + mounts.setdefault(mount, row.value) + + return await render_template( + "detail.html", host=host, reg=reg, current=current, mounts=mounts, + ) +``` + +- [ ] **Step 2: Create `plugins/host_agent/templates/detail.html`** + +```html +{% extends "base.html" %} +{% block title %}{{ host.name }} — Host Agent{% endblock %} +{% block content %} +

{{ host.name }}

+ +
+
+
CPU
+
{{ "%.1f"|format(current.cpu_pct) if current.cpu_pct is defined else "—" }}%
+
+
+
Memory
+
{{ "%.1f"|format(current.mem_used_pct) if current.mem_used_pct is defined else "—" }}%
+
+
+
Load 1m
+
{{ "%.2f"|format(current.load_1m) if current.load_1m is defined else "—" }}
+
+
+
Uptime
+
{{ (current.uptime_secs // 86400) if current.uptime_secs is defined else "—" }}d
+
+
+ +
+

Storage

+ + + + {% for mount, pct in mounts.items() %} + + {% else %} + + {% endfor %} + +
MountUsed %
{{ mount }}{{ "%.1f"|format(pct) }}%
No data.
+
+ +
+

Agent

+
+
Version
{{ reg.agent_version if reg else "—" }}
+
Kernel
{{ reg.kernel if reg else "—" }}
+
Distro
{{ reg.distro if reg else "—" }}
+
Arch
{{ reg.arch if reg else "—" }}
+
Last seen
{{ reg.last_seen_at if reg and reg.last_seen_at else "never" }}
+
+
+ +
+ Loading history… +
+{% endblock %} +``` + +- [ ] **Step 3: Add test** + +```python +# Append to tests/plugins/host_agent/test_ingest_route.py +async def test_host_detail_page_renders(client, registered_host): + await client.post( + "/plugins/host_agent/ingest", + headers={"Authorization": f"Bearer {registered_host['token']}"}, + data=json.dumps(_payload()), + ) + resp = await client.get(f"/plugins/host_agent/{registered_host['host'].id}/") + assert resp.status_code == 200 + text = (await resp.get_data()).decode() + assert "testhost" in text + assert "Storage" in text +``` + +- [ ] **Step 4: Run** + +Run: `pytest tests/plugins/host_agent/test_ingest_route.py::test_host_detail_page_renders -v` +Expected: PASS. + +- [ ] **Step 5: Commit** + +```bash +git add plugins/host_agent/routes.py plugins/host_agent/templates/detail.html tests/plugins/host_agent/test_ingest_route.py +git commit -m "feat(host_agent): per-host detail page" +``` + +--- + +## Task 12: Stale-agent scheduler + +**Files:** +- Modify: `plugins/host_agent/scheduler.py` +- Create: `tests/plugins/host_agent/test_scheduler.py` + +The scheduler computes a "stale" view from `last_seen_at` — the spec notes this may become redundant once alert rules cover the same ground. It's a one-query safety net. We implement it as a simple function that callers can invoke; the scheduled task is a wrapper that logs stale hosts (no column update — staleness is computed on read). + +- [ ] **Step 1: Write failing test** + +```python +# tests/plugins/host_agent/test_scheduler.py +from datetime import datetime, timedelta, timezone +import pytest +from sqlalchemy import select + +from roundtable.core.db import get_session +from roundtable.models.hosts import Host +from plugins.host_agent.models import HostAgentRegistration +from plugins.host_agent.scheduler import find_stale_registrations + +pytestmark = pytest.mark.asyncio + + +async def test_find_stale_returns_old_registrations(app): + now = datetime.now(timezone.utc) + async with get_session() as session: + h1 = Host(name="fresh", address="") + h2 = Host(name="stale", address="") + h3 = Host(name="never", address="") + session.add_all([h1, h2, h3]) + await session.flush() + session.add_all([ + HostAgentRegistration(host_id=h1.id, token_hash="h1", + last_seen_at=now - timedelta(seconds=30)), + HostAgentRegistration(host_id=h2.id, token_hash="h2", + last_seen_at=now - timedelta(seconds=600)), + HostAgentRegistration(host_id=h3.id, token_hash="h3", + last_seen_at=None), + ]) + await session.commit() + + stale = await find_stale_registrations(stale_after_seconds=180) + stale_names = {s["host_name"] for s in stale} + assert "stale" in stale_names + assert "fresh" not in stale_names + # never-seen hosts are not "stale" — they're unregistered-in-practice; skip. + assert "never" not in stale_names +``` + +- [ ] **Step 2: Run — expect failure** + +Run: `pytest tests/plugins/host_agent/test_scheduler.py -v` +Expected: FAIL. + +- [ ] **Step 3: Implement scheduler** + +Replace `plugins/host_agent/scheduler.py`: + +```python +# plugins/host_agent/scheduler.py +from __future__ import annotations +from datetime import datetime, timedelta, timezone + +from sqlalchemy import select + +from roundtable.core.db import get_session +from roundtable.models.hosts import Host +from .models import HostAgentRegistration + + +async def find_stale_registrations(stale_after_seconds: int = 180) -> list[dict]: + cutoff = datetime.now(timezone.utc) - timedelta(seconds=stale_after_seconds) + async with get_session() as session: + rows = (await session.execute( + select(HostAgentRegistration).where( + HostAgentRegistration.last_seen_at.isnot(None), + HostAgentRegistration.last_seen_at < cutoff, + ) + )).scalars().all() + if not rows: + return [] + hosts = { + h.id: h for h in (await session.execute( + select(Host).where(Host.id.in_([r.host_id for r in rows])) + )).scalars().all() + } + return [ + { + "host_id": r.host_id, + "host_name": hosts[r.host_id].name if r.host_id in hosts else "?", + "last_seen_at": r.last_seen_at, + } + for r in rows + ] + + +def make_stale_task(app): + async def _task(): + stale = await find_stale_registrations() + if stale and app is not None: + app.logger.info("host_agent: %d stale agent(s): %s", + len(stale), [s["host_name"] for s in stale]) + return {"name": "host_agent.mark_stale", "interval_seconds": 60, "func": _task} +``` + +- [ ] **Step 4: Run — expect pass** + +Run: `pytest tests/plugins/host_agent/test_scheduler.py -v` +Expected: PASS. + +- [ ] **Step 5: Commit** + +```bash +git add plugins/host_agent/scheduler.py tests/plugins/host_agent/test_scheduler.py +git commit -m "feat(host_agent): stale-agent scheduler" +``` + +--- + +## Task 13: End-to-end integration test + +This task ties the agent and server together without a real subprocess — we generate a payload via the agent's `build_payload()` with mocked collectors, then POST it through the Quart test client and assert the full pipeline. + +**Files:** +- Create: `tests/plugins/host_agent/test_integration.py` + +- [ ] **Step 1: Write the test** + +```python +# tests/plugins/host_agent/test_integration.py +"""End-to-end: agent build_payload → server ingest → PluginMetric rows.""" +import json +from datetime import datetime, timezone +from unittest.mock import patch +import pytest +from sqlalchemy import select + +from roundtable.core.db import get_session +from roundtable.models.metrics import PluginMetric +from plugins.host_agent import agent as a + +pytestmark = pytest.mark.asyncio + + +async def test_full_pipeline(client, registered_host): + fake_mem = {"total_bytes": 8_000_000_000, "used_bytes": 2_000_000_000, + "available_bytes": 6_000_000_000, "swap_used_bytes": 0} + fake_storage = [{"mount": "/", "total_bytes": 500_000_000_000, + "used_bytes": 100_000_000_000}] + + with patch.object(a, "collect_cpu", return_value=7.5), \ + patch.object(a, "collect_memory", return_value=fake_mem), \ + patch.object(a, "collect_load", return_value={"1m": 0.1, "5m": 0.2, "15m": 0.3}), \ + patch.object(a, "collect_uptime", return_value=100000), \ + patch.object(a, "collect_storage", return_value=fake_storage), \ + patch.object(a, "collect_metadata", return_value={ + "kernel": "6.8", "distro": "Ubuntu", "arch": "x86_64"}): + sample = a.build_sample(mounts=["/"]) + payload = a.build_payload([sample], hostname="testhost", + metadata=a.collect_metadata()) + + resp = await client.post( + "/plugins/host_agent/ingest", + headers={"Authorization": f"Bearer {registered_host['token']}", + "Content-Type": "application/json"}, + data=json.dumps(payload), + ) + assert resp.status_code == 200 + + async with get_session() as session: + rows = (await session.execute( + select(PluginMetric).where( + PluginMetric.source_module == "host_agent", + PluginMetric.resource_name == "testhost", + ) + )).scalars().all() + by_metric = {r.metric_name: r.value for r in rows} + assert by_metric["cpu_pct"] == pytest.approx(7.5) + assert by_metric["mem_used_pct"] == pytest.approx(25.0, abs=0.1) + assert by_metric["disk_used_pct_worst"] == pytest.approx(20.0, abs=0.1) + assert by_metric["load_1m"] == pytest.approx(0.1) +``` + +- [ ] **Step 2: Run** + +Run: `pytest tests/plugins/host_agent/test_integration.py -v` +Expected: PASS. + +- [ ] **Step 3: Run the full plugin test suite** + +Run: `pytest tests/plugins/host_agent -v` +Expected: all pass. + +- [ ] **Step 4: Commit** + +```bash +git add tests/plugins/host_agent/test_integration.py +git commit -m "test(host_agent): end-to-end integration (agent → ingest → metrics)" +``` + +--- + +## Task 14: Catalog entry + +**Files:** +- Modify: `docs/plugins/index.yaml.example` + +- [ ] **Step 1: Add catalog entry** + +Append to the `plugins:` list in `docs/plugins/index.yaml.example`: + +```yaml + - name: host_agent + version: "1.0.0" + description: "Remote Linux host resource monitoring via a lightweight Python push agent" + author: "Roundtable" + license: "MIT" + min_app_version: "0.1.0" + repository_url: "https://git.fabledsword.com/bvandeusen/Roundtable-plugins" + homepage: "https://git.fabledsword.com/bvandeusen/Roundtable-plugins/src/branch/main/host_agent" + download_url: "https://git.fabledsword.com/bvandeusen/Roundtable-plugins/releases/download/host_agent-v1.0.0/host_agent.zip" + checksum_sha256: "" + tags: + - host + - monitoring + - cpu + - memory + - storage +``` + +- [ ] **Step 2: Commit** + +```bash +git add docs/plugins/index.yaml.example +git commit -m "docs(plugins): catalog entry for host_agent" +``` + +--- + +## Task 15: Fable bookkeeping + +Not a code task. After all above tasks are merged: + +- [ ] **Step 1: Transition Fable task #252 from `todo` → `done`** + +Use `fable_update_task` to set status=done on task 252 ("Implement host_agent plugin") with a short completion note linking to the spec and the commit range. + +- [ ] **Step 2: Add a Fable note summarizing what shipped** + +One-paragraph `fable_create_note` attached to Roundtable project (id 6) with: +- Link to spec: `docs/plugins/host-agent-design.md` +- Link to plan: `docs/plugins/host-agent-plan.md` +- Note any deviations from the spec discovered during implementation (e.g., auth decorator name, fixture adjustments). + +--- + +## Final validation + +Before declaring done: + +- [ ] `pytest tests/plugins/host_agent -v` — all green +- [ ] `pytest tests/` — no regressions in existing suites +- [ ] `wc -l plugins/host_agent/agent.py` — under ~350 lines +- [ ] Start the dev server, visit `/plugins/host_agent/settings/`, add a host, verify the install one-liner renders with the correct public URL and an unbuntu VM (or local shell in a throwaway container) can run it and start reporting. +- [ ] Verify the dashboard `host_resources` widget shows the new host after ~30s.