cayce
ffe425320d
- audit() writes are serialized: pg_advisory_xact_lock on Postgres; in-process RLock held until session commit/rollback on SQLite (test: 4 concurrent writers x 25 entries -> chain verifies, 100 rows) - legacy rows (entry_hash NULL, pre-0002) are skipped: the chain begins GENESIS at the first hashed row; migrated DBs verify clean by design - monitor anchor: publishes the last chain hash as a signed NIP-01 note once per day, logs it in the audit log (self-verified before publish) tests: concurrent writers; legacy-genesis verification; anchor signs, logs once per day.
254 lines
8.1 KiB
Python
254 lines
8.1 KiB
Python
"""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 _anchor(cfg: Config, sess) -> int:
|
|
"""Anchor core (injectable session for tests). Returns 0 ok / 0 skip."""
|
|
from cayceops.components.publisher import nostr
|
|
from cayceops.config import ConfigError
|
|
from cayceops.models import AuditEntry, audit
|
|
|
|
if not cfg.nostr_private_key or not cfg.nostr_relays:
|
|
raise ConfigError(
|
|
"anchor requires CAYCE_OPS_NOSTR_PRIVATE_KEY and CAYCE_OPS_NOSTR_RELAYS"
|
|
)
|
|
if not cfg.deployment_id:
|
|
raise ConfigError("anchor requires CAYCE_OPS_DEPLOYMENT_ID")
|
|
|
|
last = (
|
|
sess.query(AuditEntry)
|
|
.filter(AuditEntry.entry_hash.isnot(None))
|
|
.order_by(AuditEntry.id.desc())
|
|
.first()
|
|
)
|
|
if last is None:
|
|
print("nothing to anchor: chain empty")
|
|
return 0
|
|
today = datetime.now(UTC).date()
|
|
last_anchor = (
|
|
sess.query(AuditEntry)
|
|
.filter(AuditEntry.action == "anchor")
|
|
.order_by(AuditEntry.id.desc())
|
|
.first()
|
|
)
|
|
if last_anchor is not None and last_anchor.ts.date() == today:
|
|
print("already anchored today")
|
|
return 0
|
|
|
|
secret = nostr.normalize_key(cfg.nostr_private_key)
|
|
event = nostr.build_event(
|
|
secret,
|
|
f"cayce-ops audit anchor deployment={cfg.deployment_id} "
|
|
f"last_hash={last.entry_hash} entries={last.id}",
|
|
tags=[["t", "cayce-ops"], ["t", "audit-anchor"]],
|
|
)
|
|
if not nostr.verify_event(event):
|
|
raise RuntimeError("anchor event failed self-verification")
|
|
results = nostr.publish(cfg.nostr_relays, event)
|
|
audit(
|
|
sess,
|
|
actor="daemon:monitor",
|
|
action="anchor",
|
|
detail={
|
|
"last_hash": last.entry_hash,
|
|
"last_entry": last.id,
|
|
"event_id": event["id"],
|
|
"relays": results,
|
|
"deployment": cfg.deployment_id,
|
|
},
|
|
)
|
|
print(f"anchored: {event['id']}")
|
|
return 0
|
|
|
|
|
|
def anchor(cfg: Config | None = None) -> int:
|
|
"""CLI: anchor once per day (external anchoring, audit round 2 point 4)."""
|
|
from cayceops.config import ConfigError, load
|
|
from cayceops.db import engine
|
|
|
|
cfg = cfg or load()
|
|
try:
|
|
with session_scope() as sess:
|
|
try:
|
|
return _anchor(cfg, sess)
|
|
except ConfigError as exc:
|
|
print(f"anchor refused: {exc}")
|
|
return 2
|
|
except Exception as exc: # noqa: BLE001 — CLI boundary
|
|
print(f"anchor failed: {str(exc)[:200]}")
|
|
return 1
|
|
finally:
|
|
engine().dispose()
|
|
|
|
|
|
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()
|