cayce-ops/cayceops/components/executor/executor.py

"""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


def _load_cfg():
    from cayceops.config import load

    return load()


# --- human transitions -------------------------------------------------------

# message signed by the operator key when CAYCE_OPS_OPERATOR_PUBKEY is set
def approval_message(proposal_id: int) -> bytes:
    import hashlib

    return hashlib.sha256(f"cayce-ops:approve:{proposal_id}".encode()).digest()


def approve(
    proposal_id: int,
    actor: Actor,
    cfg: Config | None = None,
    sess=None,
    signature: str | None = None,
) -> int:
    """Human approves a proposal. Audited with the operator's identity.

    When CAYCE_OPS_OPERATOR_PUBKEY is configured, `signature` (hex, BIP-340
    over approval_message(id)) is REQUIRED and verified against that key —
    'actor' then is a label, the key is the proof (audit finding 1).
    """
    from cayceops.components.publisher.nostr import verify_schnorr

    cfg = cfg or _load_cfg()
    if cfg.operator_pubkey:
        ok = bool(signature) and verify_schnorr(
            signature, approval_message(proposal_id), cfg.operator_pubkey
        )
        if not ok and sess is not None:
            audit(
                sess,
                actor=actor,
                action="approve",
                proposal_id=proposal_id,
                result="error",
                detail={
                    "reason": (
                        "operator pubkey configured: signature required"
                        if not signature
                        else "invalid approval signature"
                    )
                },
            )
        if not ok:
            return 2
    return _decide(proposal_id, actor, ProposalStatus.APPROVED, cfg, sess, signature)


def reject(proposal_id: int, actor: Actor, cfg: Config | None = None, sess=None) -> int:
    """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,
    signature: str | None = None,
) -> 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
        detail: dict = {}
        if decision == ProposalStatus.APPROVED:
            proposal.approved_by = actor
            proposal.approved_at = datetime.now(UTC)
            if signature:
                detail["signature"] = signature[:32] + "..."
        verb = "approve" if decision == ProposalStatus.APPROVED else "reject"
        audit(sess, actor=actor, action=verb, proposal_id=proposal_id, detail=detail)
        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
    # most relevant first (match_score = keyword relevance, NOT risk — the
    # audit's naming finding; ascending order approved noise first)
    candidates = (
        sess.query(Proposal)
        .filter(
            Proposal.status == ProposalStatus.PROPOSED,
            Proposal.kind.in_(cfg.autonomous_kinds),
        )
        .order_by(Proposal.match_score.desc(), 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,
                "match_score": proposal.match_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]