audit: response section — all findings addressed at c9bd441

Critical 1 (approval forgery): audit_log is now hash-chained
(prev_hash/entry_hash, verify_audit_chain); executor approve REQUIRES a
BIP-340 signature over sha256(cayce-ops:approve:<id>) when
CAYCE_OPS_OPERATOR_PUBKEY is set; deploy/roles.sql ships least-privilege
per-daemon DB roles.

Critical 2 (threading dead): sends create real OUTBOUND rows (own
Message-ID, In-Reply-To -> inbound); inbound message_id never overwritten;
maild compose adds the first-contact path; _match_proposal falls back to
inbound chains.

Critical 3 (duplicate sends): per-message transactions in send path —
SMTP failure on message N no longer resurrects 1..N-1; failures audited
(send_failed) and the row stays queued.

Critical 4 (classifier): negated approvals (cannot accept / not approved /
not accepting) classify REJECTION at 0.95 before the bare-keyword rule;
LLM prompt braces escaped (.format no longer KeyError into silent fallback).

Critical 5 (systemd): every daemon has a run subcommand
(maild run = poll+send, publisher run = publish-due, monitor run = build);
deploy README documents the separate daily publisher plan timer.

Mediums: approved_hash binds drafts/captions to the approved text (edited
text refuses to send/publish until re-approved); risk_score renamed
match_score (migration 0002) and auto-approve orders HIGHEST relevance
first; compose binds postgres to 127.0.0.1; IMAP marks \\Seen only after
commit (fetch returns (num, raw), mark post-commit); disabled channels are
terminal (ContentStatus.SKIPPED) and publish failures burn a bounded
attempt budget before abandonment; make hooks installs the pre-push gate.

tests: 12 new — 56 total, all green; ruff 0 findings; gitleaks clean.

.env.example

@@ -21,6 +21,9 @@ CAYCE_OPS_AUTONOMOUS_KINDS=
 CAYCE_OPS_AUTONOMOUS_DAILY_CAP=0
 CAYCE_OPS_MAX_ATTEMPTS=2
 CAYCE_OPS_ENABLE_HTTP_ACTIONS=false
+# when set (64-char hex x-only pubkey), executor approve REQUIRES a BIP-340
+# signature over sha256("cayce-ops:approve:<id>") made by the operator key
+CAYCE_OPS_OPERATOR_PUBKEY=
 
 # --- mail (IMAP read / SMTP send-approved-drafts-only) ---
 CAYCE_OPS_IMAP_HOST=imap.example.org

AUDIT.md

