Decision Logic Change Risk Chain:

  Modify rule threshold        Modify tree branch         Modify constraints
       |                            |                          |
       v                            v                          v
  +-------------------------------------------------------------+
  |              Deployed without validation                     |
  |  -> Approval rate anomaly (spike or drop)                    |
  |  -> Compliance risk (e.g., wrongful credit rejection)        |
  |  -> Business loss (e.g., uneven inventory allocation)        |
  +-------------------------------------------------------------+

Dry-Run Mode Overview:

  +----------------+    +----------------+    +----------------+
  | Shadow Mode    |    | What-If Mode   |    | Replay Mode    |
  +-------+--------+    +-------+--------+    +-------+--------+
          |                     |                     |
          v                     v                     v
  Parallel execution     Manual scenario        Historical data
  on production traffic  construction           batch replay
  Compare old vs new     Simulation             Statistical comparison
  Zero risk              Exploratory            Comprehensive

from __future__ import annotations
import asyncio
from dataclasses import dataclass, field
from datetime import datetime
from typing import Any


@dataclass
class ShadowConfig:
    """Shadow mode configuration"""
    shadow_id: str
    description: str
    primary_version: str           # Current production version
    shadow_version: str            # Candidate version
    sample_rate: float = 1.0       # Sampling rate 0.0-1.0
    enabled: bool = True
    start_time: datetime | None = None
    end_time: datetime | None = None
    tags: dict[str, str] = field(default_factory=dict)


@dataclass
class ShadowResult:
    """Shadow comparison result"""
    request_id: str
    primary_decision: str
    shadow_decision: str
    primary_confidence: float
    shadow_confidence: float
    is_divergent: bool
    divergence_details: dict[str, Any] = field(default_factory=dict)
    primary_latency_ms: float = 0.0
    shadow_latency_ms: float = 0.0
    timestamp: datetime = field(default_factory=datetime.utcnow)


class ShadowModeExecutor:
    """Shadow mode executor"""

    def __init__(self, primary_engine, shadow_engine,
                 config: ShadowConfig):
        self._primary = primary_engine
        self._shadow = shadow_engine
        self._config = config
        self._results: list[ShadowResult] = []
        self._divergence_count = 0
        self._total_count = 0

    async def execute(self, context) -> tuple[Any, ShadowResult | None]:
        """Execute shadow mode decision"""
        import time
        import random

        # Primary engine always executes (returned to user)
        start = time.monotonic()
        primary_result = self._primary.evaluate(context)
        primary_latency = (time.monotonic() - start) * 1000

        # Shadow engine executes based on sample rate
        shadow_result_obj = None
        if (self._config.enabled and
                random.random() < self._config.sample_rate):

            start = time.monotonic()
            try:
                shadow_result = self._shadow.evaluate(context)
                shadow_latency = (time.monotonic() - start) * 1000

                shadow_result_obj = self._compare(
                    context, primary_result, shadow_result,
                    primary_latency, shadow_latency
                )
                self._results.append(shadow_result_obj)
                self._total_count += 1

                if shadow_result_obj.is_divergent:
                    self._divergence_count += 1

            except Exception as e:
                shadow_result_obj = ShadowResult(
                    request_id=context.context_id,
                    primary_decision=primary_result.decision,
                    shadow_decision=f"ERROR: {e}",
                    primary_confidence=primary_result.confidence,
                    shadow_confidence=0.0,
                    is_divergent=True,
                    divergence_details={"error": str(e)},
                    primary_latency_ms=primary_latency,
                )

        return primary_result, shadow_result_obj

    def _compare(self, context, primary, shadow,
                 p_latency: float, s_latency: float) -> ShadowResult:
        is_divergent = primary.decision != shadow.decision
        details = {}

        if is_divergent:
            details["decision_change"] = {
                "from": primary.decision,
                "to": shadow.decision,
            }
            details["confidence_delta"] = (
                shadow.confidence - primary.confidence
            )

        conf_threshold = 0.1
        if abs(primary.confidence - shadow.confidence) > conf_threshold:
            details["confidence_shift"] = {
                "primary": primary.confidence,
                "shadow": shadow.confidence,
                "delta": shadow.confidence - primary.confidence,
            }

        return ShadowResult(
            request_id=context.context_id,
            primary_decision=primary.decision,
            shadow_decision=shadow.decision,
            primary_confidence=primary.confidence,
            shadow_confidence=shadow.confidence,
            is_divergent=is_divergent,
            divergence_details=details,
            primary_latency_ms=p_latency,
            shadow_latency_ms=s_latency,
        )

    @property
    def divergence_rate(self) -> float:
        if self._total_count == 0:
            return 0.0
        return self._divergence_count / self._total_count

    def generate_report(self) -> ShadowReport:
        return ShadowReportGenerator.generate(
            self._config, self._results
        )

