monitor: static status page — health, queue, gates, audit tail

- self-contained HTML (no external assets/JS), everything html-escaped
- daemon health from daemon_runs (last status + last run per daemon)
- proposal queue counts by state; human-gate counters (mail drafts awaiting
  approval, content draft/approved)
- audit tail: actor/action/result only — no detail payloads, no addresses,
  no subjects, no content bodies (tests assert the negatives)
- DaemonRun bookkeeping like every other daemon

tests: 3 new (sections render, leak-negatives, XSS escape) — 44 total

cayceops/components/executor/__init__.py

@@ -0,0 +1 @@
+"""executor package."""

cayceops/components/executor/executor.py

@@ -0,0 +1,246 @@
+"""executor — executes APPROVED proposals through registered action plugins.
+
+Guarantees
+----------
+- Only ``approved`` proposals execute (``AUTONOMOUS_MODE=false`` default).
+- With ``AUTONOMOUS_MODE=true`` the executor gains exactly one capability:
+  bounded auto-approval of ``proposed`` items whose ``kind`` is in
+  ``CAYCE_OPS_AUTONOMOUS_KINDS``, capped by ``CAYCE_OPS_AUTONOMOUS_DAILY_CAP``,
+  recorded with actor ``auto:policy`` (DECISIONS D5).
+- Every transition writes an ``audit_log`` row: who approved, what ran,
+  plugin output, success/failure.
+- Unknown proposal kinds fall back to the ``log_action`` plugin (never a
+  network side effect). See DECISIONS D9.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Callable
+from datetime import UTC, datetime
+from typing import Any
+
+from sqlalchemy import func, select
+
+from cayceops.components.executor import plugins as plugin_registry
+from cayceops.config import Config
+from cayceops.db import session_scope
+from cayceops.models import AuditEntry, DaemonRun, Proposal, ProposalStatus, audit
+
+Actor = str
+
+
+# --- human transitions -------------------------------------------------------
+
+
+def approve(proposal_id: int, actor: Actor, cfg: Config | None = None, sess=None) -> int:
+    """Human approves a proposal. Audited with the operator's identity."""
+    return _decide(proposal_id, actor, ProposalStatus.APPROVED, cfg, sess)
+
+
+def reject(proposal_id: int, actor: Actor, cfg: Config | None = None, sess=None) -> int:
+    """Human rejects a proposal. Audited with the operator's identity."""
+    return _decide(proposal_id, actor, ProposalStatus.REJECTED, cfg, sess)
+
+
+def _decide(
+    proposal_id: int,
+    actor: Actor,
+    decision: ProposalStatus,
+    cfg: Config | None,
+    sess,
+) -> int:
+    owned = False
+    if sess is None:
+        outer = session_scope()
+        sess = outer.__enter__()
+        owned = True
+    try:
+        proposal = sess.get(Proposal, proposal_id)
+        if proposal is None:
+            return 1
+        if proposal.status != ProposalStatus.PROPOSED:
+            verb = "approve" if decision == ProposalStatus.APPROVED else "reject"
+            audit(
+                sess,
+                actor=actor,
+                action=verb,
+                proposal_id=proposal_id,
+                result="error",
+                detail={"reason": f"invalid state: {proposal.status.value}"},
+            )
+            return 1
+        proposal.status = decision
+        if decision == ProposalStatus.APPROVED:
+            proposal.approved_by = actor
+            proposal.approved_at = datetime.now(UTC)
+        verb = "approve" if decision == ProposalStatus.APPROVED else "reject"
+        audit(sess, actor=actor, action=verb, proposal_id=proposal_id)
+        return 0
+    finally:
+        if owned:
+            outer.__exit__(None, None, None)
+
+
+# --- autonomous policy (the ONLY non-human approval path, D5) -----------------
+
+
+def _auto_approve(cfg: Config, sess) -> int:
+    """Auto-approve proposed items per policy. Returns count approved."""
+    if not cfg.autonomous_mode or not cfg.autonomous_kinds:
+        return 0
+    midnight = datetime.now(UTC).replace(hour=0, minute=0, second=0, microsecond=0)
+    used = (
+        sess.query(func.count(AuditEntry.id))
+        .filter(
+            AuditEntry.actor == "auto:policy",
+            AuditEntry.action == "approve",
+            AuditEntry.ts >= midnight,
+        )
+        .scalar()
+    )
+    budget = max(0, cfg.autonomous_daily_cap - int(used or 0))
+    approved = 0
+    if budget == 0:
+        return 0
+    candidates = (
+        sess.query(Proposal)
+        .filter(
+            Proposal.status == ProposalStatus.PROPOSED,
+            Proposal.kind.in_(cfg.autonomous_kinds),
+        )
+        .order_by(Proposal.risk_score.asc(), Proposal.id.asc())
+        .limit(budget)
+        .all()
+    )
+    for proposal in candidates:
+        proposal.status = ProposalStatus.APPROVED
+        proposal.approved_by = "auto:policy"
+        proposal.approved_at = datetime.now(UTC)
+        audit(
+            sess,
+            actor="auto:policy",
+            action="approve",
+            proposal_id=proposal.id,
+            detail={
+                "reason": "autonomous_mode policy",
+                "kind": proposal.kind,
+                "risk_score": proposal.risk_score,
+                "daily_cap": cfg.autonomous_daily_cap,
+            },
+        )
+        approved += 1
+    return approved
+
+
+# --- execution ----------------------------------------------------------------
+
+
+def execute_one(proposal: Proposal, cfg: Config, sess) -> bool:
+    """Run the plugin for one approved proposal. Returns True on success."""
+    proposal.status = ProposalStatus.RUNNING
+    proposal.attempts += 1
+    audit(
+        sess,
+        actor="daemon:executor",
+        action="execute_start",
+        proposal_id=proposal.id,
+        detail={"plugin": proposal.kind, "attempt": proposal.attempts},
+    )
+    plugin = plugin_registry.REGISTRY.get(proposal.kind)
+    try:
+        if plugin is None:
+            outcome: dict[str, Any] = plugin_registry.log_action(proposal, cfg, sess)
+            outcome["plugin_fallback"] = True
+            outcome["note"] = f"no plugin registered for kind {proposal.kind!r}"
+        else:
+            outcome = plugin(proposal, cfg, sess)
+    except Exception as exc:  # noqa: BLE001 — plugin boundary: capture for audit
+        audit(
+            sess,
+            actor="daemon:executor",
+            action="execute_error",
+            proposal_id=proposal.id,
+            result="error",
+            detail={"error": str(exc)[:300], "attempt": proposal.attempts},
+        )
+        if proposal.attempts >= cfg.max_attempts:
+            proposal.status = ProposalStatus.FAILED
+            audit(
+                sess,
+                actor="daemon:executor",
+                action="fail",
+                proposal_id=proposal.id,
+                result="error",
+                detail={"attempts": proposal.attempts},
+            )
+        else:
+            proposal.status = ProposalStatus.APPROVED  # retry next run
+        return False
+
+    proposal.status = ProposalStatus.DONE
+    audit(
+        sess,
+        actor="daemon:executor",
+        action="execute_done",
+        proposal_id=proposal.id,
+        detail={"outcome": outcome},
+    )
+    return True
+
+
+def run(cfg: Config | None = None, sess=None) -> int:
+    """CLI entry: auto-approve per policy, then execute all approved items."""
+    from cayceops.config import load
+    from cayceops.db import engine as _engine
+
+    cfg = cfg or load()
+    owned = False
+    if sess is None:
+        outer = session_scope()
+        sess = outer.__enter__()
+        owned = True
+    try:
+        run_row = DaemonRun(daemon="executor")
+        sess.add(run_row)
+        sess.flush()
+        try:
+            auto = _auto_approve(cfg, sess)
+            pending = (
+                sess.query(Proposal)
+                .filter(Proposal.status == ProposalStatus.APPROVED)
+                .order_by(Proposal.approved_at.asc())
+                .all()
+            )
+            done = failed = 0
+            for proposal in pending:
+                if execute_one(proposal, cfg, sess):
+                    done += 1
+                else:
+                    failed += 1
+            run_row.status = "ok"
+            run_row.detail = {"auto_approved": auto, "done": done, "failed": failed}
+        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.finished_at = datetime.now(UTC)
+        return 0
+    finally:
+        if owned:
+            outer.__exit__(None, None, None)
+            _engine().dispose()
+
+
+def select_approved(sess) -> list[Proposal]:
+    """Helper for tests/tools: currently approved proposals, oldest first."""
+    return list(
+        sess.scalars(
+            select(Proposal)
+            .where(Proposal.status == ProposalStatus.APPROVED)
+            .order_by(Proposal.approved_at.asc())
+        ).all()
+    )
+
+
+ActionPlugin = Callable[[Proposal, Config, Any], dict]

cayceops/components/executor/plugins.py

@@ -0,0 +1,78 @@
+"""executor action plugins.
+
+A plugin maps a proposal ``kind`` to a side effect. Plugins are repo code —
+they are reviewed like everything else and must:
+
+- take ``(proposal, cfg, sess)`` and return a JSON-safe dict (the audit outcome)
+- do their own permission checks (see ``http_post`` + ENABLE_HTTP_ACTIONS)
+- raise on failure (the executor captures the exception into the audit log)
+"""
+
+from __future__ import annotations
+
+import json
+import urllib.error
+import urllib.request
+from typing import Any
+
+from cayceops.config import Config
+from cayceops.models import Proposal, audit
+
+REGISTRY: dict[str, Any] = {}
+
+
+def register(kind: str):
+    """Decorator: register an action plugin for a proposal kind."""
+
+    def deco(fn):
+        REGISTRY[kind] = fn
+        return fn
+
+    return deco
+
+
+@register("log_action")
+def log_action(proposal: Proposal, cfg: Config, sess) -> dict:
+    """No-side-effect plugin: record the execution in the audit log only."""
+    return {"action": "logged", "title": proposal.title[:200]}
+
+
+@register("http_post")
+def http_post(proposal: Proposal, cfg: Config, sess) -> dict:
+    """POST the proposal payload to a webhook.
+
+    Inert unless ``CAYCE_OPS_ENABLE_HTTP_ACTIONS=true``. Target URL comes from
+    the proposal detail (``detail['webhook_url']``) — operator-controlled data.
+    """
+    if not cfg.enable_http_actions:
+        audit(
+            sess,
+            actor="daemon:executor",
+            action="http_post_blocked",
+            proposal_id=proposal.id,
+            result="error",
+            detail={"reason": "CAYCE_OPS_ENABLE_HTTP_ACTIONS is false"},
+        )
+        raise PermissionError("http_post disabled (CAYCE_OPS_ENABLE_HTTP_ACTIONS=false)")
+
+    url = (proposal.detail or {}).get("webhook_url", "")
+    if not url:
+        raise ValueError("proposal detail.webhook_url missing")
+    payload = json.dumps(
+        {"proposal_id": proposal.id, "kind": proposal.kind, "title": proposal.title}
+    ).encode()
+    req = urllib.request.Request(
+        url,
+        data=payload,
+        headers={"Content-Type": "application/json"},
+        method="POST",
+    )
+    with urllib.request.urlopen(req, timeout=20.0) as resp:  # noqa: S310
+        return {"action": "http_post", "status": resp.status}
+
+
+# proposals from the scanner default to content_gap/affiliate/partnership kinds;
+# until dedicated plugins exist for those kinds they must not do anything
+# network-visible. register more plugins here as the operator approves them.
+for _safe_kind in ("affiliate", "partnership", "content_gap", "other"):
+    REGISTRY.setdefault(_safe_kind, log_action)

