scanner + executor: discovery files proposals; execution gated on approval

scanner (read-only):
- sources from CAYCE_OPS_SCANNER_SOURCES JSON (rss + page types), stdlib only
- keyword scoring -> risk_score; kind via heuristics (affiliate/partnership/content_gap)
- idempotent: sha1 dedupe_key; repeated scans skip existing
- every proposal audited (actor daemon:scanner); source errors recorded, never fatal

executor (human gate):
- approve/reject audited with operator identity; only proposed -> approved
- AUTONOMOUS_MODE=true adds exactly one path: policy auto-approval
  (kinds allowlist + daily cap, actor auto:policy) — DECISIONS D5
- plugin registry by kind; unknown kinds fall back to log_action (D9)
- retries with max_attempts, then failed; every transition audited
- http_post plugin inert unless CAYCE_OPS_ENABLE_HTTP_ACTIONS=true

tests: 13 new (dedupe, keyword filter, autonomy bounds, retry-then-fail,
blocked http_post, fallback plugin) — 20 total passing

cayceops/components/executor/__init__.py

@@ -0,0 +1 @@
+"""executor package."""

cayceops/components/executor/executor.py

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

cayceops/components/executor/plugins.py

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

cayceops/components/scanner/__init__.py

@@ -0,0 +1 @@
+"""scanner package."""

cayceops/components/scanner/scanner.py

@@ -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()

tests/test_executor.py

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

tests/test_scanner.py

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