#!/usr/bin/env python3
"""Suite de regresión y hardening UCEPSA v0.3.13."""

from __future__ import annotations

import argparse
import hashlib
import json
import os
import sys
import time
from dataclasses import dataclass
from datetime import date, datetime
from decimal import Decimal
from typing import Any, Callable

import psycopg2
from psycopg2.extras import Json, RealDictCursor


VERSION = "0.3.13"
POLICY_CODE = "CONTEXT_REGRESSION_V1"


def env_first(*names: str, default: str | None = None) -> str | None:
    for name in names:
        value = os.getenv(name)
        if value not in (None, ""):
            return value
    return default


def connect():
    database = env_first("PGDATABASE", "POSTGRES_DB")
    user = env_first("PGUSER", "POSTGRES_USER")
    password = env_first("PGPASSWORD", "POSTGRES_PASSWORD")

    missing = [
        name
        for name, value in (
            ("PGDATABASE", database),
            ("PGUSER", user),
            ("PGPASSWORD", password),
        )
        if not value
    ]
    if missing:
        raise RuntimeError(
            "Faltan variables: " + ", ".join(missing)
        )

    return psycopg2.connect(
        host=env_first(
            "PGHOST",
            default="mv_ucepsa_postgres_hot",
        ),
        port=int(
            env_first("PGPORT", default="5432")
            or "5432"
        ),
        dbname=database,
        user=user,
        password=password,
        connect_timeout=10,
        application_name="context_regression_v0313",
    )


def json_safe(value: Any) -> Any:
    if isinstance(value, dict):
        return {
            str(key): json_safe(item)
            for key, item in value.items()
        }
    if isinstance(value, (list, tuple, set)):
        return [
            json_safe(item)
            for item in value
        ]
    if isinstance(value, (datetime, date)):
        return value.isoformat()
    if isinstance(value, Decimal):
        return str(value)
    return value


def print_json(value: Any) -> None:
    print(
        json.dumps(
            json_safe(value),
            ensure_ascii=False,
            indent=2,
        )
    )


@dataclass(frozen=True)
class TestCase:
    code: str
    group: str
    title: str
    severity: str
    failure_status: str
    sql: str
    expected: dict[str, Any]
    detail_pass: str
    detail_fail: str
    performance_only: bool = False