cayceops/components/maild/__init__.py

@@ -0,0 +1 @@
+"""maild package."""

cayceops/components/maild/classify.py

@@ -0,0 +1,68 @@
+"""maild classification: rules first, LLM optional, conservative threshold.
+
+The LLM never decides alone: any result below ``CAYCE_OPS_LLM_REVIEW_THRESHOLD``
+confidence becomes ``needs_review``. With the ``null`` provider (default) only
+deterministic rules run.
+"""
+
+from __future__ import annotations
+
+import json
+import re
+
+from cayceops.config import Config
+from cayceops.models import MailClassification
+
+_RULES: list[tuple[MailClassification, re.Pattern[str], float]] = [
+    # (class, pattern, confidence when matched)
+    (MailClassification.REJECTION, re.compile(r"\b(declin|reject|not able to|unable to|unfortunately|we regret|won.t be able)\w*", re.I), 0.9),
+    (MailClassification.APPROVAL, re.compile(r"\b(approv|accept|welcome to the|you.re (in|approved)|live|activated|glad to)\w*", re.I), 0.85),
+    (MailClassification.QUESTION, re.compile(r"\b(could you|can you|please (provide|clarify|confirm)|what (is|are)|how (do|does|can)|need more (info|details))\b", re.I), 0.7),
+]
+
+_LLM_PROMPT = """Classify this merchant email into exactly one of:
+approval, rejection, question, other.
+Reply with strict JSON: {"classification": "...", "confidence": 0.0-1.0}
+
+Subject and body:
+{body}"""
+
+
+def _rules(text: str) -> tuple[MailClassification, float]:
+    for kind, pattern, confidence in _RULES:
+        if pattern.search(text):
+            return kind, confidence
+    return MailClassification.OTHER, 0.3
+
+
+def _llm_classify(text: str, cfg: Config) -> tuple[MailClassification, float] | None:
+    """Ask the configured LLM. Returns None on any failure (rules take over)."""
+    from cayceops.llm import get_provider
+
+    try:
+        provider = get_provider(cfg)
+        raw = provider.complete(_LLM_PROMPT.format(body=text[:4000]), system="You classify merchant emails. Reply with JSON only.")
+        data = json.loads(re.search(r"\{.*\}", raw, re.S).group(0))
+        kind = MailClassification(data["classification"].lower())
+        confidence = float(data["confidence"])
+    except Exception:  # noqa: BLE001 — LLM is best-effort by design
+        return None
+    if kind not in MailClassification or kind is MailClassification.NEEDS_REVIEW:
+        return None
+    return kind, min(max(confidence, 0.0), 1.0)
+
+
+def classify(text: str, cfg: Config) -> tuple[MailClassification, float]:
+    """Classify merchant text. Degrades to rules when no LLM is configured."""
+    if cfg.llm_provider != "null":
+        llm_result = _llm_classify(text, cfg)
+        if llm_result is not None:
+            kind, confidence = llm_result
+        else:  # LLM failed — fall through to rules
+            kind, confidence = _rules(text)
+    else:
+        kind, confidence = _rules(text)
+
+    if confidence < cfg.llm_review_threshold:
+        return MailClassification.NEEDS_REVIEW, confidence
+    return kind, confidence

cayceops/components/maild/maild.py