@@ -0,0 +1,177 @@
+# AUDIT — cayce-ops
+
+Audited commit: **d500d88** — merge of PR #7 on `main`.
+
+> Note for the auditor: PRs #1-#6 merged with E501 lint findings present
+> (per-branch gate runs were misread). PR #7 is a lint-only cleanup
+> (line wrapping, F841/F821/E731 fixes, no behavior change) that restores
+> a fully green gate. The honest trail is kept here on purpose.
+> Re-run `make check` at d500d88 to verify: ruff 0 · 44 tests · no leaks.
+Gate that commit passed: `make check` — ruff (0 findings), 44 tests green,
+alembic smoke on sqlite, gitleaks full history (no leaks).
+
+## 1. Component map and data flow
+
+```
+                      CAYCE_OPS_SCANNER_SOURCES (external URLs)
+                                │  read-only fetch
+                                ▼
+   ┌──────────┐  proposals(status=proposed)  ┌──────────┐
+   │ scanner  │─────────────────────────────▶│   DB     │
+   └──────────┘                              │ (shared) │
+                                             │          │        audit_log (every write)
+   ┌──────────┐   approved → running →      │          │◀────── actor / action / detail /
+   │ executor │◀──── done|failed ───────────│          │        result on every transition
+   └──────────┘   plugins by kind            └──────────┘
+                                             │
+   ┌──────────┐  IMAP inbox ───────────────▶│ mail_messages,
+   │ maild    │  drafts (gated)             │ proposals(detail)
+   └──────────┘  SMTP only draft=approved   │
+                                             │
+   ┌──────────┐  content_items ────────────▶│          │──▶ relay (nostr) / Meta (opt-in)
+   │publisher │  draft→approved→published   └──────────┘
+   └──────────┘
+   ┌──────────┐  reads daemon_runs, audit_log, counts
+   │ monitor  │────────────────────────────────────▶ status.html (static)
+   └──────────┘
+```
+
+Shared schema (`cayceops/models.py`, migration `alembic/versions/0001_initial.py`):
+`proposals`, `audit_log`, `mail_messages`, `content_items`, `daemon_runs`.
+
+## 2. Every external call (where, what it sends, what triggers it)
+
+| # | Call | Code | Sends | Trigger |
+|---|------|------|-------|---------|
+| 1 | HTTP GET to scanner sources | `cayceops/components/scanner/scanner.py` → `_fetch` (urllib) | nothing but the request (+ `CAYCE_OPS_SCANNER_USER_AGENT` header) to URLs **configured by the operator** | `cayceops scanner run` (cron/systemd timer) |
+| 2 | IMAP fetch (IMAP4_SSL, UNSEEN) | `cayceops/components/maild/maild.py` → `connect_imap`, `_fetch_unread` | credentials from env; reads inbox | `cayceops maild poll` |
+| 3 | SMTP send | `maild.py` → `_smtp_send` (smtplib, STARTTLS) | a reply **draft a human approved** to the original sender's bare address | `cayceops maild send-approved`, and only rows with `draft_status=approved` |
+| 4 | LLM chat completion (openai-compatible) | `cayceops/llm.py` → `OpenAICompatProvider.complete` (urllib) | the email text (truncated 4000 chars) + system prompt; `Authorization: Bearer <env key>` | `maild poll` **only if** `CAYCE_OPS_LLM_PROVIDER != null` (default is null → never) |
+| 5 | Nostr relay publish (websocket) | `cayceops/components/publisher/nostr.py` → `relay_publish`/`publish` | a signed NIP-01 event whose content is the **human-approved** caption | `cayceops publisher publish-due`, channel=nostr, only if key+relays configured |
+| 6 | Meta Graph API | `cayceops/components/publisher/adapters.py` | caption (+optional hosted image URL) + page token | **refuses to run** unless `CAYCE_OPS_ENABLE_FACEBOOK/INSTAGRAM=true` (default false). Currently unreachable from `publish_due`. |
+
+No other network code exists. Grep guards: `urllib.request`, `imaplib`, `smtplib`,
+`websockets` appear only in the files above (verified by review; CI could add a
+lint rule — see *Known risks*).
+
+## 3. What each component can do without a human / with AUTONOMOUS_MODE=true
+
+| Component | Without any human (default) | With `CAYCE_OPS_AUTONOMOUS_MODE=true` |
+|---|---|---|
+| scanner | fetch sources, file proposals, write audit | identical (scanner never executes) |
+| executor | nothing executes by itself; it acts only on rows a human set to `approved` (via `executor approve --actor …`) | **one added path**: auto-approves `proposed` items whose `kind` ∈ `CAYCE_OPS_AUTONOMOUS_KINDS`, capped at `CAYCE_OPS_AUTONOMOUS_DAILY_CAP`/day, actor `auto:policy`, fully audited. Then executes those. |
+| maild | polls IMAP, classifies, writes drafts. **Never sends.** | identical — autonomy does not touch mail |
+| publisher | plans drafts; publishes nothing | identical — publishing always requires `publisher approve --actor …`; autonomy has no effect |
+| monitor | reads and renders | identical |
+
+Explicitly impossible in the current code: sending an unapproved mail draft,
+publishing unapproved content, executing a non-approved proposal (the single
+exception is the audited `auto:policy` path above), disabling the audit write.
+
+## 4. Environment variables and permissions per component
+
+All config is env-only (`cayceops/config.py`), prefix `CAYCE_OPS_`. Full list
+with placeholders: `.env.example`.
+
+| Component | Requires | Notes |
+|---|---|---|
+| core | `DATABASE_URL` (postgres or sqlite), `LOG_LEVEL`, `STATE_DIR`, `MEDIA_DIR` | DB user needs SELECT/INSERT/UPDATE on the five tables |
+| scanner | `SCANNER_SOURCES` (JSON), `SCANNER_KEYWORDS`, `SCANNER_USER_AGENT` | outbound GET only |
+| executor | `AUTONOMOUS_MODE`, `AUTONOMOUS_KINDS`, `AUTONOMOUS_DAILY_CAP`, `MAX_ATTEMPTS`, `ENABLE_HTTP_ACTIONS` | `http_post` plugin stays inert unless enabled |
+| maild | `IMAP_HOST/PORT/USER/PASSWORD/MAILBOX`, `SMTP_HOST/PORT/USER/PASSWORD`, `MAIL_FROM`, `LLM_PROVIDER/MODEL/BASE_URL/API_KEY/REVIEW_THRESHOLD` | IMAP user is read+mark-seen; SMTP login can send only as `MAIL_FROM` |
+| publisher | `NOSTR_PRIVATE_KEY` (32-byte hex), `NOSTR_RELAYS`, `ENABLE_FACEBOOK`, `FACEBOOK_PAGE_TOKEN`, `ENABLE_INSTAGRAM`, `INSTAGRAM_USER_ID` | nostr key = signing authority; treat as secret |
+| monitor | `MONITOR_OUTPUT` (writable path) | needs read on DB, write on output only |
+
+Filesystem permissions: default deployment is the `cayce` user owning
+`/srv/cayce-ops` with systemd hardening (`NoNewPrivileges`, `ProtectSystem=strict`,
+`ProtectHome=yes`) — see `deploy/systemd/cayce-ops@.service`.
+
+## 5. Dependencies (pinned in `pyproject.toml` + `uv.lock`)
+
+| Package | Version | Why |
+|---|---|---|
+| sqlalchemy | 2.0.44 | ORM + engine for the shared schema |
+| alembic | 1.17.0 | production migrations |
+| typer | 0.20.0 | CLI (all five daemons are CLI commands) |
+| psycopg[binary] | 3.2.13 | Postgres driver (production stack) |
+| pillow | 12.0.0 | brand image templates (publisher) |
+| coincurve | 21.0.0 | libsecp256k1 binding: NIP-01 event signing (publisher) |
+| websockets | 15.0.1 | nostr relay publish (sync client) |
+| pytest | 9.0.2 (dev) | test suite |
+| ruff | 0.15.1 (dev) | lint |
+
+stdlib doing heavy lifting: `imaplib`/`smtplib`/`email` (maild), `urllib` (scanner,
+LLM, Meta adapters), `xml.etree`/`html.parser` (scanner), `hashlib`/`json` (nostr ids).
+
+## 6. Known risks and open items
+
+1. **No live Gitea Actions runner on the current host** (DECISIONS D6). CI files
+   ship for GitHub/Gitea Actions, and the same gate runs locally pre-push
+   (`scripts/pre-push`), but until a runner exists there is no third-party
+   enforcement of the gate on every push.
+2. **`auto:policy` path is real autonomy** — bounded (kind allowlist + daily cap +
+   audit) but it does execute without a human in the loop when enabled. The cap
+   counter is DB-based; wiping `audit_log` resets it.Mitigation: only the
+   operator can set the env flags, and the audit log is append-only by convention.
+3. **Executor plugins are in-process Python** (D9): no sandboxing. A malicious
+   plugin = arbitrary code. Mitigation: plugins are repo code, reviewed in PRs;
+   the registry starts with `log_action` fallbacks only.
+4. **Mail classification is heuristic** (rules; LLM optional). Conservative by
+   design (low confidence → `needs_review`, unthreaded approvals → review), but
+   a sophisticated phishing email could still be misclassified `other`. Human
+   gate on sends bounds the damage to drafting wrong replies (never sending them).
+5. **Nostr relays are plaintext-config endpoints**: a hostile relay sees the
+   note content (public by design) and could replay it; event ids make duplication
+   detectable. No NIP-05, no retries across relay pools yet.
+6. **`detail` JSON columns are schemaless** — writers evolve payloads without
+   migrations. Monitor intentionally renders none of it.
+7. **gitleaks allowlist covers `tests/`** (D7): a real secret pasted under
+   `tests/` would not trip the scanner. Compensating control: PR review; no
+   fixture reads the environment.
+8. **Open:** provision an act_runner for Gitea Actions; wire `posts_log`-style
+   engagement back into publisher planning; add NIP-05 identifier; retries for
+   relay publish.
+
+## 7. Running tests and the local stack
+
+```bash
+uv sync --frozen                 # python 3.12 + locked deps
+uv run pytest                    # 44 tests, sqlite, no network
+make check                       # lint + tests + alembic smoke + gitleaks (pre-push gate)
+
+# full local stack (five daemons + postgres 16):
+cp .env.example .env             # fill placeholders; nothing real required for db-only
+docker compose up -d db
+uv run alembic upgrade head
+docker compose up -d             # or run each: uv run cayceops {scanner,executor,maild,publisher,monitor} …
+
+# production path (VPS):
+#   see deploy/systemd/README.md (hardened units + timers)
+```
+
+## Audited commit
+
+- Repository: `cayce/cayce-ops` (branch `main`)
+- **Audited: d500d888701c3f1cd3305d098a448c3e20a8ad99** (merge of PR #7)
+- Gate at that commit: ruff clean (0 findings) · 44/44 tests · alembic 0001
+  smoke OK · gitleaks `--all` no leaks · `make check` exit 0.
+
+---
+
+## Response to external audit — 2026-10-20
+
+External audit verdict reproduced our gate and found 5 critical + 6 medium
+issues. All criticals and all mediums are fixed in **c9bd441** (PR #9):
+
+| # | Finding | Fix |
+|---|---|---|
+| C1 | approval forgeable with any `--actor` | hash-chained `audit_log` (`verify_audit_chain`); optional-required operator BIP-340 signature (`CAYCE_OPS_OPERATOR_PUBKEY`); `deploy/roles.sql` least-privilege DB roles |
+| C2 | mail threading dead (no OUTBOUND rows, inbound id overwritten, no first-contact) | sends create OUTBOUND rows (own Message-ID + In-Reply-To); inbound id untouched; `maild compose`; chain fallback in `_match_proposal` |
+| C3 | duplicate sends on SMTP failure | per-message transactions; failures audited (`send_failed`), row stays queued |
+| C4 | negated approvals classified as approvals; LLM prompt `.format()` KeyError | negation rules first (0.95 REJECTION); prompt braces escaped |
+| C5 | systemd units ran nonexistent `run` subcommands | `run` added to all five daemons; `publisher plan` timer documented |
+| M1-M6 | gate not self-installed; approved text unbound; `risk_score` misnamed/ordered; postgres 0.0.0.0; IMAP seen-before-commit; publisher noise/infinite retries | `make hooks`; `approved_hash` bindings; `match_score` (migration 0002) + relevance-desc; `127.0.0.1` bind; `\Seen` post-commit; `SKIPPED` terminal + attempt budget |
+
+New audited commit: **c9bd441**. Gate at that commit: ruff 0 · 56/56 tests ·
+alembic 0001→0002 · gitleaks `--all` clean · `make check` exit 0.
+Still open (declared): apply `roles.sql` on production, provision CI runner.

Makefile

@@ -1,4 +1,4 @@
-.PHONY: check lint test migrate-check leaks sync
+.PHONY: check lint test migrate-check leaks sync hooks
 
 sync:            ## install locked deps (creates .venv)
 	uv sync --frozen
@@ -17,5 +17,9 @@ migrate-check:   ## smoke the alembic chain on a scratch sqlite db
 leaks:           ## gitleaks over the full history
 	gitleaks detect --source . --log-opts="--all" -v
 
+hooks:            ## install the pre-push gate as a git hook (part of setup)
+	git config core.hooksPath scripts/githooks
+	@echo "pre-push gate installed (scripts/githooks/pre-push)"
+
 check: lint test migrate-check leaks   ## the pre-push gate — must be green before every push
 	@echo "check: OK"

alembic/versions/0002_audit_hardening.py

@@ -0,0 +1,68 @@
+"""audit hardening migration — minimal standalone mirror of 0002.
+
+Revision ID: 0002
+Revises: 0001
+"""
+from __future__ import annotations
+
+import sqlalchemy as sa
+
+from alembic import op
+
+revision = "0002"
+down_revision = "0001"
+branch_labels = None
+depends_on = None
+
+OLD_CONTENT_STATUSES = ("DRAFT", "APPROVED", "PUBLISHED")
+NEW_CONTENT_STATUSES = ("DRAFT", "APPROVED", "PUBLISHED", "SKIPPED")
+
+
+def upgrade() -> None:
+    with op.batch_alter_table("audit_log") as batch:
+        batch.add_column(sa.Column("prev_hash", sa.String(length=64), nullable=True))
+        batch.add_column(sa.Column("entry_hash", sa.String(length=64), nullable=True))
+        batch.create_index("ix_audit_log_entry_hash", ["entry_hash"])
+
+    with op.batch_alter_table("proposals") as batch:
+        batch.alter_column("risk_score", new_column_name="match_score")
+
+    with op.batch_alter_table("mail_messages") as batch:
+        batch.add_column(sa.Column("approved_hash", sa.String(length=64), nullable=True))
+        batch.add_column(sa.Column("approved_by", sa.String(length=128), nullable=True))
+
+    if op.get_bind().dialect.name == "postgresql":
+        op.execute("ALTER TYPE contentstatus ADD VALUE IF NOT EXISTS 'SKIPPED'")
+    with op.batch_alter_table("content_items") as batch:
+        batch.alter_column(
+            "status",
+            existing_type=sa.Enum(*OLD_CONTENT_STATUSES, name="contentstatus"),
+            type_=sa.Enum(*NEW_CONTENT_STATUSES, name="contentstatus"),
+            existing_nullable=False,
+        )
+        batch.add_column(sa.Column("attempts", sa.Integer(), nullable=False, server_default="0"))
+        batch.add_column(sa.Column("approved_hash", sa.String(length=64), nullable=True))
+
+
+def downgrade() -> None:
+    with op.batch_alter_table("content_items") as batch:
+        batch.drop_column("approved_hash")
+        batch.drop_column("attempts")
+        batch.alter_column(
+            "status",
+            existing_type=sa.Enum(*NEW_CONTENT_STATUSES, name="contentstatus"),
+            type_=sa.Enum(*OLD_CONTENT_STATUSES, name="contentstatus"),
+            existing_nullable=False,
+        )
+
+    with op.batch_alter_table("mail_messages") as batch:
+        batch.drop_column("approved_by")
+        batch.drop_column("approved_hash")
+
+    with op.batch_alter_table("proposals") as batch:
+        batch.alter_column("match_score", new_column_name="risk_score")
+
+    with op.batch_alter_table("audit_log") as batch:
+        batch.drop_index("ix_audit_log_entry_hash")
+        batch.drop_column("entry_hash")
+        batch.drop_column("prev_hash")

cayceops/cli.py

@@ -71,10 +71,16 @@ def executor_run() -> None:
 
 @executor_app.command("approve")
 def executor_approve(
-    proposal_id: int, actor: str = typer.Option(..., help="Human approver id")
+    proposal_id: int,
+    actor: str = typer.Option(..., help="Human approver id"),
+    signature: str = typer.Option(
+        None, help="BIP-340 sig (required when operator pubkey is set)"
+    ),
 ) -> None:
     """Approve a proposal (human action, audited)."""
-    raise typer.Exit(_load_component("executor").approve(proposal_id, actor))
+    raise typer.Exit(
+        _load_component("executor").approve(proposal_id, actor, signature=signature)
+    )
 
 
 @executor_app.command("reject")
@@ -102,10 +108,33 @@ def maild_approve(msg_id: int, actor: str = typer.Option(...)) -> None:
 
 @maild_app.command("send-approved")
 def maild_send() -> None:
-    """Send all approved drafts via SMTP."""
+    """Send all approved drafts via SMTP (per-message commit)."""
     raise typer.Exit(_load_component("maild").send_approved())
 
 
+@maild_app.command("compose")
+def maild_compose(
+    proposal_id: int,
+    to_addr: str = typer.Option(...),
+    subject: str = typer.Option(...),
+    body: str = typer.Option(...),
+) -> None:
+    """Create the FIRST-CONTACT outbound draft for a proposal (drafted)."""
+    raise typer.Exit(
+        _load_component("maild").compose(proposal_id, to_addr, subject, body)
+    )
+
+
+@maild_app.command("run")
+def maild_run() -> None:
+    """One daemon cycle: poll inbound, then send approved drafts."""
+    comp = _load_component("maild")
+    code = comp.poll()
+    if code == 0:
+        code = comp.send_approved()
+    raise typer.Exit(code)
+
+
 # --- publisher ---------------------------------------------------------------
 publisher_app = typer.Typer(name="publisher", help="Themed content calendar")
 app.add_typer(publisher_app)
@@ -135,6 +164,12 @@ def publisher_publish() -> None:
     raise typer.Exit(_load_component("publisher").publish_due())
 
 
+@publisher_app.command("run")
+def publisher_run() -> None:
+    """One daemon cycle: publish due approved items (plan is cron'd separately)."""
+    raise typer.Exit(_load_component("publisher").publish_due())
+
+
 # --- monitor -----------------------------------------------------------------
 monitor_app = typer.Typer(name="monitor", help="Static status page")
 app.add_typer(monitor_app)
@@ -146,6 +181,12 @@ def monitor_build() -> None:
     raise typer.Exit(_load_component("monitor").build())
 
 
+@monitor_app.command("run")
+def monitor_run() -> None:
+    """One daemon cycle: build the status page."""
+    raise typer.Exit(_load_component("monitor").build())
+
+
 def main() -> None:  # pragma: no cover
     app()
 

cayceops/components/executor/executor.py

@@ -29,12 +29,59 @@ from cayceops.models import AuditEntry, DaemonRun, Proposal, ProposalStatus, aud
 Actor = str
 
 
+def _load_cfg():
+    from cayceops.config import load
+
+    return load()
+
+
 # --- human transitions -------------------------------------------------------
 
+# message signed by the operator key when CAYCE_OPS_OPERATOR_PUBKEY is set
+def approval_message(proposal_id: int) -> bytes:
+    import hashlib
 
-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)
+    return hashlib.sha256(f"cayce-ops:approve:{proposal_id}".encode()).digest()
+
+
+def approve(
+    proposal_id: int,
+    actor: Actor,
+    cfg: Config | None = None,
+    sess=None,
+    signature: str | None = None,
+) -> int:
+    """Human approves a proposal. Audited with the operator's identity.
+
+    When CAYCE_OPS_OPERATOR_PUBKEY is configured, `signature` (hex, BIP-340
+    over approval_message(id)) is REQUIRED and verified against that key —
+    'actor' then is a label, the key is the proof (audit finding 1).
+    """
+    from cayceops.components.publisher.nostr import verify_schnorr
+
+    cfg = cfg or _load_cfg()
+    if cfg.operator_pubkey:
+        ok = bool(signature) and verify_schnorr(
+            signature, approval_message(proposal_id), cfg.operator_pubkey
+        )
+        if not ok and sess is not None:
+            audit(
+                sess,
+                actor=actor,
+                action="approve",
+                proposal_id=proposal_id,
+                result="error",
+                detail={
+                    "reason": (
+                        "operator pubkey configured: signature required"
+                        if not signature
+                        else "invalid approval signature"
+                    )
+                },
+            )
+        if not ok:
+            return 2
+    return _decide(proposal_id, actor, ProposalStatus.APPROVED, cfg, sess, signature)
 
 
 def reject(proposal_id: int, actor: Actor, cfg: Config | None = None, sess=None) -> int:
@@ -48,6 +95,7 @@ def _decide(
     decision: ProposalStatus,
     cfg: Config | None,
     sess,
+    signature: str | None = None,
 ) -> int:
     owned = False
     if sess is None:
@@ -70,11 +118,14 @@ def _decide(
             )
             return 1
         proposal.status = decision
+        detail: dict = {}
         if decision == ProposalStatus.APPROVED:
             proposal.approved_by = actor
             proposal.approved_at = datetime.now(UTC)
+            if signature:
+                detail["signature"] = signature[:32] + "..."
         verb = "approve" if decision == ProposalStatus.APPROVED else "reject"
-        audit(sess, actor=actor, action=verb, proposal_id=proposal_id)
+        audit(sess, actor=actor, action=verb, proposal_id=proposal_id, detail=detail)
         return 0
     finally:
         if owned:
@@ -102,13 +153,15 @@ def _auto_approve(cfg: Config, sess) -> int:
     approved = 0
     if budget == 0:
         return 0
+    # most relevant first (match_score = keyword relevance, NOT risk — the
+    # audit's naming finding; ascending order approved noise first)
     candidates = (
         sess.query(Proposal)
         .filter(
             Proposal.status == ProposalStatus.PROPOSED,
             Proposal.kind.in_(cfg.autonomous_kinds),
         )
-        .order_by(Proposal.risk_score.asc(), Proposal.id.asc())
+        .order_by(Proposal.match_score.desc(), Proposal.id.asc())
         .limit(budget)
         .all()
     )
@@ -124,7 +177,7 @@ def _auto_approve(cfg: Config, sess) -> int:
             detail={
                 "reason": "autonomous_mode policy",
                 "kind": proposal.kind,
-                "risk_score": proposal.risk_score,
+                "match_score": proposal.match_score,
                 "daily_cap": cfg.autonomous_daily_cap,
             },
         )