def contract_tests() -> list[TestCase]:
    return [
        TestCase(
            code="SCHEMA-001",
            group="SCHEMA",
            title="Objetos canónicos y de gobernanza disponibles",
            severity="CRITICAL",
            failure_status="FAIL",
            expected={"violations": 0},
            detail_pass="Todos los objetos requeridos existen.",
            detail_fail="Faltan tablas o vistas requeridas.",
            sql=r"""
            SELECT COUNT(*) AS violations
            FROM (
                VALUES
                    (to_regclass('mv_loss_intelligence.production_context_sessions')),
                    (to_regclass('mv_loss_intelligence.odoo_shopfloor_sessions')),
                    (to_regclass('mv_reports_ucepsa_prod.li_canonical_context_segments_v1')),
                    (to_regclass('mv_reports_ucepsa_prod.li_stop_canonical_context_shadow_v1')),
                    (to_regclass('mv_reports_ucepsa_prod.li_context_arbitration_conflicts_v1')),
                    (to_regclass('mv_reports_ucepsa_prod.li_context_review_case_history_v1')),
                    (to_regclass('mv_reports_ucepsa_prod.li_context_decision_group_history_v1')),
                    (to_regclass('mv_reports_ucepsa_prod.li_context_governance_daily_agenda_v1')),
                    (to_regclass('mv_reports_ucepsa_prod.li_context_governance_daily_summary_v1')),
                    (to_regclass('mv_loss_intelligence.context_governance_agenda_snapshots')),
                    (to_regclass('mv_loss_intelligence.context_regression_runs')),
                    (to_regclass('mv_loss_intelligence.context_regression_results'))
            ) AS required(obj)
            WHERE obj IS NULL
            """,
        ),
        TestCase(
            code="CANON-001",
            group="CANONICAL",
            title="Intervalos canónicos temporalmente válidos",
            severity="CRITICAL",
            failure_status="FAIL",
            expected={"violations": 0},
            detail_pass="No hay intervalos invertidos ni duraciones negativas.",
            detail_fail="Existen segmentos canónicos temporalmente inválidos.",
            sql=r"""
            SELECT
                COUNT(*) FILTER (
                    WHERE segment_to <= segment_from
                )
                + COUNT(*) FILTER (
                    WHERE segment_duration_s < 0
                ) AS violations,
                COUNT(*) AS segment_count
            FROM
                mv_reports_ucepsa_prod
                .li_canonical_context_segments_v1
            """,
        ),
        TestCase(
            code="CANON-002",
            group="CANONICAL",
            title="Arbitraje sin prioridad silenciosa",
            severity="CRITICAL",
            failure_status="FAIL",
            expected={"violations": 0},
            detail_pass="Una fuente produce contexto; cero o varias bloquean la atribución.",
            detail_fail="La semántica canónica permite claves ambiguas o fuentes supporting promovidas.",
            sql=r"""
            SELECT
                COUNT(*) FILTER (
                    WHERE authoritative_candidate_count = 1
                      AND canonical_context_key IS NULL
                )
                + COUNT(*) FILTER (
                    WHERE authoritative_candidate_count <> 1
                      AND canonical_context_key IS NOT NULL
                )
                + COUNT(*) FILTER (
                    WHERE canonical_context_source IN (
                        'ODOO_ORDER_CONTEXT',
                        'INFERRED_SHADOW'
                    )
                )
                + COUNT(*) FILTER (
                    WHERE canonical_context_status =
                          'CONTEXT_CONFLICT'
                      AND NOT review_required
                ) AS violations
            FROM
                mv_reports_ucepsa_prod
                .li_canonical_context_segments_v1
            """,
        ),
        TestCase(
            code="CANON-003",
            group="CANONICAL",
            title="Paros abiertos, ambiguos o no productivos bloqueados",
            severity="CRITICAL",
            failure_status="FAIL",
            expected={"violations": 0},
            detail_pass="Ningún paro no elegible aparece como técnicamente elegible.",
            detail_fail="Hay paros abiertos, ambiguos o no productivos promovidos.",
            sql=r"""
            SELECT
                COUNT(*) FILTER (
                    WHERE source_status = 'OPEN'
                      AND technically_eligible
                )
                + COUNT(*) FILTER (
                    WHERE NOT canonical_productive_context
                      AND technically_eligible
                )
                + COUNT(*) FILTER (
                    WHERE authoritative_candidate_count <> 1
                      AND technically_eligible
                ) AS violations
            FROM
                mv_reports_ucepsa_prod
                .li_stop_canonical_context_shadow_v1
            """,
        ),
        TestCase(
            code="CANON-004",
            group="CANONICAL",
            title="Slices de paro no superan el paro raw",
            severity="CRITICAL",
            failure_status="FAIL",
            expected={"violations": 0},
            detail_pass="La segmentación conserva la duración del paro.",
            detail_fail="La suma de slices supera la duración raw.",
            sql=r"""
            WITH totals AS (
                SELECT
                    source_stop_id,
                    MAX(raw_duration_s)
                        AS raw_duration_s,
                    SUM(segment_duration_s)
                        AS sliced_duration_s
                FROM
                    mv_reports_ucepsa_prod
                    .li_stop_canonical_context_slices_v1
                GROUP BY source_stop_id
            )
            SELECT COUNT(*) AS violations
            FROM totals
            WHERE sliced_duration_s >
                  raw_duration_s + 1::numeric
            """,
        ),
        TestCase(
            code="LIVE-001",
            group="LIVE",
            title="Una fila viva por cada cortadora piloto",
            severity="CRITICAL",
            failure_status="FAIL",
            expected={
                "row_count": 3,
                "distinct_machine_count": 3,
                "official_rows": 0,
            },
            detail_pass="CORT-00, CORT-01 y CORT-02 tienen una única fila viva.",
            detail_fail="La cardinalidad viva no coincide con las tres cortadoras piloto.",
            sql=r"""
            SELECT
                COUNT(*) AS row_count,
                COUNT(
                    DISTINCT machine_id
                ) AS distinct_machine_count,
                COUNT(*) FILTER (
                    WHERE official_ledger_eligible
                ) AS official_rows,
                ARRAY_AGG(
                    machine_id
                    ORDER BY machine_id
                ) AS machines
            FROM
                mv_reports_ucepsa_prod
                .li_machine_canonical_context_live_v1
            """,
        ),
        TestCase(
            code="WO340-001",
            group="HISTORICAL_CONTRACT",
            title="Workorder 340 cerrado con cantidad final confirmada",
            severity="CRITICAL",
            failure_status="FAIL",
            expected={"violations": 0, "row_count": 1},
            detail_pass="WO 340 conserva cierre, operario y 100 kg finales.",
            detail_fail="El contrato histórico del workorder 340 ha cambiado.",
            sql=r"""
            SELECT
                COUNT(*) AS row_count,
                COUNT(*) FILTER (
                    WHERE NOT (
                        workorder_state = 'done'
                        AND production_state = 'done'
                        AND session_status = 'CLOSED'
                        AND started_at =
                            '2026-07-20 07:36:19+02'::timestamptz
                        AND ended_at =
                            '2026-07-20 14:08:41+02'::timestamptz
                        AND operator_employee_id = 24
                        AND operator_name =
                            'ESTEBAN VAQUERIZO GARCIA'
                        AND planned_quantity = 100
                        AND declared_quantity = 100
                        AND posted_quantity = 100
                        AND remaining_planned_quantity = 0
                        AND quantity_status =
                            'FINAL_POSTED'
                        AND quantity_confirmation_status =
                            'EVIDENCE_CONFIRMED'
                        AND confidence = 'HIGH'
                        AND NOT official_eligible
                    )
                ) AS violations
            FROM
                mv_loss_intelligence.odoo_shopfloor_sessions
            WHERE odoo_workorder_id = 340
            """,
        ),
        TestCase(
            code="WO340-002",
            group="HISTORICAL_CONTRACT",
            title="WO 340 no se extiende después de su cierre",
            severity="CRITICAL",
            failure_status="FAIL",
            expected={"violations": 0},
            detail_pass="La actividad posterior al cierre no se atribuye al WO 340.",
            detail_fail="Existen segmentos canónicos del WO 340 posteriores al cierre.",
            sql=r"""
            WITH session_340 AS (
                SELECT ended_at
                FROM
                    mv_loss_intelligence.odoo_shopfloor_sessions
                WHERE odoo_workorder_id = 340
                LIMIT 1
            )
            SELECT COUNT(*) AS violations
            FROM
                mv_reports_ucepsa_prod
                .li_canonical_context_segments_v1 s
            CROSS JOIN session_340 w
            WHERE s.machine_id = 'CORT-02'
              AND s.odoo_workorder_id = 340
              AND s.segment_from >=
                  w.ended_at
            """,
        ),
        TestCase(
            code="LEGACY-001",
            group="HISTORICAL_CONTRACT",
            title="Contexto legacy del episodio 1 autorizado",
            severity="CRITICAL",
            failure_status="FAIL",
            expected={"violations": 0, "row_count": 1},
            detail_pass="El episodio 1 conserva su contexto legacy autorizado.",
            detail_fail="El contexto retrospectivo del episodio 1 ha cambiado.",
            sql=r"""
            SELECT
                COUNT(*) AS row_count,
                COUNT(*) FILTER (
                    WHERE NOT (
                        machine_id = 'CORT-00'
                        AND source_type = 'LEGACY_ERP'
                        AND started_at =
                            '2026-07-20 12:58:45.457786+02'::timestamptz
                        AND ended_at =
                            '2026-07-20 13:23:56.594486+02'::timestamptz
                        AND status = 'CLOSED'
                        AND authorization_status =
                            'AUTHORIZED'
                        AND confidence = 'MEDIUM'
                        AND authorized_by = 'Juan Pablo'
                        AND NOT official_eligible
                    )
                ) AS violations
            FROM
                mv_loss_intelligence.production_context_sessions
            WHERE source_ref =
                  'SHOPFLOOR_INCIDENT_EPISODE:1'
            """,
        ),
        TestCase(
            code="LEGACY-002",
            group="HISTORICAL_CONTRACT",
            title="Cobertura canónica exacta del contexto legacy",
            severity="CRITICAL",
            failure_status="FAIL",
            expected={"violations": 0},
            detail_pass="El intervalo autorizado queda cubierto por LEGACY_ERP sin huecos ni conflicto.",
            detail_fail="La cobertura canónica del intervalo legacy no coincide con el intervalo autorizado.",
            sql=r"""
            WITH target AS (
                SELECT
                    started_at,
                    ended_at,
                    EXTRACT(
                        EPOCH FROM (
                            ended_at
                            - started_at
                        )
                    )::numeric AS target_s
                FROM
                    mv_loss_intelligence
                    .production_context_sessions
                WHERE source_ref =
                      'SHOPFLOOR_INCIDENT_EPISODE:1'
                LIMIT 1
            ),
            coverage AS (
                SELECT
                    COALESCE(
                        SUM(
                            EXTRACT(
                                EPOCH FROM (
                                    LEAST(
                                        s.segment_to,
                                        t.ended_at
                                    )
                                    - GREATEST(
                                        s.segment_from,
                                        t.started_at
                                    )
                                )
                            )
                        ) FILTER (
                            WHERE s.canonical_context_status =
                                  'LEGACY_ERP'
                              AND s.canonical_context_source =
                                  'LEGACY_ERP'
                        ),
                        0
                    )::numeric AS covered_s,
                    COUNT(*) FILTER (
                        WHERE s.canonical_context_status <>
                              'LEGACY_ERP'
                    ) AS wrong_status_segments,
                    MAX(t.target_s) AS target_s
                FROM
                    mv_reports_ucepsa_prod
                    .li_canonical_context_segments_v1 s
                CROSS JOIN target t
                WHERE s.machine_id = 'CORT-00'
                  AND s.segment_to >
                      t.started_at
                  AND s.segment_from <
                      t.ended_at
            )
            SELECT
                CASE
                    WHEN ABS(
                        covered_s
                        - target_s
                    ) <= 1
                     AND wrong_status_segments = 0
                        THEN 0
                    ELSE 1
                END AS violations,
                covered_s,
                target_s,
                wrong_status_segments
            FROM coverage
            """,
        ),
        TestCase(
            code="CONFLICT-001",
            group="HISTORICAL_CONTRACT",
            title="Doble inicio 649/698 conservado como conflicto",
            severity="CRITICAL",
            failure_status="FAIL",
            expected={"violations": 0},
            detail_pass="El solape histórico de WO 649 y 698 sigue documentado.",
            detail_fail="No existe el conflicto histórico esperado entre WO 649 y 698.",
            sql=r"""
            SELECT
                CASE
                    WHEN COUNT(*) >= 1
                        THEN 0
                    ELSE 1
                END AS violations,
                COUNT(*) AS conflict_rows,
                COALESCE(
                    SUM(segment_duration_s),
                    0
                ) AS conflict_seconds
            FROM
                mv_reports_ucepsa_prod
                .li_context_arbitration_conflicts_v1
            WHERE machine_id = 'CORT-00'
              AND conflict_type =
                  'MULTIPLE_ODOO_SHOPFLOOR_SESSIONS'
              AND authoritative_context_keys @>
                  ARRAY[
                    'ODOO_SHOPFLOOR:WORKORDER:649',
                    'ODOO_SHOPFLOOR:WORKORDER:698'
                  ]::text[]
            """,
        ),
        TestCase(
            code="CONFLICT-002",
            group="HISTORICAL_CONTRACT",
            title="El conflicto 649/698 nunca recibe prioridad canónica",
            severity="CRITICAL",
            failure_status="FAIL",
            expected={"violations": 0},
            detail_pass="Todos los tramos del doble inicio permanecen bloqueados.",
            detail_fail="Algún tramo del conflicto recibió una clave canónica.",
            sql=r"""
            WITH conflict_window AS (
                SELECT
                    MIN(segment_from) AS conflict_from,
                    MAX(segment_to) AS conflict_to
                FROM
                    mv_reports_ucepsa_prod
                    .li_context_arbitration_conflicts_v1
                WHERE machine_id = 'CORT-00'
                  AND authoritative_context_keys @>
                      ARRAY[
                        'ODOO_SHOPFLOOR:WORKORDER:649',
                        'ODOO_SHOPFLOOR:WORKORDER:698'
                      ]::text[]
            )
            SELECT COUNT(*) AS violations
            FROM
                mv_reports_ucepsa_prod
                .li_canonical_context_segments_v1 s
            CROSS JOIN conflict_window c
            WHERE c.conflict_from IS NOT NULL
              AND s.machine_id = 'CORT-00'
              AND s.segment_to >
                  c.conflict_from
              AND s.segment_from <
                  c.conflict_to
              AND (
                    s.canonical_context_status <>
                        'CONTEXT_CONFLICT'
                    OR s.canonical_context_key
                       IS NOT NULL
                    OR s.authoritative_candidate_count <= 1
                  )
            """,
        ),
        TestCase(
            code="CASE-001",
            group="GOVERNANCE",
            title="Episodios 7/8/9 agrupados en un caso raíz",
            severity="CRITICAL",
            failure_status="FAIL",
            expected={
                "distinct_case_count": 1,
                "evidence_count": 3,
            },
            detail_pass="Las tres evidencias del doble inicio producen una sola tarea raíz.",
            detail_fail="Los episodios 7/8/9 no están agrupados correctamente.",
            sql=r"""
            SELECT
                COUNT(
                    DISTINCT case_id
                ) AS distinct_case_count,
                COUNT(*) AS evidence_count
            FROM
                mv_loss_intelligence
                .context_review_case_evidence
            WHERE episode_id IN (
                7,
                8,
                9
            )
            """,
        ),
        TestCase(
            code="DECISION-001",
            group="GOVERNANCE",
            title="Casos próximos consolidados en tres decisiones",
            severity="CRITICAL",
            failure_status="FAIL",
            expected={"violations": 0},
            detail_pass="Los grupos 1-4, 5-6 y 7-8 generan una decisión cada uno.",
            detail_fail="La consolidación temporal de decisiones ha cambiado.",
            sql=r"""
            WITH checks AS (
                SELECT
                    (
                        SELECT COUNT(
                            DISTINCT decision_group_key
                        )
                        FROM
                            mv_reports_ucepsa_prod
                            .li_context_decision_group_case_assignments_v1
                        WHERE case_id IN (
                            1,
                            2,
                            3,
                            4
                        )
                    ) AS group_1_4,
                    (
                        SELECT COUNT(*)
                        FROM
                            mv_reports_ucepsa_prod
                            .li_context_decision_group_case_assignments_v1
                        WHERE case_id IN (
                            1,
                            2,
                            3,
                            4
                        )
                    ) AS linked_1_4,
                    (
                        SELECT COUNT(
                            DISTINCT decision_group_key
                        )
                        FROM
                            mv_reports_ucepsa_prod
                            .li_context_decision_group_case_assignments_v1
                        WHERE case_id IN (
                            5,
                            6
                        )
                    ) AS group_5_6,
                    (
                        SELECT COUNT(*)
                        FROM
                            mv_reports_ucepsa_prod
                            .li_context_decision_group_case_assignments_v1
                        WHERE case_id IN (
                            5,
                            6
                        )
                    ) AS linked_5_6,
                    (
                        SELECT COUNT(
                            DISTINCT decision_group_key
                        )
                        FROM
                            mv_reports_ucepsa_prod
                            .li_context_decision_group_case_assignments_v1
                        WHERE case_id IN (
                            7,
                            8
                        )
                    ) AS group_7_8,
                    (
                        SELECT COUNT(*)
                        FROM
                            mv_reports_ucepsa_prod
                            .li_context_decision_group_case_assignments_v1
                        WHERE case_id IN (
                            7,
                            8
                        )
                    ) AS linked_7_8
            )
            SELECT
                CASE
                    WHEN group_1_4 = 1
                     AND linked_1_4 = 4
                     AND group_5_6 = 1
                     AND linked_5_6 = 2
                     AND group_7_8 = 1
                     AND linked_7_8 = 2
                        THEN 0
                    ELSE 1
                END AS violations,
                *
            FROM checks
            """,
        ),
        TestCase(
            code="DECISION-002",
            group="GOVERNANCE",
            title="Cada caso raíz pertenece a una única decisión",
            severity="CRITICAL",
            failure_status="FAIL",
            expected={"violations": 0},
            detail_pass="No hay casos raíz duplicados entre decisiones.",
            detail_fail="Un caso raíz pertenece a varias decisiones.",
            sql=r"""
            SELECT COUNT(*) AS violations
            FROM (
                SELECT
                    case_id,
                    COUNT(*) AS n
                FROM
                    mv_reports_ucepsa_prod
                    .li_context_decision_group_case_assignments_v1
                GROUP BY case_id
                HAVING COUNT(*) > 1
            ) duplicated
            """,
        ),
        TestCase(
            code="AGENDA-001",
            group="GOVERNANCE",
            title="Una fila de agenda por decisión pendiente",
            severity="CRITICAL",
            failure_status="FAIL",
            expected={"violations": 0},
            detail_pass="La agenda no duplica ni omite decisiones pendientes.",
            detail_fail="La agenda y el backlog de decisiones no coinciden.",
            sql=r"""
            WITH counts AS (
                SELECT
                    (
                        SELECT COUNT(*)
                        FROM
                            mv_reports_ucepsa_prod
                            .li_context_governance_daily_agenda_v1
                    ) AS agenda_rows,
                    (
                        SELECT COUNT(*)
                        FROM
                            mv_reports_ucepsa_prod
                            .li_context_decision_group_backlog_v1
                    ) AS backlog_rows,
                    (
                        SELECT COUNT(*)
                        FROM (
                            SELECT
                                decision_group_key
                            FROM
                                mv_reports_ucepsa_prod
                                .li_context_governance_daily_agenda_v1
                            GROUP BY decision_group_key
                            HAVING COUNT(*) > 1
                        ) d
                    ) AS duplicates
            )
            SELECT
                CASE
                    WHEN agenda_rows =
                         backlog_rows
                     AND duplicates = 0
                        THEN 0
                    ELSE 1
                END AS violations,
                *
            FROM counts
            """,
        ),
        TestCase(
            code="AGENDA-002",
            group="GOVERNANCE",
            title="Ranking y plazos de agenda válidos",
            severity="CRITICAL",
            failure_status="FAIL",
            expected={"violations": 0},
            detail_pass="El ranking es continuo y los tramos de atención son válidos.",
            detail_fail="La agenda contiene rankings o estados de atención inválidos.",
            sql=r"""
            WITH ranked AS (
                SELECT
                    *,
                    ROW_NUMBER() OVER (
                        ORDER BY agenda_rank
                    ) AS expected_rank
                FROM
                    mv_reports_ucepsa_prod
                    .li_context_governance_daily_agenda_v1
            )
            SELECT
                COUNT(*) FILTER (
                    WHERE agenda_rank <>
                          expected_rank
                )
                + COUNT(*) FILTER (
                    WHERE attention_status NOT IN (
                        'RECENT',
                        'ATTENTION',
                        'OVERDUE'
                    )
                       OR pending_age_hours < 0
                       OR review_due_at <
                          first_seen_at
                ) AS violations
            FROM ranked
            """,
        ),
        TestCase(
            code="SNAPSHOT-001",
            group="GOVERNANCE",
            title="Snapshots de agenda internamente coherentes",
            severity="CRITICAL",
            failure_status="FAIL",
            expected={"violations": 0},
            detail_pass="Las cabeceras y los elementos de snapshot coinciden.",
            detail_fail="Un snapshot declara un número de elementos incorrecto.",
            sql=r"""
            SELECT COUNT(*) AS violations,
                   COUNT(*) AS snapshot_count
            FROM
                mv_loss_intelligence
                .context_governance_agenda_snapshots s
            WHERE s.item_count <>
                  (
                      SELECT COUNT(*)
                      FROM
                          mv_loss_intelligence
                          .context_governance_agenda_snapshot_items i
                      WHERE i.snapshot_id =
                            s.snapshot_id
                  )
            """,
        ),
        TestCase(
            code="SNAPSHOT-002",
            group="HISTORICAL_CONTRACT",
            title="Snapshot 1 conserva la primera agenda de tres decisiones",
            severity="CRITICAL",
            failure_status="FAIL",
            expected={"violations": 0, "row_count": 1},
            detail_pass="El primer snapshot conserva tres preguntas y modo SHADOW.",
            detail_fail="El snapshot 1 no coincide con la primera agenda guardada.",
            sql=r"""
            SELECT
                COUNT(*) AS row_count,
                COUNT(*) FILTER (
                    WHERE NOT (
                        agenda_local_date =
                            DATE '2026-07-20'
                        AND generated_by =
                            'Víctor Fraile'
                        AND item_count = 3
                        AND source_state_hash IS NOT NULL
                        AND length(
                            source_state_hash
                        ) = 64
                        AND NOT official_eligible
                    )
                ) AS violations
            FROM
                mv_loss_intelligence
                .context_governance_agenda_snapshots
            WHERE snapshot_id = 1
            """,
        ),
        TestCase(
            code="REVIEW-001",
            group="GOVERNANCE",
            title="Conteos pendientes coherentes en decisiones y casos",
            severity="CRITICAL",
            failure_status="FAIL",
            expected={"violations": 0},
            detail_pass="Cada decisión declara exactamente sus casos raíz pendientes.",
            detail_fail="El conteo pendiente de alguna decisión no coincide con sus casos.",
            sql=r"""
            SELECT COUNT(*) AS violations
            FROM
                mv_reports_ucepsa_prod
                .li_context_decision_group_history_v1 d
            WHERE d.pending_root_case_count <>
                  (
                      SELECT COUNT(*)
                      FROM
                          mv_reports_ucepsa_prod
                          .li_context_decision_group_cases_v1 c
                      WHERE c.decision_group_key =
                            d.decision_group_key
                        AND c.case_review_status =
                            'PENDING'
                  )
            """,
        ),
        TestCase(
            code="READINESS-001",
            group="READINESS",
            title="Readiness v3 coherente con sus gates",
            severity="CRITICAL",
            failure_status="FAIL",
            expected={"violations": 0},
            detail_pass="READY solo aparece cuando todos los gates pasan.",
            detail_fail="Existe una máquina READY con algún gate fallido.",
            sql=r"""
            SELECT COUNT(*) AS violations
            FROM
                mv_reports_ucepsa_prod
                .li_canonical_context_readiness_gate_v3
            WHERE readiness_status =
                  'READY_FOR_CONTROLLED_SHADOW'
              AND NOT (
                  technical_gate_pass
                  AND observation_gate_pass
                  AND review_gate_pass
                  AND coverage_gate_pass
              )
            """,
        ),
        TestCase(
            code="READINESS-002",
            group="READINESS",
            title="Consolidar nunca aumenta la carga humana",
            severity="CRITICAL",
            failure_status="FAIL",
            expected={"violations": 0},
            detail_pass="Las decisiones pendientes nunca superan a los casos raíz.",
            detail_fail="La consolidación produce más decisiones que casos raíz.",
            sql=r"""
            SELECT COUNT(*) AS violations
            FROM
                mv_reports_ucepsa_prod
                .li_canonical_context_readiness_gate_v3
            WHERE pending_decision_group_count >
                  pending_root_case_count
            """,
        ),
        TestCase(
            code="SHADOW-001",
            group="SHADOW_GUARD",
            title="Ninguna capa SHADOW habilita datos oficiales",
            severity="CRITICAL",
            failure_status="FAIL",
            expected={"violations": 0},
            detail_pass="Todas las guardas oficiales permanecen a cero.",
            detail_fail="Algún objeto SHADOW contiene filas oficiales.",
            sql=r"""
            SELECT
                (
                    SELECT COUNT(*)
                    FROM
                        mv_loss_intelligence
                        .production_context_sessions
                    WHERE official_eligible
                )
                + (
                    SELECT COUNT(*)
                    FROM
                        mv_loss_intelligence
                        .odoo_shopfloor_sessions
                    WHERE official_eligible
                )
                + (
                    SELECT COUNT(*)
                    FROM
                        mv_loss_intelligence
                        .context_review_cases
                    WHERE official_eligible
                )
                + (
                    SELECT COUNT(*)
                    FROM
                        mv_loss_intelligence
                        .context_governance_agenda_snapshots
                    WHERE official_eligible
                )
                + (
                    SELECT COUNT(*)
                    FROM
                        mv_loss_intelligence
                        .context_governance_agenda_snapshot_items
                    WHERE official_eligible
                )
                + (
                    SELECT COUNT(*)
                    FROM
                        mv_reports_ucepsa_prod
                        .li_canonical_context_segments_v1
                    WHERE official_ledger_eligible
                )
                + (
                    SELECT COUNT(*)
                    FROM
                        mv_reports_ucepsa_prod
                        .li_stop_canonical_context_shadow_v1
                    WHERE official_ledger_eligible
                )
                + (
                    SELECT COUNT(*)
                    FROM
                        mv_reports_ucepsa_prod
                        .li_context_decision_group_history_v1
                    WHERE official_eligible
                )
                + (
                    SELECT COUNT(*)
                    FROM
                        mv_reports_ucepsa_prod
                        .li_context_governance_daily_summary_v1
                    WHERE official_eligible
                )
                + (
                    SELECT COUNT(*)
                    FROM
                        mv_reports_ucepsa_prod
                        .li_operator_stop_queue_v2
                ) AS violations
            """,
        ),
        TestCase(
            code="HEALTH-001",
            group="RUNTIME_HEALTH",
            title="Sincronización y colectores saludables",
            severity="WARNING",
            failure_status="WARN",
            expected={"unhealthy_count": 0},
            detail_pass="Los componentes de captura y gobernanza están saludables.",
            detail_fail="Algún componente operativo no está HEALTHY.",
            sql=r"""
            SELECT
                (
                    CASE
                        WHEN (
                            SELECT MAX(
                                sync_health_status
                            )
                            FROM
                                mv_reports_ucepsa_prod
                                .li_shopfloor_sync_health_v1
                        ) = 'HEALTHY'
                            THEN 0
                        ELSE 1
                    END
                )
                + (
                    CASE
                        WHEN (
                            SELECT MAX(
                                collector_health_status
                            )
                            FROM
                                mv_reports_ucepsa_prod
                                .li_shopfloor_context_incident_collector_health_v1
                        ) = 'HEALTHY'
                            THEN 0
                        ELSE 1
                    END
                )
                + (
                    CASE
                        WHEN (
                            SELECT MAX(
                                builder_health_status
                            )
                            FROM
                                mv_reports_ucepsa_prod
                                .li_context_review_case_builder_health_v1
                        ) = 'HEALTHY'
                            THEN 0
                        ELSE 1
                    END
                ) AS unhealthy_count,
                (
                    SELECT MAX(
                        sync_health_status
                    )
                    FROM
                        mv_reports_ucepsa_prod
                        .li_shopfloor_sync_health_v1
                ) AS sync_health,
                (
                    SELECT MAX(
                        collector_health_status
                    )
                    FROM
                        mv_reports_ucepsa_prod
                        .li_shopfloor_context_incident_collector_health_v1
                ) AS collector_health,
                (
                    SELECT MAX(
                        builder_health_status
                    )
                    FROM
                        mv_reports_ucepsa_prod
                        .li_context_review_case_builder_health_v1
                ) AS root_case_builder_health
            """,
        ),
        TestCase(
            code="PERF-001",
            group="PERFORMANCE",
            title="Consulta de agenda dentro del umbral orientativo",
            severity="WARNING",
            failure_status="WARN",
            expected={"duration_ms_below_policy": True},
            detail_pass="La agenda responde dentro del umbral orientativo.",
            detail_fail="La consulta de agenda supera el umbral orientativo.",
            performance_only=True,
            sql=r"""
            SELECT COUNT(*) AS agenda_rows
            FROM
                mv_reports_ucepsa_prod
                .li_context_governance_daily_agenda_v1
            """,
        ),
    ]