@@ -0,0 +1,364 @@
+"""maild — classify merchant replies, draft responses, gated SMTP.
+
+Flow
+----
+1. ``poll``: fetch unread inbound mail (IMAP), parse, thread to the proposal
+   via ``In-Reply-To`` of a previously-sent draft, classify (rules + optional
+   LLM), store, and create a REPLY DRAFT (never sent automatically).
+2. ``approve_draft``: a human approves a specific draft (audited).
+3. ``send_approved``: SMTP sends *only* drafts in state ``approved``.
+
+Classification is conservative: anything below the confidence threshold becomes
+``needs_review`` and no draft is generated.
+
+Public signatures match the CLI: ``poll()``, ``approve_draft(id, actor)``,
+``send_approved()``. The ``_*`` variants take an explicit session/fetcher for
+tests and composition.
+"""
+
+from __future__ import annotations
+
+import email
+import email.policy
+import imaplib
+import smtplib
+from datetime import UTC, datetime
+from email.message import EmailMessage
+
+from cayceops.components.maild.classify import classify as _classify_text
+from cayceops.config import Config
+from cayceops.db import session_scope
+from cayceops.models import (
+    DaemonRun,
+    DraftStatus,
+    MailClassification,
+    MailDirection,
+    MailMessage,
+    Proposal,
+    audit,
+)
+
+# --- IMAP ---------------------------------------------------------------------
+
+
+def connect_imap(cfg: Config) -> imaplib.IMAP4_SSL:
+    client = imaplib.IMAP4_SSL(cfg.imap_host, cfg.imap_port)
+    client.login(cfg.imap_user, cfg.imap_password)
+    client.select(cfg.imap_mailbox)
+    return client
+
+
+def _fetch_unread(cfg: Config) -> list[bytes]:
+    """Fetch raw unread messages. Returns list of raw RFC822 bytes."""
+    client = connect_imap(cfg)
+    try:
+        _, data = client.search(None, "UNSEEN")
+        raw_messages: list[bytes] = []
+        for num in data[0].split():
+            _, msg_data = client.fetch(num, "(RFC822)")
+            raw_messages.append(msg_data[0][1])
+            client.store(num, "+FLAGS", "\\Seen")
+        return raw_messages
+    finally:
+        client.logout()
+
+
+# --- parsing ------------------------------------------------------------------
+
+
+def parse_raw(raw: bytes) -> dict:
+    """Parse raw RFC822 bytes into our flat dict shape."""
+    msg = email.message_from_bytes(raw, policy=email.policy.default)
+    body = ""
+    if msg.is_multipart():
+        for part in msg.walk():
+            if part.get_content_type() == "text/plain" and not part.get_filename():
+                body = part.get_content()
+                break
+    else:
+        body = msg.get_content()
+    reply_to = msg.get("In-Reply-To") or msg.get("References") or ""
+    message_id = (msg.get("Message-ID") or "").strip()
+    if not message_id:
+        message_id = f"<generated-{datetime.now(UTC).timestamp()}@localhost>"
+    # store the bare address (envelope-safe); display names stay out of the DB
+    from_addr = email.utils.parseaddr(msg.get("From") or "")[1].strip()[:320]
+    return {
+        "message_id": message_id[:998],
+        "in_reply_to": reply_to.split()[-1].strip() if reply_to.strip() else None,
+        "from_addr": from_addr,
+        "subject": (msg.get("Subject") or "").strip()[:998],
+        "body": body[:20000],
+    }
+
+
+# --- proposal threading -------------------------------------------------------
+
+
+def _match_proposal(parsed: dict, sess) -> Proposal | None:
+    """Thread an inbound mail to its proposal via In-Reply-To of sent drafts."""
+    irt = parsed.get("in_reply_to")
+    if not irt:
+        return None
+    sent = (
+        sess.query(MailMessage)
+        .filter(
+            MailMessage.direction == MailDirection.OUTBOUND,
+            MailMessage.message_id == irt,
+        )
+        .one_or_none()
+    )
+    if sent is None or sent.proposal_id is None:
+        return None
+    return sess.get(Proposal, sent.proposal_id)
+
+
+# --- drafts -------------------------------------------------------------------
+
+_DRAFT_TEMPLATES = {
+    MailClassification.APPROVAL: (
+        "Hello,\n\nThank you for the approval regarding \"{title}\". "
+        "We will proceed on our side and follow up with the next steps.\n\nBest regards,\nCayce"
+    ),
+    MailClassification.REJECTION: (
+        "Hello,\n\nThank you for considering \"{title}\". "
+        "We appreciate the reply and would welcome the chance to revisit in the future.\n\n"
+        "Best regards,\nCayce"
+    ),
+    MailClassification.QUESTION: (
+        "Hello,\n\nRegarding \"{title}\": {question}\n\n"
+        "Could you help us with the details so we can move forward?\n\nBest regards,\nCayce"
+    ),
+}
+
+
+def _build_draft(parsed: dict, kind: MailClassification, proposal: Proposal | None) -> str:
+    title = proposal.title if proposal else (parsed.get("subject") or "your message")
+    template = _DRAFT_TEMPLATES.get(kind)
+    if template is None:
+        return ""
+    question = (parsed.get("body") or "")[:200].strip() or "we had a question"
+    return template.format(title=title[:120], question=question)
+
+
+# --- inbound pass ---------------------------------------------------------------
+
+
+def _poll(cfg: Config, sess, fetcher=None) -> dict:
+    """One inbound pass: parse, classify, thread, draft. Returns stats."""
+    stats = {"processed": 0, "drafted": 0, "needs_review": 0, "skipped": 0}
+    for raw in (fetcher or _fetch_unread)(cfg):
+        parsed = parse_raw(raw)
+        if (
+            sess.query(MailMessage)
+            .filter(MailMessage.message_id == parsed["message_id"])
+            .one_or_none()
+        ):
+            stats["skipped"] += 1
+            continue
+
+        proposal = _match_proposal(parsed, sess)
+        kind, confidence = _classify_text(f"{parsed['subject']}\n{parsed['body']}", cfg)
+        if proposal is None and kind == MailClassification.APPROVAL:
+            # an approval we cannot thread is suspicious — human must look
+            kind = MailClassification.NEEDS_REVIEW
+            confidence = min(confidence, 0.4)
+
+        record = MailMessage(
+            direction=MailDirection.INBOUND,
+            message_id=parsed["message_id"],
+            in_reply_to=parsed["in_reply_to"],
+            from_addr=parsed["from_addr"] or None,
+            subject=parsed["subject"] or None,
+            body=parsed["body"],
+            classification=kind,
+            confidence=confidence,
+            proposal_id=proposal.id if proposal else None,
+            processed_at=datetime.now(UTC),
+        )
+        sess.add(record)
+        sess.flush()
+        audit(
+            sess,
+            actor="daemon:maild",
+            action="classify",
+            proposal_id=proposal.id if proposal else None,
+            detail={"classification": kind.value, "confidence": confidence},
+        )
+        stats["processed"] += 1
+
+        if kind == MailClassification.NEEDS_REVIEW:
+            stats["needs_review"] += 1
+            continue
+
+        if proposal is not None:
+            proposal.detail = {
+                **(proposal.detail or {}),
+                "merchant_status": kind.value,
+                "merchant_reply_at": datetime.now(UTC).isoformat(),
+            }
+
+        draft = _build_draft(parsed, kind, proposal)
+        if draft:
+            record.draft_reply = draft
+            record.draft_status = DraftStatus.DRAFTED
+            audit(
+                sess,
+                actor="daemon:maild",
+                action="draft_reply",
+                proposal_id=proposal.id if proposal else None,
+                detail={"mail_id": record.id},
+            )
+            stats["drafted"] += 1
+    return stats
+
+
+# --- human gate: approve draft -------------------------------------------------
+
+
+def _approve_draft(mail_id: int, actor: str, sess) -> int:
+    """Human approves a drafted reply. Only then may SMTP send it."""
+    record = sess.get(MailMessage, mail_id)
+    if record is None or record.direction != MailDirection.INBOUND:
+        return 1
+    if record.draft_status != DraftStatus.DRAFTED:
+        audit(
+            sess,
+            actor=actor,
+            action="approve_draft",
+            proposal_id=record.proposal_id,
+            result="error",
+            detail={"reason": f"invalid draft state: {record.draft_status.value}"},
+        )
+        return 1
+    record.draft_status = DraftStatus.APPROVED
+    audit(
+        sess,
+        actor=actor,
+        action="approve_draft",
+        proposal_id=record.proposal_id,
+        detail={"mail_id": record.id},
+    )
+    return 0
+
+
+# --- SMTP: only approved drafts ------------------------------------------------
+
+
+def _smtp_send(cfg: Config, to_addr: str, subject: str, body: str, message_id: str) -> None:
+    msg = EmailMessage()
+    msg["From"] = cfg.mail_from
+    msg["To"] = to_addr
+    msg["Subject"] = subject
+    msg["Message-ID"] = message_id
+    msg.set_content(body)
+    with smtplib.SMTP(cfg.smtp_host, cfg.smtp_port, timeout=30) as server:
+        server.starttls()
+        server.login(cfg.smtp_user, cfg.smtp_password)
+        server.send_message(msg)
+
+
+def _approved_by(sess, mail_id: int) -> str:
+    from cayceops.models import AuditEntry
+
+    for entry in sess.query(AuditEntry).all():
+        if entry.action == "approve_draft" and entry.detail.get("mail_id") == mail_id:
+            return entry.actor
+    return "unknown"
+
+
+def _send_approved(cfg: Config, sess, sender=None) -> dict:
+    """Send every APPROVED draft. Anything else stays untouched."""
+    stats = {"sent": 0, "blocked": 0}
+    pending = (
+        sess.query(MailMessage)
+        .filter(
+            MailMessage.direction == MailDirection.INBOUND,
+            MailMessage.draft_status == DraftStatus.APPROVED,
+        )
+        .all()
+    )
+    for record in pending:
+        to_addr = record.from_addr or ""
+        if not to_addr:
+            record.draft_status = DraftStatus.SKIPPED
+            stats["blocked"] += 1
+            continue
+        message_id = (
+            f"<cayce-draft-{record.id}-{int(datetime.now(UTC).timestamp())}@cayce-co.xyz>"
+        )
+        (sender or _smtp_send)(
+            cfg,
+            to_addr,
+            f"Re: {record.subject or 'your message'}",
+            record.draft_reply or "",
+            message_id,
+        )
+        approved_by = _approved_by(sess, record.id)
+        record.draft_status = DraftStatus.SENT
+        # store the Message-ID so future replies thread to this draft
+        record.message_id = message_id
+        audit(
+            sess,
+            actor="daemon:maild",
+            action="send_draft",
+            proposal_id=record.proposal_id,
+            detail={
+                "mail_id": record.id,
+                "message_id": message_id,
+                "approved_by": approved_by,
+            },
+        )
+        stats["sent"] += 1
+    return stats
+
+
+# --- CLI wrappers (session scope lives here) ------------------------------------
+
+
+def poll(cfg: Config | None = None) -> int:
+    from cayceops.config import load
+    from cayceops.db import engine
+
+    cfg = cfg or load()
+    try:
+        with session_scope() as sess:
+            run_row = DaemonRun(daemon="maild")
+            sess.add(run_row)
+            sess.flush()
+            try:
+                stats = _poll(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 = stats
+            run_row.finished_at = datetime.now(UTC)
+        return 0
+    finally:
+        engine().dispose()
+
+
+def approve_draft(mail_id: int, actor: str) -> int:
+    from cayceops.db import engine
+
+    try:
+        with session_scope() as sess:
+            return _approve_draft(mail_id, actor, sess)
+    finally:
+        engine().dispose()
+
+
+def send_approved(cfg: Config | None = None) -> int:
+    from cayceops.config import load
+    from cayceops.db import engine
+
+    cfg = cfg or load()
+    try:
+        with session_scope() as sess:
+            _send_approved(cfg, sess)
+        return 0
+    finally:
+        engine().dispose()

cayceops/components/monitor/__init__.py

@@ -0,0 +1 @@
+"""monitor package."""

cayceops/components/monitor/monitor.py

@@ -0,0 +1,160 @@
+"""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(f"<tr><td>{name}</td><td class=muted>never run</td><td>—</td></tr>")
+        else:
+            cls = "ok" if last.status == "ok" else "error"
+            rows.append(
+                f"<tr><td>{html.escape(name)}</td>"
+                f"<td class={cls}>{html.escape(last.status)}</td>"
+                f"<td>{_fmt_ts(last.finished_at or last.started_at)}</td></tr>"
+            )
+    return "<table><tr><th>daemon</th><th>last status</th><th>last run</th></tr>" + "".join(rows) + "</table>"
+
+
+def _queue_cards(sess) -> str:
+    counts = dict(sess.query(Proposal.status, func.count(Proposal.id)).group_by(Proposal.status).all())
+    cards = [
+        f'<div class=card>{html.escape(status.value)}: <strong>{counts.get(status, 0)}</strong></div>'
+        for status in ProposalStatus
+    ]
+    return "".join(cards)
+
+
+def _recent_audit(sess, limit: int = 20) -> str:
+    rows = []
+    for entry in sess.query(AuditEntry).order_by(AuditEntry.ts.desc()).limit(limit).all():
+        cls = "ok" if entry.result == "ok" else "error"
+        rows.append(
+            f"<tr><td>{_fmt_ts(entry.ts)}</td><td>{html.escape(entry.actor)}</td>"
+            f"<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()

cayceops/components/publisher/__init__.py

@@ -0,0 +1 @@
+"""publisher package."""

cayceops/components/publisher/adapters.py

@@ -0,0 +1,61 @@
+"""Optional Facebook/Instagram adapters (Graph API).
+
+DISABLED BY DEFAULT. Each adapter refuses to run unless its enable flag is
+true (``CAYCE_OPS_ENABLE_FACEBOOK`` / ``CAYCE_OPS_ENABLE_INSTAGRAM``), so the
+default deployment never touches Meta endpoints. Tokens are env-only and are
+never logged.
+"""
+
+from __future__ import annotations
+
+import json
+import urllib.parse
+import urllib.request
+
+GRAPH = "https://graph.facebook.com/v21.0"
+
+
+def _post(url: str, payload: dict) -> dict:
+    data = urllib.parse.urlencode(payload).encode()
+    req = urllib.request.Request(url, data=data, method="POST")  # noqa: S310
+    with urllib.request.urlopen(req, timeout=30) as resp:
+        return json.loads(resp.read().decode())
+
+
+def post_facebook(cfg, caption: str, media_path: str | None = None) -> dict:
+    """Post to the page feed (text) or /photos (with local media). Gated."""
+    if not cfg.enable_facebook:
+        raise PermissionError("facebook adapter disabled (CAYCE_OPS_ENABLE_FACEBOOK=false)")
+    if not cfg.facebook_page_token:
+        raise ValueError("CAYCE_OPS_FACEBOOK_PAGE_TOKEN missing")
+    if media_path:
+        # photo posts require a reachable URL or multipart upload; we use the
+        # simple URL form — media must be hosted (out of scope for local files)
+        result = _post(
+            f"{GRAPH}/me/photos",
+            {"url": media_path, "caption": caption, "access_token": cfg.facebook_page_token},
+        )
+    else:
+        result = _post(
+            f"{GRAPH}/me/feed",
+            {"message": caption, "access_token": cfg.facebook_page_token},
+        )
+    return {"post_id": result.get("post_id") or result.get("id", "")}
+
+
+def post_instagram(cfg, caption: str, image_url: str) -> dict:
+    """Two-step IG publish (container -> publish). Gated."""
+    if not cfg.enable_instagram:
+        raise PermissionError("instagram adapter disabled (CAYCE_OPS_ENABLE_INSTAGRAM=false)")
+    if not cfg.instagram_user_id:
+        raise ValueError("CAYCE_OPS_INSTAGRAM_USER_ID missing")
+    token = cfg.facebook_page_token  # IG Graph publishing uses the page token
+    container = _post(
+        f"{GRAPH}/{cfg.instagram_user_id}/media",
+        {"image_url": image_url, "caption": caption, "access_token": token},
+    )
+    published = _post(
+        f"{GRAPH}/{cfg.instagram_user_id}/media_publish",
+        {"creation_id": container["id"], "access_token": token},
+    )
+    return {"post_id": published.get("id", "")}

cayceops/components/publisher/images.py

@@ -0,0 +1,72 @@
+"""Brand image templates (Pillow). 1080x1350 portrait, two brand styles.
+
+Deterministic, template-driven graphics — deliberately not AI-generated media
+(DECISIONS: nothing here trips AI-media disclosure requirements; it is typeset
+design). Fonts: Pillow's bundled default (scalable), so no system font deps.
+"""
+
+from __future__ import annotations
+
+import os
+
+from PIL import Image, ImageDraw, ImageFont
+
+WIDTH, HEIGHT = 1080, 1350
+
+_BRANDS = {
+    # bitcoinhotels — warm gold on near-black
+    "bh": {"bg": (20, 18, 16), "accent": (201, 168, 76), "text": (242, 236, 224), "muted": (150, 140, 120)},
+    # hardcap — bitcoin orange on cool black
+    "hc": {"bg": (13, 13, 16), "accent": (247, 147, 26), "text": (250, 250, 250), "muted": (120, 120, 130)},
+}
+
+
+def _font(size: int, bold: bool = False):
+    # Pillow >=10: load_default(size=...) returns a scalable bundled font
+    return ImageFont.load_default(size=size)
+
+
+def _vgrad(top: tuple, bottom: tuple) -> Image.Image:
+    """Simple two-color vertical gradient."""
+    img = Image.new("RGB", (WIDTH, HEIGHT))
+    for y in range(HEIGHT):
+        t = y / (HEIGHT - 1)
+        row = tuple(int(top[i] + (bottom[i] - top[i]) * t) for i in range(3))
+        ImageDraw.Draw(img).line([(0, y), (WIDTH, y)], fill=row)
+    return img
+
+
+def render_brand_image(
+    *,
+    brand: str = "bh",
+    kicker: str | None,
+    lines: list[str],
+    tagline: list[str] | None = None,
+    url: str | None = None,
+    out_path: str,
+) -> str:
+    """Render one branded poster. Returns out_path. Raises on unknown brand."""
+    if brand not in _BRANDS:
+        raise ValueError(f"unknown brand {brand!r} (have: {sorted(_BRANDS)})")
+    palette = _BRANDS[brand]
+    img = _vgrad(palette["bg"], tuple(max(0, c - 8) for c in palette["bg"]))
+    draw = ImageDraw.Draw(img)
+
+    x, y = 90, 240
+    if kicker:
+        draw.text((x, y), kicker, font=_font(40), fill=palette["muted"])
+        y += 110
+    for line in lines:
+        draw.text((x, y), line, font=_font(96, bold=True), fill=palette["text"])
+        y += 118
+    y += 24
+    for t in tagline or []:
+        draw.text((x, y), t, font=_font(38), fill=palette["muted"])
+        y += 58
+    draw.rectangle([x, y + 30, x + 470, y + 33], fill=palette["accent"])
+    if url:
+        draw.text((x, y + 110), url, font=_font(44), fill=palette["accent"])
+
+    os.makedirs(os.path.dirname(out_path) or ".", exist_ok=True)
+    img.save(out_path, "JPEG", quality=92)
+    return out_path

cayceops/components/publisher/nostr.py

@@ -0,0 +1,166 @@
+"""Minimal Nostr client: NIP-01 kind-1 notes (build, sign, verify, publish).
+
+Implemented in-repo (DECISIONS D4) so every byte of the signing path is
+auditable: event id = sha256 of the canonical NIP-01 serialization; signature =
+BIP-340 Schnorr over the id. Keys are 32-byte hex. Relays are plain env config.
+
+Publishing is fire-and-wait: open a websocket per relay, send the EVENT, wait
+for the OK frame (or timeout), and return the per-relay results.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+from typing import Any
+
+from coincurve import PrivateKey, PublicKey
+
+KIND_NOTE = 1
+
+
+def normalize_key(private_key_hex: str) -> bytes:
+    """Accept 64-char hex (with or without 0x). Anything else is a config error."""
+    key = private_key_hex.strip().lower().removeprefix("0x")
+    if len(key) != 64 or not all(c in "0123456789abcdef" for c in key):
+        raise ValueError("CAYCE_OPS_NOSTR_PRIVATE_KEY must be 64-char hex (32 bytes)")
+    return bytes.fromhex(key)
+
+
+def pubkey_from_secret(secret: bytes) -> str:
+    return PublicKey.from_secret(secret).format().hex()[2:]
+
+
+def build_event(
+    secret: bytes,
+    content: str,
+    tags: list[list[str]] | None = None,
+    created_at: int | None = None,
+    kind: int = KIND_NOTE,
+) -> dict[str, Any]:
+    """Build and sign a NIP-01 event. Deterministic given the same inputs."""
+    created_at = int(created_at if created_at is not None else __import__("time").time())
+    pubkey = pubkey_from_secret(secret)
+    serialized = json.dumps(
+        [0, pubkey, created_at, kind, tags or [], content],
+        separators=(",", ":"),  # canonical NIP-01 form
+        ensure_ascii=False,
+    )
+    event_id = hashlib.sha256(serialized.encode("utf-8")).hexdigest()
+    # zero aux randomness -> fully deterministic signatures (BIP-340 permits it;
+    # verification is unaffected either way)
+    signature = (
+        PrivateKey(secret)
+        .sign_schnorr(bytes.fromhex(event_id), aux_randomness=b"\x00" * 32)
+        .hex()
+    )
+    return {
+        "pubkey": pubkey,
+        "created_at": created_at,
+        "kind": kind,
+        "tags": tags or [],
+        "content": content,
+        "id": event_id,
+        "sig": signature,
+    }
+
+
+def _schnorr_verify(sig32: bytes, msg32: bytes, xonly_pub32: bytes) -> bool:
+    """BIP-340 reference verification (pure Python, public domain spec code).
+
+    Signing runs in coincurve (libsecp256k1); verification is pure Python so
+    the auditor can read every step without CFFI indirection.
+    """
+    P = 2**256 - 2**32 - 977
+    N = 0xFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFEBAAEDCE6AF48A03BBFD25E8CD0364141
+
+    def tag_hash(tag: str, msg: bytes) -> bytes:
+        t = hashlib.sha256(tag.encode()).digest()
+        return hashlib.sha256(t + t + msg).digest()
+
+    def lift_x(b: int):
+        if b >= P:
+            return None
+        y_sq = (pow(b, 3, P) + 7) % P
+        y = pow(y_sq, (P + 1) // 4, P)
+        if pow(y, 2, P) != y_sq:
+            return None
+        return (b, y if y % 2 == 0 else P - y)
+
+    def point_add(p1, p2):
+        if p1 is None:
+            return p2
+        if p2 is None:
+            return p1
+        if p1[0] == p2[0] and p1[1] != p2[1]:
+            return None
+        if p1 == p2:
+            lam = (3 * p1[0] * p1[0] * pow(2 * p1[1], P - 2, P)) % P
+        else:
+            lam = ((p2[1] - p1[1]) * pow(p2[0] - p1[0], P - 2, P)) % P
+        x3 = (lam * lam - p1[0] - p2[0]) % P
+        return (x3, (lam * (p1[0] - x3) - p1[1]) % P)
+
+    def point_mul(p, scalar):
+        acc = None
+        for i in range(256):
+            if (scalar >> i) & 1:
+                acc = point_add(acc, p)
+            p = point_add(p, p)
+        return acc
+
+    if len(sig32) != 64 or len(msg32) != 32 or len(xonly_pub32) != 32:
+        return False
+    r = int.from_bytes(sig32[:32], "big")
+    s = int.from_bytes(sig32[32:], "big")
+    if not (0 <= r < P and 0 <= s < N):
+        return False
+    pub_point = lift_x(int.from_bytes(xonly_pub32, "big"))
+    if pub_point is None:
+        return False
+    e = int.from_bytes(tag_hash("BIP0340/challenge", sig32[:32] + xonly_pub32 + msg32), "big") % N
+    r_point = point_add(point_mul((Gx := 0x79BE667EF9DCBBAC55A06295CE870B07029BFCDB2DCE28D959F2815B16F81798, Gy := 0x483ADA7726A3C4655DA4FBFC0E1108A8FD17B448A68554199C47D08FFB10D4B8), s), point_mul(pub_point, (N - e) % N))
+    if r_point is None or r_point[1] % 2 != 0:
+        return False
+    return r_point[0] == r
+
+
+def verify_event(event: dict[str, Any]) -> bool:
+    """Recompute the id and verify the Schnorr signature against the pubkey."""
+    serialized = json.dumps(
+        [0, event["pubkey"], event["created_at"], event["kind"], event["tags"], event["content"]],
+        separators=(",", ":"),
+        ensure_ascii=False,
+    )
+    if hashlib.sha256(serialized.encode("utf-8")).hexdigest() != event["id"]:
+        return False
+    compressed = PublicKey(bytes.fromhex("02" + event["pubkey"])).format()  # 33-byte, prefix 02
+    return _schnorr_verify(
+        bytes.fromhex(event["sig"]), bytes.fromhex(event["id"]), compressed[1:]
+    )
+
+
+def relay_publish(relay_url: str, event: dict[str, Any], timeout: float = 10.0) -> str:
+    """Publish one event to one relay; return the relay's OK message."""
+    from websockets.sync.client import connect
+
+    with connect(relay_url, close_timeout=timeout) as ws:
+        ws.send(json.dumps(["EVENT", event]))
+        reply = json.loads(ws.recv(timeout=timeout))
+    if not (isinstance(reply, list) and len(reply) >= 3 and reply[0] == "OK" and reply[2] is True):
+        raise RuntimeError(f"relay {relay_url} rejected event: {str(reply)[:200]}")
+    return str(reply[1])
+
+
+def publish(relays: list[str], event: dict[str, Any], timeout: float = 10.0) -> list[dict[str, str]]:
+    """Publish to every relay; report per-relay outcome (never raises unless all fail)."""
+    results = []
+    for relay in relays:
+        try:
+            relay_publish(relay, event, timeout)
+            results.append({"relay": relay, "ok": "true"})
+        except Exception as exc:  # noqa: BLE001 — per-relay isolation
+            results.append({"relay": relay, "ok": "false", "error": str(exc)[:200]})
+    if relays and all(r["ok"] == "false" for r in results):
+        raise RuntimeError(f"all relays failed: {results}")
+    return results

cayceops/components/publisher/publisher.py

@@ -0,0 +1,266 @@
+"""publisher — themed content calendar with a hard human gate on publishing.
+
+State machine: draft -> approved -> published. Only ``approved`` items ever
+publish. AUTONOMOUS_MODE has NO effect on this component: content always
+requires explicit human approval (documented in AUDIT.md).
+
+Channels: nostr (NIP-01, primary) — facebook/instagram adapters exist but are
+disabled unless explicitly enabled via env (DECISIONS: default deployment never
+touches Meta).
+"""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime, timedelta
+
+from cayceops.components.publisher import images, nostr
+from cayceops.config import Config
+from cayceops.db import session_scope
+from cayceops.models import ContentItem, ContentStatus, DaemonRun, audit
+
+# rotation: one theme per slot; channel rotation nostr -> facebook -> instagram
+THEMES = [
+    ("lightning-abroad", "bitcoinhotels.org/guides/lightning-abroad"),
+    ("seed-strategy", "hardcap.news/learn/travel-security-kit"),
+    ("atm-fees", "bitcoinhotels.org/guides/best-bitcoin-atms-by-city"),
+    ("elsalvador-budget", "bitcoinhotels.org/guides/living-on-sats-in-el-salvador"),
+]
+
+_CAPTION_TEMPLATE = "{hook}\n\n{body}\n\n{link}"
+
+
+def _caption(theme: str, link: str) -> str:
+    hooks = {
+        "lightning-abroad": "Paying abroad with Lightning beats your bank.",
+        "seed-strategy": "The seed strategy you pick decides how the border story goes.",
+        "atm-fees": "The first ATM you find charges 10-15% over spot.",
+        "elsalvador-budget": "A real weekly budget on a Bitcoin standard.",
+    }
+    bodies = {
+        "lightning-abroad": "No foreign transaction fees, no dynamic currency conversion, no declines. Field-tested guide.",
+        "seed-strategy": "Steel plate, memorized + decoy, or split custody — the full breakdown, brutally honest.",
+        "atm-fees": "The prepared traveler pays 4-6%. The difference is knowing the operator and the no-KYC cap.",
+        "elsalvador-budget": "Surf hostels, comidas, coworking — the corners where you still need dollars, and the 90-day protocol.",
+    }
+    return _CAPTION_TEMPLATE.format(
+        hook=hooks.get(theme, theme.replace("-", " ").title()),
+        body=bodies.get(theme, "Field-tested notes."),
+        link=link,
+    )
+
+
+# --- calendar planning ----------------------------------------------------------
+
+
+def _plan(cfg: Config, sess, days: int = 4) -> dict:
+    """Fill the calendar: one draft per theme/channel pair not already queued.
+
+    Idempotent by (theme, channel): existing draft/approved items block a
+    duplicate. Drafts are scheduled one per day, starting tomorrow.
+    """
+    created = 0
+    for idx, (theme, link) in enumerate(THEMES[:days]):
+        channel = ["nostr", "facebook", "instagram", "nostr"][idx % 4]
+        exists = (
+            sess.query(ContentItem)
+            .filter(
+                ContentItem.theme == theme,
+                ContentItem.channel == channel,
+                ContentItem.status != ContentStatus.PUBLISHED,
+            )
+            .one_or_none()
+        )
+        if exists:
+            continue
+        item = ContentItem(
+            theme=theme,
+            channel=channel,
+            caption=_caption(theme, link),
+            link=link,
+            status=ContentStatus.DRAFT,
+            scheduled_for=datetime.now(UTC).replace(hour=13, minute=0, second=0, microsecond=0)
+            + timedelta(days=idx + 1),
+        )
+        sess.add(item)
+        sess.flush()
+        audit(sess, actor="daemon:publisher", action="plan_draft", detail={"item": item.id, "theme": theme, "channel": channel})
+        created += 1
+    return {"created": created}
+
+
+# --- human gate -------------------------------------------------------------------
+
+
+def _approve(item_id: int, actor: str, sess) -> int:
+    item = sess.get(ContentItem, item_id)
+    if item is None or item.status != ContentStatus.DRAFT:
+        audit(
+            sess,
+            actor=actor,
+            action="approve_content",
+            result="error",
+            detail={"reason": f"item {item_id} not in draft state"},
+        )
+        return 1
+    item.status = ContentStatus.APPROVED
+    audit(sess, actor=actor, action="approve_content", detail={"item": item.id, "channel": item.channel})
+    return 0
+
+
+# --- image rendering ----------------------------------------------------------------
+
+
+def _render_image(item: ContentItem, cfg: Config) -> str | None:
+    """Render the brand image for a nostr item (per its theme)."""
+    brand = "hc" if "hardcap" in (item.link or "") else "bh"
+    title = item.theme.replace("-", " ").upper().split()
+    lines = [" ".join(title[:2])] if title else [item.theme]
+    out = images.render_brand_image(
+        brand=brand,
+        kicker="C A Y C E",
+        lines=lines,
+        tagline=[(item.link or "")[:70]],
+        url=(item.link or "")[:60],
+        out_path=f"{cfg.media_dir}/item-{item.id}.jpg",
+    )
+    item.media_path = out
+    return out
+
+
+# --- publishing ----------------------------------------------------------------------
+
+
+def _publish_nostr(item: ContentItem, cfg: Config, sess, sender=None) -> str:
+    if not cfg.nostr_private_key or not cfg.nostr_relays:
+        raise RuntimeError("nostr channel requires CAYCE_OPS_NOSTR_PRIVATE_KEY and CAYCE_OPS_NOSTR_RELAYS")
+    secret = nostr.normalize_key(cfg.nostr_private_key)
+    event = nostr.build_event(secret, item.caption, tags=[["t", item.theme]])
+    if not nostr.verify_event(event):  # pragma: no cover — build/verify symmetry
+        raise RuntimeError("event failed self-verification before publish")
+    results = (sender or nostr.publish)(cfg.nostr_relays, event)
+    audit(
+        sess,
+        actor="daemon:publisher",
+        action="nostr_relay_results",
+        detail={"event": event["id"], "relays": results},
+    )
+    return event["id"]
+
+
+def _publish_due(cfg: Config, sess, nostr_sender=None) -> dict:
+    """Publish approved items that are due. Non-nostr channels are skipped
+    unless their adapters are explicitly enabled (default: skipped, audited)."""
+    stats = {"published": 0, "skipped": 0, "failed": 0}
+    due = (
+        sess.query(ContentItem)
+        .filter(ContentItem.status == ContentStatus.APPROVED)
+        .all()
+    )
+    now = datetime.now(UTC)
+    for item in due:
+        if item.scheduled_for and item.scheduled_for > now:
+            continue  # not due yet
+        try:
+            if item.channel == "nostr":
+                item.post_ref = _publish_nostr(item, cfg, sess, sender=nostr_sender)
+            else:
+                audit(
+                    sess,
+                    actor="daemon:publisher",
+                    action="publish_skipped",
+                    detail={
+                        "item": item.id,
+                        "channel": item.channel,
+                        "reason": f"{item.channel} adapter disabled by default",
+                    },
+                )
+                stats["skipped"] += 1
+                continue
+        except Exception as exc:  # noqa: BLE001 — publish boundary: audit and continue
+            item.error = str(exc)[:300]
+            audit(
+                sess,
+                actor="daemon:publisher",
+                action="publish_failed",
+                result="error",
+                detail={"item": item.id, "error": item.error},
+            )
+            stats["failed"] += 1
+            continue
+        item.status = ContentStatus.PUBLISHED
+        item.published_at = datetime.now(UTC)
+        audit(
+            sess,
+            actor="daemon:publisher",
+            action="publish_content",
+            detail={"item": item.id, "channel": item.channel, "ref": item.post_ref},
+        )
+        stats["published"] += 1
+    return stats
+
+
+# --- CLI wrappers ---------------------------------------------------------------------
+
+
+def plan(cfg: Config | None = None) -> int:
+    from cayceops.config import load
+    from cayceops.db import engine
+
+    cfg = cfg or load()
+    try:
+        with session_scope() as sess:
+            run_row = DaemonRun(daemon="publisher")
+            sess.add(run_row)
+            sess.flush()
+            stats = _plan(cfg, sess)
+            run_row.status = "ok"
+            run_row.detail = stats
+            run_row.finished_at = datetime.now(UTC)
+        return 0
+    finally:
+        engine().dispose()
+
+
+def publish_due(cfg: Config | None = None) -> int:
+    from cayceops.config import load
+    from cayceops.db import engine
+
+    cfg = cfg or load()
+    try:
+        with session_scope() as sess:
+            run_row = DaemonRun(daemon="publisher")
+            sess.add(run_row)
+            sess.flush()
+            stats = _publish_due(cfg, sess)
+            run_row.status = "ok"
+            run_row.detail = stats
+            run_row.finished_at = datetime.now(UTC)
+        return 0
+    finally:
+        engine().dispose()
+
+
+def make_image(item_id: int, cfg: Config | None = None) -> int:
+    from cayceops.config import load
+    from cayceops.db import engine
+
+    cfg = cfg or load()
+    try:
+        with session_scope() as sess:
+            item = sess.get(ContentItem, item_id)
+            if item is None:
+                return 1
+            print(_render_image(item, cfg))
+        return 0
+    finally:
+        engine().dispose()
+
+
+def approve(item_id: int, actor: str) -> int:
+    from cayceops.db import engine
+
+    try:
+        with session_scope() as sess:
+            return _approve(item_id, actor, sess)
+    finally:
+        engine().dispose()

cayceops/components/scanner/__init__.py

@@ -0,0 +1 @@
+"""scanner package."""

cayceops/components/scanner/scanner.py

@@ -0,0 +1,209 @@
+"""scanner — opportunity discovery. Proposes; never executes.
+
+Sources come from ``CAYCE_OPS_SCANNER_SOURCES`` (JSON array):
+    [{"type": "rss",  "url": "https://example.org/feed.xml"},
+     {"type": "page", "url": "https://example.org/partners"}]
+
+Every discovered candidate becomes a ``proposals`` row with status ``proposed``
+and a stable ``dedupe_key`` (sha1 of url-or-title) so repeated scans stay
+idempotent. The scanner performs no side effects beyond writing proposals and
+audit entries.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import re
+import urllib.error
+import urllib.request
+from datetime import UTC, datetime
+from html.parser import HTMLParser
+from xml.etree import ElementTree
+
+from cayceops.config import Config
+from cayceops.db import session_scope
+from cayceops.models import DaemonRun, Proposal, ProposalSource, ProposalStatus, audit
+
+# kind classification heuristics (checked in order)
+_KIND_PATTERNS = [
+    ("partnership", re.compile(r"partner|collaborat|integrat", re.I)),
+    ("affiliate", re.compile(r"affiliate|referral|commission|revenue\s*share", re.I)),
+]
+DEFAULT_KIND = "content_gap"
+
+
+class _LinkCollector(HTMLParser):
+    """Minimal HTML parser: page <title> + anchor (href, text) pairs."""
+
+    def __init__(self) -> None:
+        super().__init__()
+        self.title: str = ""
+        self.links: list[tuple[str, str]] = []
+        self._in_title = False
+        self._in_anchor = False
+        self._href = ""
+        self._text: list[str] = []
+
+    def handle_starttag(self, tag: str, attrs: list[tuple[str, str]]) -> None:
+        if tag == "title":
+            self._in_title = True
+        elif tag == "a":
+            self._in_anchor = True
+            self._href = dict(attrs).get("href", "")
+            self._text = []
+
+    def handle_endtag(self, tag: str) -> None:
+        if tag == "title":
+            self._in_title = False
+        elif tag == "a" and self._in_anchor:
+            self._in_anchor = False
+            text = " ".join("".join(self._text).split())
+            if self._href and text:
+                self.links.append((self._href, text))
+
+    def handle_data(self, data: str) -> None:
+        if self._in_title:
+            self.title += data
+        if self._in_anchor:
+            self._text.append(data)
+
+
+def _fetch(url: str, cfg: Config, timeout: float = 20.0) -> bytes:
+    req = urllib.request.Request(
+        url, headers={"User-Agent": cfg.scanner_user_agent}, method="GET"
+    )
+    with urllib.request.urlopen(req, timeout=timeout) as resp:  # noqa: S310
+        return resp.read()
+
+
+def _parse_rss(body: bytes) -> list[tuple[str, str, str]]:
+    """Return (title, link, description) tuples from an RSS/Atom feed."""
+    out: list[tuple[str, str, str]] = []
+    root = ElementTree.fromstring(body)
+    ns = {"atom": "http://www.w3.org/2005/Atom"}
+    for item in root.iter("item"):  # RSS 2.0
+        title = item.findtext("title") or ""
+        link = item.findtext("link") or ""
+        desc = re.sub(r"<[^>]+>", " ", item.findtext("description") or "")
+        out.append((" ".join(title.split()), link, " ".join(desc.split())[:500]))
+    for entry in root.findall("atom:entry", ns):  # Atom
+        title = entry.findtext("atom:title", namespaces=ns) or ""
+        link_el = entry.find("atom:link", ns)
+        link = link_el.get("href", "") if link_el is not None else ""
+        desc = re.sub(r"<[^>]+>", " ", entry.findtext("atom:summary", namespaces=ns) or "")
+        out.append((" ".join(title.split()), link, " ".join(desc.split())[:500]))
+    return out
+
+
+def _parse_page(body: bytes) -> list[tuple[str, str, str]]:
+    """Return (title=anchor text, link=href, description=page title) tuples."""
+    parser = _LinkCollector()
+    parser.feed(body.decode("utf-8", errors="replace"))
+    page_title = " ".join(parser.title.split())
+    return [(text, href, f"linked from page: {page_title}") for href, text in parser.links]
+
+
+def _classify(title: str, description: str) -> str:
+    blob = f"{title}\n{description}"
+    for kind, pattern in _KIND_PATTERNS:
+        if pattern.search(blob):
+            return kind
+    return DEFAULT_KIND
+
+
+def _score(title: str, description: str, keywords: list[str]) -> float:
+    blob = f"{title}\n{description}".lower()
+    hits = sum(1 for kw in keywords if kw.lower() in blob)
+    return round(min(hits / max(1, len(keywords)), 1.0), 3)
+
+
+def scan(cfg: Config, sess, fetch=_fetch) -> dict:
+    """Run one scan pass against all configured sources. Returns run stats."""
+    sources = json.loads(cfg.scanner_sources or "[]")
+    stats = {"sources": len(sources), "candidates": 0, "proposed": 0, "skipped_existing": 0}
+    existing = {
+        key
+        for (key,) in sess.query(Proposal.dedupe_key).filter(
+            Proposal.dedupe_key.isnot(None)
+        )
+    }
+
+    for src in sources:
+        url = src.get("url", "")
+        stype = src.get("type", "page")
+        if not url:
+            continue
+        try:
+            body = fetch(url, cfg)
+        except (urllib.error.URLError, OSError) as exc:
+            stats.setdefault("source_errors", []).append({"url": url, "error": str(exc)[:200]})
+            continue
+        items = _parse_rss(body) if stype == "rss" else _parse_page(body)
+        stats["candidates"] += len(items)
+
+        for title, link, desc in items:
+            text = title or link
+            if not text:
+                continue
+            keywords = cfg.scanner_keywords
+            score = _score(title or link, desc, keywords)
+            if score <= 0 and stype != "page":
+                continue  # feeds must match at least one keyword; pages keep links
+            dedupe = hashlib.sha1((link or title).encode()).hexdigest()[:32]  # noqa: S324
+            if dedupe in existing:
+                stats["skipped_existing"] += 1
+                continue
+            proposal = Proposal(
+                source=ProposalSource.SCANNER,
+                kind=_classify(text, desc),
+                title=text[:512],
+                url=(link or None)[:1024] if (link or None) else None,
+                detail={
+                    "source_type": stype,
+                    "source_url": url,
+                    "summary": desc,
+                    "scanned_at": datetime.now(UTC).isoformat(),
+                },
+                risk_score=score,
+                status=ProposalStatus.PROPOSED,
+                dedupe_key=dedupe,
+            )
+            sess.add(proposal)
+            sess.flush()
+            audit(
+                sess,
+                actor="daemon:scanner",
+                action="propose",
+                proposal_id=proposal.id,
+                detail={"kind": proposal.kind, "score": score, "source": stype},
+            )
+            existing.add(dedupe)
+            stats["proposed"] += 1
+    return stats
+
+
+def run(cfg: Config | None = None) -> int:
+    """CLI entry: one scan pass, wrapped in a DaemonRun record."""
+    from cayceops.config import load
+    from cayceops.db import engine
+
+    cfg = cfg or load()
+    try:
+        with session_scope() as sess:
+            run_row = DaemonRun(daemon="scanner")
+            sess.add(run_row)
+            sess.flush()
+            try:
+                stats = scan(cfg, sess)
+            except Exception as exc:  # noqa: BLE001 — daemon boundary: record and fail cleanly
+                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 = stats
+            run_row.finished_at = datetime.now(UTC)
+        return 0
+    finally:
+        engine().dispose()

tests/test_executor.py

@@ -0,0 +1,133 @@
+"""executor tests — human gate, autonomy bounds, retries, audit chain."""
+
+from __future__ import annotations
+
+import pytest
+
+from cayceops.components.executor import executor, plugins
+from cayceops.models import AuditEntry, Proposal, ProposalStatus
+
+
+def _proposal(kind="content_gap", title="Test item", **kw):
+    return Proposal(kind=kind, title=title, **kw)
+
+
+def test_approve_requires_proposed_state(cfg, sess):
+    p = _proposal()
+    sess.add(p)
+    sess.flush()
+
+    assert executor.approve(p.id, "gustavo", cfg(), sess) == 0
+    # second approve must fail (state is now approved)
+    assert executor.approve(p.id, "gustavo", cfg(), sess) == 1
+    assert p.approved_by == "gustavo"
+    assert p.status == ProposalStatus.APPROVED
+
+
+def test_reject_records_actor(cfg, sess):
+    p = _proposal()
+    sess.add(p)
+    sess.flush()
+    assert executor.reject(p.id, "gustavo", cfg(), sess) == 0
+    assert p.status == ProposalStatus.REJECTED
+
+
+def test_run_executes_only_approved(cfg, sess):
+    approved = _proposal(title="Approved item")
+    proposed = _proposal(title="Proposed item")
+    sess.add_all([approved, proposed])
+    sess.flush()
+    executor.approve(approved.id, "gustavo", cfg(), sess)
+
+    config = cfg()  # AUTONOMOUS_MODE=false
+    assert executor.run(config, sess) == 0
+
+    assert approved.status == ProposalStatus.DONE
+    assert proposed.status == ProposalStatus.PROPOSED  # untouched
+
+    acts = [a.action for a in sess.query(AuditEntry).order_by(AuditEntry.id).all()]
+    assert "execute_start" in acts and "execute_done" in acts
+    assert acts.count("approve") == 1
+
+
+def test_autonomy_off_by_default(cfg, sess):
+    p = _proposal(kind="content_gap", title="Never auto-runs")
+    sess.add(p)
+    sess.flush()
+    config = cfg()  # autonomous_mode False even with kinds configured
+    executor._auto_approve(config, sess)
+    assert p.status == ProposalStatus.PROPOSED
+
+
+def test_autonomy_respects_kinds_and_daily_cap(cfg, sess):
+    p1 = _proposal(kind="content_gap", title="auto ok")
+    p2 = _proposal(kind="content_gap", title="auto over cap")
+    p3 = _proposal(kind="affiliate", title="kind not allowed")
+    sess.add_all([p1, p2, p3])
+    sess.flush()
+
+    config = cfg(
+        CAYCE_OPS_AUTONOMOUS_MODE="true",
+        CAYCE_OPS_AUTONOMOUS_KINDS="content_gap",
+        CAYCE_OPS_AUTONOMOUS_DAILY_CAP="1",
+    )
+    approved = executor._auto_approve(config, sess)
+    assert approved == 1
+    assert p1.status == ProposalStatus.APPROVED
+    assert p2.status == ProposalStatus.PROPOSED  # cap reached
+    assert p3.status == ProposalStatus.PROPOSED  # kind not in policy
+
+    actor = sess.query(AuditEntry).filter_by(action="approve").one().actor
+    assert actor == "auto:policy"
+
+
+def test_failing_plugin_retries_then_fails(cfg, sess, monkeypatch):
+    p = _proposal(kind="affiliate", title="Will fail")
+    sess.add(p)
+    sess.flush()
+    executor.approve(p.id, "gustavo", cfg(), sess)
+
+    calls = {"n": 0}
+
+    def boom(proposal, cfg, sess):
+        calls["n"] += 1
+        raise RuntimeError("upstream exploded")
+
+    monkeypatch.setitem(plugins.REGISTRY, "affiliate", boom)
+    config = cfg(CAYCE_OPS_MAX_ATTEMPTS="2")
+
+    executor.run(config, sess)  # attempt 1 -> back to approved for retry
+    assert p.status == ProposalStatus.APPROVED
+    assert p.attempts == 1
+
+    executor.run(config, sess)  # attempt 2 -> failed for good
+    assert p.status == ProposalStatus.FAILED
+    assert p.attempts == 2
+    assert calls["n"] == 2
+
+    errors = sess.query(AuditEntry).filter_by(result="error").all()
+    assert any(a.action == "fail" for a in errors)
+
+
+def test_http_post_blocked_when_disabled(cfg, sess):
+    p = _proposal(kind="affiliate", title=" webhook target")
+    p.detail = {"webhook_url": "https://hooks.example.org/ingest"}
+    sess.add(p)
+    sess.flush()
+
+    config = cfg()  # CAYCE_OPS_ENABLE_HTTP_ACTIONS=false
+    with pytest.raises(PermissionError):
+        plugins.http_post(p, config, sess)
+
+
+def test_unknown_kind_falls_back_to_log_action(cfg, sess):
+    p = _proposal(kind="brand_new_kind", title="No plugin yet")
+    sess.add(p)
+    sess.flush()
+    executor.approve(p.id, "gustavo", cfg(), sess)
+
+    assert executor.run(cfg(), sess) == 0
+    assert p.status == ProposalStatus.DONE
+
+    done = sess.query(AuditEntry).filter_by(action="execute_done").one()
+    assert done.detail["outcome"]["plugin_fallback"] is True

tests/test_maild.py

@@ -0,0 +1,158 @@
+"""maild tests — parsing, classification, threading, gated send. Invented data only."""
+
+from __future__ import annotations
+
+from cayceops.components.maild import classify as clf
+from cayceops.components.maild import maild
+from cayceops.models import (
+    AuditEntry,
+    DraftStatus,
+    MailClassification,
+    MailDirection,
+    MailMessage,
+    Proposal,
+    ProposalStatus,
+)
+
+OUTBOUND_MSGID = "<cayce-draft-1-1000@cayce-co.xyz>"
+
+
+def _eml(subject: str, body: str, in_reply_to: str | None = None, msgid: str | None = None) -> bytes:
+    lines = [
+        "From: Partner Desk <partners@acme.example.org>",
+        "To: agent@cayce-co.xyz",
+        f"Subject: {subject}",
+        f"Message-ID: {msgid or '<acme-reply-1@acme.example.org>'}",
+    ]
+    if in_reply_to:
+        lines.append(f"In-Reply-To: {in_reply_to}")
+    lines += ["MIME-Version: 1.0", "Content-Type: text/plain; charset=utf-8", "", body]
+    return "\r\n".join(lines).encode()
+
+
+def _proposal_with_sent_draft(sess) -> Proposal:
+    p = Proposal(kind="partnership", title="Acme hotel affiliate program", status=ProposalStatus.APPROVED)
+    sess.add(p)
+    sess.flush()
+    sess.add(
+        MailMessage(
+            direction=MailDirection.OUTBOUND,
+            message_id=OUTBOUND_MSGID,
+            to_addr="partners@acme.example.org",
+            subject="[cayce] partnership inquiry",
+            proposal_id=p.id,
+            draft_status=DraftStatus.SENT,
+        )
+    )
+    sess.flush()
+    return p
+
+
+# --- classifier ---------------------------------------------------------------
+
+
+def test_rules_classification(cfg):
+    config = cfg()
+    assert clf.classify("Your application was approved, welcome to the program", config)[0] is MailClassification.APPROVAL
+    assert clf.classify("Unfortunately we cannot accept new partners", config)[0] is MailClassification.REJECTION
+    assert clf.classify("Could you provide your traffic numbers?", config)[0] is MailClassification.QUESTION
+    kind, confidence = clf.classify("grocery list", config)
+    assert kind is MailClassification.OTHER or kind is MailClassification.NEEDS_REVIEW
+
+
+def test_low_confidence_becomes_needs_review(cfg):
+    config = cfg(CAYCE_OPS_LLM_REVIEW_THRESHOLD="0.95")
+    kind, confidence = clf.classify("Your application was approved, welcome", config)
+    assert kind is MailClassification.NEEDS_REVIEW
+    assert confidence < 0.95
+
+
+# --- inbound pass ---------------------------------------------------------------
+
+
+def test_poll_threads_approval_and_drafts_reply(cfg, sess):
+    p = _proposal_with_sent_draft(sess)
+    raw = _eml("Re: [cayce] partnership inquiry", "Approved! Welcome to the Acme partner program.", in_reply_to=OUTBOUND_MSGID)
+
+    stats = maild._poll(cfg(), sess, fetcher=lambda cfg: [raw])
+    assert stats["processed"] == 1 and stats["drafted"] == 1
+
+    inbound = sess.query(MailMessage).filter_by(direction=MailDirection.INBOUND).one()
+    assert inbound.classification is MailClassification.APPROVAL
+    assert inbound.proposal_id == p.id
+    assert inbound.draft_status is DraftStatus.DRAFTED
+    assert inbound.draft_reply and "approval" in inbound.draft_reply.lower()
+
+    assert p.detail["merchant_status"] == "approval"
+    assert p.status is ProposalStatus.APPROVED  # lifecycle untouched by maild
+
+
+def test_poll_idempotent_on_message_id(cfg, sess):
+    _proposal_with_sent_draft(sess)
+    raw = _eml("Re: hi", "Approved, welcome to the program.", in_reply_to=OUTBOUND_MSGID)
+    fetcher = lambda cfg: [raw]
+    assert maild._poll(cfg(), sess, fetcher=fetcher)["processed"] == 1
+    assert maild._poll(cfg(), sess, fetcher=fetcher)["skipped"] == 1
+
+
+def test_unthreaded_approval_needs_review(cfg, sess):
+    raw = _eml("Great news", "Approved! Welcome aboard.")  # no In-Reply-To
+    stats = maild._poll(cfg(), sess, fetcher=lambda cfg: [raw])
+    inbound = sess.query(MailMessage).filter_by(direction=MailDirection.INBOUND).one()
+    assert inbound.classification is MailClassification.NEEDS_REVIEW
+    assert stats["needs_review"] == 1
+    assert inbound.draft_status is DraftStatus.NONE  # no draft for review items
+
+
+def test_question_gets_clarification_draft(cfg, sess):
+    _proposal_with_sent_draft(sess)
+    raw = _eml("Re: inquiry", "Could you provide your monthly traffic?", in_reply_to=OUTBOUND_MSGID)
+    maild._poll(cfg(), sess, fetcher=lambda cfg: [raw])
+    inbound = sess.query(MailMessage).filter_by(direction=MailDirection.INBOUND).one()
+    assert inbound.classification is MailClassification.QUESTION
+    assert inbound.draft_status is DraftStatus.DRAFTED
+    assert "traffic" in inbound.draft_reply
+
+
+# --- human gate -------------------------------------------------------------------
+
+
+def test_send_blocked_without_approval(cfg, sess, monkeypatch):
+    _proposal_with_sent_draft(sess)
+    raw = _eml("Re: inquiry", "Approved, welcome to the program.", in_reply_to=OUTBOUND_MSGID)
+    maild._poll(cfg(), sess, fetcher=lambda cfg: [raw])
+
+    sent = []
+    config = cfg(CAYCE_OPS_MAIL_FROM="agent@cayce-co.xyz")
+    stats = maild._send_approved(config, sess, sender=lambda *a, **k: sent.append(a))
+    assert stats["sent"] == 0 and sent == []
+    assert sess.query(MailMessage).filter_by(direction=MailDirection.INBOUND).one().draft_status is DraftStatus.DRAFTED
+
+
+def test_send_requires_explicit_human_approval(cfg, sess):
+    _proposal_with_sent_draft(sess)
+    raw = _eml("Re: inquiry", "Approved, welcome to the program.", in_reply_to=OUTBOUND_MSGID)
+    maild._poll(cfg(), sess, fetcher=lambda cfg: [raw])
+    inbound = sess.query(MailMessage).filter_by(direction=MailDirection.INBOUND).one()
+
+    assert maild._approve_draft(inbound.id, "gustavo", sess) == 0
+
+    sent = []
+    config = cfg(CAYCE_OPS_MAIL_FROM="agent@cayce-co.xyz")
+    stats = maild._send_approved(config, sess, sender=lambda *a, **k: sent.append(a))
+    assert stats["sent"] == 1
+    assert inbound.draft_status is DraftStatus.SENT
+    assert sent[0][1] == "partners@acme.example.org"
+
+    row = sess.query(AuditEntry).filter_by(action="send_draft").one()
+    assert "gustavo" in str(row.detail)
+
+
+def test_double_approve_is_rejected(cfg, sess):
+    _proposal_with_sent_draft(sess)
+    raw = _eml("Re: inquiry", "Approved, welcome to the program.", in_reply_to=OUTBOUND_MSGID)
+    maild._poll(cfg(), sess, fetcher=lambda cfg: [raw])
+    inbound = sess.query(MailMessage).filter_by(direction=MailDirection.INBOUND).one()
+
+    assert maild._approve_draft(inbound.id, "gustavo", sess) == 0
+    assert maild._approve_draft(inbound.id, "gustavo", sess) == 1

tests/test_monitor.py

@@ -0,0 +1,69 @@
+"""monitor tests — page renders, no sensitive data, escapes input."""
+
+from __future__ import annotations
+
+from cayceops.components.monitor import monitor
+from cayceops.models import ContentItem, DaemonRun, MailMessage, Proposal
+
+
+def _seed(sess):
+    sess.add(DaemonRun(daemon="scanner", status="ok", detail={"proposed": 3}))
+    sess.add(Proposal(kind="affiliate", title="Title with <script>alert(1)</script>"))
+    sess.add(
+        MailMessage(
+            direction="in",
+            from_addr="secret-person@private.example.org",
+            subject="private subject",
+            classification="approval",
+            draft_status="drafted",
+        )
+    )
+    sess.add(ContentItem(theme="atm-fees", channel="nostr", caption="caption", status="draft"))
+    sess.flush()
+
+
+def test_build_renders_all_sections(cfg, sess, tmp_path):
+    _seed(sess)
+    out = tmp_path / "status.html"
+    monitor._build(cfg(CAYCE_OPS_MONITOR_OUTPUT=str(out)), sess)
+    page = out.read_text()
+
+    assert "daemon health" in page
+    assert "proposal queue" in page
+    assert "recent audit activity" in page
+    assert "scanner" in page
+    assert "proposed: <strong>1</strong>" in page
+    assert "mail drafts awaiting approval: <strong>1</strong>" in page
+
+
+def test_build_leaks_no_sensitive_data(cfg, sess, tmp_path):
+    _seed(sess)
+    out = tmp_path / "status.html"
+    monitor._build(cfg(CAYCE_OPS_MONITOR_OUTPUT=str(out)), sess)
+    page = out.read_text()
+
+    assert "secret-person@private.example.org" not in page  # no addresses
+    assert "private subject" not in page  # no subjects
+    assert "caption" not in page  # no content bodies
+    assert "Title with" not in page  # proposal titles not rendered
+
+
+def test_build_escapes_injected_content(cfg, sess, tmp_path):
+    from datetime import UTC, datetime
+
+    from cayceops.models import AuditEntry
+
+    sess.add(
+        AuditEntry(
+            ts=datetime.now(UTC),
+            actor="<script>x</script>",
+            action="probe",
+            detail={},
+        )
+    )
+    sess.flush()
+    out = tmp_path / "status.html"
+    monitor._build(cfg(CAYCE_OPS_MONITOR_OUTPUT=str(out)), sess)
+    page = out.read_text()
+    assert "<script>x</script>" not in page
+    assert "&lt;script&gt;x&lt;/script&gt;" in page

tests/test_publisher.py

@@ -0,0 +1,162 @@
+"""publisher tests — calendar, human gate, nostr publish, disabled adapters."""
+
+from __future__ import annotations
+
+import os
+
+import pytest
+
+from cayceops.components.publisher import adapters, nostr, publisher
+from cayceops.models import AuditEntry, ContentItem, ContentStatus
+
+# invented test key (never a real nostr key — allowlisted in .gitleaks.toml)
+TEST_KEY = "11" * 32
+
+
+def _item(channel="nostr", theme="atm-fees", scheduled=None, status=ContentStatus.APPROVED):
+    return ContentItem(
+        theme=theme,
+        channel=channel,
+        caption="The first ATM you find charges 10-15% over spot.\n\nbitcoinhotels.org/guides/atm",
+        link="bitcoinhotels.org/guides/atm",
+        status=status,
+        scheduled_for=scheduled,
+    )
+
+
+# --- nostr client -------------------------------------------------------------
+
+
+def test_event_build_verify_roundtrip():
+    event = nostr.build_event(bytes.fromhex(TEST_KEY), "gm", tags=[["t", "bitcoin"]], created_at=1700000000)
+    assert nostr.verify_event(event)
+    assert len(event["id"]) == 64 and len(event["sig"]) == 128
+
+    tampered = {**event, "content": "gm (edited)"}
+    assert not nostr.verify_event(tampered)
+
+
+def test_event_is_deterministic():
+    a = nostr.build_event(bytes.fromhex(TEST_KEY), "gm", created_at=1700000000)
+    b = nostr.build_event(bytes.fromhex(TEST_KEY), "gm", created_at=1700000000)
+    assert a == b
+
+
+def test_key_validation():
+    with pytest.raises(ValueError):
+        nostr.normalize_key("not-a-key")
+    assert nostr.normalize_key("0x" + TEST_KEY) == bytes.fromhex(TEST_KEY)
+
+
+# --- calendar ------------------------------------------------------------------
+
+
+def test_plan_creates_idempotent_drafts(cfg, sess):
+    config = cfg()
+    first = publisher._plan(config, sess)
+    second = publisher._plan(config, sess)
+    assert first["created"] == 4
+    assert second["created"] == 0  # same themes+channels already queued
+
+
+def test_approve_gate(cfg, sess):
+    item = _item(status=ContentStatus.DRAFT)
+    sess.add(item)
+    sess.flush()
+
+    assert publisher._approve(item.id, "gustavo", sess) == 0
+    assert item.status is ContentStatus.APPROVED
+    assert publisher._approve(item.id, "gustavo", sess) == 1  # not draft anymore
+
+    actor = sess.query(AuditEntry).filter_by(action="approve_content").first().actor
+    assert actor == "gustavo"
+
+
+# --- publishing ------------------------------------------------------------------
+
+
+def test_publish_due_nostr_happy_path(cfg, sess):
+    item = _item()  # approved, scheduled None -> due now
+    sess.add(item)
+    sess.flush()
+
+    relays_seen = []
+
+    def fake_sender(relays, event, timeout=10.0):
+        relays_seen.append(relays)
+        assert nostr.verify_event(event)
+        return [{"relay": r, "ok": "true"} for r in relays]
+
+    config = cfg(
+        CAYCE_OPS_NOSTR_PRIVATE_KEY=TEST_KEY,
+        CAYCE_OPS_NOSTR_RELAYS="wss://relay.example.org,wss://other.example.org",
+    )
+    stats = publisher._publish_due(config, sess, nostr_sender=fake_sender)
+
+    assert stats["published"] == 1
+    assert item.status is ContentStatus.PUBLISHED
+    assert item.post_ref and len(item.post_ref) == 64
+    assert item.published_at is not None
+    assert "wss://relay.example.org" in relays_seen[0]
+
+
+def test_future_scheduled_item_not_published(cfg, sess):
+    from datetime import UTC, datetime, timedelta
+
+    item = _item(scheduled=datetime.now(UTC) + timedelta(hours=2))
+    sess.add(item)
+    sess.flush()
+    config = cfg(CAYCE_OPS_NOSTR_PRIVATE_KEY=TEST_KEY, CAYCE_OPS_NOSTR_RELAYS="wss://r.example.org")
+    stats = publisher._publish_due(config, sess, nostr_sender=lambda r, e, timeout=10: [])
+    assert stats["published"] == 0
+    assert item.status is ContentStatus.APPROVED
+
+
+def test_nostr_without_config_fails_clean(cfg, sess):
+    item = _item()
+    sess.add(item)
+    sess.flush()
+    config = cfg()  # no key, no relays
+    stats = publisher._publish_due(config, sess)
+    assert stats["failed"] == 1
+    assert item.status is ContentStatus.APPROVED  # stays approved, error recorded
+    assert item.error and "nostr" in item.error
+
+
+def test_non_nostr_channels_skipped_by_default(cfg, sess):
+    item = _item(channel="facebook", theme="lightning-abroad")
+    sess.add(item)
+    sess.flush()
+    stats = publisher._publish_due(cfg(), sess)
+    assert stats["skipped"] == 1
+    assert item.status is ContentStatus.APPROVED
+    skip = sess.query(AuditEntry).filter_by(action="publish_skipped").one()
+    assert skip.detail["channel"] == "facebook"
+
+
+def test_meta_adapters_disabled_by_default(cfg):
+    config = cfg()
+    with pytest.raises(PermissionError):
+        adapters.post_facebook(config, "hello")
+    with pytest.raises(PermissionError):
+        adapters.post_instagram(config, "hello", "https://img.example.org/x.jpg")
+
+
+# --- images ------------------------------------------------------------------
+
+
+def test_brand_image_render(tmp_path, cfg):
+    config = cfg(CAYCE_OPS_MEDIA_DIR=str(tmp_path))
+    out = publisher._render_image(_item(), config)
+    assert os.path.exists(out)
+    from PIL import Image
+
+    with Image.open(out) as img:
+        assert img.size == (1080, 1350)
+
+
+def test_unknown_brand_rejected():
+    from cayceops.components.publisher import images
+
+    with pytest.raises(ValueError):
+        images.render_brand_image(brand="xx", kicker=None, lines=["A"], out_path="/tmp/x.jpg")

tests/test_scanner.py

@@ -0,0 +1,112 @@
+"""scanner tests — discovery, classification, dedupe. No network (injected fetch)."""
+
+from __future__ import annotations
+
+import json
+
+from cayceops.components.scanner import scanner
+from cayceops.models import Proposal, ProposalStatus
+
+RSS = b"""<?xml version="1.0"?>
+<rss><channel>
+  <item><title>Acme launches affiliate program for hotels</title>
+    <link>https://acme.example.org/affiliates</link>
+    <description>Earn commission on every booking.</description></item>
+  <item><title>Weekly crypto digest</title>
+    <link>https://news.example.org/digest</link>
+    <description>Everything unrelated.</description></item>
+</channel></rss>"""
+
+PAGE = b"""<html><head><title>Partners</title></head><body>
+  <a href="https://pay.example.org/partner">Become a payment partner</a>
+  <a href="https://pay.example.org/docs">API docs</a>
+</body></html>"""
+
+
+def _fake_fetch(responses):
+    def fetch(url, cfg, timeout=20.0):
+        return responses[url]
+
+    return fetch
+
+
+SOURCES = [
+    {"type": "rss", "url": "https://feeds.example.org/news.xml"},
+    {"type": "page", "url": "https://pay.example.org/partners"},
+]
+
+
+def _cfg(cfg, **kw):
+    return cfg(
+        CAYCE_OPS_SCANNER_SOURCES=json.dumps(SOURCES),
+        CAYCE_OPS_SCANNER_KEYWORDS="affiliate,partner,bitcoin,commission",
+        **kw,
+    )
+
+
+def test_scan_files_proposals_with_kind_and_score(cfg, sess):
+    config = _cfg(cfg)
+    stats = scanner.scan(config, sess, fetch=_fake_fetch({
+        "https://feeds.example.org/news.xml": RSS,
+        "https://pay.example.org/partners": PAGE,
+    }))
+
+    assert stats["proposed"] == 3  # affiliate item + both page links (pages keep all)
+    proposals = sess.query(Proposal).all()
+    by_title = {p.title: p for p in proposals}
+
+    affiliate = by_title["Acme launches affiliate program for hotels"]
+    assert affiliate.kind == "affiliate"
+    assert affiliate.status == ProposalStatus.PROPOSED
+    assert affiliate.risk_score > 0
+    assert affiliate.dedupe_key
+
+    partner = by_title["Become a payment partner"]
+    assert partner.kind == "partnership"
+
+
+def test_scan_is_idempotent_via_dedupe(cfg, sess):
+    config = _cfg(cfg)
+    scanner.scan(config, sess, fetch=_fake_fetch({
+        "https://feeds.example.org/news.xml": RSS,
+        "https://pay.example.org/partners": PAGE,
+    }))
+    stats2 = scanner.scan(config, sess, fetch=_fake_fetch({
+        "https://feeds.example.org/news.xml": RSS,
+        "https://pay.example.org/partners": PAGE,
+    }))
+    assert stats2["proposed"] == 0
+    assert stats2["skipped_existing"] == 3
+
+
+def test_scan_filters_keywordless_feed_items(cfg, sess):
+    config = _cfg(cfg)
+    scanner.scan(config, sess, fetch=_fake_fetch({
+        "https://feeds.example.org/news.xml": RSS,
+        "https://pay.example.org/partners": PAGE,
+    }))
+    titles = {p.title for p in sess.query(Proposal).all()}
+    assert "Weekly crypto digest" not in titles  # no keyword match in feed -> dropped
+
+
+def test_scan_records_audit_for_each_proposal(cfg, sess):
+    from cayceops.models import AuditEntry
+
+    config = _cfg(cfg)
+    scanner.scan(config, sess, fetch=_fake_fetch({
+        "https://feeds.example.org/news.xml": RSS,
+        "https://pay.example.org/partners": PAGE,
+    }))
+    acts = [a.action for a in sess.query(AuditEntry).all()]
+    assert acts.count("propose") == 3
+    assert all(a.actor == "daemon:scanner" for a in sess.query(AuditEntry).all())
+
+
+def test_scan_survives_source_error(cfg, sess):
+    def broken_fetch(url, cfg, timeout=20.0):
+        raise OSError("connection refused")
+
+    config = _cfg(cfg)
+    stats = scanner.scan(config, sess, fetch=broken_fetch)
+    assert stats["proposed"] == 0
+    assert len(stats["source_errors"]) == 2