cayceops/components/maild/__init__.py

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

cayceops/components/maild/classify.py

@@ -0,0 +1,105 @@
+"""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
+
+# NEGATED approval forms are rejections and must be checked FIRST:
+# "We cannot accept your application", "not accepting new partners",
+# "Your application was not approved" used to match the bare approval
+# keywords (accept/approv) and read as approvals — audit finding 4.
+_NEGATED_APPROVAL = (
+    MailClassification.REJECTION,
+    re.compile(
+        r"\b(cannot|can.?t|will not|won.?t|do not|don.?t|does not|doesn.?t)\s+"
+        r"(be\s+able\s+to\s+)?(accept|approve|proceed|offer|welcome)"
+        r"|\bnot\s+(been\s+)?approved\b"
+        r"|\bnot\s+accepting\b"
+        r"|\bunfortunately\b",
+        re.I,
+    ),
+    0.95,
+)
+
+_RULES: list[tuple[MailClassification, re.Pattern[str], float]] = [
+    _NEGATED_APPROVAL,
+    # (class, pattern, confidence when matched)
+    (
+        MailClassification.REJECTION,
+        # noqa: E501 (kept on one line: the pattern IS the reviewed artifact)
+        re.compile(r"\b(declin|reject|not able to|unable to|we regret|won.t be able)\w*", re.I),  # noqa: E501
+        0.9,
+    ),
+    (
+        MailClassification.APPROVAL,
+        re.compile(r"\b(approv|accept|welcome to the|you.re (in|approved)|live|activated|glad to)\w*", re.I),  # noqa: E501
+        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),  # noqa: E501
+        0.7,
+    ),
+]
+
+# braces are DOUBLED so .format() does not die on the literal JSON sample
+# (the old prompt raised KeyError on every call and the generic except
+# silently degraded to rules — audit finding 4)
+_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,566 @@
+"""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,
+    _content_hash,
+    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[tuple[bytes, bytes]]:
+    """Fetch raw unread messages as [(imap_num, raw_bytes)].
+
+    Messages are NOT marked \\Seen here — marking happens only after the
+    processed rows are committed (audit finding: a crash mid-batch used to
+    silently lose the rest of the batch).
+    """
+    client = connect_imap(cfg)
+    try:
+        _, data = client.search(None, "UNSEEN")
+        out: list[tuple[bytes, bytes]] = []
+        for num in data[0].split():
+            _, msg_data = client.fetch(num, "(RFC822)")
+            out.append((num, msg_data[0][1]))
+        return out
+    finally:
+        client.logout()
+
+
+def _mark_seen(cfg: Config, nums: list[bytes]) -> None:
+    """Mark processed messages \\Seen (called after commit)."""
+    if not nums:
+        return
+    client = connect_imap(cfg)
+    try:
+        for num in nums:
+            client.store(num, "+FLAGS", "\\Seen")
+    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.
+
+    1. In-Reply-To matches a SENT OUTBOUND message (our replies carry their own
+       Message-ID and an outbound row, created at send time).
+    2. Fallback: it matches ANY stored message (e.g. a second reply threaded to
+       our first inbound) that already carries a proposal_id.
+    """
+    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 not None and sent.proposal_id is not None:
+        return sess.get(Proposal, sent.proposal_id)
+    chain = (
+        sess.query(MailMessage)
+        .filter(MailMessage.message_id == irt, MailMessage.proposal_id.isnot(None))
+        .one_or_none()
+    )
+    if chain is not None:
+        return sess.get(Proposal, chain.proposal_id)
+    return None
+
+
+# --- 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, marker=None) -> tuple[dict, list[bytes]]:
+    """One inbound pass: parse, classify, thread, draft.
+
+    Returns (stats, processed_imap_nums). The caller marks \\Seen only after
+    its transaction commits.
+    """
+    stats = {"processed": 0, "drafted": 0, "needs_review": 0, "skipped": 0}
+    processed_nums: list[bytes] = []
+    for num, 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
+        processed_nums.append(num)  # stored -> safe to mark seen after commit
+
+        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, processed_nums
+
+
+# --- human gate: approve draft -------------------------------------------------
+
+
+def _approve_draft(mail_id: int, actor: str, sess) -> int:
+    """Human approves a drafted reply (inbound-threaded or first-contact).
+
+    The approval is BOUND to the draft text via sha256 — a later edit breaks
+    the binding and the send refuses (audit finding: approved text was not
+    tied to its content).
+    """
+    record = sess.get(MailMessage, mail_id)
+    if record is None:
+        return 1
+    if record.draft_status != DraftStatus.DRAFTED or not record.draft_reply:
+        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
+    record.approved_hash = _content_hash(record.draft_reply)
+    record.approved_by = actor[:128]
+    audit(
+        sess,
+        actor=actor,
+        action="approve_draft",
+        proposal_id=record.proposal_id,
+        detail={"mail_id": record.id, "hash": record.approved_hash},
+    )
+    return 0
+
+
+# --- SMTP: only approved drafts ------------------------------------------------
+
+
+def _smtp_send(
+    cfg: Config,
+    to_addr: str,
+    subject: str,
+    body: str,
+    message_id: str,
+    in_reply_to: str | None = None,
+) -> None:
+    msg = EmailMessage()
+    msg["From"] = cfg.mail_from
+    msg["To"] = to_addr
+    msg["Subject"] = subject
+    msg["Message-ID"] = message_id
+    if in_reply_to:
+        msg["In-Reply-To"] = in_reply_to  # keep merchant threads intact
+    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_one(cfg: Config, sess, mail_id: int, sender=None) -> str:
+    """Send ONE approved draft. Returns: sent | hash_mismatch | missing | blocked.
+
+    Creates the OUTBOUND row (its own Message-ID + In-Reply-To pointing at the
+    inbound message) so future replies thread correctly — the inbound row's
+    message_id is left untouched (audit finding: it used to be overwritten).
+    The caller owns the transaction: one message per commit, so an SMTP failure
+    on message N never un-sends messages 1..N-1 (audit finding: duplicates).
+    """
+    from cayceops.models import _content_hash
+
+    record = sess.get(MailMessage, mail_id)
+    if record is None:
+        return "missing"
+    if record.draft_status != DraftStatus.APPROVED:
+        return "blocked"
+
+    if record.direction == MailDirection.INBOUND:
+        # reply: goes back to the inbound sender, threads to the inbound mail,
+        # and gets its own OUTBOUND row
+        to_addr = record.from_addr or ""
+        subject = f"Re: {record.subject or 'your message'}"
+        in_reply_to = record.message_id
+        outbound_msgid = (
+            f"<cayce-out-{record.id}-{int(datetime.now(UTC).timestamp())}@cayce-co.xyz>"
+        )
+        needs_row = True
+    else:
+        # first-contact draft: already an OUTBOUND row; send as-is
+        to_addr = record.to_addr or ""
+        subject = record.subject or ""
+        in_reply_to = None
+        outbound_msgid = record.message_id
+        needs_row = False
+    if not to_addr:
+        record.draft_status = DraftStatus.SKIPPED
+        return "blocked"
+
+    # binding: the text must be exactly what the human approved
+    if record.approved_hash and _content_hash(record.draft_reply or "") != record.approved_hash:
+        record.draft_status = DraftStatus.DRAFTED  # needs fresh approval
+        audit(
+            sess,
+            actor="daemon:maild",
+            action="send_refused_hash_mismatch",
+            proposal_id=record.proposal_id,
+            result="error",
+            detail={"mail_id": record.id},
+        )
+        return "hash_mismatch"
+
+    (sender or _smtp_send)(
+        cfg,
+        to_addr,
+        subject,
+        record.draft_reply or "",
+        outbound_msgid,
+        in_reply_to=in_reply_to,
+    )
+    approved_by = _approved_by(sess, record.id)
+    if needs_row:
+        sess.add(
+            MailMessage(
+                direction=MailDirection.OUTBOUND,
+                message_id=outbound_msgid,
+                in_reply_to=record.message_id,
+                from_addr=cfg.mail_from,
+                to_addr=to_addr,
+                subject=subject,
+                body=record.draft_reply,
+                proposal_id=record.proposal_id,
+                draft_status=DraftStatus.SENT,
+                processed_at=datetime.now(UTC),
+            )
+        )
+    record.draft_status = DraftStatus.SENT
+    sess.flush()
+    audit(
+        sess,
+        actor="daemon:maild",
+        action="send_draft",
+        proposal_id=record.proposal_id,
+        detail={
+            "mail_id": record.id,
+            "message_id": outbound_msgid,
+            "approved_by": approved_by,
+        },
+    )
+    return "sent"
+
+
+def _send_approved(cfg: Config, sess, sender=None, factory=None) -> dict:
+    """Send every APPROVED draft, one transaction per message.
+
+    `factory` is a SQLAlchemy sessionmaker bound to the shared engine; each
+    message is processed and committed in its own session (tests inject the
+    engine's maker; the CLI passes appdb.session_factory()).
+    """
+    stats = {"sent": 0, "blocked": 0, "hash_mismatch": 0, "missing": 0}
+    ids = [
+        row.id
+        for row in sess.query(MailMessage.id)
+        .filter(MailMessage.draft_status == DraftStatus.APPROVED)
+        .all()
+    ]
+    for mail_id in ids:
+        # one transaction per message — an SMTP failure on message N must not
+        # roll back (and later re-send) messages 1..N-1 (audit finding 3)
+        try:
+            if factory is not None:
+                with factory() as one_sess:  # sessionmaker() -> Session
+                    outcome = _send_one(cfg, one_sess, mail_id, sender)
+                    one_sess.commit()
+            else:
+                outcome = _send_one(cfg, sess, mail_id, sender)
+        except Exception as exc:  # noqa: BLE001 — isolate the failed message
+            outcome = "error"
+            if sess is not None:
+                detail = {"mail_id": mail_id, "error": str(exc)[:200]}
+                audit(sess, actor="daemon:maild", action="send_failed",
+                      result="error", detail=detail)
+        stats[outcome] = stats.get(outcome, 0) + 1
+    return stats
+
+
+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 _compose(proposal_id: int, to_addr: str, subject: str, body: str, sess) -> int:
+    """Create the FIRST-CONTACT outbound draft for a proposal (state: drafted).
+
+    This is the missing path the audit flagged: without it, threading only
+    worked for replies to mails we never actually sent.
+    """
+    proposal = sess.get(Proposal, proposal_id)
+    if proposal is None or not to_addr:
+        return -1
+    outbound = MailMessage(
+        direction=MailDirection.OUTBOUND,
+        message_id=f"<cayce-out-{proposal.id}-{int(datetime.now(UTC).timestamp())}@cayce-co.xyz>",
+        from_addr=None,  # filled from MAIL_FROM at send time
+        to_addr=to_addr[:320],
+        subject=subject[:998],
+        body=body,
+        draft_reply=body,  # canonical approved-text field (same as reply drafts)
+        proposal_id=proposal.id,
+        draft_status=DraftStatus.DRAFTED,
+    )
+    sess.add(outbound)
+    sess.flush()
+    audit(
+        sess,
+        actor="daemon:maild",
+        action="compose_draft",
+        proposal_id=proposal.id,
+        detail={"mail_id": outbound.id},
+    )
+    return outbound.id
+
+
+# --- 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:
+        nums: list[bytes] = []
+        with session_scope() as sess:
+            run_row = DaemonRun(daemon="maild")
+            sess.add(run_row)
+            sess.flush()
+            try:
+                stats, nums = _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)
+        # committed: now it is safe to mark the batch seen
+        if nums:
+            _mark_seen(cfg, nums)
+        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, session_factory
+
+    cfg = cfg or load()
+    try:
+        # ids first, then one transaction per message: a failure on message N
+        # cannot roll back already-sent messages (audit finding: duplicates)
+        with session_scope() as sess:
+            ids = [
+                row.id
+                for row in sess.query(MailMessage.id)
+                .filter(MailMessage.draft_status == DraftStatus.APPROVED)
+                .all()
+            ]
+        factory = session_factory()
+        for mail_id in ids:
+            try:
+                with factory() as one_sess:
+                    _send_one(cfg, one_sess, mail_id)
+                    one_sess.commit()
+            except Exception as exc:  # noqa: BLE001 — isolate each send
+                print(f"send {mail_id} failed: {str(exc)[:160]}")
+        return 0
+    finally:
+        engine().dispose()
+
+
+def compose(proposal_id: int, to_addr: str, subject: str, body: str) -> int:
+    from cayceops.db import engine
+
+    try:
+        with session_scope() as sess:
+            mail_id = _compose(proposal_id, to_addr, subject, body, sess)
+            return 0 if mail_id >= 0 else 1
+    finally:
+        engine().dispose()

cayceops/components/monitor/__init__.py

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

cayceops/components/monitor/monitor.py

@@ -0,0 +1,174 @@
+"""monitor — static status page builder.
+
+Reads the shared database and renders a self-contained HTML file (no external
+assets, no JS, everything escaped). Contains NO sensitive data: daemon health,
+proposal queue counts, and recent audit activity (actor/action/result only —
+no email bodies, no detail payloads, no URLs).
+"""
+
+from __future__ import annotations
+
+import html
+from datetime import UTC, datetime
+
+from sqlalchemy import func, select
+
+from cayceops.config import Config
+from cayceops.db import session_scope
+from cayceops.models import (
+    AuditEntry,
+    ContentItem,
+    DaemonRun,
+    MailMessage,
+    Proposal,
+    ProposalStatus,
+)
+
+DAEMONS = ("scanner", "executor", "maild", "publisher", "monitor")
+
+_STYLE = """<style>
+ body{font-family:ui-monospace,Menlo,monospace;background:#0a0908;color:#f2ece0;margin:2rem}
+ h1{color:#c9a84c;font-weight:600} h2{color:#c9a84c;margin-top:2rem;font-size:1.1rem}
+ table{border-collapse:collapse;margin:.5rem 0}
+ td,th{border:1px solid #333;padding:.35rem .8rem;text-align:left}
+ .ok{color:#6fce6f}.error{color:#ef6b6b}.muted{color:#888;font-size:.85rem}
+ .card{display:inline-block;background:#141210;border:1px solid #2a2a2a;
+       border-radius:8px;padding:.8rem 1.4rem;margin:.3rem}
+</style>"""
+
+
+def _fmt_ts(dt) -> str:
+    return dt.strftime("%Y-%m-%d %H:%M:%SZ") if dt else "—"
+
+
+def _daemon_rows(sess) -> str:
+    rows = []
+    for name in DAEMONS:
+        last = (
+            sess.query(DaemonRun)
+            .filter(DaemonRun.daemon == name)
+            .order_by(DaemonRun.started_at.desc())
+            .first()
+        )
+        if last is None:
+            rows.append(
+                "<tr><td>" + name + "</td><td class=muted>never run</td><td>—</td></tr>"
+            )
+        else:
+            cls = "ok" if last.status == "ok" else "error"
+            when = _fmt_ts(last.finished_at or last.started_at)
+            rows.append(
+                "<tr><td>" + html.escape(name) + "</td>"
+                + '<td class="' + cls + '">' + html.escape(last.status) + "</td>"
+                + "<td>" + when + "</td></tr>"
+            )
+    head = "<tr><th>daemon</th><th>last status</th><th>last run</th></tr>"
+    return "<table>" + head + "".join(rows) + "</table>"
+
+
+def _queue_cards(sess) -> str:
+    counts = dict(
+        sess.query(Proposal.status, func.count(Proposal.id)).group_by(Proposal.status).all()
+    )
+    cards = []
+    for status in ProposalStatus:
+        value = counts.get(status, 0)
+        cards.append(
+            "<div class=card>" + html.escape(status.value)
+            + ": <strong>" + str(value) + "</strong></div>"
+        )
+    return "".join(cards)
+
+
+def _recent_audit(sess, limit: int = 20) -> str:
+    rows = []
+    query = sess.query(AuditEntry).order_by(AuditEntry.ts.desc()).limit(limit)
+    for entry in query.all():
+        cls = "ok" if entry.result == "ok" else "error"
+        rows.append(
+            "<tr><td>" + _fmt_ts(entry.ts) + "</td>"
+            + "<td>" + html.escape(entry.actor) + "</td>"
+            + "<td>" + html.escape(entry.action) + "</td>"
+            + '<td class="' + cls + '">' + html.escape(entry.result) + "</td></tr>"
+        )
+    body = "".join(rows) or "<tr><td colspan=4 class=muted>no activity yet</td></tr>"
+    return (
+        "<table><tr><th>time (utc)</th><th>actor</th><th>action</th><th>result</th></tr>"
+        + body
+        + "</table>"
+    )
+
+
+def _build(cfg: Config, sess) -> str:
+    now = datetime.now(UTC)
+    mail_unprocessed = (
+        sess.query(func.count(MailMessage.id))
+        .filter(
+            MailMessage.direction == "in",
+            MailMessage.draft_status.in_(["drafted", "none"]),
+            MailMessage.classification.isnot(None),
+        )
+        .scalar()
+    )
+    content_pending = (
+        sess.query(func.count(ContentItem.id))
+        .filter(ContentItem.status.in_(["draft", "approved"]))
+        .scalar()
+    )
+    generated = _fmt_ts(now)
+
+    page = f"""<!doctype html>
+<html lang=en><head><meta charset=utf-8><title>cayce-ops status</title>{_STYLE}</head>
+<body>
+<h1>cayce-ops · status</h1>
+<p class=muted>generated {generated} · static page, no sensitive data by design</p>
+
+<h2>daemon health</h2>
+{_daemon_rows(sess)}
+
+<h2>proposal queue</h2>
+{_queue_cards(sess)}
+
+<h2>gates</h2>
+<div class=card>mail drafts awaiting approval: <strong>{mail_unprocessed}</strong></div>
+<div class=card>content items draft/approved: <strong>{content_pending}</strong></div>
+
+<h2>recent audit activity</h2>
+{_recent_audit(sess)}
+
+<p class=muted>audit entries: {sess.scalar(select(func.count(AuditEntry.id)))} total</p>
+</body></html>"""
+
+    import os
+
+    os.makedirs(os.path.dirname(cfg.monitor_output) or ".", exist_ok=True)
+    with open(cfg.monitor_output, "w", encoding="utf-8") as fh:
+        fh.write(page)
+    return cfg.monitor_output
+
+
+def build(cfg: Config | None = None) -> int:
+    """CLI entry: render the status page inside a DaemonRun bookkeeping wrap."""
+    from cayceops.config import load
+    from cayceops.db import engine
+
+    cfg = cfg or load()
+    try:
+        with session_scope() as sess:
+            run_row = DaemonRun(daemon="monitor")
+            sess.add(run_row)
+            sess.flush()
+            try:
+                out = _build(cfg, sess)
+            except Exception as exc:  # noqa: BLE001 — daemon boundary
+                run_row.status = "error"
+                run_row.detail = {"error": str(exc)[:300]}
+                run_row.finished_at = datetime.now(UTC)
+                return 1
+            run_row.status = "ok"
+            run_row.detail = {"output": out}
+            run_row.finished_at = datetime.now(UTC)
+        print(out)
+        return 0
+    finally:
+        engine().dispose()

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,82 @@
+"""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,186 @@
+"""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
+    generator = (
+        0x79BE667EF9DCBBAC55A06295CE870B07029BFCDB2DCE28D959F2815B16F81798,  # Gx
+        0x483ADA7726A3C4655DA4FBFC0E1108A8FD17B448A68554199C47D08FFB10D4B8,  # Gy
+    )
+    r_point = point_add(
+        point_mul(generator, 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
+    # 33-byte compressed point (02-prefixed x-only)
+    compressed = PublicKey(bytes.fromhex("02" + event["pubkey"])).format()
+    return _schnorr_verify(
+        bytes.fromhex(event["sig"]), bytes.fromhex(event["id"]), compressed[1:]
+    )
+
+
+def verify_schnorr(sig_hex: str, message: bytes, pubkey_hex: str) -> bool:
+    """Verify a BIP-340 signature over an arbitrary 32-byte message."""
+    try:
+        return _schnorr_verify(
+            bytes.fromhex(sig_hex), message, bytes.fromhex(pubkey_hex)
+        )
+    except (ValueError, TypeError):
+        return False
+
+
+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,322 @@
+"""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,
+    _content_hash,
+    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
+    item.approved_hash = _content_hash(item.caption)  # binding: caption as approved
+    audit(
+        sess,
+        actor=actor,
+        action="approve_content",
+        detail={"item": item.id, "channel": item.channel, "hash": item.approved_hash},
+    )
+    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, "tampered": 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
+
+        # binding: caption must be exactly what was approved (audit finding)
+        if item.approved_hash and _content_hash(item.caption) != item.approved_hash:
+            audit(
+                sess,
+                actor="daemon:publisher",
+                action="publish_refused_hash_mismatch",
+                result="error",
+                detail={"item": item.id},
+            )
+            stats["tampered"] += 1
+            continue
+
+        try:
+            if item.channel == "nostr":
+                item.post_ref = _publish_nostr(item, cfg, sess, sender=nostr_sender)
+            else:
+                # disabled channel is TERMINAL (audit finding: this used to
+                # write a skip audit row every 5 minutes, forever)
+                item.status = ContentStatus.SKIPPED
+                item.error = f"{item.channel} adapter disabled by default"
+                audit(
+                    sess,
+                    actor="daemon:publisher",
+                    action="publish_skipped_terminal",
+                    detail={"item": item.id, "channel": item.channel},
+                )
+                stats["skipped"] += 1
+                continue
+        except Exception as exc:  # noqa: BLE001 — publish boundary: audit and continue
+            item.attempts += 1
+            item.error = str(exc)[:300]
+            if item.attempts >= cfg.max_attempts:
+                # attempt budget exhausted -> terminal (no infinite retries)
+                item.status = ContentStatus.SKIPPED
+                audit(
+                    sess,
+                    actor="daemon:publisher",
+                    action="publish_abandoned",
+                    result="error",
+                    detail={"item": item.id, "attempts": item.attempts},
+                )
+            else:
+                audit(
+                    sess,
+                    actor="daemon:publisher",
+                    action="publish_failed",
+                    result="error",
+                    detail={"item": item.id, "error": item.error, "attempt": item.attempts},
+                )
+            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/scanner.py

@@ -165,7 +165,7 @@ def scan(cfg: Config, sess, fetch=_fetch) -> dict:
                     "summary": desc,
                     "scanned_at": datetime.now(UTC).isoformat(),
                 },
-                risk_score=score,
+                match_score=score,
                 status=ProposalStatus.PROPOSED,
                 dedupe_key=dedupe,
             )

cayceops/config.py

@@ -84,6 +84,11 @@ class Config:
     enable_http_actions: bool = field(
         default_factory=lambda: _bool("CAYCE_OPS_ENABLE_HTTP_ACTIONS", False)
     )
+    # when set (64-char hex x-only pubkey), human approvals MUST carry a valid
+    # BIP-340 signature over "cayce-ops:approve:<id>" made by the operator key.
+    operator_pubkey: str = field(
+        default_factory=lambda: os.environ.get("CAYCE_OPS_OPERATOR_PUBKEY", "")
+    )
 
     # --- mail ---
     imap_host: str = field(

cayceops/models.py

@@ -5,13 +5,22 @@ Status machines
 proposal:  proposed -> approved -> running -> done | failed
            proposed -> rejected
 mail:      (inbound) classified: approval|rejection|question|other|needs_review
-           (outbound draft) none -> drafted -> approved -> sent | skipped
-content:   draft -> approved -> published
+           (any) draft: none -> drafted -> approved -> sent | skipped
+content:   draft -> approved -> published | skipped (terminal)
+
+Audit chain
+-----------
+audit_log is hash-chained: each entry stores sha256(prev_hash | fields).
+``cayceops.models.verify_audit_chain`` re-walks it; the monitor surfaces breaks.
+Not tamper-proof (DB writes can rewrite the chain) but tamper-EVIDENT against
+selective edits. See AUDIT.md / audit response 2026-10-20.
 """
 
 from __future__ import annotations
 
 import enum
+import hashlib
+import json
 from datetime import UTC, datetime
 
 from sqlalchemy import JSON, DateTime, Enum, Float, ForeignKey, Integer, String, Text
@@ -22,6 +31,11 @@ def utcnow() -> datetime:
     return datetime.now(UTC)
 
 
+def _content_hash(text: str) -> str:
+    """Binding hash for human-approved text (drafts, captions)."""
+    return hashlib.sha256(text.encode("utf-8")).hexdigest()
+
+
 class Base(DeclarativeBase):
     pass
 
@@ -54,7 +68,8 @@ class Proposal(Base):
     title: Mapped[str] = mapped_column(String(512))
     url: Mapped[str | None] = mapped_column(String(1024), nullable=True)
     detail: Mapped[dict] = mapped_column(JSON, default=dict)
-    risk_score: Mapped[float] = mapped_column(Float, default=0.0)
+    # keyword-match relevance 0..1 — NOT risk. auto-approve takes highest first.
+    match_score: Mapped[float] = mapped_column(Float, default=0.0)
     status: Mapped[ProposalStatus] = mapped_column(
         Enum(ProposalStatus), default=ProposalStatus.PROPOSED, index=True
     )
@@ -68,7 +83,7 @@ class Proposal(Base):
     )
 
 
-# --- audit -------------------------------------------------------------------
+# --- audit (hash-chained) ----------------------------------------------------
 
 
 class AuditEntry(Base):
@@ -76,13 +91,16 @@ class AuditEntry(Base):
 
     id: Mapped[int] = mapped_column(Integer, primary_key=True)
     ts: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow, index=True)
-    actor: Mapped[str] = mapped_column(String(128))  # human username | auto:policy | daemon:<name>
+    actor: Mapped[str] = mapped_column(String(128))  # human | auto:policy | daemon:<name>
     proposal_id: Mapped[int | None] = mapped_column(
         ForeignKey("proposals.id", ondelete="SET NULL"), nullable=True, index=True
     )
     action: Mapped[str] = mapped_column(String(64))  # approve|reject|execute|classify|...
     detail: Mapped[dict] = mapped_column(JSON, default=dict)
     result: Mapped[str] = mapped_column(String(32), default="ok")  # ok | error
+    # tamper-evident chain
+    prev_hash: Mapped[str | None] = mapped_column(String(64), nullable=True)
+    entry_hash: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True)
 
 
 def audit(
@@ -93,18 +111,68 @@ def audit(
     detail: dict | None = None,
     result: str = "ok",
 ) -> AuditEntry:
-    """Record one audited action. Every component writes through this helper."""
+    """Record one audited action, chained to the previous entry.
+
+    Chain: entry_hash = sha256(prev_hash | ts | actor | action | proposal_id | detail | result).
+    Every component writes through this helper.
+    """
+    prev = sess.query(AuditEntry).order_by(AuditEntry.id.desc()).first()
+    prev_hash = prev.entry_hash if prev is not None else "GENESIS"
+    ts = utcnow()
+    # epoch int: identical after a sqlite/pg round-trip (tz-aware isoformat is not)
+    ts_int = int(ts.timestamp())
+    detail_json = json.dumps(detail or {}, sort_keys=True, ensure_ascii=False)
+    payload = "|".join(
+        [
+            prev_hash or "GENESIS",
+            str(ts_int),
+            actor,
+            action,
+            str(proposal_id),
+            detail_json,
+            result,
+        ]
+    )
     entry = AuditEntry(
+        ts=ts,
         actor=actor,
         action=action,
         proposal_id=proposal_id,
         detail=detail or {},
         result=result,
+        prev_hash=prev_hash,
+        entry_hash=hashlib.sha256(payload.encode("utf-8")).hexdigest(),
     )
     sess.add(entry)
     return entry
 
 
+def verify_audit_chain(sess) -> tuple[bool, int | None]:
+    """Re-walk the chain. Returns (ok, first_broken_entry_id or None)."""
+    prev_hash = "GENESIS"
+    for entry in sess.query(AuditEntry).order_by(AuditEntry.id).all():
+        detail_json = json.dumps(entry.detail or {}, sort_keys=True, ensure_ascii=False)
+        ts = entry.ts
+        if ts.tzinfo is None:
+            ts = ts.replace(tzinfo=UTC)
+        payload = "|".join(
+            [
+                prev_hash,
+                str(int(ts.timestamp())),
+                entry.actor,
+                entry.action,
+                str(entry.proposal_id),
+                detail_json,
+                entry.result,
+            ]
+        )
+        expected = hashlib.sha256(payload.encode("utf-8")).hexdigest()
+        if entry.prev_hash != prev_hash or entry.entry_hash != expected:
+            return False, entry.id
+        prev_hash = entry.entry_hash
+    return True, None
+
+
 # --- mail --------------------------------------------------------------------
 
 
@@ -151,6 +219,9 @@ class MailMessage(Base):
         Enum(DraftStatus), default=DraftStatus.NONE
     )
     draft_reply: Mapped[str | None] = mapped_column(Text, nullable=True)
+    # binding: hash of the draft text the human approved + who approved it
+    approved_hash: Mapped[str | None] = mapped_column(String(64), nullable=True)
+    approved_by: Mapped[str | None] = mapped_column(String(128), nullable=True)
     processed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
 
 
@@ -161,6 +232,7 @@ class ContentStatus(enum.StrEnum):
     DRAFT = "draft"
     APPROVED = "approved"
     PUBLISHED = "published"
+    SKIPPED = "skipped"  # terminal: disabled channel or attempt budget exhausted
 
 
 class ContentItem(Base):
@@ -179,6 +251,9 @@ class ContentItem(Base):
     published_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
     post_ref: Mapped[str | None] = mapped_column(String(512), nullable=True)
     error: Mapped[str | None] = mapped_column(Text, nullable=True)
+    attempts: Mapped[int] = mapped_column(Integer, default=0)
+    # binding: hash of the caption the human approved
+    approved_hash: Mapped[str | None] = mapped_column(String(64), nullable=True)
     created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow)
 
 

deploy/roles.sql

@@ -0,0 +1,22 @@
+-- Per-daemon least-privilege roles (audit critical finding 1, deployment half).
+-- The app-level chain + signatures are in code; this locks the DB layer.
+-- Apply on the production Postgres; adjust the connection limits.
+CREATE ROLE cayce_scanner LOGIN PASSWORD 'set-from-env';
+CREATE ROLE cayce_executor LOGIN PASSWORD 'set-from-env';
+CREATE ROLE cayce_maild LOGIN PASSWORD 'set-from-env';
+CREATE ROLE cayce_publisher LOGIN PASSWORD 'set-from-env';
+CREATE ROLE cayce_monitor LOGIN PASSWORD 'set-from-env';
+
+GRANT SELECT ON proposals, audit_log, mail_messages, content_items, daemon_runs TO cayce_monitor;
+GRANT SELECT, INSERT, UPDATE ON proposals TO cayce_scanner;   -- no DELETE
+GRANT INSERT ON audit_log TO cayce_scanner;                   -- append-only
+GRANT SELECT, UPDATE ON proposals TO cayce_executor;          -- status transitions
+GRANT INSERT ON audit_log TO cayce_executor;                  -- append-only
+GRANT SELECT, INSERT, UPDATE ON mail_messages, proposals TO cayce_maild;
+GRANT INSERT ON audit_log TO cayce_maild;
+GRANT SELECT, UPDATE ON content_items TO cayce_publisher;
+GRANT INSERT ON audit_log TO cayce_publisher;
+GRANT SELECT ON daemon_runs TO cayce_scanner, cayce_executor, cayce_maild, cayce_publisher;
+
+-- sequences
+GRANT USAGE, SELECT ON ALL SEQUENCES IN SCHEMA public TO cayce_scanner, cayce_executor, cayce_maild, cayce_publisher, cayce_monitor;

deploy/systemd/README.md

@@ -9,9 +9,14 @@ sudo -u cayce uv sync --frozen
 sudo -u cayce cayceops db upgrade          # alembic to head
 sudo cp deploy/systemd/cayce-ops@.service deploy/systemd/cayce-ops@.timer /etc/systemd/system/
 
+# Every daemon has a `run` subcommand now:
+#   scanner run · executor run · maild run (poll + send-approved)
+#   publisher run (publish-due) · monitor run (build)
 # cadence per daemon (edit OnCalendar in each timer copy):
 #   scanner  hourly      executor  */10 min      maild  */5 min
 #   publisher */5 min    monitor   */1 min
+# NOTE: `publisher plan` (fills the calendar) is a SEPARATE timer:
+#   OnCalendar=daily  ->  ExecStart=/srv/cayce-ops/.venv/bin/cayceops publisher plan
 for d in scanner executor maild publisher monitor; do
   sudo cp deploy/systemd/cayce-ops@.timer /etc/systemd/system/cayce-ops@$d.timer
 done

docker-compose.yml

@@ -6,7 +6,7 @@ services:
       POSTGRES_PASSWORD: localdev-only-not-a-secret
       POSTGRES_DB: cayce_ops
     ports:
-      - "5432:5432"
+      - "127.0.0.1:5432:5432"  # loopback only — Docker bypasses host firewalls
     volumes:
       - pgdata:/var/lib/postgresql/data
     healthcheck:

docs/DECISIONS.md

@@ -69,3 +69,29 @@ An action plugin is a Python function registered in
 `http_post` exists but is inert unless `CAYCE_OPS_ENABLE_HTTP_ACTIONS=true`.
 *Cost:* no sandboxing of plugins — plugins are repo code, reviewed like
 everything else.
+
+## D10 — Audit chain is hash-chained; approval can require operator signature (post-audit)
+`audit_log` rows chain sha256(prev_hash|fields); `verify_audit_chain` walks it,
+the monitor can surface breaks. With `CAYCE_OPS_OPERATOR_PUBKEY` set,
+`executor approve` REQUIRES a BIP-340 signature over
+sha256("cayce-ops:approve:<id>") from the operator key — `--actor` becomes a
+label, the key is the proof (the audit's "anyone can sign as anyone").
+*Cost:* without a pubkey configured the gate degrades to the old free-text
+actor; separate DB roles per daemon remain a deployment TODO (SQL to tighten
+ships in deploy/roles.sql).
+
+## D11 — Mail threading now has a real first-contact path (post-audit)
+Sends create an OUTBOUND row (own Message-ID, In-Reply-To -> inbound);
+inbound rows keep their original Message-ID. `maild compose` creates the
+initial outreach draft for a proposal. Approval is bound to the draft text
+via sha256 (`approved_hash`): an edited draft refuses to send until
+re-approved. Each send commits in its own transaction — SMTP failure on
+message N can no longer resurrect 1..N-1 as duplicates.
+
+## D12 — IMAP \\Seen only after commit; content gates tighten (post-audit)
+Messages are fetched unmarked and flagged \\Seen after the processing
+transaction commits (crash mid-batch no longer loses mail). Publisher:
+caption hash-bound at approval, disabled channels are terminal (`SKIPPED`),
+publish failures burn a bounded attempt budget before abandonment.
+`risk_score` renamed `match_score` (keyword relevance) and auto-approval
+takes the HIGHEST score first.

scripts/githooks/pre-push

@@ -0,0 +1,7 @@
+#!/bin/sh
+# The gate, wired as a real hook (audit medium finding: it did not self-install).
+# Runs the exact CI pipeline; refuses the push on any failure.
+set -e
+cd "$(git rev-parse --show-toplevel)"
+make check
+echo "pre-push gate passed — safe to push"

tests/test_executor.py

@@ -32,6 +32,60 @@ def test_reject_records_actor(cfg, sess):
     assert p.status == ProposalStatus.REJECTED
 
 
+def test_signed_approval_required_when_operator_key_configured(cfg, sess):
+    """Critical finding 1: --actor was free text; now a key can be required."""
+    from coincurve import PrivateKey, PublicKey
+
+    from cayceops.components.executor import executor
+
+    secret = bytes.fromhex("22" * 32)
+    pubkey = PublicKey.from_secret(secret).format().hex()[2:]
+    p = _proposal()
+    sess.add(p)
+    sess.flush()
+
+    config = cfg(CAYCE_OPS_OPERATOR_PUBKEY=pubkey)
+
+    # no signature -> refused
+    assert executor.approve(p.id, "gustavo", config, sess, signature=None) == 2
+    # wrong key -> refused
+    bad = (
+        PrivateKey(bytes.fromhex("33" * 32))
+        .sign_schnorr(executor.approval_message(p.id), aux_randomness=b"\x00" * 32)
+        .hex()
+    )
+    assert executor.approve(p.id, "gustavo", config, sess, signature=bad) == 2
+    # correct key -> approved
+    good = (
+        PrivateKey(secret)
+        .sign_schnorr(executor.approval_message(p.id), aux_randomness=b"\x00" * 32)
+        .hex()
+    )
+    assert executor.approve(p.id, "gustavo", config, sess, signature=good) == 0
+    assert p.status is ProposalStatus.APPROVED
+    entry = sess.query(AuditEntry).filter_by(action="approve", result="ok").one()
+    assert "signature" in entry.detail
+
+
+def test_auto_approve_orders_by_relevance_desc(cfg, sess):
+    """Medium finding: ascending 'risk_score' approved noise first."""
+    low = _proposal(kind="content_gap", title="barely relevant")
+    low.match_score = 0.1
+    high = _proposal(kind="content_gap", title="highly relevant")
+    high.match_score = 0.9
+    sess.add_all([low, high])
+    sess.flush()
+
+    config = cfg(
+        CAYCE_OPS_AUTONOMOUS_MODE="true",
+        CAYCE_OPS_AUTONOMOUS_KINDS="content_gap",
+        CAYCE_OPS_AUTONOMOUS_DAILY_CAP="1",
+    )
+    executor._auto_approve(config, sess)
+    assert high.status is ProposalStatus.APPROVED
+    assert low.status is ProposalStatus.PROPOSED
+
+
 def test_run_executes_only_approved(cfg, sess):
     approved = _proposal(title="Approved item")
     proposed = _proposal(title="Proposed item")

tests/test_maild.py

@@ -0,0 +1,351 @@
+"""maild tests — parsing, classification, threading, gated send. Invented data only."""
+
+from __future__ import annotations
+
+from types import SimpleNamespace
+
+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>"
+
+STUB_CFG = SimpleNamespace(mail_from="agent@cayce-co.xyz")
+
+
+def _eml(subject, body, in_reply_to=None, msgid=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:
+    """A proposal whose first-contact mail went through the REAL path:
+    compose -> approve -> send (creating a true OUTBOUND row)."""
+    p = Proposal(
+        kind="partnership",
+        title="Acme hotel affiliate program",
+        status=ProposalStatus.APPROVED,
+    )
+    sess.add(p)
+    sess.flush()
+    mail_id = maild._compose(
+        p.id, "partners@acme.example.org", "[cayce] partnership inquiry", "Hello!", sess
+    )
+    assert maild._approve_draft(mail_id, "gustavo", sess) == 0
+    assert maild._send_one(STUB_CFG, sess, mail_id, sender=lambda *a, **k: None) == "sent"
+    out = (
+        sess.query(MailMessage)
+        .filter_by(direction=MailDirection.OUTBOUND, proposal_id=p.id)
+        .one()
+    )
+    out.message_id = OUTBOUND_MSGID  # deterministic id for the fixtures
+    sess.flush()
+    return p
+
+
+def _fetcher(raws: list[bytes]):
+    return lambda cfg: [(bytes(str(i), "ascii"), raw) for i, raw in enumerate(raws)]
+
+
+# --- 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, _ = clf.classify("grocery list", config)
+    assert kind is MailClassification.OTHER or kind is MailClassification.NEEDS_REVIEW
+
+
+def test_negated_approvals_are_rejections(cfg):
+    """Critical finding 4: 'cannot accept'/'not approved' read as approvals."""
+    config = cfg()
+    cases = [
+        "We cannot accept your application",
+        "We are not accepting new partners at this time",
+        "Your application was not approved",
+        "We can't approve this request",
+        "We do not accept external affiliates",
+    ]
+    for text in cases:
+        kind, confidence = clf.classify(text, config)
+        assert kind is MailClassification.REJECTION, f"{text!r} -> {kind}"
+        assert confidence >= 0.9
+
+
+def test_llm_prompt_format_survives_json_braces():
+    """Critical finding 4: .format() raised KeyError on the JSON sample."""
+    formatted = clf._LLM_PROMPT.format(body="x")
+    assert '{"classification"' in formatted  # literal braces survive
+
+
+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, nums = maild._poll(cfg(), sess, fetcher=_fetcher([raw]))
+    assert stats["processed"] == 1 and stats["drafted"] == 1
+    assert len(nums) == 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)
+
+    first, second = maild._poll(cfg(), sess, fetcher=_fetcher([raw])), maild._poll(
+        cfg(), sess, fetcher=_fetcher([raw])
+    )
+    assert first[0]["processed"] == 1
+    assert second[0]["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=_fetcher([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=_fetcher([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):
+    _proposal_with_sent_draft(sess)
+    raw = _eml("Re: inquiry", "Approved, welcome to the program.", in_reply_to=OUTBOUND_MSGID)
+    maild._poll(cfg(), sess, fetcher=_fetcher([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 stats["blocked"] + stats["missing"] >= 0  # no crash paths
+    assert config.mail_from == "agent@cayce-co.xyz"
+    row = sess.query(MailMessage).filter_by(direction=MailDirection.INBOUND).one()
+    assert row.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=_fetcher([raw]))
+    inbound = sess.query(MailMessage).filter_by(direction=MailDirection.INBOUND).one()
+    original_id = inbound.message_id
+
+    assert maild._approve_draft(inbound.id, "gustavo", sess) == 0
+
+    sent = []
+    config = cfg(CAYCE_OPS_MAIL_FROM="agent@cayce-co.xyz")
+
+    def sender(cfg, to, subject, body, message_id, in_reply_to=None):
+        sent.append((to, subject, body, message_id, in_reply_to))
+
+    stats = maild._send_approved(config, sess, sender=sender)
+    assert stats["sent"] == 1
+    assert inbound.draft_status is DraftStatus.SENT
+    to, subject, body, message_id, in_reply_to = sent[0]
+    assert to == "partners@acme.example.org"
+    assert in_reply_to == original_id  # In-Reply-To threads to the inbound mail
+
+    # a real OUTBOUND row now exists (audit fix: they were never created)
+    out = (
+        sess.query(MailMessage)
+        .filter_by(direction=MailDirection.OUTBOUND, message_id=message_id)
+        .one()
+    )
+    assert out.in_reply_to == original_id
+    assert out.proposal_id == inbound.proposal_id
+    assert inbound.message_id == original_id  # inbound id NOT overwritten
+
+    row = (
+        sess.query(AuditEntry)
+        .filter_by(action="send_draft")
+        .order_by(AuditEntry.id.desc())
+        .first()
+    )
+    assert row is not None and "gustavo" in str(row.detail)
+
+
+def test_double_approve_is_rejected(cfg, sess):
+    _proposal_with_sent_draft(sess)
+    raw = _eml("Re: inquiry", "Approved, welcome.", in_reply_to=OUTBOUND_MSGID)
+    maild._poll(cfg(), sess, fetcher=_fetcher([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
+
+
+# --- audit corrections ----------------------------------------------------------
+
+
+def test_send_failure_does_not_resurrect_sent_mails(cfg, sess):
+    """Critical finding 3: SMTP failure mid-batch rolled back the whole
+    transaction, re-sending already-delivered mails 5 minutes later."""
+    config = cfg(CAYCE_OPS_MAIL_FROM="agent@cayce-co.xyz")
+    for i in range(3):
+        # unique first-contact thread per proposal (real compose -> send path)
+        p = Proposal(kind="partnership", title=f"Acme partner {i}")
+        sess.add(p)
+        sess.flush()
+        mail_id = maild._compose(
+            p.id, f"desk{i}@acme.example.org", f"[cayce] inquiry {i}", "Hello!", sess
+        )
+        assert maild._approve_draft(mail_id, "gustavo", sess) == 0
+        assert maild._send_one(STUB_CFG, sess, mail_id, sender=lambda *a, **k: None) == "sent"
+        sess.flush()
+        out = sess.get(MailMessage, mail_id)
+        raw = _eml(
+            f"Re: {i}",
+            "Approved, welcome to the program.",
+            in_reply_to=out.message_id,
+            msgid=f"<acme-{i}@acme.example.org>",
+        )
+        maild._poll(config, sess, fetcher=_fetcher([raw]))
+        inbound = (
+            sess.query(MailMessage)
+            .filter_by(direction=MailDirection.INBOUND, message_id=f"<acme-{i}@acme.example.org>")
+            .one()
+        )
+        maild._approve_draft(inbound.id, "gustavo", sess)
+    sess.commit()  # per-message factory sessions only see committed rows
+
+    calls = {"n": 0}
+
+    def flaky_smtp(cfg, to, subject, body, message_id, in_reply_to=None):
+        calls["n"] += 1
+        if calls["n"] == 3:
+            raise OSError("smtp connection lost")
+
+    import cayceops.db as appdb
+
+    stats = maild._send_approved(
+        config, sess, sender=flaky_smtp, factory=appdb.session_factory()
+    )
+
+    assert calls["n"] == 3
+    assert stats["sent"] == 2 and stats["error"] == 1  # no rollback of 1..2
+    still_approved = (
+        sess.query(MailMessage)
+        .filter_by(direction=MailDirection.INBOUND, draft_status=DraftStatus.APPROVED)
+        .count()
+    )
+    assert still_approved == 1  # only the failed one stays queued
+
+
+def test_edited_draft_refuses_to_send(cfg, sess):
+    """Medium finding: approved text was not bound to its content."""
+    _proposal_with_sent_draft(sess)
+    raw = _eml("Re: inquiry", "Approved, welcome to the program.", in_reply_to=OUTBOUND_MSGID)
+    maild._poll(cfg(), sess, fetcher=_fetcher([raw]))
+    inbound = sess.query(MailMessage).filter_by(direction=MailDirection.INBOUND).one()
+    assert maild._approve_draft(inbound.id, "gustavo", sess) == 0
+
+    inbound.draft_reply = "TOTALLY EDITED TEXT after approval"  # tamper
+
+    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 stats["hash_mismatch"] == 1
+    assert sent == []
+    assert inbound.draft_status is DraftStatus.DRAFTED  # needs fresh approval
+    action = sess.query(AuditEntry).filter_by(action="send_refused_hash_mismatch").one()
+    assert action.result == "error"
+
+
+def test_compose_creates_first_contact_draft(cfg, sess):
+    """Critical finding 2: there was no path for the first contact mail."""
+    p = Proposal(kind="partnership", title="New hotel partner")
+    sess.add(p)
+    sess.flush()
+
+    mail_id = maild._compose(
+        p.id, "desk@hotel.example.org", "Partnership", "We would like to...", sess
+    )
+    assert mail_id > 0
+    out = sess.get(MailMessage, mail_id)
+    assert out.direction is MailDirection.OUTBOUND
+    assert out.draft_status is DraftStatus.DRAFTED
+    assert out.proposal_id == p.id
+
+    # approve + send -> an inbound reply can now thread to this proposal
+    assert maild._approve_draft(mail_id, "gustavo", sess) == 0
+    stats = maild._send_approved(STUB_CFG, sess, sender=lambda *a, **k: None)
+    assert stats["sent"] == 1
+    sess.flush()
+
+    reply = _eml("Re: Partnership", "Approved, welcome!", in_reply_to=out.message_id)
+    assert maild._match_proposal(maild.parse_raw(reply), sess) is p
+
+
+def test_mark_seen_returned_for_post_commit_processing(cfg, sess):
+    """Medium finding: \\\\Seen was set before processing (crash lost mail)."""
+    raw = _eml("Re: hi", "Approved, welcome to the program.")
+    stats, nums = maild._poll(cfg(), sess, fetcher=_fetcher([raw]))
+    assert stats["processed"] == 1
+    assert nums == [b"0"]  # caller marks seen only after commit

tests/test_models.py

@@ -90,3 +90,23 @@ def test_mail_and_content_and_runs(sess):
     assert m.id and c.id and r.id
     assert m.draft_status == "drafted"
     assert c.status == "draft"
+
+
+def test_audit_chain_and_verification(sess):
+    from cayceops.models import verify_audit_chain
+
+    p = Proposal(kind="other", title="chain")
+    sess.add(p)
+    sess.flush()
+    for i in range(3):
+        audit(sess, actor=f"actor-{i}", action="step", proposal_id=p.id, detail={"i": i})
+    sess.flush()
+
+    ok, broken = verify_audit_chain(sess)
+    assert ok and broken is None
+
+    # tamper with one entry -> chain breaks at that id
+    first = sess.query(AuditEntry).order_by(AuditEntry.id).first()
+    first.actor = "tampered"
+    ok, broken = verify_audit_chain(sess)
+    assert not ok and broken == first.id

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,224 @@
+"""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]  # noqa: E501
+
+
+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_terminal(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.SKIPPED  # terminal (audit fix)
+    skip = sess.query(AuditEntry).filter_by(action="publish_skipped_terminal").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")
+
+
+# --- audit corrections ----------------------------------------------------------
+
+
+def test_publish_refuses_tampered_caption(cfg, sess):
+    """Medium finding: approved caption was not bound to content."""
+    item = _item(status=ContentStatus.DRAFT)
+    sess.add(item)
+    sess.flush()
+    assert publisher._approve(item.id, "gustavo", sess) == 0
+
+    item.caption = item.caption + " TAMPERED"
+    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 and stats["tampered"] == 1
+    assert item.status is ContentStatus.APPROVED  # stays, needs re-approval
+    action = sess.query(AuditEntry).filter_by(action="publish_refused_hash_mismatch").one()
+    assert action.result == "error"
+
+
+def test_disabled_channel_is_terminal(cfg, sess):
+    """Medium finding: disabled channels wrote skip audits every 5 minutes."""
+    item = _item(channel="facebook", theme="lightning-abroad")
+    sess.add(item)
+    sess.flush()
+    config = cfg()
+
+    stats1 = publisher._publish_due(config, sess)
+    assert stats1["skipped"] == 1
+    assert item.status is ContentStatus.SKIPPED  # terminal
+
+    stats2 = publisher._publish_due(config, sess)
+    assert stats2["skipped"] == 0  # never revisited
+    skips = sess.query(AuditEntry).filter_by(action="publish_skipped_terminal").count()
+    assert skips == 1
+
+
+def test_publish_attempts_bounded(cfg, sess, monkeypatch):
+    """Medium finding: nostr failures retried without limit."""
+    from cayceops.components.publisher import nostr as nostr_mod
+
+    item = _item()
+    sess.add(item)
+    sess.flush()
+
+    def always_fail(relays, event, timeout=10.0):
+        raise RuntimeError("relay down")
+
+    monkeypatch.setattr(nostr_mod, "publish", always_fail)
+    config = cfg(
+        CAYCE_OPS_NOSTR_PRIVATE_KEY=TEST_KEY,
+        CAYCE_OPS_NOSTR_RELAYS="wss://r.example.org",
+        CAYCE_OPS_MAX_ATTEMPTS="1",
+    )
+
+    publisher._publish_due(config, sess, nostr_sender=always_fail)
+    assert item.attempts == 1
+    assert item.status is ContentStatus.SKIPPED  # budget exhausted -> terminal

tests/test_scanner.py

@@ -58,7 +58,7 @@ def test_scan_files_proposals_with_kind_and_score(cfg, sess):
     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.match_score > 0
     assert affiliate.dedupe_key
 
     partner = by_title["Become a payment partner"]