@dataclass
class ShadowReport:
    """Shadow mode analysis report"""
    shadow_id: str
    total_requests: int
    divergent_requests: int
    divergence_rate: float
    avg_primary_latency_ms: float
    avg_shadow_latency_ms: float
    decision_distribution_primary: dict[str, int]
    decision_distribution_shadow: dict[str, int]
    top_divergence_patterns: list[dict]
    recommendation: str


class ShadowReportGenerator:
    """Shadow report generator"""

    @staticmethod
    def generate(config: ShadowConfig,
                 results: list[ShadowResult]) -> ShadowReport:
        total = len(results)
        divergent = [r for r in results if r.is_divergent]

        primary_dist: dict[str, int] = {}
        shadow_dist: dict[str, int] = {}
        for r in results:
            primary_dist[r.primary_decision] = (
                primary_dist.get(r.primary_decision, 0) + 1
            )
            shadow_dist[r.shadow_decision] = (
                shadow_dist.get(r.shadow_decision, 0) + 1
            )

        patterns: dict[str, int] = {}
        for r in divergent:
            key = f"{r.primary_decision} -> {r.shadow_decision}"
            patterns[key] = patterns.get(key, 0) + 1

        top_patterns = sorted(
            [{"pattern": k, "count": v} for k, v in patterns.items()],
            key=lambda x: x["count"], reverse=True
        )[:10]

        div_rate = len(divergent) / total if total > 0 else 0
        if div_rate < 0.01:
            rec = "Divergence rate very low (<1%). Recommend proceeding to production."
        elif div_rate < 0.05:
            rec = "Divergence rate acceptable (1-5%). Review divergent samples before deploying."
        elif div_rate < 0.15:
            rec = "Divergence rate elevated (5-15%). Analyze patterns and re-validate."
        else:
            rec = "Divergence rate too high (>15%). Do not deploy. Review logic changes."

        return ShadowReport(
            shadow_id=config.shadow_id,
            total_requests=total,
            divergent_requests=len(divergent),
            divergence_rate=div_rate,
            avg_primary_latency_ms=(
                sum(r.primary_latency_ms for r in results) / total
                if total > 0 else 0
            ),
            avg_shadow_latency_ms=(
                sum(r.shadow_latency_ms for r in results) / total
                if total > 0 else 0
            ),
            decision_distribution_primary=primary_dist,
            decision_distribution_shadow=shadow_dist,
            top_divergence_patterns=top_patterns,
            recommendation=rec,
        )

