Add auditable research candidate promotion policy

docs/research-discovery.md

@@ -97,12 +97,16 @@ secopsai research monitor create --ecosystem nuget --watchlist-id WL-... --inter
 secopsai research monitor list
 secopsai research monitor run-due --limit 25
 secopsai research candidate list
+secopsai research candidate promotion-policy --ecosystem all
+secopsai research candidate run-promotion-policy --ecosystem all
 secopsai research campaign correlate
 secopsai research campaign list
 ```
 
 The dashboard provides the same actions under Research Discovery. Use **Add watchlist**, **Create monitor**, **Run due monitors**, **Compare exact packages**, and **Correlate campaigns**. Protected writes require the research action token.
 
+Candidate promotion is disabled by default. Configure it with `research candidate promotion-policy --set`, then run `run-promotion-policy` without `--apply` to preview every decision. The policy evaluates a minimum score, evidence-reference count, optional publisher evidence, and ecosystem scope. Adding `--apply` may create draft Research Cases only; it never records a malicious verdict. Each promotion persists the candidate, exact policy, reasons, actor, resulting case ID, and event timestamp for audit and rollback review.
+
 On macOS, install the local due-monitor trigger as a background service:
 
 ```bash

secopsai/cli.py

@@ -137,6 +137,7 @@
     create_monitor as create_research_monitor,
     create_watchlist as create_research_watchlist,
     get_candidate as get_research_candidate,
+    get_promotion_policy as get_research_promotion_policy,
     ingest_registry_metadata,
     list_candidates as list_research_candidates,
     list_alerts as list_research_alerts,
@@ -146,6 +147,8 @@
     run_monitor as run_research_monitor,
     run_due_monitors as run_due_research_monitors,
     recover_stale_monitor_runs,
