cayce
a2de419467
- README: the stack belongs to Cayce, an autonomous entity operating under the Próspera ZEDE agentic-companies jurisdiction pilot; the experiment is managed by Cayce itself - tests: approver actor renamed to the generic label operator - zero occurrences of personal names in the tree (verified with grep)
243 lines
8.4 KiB
Python
243 lines
8.4 KiB
Python
"""executor tests — human gate (signed), autonomy bounds, retries, audit chain.
|
|
|
|
Since audit round 2, approvals REQUIRE an operator signature bound to the
|
|
proposal's current content + deployment id (tests sign with the conftest
|
|
operator key 22*32, deployment 'itest').
|
|
"""
|
|
|
|
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, signer):
|
|
p = _proposal()
|
|
sess.add(p)
|
|
sess.flush()
|
|
|
|
assert (
|
|
executor.approve(p.id, "operator", cfg(), sess, signature=signer.proposal(p)) == 0
|
|
)
|
|
# second approve must fail (state is now approved)
|
|
assert (
|
|
executor.approve(p.id, "operator", cfg(), sess, signature=signer.proposal(p)) == 1
|
|
)
|
|
assert p.approved_by == "operator"
|
|
assert p.status is ProposalStatus.APPROVED
|
|
# FULL signature stored on the row
|
|
assert p.operator_signature and len(p.operator_signature) == 128
|
|
|
|
|
|
def test_reject_records_actor(cfg, sess):
|
|
p = _proposal()
|
|
sess.add(p)
|
|
sess.flush()
|
|
assert executor.reject(p.id, "operator", cfg(), sess) == 0
|
|
assert p.status is ProposalStatus.REJECTED
|
|
|
|
|
|
def test_run_executes_only_approved(cfg, sess, signer):
|
|
approved = _proposal(title="Approved item")
|
|
proposed = _proposal(title="Proposed item")
|
|
sess.add_all([approved, proposed])
|
|
sess.flush()
|
|
assert (
|
|
executor.approve(
|
|
approved.id, "operator", cfg(), sess, signature=signer.proposal(approved)
|
|
)
|
|
== 0
|
|
)
|
|
|
|
config = cfg() # AUTONOMOUS_MODE=false
|
|
assert executor.run(config, sess) == 0
|
|
|
|
assert approved.status is ProposalStatus.DONE
|
|
assert proposed.status is 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 is ProposalStatus.PROPOSED
|
|
|
|
|
|
def test_autonomy_cannot_approve_anymore(cfg, sess):
|
|
"""Round-2 1a: with mandatory signatures there is NO non-human approval
|
|
path. AUTONOMOUS_MODE no longer approves anything (D13)."""
|
|
low = _proposal(kind="content_gap", title="barely relevant")
|
|
low.match_score = 0.1
|
|
high = _proposal(kind="content_gap", title="highly relevant")
|
|
high.match_score = 0.9
|
|
sess.add_all([low, high])
|
|
sess.flush()
|
|
config = cfg(
|
|
CAYCE_OPS_AUTONOMOUS_MODE="true",
|
|
CAYCE_OPS_AUTONOMOUS_KINDS="content_gap",
|
|
CAYCE_OPS_AUTONOMOUS_DAILY_CAP="1",
|
|
)
|
|
assert executor._auto_approve(config, sess) == 0
|
|
assert low.status is ProposalStatus.PROPOSED
|
|
assert high.status is ProposalStatus.PROPOSED
|
|
entry = sess.query(AuditEntry).filter_by(action="autonomy_disabled").one()
|
|
assert entry.result == "error"
|
|
|
|
|
|
def test_signed_approval_mandatory_and_content_bound(cfg, sess, signer):
|
|
"""Round-2 1a/1b/1c: signature mandatory, bound to content+deployment,
|
|
stored COMPLETE on the row and in the audit log. Signatures for other
|
|
deployments or other content are refused."""
|
|
from cayceops.signing import proposal_content_hash
|
|
|
|
p = _proposal(kind="partnership", title="Acme partner", url="https://acme.example.org")
|
|
p.detail = {"region": "latam"}
|
|
sess.add(p)
|
|
sess.flush()
|
|
|
|
chash = proposal_content_hash(p.kind, p.title, p.url, p.detail)
|
|
|
|
# no signature -> refused
|
|
assert executor.approve(p.id, "operator", cfg(), sess, signature=None) == 2
|
|
# signature for ANOTHER deployment -> refused (required test)
|
|
other_dep = signer.raw(
|
|
signer.msg("proposal", "other-deploy", p.id, chash)
|
|
)
|
|
assert executor.approve(p.id, "operator", cfg(), sess, signature=other_dep) == 2
|
|
# signature over OTHER CONTENT -> refused (required test)
|
|
other_content = signer.raw(signer.msg("proposal", "itest", p.id, "00" * 32))
|
|
assert executor.approve(p.id, "operator", cfg(), sess, signature=other_content) == 2
|
|
# valid signature -> approved
|
|
assert (
|
|
executor.approve(p.id, "operator", cfg(), sess, signature=signer.proposal(p)) == 0
|
|
)
|
|
assert p.status is ProposalStatus.APPROVED
|
|
entry = sess.query(AuditEntry).filter_by(action="approve", result="ok").one()
|
|
assert entry.detail["signature"] == p.operator_signature
|
|
|
|
|
|
def test_approve_fails_audited_from_cli_wrapper(cfg, sess):
|
|
"""Round-2 1f: signature failures leave an audit record even when the
|
|
command runs through its CLI wrapper (the wrapper passes sess)."""
|
|
p = _proposal()
|
|
sess.add(p)
|
|
sess.flush()
|
|
|
|
code = executor.approve(p.id, "operator", cfg(), sess, signature=None)
|
|
assert code == 2
|
|
entry = sess.query(AuditEntry).filter_by(action="approve", result="error").one()
|
|
assert "signature required" in entry.detail["reason"]
|
|
|
|
|
|
def test_sql_approved_row_is_refused_at_execution(cfg, sess, monkeypatch):
|
|
"""Round-2 1d (required test): a row put in `approved` by raw SQL without
|
|
a signature must not execute; it reverts to `proposed` with an error audit."""
|
|
p = _proposal()
|
|
p.status = ProposalStatus.APPROVED # raw-SQL style: no signature
|
|
sess.add(p)
|
|
sess.flush()
|
|
|
|
executed = []
|
|
|
|
def spy_plugin(proposal, cfg, sess):
|
|
executed.append(proposal.id)
|
|
return {}
|
|
|
|
monkeypatch.setitem(plugins.REGISTRY, "content_gap", spy_plugin)
|
|
assert executor.run(cfg(), sess) == 0
|
|
assert executed == []
|
|
assert p.status is ProposalStatus.PROPOSED # reverted
|
|
refused = sess.query(AuditEntry).filter_by(action="execute_refused").one()
|
|
assert refused.result == "error"
|
|
|
|
|
|
def test_execution_verifies_signature_against_current_content(cfg, sess, signer, monkeypatch):
|
|
"""Round-2 1d: signature over content that changed after approval is
|
|
refused at execution time."""
|
|
p = _proposal(kind="partnership", title="original")
|
|
sess.add(p)
|
|
sess.flush()
|
|
assert (
|
|
executor.approve(p.id, "operator", cfg(), sess, signature=signer.proposal(p)) == 0
|
|
)
|
|
|
|
p.title = "EDITED after approval" # content changed under the signature
|
|
|
|
executed = []
|
|
monkeypatch.setitem(
|
|
plugins.REGISTRY,
|
|
"partnership",
|
|
lambda proposal, cfg, sess: executed.append(proposal.id) or {},
|
|
)
|
|
assert executor.run(cfg(), sess) == 0
|
|
assert executed == []
|
|
assert p.status is ProposalStatus.PROPOSED
|
|
|
|
|
|
def test_failing_plugin_retries_then_fails(cfg, sess, signer, monkeypatch):
|
|
p = _proposal(kind="affiliate", title="Will fail")
|
|
sess.add(p)
|
|
sess.flush()
|
|
assert (
|
|
executor.approve(p.id, "operator", cfg(), sess, signature=signer.proposal(p)) == 0
|
|
)
|
|
|
|
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 is ProposalStatus.APPROVED
|
|
assert p.attempts == 1
|
|
|
|
executor.run(config, sess) # attempt 2 -> failed for good
|
|
assert p.status is 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, signer):
|
|
p = _proposal(kind="brand_new_kind", title="No plugin yet")
|
|
sess.add(p)
|
|
sess.flush()
|
|
assert (
|
|
executor.approve(p.id, "operator", cfg(), sess, signature=signer.proposal(p)) == 0
|
|
)
|
|
|
|
assert executor.run(cfg(), sess) == 0
|
|
assert p.status is ProposalStatus.DONE
|
|
|
|
done = sess.query(AuditEntry).filter_by(action="execute_done").one()
|
|
assert done.detail["outcome"]["plugin_fallback"] is True
|