Shadow Mode Prerequisites¶

# Example: Trading Domain Proposal Schema
proposal_schema:
  domain: "trading"
  entity: "trade_order"

  mapping:
    proposal_id: "order_id"
    timestamp: "order_timestamp"

    risk_inputs:
      R_base: "historical_var_95"        # Value-at-Risk from past 30 days
      R_prop: "projected_var_95"         # VaR if this trade executes

    profit_inputs:
      P_base: "baseline_expected_return"
      P_prop: "projected_expected_return"

    complexity_inputs:
      C_S: "instrument_count"            # Number of instruments
      C_D: "counterparty_count"          # New counterparties

    novelty_inputs:
      features: ["asset_class", "notional", "tenor", "strategy_type"]
      similarity_method: "cosine_distance_to_centroid"

    quality_inputs:
      validity: "model_validation_score"
      consistency: "internal_limit_check"
      support: "backtesting_score"

┌─────────────────────────────────────────────────────────────────────┐
│                        PROPOSAL INTAKE                               │
│                              │                                       │
│                              ▼                                       │
│  ┌───────────────────────────────────────────────────────────────┐  │
│  │                    EXISTING SYSTEM                             │  │
│  │  ┌─────────────┐    ┌─────────────┐    ┌─────────────┐        │  │
│  │  │   Rules     │───►│   Human     │───►│  Decision   │        │  │
│  │  │   Engine    │    │   Review    │    │  (ACTUAL)   │────────┼──┼──► Production
│  │  └─────────────┘    └─────────────┘    └──────┬──────┘        │  │
│  └───────────────────────────────────────────────┼───────────────┘  │
│                                                  │                   │
│                                                  │ Log for           │
│                                                  │ comparison        │
│                                                  ▼                   │
│  ┌───────────────────────────────────────────────────────────────┐  │
│  │                    SHADOW GUARDRAIL                            │  │
│  │  ┌─────────────┐    ┌─────────────┐    ┌─────────────┐        │  │
│  │  │  Compute    │───►│   Apply     │───►│  Decision   │        │  │
│  │  │  Metrics    │    │ Thresholds  │    │  (SHADOW)   │────────┼──┼──► Telemetry Only
│  │  └─────────────┘    └─────────────┘    └─────────────┘        │  │
│  └───────────────────────────────────────────────────────────────┘  │
│                                                                      │
│                              │                                       │
│                              ▼                                       │
│                    ┌─────────────────┐                              │
│                    │   COMPARISON    │                              │
│                    │   & TELEMETRY   │                              │
│                    └─────────────────┘                              │
└─────────────────────────────────────────────────────────────────────┘

# Baseline snapshot requirements
baseline_snapshot:
  version: "1.0"
  created_at: "2025-12-26T00:00:00Z"
  sha256_hash: "<computed at creation>"

  period:
    start: "2025-09-26"
    end: "2025-12-25"
    days: 91

  statistics:
    total_proposals: 2847
    approved: 2102
    rejected: 745

  distributions:
    risk_scores:
      mean: 0.23
      std: 0.15
      p50: 0.19
      p90: 0.42
      p99: 0.71
      histogram_bins: [0.0, 0.1, 0.2, ..., 1.0]
      histogram_counts: [142, 387, 521, ...]

    profit_scores:
      mean: 0.45
      std: 0.22
      # ... similar structure

    novelty_scores:
      # ...

    complexity_scores:
      # ...

-- Example: Export proposals from last 90 days
SELECT
    proposal_id,
    created_at,
    risk_score,
    profit_score,
    novelty_score,
    complexity_score,
    quality_score,
    decision,
    decision_maker
FROM proposals
WHERE created_at >= CURRENT_DATE - INTERVAL '90 days'
  AND status = 'completed';

import numpy as np
import hashlib
import json

