DataOps-platform/app/core/mcp/canary.py

"""Trusted Canary verification based on engine state and bounded comparisons."""

from __future__ import annotations

TERMINAL_FAILURE_STATES = {
    "FAILED",
    "KILLED",
    "CANCELLED",
    "WARNING",
}
SAFE_METRICS = {
    "row_count_delta",
    "duration_delta_seconds",
    "error_count_delta",
    "output_hash_match",
}


class CanaryVerifier:
    def __init__(
        self,
        *,
        plans,
        engine,
        comparator=None,
        verifier_id="dataops-canary-verifier",
    ):
        self.plans = plans
        self.engine = engine
        self.comparator = comparator
        self.verifier_id = verifier_id

    def verify(self, candidate_id, *, baseline_execution_id=None):
        evidence = self.plans.get_canary_evidence(candidate_id)
        if not isinstance(evidence, dict):
            raise ValueError("canary evidence does not exist")
        if evidence.get("status") != "started":
            raise ValueError("canary evidence is already finalized")
        execution_id = evidence.get("execution_id")
        execution = self.engine.get_execution(execution_id)
        state = str((execution.get("state") or {}).get("current") or "").upper()
        if state not in {"SUCCESS", *TERMINAL_FAILURE_STATES}:
            return evidence

        equivalent = state == "SUCCESS"
        metrics = {}
        if state == "SUCCESS" and baseline_execution_id is not None:
            if self.comparator is None:
                raise RuntimeError("canary comparator is not configured")
            comparison = self.comparator.compare(baseline_execution_id, execution_id)
            equivalent = bool(comparison.get("equivalent", False))
            metrics = {
                key: value
                for key, value in dict(comparison.get("metrics") or {}).items()
                if key in SAFE_METRICS
                and (value is None or isinstance(value, (bool, int, float)))
            }

        status = "passed" if state == "SUCCESS" and equivalent else "failed"
        summary = {
            "engine_state": state,
            "equivalent": equivalent,
            "metrics": metrics,
        }
        finalized = {
            "candidate_id": candidate_id,
            "execution_id": execution_id,
            "status": status,
            "sample_runs": 1,
            "verified_by": self.verifier_id,
            "verification_summary": summary,
        }
        return self.plans.record_canary_verification(candidate_id, finalized)