Compare commits
2
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
73ccdf7bdf | ||
|
|
00fa3777ab |
@@ -0,0 +1 @@
|
||||
"""executor package."""
|
||||
@@ -0,0 +1,246 @@
|
||||
"""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
|
||||
|
||||
|
||||
# --- human transitions -------------------------------------------------------
|
||||
|
||||
|
||||
def approve(proposal_id: int, actor: Actor, cfg: Config | None = None, sess=None) -> int:
|
||||
"""Human approves a proposal. Audited with the operator's identity."""
|
||||
return _decide(proposal_id, actor, ProposalStatus.APPROVED, cfg, sess)
|
||||
|
||||
|
||||
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,
|
||||
) -> 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
|
||||
if decision == ProposalStatus.APPROVED:
|
||||
proposal.approved_by = actor
|
||||
proposal.approved_at = datetime.now(UTC)
|
||||
verb = "approve" if decision == ProposalStatus.APPROVED else "reject"
|
||||
audit(sess, actor=actor, action=verb, proposal_id=proposal_id)
|
||||
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
|
||||
candidates = (
|
||||
sess.query(Proposal)
|
||||
.filter(
|
||||
Proposal.status == ProposalStatus.PROPOSED,
|
||||
Proposal.kind.in_(cfg.autonomous_kinds),
|
||||
)
|
||||
.order_by(Proposal.risk_score.asc(), 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,
|
||||
"risk_score": proposal.risk_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]
|
||||
@@ -0,0 +1,78 @@
|
||||
"""executor action plugins.
|
||||
|
||||
A plugin maps a proposal ``kind`` to a side effect. Plugins are repo code —
|
||||
they are reviewed like everything else and must:
|
||||
|
||||
- take ``(proposal, cfg, sess)`` and return a JSON-safe dict (the audit outcome)
|
||||
- do their own permission checks (see ``http_post`` + ENABLE_HTTP_ACTIONS)
|
||||
- raise on failure (the executor captures the exception into the audit log)
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import urllib.error
|
||||
import urllib.request
|
||||
from typing import Any
|
||||
|
||||
from cayceops.config import Config
|
||||
from cayceops.models import Proposal, audit
|
||||
|
||||
REGISTRY: dict[str, Any] = {}
|
||||
|
||||
|
||||
def register(kind: str):
|
||||
"""Decorator: register an action plugin for a proposal kind."""
|
||||
|
||||
def deco(fn):
|
||||
REGISTRY[kind] = fn
|
||||
return fn
|
||||
|
||||
return deco
|
||||
|
||||
|
||||
@register("log_action")
|
||||
def log_action(proposal: Proposal, cfg: Config, sess) -> dict:
|
||||
"""No-side-effect plugin: record the execution in the audit log only."""
|
||||
return {"action": "logged", "title": proposal.title[:200]}
|
||||
|
||||
|
||||
@register("http_post")
|
||||
def http_post(proposal: Proposal, cfg: Config, sess) -> dict:
|
||||
"""POST the proposal payload to a webhook.
|
||||
|
||||
Inert unless ``CAYCE_OPS_ENABLE_HTTP_ACTIONS=true``. Target URL comes from
|
||||
the proposal detail (``detail['webhook_url']``) — operator-controlled data.
|
||||
"""
|
||||
if not cfg.enable_http_actions:
|
||||
audit(
|
||||
sess,
|
||||
actor="daemon:executor",
|
||||
action="http_post_blocked",
|
||||
proposal_id=proposal.id,
|
||||
result="error",
|
||||
detail={"reason": "CAYCE_OPS_ENABLE_HTTP_ACTIONS is false"},
|
||||
)
|
||||
raise PermissionError("http_post disabled (CAYCE_OPS_ENABLE_HTTP_ACTIONS=false)")
|
||||
|
||||
url = (proposal.detail or {}).get("webhook_url", "")
|
||||
if not url:
|
||||
raise ValueError("proposal detail.webhook_url missing")
|
||||
payload = json.dumps(
|
||||
{"proposal_id": proposal.id, "kind": proposal.kind, "title": proposal.title}
|
||||
).encode()
|
||||
req = urllib.request.Request(
|
||||
url,
|
||||
data=payload,
|
||||
headers={"Content-Type": "application/json"},
|
||||
method="POST",
|
||||
)
|
||||
with urllib.request.urlopen(req, timeout=20.0) as resp: # noqa: S310
|
||||
return {"action": "http_post", "status": resp.status}
|
||||
|
||||
|
||||
# proposals from the scanner default to content_gap/affiliate/partnership kinds;
|
||||
# until dedicated plugins exist for those kinds they must not do anything
|
||||
# network-visible. register more plugins here as the operator approves them.
|
||||
for _safe_kind in ("affiliate", "partnership", "content_gap", "other"):
|
||||
REGISTRY.setdefault(_safe_kind, log_action)
|
||||
@@ -0,0 +1 @@
|
||||
"""scanner package."""
|
||||
@@ -0,0 +1,209 @@
|
||||
"""scanner — opportunity discovery. Proposes; never executes.
|
||||
|
||||
Sources come from ``CAYCE_OPS_SCANNER_SOURCES`` (JSON array):
|
||||
[{"type": "rss", "url": "https://example.org/feed.xml"},
|
||||
{"type": "page", "url": "https://example.org/partners"}]
|
||||
|
||||
Every discovered candidate becomes a ``proposals`` row with status ``proposed``
|
||||
and a stable ``dedupe_key`` (sha1 of url-or-title) so repeated scans stay
|
||||
idempotent. The scanner performs no side effects beyond writing proposals and
|
||||
audit entries.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import json
|
||||
import re
|
||||
import urllib.error
|
||||
import urllib.request
|
||||
from datetime import UTC, datetime
|
||||
from html.parser import HTMLParser
|
||||
from xml.etree import ElementTree
|
||||
|
||||
from cayceops.config import Config
|
||||
from cayceops.db import session_scope
|
||||
from cayceops.models import DaemonRun, Proposal, ProposalSource, ProposalStatus, audit
|
||||
|
||||
# kind classification heuristics (checked in order)
|
||||
_KIND_PATTERNS = [
|
||||
("partnership", re.compile(r"partner|collaborat|integrat", re.I)),
|
||||
("affiliate", re.compile(r"affiliate|referral|commission|revenue\s*share", re.I)),
|
||||
]
|
||||
DEFAULT_KIND = "content_gap"
|
||||
|
||||
|
||||
class _LinkCollector(HTMLParser):
|
||||
"""Minimal HTML parser: page <title> + anchor (href, text) pairs."""
|
||||
|
||||
def __init__(self) -> None:
|
||||
super().__init__()
|
||||
self.title: str = ""
|
||||
self.links: list[tuple[str, str]] = []
|
||||
self._in_title = False
|
||||
self._in_anchor = False
|
||||
self._href = ""
|
||||
self._text: list[str] = []
|
||||
|
||||
def handle_starttag(self, tag: str, attrs: list[tuple[str, str]]) -> None:
|
||||
if tag == "title":
|
||||
self._in_title = True
|
||||
elif tag == "a":
|
||||
self._in_anchor = True
|
||||
self._href = dict(attrs).get("href", "")
|
||||
self._text = []
|
||||
|
||||
def handle_endtag(self, tag: str) -> None:
|
||||
if tag == "title":
|
||||
self._in_title = False
|
||||
elif tag == "a" and self._in_anchor:
|
||||
self._in_anchor = False
|
||||
text = " ".join("".join(self._text).split())
|
||||
if self._href and text:
|
||||
self.links.append((self._href, text))
|
||||
|
||||
def handle_data(self, data: str) -> None:
|
||||
if self._in_title:
|
||||
self.title += data
|
||||
if self._in_anchor:
|
||||
self._text.append(data)
|
||||
|
||||
|
||||
def _fetch(url: str, cfg: Config, timeout: float = 20.0) -> bytes:
|
||||
req = urllib.request.Request(
|
||||
url, headers={"User-Agent": cfg.scanner_user_agent}, method="GET"
|
||||
)
|
||||
with urllib.request.urlopen(req, timeout=timeout) as resp: # noqa: S310
|
||||
return resp.read()
|
||||
|
||||
|
||||
def _parse_rss(body: bytes) -> list[tuple[str, str, str]]:
|
||||
"""Return (title, link, description) tuples from an RSS/Atom feed."""
|
||||
out: list[tuple[str, str, str]] = []
|
||||
root = ElementTree.fromstring(body)
|
||||
ns = {"atom": "http://www.w3.org/2005/Atom"}
|
||||
for item in root.iter("item"): # RSS 2.0
|
||||
title = item.findtext("title") or ""
|
||||
link = item.findtext("link") or ""
|
||||
desc = re.sub(r"<[^>]+>", " ", item.findtext("description") or "")
|
||||
out.append((" ".join(title.split()), link, " ".join(desc.split())[:500]))
|
||||
for entry in root.findall("atom:entry", ns): # Atom
|
||||
title = entry.findtext("atom:title", namespaces=ns) or ""
|
||||
link_el = entry.find("atom:link", ns)
|
||||
link = link_el.get("href", "") if link_el is not None else ""
|
||||
desc = re.sub(r"<[^>]+>", " ", entry.findtext("atom:summary", namespaces=ns) or "")
|
||||
out.append((" ".join(title.split()), link, " ".join(desc.split())[:500]))
|
||||
return out
|
||||
|
||||
|
||||
def _parse_page(body: bytes) -> list[tuple[str, str, str]]:
|
||||
"""Return (title=anchor text, link=href, description=page title) tuples."""
|
||||
parser = _LinkCollector()
|
||||
parser.feed(body.decode("utf-8", errors="replace"))
|
||||
page_title = " ".join(parser.title.split())
|
||||
return [(text, href, f"linked from page: {page_title}") for href, text in parser.links]
|
||||
|
||||
|
||||
def _classify(title: str, description: str) -> str:
|
||||
blob = f"{title}\n{description}"
|
||||
for kind, pattern in _KIND_PATTERNS:
|
||||
if pattern.search(blob):
|
||||
return kind
|
||||
return DEFAULT_KIND
|
||||
|
||||
|
||||
def _score(title: str, description: str, keywords: list[str]) -> float:
|
||||
blob = f"{title}\n{description}".lower()
|
||||
hits = sum(1 for kw in keywords if kw.lower() in blob)
|
||||
return round(min(hits / max(1, len(keywords)), 1.0), 3)
|
||||
|
||||
|
||||
def scan(cfg: Config, sess, fetch=_fetch) -> dict:
|
||||
"""Run one scan pass against all configured sources. Returns run stats."""
|
||||
sources = json.loads(cfg.scanner_sources or "[]")
|
||||
stats = {"sources": len(sources), "candidates": 0, "proposed": 0, "skipped_existing": 0}
|
||||
existing = {
|
||||
key
|
||||
for (key,) in sess.query(Proposal.dedupe_key).filter(
|
||||
Proposal.dedupe_key.isnot(None)
|
||||
)
|
||||
}
|
||||
|
||||
for src in sources:
|
||||
url = src.get("url", "")
|
||||
stype = src.get("type", "page")
|
||||
if not url:
|
||||
continue
|
||||
try:
|
||||
body = fetch(url, cfg)
|
||||
except (urllib.error.URLError, OSError) as exc:
|
||||
stats.setdefault("source_errors", []).append({"url": url, "error": str(exc)[:200]})
|
||||
continue
|
||||
items = _parse_rss(body) if stype == "rss" else _parse_page(body)
|
||||
stats["candidates"] += len(items)
|
||||
|
||||
for title, link, desc in items:
|
||||
text = title or link
|
||||
if not text:
|
||||
continue
|
||||
keywords = cfg.scanner_keywords
|
||||
score = _score(title or link, desc, keywords)
|
||||
if score <= 0 and stype != "page":
|
||||
continue # feeds must match at least one keyword; pages keep links
|
||||
dedupe = hashlib.sha1((link or title).encode()).hexdigest()[:32] # noqa: S324
|
||||
if dedupe in existing:
|
||||
stats["skipped_existing"] += 1
|
||||
continue
|
||||
proposal = Proposal(
|
||||
source=ProposalSource.SCANNER,
|
||||
kind=_classify(text, desc),
|
||||
title=text[:512],
|
||||
url=(link or None)[:1024] if (link or None) else None,
|
||||
detail={
|
||||
"source_type": stype,
|
||||
"source_url": url,
|
||||
"summary": desc,
|
||||
"scanned_at": datetime.now(UTC).isoformat(),
|
||||
},
|
||||
risk_score=score,
|
||||
status=ProposalStatus.PROPOSED,
|
||||
dedupe_key=dedupe,
|
||||
)
|
||||
sess.add(proposal)
|
||||
sess.flush()
|
||||
audit(
|
||||
sess,
|
||||
actor="daemon:scanner",
|
||||
action="propose",
|
||||
proposal_id=proposal.id,
|
||||
detail={"kind": proposal.kind, "score": score, "source": stype},
|
||||
)
|
||||
existing.add(dedupe)
|
||||
stats["proposed"] += 1
|
||||
return stats
|
||||
|
||||
|
||||
def run(cfg: Config | None = None) -> int:
|
||||
"""CLI entry: one scan pass, wrapped in a DaemonRun record."""
|
||||
from cayceops.config import load
|
||||
from cayceops.db import engine
|
||||
|
||||
cfg = cfg or load()
|
||||
try:
|
||||
with session_scope() as sess:
|
||||
run_row = DaemonRun(daemon="scanner")
|
||||
sess.add(run_row)
|
||||
sess.flush()
|
||||
try:
|
||||
stats = scan(cfg, sess)
|
||||
except Exception as exc: # noqa: BLE001 — daemon boundary: record and fail cleanly
|
||||
run_row.status = "error"
|
||||
run_row.detail = {"error": str(exc)[:300]}
|
||||
run_row.finished_at = datetime.now(UTC)
|
||||
return 1
|
||||
run_row.status = "ok"
|
||||
run_row.detail = stats
|
||||
run_row.finished_at = datetime.now(UTC)
|
||||
return 0
|
||||
finally:
|
||||
engine().dispose()
|
||||
@@ -0,0 +1,133 @@
|
||||
"""executor tests — human gate, autonomy bounds, retries, audit chain."""
|
||||
|
||||
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):
|
||||
p = _proposal()
|
||||
sess.add(p)
|
||||
sess.flush()
|
||||
|
||||
assert executor.approve(p.id, "gustavo", cfg(), sess) == 0
|
||||
# second approve must fail (state is now approved)
|
||||
assert executor.approve(p.id, "gustavo", cfg(), sess) == 1
|
||||
assert p.approved_by == "gustavo"
|
||||
assert p.status == ProposalStatus.APPROVED
|
||||
|
||||
|
||||
def test_reject_records_actor(cfg, sess):
|
||||
p = _proposal()
|
||||
sess.add(p)
|
||||
sess.flush()
|
||||
assert executor.reject(p.id, "gustavo", cfg(), sess) == 0
|
||||
assert p.status == ProposalStatus.REJECTED
|
||||
|
||||
|
||||
def test_run_executes_only_approved(cfg, sess):
|
||||
approved = _proposal(title="Approved item")
|
||||
proposed = _proposal(title="Proposed item")
|
||||
sess.add_all([approved, proposed])
|
||||
sess.flush()
|
||||
executor.approve(approved.id, "gustavo", cfg(), sess)
|
||||
|
||||
config = cfg() # AUTONOMOUS_MODE=false
|
||||
assert executor.run(config, sess) == 0
|
||||
|
||||
assert approved.status == ProposalStatus.DONE
|
||||
assert proposed.status == 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 == ProposalStatus.PROPOSED
|
||||
|
||||
|
||||
def test_autonomy_respects_kinds_and_daily_cap(cfg, sess):
|
||||
p1 = _proposal(kind="content_gap", title="auto ok")
|
||||
p2 = _proposal(kind="content_gap", title="auto over cap")
|
||||
p3 = _proposal(kind="affiliate", title="kind not allowed")
|
||||
sess.add_all([p1, p2, p3])
|
||||
sess.flush()
|
||||
|
||||
config = cfg(
|
||||
CAYCE_OPS_AUTONOMOUS_MODE="true",
|
||||
CAYCE_OPS_AUTONOMOUS_KINDS="content_gap",
|
||||
CAYCE_OPS_AUTONOMOUS_DAILY_CAP="1",
|
||||
)
|
||||
approved = executor._auto_approve(config, sess)
|
||||
assert approved == 1
|
||||
assert p1.status == ProposalStatus.APPROVED
|
||||
assert p2.status == ProposalStatus.PROPOSED # cap reached
|
||||
assert p3.status == ProposalStatus.PROPOSED # kind not in policy
|
||||
|
||||
actor = sess.query(AuditEntry).filter_by(action="approve").one().actor
|
||||
assert actor == "auto:policy"
|
||||
|
||||
|
||||
def test_failing_plugin_retries_then_fails(cfg, sess, monkeypatch):
|
||||
p = _proposal(kind="affiliate", title="Will fail")
|
||||
sess.add(p)
|
||||
sess.flush()
|
||||
executor.approve(p.id, "gustavo", cfg(), sess)
|
||||
|
||||
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 == ProposalStatus.APPROVED
|
||||
assert p.attempts == 1
|
||||
|
||||
executor.run(config, sess) # attempt 2 -> failed for good
|
||||
assert p.status == 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):
|
||||
p = _proposal(kind="brand_new_kind", title="No plugin yet")
|
||||
sess.add(p)
|
||||
sess.flush()
|
||||
executor.approve(p.id, "gustavo", cfg(), sess)
|
||||
|
||||
assert executor.run(cfg(), sess) == 0
|
||||
assert p.status == ProposalStatus.DONE
|
||||
|
||||
done = sess.query(AuditEntry).filter_by(action="execute_done").one()
|
||||
assert done.detail["outcome"]["plugin_fallback"] is True
|
||||
@@ -0,0 +1,112 @@
|
||||
"""scanner tests — discovery, classification, dedupe. No network (injected fetch)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
|
||||
from cayceops.components.scanner import scanner
|
||||
from cayceops.models import Proposal, ProposalStatus
|
||||
|
||||
RSS = b"""<?xml version="1.0"?>
|
||||
<rss><channel>
|
||||
<item><title>Acme launches affiliate program for hotels</title>
|
||||
<link>https://acme.example.org/affiliates</link>
|
||||
<description>Earn commission on every booking.</description></item>
|
||||
<item><title>Weekly crypto digest</title>
|
||||
<link>https://news.example.org/digest</link>
|
||||
<description>Everything unrelated.</description></item>
|
||||
</channel></rss>"""
|
||||
|
||||
PAGE = b"""<html><head><title>Partners</title></head><body>
|
||||
<a href="https://pay.example.org/partner">Become a payment partner</a>
|
||||
<a href="https://pay.example.org/docs">API docs</a>
|
||||
</body></html>"""
|
||||
|
||||
|
||||
def _fake_fetch(responses):
|
||||
def fetch(url, cfg, timeout=20.0):
|
||||
return responses[url]
|
||||
|
||||
return fetch
|
||||
|
||||
|
||||
SOURCES = [
|
||||
{"type": "rss", "url": "https://feeds.example.org/news.xml"},
|
||||
{"type": "page", "url": "https://pay.example.org/partners"},
|
||||
]
|
||||
|
||||
|
||||
def _cfg(cfg, **kw):
|
||||
return cfg(
|
||||
CAYCE_OPS_SCANNER_SOURCES=json.dumps(SOURCES),
|
||||
CAYCE_OPS_SCANNER_KEYWORDS="affiliate,partner,bitcoin,commission",
|
||||
**kw,
|
||||
)
|
||||
|
||||
|
||||
def test_scan_files_proposals_with_kind_and_score(cfg, sess):
|
||||
config = _cfg(cfg)
|
||||
stats = scanner.scan(config, sess, fetch=_fake_fetch({
|
||||
"https://feeds.example.org/news.xml": RSS,
|
||||
"https://pay.example.org/partners": PAGE,
|
||||
}))
|
||||
|
||||
assert stats["proposed"] == 3 # affiliate item + both page links (pages keep all)
|
||||
proposals = sess.query(Proposal).all()
|
||||
by_title = {p.title: p for p in proposals}
|
||||
|
||||
affiliate = by_title["Acme launches affiliate program for hotels"]
|
||||
assert affiliate.kind == "affiliate"
|
||||
assert affiliate.status == ProposalStatus.PROPOSED
|
||||
assert affiliate.risk_score > 0
|
||||
assert affiliate.dedupe_key
|
||||
|
||||
partner = by_title["Become a payment partner"]
|
||||
assert partner.kind == "partnership"
|
||||
|
||||
|
||||
def test_scan_is_idempotent_via_dedupe(cfg, sess):
|
||||
config = _cfg(cfg)
|
||||
scanner.scan(config, sess, fetch=_fake_fetch({
|
||||
"https://feeds.example.org/news.xml": RSS,
|
||||
"https://pay.example.org/partners": PAGE,
|
||||
}))
|
||||
stats2 = scanner.scan(config, sess, fetch=_fake_fetch({
|
||||
"https://feeds.example.org/news.xml": RSS,
|
||||
"https://pay.example.org/partners": PAGE,
|
||||
}))
|
||||
assert stats2["proposed"] == 0
|
||||
assert stats2["skipped_existing"] == 3
|
||||
|
||||
|
||||
def test_scan_filters_keywordless_feed_items(cfg, sess):
|
||||
config = _cfg(cfg)
|
||||
scanner.scan(config, sess, fetch=_fake_fetch({
|
||||
"https://feeds.example.org/news.xml": RSS,
|
||||
"https://pay.example.org/partners": PAGE,
|
||||
}))
|
||||
titles = {p.title for p in sess.query(Proposal).all()}
|
||||
assert "Weekly crypto digest" not in titles # no keyword match in feed -> dropped
|
||||
|
||||
|
||||
def test_scan_records_audit_for_each_proposal(cfg, sess):
|
||||
from cayceops.models import AuditEntry
|
||||
|
||||
config = _cfg(cfg)
|
||||
scanner.scan(config, sess, fetch=_fake_fetch({
|
||||
"https://feeds.example.org/news.xml": RSS,
|
||||
"https://pay.example.org/partners": PAGE,
|
||||
}))
|
||||
acts = [a.action for a in sess.query(AuditEntry).all()]
|
||||
assert acts.count("propose") == 3
|
||||
assert all(a.actor == "daemon:scanner" for a in sess.query(AuditEntry).all())
|
||||
|
||||
|
||||
def test_scan_survives_source_error(cfg, sess):
|
||||
def broken_fetch(url, cfg, timeout=20.0):
|
||||
raise OSError("connection refused")
|
||||
|
||||
config = _cfg(cfg)
|
||||
stats = scanner.scan(config, sess, fetch=broken_fetch)
|
||||
assert stats["proposed"] == 0
|
||||
assert len(stats["source_errors"]) == 2
|
||||
Reference in New Issue
Block a user