Shadow Mode Analysis Report:
===================================

  Config: shadow-credit-v3 (Sample Rate: 100%)
  Period: 2026-03-20 08:00 ~ 2026-03-20 20:00

  Total Requests:     12,847
  Divergent Requests:    641
  Divergence Rate:     4.99%

  Decision Distribution Comparison:
  ---------------------------------
  Decision Type     | Production | Shadow
  approve           |     6,421  |  6,892  (+7.3%)
  conditional       |     3,208  |  2,891  (-9.9%)
  reject            |     3,218  |  3,064  (-4.8%)

  Top Divergence Patterns:
  ---------------------------------
  reject -> conditional    :  312 (48.7%)
  conditional -> approve   :  198 (30.9%)
  approve -> conditional   :   89 (13.9%)
  conditional -> reject    :   42 ( 6.5%)

  Latency Comparison:
  ---------------------------------
  Production P50:  2.1ms    Shadow P50:  2.3ms
  Production P99:  8.7ms    Shadow P99:  9.1ms

  Recommendation: Divergence rate acceptable (1-5%). Review samples before deploying.

@dataclass
class WhatIfScenario:
    """Hypothetical scenario"""
    scenario_id: str
    name: str
    base_inputs: dict[str, Any]
    modifications: list[WhatIfModification]
    description: str = ""


@dataclass
class WhatIfModification:
    """Single hypothetical modification"""
    variable: str
    original_value: Any
    hypothetical_value: Any
    description: str = ""


@dataclass
class WhatIfResult:
    """Hypothetical analysis result"""
    scenario_id: str
    original_decision: str
    original_confidence: float
    hypothetical_decision: str
    hypothetical_confidence: float
    decision_changed: bool
    modifications_applied: list[WhatIfModification]
    sensitivity: dict[str, float]


class WhatIfAnalyzer:
    """What-If analyzer"""

    def __init__(self, decision_engine):
        self._engine = decision_engine

    def analyze(self, scenario: WhatIfScenario,
                context_builder) -> WhatIfResult:
        """Execute hypothetical analysis"""
        original_ctx = context_builder(scenario.base_inputs)
        original = self._engine.evaluate(original_ctx)

        modified_inputs = dict(scenario.base_inputs)
        for mod in scenario.modifications:
            modified_inputs[mod.variable] = mod.hypothetical_value

        modified_ctx = context_builder(modified_inputs)
        modified = self._engine.evaluate(modified_ctx)

        sensitivity = self._compute_sensitivity(
            scenario.base_inputs, scenario.modifications, context_builder
        )

        return WhatIfResult(
            scenario_id=scenario.scenario_id,
            original_decision=original.decision,
            original_confidence=original.confidence,
            hypothetical_decision=modified.decision,
            hypothetical_confidence=modified.confidence,
            decision_changed=original.decision != modified.decision,
            modifications_applied=scenario.modifications,
            sensitivity=sensitivity,
        )

    def _compute_sensitivity(self, base: dict,
                             modifications: list[WhatIfModification],
                             builder) -> dict[str, float]:
        """Compute independent sensitivity for each variable"""
        sensitivities = {}
        base_ctx = builder(base)
        base_result = self._engine.evaluate(base_ctx)

        for mod in modifications:
            single_mod = dict(base)
            single_mod[mod.variable] = mod.hypothetical_value
            mod_ctx = builder(single_mod)
            mod_result = self._engine.evaluate(mod_ctx)

            conf_delta = abs(mod_result.confidence - base_result.confidence)
            decision_change = 1.0 if mod_result.decision != base_result.decision else 0.0
            sensitivities[mod.variable] = conf_delta + decision_change

        return sensitivities

    def sweep(self, base_inputs: dict, variable: str,
              values: list, builder) -> list[dict]:
        """Variable sweep: trace decision changes across a range"""
        results = []
        for val in values:
            inputs = dict(base_inputs)
            inputs[variable] = val
            ctx = builder(inputs)
            result = self._engine.evaluate(ctx)
            results.append({
                "value": val,
                "decision": result.decision,
                "confidence": result.confidence,
            })
        return results

