Compare commits

...
12 Commits
Author SHA1 Message Date
cayce a282aedb94 lint: wrap long lines (E501), fix F841/F821/E731 — make check fully green
All 32 findings resolved: regex literals kept single-line with noqa
(pattern IS the reviewed artifact), HTML rows string-concatenated for
line length, walrus tuple in BIP-340 verify split into named generator,
unused pytest import dropped. No behavior changes; 44/44 tests pass.
Full-gate discipline restored: ruff exit 0.
2026-10-19 22:03:53 +00:00
cayce abbaf7e4e2 Merge PR #6: audit — AUDIT.md for operator review
ci / lint-test-leaks (push) Canceled after 0s
2026-10-19 21:58:47 +00:00
cayce 6d77b3eff6 audit: AUDIT.md for operator review
ci / lint-test-leaks (pull_request) Canceled after 0s
Component map + data flow, every external call (where/what/trigger),
human-gate vs AUTONOMOUS_MODE matrix, env+permissions per component,
pinned deps with purpose, known risks (8, each with mitigation),
test+stack instructions. Audited-commit hash lands in the issue.
2026-10-19 21:58:46 +00:00
cayce 3f07362b7a Merge PR #5: monitor
ci / lint-test-leaks (push) Canceled after 0s
Static status page with privacy assertions and full escaping. 44 tests green.
2026-10-19 21:57:32 +00:00
cayce f61372e28e monitor: static status page — health, queue, gates, audit tail
ci / lint-test-leaks (pull_request) Canceled after 0s
- self-contained HTML (no external assets/JS), everything html-escaped
- daemon health from daemon_runs (last status + last run per daemon)
- proposal queue counts by state; human-gate counters (mail drafts awaiting
  approval, content draft/approved)
- audit tail: actor/action/result only — no detail payloads, no addresses,
  no subjects, no content bodies (tests assert the negatives)
- DaemonRun bookkeeping like every other daemon

tests: 3 new (sections render, leak-negatives, XSS escape) — 44 total
2026-10-19 21:57:31 +00:00
cayce 25c7db3a31 Merge PR #4: publisher
ci / lint-test-leaks (push) Canceled after 0s
Themed calendar + NIP-01 nostr (deterministic schnorr, self-verify),
FB/IG adapters disabled by default, hard human gate. 41 tests green.
2026-10-19 21:54:48 +00:00
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
cayce 73ccdf7bdf scanner + executor: discovery files proposals; execution gated on approval
ci / lint-test-leaks (pull_request) Canceled after 0s
scanner (read-only):
- sources from CAYCE_OPS_SCANNER_SOURCES JSON (rss + page types), stdlib only
- keyword scoring -> risk_score; kind via heuristics (affiliate/partnership/content_gap)
- idempotent: sha1 dedupe_key; repeated scans skip existing
- every proposal audited (actor daemon:scanner); source errors recorded, never fatal

executor (human gate):
- approve/reject audited with operator identity; only proposed -> approved
- AUTONOMOUS_MODE=true adds exactly one path: policy auto-approval
  (kinds allowlist + daily cap, actor auto:policy) — DECISIONS D5
- plugin registry by kind; unknown kinds fall back to log_action (D9)
- retries with max_attempts, then failed; every transition audited
- http_post plugin inert unless CAYCE_OPS_ENABLE_HTTP_ACTIONS=true

