cayce-ops/cayceops/components/monitor/monitor.py

"""monitor — static status page builder.

Reads the shared database and renders a self-contained HTML file (no external
assets, no JS, everything escaped). Contains NO sensitive data: daemon health,
proposal queue counts, and recent audit activity (actor/action/result only —
no email bodies, no detail payloads, no URLs).
"""

from __future__ import annotations

import html
from datetime import UTC, datetime

from sqlalchemy import func, select

from cayceops.config import Config
from cayceops.db import session_scope
from cayceops.models import (
    AuditEntry,
    ContentItem,
    DaemonRun,
    MailMessage,
    Proposal,
    ProposalStatus,
)

DAEMONS = ("scanner", "executor", "maild", "publisher", "monitor")

_STYLE = """<style>
 body{font-family:ui-monospace,Menlo,monospace;background:#0a0908;color:#f2ece0;margin:2rem}
 h1{color:#c9a84c;font-weight:600} h2{color:#c9a84c;margin-top:2rem;font-size:1.1rem}
 table{border-collapse:collapse;margin:.5rem 0}
 td,th{border:1px solid #333;padding:.35rem .8rem;text-align:left}
 .ok{color:#6fce6f}.error{color:#ef6b6b}.muted{color:#888;font-size:.85rem}
 .card{display:inline-block;background:#141210;border:1px solid #2a2a2a;
       border-radius:8px;padding:.8rem 1.4rem;margin:.3rem}
</style>"""


def _fmt_ts(dt) -> str:
    return dt.strftime("%Y-%m-%d %H:%M:%SZ") if dt else "—"


def _daemon_rows(sess) -> str:
    rows = []
    for name in DAEMONS:
        last = (
            sess.query(DaemonRun)
            .filter(DaemonRun.daemon == name)
            .order_by(DaemonRun.started_at.desc())
            .first()
        )
        if last is None:
            rows.append(
                "<tr><td>" + name + "</td><td class=muted>never run</td><td>—</td></tr>"
            )
        else:
            cls = "ok" if last.status == "ok" else "error"
            when = _fmt_ts(last.finished_at or last.started_at)
            rows.append(
                "<tr><td>" + html.escape(name) + "</td>"
                + '<td class="' + cls + '">' + html.escape(last.status) + "</td>"
                + "<td>" + when + "</td></tr>"
            )
    head = "<tr><th>daemon</th><th>last status</th><th>last run</th></tr>"
    return "<table>" + head + "".join(rows) + "</table>"


def _queue_cards(sess) -> str:
    counts = dict(
        sess.query(Proposal.status, func.count(Proposal.id)).group_by(Proposal.status).all()
    )
    cards = []
    for status in ProposalStatus:
        value = counts.get(status, 0)
        cards.append(
            "<div class=card>" + html.escape(status.value)
            + ": <strong>" + str(value) + "</strong></div>"
        )
    return "".join(cards)


def _recent_audit(sess, limit: int = 20) -> str:
    rows = []
    query = sess.query(AuditEntry).order_by(AuditEntry.ts.desc()).limit(limit)
    for entry in query.all():
        cls = "ok" if entry.result == "ok" else "error"
        rows.append(
            "<tr><td>" + _fmt_ts(entry.ts) + "</td>"
            + "<td>" + html.escape(entry.actor) + "</td>"
            + "<td>" + html.escape(entry.action) + "</td>"
            + '<td class="' + cls + '">' + html.escape(entry.result) + "</td></tr>"
        )
    body = "".join(rows) or "<tr><td colspan=4 class=muted>no activity yet</td></tr>"
    return (
        "<table><tr><th>time (utc)</th><th>actor</th><th>action</th><th>result</th></tr>"
        + body
        + "</table>"
    )


def _build(cfg: Config, sess) -> str:
    now = datetime.now(UTC)
    mail_unprocessed = (
        sess.query(func.count(MailMessage.id))
        .filter(
            MailMessage.direction == "in",
            MailMessage.draft_status.in_(["drafted", "none"]),
            MailMessage.classification.isnot(None),
        )
        .scalar()
    )
    content_pending = (
        sess.query(func.count(ContentItem.id))
        .filter(ContentItem.status.in_(["draft", "approved"]))
        .scalar()
    )
    generated = _fmt_ts(now)

    page = f"""<!doctype html>
<html lang=en><head><meta charset=utf-8><title>cayce-ops status</title>{_STYLE}</head>
<body>
<h1>cayce-ops · status</h1>
<p class=muted>generated {generated} · static page, no sensitive data by design</p>

<h2>daemon health</h2>
{_daemon_rows(sess)}

<h2>proposal queue</h2>
{_queue_cards(sess)}

<h2>gates</h2>
<div class=card>mail drafts awaiting approval: <strong>{mail_unprocessed}</strong></div>
<div class=card>content items draft/approved: <strong>{content_pending}</strong></div>

<h2>recent audit activity</h2>
{_recent_audit(sess)}

<p class=muted>audit entries: {sess.scalar(select(func.count(AuditEntry.id)))} total</p>
</body></html>"""

    import os

    os.makedirs(os.path.dirname(cfg.monitor_output) or ".", exist_ok=True)
    with open(cfg.monitor_output, "w", encoding="utf-8") as fh:
        fh.write(page)
    return cfg.monitor_output


def build(cfg: Config | None = None) -> int:
    """CLI entry: render the status page inside a DaemonRun bookkeeping wrap."""
    from cayceops.config import load
    from cayceops.db import engine

    cfg = cfg or load()
    try:
        with session_scope() as sess:
            run_row = DaemonRun(daemon="monitor")
            sess.add(run_row)
            sess.flush()
            try:
                out = _build(cfg, sess)
            except Exception as exc:  # noqa: BLE001 — daemon boundary
                run_row.status = "error"
                run_row.detail = {"error": str(exc)[:300]}
                run_row.finished_at = datetime.now(UTC)
                return 1
            run_row.status = "ok"
            run_row.detail = {"output": out}
            run_row.finished_at = datetime.now(UTC)
        print(out)
        return 0
    finally:
        engine().dispose()