Variable Sweep: credit_score from 400 to 800

  credit_score | Decision          | Confidence
  -------------|-------------------|----------
  400          | reject            | 0.95
  450          | reject            | 0.92
  500          | reject            | 0.88
  550          | reject            | 0.78  <- uncertainty begins
  600          | conditional       | 0.65
  620          | conditional       | 0.72
  650          | conditional       | 0.80
  700          | approve           | 0.88  <- decision flip point
  750          | approve           | 0.93
  800          | approve           | 0.97

  Decision flip point: credit_score = 700
  Sensitive range: [550, 700] -- small changes significantly affect decision

from datetime import datetime, timedelta


@dataclass
class ReplayConfig:
    """Replay configuration"""
    replay_id: str
    source: str                      # "database", "file", "kafka"
    time_range: tuple[datetime, datetime]
    new_engine_version: str
    batch_size: int = 1000
    max_records: int = 100_000


@dataclass
class ReplayStats:
    """Replay statistics"""
    total: int = 0
    same_decision: int = 0
    different_decision: int = 0
    new_approve_rate: float = 0.0
    old_approve_rate: float = 0.0
    latency_improvement_pct: float = 0.0
    error_count: int = 0


class ReplayEngine:
    """Historical decision replay engine"""

    def __init__(self, old_engine, new_engine, decision_store):
        self._old = old_engine
        self._new = new_engine
        self._store = decision_store

    async def replay(self, config: ReplayConfig) -> ReplayStats:
        """Replay historical decisions"""
        stats = ReplayStats()
        old_approves = 0
        new_approves = 0

        records = await self._store.query(
            start=config.time_range[0],
            end=config.time_range[1],
            limit=config.max_records,
        )

        for batch_start in range(0, len(records), config.batch_size):
            batch = records[batch_start:batch_start + config.batch_size]

            for record in batch:
                try:
                    old_result = record["original_decision"]
                    new_result = self._new.evaluate(
                        self._rebuild_context(record)
                    )

                    stats.total += 1

                    if old_result == new_result.decision:
                        stats.same_decision += 1
                    else:
                        stats.different_decision += 1

                    if old_result in ("approve", "conditional_approve"):
                        old_approves += 1
                    if new_result.decision in ("approve", "conditional_approve"):
                        new_approves += 1

                except Exception:
                    stats.error_count += 1

        if stats.total > 0:
            stats.old_approve_rate = old_approves / stats.total
            stats.new_approve_rate = new_approves / stats.total

        return stats

    def _rebuild_context(self, record: dict):
        builder = DecisionContextBuilder(record["domain"])
        for k, v in record.get("inputs", {}).items():
            builder = builder.with_inputs(**{k: v})
        return builder.build()

Historical Replay Report:
===================================

  Replay ID: replay-credit-v3-upgrade
  Time Range: 2026-02-01 ~ 2026-03-01
  Total Records: 89,421

  Decision Consistency:
  ---------------------------------
  Same:      84,602 (94.6%)
  Different:  4,819 ( 5.4%)

  Approval Rate Comparison:
  ---------------------------------
  Old version:  62.3%
  New version:  64.8%  (+2.5pp)

  Divergence Details:
  ---------------------------------
  reject -> approve        :  2,104 (43.7%)
  reject -> conditional    :  1,287 (26.7%)
  conditional -> approve   :    891 (18.5%)
  approve -> conditional   :    412 ( 8.5%)
  approve -> reject        :    125 ( 2.6%)

  Risk Assessment:
  - 2,104 new approvals -- evaluate default risk
  - 412 downgraded to conditional -- customer experience impact
  - 125 new rejections -- compliance review needed

syntax = "proto3";
package onto.decision.dryrun.v1;

service DryRunService {
    rpc CreateShadow(CreateShadowRequest) returns (CreateShadowResponse);
    rpc StopShadow(StopShadowRequest) returns (StopShadowResponse);
    rpc GetShadowReport(GetReportRequest) returns (ShadowReportResponse);
    rpc WhatIf(WhatIfRequest) returns (WhatIfResponse);
    rpc VariableSweep(SweepRequest) returns (SweepResponse);
    rpc StartReplay(ReplayRequest) returns (stream ReplayProgress);
}

