Compare commits

...
Author SHA1 Message Date
cayce aabd6bafc7 publisher: themed calendar, NIP-01 nostr adapter, gated brand images
ci / lint-test-leaks (pull_request) Canceled after 0s
- calendar: one draft per theme/channel (idempotent), rotated schedule
- human gate: approve() audited; AUTONOMOUS_MODE has no effect here by design
- nostr (primary): in-repo NIP-01 client (D4) — canonical serialization,
  sha256 id, BIP-340 schnorr sign via coincurve with zero aux randomness
  (deterministic), pure-python reference verification before publish,
  per-relay publish with isolated errors, event id stored as post_ref
- facebook/instagram adapters: exist but refuse to run unless explicitly
  enabled via env (default deployment never touches Meta)
- images: Pillow brand templates (bh gold / hc orange), 1080x1350,
  template-driven typeset design (not AI-generated media)
- failure isolation: config errors audited, item stays approved

tests: 12 new (event roundtrip+tamper, determinism, key validation,
calendar idempotence, approve gate, nostr happy path, future schedule,
clean config failure, disabled meta adapters, image render) — 41 total
2026-10-19 21:54:39 +00:00
cayce 15072a0cd2 Merge PR #3: mail-negotiator
ci / lint-test-leaks (push) Canceled after 0s
IMAP classification (rules + optional LLM), conservative needs_review,
reply drafts gated on explicit human approval, SMTP approved-only,
message-id threading, full audit chain. 29 tests green.
2026-10-19 21:47:29 +00:00
cayce 5b259bc663 mail-negotiator: classify merchant replies, draft gated, audited sends
ci / lint-test-leaks (pull_request) Canceled after 0s
- IMAP poll (UNSEEN, stdlib imaplib) -> parse (bare address via email.utils)
- threading: In-Reply-To -> outbound sent-draft message-id -> proposal
- classify: deterministic rules; optional LLM behind provider interface;
  below-threshold confidence -> needs_review (never guessed)
- unthreadable approvals flagged needs_review (suspicious by definition)
- merchant_status recorded on proposal detail; proposal lifecycle untouched
- reply drafts generated from templates (drafted), NEVER auto-sent
- SMTP sends only draft_status=approved (human approve_draft, audited);
  sent drafts store their Message-ID for future threading
- idempotent on Message-ID; DaemonRun wrapper for monitor