def create_baseline(proposals_df):
    baseline = {
        "version": "1.0",
        "created_at": datetime.now().isoformat(),
        "period": {
            "start": proposals_df['created_at'].min().isoformat(),
            "end": proposals_df['created_at'].max().isoformat(),
            "days": (proposals_df['created_at'].max() -
                    proposals_df['created_at'].min()).days
        },
        "statistics": {
            "total_proposals": len(proposals_df),
            "approved": len(proposals_df[proposals_df['decision'] == 'approved']),
            "rejected": len(proposals_df[proposals_df['decision'] == 'rejected'])
        },
        "distributions": {}
    }

    for metric in ['risk_score', 'profit_score', 'novelty_score',
                   'complexity_score', 'quality_score']:
        values = proposals_df[metric].dropna()
        hist, bins = np.histogram(values, bins=10, range=(0, 1))

        baseline["distributions"][metric] = {
            "mean": float(np.mean(values)),
            "std": float(np.std(values)),
            "p50": float(np.percentile(values, 50)),
            "p90": float(np.percentile(values, 90)),
            "p99": float(np.percentile(values, 99)),
            "histogram_bins": bins.tolist(),
            "histogram_counts": hist.tolist()
        }

    # Compute hash for integrity verification
    baseline_json = json.dumps(baseline, sort_keys=True)
    baseline["sha256_hash"] = hashlib.sha256(baseline_json.encode()).hexdigest()

    return baseline

# Save baseline snapshot
python create_baseline.py --output baseline-v1.0.json

# Create immutable reference
sha256sum baseline-v1.0.json > baseline-v1.0.sha256

# Store in version control or immutable storage
aws s3 cp baseline-v1.0.json s3://guardrail-baselines/ --metadata sha256=$(cat baseline-v1.0.sha256)

proposal_schema:
  domain: "customer_service_agent"
  entity: "agent_action"

  mapping:
    proposal_id: "action_id"           # UUID for each agent action
    timestamp: "action_timestamp"

    risk_inputs:
      R_base: 0.15                      # Historical average risk for this action type
      R_prop:                           # Computed from:
        - data_sensitivity_score        # 0-1: How sensitive is the data accessed?
        - financial_impact_score        # 0-1: Could this cost money?
        - compliance_risk_score         # 0-1: Regulatory implications?
        # R_prop = weighted average of above

    profit_inputs:
      P_base: 0.60                      # Historical customer satisfaction baseline
      P_prop:
        - resolution_probability        # 0-1: Will this resolve the issue?
        - customer_sentiment_score      # 0-1: Predicted customer satisfaction
        - efficiency_score              # 0-1: Time/cost savings

    complexity_inputs:
      C_S: "systems_accessed_count"     # Number of backend systems touched
      C_D: "external_api_calls"         # External services called

    novelty_inputs:
      features:
        - action_type                   # "respond", "lookup", "transact", "escalate"
        - query_category                # "balance", "payment", "dispute", etc.
        - customer_tier                 # "standard", "premium", "vip"
        - account_type                  # "checking", "savings", "investment"
      similarity_method: "embedding_cosine_distance"
      embedding_model: "text-embedding-3-small"

    quality_inputs:
      validity: "response_format_valid"           # Does response follow required format?
      consistency: "no_contradictions_detected"   # Consistent with previous statements?
      support: "citations_provided"               # Are claims backed by data?

Customer ──► AI Agent ──► [Proposed Response]
                                │
                ┌───────────────┴───────────────┐
                │                               │
                ▼                               ▼
    ┌───────────────────┐           ┌───────────────────┐
    │  EXISTING SYSTEM  │           │  SHADOW GUARDRAIL │
    │                   │           │                   │
    │  Risk Score Check │           │  Compute Metrics  │
    │  (simple rules)   │           │  (full framework) │
    │        │          │           │        │          │
    │        ▼          │           │        ▼          │
    │  High Risk?       │           │  Log Decision     │
    │  ├─ Yes: Human    │           │  + All Metrics    │
    │  └─ No: Auto-send │           │                   │
    └─────────┬─────────┘           └─────────┬─────────┘
              │                               │
              ▼                               ▼
    [Actual Decision]              [Telemetry Database]
              │
              ▼
         Customer

# Current simple rule-based system
def existing_risk_check(response):
    risk_score = 0

    # Rule 1: Financial amounts trigger review
    if contains_dollar_amount(response) and amount > 1000:
        risk_score += 0.4

    # Rule 2: Account changes trigger review
    if action_type in ["transact", "modify_account"]:
        risk_score += 0.3

    # Rule 3: Sensitive topics
    if topic in ["fraud", "complaint", "legal"]:
        risk_score += 0.3

    if risk_score >= 0.5:
        return "human_review"
    else:
        return "auto_approve"