message WhatIfRequest {
    string domain = 1;
    map<string, string> base_inputs = 2;
    repeated Modification modifications = 3;
}

message Modification {
    string variable = 1;
    string original_value = 2;
    string hypothetical_value = 3;
}

message WhatIfResponse {
    string original_decision = 1;
    double original_confidence = 2;
    string hypothetical_decision = 3;
    double hypothetical_confidence = 4;
    bool decision_changed = 5;
    map<string, double> sensitivity = 6;
}

class DryRunServiceImpl:
    """Dry-Run gRPC service"""

    def __init__(self, engine_registry, decision_store):
        self._registry = engine_registry
        self._store = decision_store
        self._active_shadows: dict[str, ShadowModeExecutor] = {}

    async def CreateShadow(self, request, context):
        primary = self._registry.get(request.primary_version)
        shadow = self._registry.get(request.shadow_version)

        config = ShadowConfig(
            shadow_id=request.shadow_id,
            description=request.description,
            primary_version=request.primary_version,
            shadow_version=request.shadow_version,
            sample_rate=request.sample_rate,
        )

        executor = ShadowModeExecutor(primary, shadow, config)
        self._active_shadows[config.shadow_id] = executor

        return {"shadow_id": config.shadow_id, "status": "active"}

    async def WhatIf(self, request, context):
        engine = self._registry.get_current()
        analyzer = WhatIfAnalyzer(engine)

        modifications = [
            WhatIfModification(
                variable=m.variable,
                original_value=m.original_value,
                hypothetical_value=self._parse(m.hypothetical_value),
            )
            for m in request.modifications
        ]

        scenario = WhatIfScenario(
            scenario_id=f"whatif-{id(request)}",
            name="ad-hoc",
            base_inputs={k: self._parse(v)
                         for k, v in request.base_inputs.items()},
            modifications=modifications,
        )

        result = analyzer.analyze(
            scenario,
            lambda inputs: DecisionContextBuilder(request.domain)
            .with_inputs(**inputs).build()
        )

        return {
            "original_decision": result.original_decision,
            "original_confidence": result.original_confidence,
            "hypothetical_decision": result.hypothetical_decision,
            "hypothetical_confidence": result.hypothetical_confidence,
            "decision_changed": result.decision_changed,
            "sensitivity": result.sensitivity,
        }

    def _parse(self, s: str):
        try:
            return int(s)
        except ValueError:
            pass
        try:
            return float(s)
        except ValueError:
            return s

class ShadowGuard:
    """Shadow mode safety guard"""

    def __init__(self, max_latency_ratio: float = 2.0,
                 max_error_rate: float = 0.05):
        self._max_latency_ratio = max_latency_ratio
        self._max_error_rate = max_error_rate
        self._error_count = 0
        self._total_count = 0

    def check(self, result: ShadowResult) -> bool:
        """Check if shadow execution is safe"""
        self._total_count += 1

        if result.primary_latency_ms > 0:
            ratio = result.shadow_latency_ms / result.primary_latency_ms
            if ratio > self._max_latency_ratio:
                return False

        if result.shadow_decision.startswith("ERROR"):
            self._error_count += 1

        error_rate = self._error_count / self._total_count
        if error_rate > self._max_error_rate:
            return False

        return True

    def should_disable(self) -> bool:
        """Whether shadow mode should be disabled"""
        if self._total_count < 100:
            return False
        error_rate = self._error_count / self._total_count
        return error_rate > self._max_error_rate * 2