def suite_hash(tests: list[TestCase]) -> str:
    payload = [
        {
            "code": test.code,
            "sql": " ".join(
                test.sql.split()
            ),
            "expected": test.expected,
            "failure_status":
                test.failure_status,
        }
        for test in tests
    ]
    return hashlib.sha256(
        json.dumps(
            payload,
            sort_keys=True,
            separators=(",", ":"),
        ).encode("utf-8")
    ).hexdigest()


def values_match(
    actual: dict[str, Any],
    expected: dict[str, Any],
) -> bool:
    for key, expected_value in expected.items():
        if key not in actual:
            return False
        actual_value = actual[key]
        if isinstance(expected_value, bool):
            if bool(actual_value) is not expected_value:
                return False
        elif actual_value != expected_value:
            return False
    return True


def load_policy(cursor) -> dict[str, Any]:
    cursor.execute(
        """
        SELECT *
        FROM
            mv_loss_intelligence.context_regression_policies
        WHERE tenant = 'ucepsa'
          AND site = 'ucepsa_onpremise'
          AND policy_code = %s
          AND active
        LIMIT 1
        """,
        (POLICY_CODE,),
    )
    row = cursor.fetchone()
    if not row:
        raise RuntimeError(
            f"No existe política activa {POLICY_CODE!r}"
        )
    return dict(row)