-- Export 90 days of agent actions for baseline
SELECT
    action_id,
    action_timestamp,
    action_type,
    query_category,
    customer_tier,

    -- Risk components
    data_sensitivity_score,
    financial_impact_score,
    compliance_risk_score,

    -- Profit components
    resolution_probability,
    customer_sentiment_score,
    efficiency_score,

    -- Complexity
    systems_accessed_count,
    external_api_calls,

    -- Quality
    response_format_valid,
    no_contradictions_detected,
    citations_provided,

    -- Outcomes
    existing_system_decision,    -- 'auto_approve' or 'human_review'
    human_decision,              -- 'approve', 'reject', 'modify', NULL if auto
    customer_feedback_score,     -- 1-5 stars, NULL if not provided
    escalation_occurred          -- Did customer ask for human after?

FROM agent_actions
WHERE action_timestamp >= CURRENT_DATE - INTERVAL '90 days'
  AND action_status = 'completed';

baseline_snapshot:
  version: "1.0"
  created_at: "2025-12-26T00:00:00Z"
  sha256_hash: "a1b2c3d4e5f6..."

  period:
    start: "2025-09-27"
    end: "2025-12-25"
    days: 90

  statistics:
    total_actions: 127,543
    auto_approved: 98,234 (77%)
    human_reviewed: 29,309 (23%)
    human_approved: 27,891 (95% of reviewed)
    human_rejected: 1,418 (5% of reviewed)

  distributions:
    risk_scores:
      mean: 0.23
      std: 0.18
      p50: 0.18
      p90: 0.47
      p99: 0.72
      # Note: Existing system uses 0.5 threshold

    novelty_scores:
      mean: 0.31
      std: 0.22
      p50: 0.26
      p90: 0.61
      p99: 0.89
      # High novelty often correlates with escalations

    # Key insight from baseline:
    insights:
      - "23% of actions go to human review"
      - "Only 5% of reviewed actions are rejected"
      - "High novelty (>0.6) correlates with 3x escalation rate"
      - "Actions accessing 3+ systems have 2x rejection rate"

CREATE TABLE guardrail_telemetry (
    -- Identifiers
    telemetry_id        UUID PRIMARY KEY,
    action_id           UUID NOT NULL,
    timestamp           TIMESTAMPTZ NOT NULL,

    -- Proposal Context
    action_type         VARCHAR(50),
    query_category      VARCHAR(100),
    customer_tier       VARCHAR(20),

    -- Guardrail Metrics
    risk_score          FLOAT,
    profit_score        FLOAT,
    novelty_score       FLOAT,
    complexity_score    FLOAT,
    quality_score       FLOAT,

    -- Sub-metrics (for debugging)
    data_sensitivity    FLOAT,
    financial_impact    FLOAT,
    compliance_risk     FLOAT,

    -- Decisions
    guardrail_decision  VARCHAR(20),  -- 'pass' or 'fail'
    existing_decision   VARCHAR(20),  -- 'auto_approve' or 'human_review'
    human_decision      VARCHAR(20),  -- 'approve', 'reject', 'modify', NULL

    -- KL Drift Tracking
    kl_divergence       FLOAT,
    drift_status        VARCHAR(20),  -- 'normal', 'warning', 'critical'

    -- Traceability
    param_snapshot_id   VARCHAR(50),  -- 'guardrail-v1.1-freeze'
    baseline_hash       VARCHAR(64),
    embedding_hash      VARCHAR(64),

    -- Indexes
    INDEX idx_timestamp (timestamp),
    INDEX idx_action_id (action_id),
    INDEX idx_drift_status (drift_status)
);

-- Agreement Rate: How often does guardrail match existing system?
SELECT
    DATE(timestamp) as date,
    COUNT(*) as total,
    SUM(CASE WHEN guardrail_decision = 'fail'
             AND existing_decision = 'human_review' THEN 1 ELSE 0 END) as both_flagged,
    SUM(CASE WHEN guardrail_decision = 'pass'
             AND existing_decision = 'auto_approve' THEN 1 ELSE 0 END) as both_passed,
    SUM(CASE WHEN guardrail_decision = 'fail'
             AND existing_decision = 'auto_approve' THEN 1 ELSE 0 END) as guardrail_stricter,
    SUM(CASE WHEN guardrail_decision = 'pass'
             AND existing_decision = 'human_review' THEN 1 ELSE 0 END) as guardrail_lenient
