cayce-ops/cayceops/models.py

"""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:<name>
    proposal_id: Mapped[int | None] = mapped_column(
        ForeignKey("proposals.id", ondelete="SET NULL"), nullable=True, index=True
    )
    action: Mapped[str] = mapped_column(String(64))  # approve|reject|execute|classify|...
    detail: Mapped[dict] = mapped_column(JSON, default=dict)
    result: Mapped[str] = mapped_column(String(32), default="ok")  # ok | error
    # tamper-evident chain
    prev_hash: Mapped[str | None] = mapped_column(String(64), nullable=True)
    entry_hash: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True)


# 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).

    Legacy rows (entry_hash IS NULL, written before migration 0002) are
    skipped and documented: the chain begins GENESIS at the first hashed row,
    so a migrated database verifies clean without any tampering.
    """
    prev_hash: str | None = None  # None = still in legacy prefix
    for entry in sess.query(AuditEntry).order_by(AuditEntry.id).all():
        if entry.entry_hash is None:
            continue  # legacy row: outside the chain by documented genesis
        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)