def execute_test(
    cursor,
    test: TestCase,
    performance_warning_ms: int,
) -> dict[str, Any]:
    started = time.perf_counter()
    savepoint = (
        "regression_"
        + test.code.lower().replace("-", "_")
    )
    cursor.execute(
        f"SAVEPOINT {savepoint}"
    )

    try:
        cursor.execute(test.sql)
        row = cursor.fetchone()
        actual = (
            dict(row)
            if row is not None
            else {}
        )
        duration_ms = (
            time.perf_counter()
            - started
        ) * 1000.0

        if test.performance_only:
            passed = (
                duration_ms <=
                performance_warning_ms
            )
            actual[
                "duration_ms_below_policy"
            ] = passed
            actual[
                "performance_warning_ms"
            ] = performance_warning_ms
        else:
            passed = values_match(
                actual,
                test.expected,
            )

        status = (
            "PASS"
            if passed
            else test.failure_status
        )
        detail = (
            test.detail_pass
            if passed
            else test.detail_fail
        )

        cursor.execute(
            f"RELEASE SAVEPOINT {savepoint}"
        )

        return {
            "test_code": test.code,
            "test_group": test.group,
            "test_title": test.title,
            "status": status,
            "severity": test.severity,
            "expected_json": test.expected,
            "actual_json": actual,
            "detail": detail,
            "duration_ms": round(
                duration_ms,
                2,
            ),
            "evidence_json": {
                "sql_hash": hashlib.sha256(
                    " ".join(
                        test.sql.split()
                    ).encode("utf-8")
                ).hexdigest(),
            },
        }

    except Exception as error:
        duration_ms = (
            time.perf_counter()
            - started
        ) * 1000.0
        cursor.execute(
            f"ROLLBACK TO SAVEPOINT {savepoint}"
        )
        cursor.execute(
            f"RELEASE SAVEPOINT {savepoint}"
        )

        return {
            "test_code": test.code,
            "test_group": test.group,
            "test_title": test.title,
            "status": "ERROR",
            "severity": test.severity,
            "expected_json": test.expected,
            "actual_json": {},
            "detail": str(error),
            "duration_ms": round(
                duration_ms,
                2,
            ),
            "evidence_json": {
                "error_type":
                    type(error).__name__,
            },
        }