FROM guardrail_telemetry
WHERE timestamp >= CURRENT_DATE - INTERVAL '7 days'
GROUP BY DATE(timestamp)
ORDER BY date;

-- False Negative Detection: Actions guardrail passed but humans rejected
SELECT *
FROM guardrail_telemetry
WHERE guardrail_decision = 'pass'
  AND human_decision = 'reject'
  AND timestamp >= CURRENT_DATE - INTERVAL '30 days';

from dataclasses import dataclass
from datetime import datetime
import hashlib
import json

@dataclass
class AgentAction:
    action_id: str
    action_type: str
    query_category: str
    customer_tier: str
    proposed_response: str
    context: dict

class ShadowGuardrail:
    def __init__(self, config):
        self.param_snapshot_id = config['param_snapshot_id']
        self.baseline = self.load_baseline(config['baseline_path'])
        self.telemetry_client = TelemetryClient(config['telemetry_endpoint'])

    def evaluate(self, action: AgentAction) -> dict:
        """
        Evaluate action through guardrail framework.
        Returns decision and metrics (does NOT block action).
        """
        start_time = datetime.now()

        # Compute all metrics
        risk_score = self.compute_risk_score(action)
        profit_score = self.compute_profit_score(action)
        novelty_score = self.compute_novelty_score(action)
        complexity_score = self.compute_complexity_score(action)
        quality_score = self.compute_quality_score(action)

        # Apply guardrail logic
        guardrail_decision = self.apply_thresholds(
            risk_score, profit_score, novelty_score,
            complexity_score, quality_score
        )

        # Compute KL divergence for drift detection
        kl_divergence = self.compute_kl_divergence(risk_score)
        drift_status = self.get_drift_status(kl_divergence)

        # Build telemetry record
        telemetry = {
            "action_id": action.action_id,
            "timestamp": datetime.now().isoformat(),
            "action_type": action.action_type,
            "query_category": action.query_category,
            "customer_tier": action.customer_tier,
            "risk_score": risk_score,
            "profit_score": profit_score,
            "novelty_score": novelty_score,
            "complexity_score": complexity_score,
            "quality_score": quality_score,
            "guardrail_decision": guardrail_decision,
            "kl_divergence": kl_divergence,
            "drift_status": drift_status,
            "param_snapshot_id": self.param_snapshot_id,
            "baseline_hash": self.baseline['sha256_hash'],
            "latency_ms": (datetime.now() - start_time).total_seconds() * 1000
        }

        # Log telemetry (async, non-blocking)
        self.telemetry_client.log_async(telemetry)

        return {
            "decision": guardrail_decision,
            "metrics": telemetry
        }

    def compute_risk_score(self, action: AgentAction) -> float:
        # Implementation based on proposal schema
        data_sensitivity = self.score_data_sensitivity(action)
        financial_impact = self.score_financial_impact(action)
        compliance_risk = self.score_compliance_risk(action)

        # Weighted combination
        weights = [0.4, 0.35, 0.25]
        raw_score = (
            weights[0] * data_sensitivity +
            weights[1] * financial_impact +
            weights[2] * compliance_risk
        )

        # Normalize relative to baseline
        baseline_mean = self.baseline['distributions']['risk_scores']['mean']
        epsilon = 0.01
        normalized = (raw_score - baseline_mean) / max(baseline_mean, epsilon)

        return max(0, min(1, (normalized + 1) / 2))  # Scale to 0-1

    # ... other scoring methods ...


# Integration point: wrap existing decision function
def process_agent_action(action: AgentAction):
    """
    Main processing function - integrates shadow guardrail.
    """
    # 1. Run shadow guardrail (non-blocking)
    shadow_result = shadow_guardrail.evaluate(action)

    # 2. Run existing system (this is what actually decides)
    existing_decision = existing_risk_check(action)

    # 3. Log comparison (for calibration analysis)
    shadow_guardrail.telemetry_client.log_comparison(
        action_id=action.action_id,
        guardrail_decision=shadow_result['decision'],
        existing_decision=existing_decision
    )

    # 4. Return existing system's decision (shadow doesn't affect outcome)
    return existing_decision