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

GATE_ERROR = 2  # missing/invalid configuration or signature


def _signing(cfg: Config) -> tuple[str, str]:
    """(deployment_id, operator_pubkey) or ConfigError. No degraded mode."""
    from cayceops.config import ConfigError

    if not cfg.operator_pubkey or not cfg.deployment_id:
        raise ConfigError(
            "approvals disabled: CAYCE_OPS_OPERATOR_PUBKEY and "
            "CAYCE_OPS_DEPLOYMENT_ID are both required"
        )
    return cfg.deployment_id, cfg.operator_pubkey


def approve(
    proposal_id: int,
    actor: Actor,
    cfg: Config | None = None,
    sess=None,
    signature: str | None = None,
) -> int:
    """Human approves a proposal. Signature is MANDATORY and bound to the
    proposal's current content and this deployment (audit finding 1)."""
    from cayceops.config import ConfigError
    from cayceops.signing import (
        SignatureError,
        proposal_content_hash,
        verify_signature,
    )

    cfg = cfg or _load_cfg()
    try:
        deployment_id, pubkey = _signing(cfg)
    except ConfigError as exc:
        if sess is not None:
            audit(
                sess,
                actor=actor,
                action="approve",
                proposal_id=proposal_id,
                result="error",
                detail={"reason": str(exc)},
            )
        return GATE_ERROR

    proposal = sess.get(Proposal, proposal_id) if sess is not None else None
    if proposal is None:
        return 1
    chash = proposal_content_hash(
        proposal.kind, proposal.title, proposal.url, proposal.detail
    )
    try:
        if not signature:
            raise SignatureError("signature required")
        if not verify_signature(
            "proposal", deployment_id, proposal_id, chash, signature, pubkey
        ):
            raise SignatureError("signature invalid for this content/deployment")
    except SignatureError as exc:
        audit(
            sess,
            actor=actor,
            action="approve",
            proposal_id=proposal_id,
            result="error",
            detail={"reason": str(exc), "content_hash": chash},
        )
        return GATE_ERROR

    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)
            proposal.operator_signature = signature  # FULL signature, row + audit
            detail["signature"] = signature or ""
        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:
    """DISABLED (audit round 2, point 1a): approvals require operator
    signatures bound to content — there is no non-human approval path.
    AUTONOMOUS_MODE no longer grants approval powers (DECISIONS D13)."""
    if cfg.autonomous_mode and cfg.autonomous_kinds:
        audit(
            sess,
            actor="auto:policy",
            action="autonomy_disabled",
            result="error",
            detail={
                "reason": "mandatory operator signatures: no non-human approvals",
            },
        )
    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 _verify_approval_signature(proposal: Proposal, cfg: Config, sess) -> bool:
    """Run-time gate (audit round 2, 1d): a row in `approved` MUST carry a
    valid operator signature for its CURRENT content and THIS deployment.
    Rows approved by raw SQL (no signature) or edited after approval are
    rejected, audited as error, and revert to `proposed`."""
    from cayceops.signing import (
        SignatureError,
        proposal_content_hash,
        verify_signature,
    )

    try:
        deployment_id, pubkey = _signing(cfg)
    except Exception as exc:  # noqa: BLE001 — ConfigError at daemon boundary
        audit(
            sess,
            actor="daemon:executor",
            action="execute_refused",
            proposal_id=proposal.id,
            result="error",
            detail={"reason": str(exc)},
        )
        return False
    chash = proposal_content_hash(
        proposal.kind, proposal.title, proposal.url, proposal.detail
    )
    try:
        if not verify_signature(
            "proposal",
            deployment_id,
            proposal.id,
            chash,
            proposal.operator_signature,
            pubkey,
        ):
            raise SignatureError(
                "no valid operator signature for current content/deployment"
            )
    except SignatureError as exc:
        proposal.status = ProposalStatus.PROPOSED  # back to pre-approval
        proposal.operator_signature = None
        audit(
            sess,
            actor="daemon:executor",
            action="execute_refused",
            proposal_id=proposal.id,
            result="error",
            detail={
                "reason": str(exc),
                "content_hash": chash,
                "had_signature": bool(proposal.operator_signature),
            },
        )
        return False
    return True


def execute_one(proposal: Proposal, cfg: Config, sess) -> bool:
    """Run the plugin for one approved proposal. Returns True on success."""
    if not _verify_approval_signature(proposal, cfg, sess):
        return False
    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:
            _signing(cfg)  # refuse to run without signing config (no degraded mode)
        except Exception as exc:  # noqa: BLE001 — ConfigError at boundary
            run_row.status = "error"
            run_row.detail = {"refused": str(exc)[:200]}
            run_row.finished_at = datetime.now(UTC)
            return 2
        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]