def calculate_overall(
    results: list[dict[str, Any]],
) -> str:
    statuses = {
        result["status"]
        for result in results
    }
    if "ERROR" in statuses:
        return "ERROR"
    if "FAIL" in statuses:
        return "FAIL"
    if "WARN" in statuses:
        return "WARN"
    return "PASS"


def run_suite(
    triggered_by: str,
    trigger_kind: str,
    notes: str | None,
    git_commit: str | None,
    dry_run: bool,
) -> dict[str, Any]:
    tests = contract_tests()
    state_hash = suite_hash(tests)
    suite_started = time.perf_counter()

    with connect() as connection:
        with connection.cursor(
            cursor_factory=RealDictCursor,
        ) as cursor:
            policy = load_policy(cursor)
            performance_warning_ms = int(
                policy[
                    "performance_warning_ms"
                ]
            )

            run_id = None
            if not dry_run:
                cursor.execute(
                    """
                    INSERT INTO
                        mv_loss_intelligence
                        .context_regression_runs (
                            tenant,
                            site,
                            policy_code,
                            suite_version,
                            overall_status,
                            triggered_by,
                            trigger_kind,
                            source_git_commit,
                            source_state_hash,
                            notes,
                            official_eligible
                        )
                    VALUES (
                        'ucepsa',
                        'ucepsa_onpremise',
                        %s,
                        %s,
                        'RUNNING',
                        %s,
                        %s,
                        %s,
                        %s,
                        %s,
                        false
                    )
                    RETURNING run_id
                    """,
                    (
                        POLICY_CODE,
                        VERSION,
                        triggered_by,
                        trigger_kind,
                        git_commit,
                        state_hash,
                        notes,
                    ),
                )
                run_id = int(
                    cursor.fetchone()["run_id"]
                )
                connection.commit()

            results = []

            for test in tests:
                result = execute_test(
                    cursor,
                    test,
                    performance_warning_ms,
                )
                results.append(result)

                if not dry_run:
                    cursor.execute(
                        """
                        INSERT INTO
                            mv_loss_intelligence
                            .context_regression_results (
                                run_id,
                                test_code,
                                test_group,
                                test_title,
                                status,
                                severity,
                                expected_json,
                                actual_json,
                                detail,
                                duration_ms,
                                evidence_json,
                                official_eligible
                            )
                        VALUES (
                            %s, %s, %s, %s,
                            %s, %s, %s, %s,
                            %s, %s, %s, false
                        )
                        """,
                        (
                            run_id,
                            result["test_code"],
                            result["test_group"],
                            result["test_title"],
                            result["status"],
                            result["severity"],
                            Json(
                                json_safe(
                                    result[
                                        "expected_json"
                                    ]
                                )
                            ),
                            Json(
                                json_safe(
                                    result[
                                        "actual_json"
                                    ]
                                )
                            ),
                            result["detail"],
                            result["duration_ms"],
                            Json(
                                result[
                                    "evidence_json"
                                ]
                            ),
                        ),
                    )
                    connection.commit()

            overall_status = calculate_overall(
                results
            )
            counts = {
                status: sum(
                    1
                    for result in results
                    if result["status"] ==
                       status
                )
                for status in (
                    "PASS",
                    "WARN",
                    "FAIL",
                    "SKIP",
                    "ERROR",
                )
            }
            duration_ms = (
                time.perf_counter()
                - suite_started
            ) * 1000.0

            if not dry_run:
                cursor.execute(
                    """
                    UPDATE
                        mv_loss_intelligence
                        .context_regression_runs
                    SET
                        overall_status = %s,
                        finished_at = now(),
                        duration_ms = %s,
                        result_count = %s,
                        pass_count = %s,
                        warn_count = %s,
                        fail_count = %s,
                        skip_count = %s,
                        error_count = %s
                    WHERE run_id = %s
                    """,
                    (
                        overall_status,
                        round(duration_ms, 2),
                        len(results),
                        counts["PASS"],
                        counts["WARN"],
                        counts["FAIL"],
                        counts["SKIP"],
                        counts["ERROR"],
                        run_id,
                    ),
                )
                connection.commit()

    return {
        "version": VERSION,
        "status": (
            "dry_run"
            if dry_run
            else "completed"
        ),
        "run_id": run_id,
        "overall_status": overall_status,
        "duration_ms": round(
            duration_ms,
            2,
        ),
        "result_count": len(results),
        "counts": counts,
        "source_git_commit": git_commit,
        "source_state_hash": state_hash,
        "results": results,
    }