tests: 13 new (dedupe, keyword filter, autonomy bounds, retry-then-fail,
blocked http_post, fallback plugin) — 20 total passing
2026-10-19 21:42:34 +00:00
cayce 00fa3777ab Merge PR #1: skeleton — structure, schema, CI, deploy
ci / lint-test-leaks (push) Canceled after 0s
Component wiring, ORM schema + state machines, env-only config, alembic 0001,
Typer CLI, CI (ruff/pytest/alembic/gitleaks), compose + systemd, DECISIONS D1-D9.
Gate: make check green (ruff 0 errors, 7 tests pass, migration smoke ok, no leaks).
2026-10-19 21:38:06 +00:00
21 changed files with 2581 additions and 0 deletions
+152
View File
@@ -0,0 +1,152 @@
# AUDIT — cayce-ops
Audited commit: the merge of PR #6 on `main` (see *Audited commit* at the bottom).
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: merge of PR #6 (`audit: AUDIT.md for operator review`) — hash printed
in the "Ready for audit" issue that links this file.
- Gate at that commit: ruff clean · 44/44 tests · alembic 0001 smoke OK ·
gitleaks `--all` no leaks.
+1
View File
@@ -0,0 +1 @@
"""executor package."""
+246
View File
@@ -0,0 +1,246 @@
"""executor — executes APPROVED proposals through registered action plugins.
Guarantees
----------
- Only ``approved`` proposals execute (``AUTONOMOUS_MODE=false`` default).
- With ``AUTONOMOUS_MODE=true`` the executor gains exactly one capability:
bounded auto-approval of ``proposed`` items whose ``kind`` is in
``CAYCE_OPS_AUTONOMOUS_KINDS``, capped by ``CAYCE_OPS_AUTONOMOUS_DAILY_CAP``,
recorded with actor ``auto:policy`` (DECISIONS D5).
- Every transition writes an ``audit_log`` row: who approved, what ran,
plugin output, success/failure.
- Unknown proposal kinds fall back to the ``log_action`` plugin (never a
network side effect). See DECISIONS D9.
"""
from __future__ import annotations
from collections.abc import Callable
from datetime import UTC, datetime
from typing import Any
from sqlalchemy import func, select
from cayceops.components.executor import plugins as plugin_registry
from cayceops.config import Config
from cayceops.db import session_scope
from cayceops.models import AuditEntry, DaemonRun, Proposal, ProposalStatus, audit
Actor = str
# --- human transitions -------------------------------------------------------
def approve(proposal_id: int, actor: Actor, cfg: Config | None = None, sess=None) -> int:
"""Human approves a proposal. Audited with the operator's identity."""
return _decide(proposal_id, actor, ProposalStatus.APPROVED, cfg, sess)
def reject(proposal_id: int, actor: Actor, cfg: Config | None = None, sess=None) -> int:
"""Human rejects a proposal. Audited with the operator's identity."""
return _decide(proposal_id, actor, ProposalStatus.REJECTED, cfg, sess)
def _decide(
proposal_id: int,
actor: Actor,
decision: ProposalStatus,
cfg: Config | None,
sess,
) -> int:
owned = False
if sess is None:
outer = session_scope()
sess = outer.__enter__()
owned = True
try:
proposal = sess.get(Proposal, proposal_id)
if proposal is None:
return 1
if proposal.status != ProposalStatus.PROPOSED:
verb = "approve" if decision == ProposalStatus.APPROVED else "reject"
audit(
sess,
actor=actor,
action=verb,
proposal_id=proposal_id,
result="error",
detail={"reason": f"invalid state: {proposal.status.value}"},
)
return 1
proposal.status = decision
if decision == ProposalStatus.APPROVED:
proposal.approved_by = actor
proposal.approved_at = datetime.now(UTC)
verb = "approve" if decision == ProposalStatus.APPROVED else "reject"
audit(sess, actor=actor, action=verb, proposal_id=proposal_id)
return 0
finally:
if owned:
outer.__exit__(None, None, None)
# --- autonomous policy (the ONLY non-human approval path, D5) -----------------
def _auto_approve(cfg: Config, sess) -> int:
"""Auto-approve proposed items per policy. Returns count approved."""
if not cfg.autonomous_mode or not cfg.autonomous_kinds:
return 0
midnight = datetime.now(UTC).replace(hour=0, minute=0, second=0, microsecond=0)
used = (
sess.query(func.count(AuditEntry.id))
.filter(
AuditEntry.actor == "auto:policy",
AuditEntry.action == "approve",
AuditEntry.ts >= midnight,
)
.scalar()
)
budget = max(0, cfg.autonomous_daily_cap - int(used or 0))
approved = 0
if budget == 0:
return 0
candidates = (
sess.query(Proposal)
.filter(
Proposal.status == ProposalStatus.PROPOSED,
Proposal.kind.in_(cfg.autonomous_kinds),
)
.order_by(Proposal.risk_score.asc(), Proposal.id.asc())
.limit(budget)
.all()
)
for proposal in candidates:
proposal.status = ProposalStatus.APPROVED
proposal.approved_by = "auto:policy"
proposal.approved_at = datetime.now(UTC)
audit(
sess,
actor="auto:policy",
action="approve",
proposal_id=proposal.id,
detail={
"reason": "autonomous_mode policy",
"kind": proposal.kind,
"risk_score": proposal.risk_score,
"daily_cap": cfg.autonomous_daily_cap,
},
)
approved += 1
return approved
# --- execution ----------------------------------------------------------------
def execute_one(proposal: Proposal, cfg: Config, sess) -> bool:
"""Run the plugin for one approved proposal. Returns True on success."""
proposal.status = ProposalStatus.RUNNING
proposal.attempts += 1
audit(
sess,
actor="daemon:executor",
action="execute_start",
proposal_id=proposal.id,
detail={"plugin": proposal.kind, "attempt": proposal.attempts},
)
plugin = plugin_registry.REGISTRY.get(proposal.kind)
try:
if plugin is None:
outcome: dict[str, Any] = plugin_registry.log_action(proposal, cfg, sess)
outcome["plugin_fallback"] = True
outcome["note"] = f"no plugin registered for kind {proposal.kind!r}"
else:
outcome = plugin(proposal, cfg, sess)
except Exception as exc: # noqa: BLE001 — plugin boundary: capture for audit
audit(
sess,
actor="daemon:executor",
action="execute_error",
proposal_id=proposal.id,
result="error",
detail={"error": str(exc)[:300], "attempt": proposal.attempts},
)
if proposal.attempts >= cfg.max_attempts:
proposal.status = ProposalStatus.FAILED
audit(
sess,
actor="daemon:executor",
action="fail",
proposal_id=proposal.id,
result="error",
detail={"attempts": proposal.attempts},
)
else:
proposal.status = ProposalStatus.APPROVED # retry next run
return False
proposal.status = ProposalStatus.DONE
audit(
sess,
actor="daemon:executor",
action="execute_done",
proposal_id=proposal.id,
detail={"outcome": outcome},
)
return True
def run(cfg: Config | None = None, sess=None) -> int:
"""CLI entry: auto-approve per policy, then execute all approved items."""
from cayceops.config import load
from cayceops.db import engine as _engine
cfg = cfg or load()
owned = False
if sess is None:
outer = session_scope()
sess = outer.__enter__()
owned = True
try:
run_row = DaemonRun(daemon="executor")
sess.add(run_row)
sess.flush()
try:
auto = _auto_approve(cfg, sess)
pending = (
sess.query(Proposal)
.filter(Proposal.status == ProposalStatus.APPROVED)
.order_by(Proposal.approved_at.asc())
.all()
)
done = failed = 0
for proposal in pending:
if execute_one(proposal, cfg, sess):
done += 1
else:
failed += 1
run_row.status = "ok"
run_row.detail = {"auto_approved": auto, "done": done, "failed": failed}
except Exception as exc: # noqa: BLE001 — daemon boundary
run_row.status = "error"
run_row.detail = {"error": str(exc)[:300]}
run_row.finished_at = datetime.now(UTC)
return 1
run_row.finished_at = datetime.now(UTC)
return 0
finally:
if owned:
outer.__exit__(None, None, None)
_engine().dispose()
def select_approved(sess) -> list[Proposal]:
"""Helper for tests/tools: currently approved proposals, oldest first."""
return list(
sess.scalars(
select(Proposal)
.where(Proposal.status == ProposalStatus.APPROVED)
.order_by(Proposal.approved_at.asc())
).all()
)
ActionPlugin = Callable[[Proposal, Config, Any], dict]
+78
View File
@@ -0,0 +1,78 @@
"""executor action plugins.
A plugin maps a proposal ``kind`` to a side effect. Plugins are repo code —
they are reviewed like everything else and must:
- take ``(proposal, cfg, sess)`` and return a JSON-safe dict (the audit outcome)
- do their own permission checks (see ``http_post`` + ENABLE_HTTP_ACTIONS)
- raise on failure (the executor captures the exception into the audit log)
"""
from __future__ import annotations
import json
import urllib.error
import urllib.request
from typing import Any
from cayceops.config import Config
from cayceops.models import Proposal, audit
REGISTRY: dict[str, Any] = {}
def register(kind: str):
"""Decorator: register an action plugin for a proposal kind."""
def deco(fn):
REGISTRY[kind] = fn
return fn
return deco
@register("log_action")
def log_action(proposal: Proposal, cfg: Config, sess) -> dict:
"""No-side-effect plugin: record the execution in the audit log only."""
return {"action": "logged", "title": proposal.title[:200]}
@register("http_post")
def http_post(proposal: Proposal, cfg: Config, sess) -> dict:
"""POST the proposal payload to a webhook.
Inert unless ``CAYCE_OPS_ENABLE_HTTP_ACTIONS=true``. Target URL comes from
the proposal detail (``detail['webhook_url']``) — operator-controlled data.
"""
if not cfg.enable_http_actions:
audit(
sess,
actor="daemon:executor",
action="http_post_blocked",
proposal_id=proposal.id,
result="error",
detail={"reason": "CAYCE_OPS_ENABLE_HTTP_ACTIONS is false"},
)
raise PermissionError("http_post disabled (CAYCE_OPS_ENABLE_HTTP_ACTIONS=false)")
url = (proposal.detail or {}).get("webhook_url", "")
if not url:
raise ValueError("proposal detail.webhook_url missing")
payload = json.dumps(
{"proposal_id": proposal.id, "kind": proposal.kind, "title": proposal.title}
).encode()
req = urllib.request.Request(
url,
data=payload,
headers={"Content-Type": "application/json"},
method="POST",
)
with urllib.request.urlopen(req, timeout=20.0) as resp: # noqa: S310
return {"action": "http_post", "status": resp.status}
# proposals from the scanner default to content_gap/affiliate/partnership kinds;
# until dedicated plugins exist for those kinds they must not do anything
# network-visible. register more plugins here as the operator approves them.
for _safe_kind in ("affiliate", "partnership", "content_gap", "other"):
REGISTRY.setdefault(_safe_kind, log_action)
+1
View File
@@ -0,0 +1 @@
"""maild package."""
+84
View File
@@ -0,0 +1,84 @@
"""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,
# noqa: E501 (kept on one line: the pattern IS the reviewed artifact)
re.compile(r"\b(declin|reject|not able to|unable to|unfortunately|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,
),
]
_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()
+1
View File
@@ -0,0 +1 @@
"""monitor package."""
+174
View File
@@ -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()
@@ -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", "")}
+82
View File
@@ -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
+176
View File
@@ -0,0 +1,176 @@
"""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 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
+290
View File
@@ -0,0 +1,290 @@
"""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()
+1
View File
@@ -0,0 +1 @@
"""scanner package."""
+209
View File
@@ -0,0 +1,209 @@
"""scanner — opportunity discovery. Proposes; never executes.
Sources come from ``CAYCE_OPS_SCANNER_SOURCES`` (JSON array):
[{"type": "rss", "url": "https://example.org/feed.xml"},
{"type": "page", "url": "https://example.org/partners"}]
Every discovered candidate becomes a ``proposals`` row with status ``proposed``
and a stable ``dedupe_key`` (sha1 of url-or-title) so repeated scans stay
idempotent. The scanner performs no side effects beyond writing proposals and
audit entries.
"""
from __future__ import annotations
import hashlib
import json
import re
import urllib.error
import urllib.request
from datetime import UTC, datetime
from html.parser import HTMLParser
from xml.etree import ElementTree
from cayceops.config import Config
from cayceops.db import session_scope
from cayceops.models import DaemonRun, Proposal, ProposalSource, ProposalStatus, audit
# kind classification heuristics (checked in order)
_KIND_PATTERNS = [
("partnership", re.compile(r"partner|collaborat|integrat", re.I)),
("affiliate", re.compile(r"affiliate|referral|commission|revenue\s*share", re.I)),
]
DEFAULT_KIND = "content_gap"
class _LinkCollector(HTMLParser):
"""Minimal HTML parser: page <title> + anchor (href, text) pairs."""
def __init__(self) -> None:
super().__init__()
self.title: str = ""
self.links: list[tuple[str, str]] = []
self._in_title = False
self._in_anchor = False
self._href = ""
self._text: list[str] = []
def handle_starttag(self, tag: str, attrs: list[tuple[str, str]]) -> None:
if tag == "title":
self._in_title = True
elif tag == "a":
self._in_anchor = True
self._href = dict(attrs).get("href", "")
self._text = []
def handle_endtag(self, tag: str) -> None:
if tag == "title":
self._in_title = False
elif tag == "a" and self._in_anchor:
self._in_anchor = False
text = " ".join("".join(self._text).split())
if self._href and text:
self.links.append((self._href, text))
def handle_data(self, data: str) -> None:
if self._in_title:
self.title += data
if self._in_anchor:
self._text.append(data)
def _fetch(url: str, cfg: Config, timeout: float = 20.0) -> bytes:
req = urllib.request.Request(
url, headers={"User-Agent": cfg.scanner_user_agent}, method="GET"
)
with urllib.request.urlopen(req, timeout=timeout) as resp: # noqa: S310
return resp.read()
def _parse_rss(body: bytes) -> list[tuple[str, str, str]]:
"""Return (title, link, description) tuples from an RSS/Atom feed."""
out: list[tuple[str, str, str]] = []
root = ElementTree.fromstring(body)
ns = {"atom": "http://www.w3.org/2005/Atom"}
for item in root.iter("item"): # RSS 2.0
title = item.findtext("title") or ""
link = item.findtext("link") or ""
desc = re.sub(r"<[^>]+>", " ", item.findtext("description") or "")
out.append((" ".join(title.split()), link, " ".join(desc.split())[:500]))
for entry in root.findall("atom:entry", ns): # Atom
title = entry.findtext("atom:title", namespaces=ns) or ""
link_el = entry.find("atom:link", ns)
link = link_el.get("href", "") if link_el is not None else ""
desc = re.sub(r"<[^>]+>", " ", entry.findtext("atom:summary", namespaces=ns) or "")
out.append((" ".join(title.split()), link, " ".join(desc.split())[:500]))
return out
def _parse_page(body: bytes) -> list[tuple[str, str, str]]:
"""Return (title=anchor text, link=href, description=page title) tuples."""
parser = _LinkCollector()
parser.feed(body.decode("utf-8", errors="replace"))
page_title = " ".join(parser.title.split())
return [(text, href, f"linked from page: {page_title}") for href, text in parser.links]
def _classify(title: str, description: str) -> str:
blob = f"{title}\n{description}"
for kind, pattern in _KIND_PATTERNS:
if pattern.search(blob):
return kind
return DEFAULT_KIND
def _score(title: str, description: str, keywords: list[str]) -> float:
blob = f"{title}\n{description}".lower()
hits = sum(1 for kw in keywords if kw.lower() in blob)
return round(min(hits / max(1, len(keywords)), 1.0), 3)
def scan(cfg: Config, sess, fetch=_fetch) -> dict:
"""Run one scan pass against all configured sources. Returns run stats."""
sources = json.loads(cfg.scanner_sources or "[]")
stats = {"sources": len(sources), "candidates": 0, "proposed": 0, "skipped_existing": 0}
existing = {
key
for (key,) in sess.query(Proposal.dedupe_key).filter(
Proposal.dedupe_key.isnot(None)
)
}
for src in sources:
url = src.get("url", "")
stype = src.get("type", "page")
if not url:
continue
try:
body = fetch(url, cfg)
except (urllib.error.URLError, OSError) as exc:
stats.setdefault("source_errors", []).append({"url": url, "error": str(exc)[:200]})
continue
items = _parse_rss(body) if stype == "rss" else _parse_page(body)
stats["candidates"] += len(items)
for title, link, desc in items:
text = title or link
if not text:
continue
keywords = cfg.scanner_keywords
score = _score(title or link, desc, keywords)
if score <= 0 and stype != "page":
continue # feeds must match at least one keyword; pages keep links
dedupe = hashlib.sha1((link or title).encode()).hexdigest()[:32] # noqa: S324
if dedupe in existing:
stats["skipped_existing"] += 1
continue
proposal = Proposal(
source=ProposalSource.SCANNER,
kind=_classify(text, desc),
title=text[:512],
url=(link or None)[:1024] if (link or None) else None,
detail={
"source_type": stype,
"source_url": url,
"summary": desc,
"scanned_at": datetime.now(UTC).isoformat(),
},
risk_score=score,
status=ProposalStatus.PROPOSED,
dedupe_key=dedupe,
)
sess.add(proposal)
sess.flush()
audit(
sess,
actor="daemon:scanner",
action="propose",
proposal_id=proposal.id,
detail={"kind": proposal.kind, "score": score, "source": stype},
)
existing.add(dedupe)
stats["proposed"] += 1
return stats
def run(cfg: Config | None = None) -> int:
"""CLI entry: one scan pass, wrapped in a DaemonRun record."""
from cayceops.config import load
from cayceops.db import engine
cfg = cfg or load()
try:
with session_scope() as sess:
run_row = DaemonRun(daemon="scanner")
sess.add(run_row)
sess.flush()
try:
stats = scan(cfg, sess)
except Exception as exc: # noqa: BLE001 — daemon boundary: record and fail cleanly
run_row.status = "error"
run_row.detail = {"error": str(exc)[:300]}
run_row.finished_at = datetime.now(UTC)
return 1
run_row.status = "ok"
run_row.detail = stats
run_row.finished_at = datetime.now(UTC)
return 0
finally:
engine().dispose()
+133
View File
@@ -0,0 +1,133 @@
"""executor tests — human gate, autonomy bounds, retries, audit chain."""
from __future__ import annotations
import pytest
from cayceops.components.executor import executor, plugins
from cayceops.models import AuditEntry, Proposal, ProposalStatus
def _proposal(kind="content_gap", title="Test item", **kw):
return Proposal(kind=kind, title=title, **kw)
def test_approve_requires_proposed_state(cfg, sess):
p = _proposal()
sess.add(p)
sess.flush()
assert executor.approve(p.id, "gustavo", cfg(), sess) == 0
# second approve must fail (state is now approved)
assert executor.approve(p.id, "gustavo", cfg(), sess) == 1
assert p.approved_by == "gustavo"
assert p.status == ProposalStatus.APPROVED
def test_reject_records_actor(cfg, sess):
p = _proposal()
sess.add(p)
sess.flush()
assert executor.reject(p.id, "gustavo", cfg(), sess) == 0
assert p.status == ProposalStatus.REJECTED
def test_run_executes_only_approved(cfg, sess):
approved = _proposal(title="Approved item")
proposed = _proposal(title="Proposed item")
sess.add_all([approved, proposed])
sess.flush()
executor.approve(approved.id, "gustavo", cfg(), sess)
config = cfg() # AUTONOMOUS_MODE=false
assert executor.run(config, sess) == 0
assert approved.status == ProposalStatus.DONE
assert proposed.status == ProposalStatus.PROPOSED # untouched
acts = [a.action for a in sess.query(AuditEntry).order_by(AuditEntry.id).all()]
assert "execute_start" in acts and "execute_done" in acts
assert acts.count("approve") == 1
def test_autonomy_off_by_default(cfg, sess):
p = _proposal(kind="content_gap", title="Never auto-runs")
sess.add(p)
sess.flush()
config = cfg() # autonomous_mode False even with kinds configured
executor._auto_approve(config, sess)
assert p.status == ProposalStatus.PROPOSED
def test_autonomy_respects_kinds_and_daily_cap(cfg, sess):
p1 = _proposal(kind="content_gap", title="auto ok")
p2 = _proposal(kind="content_gap", title="auto over cap")
p3 = _proposal(kind="affiliate", title="kind not allowed")
sess.add_all([p1, p2, p3])
sess.flush()
config = cfg(
CAYCE_OPS_AUTONOMOUS_MODE="true",
CAYCE_OPS_AUTONOMOUS_KINDS="content_gap",
CAYCE_OPS_AUTONOMOUS_DAILY_CAP="1",
)
approved = executor._auto_approve(config, sess)
assert approved == 1
assert p1.status == ProposalStatus.APPROVED
assert p2.status == ProposalStatus.PROPOSED # cap reached
assert p3.status == ProposalStatus.PROPOSED # kind not in policy
actor = sess.query(AuditEntry).filter_by(action="approve").one().actor
assert actor == "auto:policy"
def test_failing_plugin_retries_then_fails(cfg, sess, monkeypatch):
p = _proposal(kind="affiliate", title="Will fail")
sess.add(p)
sess.flush()
executor.approve(p.id, "gustavo", cfg(), sess)
calls = {"n": 0}
def boom(proposal, cfg, sess):
calls["n"] += 1
raise RuntimeError("upstream exploded")
monkeypatch.setitem(plugins.REGISTRY, "affiliate", boom)
config = cfg(CAYCE_OPS_MAX_ATTEMPTS="2")
executor.run(config, sess) # attempt 1 -> back to approved for retry
assert p.status == ProposalStatus.APPROVED
assert p.attempts == 1
executor.run(config, sess) # attempt 2 -> failed for good
assert p.status == ProposalStatus.FAILED
assert p.attempts == 2
assert calls["n"] == 2
errors = sess.query(AuditEntry).filter_by(result="error").all()
assert any(a.action == "fail" for a in errors)
def test_http_post_blocked_when_disabled(cfg, sess):
p = _proposal(kind="affiliate", title=" webhook target")
p.detail = {"webhook_url": "https://hooks.example.org/ingest"}
sess.add(p)
sess.flush()
config = cfg() # CAYCE_OPS_ENABLE_HTTP_ACTIONS=false
with pytest.raises(PermissionError):
plugins.http_post(p, config, sess)
def test_unknown_kind_falls_back_to_log_action(cfg, sess):
p = _proposal(kind="brand_new_kind", title="No plugin yet")
sess.add(p)
sess.flush()
executor.approve(p.id, "gustavo", cfg(), sess)
assert executor.run(cfg(), sess) == 0
assert p.status == ProposalStatus.DONE
done = sess.query(AuditEntry).filter_by(action="execute_done").one()
assert done.detail["outcome"]["plugin_fallback"] is True
+182
View File
@@ -0,0 +1,182 @@
"""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()
ok = clf.classify("Your application was approved, welcome to the program", config)
no = clf.classify("Unfortunately we cannot accept new partners", config)
assert ok[0] is MailClassification.APPROVAL
assert no[0] is MailClassification.REJECTION
q = clf.classify("Could you provide your traffic numbers?", config)
assert q[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)
def fetcher(cfg):
return [raw]
assert maild._poll(cfg(), sess, fetcher=fetcher)["processed"] == 1
assert maild._poll(cfg(), sess, fetcher=fetcher)["skipped"] == 1 # noqa: E501
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):
_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]) # noqa: E731 (test stub)
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 == []
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=lambda cfg: [raw]) # noqa: E731 (test stub)
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]) # noqa: E731 (test stub)
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
+69
View File
@@ -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
+164
View File
@@ -0,0 +1,164 @@
"""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_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")
+112
View File
@@ -0,0 +1,112 @@
"""scanner tests — discovery, classification, dedupe. No network (injected fetch)."""
from __future__ import annotations
import json
from cayceops.components.scanner import scanner
from cayceops.models import Proposal, ProposalStatus
RSS = b"""<?xml version="1.0"?>
<rss><channel>
<item><title>Acme launches affiliate program for hotels</title>
<link>https://acme.example.org/affiliates</link>
<description>Earn commission on every booking.</description></item>
<item><title>Weekly crypto digest</title>
<link>https://news.example.org/digest</link>
<description>Everything unrelated.</description></item>
</channel></rss>"""
PAGE = b"""<html><head><title>Partners</title></head><body>
<a href="https://pay.example.org/partner">Become a payment partner</a>
<a href="https://pay.example.org/docs">API docs</a>
</body></html>"""
def _fake_fetch(responses):
def fetch(url, cfg, timeout=20.0):
return responses[url]
return fetch
SOURCES = [
{"type": "rss", "url": "https://feeds.example.org/news.xml"},
{"type": "page", "url": "https://pay.example.org/partners"},
]
def _cfg(cfg, **kw):
return cfg(
CAYCE_OPS_SCANNER_SOURCES=json.dumps(SOURCES),
CAYCE_OPS_SCANNER_KEYWORDS="affiliate,partner,bitcoin,commission",
**kw,
)
def test_scan_files_proposals_with_kind_and_score(cfg, sess):
config = _cfg(cfg)
stats = scanner.scan(config, sess, fetch=_fake_fetch({
"https://feeds.example.org/news.xml": RSS,
"https://pay.example.org/partners": PAGE,
}))
assert stats["proposed"] == 3 # affiliate item + both page links (pages keep all)
proposals = sess.query(Proposal).all()
by_title = {p.title: p for p in proposals}
affiliate = by_title["Acme launches affiliate program for hotels"]
assert affiliate.kind == "affiliate"
assert affiliate.status == ProposalStatus.PROPOSED
assert affiliate.risk_score > 0
assert affiliate.dedupe_key
partner = by_title["Become a payment partner"]
assert partner.kind == "partnership"
def test_scan_is_idempotent_via_dedupe(cfg, sess):
config = _cfg(cfg)
scanner.scan(config, sess, fetch=_fake_fetch({
"https://feeds.example.org/news.xml": RSS,
"https://pay.example.org/partners": PAGE,
}))
stats2 = scanner.scan(config, sess, fetch=_fake_fetch({
"https://feeds.example.org/news.xml": RSS,
"https://pay.example.org/partners": PAGE,
}))
assert stats2["proposed"] == 0
assert stats2["skipped_existing"] == 3
def test_scan_filters_keywordless_feed_items(cfg, sess):
config = _cfg(cfg)
scanner.scan(config, sess, fetch=_fake_fetch({
"https://feeds.example.org/news.xml": RSS,
"https://pay.example.org/partners": PAGE,
}))
titles = {p.title for p in sess.query(Proposal).all()}
assert "Weekly crypto digest" not in titles # no keyword match in feed -> dropped
def test_scan_records_audit_for_each_proposal(cfg, sess):
from cayceops.models import AuditEntry
config = _cfg(cfg)
scanner.scan(config, sess, fetch=_fake_fetch({
"https://feeds.example.org/news.xml": RSS,
"https://pay.example.org/partners": PAGE,
}))
acts = [a.action for a in sess.query(AuditEntry).all()]
assert acts.count("propose") == 3
assert all(a.actor == "daemon:scanner" for a in sess.query(AuditEntry).all())
def test_scan_survives_source_error(cfg, sess):
def broken_fetch(url, cfg, timeout=20.0):
raise OSError("connection refused")
config = _cfg(cfg)
stats = scanner.scan(config, sess, fetch=broken_fetch)
assert stats["proposed"] == 0
assert len(stats["source_errors"]) == 2