"""ORM models and lifecycle states shared by every component. Status machines --------------- proposal: proposed -> approved -> running -> done | failed proposed -> rejected mail: (inbound) classified: approval|rejection|question|other|needs_review (any) draft: none -> drafted -> approved -> sent | skipped content: draft -> approved -> published | skipped (terminal) Audit chain ----------- audit_log is hash-chained: each entry stores sha256(prev_hash | fields). ``cayceops.models.verify_audit_chain`` re-walks it; the monitor surfaces breaks. Not tamper-proof (DB writes can rewrite the chain) but tamper-EVIDENT against selective edits. See AUDIT.md / audit response 2026-10-20. """ from __future__ import annotations import enum import hashlib import json import threading from datetime import UTC, datetime from sqlalchemy import JSON, DateTime, Enum, Float, ForeignKey, Integer, String, Text, text from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column def utcnow() -> datetime: return datetime.now(UTC) def _content_hash(text: str) -> str: """Binding hash for human-approved text (drafts, captions).""" return hashlib.sha256(text.encode("utf-8")).hexdigest() class Base(DeclarativeBase): pass # --- proposals --------------------------------------------------------------- class ProposalStatus(enum.StrEnum): PROPOSED = "proposed" APPROVED = "approved" REJECTED = "rejected" RUNNING = "running" DONE = "done" FAILED = "failed" class ProposalSource(enum.StrEnum): SCANNER = "scanner" MANUAL = "manual" class Proposal(Base): __tablename__ = "proposals" id: Mapped[int] = mapped_column(Integer, primary_key=True) source: Mapped[ProposalSource] = mapped_column( Enum(ProposalSource), default=ProposalSource.SCANNER ) kind: Mapped[str] = mapped_column(String(64)) # affiliate|partnership|content_gap|other title: Mapped[str] = mapped_column(String(512)) url: Mapped[str | None] = mapped_column(String(1024), nullable=True) detail: Mapped[dict] = mapped_column(JSON, default=dict) # keyword-match relevance 0..1 — NOT risk. auto-approve takes highest first. match_score: Mapped[float] = mapped_column(Float, default=0.0) status: Mapped[ProposalStatus] = mapped_column( Enum(ProposalStatus), default=ProposalStatus.PROPOSED, index=True ) attempts: Mapped[int] = mapped_column(Integer, default=0) dedupe_key: Mapped[str | None] = mapped_column(String(256), unique=True, nullable=True) approved_by: Mapped[str | None] = mapped_column(String(128), nullable=True) approved_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) # full operator approval signature (128 hex chars) — never truncated operator_signature: Mapped[str | None] = mapped_column(Text, nullable=True) created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow) updated_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), default=utcnow, onupdate=utcnow ) # --- audit (hash-chained) ---------------------------------------------------- class AuditEntry(Base): __tablename__ = "audit_log" id: Mapped[int] = mapped_column(Integer, primary_key=True) ts: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow, index=True) actor: Mapped[str] = mapped_column(String(128)) # human | auto:policy | daemon: proposal_id: Mapped[int | None] = mapped_column( ForeignKey("proposals.id", ondelete="SET NULL"), nullable=True, index=True ) action: Mapped[str] = mapped_column(String(64)) # approve|reject|execute|classify|... detail: Mapped[dict] = mapped_column(JSON, default=dict) result: Mapped[str] = mapped_column(String(32), default="ok") # ok | error # tamper-evident chain prev_hash: Mapped[str | None] = mapped_column(String(64), nullable=True) entry_hash: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True) # reentrant: several audit() calls may share one transaction _AUDIT_LOCK = threading.RLock() _AUDIT_LOCK_PG_KEY = 7471186757 # pg_advisory_xact_lock key for cross-process def audit( sess, actor: str, action: str, proposal_id: int | None = None, detail: dict | None = None, result: str = "ok", ) -> AuditEntry: """Record one audited action, chained to the previous entry. Chain: entry_hash = sha256(prev_hash | ts_epoch | actor | action | proposal_id | detail | result). Writes are SERIALIZED (audit round 2, point 4): pg_advisory_xact_lock on Postgres, in-process lock on SQLite — concurrent writers can no longer fork the chain. Legacy rows (entry_hash IS NULL, pre-0002) are ignored when picking prev_hash: the chain starts GENESIS at the first hashed row. """ bind = sess.get_bind() if bind.dialect.name == "postgresql": # xact-scoped: PG releases the lock at commit/rollback automatically sess.execute( text("SELECT pg_advisory_xact_lock(:k)"), {"k": _AUDIT_LOCK_PG_KEY} ) prev = sess.query(AuditEntry).order_by(AuditEntry.id.desc()).first() prev_hash = prev.entry_hash if prev is not None else "GENESIS" entry = _build_entry(prev_hash, actor, action, proposal_id, detail, result) sess.add(entry) return entry # SQLite path: hold the in-process lock until the session commits or # rolls back, so concurrent writers cannot read the same prev_hash # (round-2 point 4 — verified by test_concurrent_writers_keep_chain_verifiable) _AUDIT_LOCK.acquire() released = False def _release(*_args): nonlocal released if not released: released = True try: _AUDIT_LOCK.release() except RuntimeError: # pragma: no cover pass from sqlalchemy import event event.listen(sess, "after_commit", _release) event.listen(sess, "after_rollback", _release) event.listen(sess, "after_soft_rollback", _release) try: prev = sess.query(AuditEntry).order_by(AuditEntry.id.desc()).first() prev_hash = prev.entry_hash if prev is not None else "GENESIS" entry = _build_entry(prev_hash, actor, action, proposal_id, detail, result) sess.add(entry) return entry except Exception: _release() raise def _build_entry( prev_hash: str, actor: str, action: str, proposal_id: int | None, detail: dict | None, result: str, ) -> AuditEntry: ts = utcnow() ts_int = int(ts.timestamp()) detail_json = json.dumps(detail or {}, sort_keys=True, ensure_ascii=False) payload = "|".join( [ prev_hash or "GENESIS", str(ts_int), actor, action, str(proposal_id), detail_json, result, ] ) return AuditEntry( ts=ts, actor=actor, action=action, proposal_id=proposal_id, detail=detail or {}, result=result, prev_hash=prev_hash or "GENESIS", entry_hash=hashlib.sha256(payload.encode("utf-8")).hexdigest(), ) def verify_audit_chain(sess) -> tuple[bool, int | None]: """Re-walk the chain. Returns (ok, first_broken_entry_id or None). Rows with entry_hash IS NULL are ONLY tolerated BEFORE the first hashed row (the documented legacy prefix from before migration 0002). A NULL hash AFTER the chain has started is tampering — an attacker erasing the hash of an intermediate row to hide an edit — and fails verification at that row's id. """ prev_hash: str | None = None # None = still in legacy prefix chain_started = False for entry in sess.query(AuditEntry).order_by(AuditEntry.id).all(): if entry.entry_hash is None: if chain_started: # tampering: a hash was erased in the middle of the chain return False, entry.id continue # legacy prefix row: outside the chain by genesis chain_started = True if prev_hash is None: expected_prev = "GENESIS" # first hashed row anchors the chain else: expected_prev = prev_hash detail_json = json.dumps(entry.detail or {}, sort_keys=True, ensure_ascii=False) ts = entry.ts if ts.tzinfo is None: ts = ts.replace(tzinfo=UTC) payload = "|".join( [ expected_prev, str(int(ts.timestamp())), entry.actor, entry.action, str(entry.proposal_id), detail_json, entry.result, ] ) expected = hashlib.sha256(payload.encode("utf-8")).hexdigest() if entry.prev_hash != expected_prev or entry.entry_hash != expected: return False, entry.id prev_hash = entry.entry_hash return True, None # --- mail -------------------------------------------------------------------- class MailDirection(enum.StrEnum): INBOUND = "in" OUTBOUND = "out" class MailClassification(enum.StrEnum): APPROVAL = "approval" REJECTION = "rejection" QUESTION = "question" OTHER = "other" NEEDS_REVIEW = "needs_review" class DraftStatus(enum.StrEnum): NONE = "none" DRAFTED = "drafted" APPROVED = "approved" # human approved the draft; SMTP may send it SENT = "sent" SKIPPED = "skipped" class MailMessage(Base): __tablename__ = "mail_messages" id: Mapped[int] = mapped_column(Integer, primary_key=True) direction: Mapped[MailDirection] = mapped_column(Enum(MailDirection)) message_id: Mapped[str | None] = mapped_column(String(998), nullable=True, index=True) in_reply_to: Mapped[str | None] = mapped_column(String(998), nullable=True) from_addr: Mapped[str | None] = mapped_column(String(320), nullable=True) to_addr: Mapped[str | None] = mapped_column(String(320), nullable=True) subject: Mapped[str | None] = mapped_column(String(998), nullable=True) body: Mapped[str | None] = mapped_column(Text, nullable=True) classification: Mapped[MailClassification | None] = mapped_column( Enum(MailClassification), nullable=True ) confidence: Mapped[float] = mapped_column(Float, default=0.0) proposal_id: Mapped[int | None] = mapped_column( ForeignKey("proposals.id", ondelete="SET NULL"), nullable=True, index=True ) draft_status: Mapped[DraftStatus] = mapped_column( Enum(DraftStatus), default=DraftStatus.NONE ) draft_reply: Mapped[str | None] = mapped_column(Text, nullable=True) # binding: hash of the draft text the human approved + who approved it approved_hash: Mapped[str | None] = mapped_column(String(64), nullable=True) approved_by: Mapped[str | None] = mapped_column(String(128), nullable=True) operator_signature: Mapped[str | None] = mapped_column(Text, nullable=True) processed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) # --- content ----------------------------------------------------------------- class ContentStatus(enum.StrEnum): DRAFT = "draft" APPROVED = "approved" PUBLISHED = "published" SKIPPED = "skipped" # terminal: disabled channel or attempt budget exhausted class ContentItem(Base): __tablename__ = "content_items" id: Mapped[int] = mapped_column(Integer, primary_key=True) theme: Mapped[str] = mapped_column(String(128)) channel: Mapped[str] = mapped_column(String(32)) # nostr|facebook|instagram caption: Mapped[str] = mapped_column(Text) media_path: Mapped[str | None] = mapped_column(String(1024), nullable=True) link: Mapped[str | None] = mapped_column(String(1024), nullable=True) status: Mapped[ContentStatus] = mapped_column( Enum(ContentStatus), default=ContentStatus.DRAFT, index=True ) scheduled_for: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) published_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) post_ref: Mapped[str | None] = mapped_column(String(512), nullable=True) error: Mapped[str | None] = mapped_column(Text, nullable=True) attempts: Mapped[int] = mapped_column(Integer, default=0) # binding: hash of the caption the human approved approved_hash: Mapped[str | None] = mapped_column(String(64), nullable=True) operator_signature: Mapped[str | None] = mapped_column(Text, nullable=True) created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow) # --- daemon runs (monitor input) --------------------------------------------- class DaemonRun(Base): __tablename__ = "daemon_runs" id: Mapped[int] = mapped_column(Integer, primary_key=True) # scanner | executor | maild | publisher | monitor daemon: Mapped[str] = mapped_column(String(32), index=True) started_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow) finished_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) status: Mapped[str] = mapped_column(String(16), default="ok") # ok | error detail: Mapped[dict] = mapped_column(JSON, default=dict)