+    run_promotion_policy as run_research_promotion_policy,
+    set_promotion_policy as set_research_promotion_policy,
 )
 from secopsai.research_surveillance import (
     collector_status as registry_collector_status,
@@ -1314,6 +1317,22 @@ def parse_args(argv: Optional[List[str]] = None) -> argparse.Namespace:
     candidate_show = research_candidate_sub.add_parser("show")
     candidate_show.add_argument("candidate_id")
     candidate_show.add_argument("--db-path", default=None)
+    candidate_policy = research_candidate_sub.add_parser("promotion-policy")
+    candidate_policy.add_argument("--ecosystem", default="all")
+    candidate_policy.add_argument("--set", action="store_true")
+    candidate_policy.add_argument("--enabled", choices=["true", "false"], default=None)
+    candidate_policy.add_argument("--score-threshold", type=float, default=90.0)
+    candidate_policy.add_argument("--minimum-evidence", type=int, default=2)
+    candidate_policy.add_argument("--require-publisher", action="store_true")
+    candidate_policy.add_argument("--mode", choices=["review_only", "draft_case"], default="draft_case")
+    candidate_policy.add_argument("--actor", default="operator")
+    candidate_policy.add_argument("--db-path", default=None)
+    candidate_promote = research_candidate_sub.add_parser("run-promotion-policy")
+    candidate_promote.add_argument("--ecosystem", default="all")
+    candidate_promote.add_argument("--apply", action="store_true")
+    candidate_promote.add_argument("--actor", default="operator")
+    candidate_promote.add_argument("--limit", type=int, default=100)
+    candidate_promote.add_argument("--db-path", default=None)
 
     research_collect = research_sub.add_parser("collect", help="Run global registry feed collectors")
     research_collect_sub = research_collect.add_subparsers(dest="research_collect_cmd", required=True)
@@ -2404,6 +2423,13 @@ def _run_research_automation_command(args: argparse.Namespace) -> int:
         elif args.research_cmd == "candidate":
             if args.research_candidate_cmd == "list":
                 payload = {"candidates": list_research_candidates(status=args.status, ecosystem=args.ecosystem, limit=args.limit, db_path=args.db_path)}
+            elif args.research_candidate_cmd == "promotion-policy":
+                if args.set:
+                    payload = set_research_promotion_policy(ecosystem=args.ecosystem, enabled=args.enabled == "true", score_threshold=args.score_threshold, minimum_evidence=args.minimum_evidence, require_publisher=args.require_publisher, mode=args.mode, actor=args.actor, db_path=args.db_path)
+                else:
+                    payload = get_research_promotion_policy(ecosystem=args.ecosystem, db_path=args.db_path)
+            elif args.research_candidate_cmd == "run-promotion-policy":
+                payload = run_research_promotion_policy(ecosystem=args.ecosystem, apply=args.apply, actor=args.actor, limit=args.limit, db_path=args.db_path)
             else:
                 payload = get_research_candidate(args.candidate_id, db_path=args.db_path)
         elif args.research_cmd == "collect":

secopsai/research_discovery.py

@@ -26,6 +26,7 @@
 WATCH_TYPES = {"package", "namespace", "publisher", "brand", "repository", "organization"}
 PRIORITIES = {"low", "normal", "high", "critical"}
 STATUSES = {"active", "paused", "archived", "new", "review", "dismissed", "promoted"}
+PROMOTION_MODES = {"review_only", "draft_case"}
 
 
 def _id(prefix: str) -> str:
@@ -516,6 +517,78 @@ def get_candidate(candidate_id: str, *, db_path: Optional[str] = None) -> Dict[s
     return item
 
 
+def get_promotion_policy(*, ecosystem: str = "all", db_path: Optional[str] = None) -> Dict[str, Any]:
+    """Return the durable, auditable candidate-promotion policy."""
+    soc_store.init_db(db_path)
+    ecosystem = normalize_identifier("", ecosystem or "all")
+    with closing(soc_store.connect(db_path)) as connection:
+        row = connection.execute("SELECT * FROM research_promotion_policies WHERE ecosystem = ?", (ecosystem,)).fetchone()
+    if row:
+        return dict(row)
+    return {"ecosystem": ecosystem, "enabled": 0, "score_threshold": 90.0, "minimum_evidence": 2, "require_publisher": 0, "mode": "draft_case", "updated_by": "", "created_at": None, "updated_at": None}
+
+
+def set_promotion_policy(*, ecosystem: str = "all", enabled: bool = False, score_threshold: float = 90.0, minimum_evidence: int = 2, require_publisher: bool = False, mode: str = "draft_case", actor: str = "operator", db_path: Optional[str] = None) -> Dict[str, Any]:
+    soc_store.init_db(db_path)
+    ecosystem = normalize_identifier("", ecosystem or "all")
+    if ecosystem != "all" and ecosystem not in CAPABILITIES:
+        raise ValueError("unsupported promotion-policy ecosystem")
+    threshold = float(score_threshold)
+    if not 70 <= threshold <= 100:
+        raise ValueError("score threshold must be between 70 and 100")
+    evidence_count = max(1, min(int(minimum_evidence), 8))
+    if mode not in PROMOTION_MODES:
+        raise ValueError("unsupported promotion mode")
+    now = _now()
+    with closing(soc_store.connect(db_path)) as connection:
+        connection.execute("""INSERT INTO research_promotion_policies
+            (ecosystem, enabled, score_threshold, minimum_evidence, require_publisher, mode, updated_by, created_at, updated_at)
+            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
+            ON CONFLICT(ecosystem) DO UPDATE SET enabled=excluded.enabled, score_threshold=excluded.score_threshold,
+            minimum_evidence=excluded.minimum_evidence, require_publisher=excluded.require_publisher,
+            mode=excluded.mode, updated_by=excluded.updated_by, updated_at=excluded.updated_at""",
+            (ecosystem, int(bool(enabled)), threshold, evidence_count, int(bool(require_publisher)), mode, str(actor or "operator")[:160], now, now))
+        connection.commit()
+    return get_promotion_policy(ecosystem=ecosystem, db_path=db_path)
+
+
+def run_promotion_policy(*, ecosystem: str = "all", apply: bool = False, actor: str = "operator", limit: int = 100, db_path: Optional[str] = None) -> Dict[str, Any]:
+    """Evaluate candidates deterministically and optionally create draft cases.
+
+    Promotion never records a malicious verdict. It creates an auditable draft
+    investigation only when the configured evidence gates pass.
+    """
+    policy = get_promotion_policy(ecosystem=ecosystem, db_path=db_path)
+    candidates = list_candidates(status="new", ecosystem=None if policy["ecosystem"] == "all" else policy["ecosystem"], limit=limit, db_path=db_path)
+    decisions: List[Dict[str, Any]] = []
+    for candidate in candidates:
+        evidence = candidate.get("evidence") if isinstance(candidate.get("evidence"), dict) else {}
+        evidence_refs = [key for key in ("metadata_url", "artifact_url", "publisher") if evidence.get(key)]
+        reasons = []
+        if float(candidate.get("score") or 0) < float(policy["score_threshold"]):
+            reasons.append("score_below_threshold")
+        if len(evidence_refs) < int(policy["minimum_evidence"]):
+            reasons.append("insufficient_evidence")
+        if policy["require_publisher"] and not evidence.get("publisher"):
+            reasons.append("publisher_missing")
+        eligible = not reasons and bool(policy["enabled"])
+        decision = {"candidate_id": candidate["candidate_id"], "eligible": eligible, "score": candidate["score"], "evidence_count": len(evidence_refs), "reasons": reasons or (["policy_disabled"] if not policy["enabled"] else ["eligible"]), "case_id": candidate.get("case_id")}
+        if apply and eligible and policy["mode"] == "draft_case" and not candidate.get("case_id"):
+            from secopsai.research_cases import add_evidence, add_subject, create_case
+            case = create_case(title=f"Investigate {candidate['ecosystem']} package {candidate['package']}", summary=f"Automatically promoted as a draft investigation after deterministic policy checks. Similarity score: {candidate['score']}. This is a lead, not a maliciousness verdict.", case_type="typosquatting", severity="medium", confidence=min(int(float(candidate["score"])), 80), owner=actor, metadata={"candidate_id": candidate["candidate_id"], "promotion_policy": policy}, db_path=db_path)
+            add_subject(case["case_id"], subject_type="package", name=candidate["package"], ecosystem=candidate["ecosystem"], version=candidate.get("version") or "", publisher=evidence.get("publisher") or "", registry_state="available", metadata={"candidate_id": candidate["candidate_id"], "reference_identifier": candidate.get("reference_identifier")}, db_path=db_path, actor=actor)
+            if evidence.get("metadata_url"):
+                add_evidence(case["case_id"], evidence_type="registry_metadata", title="Registry metadata captured during candidate discovery", locator=evidence["metadata_url"], provenance="official registry monitor", notes=candidate.get("reason") or "", metadata={"candidate_id": candidate["candidate_id"]}, db_path=db_path, actor=actor)
+            with closing(soc_store.connect(db_path)) as connection:
+                connection.execute("UPDATE research_candidates SET status = 'promoted', case_id = ? WHERE candidate_id = ?", (case["case_id"], candidate["candidate_id"]))
+                connection.execute("INSERT OR IGNORE INTO research_promotion_events (event_id, candidate_id, policy_ecosystem, decision, reasons_json, case_id, actor, created_at) VALUES (?, ?, ?, 'draft_case_created', ?, ?, ?, ?)", (_id("RPE"), candidate["candidate_id"], policy["ecosystem"], _json(decision["reasons"]), case["case_id"], actor, _now()))
+                connection.commit()
+            decision["case_id"] = case["case_id"]
+            decision["applied"] = True
+        decisions.append(decision)
+    return {"schema_version": "secopsai.research.promotion-policy.v1", "policy": policy, "apply": bool(apply), "evaluated": len(decisions), "eligible": sum(bool(item["eligible"]) for item in decisions), "promoted": sum(bool(item.get("applied")) for item in decisions), "decisions": decisions}
+
+
 def create_candidate_alert(candidate: Dict[str, Any], *, db_path: Optional[str] = None) -> Dict[str, Any]:
     """Create a deduplicated alert for a high-confidence candidate."""
     soc_store.init_db(db_path)

soc_store.py

@@ -721,6 +721,32 @@ def init_db(db_path: str | None = None) -> None:
                 FOREIGN KEY (case_id) REFERENCES research_cases (case_id) ON DELETE SET NULL
             );
 
+            CREATE TABLE IF NOT EXISTS research_promotion_policies (
+                ecosystem TEXT PRIMARY KEY,
+                enabled INTEGER NOT NULL DEFAULT 0,
+                score_threshold REAL NOT NULL DEFAULT 90,
+                minimum_evidence INTEGER NOT NULL DEFAULT 2,
+                require_publisher INTEGER NOT NULL DEFAULT 0,
+                mode TEXT NOT NULL DEFAULT 'draft_case',
+                updated_by TEXT NOT NULL,
+                created_at TEXT NOT NULL,
+                updated_at TEXT NOT NULL
+            );
+
+            CREATE TABLE IF NOT EXISTS research_promotion_events (
+                event_id TEXT PRIMARY KEY,
+                candidate_id TEXT NOT NULL,
+                policy_ecosystem TEXT NOT NULL,
+                decision TEXT NOT NULL,
+                reasons_json TEXT NOT NULL,
+                case_id TEXT,
+                actor TEXT NOT NULL,
+                created_at TEXT NOT NULL,
+                UNIQUE(candidate_id, policy_ecosystem, decision),
+                FOREIGN KEY (candidate_id) REFERENCES research_candidates (candidate_id) ON DELETE CASCADE,
+                FOREIGN KEY (case_id) REFERENCES research_cases (case_id) ON DELETE SET NULL
+            );
+
             CREATE TABLE IF NOT EXISTS research_campaigns (
                 campaign_id TEXT PRIMARY KEY,
                 title TEXT NOT NULL,

tests/test_research_promotion_policy.py

@@ -0,0 +1,56 @@
+from __future__ import annotations
+
+import json
+import tempfile
+from pathlib import Path
+
+import soc_store
+from secopsai.research_discovery import get_promotion_policy, run_promotion_policy, set_promotion_policy
+
+
+def _candidate(db_path: str, *, score: float = 96.0) -> str:
+    soc_store.init_db(db_path)
+    candidate_id = "CAN-POLICYTEST0001"
+    with soc_store.connect(db_path) as connection:
+        connection.execute(
+            """INSERT INTO research_candidates
+            (candidate_id, event_id, watchlist_id, ecosystem, package, version, reference_identifier,
+             score, score_components_json, reason, status, case_id, evidence_json, first_seen,
+             last_seen, algorithm_version)
+            VALUES (?, NULL, NULL, 'nuget', 'Braintree.Net.Test', '1.0.0', 'Braintree.Net', ?, '{}',
+                    'test similarity candidate', 'new', NULL, ?, '2026-07-29T00:00:00Z',
+                    '2026-07-29T00:00:00Z', 'similarity-1')""",
+            (candidate_id, score, json.dumps({"metadata_url": "https://api.nuget.org/v3/registration5-gz-semver2/braintree.net.test/index.json", "artifact_url": "https://api.nuget.org/v3-flatcontainer/braintree.net.test/1.0.0/braintree.net.test.1.0.0.nupkg", "publisher": "unexpected"})),
+        )
+        connection.commit()
+    return candidate_id
+
+
+def test_promotion_policy_is_disabled_and_preview_only_by_default() -> None:
+    with tempfile.TemporaryDirectory() as temp_dir:
+        db_path = str(Path(temp_dir) / "soc.db")
+        _candidate(db_path)
+        policy = get_promotion_policy(ecosystem="nuget", db_path=db_path)
+        assert policy["enabled"] == 0
+        preview = run_promotion_policy(ecosystem="nuget", db_path=db_path)
+        assert preview["promoted"] == 0
+        assert preview["decisions"][0]["eligible"] is False
+
+
+def test_enabled_policy_creates_auditable_draft_case_without_a_verdict() -> None:
+    with tempfile.TemporaryDirectory() as temp_dir:
+        db_path = str(Path(temp_dir) / "soc.db")
+        candidate_id = _candidate(db_path)
+        set_promotion_policy(ecosystem="nuget", enabled=True, score_threshold=90, minimum_evidence=2, require_publisher=True, actor="test-operator", db_path=db_path)
+        result = run_promotion_policy(ecosystem="nuget", apply=True, actor="test-operator", db_path=db_path)
+        assert result["promoted"] == 1
+        assert result["decisions"][0]["case_id"].startswith("RSC-")
+        with soc_store.connect(db_path) as connection:
+            candidate = connection.execute("SELECT status, case_id FROM research_candidates WHERE candidate_id = ?", (candidate_id,)).fetchone()
+            case = connection.execute("SELECT status, payload_json FROM research_cases WHERE case_id = ?", (candidate["case_id"],)).fetchone()
+            event = connection.execute("SELECT decision, actor FROM research_promotion_events WHERE candidate_id = ?", (candidate_id,)).fetchone()
+        assert candidate["status"] == "promoted"
+        assert case["status"] == "draft"
+        assert "promotion_policy" in case["payload_json"]
+        assert event["decision"] == "draft_case_created"
+        assert event["actor"] == "test-operator"