def list_runs(limit: int) -> None:
    with connect() as connection:
        with connection.cursor(
            cursor_factory=RealDictCursor,
        ) as cursor:
            cursor.execute(
                """
                SELECT *
                FROM
                    mv_reports_ucepsa_prod
                    .li_context_regression_run_history_v1
                ORDER BY
                    started_at DESC,
                    run_id DESC
                LIMIT %s
                """,
                (limit,),
            )
            print_json(
                [dict(row) for row in cursor.fetchall()]
            )


def show_run(run_id: int | None) -> None:
    with connect() as connection:
        with connection.cursor(
            cursor_factory=RealDictCursor,
        ) as cursor:
            if run_id is None:
                cursor.execute(
                    """
                    SELECT *
                    FROM
                        mv_reports_ucepsa_prod
                        .li_context_regression_latest_run_v1
                    WHERE tenant = 'ucepsa'
                      AND site = 'ucepsa_onpremise'
                    """
                )
            else:
                cursor.execute(
                    """
                    SELECT *
                    FROM
                        mv_loss_intelligence
                        .context_regression_runs
                    WHERE run_id = %s
                    """,
                    (run_id,),
                )

            run = cursor.fetchone()
            if not run:
                raise RuntimeError(
                    "No existe una ejecución de regresión."
                )

            cursor.execute(
                """
                SELECT *
                FROM
                    mv_loss_intelligence
                    .context_regression_results
                WHERE run_id = %s
                ORDER BY
                    CASE status
                        WHEN 'ERROR' THEN 1
                        WHEN 'FAIL' THEN 2
                        WHEN 'WARN' THEN 3
                        WHEN 'SKIP' THEN 4
                        ELSE 5
                    END,
                    test_group,
                    test_code
                """,
                (run["run_id"],),
            )
            results = [
                dict(row)
                for row in cursor.fetchall()
            ]

            print_json({
                "run": dict(run),
                "results": results,
            })