# Decision validation stage in .gitlab-ci.yml
decision-dry-run:
  stage: validate
  script:
    - python -m onto.decision.dryrun replay
        --config replay-config.yaml
        --new-version $CI_COMMIT_SHA
        --time-range "7d"
        --max-records 50000
    - python -m onto.decision.dryrun check
        --max-divergence-rate 0.10
        --max-new-reject-increase 0.02
  artifacts:
    paths:
      - reports/dryrun-report.json
  rules:
    - if: $CI_MERGE_REQUEST_TARGET_BRANCH == "main"
      changes:
        - decision-trees/**/*
        - intelligence-Layer/reasoning/**/*

class DryRunQualityGate:
    """Dry-Run quality gate"""

    def __init__(self, max_divergence: float = 0.10,
                 max_reject_increase: float = 0.02,
                 max_latency_increase_pct: float = 20.0):
        self._max_div = max_divergence
        self._max_reject = max_reject_increase
        self._max_latency = max_latency_increase_pct

    def evaluate(self, report) -> dict:
        checks = []

        if isinstance(report, ShadowReport):
            checks.append({
                "name": "divergence_rate",
                "value": report.divergence_rate,
                "threshold": self._max_div,
                "passed": report.divergence_rate <= self._max_div,
            })

        if isinstance(report, ReplayStats):
            reject_increase = (
                (1 - report.new_approve_rate) -
                (1 - report.old_approve_rate)
            )
            checks.append({
                "name": "reject_increase",
                "value": reject_increase,
                "threshold": self._max_reject,
                "passed": reject_increase <= self._max_reject,
            })

        all_passed = all(c["passed"] for c in checks)
        return {
            "passed": all_passed,
            "checks": checks,
            "recommendation": "PROCEED" if all_passed else "BLOCK",
        }

class EngineRegistry:
    """Decision engine version registry"""

    def __init__(self):
        self._versions: dict[str, Any] = {}
        self._current: str = ""

    def register(self, version: str, engine) -> None:
        self._versions[version] = engine

    def set_current(self, version: str) -> None:
        if version not in self._versions:
            raise ValueError(f"Version {version} not registered")
        self._current = version

    def get(self, version: str):
        engine = self._versions.get(version)
        if engine is None:
            raise ValueError(f"Version {version} not found")
        return engine

    def get_current(self):
        return self.get(self._current)

    def list_versions(self) -> list[str]:
        return list(self._versions.keys())

Sample Rate Selection Guide:

  QPS Level        | Recommended Rate | Daily Samples
  -----------------|-----------------|---------------
  < 100 QPS        | 100%            | ~8.6M
  100-1000 QPS     | 10-50%          | ~4.3M-43M
  1000-10000 QPS   | 1-10%           | ~864K-8.6M
  > 10000 QPS      | 0.1-1%          | ~864K

# Scenario: Credit rule threshold adjustment validation

# 1. Create shadow mode
config = ShadowConfig(
    shadow_id="shadow-credit-threshold-adjust",
    description="Adjust credit_score threshold from 650 to 620",
    primary_version="credit-v2.1",
    shadow_version="credit-v2.2-candidate",
    sample_rate=0.5,
)

executor = ShadowModeExecutor(
    primary_engine=registry.get("credit-v2.1"),
    shadow_engine=registry.get("credit-v2.2-candidate"),
    config=config,
)

# 2. Review report after 7 days
report = executor.generate_report()
print(f"Divergence rate: {report.divergence_rate:.1%}")
print(f"Recommendation: {report.recommendation}")

# 3. What-If analysis for edge cases
analyzer = WhatIfAnalyzer(registry.get("credit-v2.2-candidate"))
sweep_results = analyzer.sweep(
    base_inputs={"credit_score": 580, "debt_ratio": 0.45,
                 "annual_income": 200000},
    variable="credit_score",
    values=list(range(550, 750, 10)),
    builder=lambda inputs: DecisionContextBuilder("credit")
    .with_inputs(**inputs).build()
)

# 4. Quality gate check
gate = DryRunQualityGate(max_divergence=0.10)
result = gate.evaluate(report)
# {"passed": True, "recommendation": "PROCEED"}