tests: 9 new (classification rules, threshold->review, threading, idempotence,
blocked send without approval, approve->send->audit chain, double approve)
— 29 total passing
2026-10-19 21:47:18 +00:00
cayce 1b2c3cc1cf Merge PR #2: scanner + executor
ci / lint-test-leaks (push) Canceled after 0s
Discovery (read-only) + approval-gated execution with plugin registry,
audited autonomy policy, retry-then-fail. Gate: ruff clean, 20 tests green,
migration smoke ok, no leaks (full history).
2026-10-19 21:42:41 +00:00
10 changed files with 1319 additions and 0 deletions
+1
View File
@@ -0,0 +1 @@
"""maild package."""
+68
View File
@@ -0,0 +1,68 @@
"""maild classification: rules first, LLM optional, conservative threshold.
The LLM never decides alone: any result below ``CAYCE_OPS_LLM_REVIEW_THRESHOLD``
confidence becomes ``needs_review``. With the ``null`` provider (default) only
deterministic rules run.
"""
from __future__ import annotations
import json
import re
from cayceops.config import Config
from cayceops.models import MailClassification
_RULES: list[tuple[MailClassification, re.Pattern[str], float]] = [
# (class, pattern, confidence when matched)
(MailClassification.REJECTION, re.compile(r"\b(declin|reject|not able to|unable to|unfortunately|we regret|won.t be able)\w*", re.I), 0.9),
(MailClassification.APPROVAL, re.compile(r"\b(approv|accept|welcome to the|you.re (in|approved)|live|activated|glad to)\w*", re.I), 0.85),
(MailClassification.QUESTION, re.compile(r"\b(could you|can you|please (provide|clarify|confirm)|what (is|are)|how (do|does|can)|need more (info|details))\b", re.I), 0.7),
]
_LLM_PROMPT = """Classify this merchant email into exactly one of:
approval, rejection, question, other.
Reply with strict JSON: {"classification": "...", "confidence": 0.0-1.0}
Subject and body:
{body}"""
def _rules(text: str) -> tuple[MailClassification, float]:
for kind, pattern, confidence in _RULES:
if pattern.search(text):
return kind, confidence
return MailClassification.OTHER, 0.3
def _llm_classify(text: str, cfg: Config) -> tuple[MailClassification, float] | None:
"""Ask the configured LLM. Returns None on any failure (rules take over)."""
from cayceops.llm import get_provider
try:
provider = get_provider(cfg)
raw = provider.complete(_LLM_PROMPT.format(body=text[:4000]), system="You classify merchant emails. Reply with JSON only.")
data = json.loads(re.search(r"\{.*\}", raw, re.S).group(0))
kind = MailClassification(data["classification"].lower())
confidence = float(data["confidence"])
except Exception: # noqa: BLE001 — LLM is best-effort by design
return None
if kind not in MailClassification or kind is MailClassification.NEEDS_REVIEW:
return None
return kind, min(max(confidence, 0.0), 1.0)
def classify(text: str, cfg: Config) -> tuple[MailClassification, float]:
"""Classify merchant text. Degrades to rules when no LLM is configured."""
if cfg.llm_provider != "null":
llm_result = _llm_classify(text, cfg)
if llm_result is not None:
kind, confidence = llm_result
else: # LLM failed — fall through to rules
kind, confidence = _rules(text)
else:
kind, confidence = _rules(text)
if confidence < cfg.llm_review_threshold:
return MailClassification.NEEDS_REVIEW, confidence
return kind, confidence
+364
View File
@@ -0,0 +1,364 @@
"""maild — classify merchant replies, draft responses, gated SMTP.
Flow
----
1. ``poll``: fetch unread inbound mail (IMAP), parse, thread to the proposal
via ``In-Reply-To`` of a previously-sent draft, classify (rules + optional
LLM), store, and create a REPLY DRAFT (never sent automatically).
2. ``approve_draft``: a human approves a specific draft (audited).
3. ``send_approved``: SMTP sends *only* drafts in state ``approved``.
Classification is conservative: anything below the confidence threshold becomes
``needs_review`` and no draft is generated.
Public signatures match the CLI: ``poll()``, ``approve_draft(id, actor)``,
``send_approved()``. The ``_*`` variants take an explicit session/fetcher for
tests and composition.
"""
from __future__ import annotations
import email
import email.policy
import imaplib
import smtplib
from datetime import UTC, datetime
from email.message import EmailMessage
from cayceops.components.maild.classify import classify as _classify_text
from cayceops.config import Config
from cayceops.db import session_scope
from cayceops.models import (
DaemonRun,
DraftStatus,
MailClassification,
MailDirection,
MailMessage,
Proposal,
audit,
)
# --- IMAP ---------------------------------------------------------------------
def connect_imap(cfg: Config) -> imaplib.IMAP4_SSL:
client = imaplib.IMAP4_SSL(cfg.imap_host, cfg.imap_port)
client.login(cfg.imap_user, cfg.imap_password)
client.select(cfg.imap_mailbox)
return client
def _fetch_unread(cfg: Config) -> list[bytes]:
"""Fetch raw unread messages. Returns list of raw RFC822 bytes."""
client = connect_imap(cfg)
try:
_, data = client.search(None, "UNSEEN")
raw_messages: list[bytes] = []
for num in data[0].split():
_, msg_data = client.fetch(num, "(RFC822)")
raw_messages.append(msg_data[0][1])
client.store(num, "+FLAGS", "\\Seen")
return raw_messages
finally:
client.logout()
# --- parsing ------------------------------------------------------------------
def parse_raw(raw: bytes) -> dict:
"""Parse raw RFC822 bytes into our flat dict shape."""
msg = email.message_from_bytes(raw, policy=email.policy.default)
body = ""
if msg.is_multipart():
for part in msg.walk():
if part.get_content_type() == "text/plain" and not part.get_filename():
body = part.get_content()
break
else:
body = msg.get_content()
reply_to = msg.get("In-Reply-To") or msg.get("References") or ""
message_id = (msg.get("Message-ID") or "").strip()
if not message_id:
message_id = f"<generated-{datetime.now(UTC).timestamp()}@localhost>"
# store the bare address (envelope-safe); display names stay out of the DB
from_addr = email.utils.parseaddr(msg.get("From") or "")[1].strip()[:320]
return {
"message_id": message_id[:998],
"in_reply_to": reply_to.split()[-1].strip() if reply_to.strip() else None,
"from_addr": from_addr,
"subject": (msg.get("Subject") or "").strip()[:998],
"body": body[:20000],
}
# --- proposal threading -------------------------------------------------------
def _match_proposal(parsed: dict, sess) -> Proposal | None:
"""Thread an inbound mail to its proposal via In-Reply-To of sent drafts."""
irt = parsed.get("in_reply_to")
if not irt:
return None
sent = (
sess.query(MailMessage)
.filter(
MailMessage.direction == MailDirection.OUTBOUND,
MailMessage.message_id == irt,
)
.one_or_none()
)
if sent is None or sent.proposal_id is None:
return None
return sess.get(Proposal, sent.proposal_id)
# --- drafts -------------------------------------------------------------------
_DRAFT_TEMPLATES = {
MailClassification.APPROVAL: (
"Hello,\n\nThank you for the approval regarding \"{title}\". "
"We will proceed on our side and follow up with the next steps.\n\nBest regards,\nCayce"
),
MailClassification.REJECTION: (
"Hello,\n\nThank you for considering \"{title}\". "
"We appreciate the reply and would welcome the chance to revisit in the future.\n\n"
"Best regards,\nCayce"
),
MailClassification.QUESTION: (
"Hello,\n\nRegarding \"{title}\": {question}\n\n"
"Could you help us with the details so we can move forward?\n\nBest regards,\nCayce"
),
}
def _build_draft(parsed: dict, kind: MailClassification, proposal: Proposal | None) -> str:
title = proposal.title if proposal else (parsed.get("subject") or "your message")
template = _DRAFT_TEMPLATES.get(kind)
if template is None:
return ""
question = (parsed.get("body") or "")[:200].strip() or "we had a question"
return template.format(title=title[:120], question=question)
# --- inbound pass ---------------------------------------------------------------
def _poll(cfg: Config, sess, fetcher=None) -> dict:
"""One inbound pass: parse, classify, thread, draft. Returns stats."""
stats = {"processed": 0, "drafted": 0, "needs_review": 0, "skipped": 0}
for raw in (fetcher or _fetch_unread)(cfg):
parsed = parse_raw(raw)
if (
sess.query(MailMessage)
.filter(MailMessage.message_id == parsed["message_id"])
.one_or_none()
):
stats["skipped"] += 1
continue
proposal = _match_proposal(parsed, sess)
kind, confidence = _classify_text(f"{parsed['subject']}\n{parsed['body']}", cfg)
if proposal is None and kind == MailClassification.APPROVAL:
# an approval we cannot thread is suspicious — human must look
kind = MailClassification.NEEDS_REVIEW
confidence = min(confidence, 0.4)
record = MailMessage(
direction=MailDirection.INBOUND,
message_id=parsed["message_id"],
in_reply_to=parsed["in_reply_to"],
from_addr=parsed["from_addr"] or None,
subject=parsed["subject"] or None,
body=parsed["body"],
classification=kind,
confidence=confidence,
proposal_id=proposal.id if proposal else None,
processed_at=datetime.now(UTC),
)
sess.add(record)
sess.flush()
audit(
sess,
actor="daemon:maild",
action="classify",
proposal_id=proposal.id if proposal else None,
detail={"classification": kind.value, "confidence": confidence},
)
stats["processed"] += 1
if kind == MailClassification.NEEDS_REVIEW:
stats["needs_review"] += 1
continue
if proposal is not None:
proposal.detail = {
**(proposal.detail or {}),
"merchant_status": kind.value,
"merchant_reply_at": datetime.now(UTC).isoformat(),
}
draft = _build_draft(parsed, kind, proposal)
if draft:
record.draft_reply = draft
record.draft_status = DraftStatus.DRAFTED
audit(
sess,
actor="daemon:maild",
action="draft_reply",
proposal_id=proposal.id if proposal else None,
detail={"mail_id": record.id},
)
stats["drafted"] += 1
return stats
# --- human gate: approve draft -------------------------------------------------
def _approve_draft(mail_id: int, actor: str, sess) -> int:
"""Human approves a drafted reply. Only then may SMTP send it."""
record = sess.get(MailMessage, mail_id)
if record is None or record.direction != MailDirection.INBOUND:
return 1
if record.draft_status != DraftStatus.DRAFTED:
audit(
sess,
actor=actor,
action="approve_draft",
proposal_id=record.proposal_id,
result="error",
detail={"reason": f"invalid draft state: {record.draft_status.value}"},
)
return 1
record.draft_status = DraftStatus.APPROVED
audit(
sess,
actor=actor,
action="approve_draft",
proposal_id=record.proposal_id,
detail={"mail_id": record.id},
)
return 0
# --- SMTP: only approved drafts ------------------------------------------------
def _smtp_send(cfg: Config, to_addr: str, subject: str, body: str, message_id: str) -> None:
msg = EmailMessage()
msg["From"] = cfg.mail_from
msg["To"] = to_addr
msg["Subject"] = subject
msg["Message-ID"] = message_id
msg.set_content(body)
with smtplib.SMTP(cfg.smtp_host, cfg.smtp_port, timeout=30) as server:
server.starttls()
server.login(cfg.smtp_user, cfg.smtp_password)
server.send_message(msg)
def _approved_by(sess, mail_id: int) -> str:
from cayceops.models import AuditEntry
for entry in sess.query(AuditEntry).all():
if entry.action == "approve_draft" and entry.detail.get("mail_id") == mail_id:
return entry.actor
return "unknown"
def _send_approved(cfg: Config, sess, sender=None) -> dict:
"""Send every APPROVED draft. Anything else stays untouched."""
stats = {"sent": 0, "blocked": 0}
pending = (
sess.query(MailMessage)
.filter(
MailMessage.direction == MailDirection.INBOUND,
MailMessage.draft_status == DraftStatus.APPROVED,
)
.all()
)
for record in pending:
to_addr = record.from_addr or ""
if not to_addr:
record.draft_status = DraftStatus.SKIPPED
stats["blocked"] += 1
continue
message_id = (
f"<cayce-draft-{record.id}-{int(datetime.now(UTC).timestamp())}@cayce-co.xyz>"
)
(sender or _smtp_send)(
cfg,
to_addr,
f"Re: {record.subject or 'your message'}",
record.draft_reply or "",
message_id,
)
approved_by = _approved_by(sess, record.id)
record.draft_status = DraftStatus.SENT
# store the Message-ID so future replies thread to this draft
record.message_id = message_id
audit(
sess,
actor="daemon:maild",
action="send_draft",
proposal_id=record.proposal_id,
detail={
"mail_id": record.id,
"message_id": message_id,
"approved_by": approved_by,
},
)
stats["sent"] += 1
return stats
# --- CLI wrappers (session scope lives here) ------------------------------------
def poll(cfg: Config | None = None) -> int:
from cayceops.config import load
from cayceops.db import engine
cfg = cfg or load()
try:
with session_scope() as sess:
run_row = DaemonRun(daemon="maild")
sess.add(run_row)
sess.flush()
try:
stats = _poll(cfg, sess)
except Exception as exc: # noqa: BLE001 — daemon boundary
run_row.status = "error"
run_row.detail = {"error": str(exc)[:300]}
run_row.finished_at = datetime.now(UTC)
return 1
run_row.status = "ok"
run_row.detail = stats
run_row.finished_at = datetime.now(UTC)
return 0
finally:
engine().dispose()
def approve_draft(mail_id: int, actor: str) -> int:
from cayceops.db import engine
try:
with session_scope() as sess:
return _approve_draft(mail_id, actor, sess)
finally:
engine().dispose()
def send_approved(cfg: Config | None = None) -> int:
from cayceops.config import load
from cayceops.db import engine
cfg = cfg or load()
try:
with session_scope() as sess:
_send_approved(cfg, sess)
return 0
finally:
engine().dispose()
@@ -0,0 +1 @@
"""publisher package."""
+61
View File
@@ -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", "")}
+72
View File
@@ -0,0 +1,72 @@
"""Brand image templates (Pillow). 1080x1350 portrait, two brand styles.
Deterministic, template-driven graphics — deliberately not AI-generated media
(DECISIONS: nothing here trips AI-media disclosure requirements; it is typeset
design). Fonts: Pillow's bundled default (scalable), so no system font deps.
"""
from __future__ import annotations
import os
from PIL import Image, ImageDraw, ImageFont
WIDTH, HEIGHT = 1080, 1350
_BRANDS = {
# bitcoinhotels — warm gold on near-black
"bh": {"bg": (20, 18, 16), "accent": (201, 168, 76), "text": (242, 236, 224), "muted": (150, 140, 120)},
# hardcap — bitcoin orange on cool black
"hc": {"bg": (13, 13, 16), "accent": (247, 147, 26), "text": (250, 250, 250), "muted": (120, 120, 130)},
}
def _font(size: int, bold: bool = False):
# Pillow >=10: load_default(size=...) returns a scalable bundled font
return ImageFont.load_default(size=size)
def _vgrad(top: tuple, bottom: tuple) -> Image.Image:
"""Simple two-color vertical gradient."""
img = Image.new("RGB", (WIDTH, HEIGHT))
for y in range(HEIGHT):
t = y / (HEIGHT - 1)
row = tuple(int(top[i] + (bottom[i] - top[i]) * t) for i in range(3))
ImageDraw.Draw(img).line([(0, y), (WIDTH, y)], fill=row)
return img
def render_brand_image(
*,
brand: str = "bh",
kicker: str | None,
lines: list[str],
tagline: list[str] | None = None,
url: str | None = None,
out_path: str,
) -> str:
"""Render one branded poster. Returns out_path. Raises on unknown brand."""
if brand not in _BRANDS:
raise ValueError(f"unknown brand {brand!r} (have: {sorted(_BRANDS)})")
palette = _BRANDS[brand]
img = _vgrad(palette["bg"], tuple(max(0, c - 8) for c in palette["bg"]))
draw = ImageDraw.Draw(img)
x, y = 90, 240
if kicker:
draw.text((x, y), kicker, font=_font(40), fill=palette["muted"])
y += 110
for line in lines:
draw.text((x, y), line, font=_font(96, bold=True), fill=palette["text"])
y += 118
y += 24
for t in tagline or []:
draw.text((x, y), t, font=_font(38), fill=palette["muted"])
y += 58
draw.rectangle([x, y + 30, x + 470, y + 33], fill=palette["accent"])
if url:
draw.text((x, y + 110), url, font=_font(44), fill=palette["accent"])
os.makedirs(os.path.dirname(out_path) or ".", exist_ok=True)
img.save(out_path, "JPEG", quality=92)
return out_path
+166
View File
@@ -0,0 +1,166 @@
"""Minimal Nostr client: NIP-01 kind-1 notes (build, sign, verify, publish).
Implemented in-repo (DECISIONS D4) so every byte of the signing path is
auditable: event id = sha256 of the canonical NIP-01 serialization; signature =
BIP-340 Schnorr over the id. Keys are 32-byte hex. Relays are plain env config.
Publishing is fire-and-wait: open a websocket per relay, send the EVENT, wait
for the OK frame (or timeout), and return the per-relay results.
"""
from __future__ import annotations
import hashlib
import json
from typing import Any
from coincurve import PrivateKey, PublicKey
KIND_NOTE = 1
def normalize_key(private_key_hex: str) -> bytes:
"""Accept 64-char hex (with or without 0x). Anything else is a config error."""
key = private_key_hex.strip().lower().removeprefix("0x")
if len(key) != 64 or not all(c in "0123456789abcdef" for c in key):
raise ValueError("CAYCE_OPS_NOSTR_PRIVATE_KEY must be 64-char hex (32 bytes)")
return bytes.fromhex(key)
def pubkey_from_secret(secret: bytes) -> str:
return PublicKey.from_secret(secret).format().hex()[2:]
def build_event(
secret: bytes,
content: str,
tags: list[list[str]] | None = None,
created_at: int | None = None,
kind: int = KIND_NOTE,
) -> dict[str, Any]:
"""Build and sign a NIP-01 event. Deterministic given the same inputs."""
created_at = int(created_at if created_at is not None else __import__("time").time())
pubkey = pubkey_from_secret(secret)
serialized = json.dumps(
[0, pubkey, created_at, kind, tags or [], content],
separators=(",", ":"), # canonical NIP-01 form
ensure_ascii=False,
)
event_id = hashlib.sha256(serialized.encode("utf-8")).hexdigest()
# zero aux randomness -> fully deterministic signatures (BIP-340 permits it;
# verification is unaffected either way)
signature = (
PrivateKey(secret)
.sign_schnorr(bytes.fromhex(event_id), aux_randomness=b"\x00" * 32)
.hex()
)
return {
"pubkey": pubkey,
"created_at": created_at,
"kind": kind,
"tags": tags or [],
"content": content,
"id": event_id,
"sig": signature,
}
def _schnorr_verify(sig32: bytes, msg32: bytes, xonly_pub32: bytes) -> bool:
"""BIP-340 reference verification (pure Python, public domain spec code).
Signing runs in coincurve (libsecp256k1); verification is pure Python so
the auditor can read every step without CFFI indirection.
"""
P = 2**256 - 2**32 - 977
N = 0xFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFEBAAEDCE6AF48A03BBFD25E8CD0364141
def tag_hash(tag: str, msg: bytes) -> bytes:
t = hashlib.sha256(tag.encode()).digest()
return hashlib.sha256(t + t + msg).digest()
def lift_x(b: int):
if b >= P:
return None
y_sq = (pow(b, 3, P) + 7) % P
y = pow(y_sq, (P + 1) // 4, P)
if pow(y, 2, P) != y_sq:
return None
return (b, y if y % 2 == 0 else P - y)
def point_add(p1, p2):
if p1 is None:
return p2
if p2 is None:
return p1
if p1[0] == p2[0] and p1[1] != p2[1]:
return None
if p1 == p2:
lam = (3 * p1[0] * p1[0] * pow(2 * p1[1], P - 2, P)) % P
else:
lam = ((p2[1] - p1[1]) * pow(p2[0] - p1[0], P - 2, P)) % P
x3 = (lam * lam - p1[0] - p2[0]) % P
return (x3, (lam * (p1[0] - x3) - p1[1]) % P)
def point_mul(p, scalar):
acc = None
for i in range(256):
if (scalar >> i) & 1:
acc = point_add(acc, p)
p = point_add(p, p)
return acc
if len(sig32) != 64 or len(msg32) != 32 or len(xonly_pub32) != 32:
return False
r = int.from_bytes(sig32[:32], "big")
s = int.from_bytes(sig32[32:], "big")
if not (0 <= r < P and 0 <= s < N):
return False
pub_point = lift_x(int.from_bytes(xonly_pub32, "big"))
if pub_point is None:
return False
e = int.from_bytes(tag_hash("BIP0340/challenge", sig32[:32] + xonly_pub32 + msg32), "big") % N
r_point = point_add(point_mul((Gx := 0x79BE667EF9DCBBAC55A06295CE870B07029BFCDB2DCE28D959F2815B16F81798, Gy := 0x483ADA7726A3C4655DA4FBFC0E1108A8FD17B448A68554199C47D08FFB10D4B8), s), point_mul(pub_point, (N - e) % N))
if r_point is None or r_point[1] % 2 != 0:
return False
return r_point[0] == r
def verify_event(event: dict[str, Any]) -> bool:
"""Recompute the id and verify the Schnorr signature against the pubkey."""
serialized = json.dumps(
[0, event["pubkey"], event["created_at"], event["kind"], event["tags"], event["content"]],
separators=(",", ":"),
ensure_ascii=False,
)
if hashlib.sha256(serialized.encode("utf-8")).hexdigest() != event["id"]:
return False
compressed = PublicKey(bytes.fromhex("02" + event["pubkey"])).format() # 33-byte, prefix 02
return _schnorr_verify(
bytes.fromhex(event["sig"]), bytes.fromhex(event["id"]), compressed[1:]
)
def relay_publish(relay_url: str, event: dict[str, Any], timeout: float = 10.0) -> str:
"""Publish one event to one relay; return the relay's OK message."""
from websockets.sync.client import connect
with connect(relay_url, close_timeout=timeout) as ws:
ws.send(json.dumps(["EVENT", event]))
reply = json.loads(ws.recv(timeout=timeout))
if not (isinstance(reply, list) and len(reply) >= 3 and reply[0] == "OK" and reply[2] is True):
raise RuntimeError(f"relay {relay_url} rejected event: {str(reply)[:200]}")
return str(reply[1])
def publish(relays: list[str], event: dict[str, Any], timeout: float = 10.0) -> list[dict[str, str]]:
"""Publish to every relay; report per-relay outcome (never raises unless all fail)."""
results = []
for relay in relays:
try:
relay_publish(relay, event, timeout)
results.append({"relay": relay, "ok": "true"})
except Exception as exc: # noqa: BLE001 — per-relay isolation
results.append({"relay": relay, "ok": "false", "error": str(exc)[:200]})
if relays and all(r["ok"] == "false" for r in results):
raise RuntimeError(f"all relays failed: {results}")
return results
+266
View File
@@ -0,0 +1,266 @@
"""publisher — themed content calendar with a hard human gate on publishing.
State machine: draft -> approved -> published. Only ``approved`` items ever
publish. AUTONOMOUS_MODE has NO effect on this component: content always
requires explicit human approval (documented in AUDIT.md).
Channels: nostr (NIP-01, primary) — facebook/instagram adapters exist but are
disabled unless explicitly enabled via env (DECISIONS: default deployment never
touches Meta).
"""
from __future__ import annotations
from datetime import UTC, datetime, timedelta
from cayceops.components.publisher import images, nostr
from cayceops.config import Config
from cayceops.db import session_scope
from cayceops.models import ContentItem, ContentStatus, DaemonRun, audit
# rotation: one theme per slot; channel rotation nostr -> facebook -> instagram
THEMES = [
("lightning-abroad", "bitcoinhotels.org/guides/lightning-abroad"),
("seed-strategy", "hardcap.news/learn/travel-security-kit"),
("atm-fees", "bitcoinhotels.org/guides/best-bitcoin-atms-by-city"),
("elsalvador-budget", "bitcoinhotels.org/guides/living-on-sats-in-el-salvador"),
]
_CAPTION_TEMPLATE = "{hook}\n\n{body}\n\n{link}"
def _caption(theme: str, link: str) -> str:
hooks = {
"lightning-abroad": "Paying abroad with Lightning beats your bank.",
"seed-strategy": "The seed strategy you pick decides how the border story goes.",
"atm-fees": "The first ATM you find charges 10-15% over spot.",
"elsalvador-budget": "A real weekly budget on a Bitcoin standard.",
}
bodies = {
"lightning-abroad": "No foreign transaction fees, no dynamic currency conversion, no declines. Field-tested guide.",
"seed-strategy": "Steel plate, memorized + decoy, or split custody — the full breakdown, brutally honest.",
"atm-fees": "The prepared traveler pays 4-6%. The difference is knowing the operator and the no-KYC cap.",
"elsalvador-budget": "Surf hostels, comidas, coworking — the corners where you still need dollars, and the 90-day protocol.",
}
return _CAPTION_TEMPLATE.format(
hook=hooks.get(theme, theme.replace("-", " ").title()),
body=bodies.get(theme, "Field-tested notes."),
link=link,
)
# --- calendar planning ----------------------------------------------------------
def _plan(cfg: Config, sess, days: int = 4) -> dict:
"""Fill the calendar: one draft per theme/channel pair not already queued.
Idempotent by (theme, channel): existing draft/approved items block a
duplicate. Drafts are scheduled one per day, starting tomorrow.
"""
created = 0
for idx, (theme, link) in enumerate(THEMES[:days]):
channel = ["nostr", "facebook", "instagram", "nostr"][idx % 4]
exists = (
sess.query(ContentItem)
.filter(
ContentItem.theme == theme,
ContentItem.channel == channel,
ContentItem.status != ContentStatus.PUBLISHED,
)
.one_or_none()
)
if exists:
continue
item = ContentItem(
theme=theme,
channel=channel,
caption=_caption(theme, link),
link=link,
status=ContentStatus.DRAFT,
scheduled_for=datetime.now(UTC).replace(hour=13, minute=0, second=0, microsecond=0)
+ timedelta(days=idx + 1),
)
sess.add(item)
sess.flush()
audit(sess, actor="daemon:publisher", action="plan_draft", detail={"item": item.id, "theme": theme, "channel": channel})
created += 1
return {"created": created}
# --- human gate -------------------------------------------------------------------
def _approve(item_id: int, actor: str, sess) -> int:
item = sess.get(ContentItem, item_id)
if item is None or item.status != ContentStatus.DRAFT:
audit(
sess,
actor=actor,
action="approve_content",
result="error",
detail={"reason": f"item {item_id} not in draft state"},
)
return 1
item.status = ContentStatus.APPROVED
audit(sess, actor=actor, action="approve_content", detail={"item": item.id, "channel": item.channel})
return 0
# --- image rendering ----------------------------------------------------------------
def _render_image(item: ContentItem, cfg: Config) -> str | None:
"""Render the brand image for a nostr item (per its theme)."""
brand = "hc" if "hardcap" in (item.link or "") else "bh"
title = item.theme.replace("-", " ").upper().split()
lines = [" ".join(title[:2])] if title else [item.theme]
out = images.render_brand_image(
brand=brand,
kicker="C A Y C E",
lines=lines,
tagline=[(item.link or "")[:70]],
url=(item.link or "")[:60],
out_path=f"{cfg.media_dir}/item-{item.id}.jpg",
)
item.media_path = out
return out
# --- publishing ----------------------------------------------------------------------
def _publish_nostr(item: ContentItem, cfg: Config, sess, sender=None) -> str:
if not cfg.nostr_private_key or not cfg.nostr_relays:
raise RuntimeError("nostr channel requires CAYCE_OPS_NOSTR_PRIVATE_KEY and CAYCE_OPS_NOSTR_RELAYS")
secret = nostr.normalize_key(cfg.nostr_private_key)
event = nostr.build_event(secret, item.caption, tags=[["t", item.theme]])
if not nostr.verify_event(event): # pragma: no cover — build/verify symmetry
raise RuntimeError("event failed self-verification before publish")
results = (sender or nostr.publish)(cfg.nostr_relays, event)
audit(
sess,
actor="daemon:publisher",
action="nostr_relay_results",
detail={"event": event["id"], "relays": results},
)
return event["id"]
def _publish_due(cfg: Config, sess, nostr_sender=None) -> dict:
"""Publish approved items that are due. Non-nostr channels are skipped
unless their adapters are explicitly enabled (default: skipped, audited)."""
stats = {"published": 0, "skipped": 0, "failed": 0}
due = (
sess.query(ContentItem)
.filter(ContentItem.status == ContentStatus.APPROVED)
.all()
)
now = datetime.now(UTC)
for item in due:
if item.scheduled_for and item.scheduled_for > now:
continue # not due yet
try:
if item.channel == "nostr":
item.post_ref = _publish_nostr(item, cfg, sess, sender=nostr_sender)
else:
audit(
sess,
actor="daemon:publisher",
action="publish_skipped",
detail={
"item": item.id,
"channel": item.channel,
"reason": f"{item.channel} adapter disabled by default",
},
)
stats["skipped"] += 1
continue
except Exception as exc: # noqa: BLE001 — publish boundary: audit and continue
item.error = str(exc)[:300]
audit(
sess,
actor="daemon:publisher",
action="publish_failed",
result="error",
detail={"item": item.id, "error": item.error},
)
stats["failed"] += 1
continue
item.status = ContentStatus.PUBLISHED
item.published_at = datetime.now(UTC)
audit(
sess,
actor="daemon:publisher",
action="publish_content",
detail={"item": item.id, "channel": item.channel, "ref": item.post_ref},
)
stats["published"] += 1
return stats
# --- CLI wrappers ---------------------------------------------------------------------
def plan(cfg: Config | None = None) -> int:
from cayceops.config import load
from cayceops.db import engine
cfg = cfg or load()
try:
with session_scope() as sess:
run_row = DaemonRun(daemon="publisher")
sess.add(run_row)
sess.flush()
stats = _plan(cfg, sess)
run_row.status = "ok"
run_row.detail = stats
run_row.finished_at = datetime.now(UTC)
return 0
finally:
engine().dispose()
def publish_due(cfg: Config | None = None) -> int:
from cayceops.config import load
from cayceops.db import engine
cfg = cfg or load()
try:
with session_scope() as sess:
run_row = DaemonRun(daemon="publisher")
sess.add(run_row)
sess.flush()
stats = _publish_due(cfg, sess)
run_row.status = "ok"
run_row.detail = stats
run_row.finished_at = datetime.now(UTC)
return 0
finally:
engine().dispose()
def make_image(item_id: int, cfg: Config | None = None) -> int:
from cayceops.config import load
from cayceops.db import engine
cfg = cfg or load()
try:
with session_scope() as sess:
item = sess.get(ContentItem, item_id)
if item is None:
return 1
print(_render_image(item, cfg))
return 0
finally:
engine().dispose()
def approve(item_id: int, actor: str) -> int:
from cayceops.db import engine
try:
with session_scope() as sess:
return _approve(item_id, actor, sess)
finally:
engine().dispose()
+158
View File
@@ -0,0 +1,158 @@
"""maild tests — parsing, classification, threading, gated send. Invented data only."""
from __future__ import annotations
from cayceops.components.maild import classify as clf
from cayceops.components.maild import maild
from cayceops.models import (
AuditEntry,
DraftStatus,
MailClassification,
MailDirection,
MailMessage,
Proposal,
ProposalStatus,
)
OUTBOUND_MSGID = "<cayce-draft-1-1000@cayce-co.xyz>"
def _eml(subject: str, body: str, in_reply_to: str | None = None, msgid: str | None = None) -> bytes:
lines = [
"From: Partner Desk <partners@acme.example.org>",
"To: agent@cayce-co.xyz",
f"Subject: {subject}",
f"Message-ID: {msgid or '<acme-reply-1@acme.example.org>'}",
]
if in_reply_to:
lines.append(f"In-Reply-To: {in_reply_to}")
lines += ["MIME-Version: 1.0", "Content-Type: text/plain; charset=utf-8", "", body]
return "\r\n".join(lines).encode()
def _proposal_with_sent_draft(sess) -> Proposal:
p = Proposal(kind="partnership", title="Acme hotel affiliate program", status=ProposalStatus.APPROVED)
sess.add(p)
sess.flush()
sess.add(
MailMessage(
direction=MailDirection.OUTBOUND,
message_id=OUTBOUND_MSGID,
to_addr="partners@acme.example.org",
subject="[cayce] partnership inquiry",
proposal_id=p.id,
draft_status=DraftStatus.SENT,
)
)
sess.flush()
return p
# --- classifier ---------------------------------------------------------------
def test_rules_classification(cfg):
config = cfg()
assert clf.classify("Your application was approved, welcome to the program", config)[0] is MailClassification.APPROVAL
assert clf.classify("Unfortunately we cannot accept new partners", config)[0] is MailClassification.REJECTION
assert clf.classify("Could you provide your traffic numbers?", config)[0] is MailClassification.QUESTION
kind, confidence = clf.classify("grocery list", config)
assert kind is MailClassification.OTHER or kind is MailClassification.NEEDS_REVIEW
def test_low_confidence_becomes_needs_review(cfg):
config = cfg(CAYCE_OPS_LLM_REVIEW_THRESHOLD="0.95")
kind, confidence = clf.classify("Your application was approved, welcome", config)
assert kind is MailClassification.NEEDS_REVIEW
assert confidence < 0.95
# --- inbound pass ---------------------------------------------------------------
def test_poll_threads_approval_and_drafts_reply(cfg, sess):
p = _proposal_with_sent_draft(sess)
raw = _eml("Re: [cayce] partnership inquiry", "Approved! Welcome to the Acme partner program.", in_reply_to=OUTBOUND_MSGID)
stats = maild._poll(cfg(), sess, fetcher=lambda cfg: [raw])
assert stats["processed"] == 1 and stats["drafted"] == 1
inbound = sess.query(MailMessage).filter_by(direction=MailDirection.INBOUND).one()
assert inbound.classification is MailClassification.APPROVAL
assert inbound.proposal_id == p.id
assert inbound.draft_status is DraftStatus.DRAFTED
assert inbound.draft_reply and "approval" in inbound.draft_reply.lower()
assert p.detail["merchant_status"] == "approval"
assert p.status is ProposalStatus.APPROVED # lifecycle untouched by maild
def test_poll_idempotent_on_message_id(cfg, sess):
_proposal_with_sent_draft(sess)
raw = _eml("Re: hi", "Approved, welcome to the program.", in_reply_to=OUTBOUND_MSGID)
fetcher = lambda cfg: [raw]
assert maild._poll(cfg(), sess, fetcher=fetcher)["processed"] == 1
assert maild._poll(cfg(), sess, fetcher=fetcher)["skipped"] == 1
def test_unthreaded_approval_needs_review(cfg, sess):
raw = _eml("Great news", "Approved! Welcome aboard.") # no In-Reply-To
stats = maild._poll(cfg(), sess, fetcher=lambda cfg: [raw])
inbound = sess.query(MailMessage).filter_by(direction=MailDirection.INBOUND).one()
assert inbound.classification is MailClassification.NEEDS_REVIEW
assert stats["needs_review"] == 1
assert inbound.draft_status is DraftStatus.NONE # no draft for review items
def test_question_gets_clarification_draft(cfg, sess):
_proposal_with_sent_draft(sess)
raw = _eml("Re: inquiry", "Could you provide your monthly traffic?", in_reply_to=OUTBOUND_MSGID)
maild._poll(cfg(), sess, fetcher=lambda cfg: [raw])
inbound = sess.query(MailMessage).filter_by(direction=MailDirection.INBOUND).one()
assert inbound.classification is MailClassification.QUESTION
assert inbound.draft_status is DraftStatus.DRAFTED
assert "traffic" in inbound.draft_reply
# --- human gate -------------------------------------------------------------------
def test_send_blocked_without_approval(cfg, sess, monkeypatch):
_proposal_with_sent_draft(sess)
raw = _eml("Re: inquiry", "Approved, welcome to the program.", in_reply_to=OUTBOUND_MSGID)
maild._poll(cfg(), sess, fetcher=lambda cfg: [raw])
sent = []
config = cfg(CAYCE_OPS_MAIL_FROM="agent@cayce-co.xyz")
stats = maild._send_approved(config, sess, sender=lambda *a, **k: sent.append(a))
assert stats["sent"] == 0 and sent == []
assert sess.query(MailMessage).filter_by(direction=MailDirection.INBOUND).one().draft_status is DraftStatus.DRAFTED
def test_send_requires_explicit_human_approval(cfg, sess):
_proposal_with_sent_draft(sess)
raw = _eml("Re: inquiry", "Approved, welcome to the program.", in_reply_to=OUTBOUND_MSGID)
maild._poll(cfg(), sess, fetcher=lambda cfg: [raw])
inbound = sess.query(MailMessage).filter_by(direction=MailDirection.INBOUND).one()
assert maild._approve_draft(inbound.id, "gustavo", sess) == 0
sent = []
config = cfg(CAYCE_OPS_MAIL_FROM="agent@cayce-co.xyz")
stats = maild._send_approved(config, sess, sender=lambda *a, **k: sent.append(a))
assert stats["sent"] == 1
assert inbound.draft_status is DraftStatus.SENT
assert sent[0][1] == "partners@acme.example.org"
row = sess.query(AuditEntry).filter_by(action="send_draft").one()
assert "gustavo" in str(row.detail)
def test_double_approve_is_rejected(cfg, sess):
_proposal_with_sent_draft(sess)
raw = _eml("Re: inquiry", "Approved, welcome to the program.", in_reply_to=OUTBOUND_MSGID)
maild._poll(cfg(), sess, fetcher=lambda cfg: [raw])
inbound = sess.query(MailMessage).filter_by(direction=MailDirection.INBOUND).one()
assert maild._approve_draft(inbound.id, "gustavo", sess) == 0
assert maild._approve_draft(inbound.id, "gustavo", sess) == 1
+162
View File
@@ -0,0 +1,162 @@
"""publisher tests — calendar, human gate, nostr publish, disabled adapters."""
from __future__ import annotations
import os
import pytest
from cayceops.components.publisher import adapters, nostr, publisher
from cayceops.models import AuditEntry, ContentItem, ContentStatus
# invented test key (never a real nostr key — allowlisted in .gitleaks.toml)
TEST_KEY = "11" * 32
def _item(channel="nostr", theme="atm-fees", scheduled=None, status=ContentStatus.APPROVED):
return ContentItem(
theme=theme,
channel=channel,
caption="The first ATM you find charges 10-15% over spot.\n\nbitcoinhotels.org/guides/atm",
link="bitcoinhotels.org/guides/atm",
status=status,
scheduled_for=scheduled,
)
# --- nostr client -------------------------------------------------------------
def test_event_build_verify_roundtrip():
event = nostr.build_event(bytes.fromhex(TEST_KEY), "gm", tags=[["t", "bitcoin"]], created_at=1700000000)
assert nostr.verify_event(event)
assert len(event["id"]) == 64 and len(event["sig"]) == 128
tampered = {**event, "content": "gm (edited)"}
assert not nostr.verify_event(tampered)
def test_event_is_deterministic():
a = nostr.build_event(bytes.fromhex(TEST_KEY), "gm", created_at=1700000000)
b = nostr.build_event(bytes.fromhex(TEST_KEY), "gm", created_at=1700000000)
assert a == b
def test_key_validation():
with pytest.raises(ValueError):
nostr.normalize_key("not-a-key")
assert nostr.normalize_key("0x" + TEST_KEY) == bytes.fromhex(TEST_KEY)
# --- calendar ------------------------------------------------------------------
def test_plan_creates_idempotent_drafts(cfg, sess):
config = cfg()
first = publisher._plan(config, sess)
second = publisher._plan(config, sess)
assert first["created"] == 4
assert second["created"] == 0 # same themes+channels already queued
def test_approve_gate(cfg, sess):
item = _item(status=ContentStatus.DRAFT)
sess.add(item)
sess.flush()
assert publisher._approve(item.id, "gustavo", sess) == 0
assert item.status is ContentStatus.APPROVED
assert publisher._approve(item.id, "gustavo", sess) == 1 # not draft anymore
actor = sess.query(AuditEntry).filter_by(action="approve_content").first().actor
assert actor == "gustavo"
# --- publishing ------------------------------------------------------------------
def test_publish_due_nostr_happy_path(cfg, sess):
item = _item() # approved, scheduled None -> due now
sess.add(item)
sess.flush()
relays_seen = []
def fake_sender(relays, event, timeout=10.0):
relays_seen.append(relays)
assert nostr.verify_event(event)
return [{"relay": r, "ok": "true"} for r in relays]
config = cfg(
CAYCE_OPS_NOSTR_PRIVATE_KEY=TEST_KEY,
CAYCE_OPS_NOSTR_RELAYS="wss://relay.example.org,wss://other.example.org",
)
stats = publisher._publish_due(config, sess, nostr_sender=fake_sender)
assert stats["published"] == 1
assert item.status is ContentStatus.PUBLISHED
assert item.post_ref and len(item.post_ref) == 64
assert item.published_at is not None
assert "wss://relay.example.org" in relays_seen[0]
def test_future_scheduled_item_not_published(cfg, sess):
from datetime import UTC, datetime, timedelta
item = _item(scheduled=datetime.now(UTC) + timedelta(hours=2))
sess.add(item)
sess.flush()
config = cfg(CAYCE_OPS_NOSTR_PRIVATE_KEY=TEST_KEY, CAYCE_OPS_NOSTR_RELAYS="wss://r.example.org")
stats = publisher._publish_due(config, sess, nostr_sender=lambda r, e, timeout=10: [])
assert stats["published"] == 0
assert item.status is ContentStatus.APPROVED
def test_nostr_without_config_fails_clean(cfg, sess):
item = _item()
sess.add(item)
sess.flush()
config = cfg() # no key, no relays
stats = publisher._publish_due(config, sess)
assert stats["failed"] == 1
assert item.status is ContentStatus.APPROVED # stays approved, error recorded
assert item.error and "nostr" in item.error
def test_non_nostr_channels_skipped_by_default(cfg, sess):
item = _item(channel="facebook", theme="lightning-abroad")
sess.add(item)
sess.flush()
stats = publisher._publish_due(cfg(), sess)
assert stats["skipped"] == 1
assert item.status is ContentStatus.APPROVED
skip = sess.query(AuditEntry).filter_by(action="publish_skipped").one()
assert skip.detail["channel"] == "facebook"
def test_meta_adapters_disabled_by_default(cfg):
config = cfg()
with pytest.raises(PermissionError):
adapters.post_facebook(config, "hello")
with pytest.raises(PermissionError):
adapters.post_instagram(config, "hello", "https://img.example.org/x.jpg")
# --- images ------------------------------------------------------------------
def test_brand_image_render(tmp_path, cfg):
config = cfg(CAYCE_OPS_MEDIA_DIR=str(tmp_path))
out = publisher._render_image(_item(), config)
assert os.path.exists(out)
from PIL import Image
with Image.open(out) as img:
assert img.size == (1080, 1350)
def test_unknown_brand_rejected():
from cayceops.components.publisher import images
with pytest.raises(ValueError):
images.render_brand_image(brand="xx", kicker=None, lines=["A"], out_path="/tmp/x.jpg")