def markdown_report(
    run: dict[str, Any],
    results: list[dict[str, Any]],
) -> str:
    lines = [
        f"# Regresión de contexto UCEPSA — ejecución {run['run_id']}",
        "",
        f"- Suite: {run['suite_version']}",
        f"- Estado: {run['overall_status']}",
        f"- Inicio: {run['started_at']}",
        f"- Fin: {run['finished_at']}",
        f"- Disparada por: {run['triggered_by']}",
        f"- Tipo: {run['trigger_kind']}",
        f"- Commit: {run.get('source_git_commit') or '—'}",
        f"- Duración: {run.get('duration_ms') or '—'} ms",
        "- Modo: SHADOW",
        "- Ledger oficial: desactivado",
        "",
        "## Resumen",
        "",
        f"- PASS: {run['pass_count']}",
        f"- WARN: {run['warn_count']}",
        f"- FAIL: {run['fail_count']}",
        f"- ERROR: {run['error_count']}",
        f"- SKIP: {run['skip_count']}",
        "",
        "## Resultados",
        "",
    ]

    for result in results:
        lines.extend([
            f"### {result['status']} · {result['test_code']}",
            "",
            f"**{result['test_title']}**",
            "",
            f"- Grupo: {result['test_group']}",
            f"- Severidad: {result['severity']}",
            f"- Duración: {result['duration_ms']} ms",
            f"- Detalle: {result.get('detail') or '—'}",
            "",
            "```json",
            json.dumps(
                json_safe(
                    result["actual_json"]
                ),
                ensure_ascii=False,
                indent=2,
            ),
            "```",
            "",
        ])

    return "\n".join(lines) + "\n"


def export_markdown(run_id: int | None) -> None:
    with connect() as connection:
        with connection.cursor(
            cursor_factory=RealDictCursor,
        ) as cursor:
            if run_id is None:
                cursor.execute(
                    """
                    SELECT *
                    FROM
                        mv_reports_ucepsa_prod
                        .li_context_regression_latest_run_v1
                    WHERE tenant = 'ucepsa'
                      AND site = 'ucepsa_onpremise'
                    """
                )
            else:
                cursor.execute(
                    """
                    SELECT *
                    FROM
                        mv_loss_intelligence
                        .context_regression_runs
                    WHERE run_id = %s
                    """,
                    (run_id,),
                )
            run = cursor.fetchone()
            if not run:
                raise RuntimeError(
                    "No existe una ejecución."
                )

            cursor.execute(
                """
                SELECT *
                FROM
                    mv_loss_intelligence
                    .context_regression_results
                WHERE run_id = %s
                ORDER BY
                    test_group,
                    test_code
                """,
                (run["run_id"],),
            )
            results = [
                dict(row)
                for row in cursor.fetchall()
            ]

    sys.stdout.write(
        markdown_report(
            dict(run),
            results,
        )
    )


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description=(
            "Regresión y hardening de contexto "
            "UCEPSA."
        )
    )
    sub = parser.add_subparsers(
        dest="command",
        required=True,
    )

    run = sub.add_parser("run")
    run.add_argument(
        "--triggered-by",
        required=True,
    )
    run.add_argument(
        "--trigger-kind",
        choices=(
            "MANUAL",
            "DEPLOYMENT",
            "SCHEDULED",
            "CI",
        ),
        default="MANUAL",
    )
    run.add_argument("--notes")
    run.add_argument(
        "--git-commit",
        default=env_first(
            "SOURCE_GIT_COMMIT"
        ),
    )

    dry = sub.add_parser("dry-run")
    dry.add_argument(
        "--triggered-by",
        default="MESAVAULT",
    )
    dry.add_argument(
        "--git-commit",
        default=env_first(
            "SOURCE_GIT_COMMIT"
        ),
    )

    list_cmd = sub.add_parser("list")
    list_cmd.add_argument(
        "--limit",
        type=int,
        default=20,
    )

    show = sub.add_parser("show")
    show.add_argument(
        "--run-id",
        type=int,
    )

    export = sub.add_parser("export-markdown")
    export.add_argument(
        "--run-id",
        type=int,
    )

    return parser.parse_args()


def main() -> int:
    args = parse_args()

    if args.command == "run":
        result = run_suite(
            triggered_by=args.triggered_by,
            trigger_kind=args.trigger_kind,
            notes=args.notes,
            git_commit=args.git_commit,
            dry_run=False,
        )
        print_json(result)
        return (
            0
            if result["overall_status"] in (
                "PASS",
                "WARN",
            )
            else 1
        )

    if args.command == "dry-run":
        result = run_suite(
            triggered_by=args.triggered_by,
            trigger_kind="MANUAL",
            notes="Dry-run no persistente.",
            git_commit=args.git_commit,
            dry_run=True,
        )
        print_json(result)
        return (
            0
            if result["overall_status"] in (
                "PASS",
                "WARN",
            )
            else 1
        )

    if args.command == "list":
        list_runs(args.limit)
        return 0

    if args.command == "show":
        show_run(args.run_id)
        return 0

    if args.command == "export-markdown":
        export_markdown(args.run_id)
        return 0

    raise RuntimeError("Comando no soportado")


if __name__ == "__main__":
    raise SystemExit(main())