#!/usr/bin/env python3 """Run a strictly read-only BLIF Flow v2 shadow/cutover audit.""" from __future__ import annotations import argparse import json import os import sys from collections import Counter from datetime import datetime, timezone from pathlib import Path from typing import Any, Sequence ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(ROOT)) os.chdir(ROOT) PRODUCTION_DATABASE = "clientflow" TEST_DATABASE = "clientflow_codex_test" AUDIT = Path("/tmp/blif_flow_v2_production_shadow_audit.json") SEMANTIC = Path("/tmp/blif_flow_v2_production_semantic_compare.json") SUMMARY = Path("/tmp/blif_flow_v2_production_shadow_summary.txt") CLASSIFICATIONS = { "SEMANTICALLY_EQUIVALENT", "V1_OPERATIONAL_OVERRIDE", "V2_CORRECTS_V1", "LEGACY_ONLY", "REAL_CONFLICT", "MISSING_PROJECTION", } CURRENT_QUEUES = {"do_now", "review", "exception"} OVERRIDE_PRECEDENCE = { "scheduled_call", "due_followup", "future_followup", "integration_exception", "document_prerequisite", "fiscal_prerequisite", "safe_preserve_v1", } LEGACY_ACTIONS = { "CREATE_JASMIN_QUOTE", "NO_ACTION", "WAIT_CUSTOMER", "WAIT_PAYMENT", "WAIT_PRODUCTION", "WAIT_LOGISTICS", "WAIT_SUPPLIER", "WAIT_SCHEDULED_DATE", } REVIEW_ACTIONS = { "REVIEW", "REVIEW_REQUIRED", "REVIEW_MANUALLY", "REVIEW_RECONSTRUCTED_PROCESS", "RECONCILE_DOCUMENTS", "VALIDATE_FISCAL_CUSTOMER", "REVIEW_EXCEPTION", } NAMED_IDS = { "INSTALBEIRA": "5c33db95-fab8-477a-bddd-0b9cc8f91302", "PANORAMIC": "fd79b9a1-07e6-4f61-95e8-09eab89c155e", "ENGEXICON": "61f1c955-a372-4ea7-b9b0-b8528d74a141", "CONSTRURECUP": "e3b23ac5-84db-4763-8a31-a684e873032c", "X_MAT_CANONICAL": "dc89a466-db24-401b-bfe9-d47644b2d0c8", "X_MAT_DUPLICATE": "1816a06e-9a69-4a9b-9279-1263156892d3", "RZSOLAR_CANONICAL": "fd221608-e007-4043-a23d-07e0c119a345", "RZSOLAR_DUPLICATE": "434124fb-ac19-4d78-909a-55761d7e8daa", } def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument( "--production-readonly-audit", action="store_true", help="explicitly authorize a read-only audit of database clientflow", ) return parser def validate_execution( *, production_readonly_audit: bool, database: str, transaction_read_only: str, mode: str, ) -> None: normalized_mode = str(mode or "").strip().lower() if production_readonly_audit: if database != PRODUCTION_DATABASE: raise RuntimeError( f"--production-readonly-audit requires database {PRODUCTION_DATABASE!r}, found {database!r}" ) if transaction_read_only != "on": raise RuntimeError("production audit requires transaction_read_only=on") if normalized_mode not in {"shadow", "compare"}: raise RuntimeError("production audit requires BLIF_FLOW_V2_MODE=shadow or compare") return if database == PRODUCTION_DATABASE: raise RuntimeError("production database requires explicit --production-readonly-audit opt-in") if database != TEST_DATABASE: raise RuntimeError(f"default audit requires database {TEST_DATABASE!r}, found {database!r}") if transaction_read_only != "on": raise RuntimeError("cutover audit requires an explicit READ ONLY transaction") def _code(value: Any) -> str: return str(value or "").strip().upper() def classify_semantic_difference( *, v1_state: str | None, v1_action: str | None, v2_state: str | None, v2_action: str | None, v2_projection_present: bool = True, is_duplicate_representation: bool = False, operational_action: str | None = None, operational_precedence: str | None = None, operational_queue: str | None = None, v2_confidence: str | None = None, v2_diagnostic_status: str | None = None, ) -> tuple[str, str]: """Conservatively compare meanings rather than raw action vocabulary.""" if not v2_projection_present: return "MISSING_PROJECTION", "No persisted Flow v2 projection exists for this opportunity." v1, v2, effective = _code(v1_action), _code(v2_action), _code(operational_action) precedence = str(operational_precedence or "").strip().lower() queue = str(operational_queue or "").strip().lower() if is_duplicate_representation: return "V2_CORRECTS_V1", "Material identity suppresses a duplicate representation without deleting evidence." if precedence in OVERRIDE_PRECEDENCE and effective and (v1 == effective or queue in CURRENT_QUEUES | {"waiting"}): return "V1_OPERATIONAL_OVERRIDE", f"Explicit operational precedence {precedence} validly overlays the V2 business transition." if v1 == v2 and v1: return "SEMANTICALLY_EQUIVALENT", "V1 and V2 select the same action." if not v1 and not v2: return "SEMANTICALLY_EQUIVALENT", "Neither model has a current business action." if v1 in REVIEW_ACTIONS and (v2 in REVIEW_ACTIONS or _code(v2_state) in {"REVIEW_REQUIRED", "FISCAL_BLOCKED", "DOCUMENT_RECONCILIATION_REQUIRED"}): return "SEMANTICALLY_EQUIVALENT", "Both decisions require review/blocker handling." if v1 in {"NO_ACTION", ""} and not v2 and _code(v2_state) in {"COMPLETED", "LOST", "NO_INTEREST"}: return "SEMANTICALLY_EQUIVALENT", "Both decisions represent a terminal/non-current process." if v1.startswith("FOLLOW_UP_") and _code(v2_state) in {"AWAITING_CUSTOMER", "AWAITING_PAYMENT"}: return "V1_OPERATIONAL_OVERRIDE", "A current follow-up obligation overlays a waiting V2 business state." if v1 == "CONFIRM_PAYMENT" and _code(v2_state) == "AWAITING_PAYMENT" and not v2: return "SEMANTICALLY_EQUIVALENT", "Both decisions mean payment remains outstanding; V1 names the compatibility action." if v1 in {"VALIDATE_FISCAL_CUSTOMER", "RECONCILE_DOCUMENTS"} and not effective and ( queue in {"not_current", "backlog"} or _code(v2_state) in {"INQUIRY", "AWAITING_CUSTOMER", "AWAITING_PAYMENT"} ): return "LEGACY_ONLY", "Legacy data-hygiene/blocker vocabulary is not a current factual V2 obligation." if precedence in {"safe_diagnostic_only", "safe_ambiguous_review"}: return "V2_CORRECTS_V1", "SAFE V2 prevents ambiguous historical compatibility state from creating current work." if _code(v2_state) in {"REVIEW_REQUIRED", "FISCAL_BLOCKED", "DOCUMENT_RECONCILIATION_REQUIRED"}: return "V2_CORRECTS_V1", "V2 converts conflicting factual history into an explicit protected review state." if effective and effective == v2 and v1 != v2: return "V2_CORRECTS_V1", "The safe operational action follows the factual V2 transition rather than the legacy action." if v1 in LEGACY_ACTIONS: if v2 and _code(v2_state) not in {"AWAITING_CUSTOMER", "AWAITING_PAYMENT"}: return "V2_CORRECTS_V1", "V2 replaces a legacy compatibility action with a factual business transition." return "LEGACY_ONLY", "V1 action is compatibility vocabulary with no native V2 business transition." if v2 and str(v2_confidence or "").lower() == "high" and str(v2_diagnostic_status or "").lower() == "clear": return "V2_CORRECTS_V1", "High-confidence factual V2 transition corrects a different legacy action." if effective and v1 == effective: return "V1_OPERATIONAL_OVERRIDE", "V1 matches the safe operational overlay rather than the business action." return "REAL_CONFLICT", "The available evidence does not establish equivalence, a valid override, or a safe V2 correction." def _write(path: Path, value: Any) -> None: path.write_text(json.dumps(value, ensure_ascii=False, indent=2, default=str), encoding="utf-8") def _queue_totals(rows: list[dict[str, Any]], key: str) -> dict[str, int]: counts = Counter(str(row[key].get("operational_queue") or row[key].get("effective_operational_queue") or "not_current") for row in rows) return { "current": sum(counts[name] for name in CURRENT_QUEUES), "do_now": counts["do_now"], "review": counts["review"], "waiting": counts["waiting"], "backlog": counts["backlog"], "not_current": counts["not_current"], } def _operation_totals(source: dict[str, Any]) -> dict[str, int]: return {"current": int(source.get("current_work") or 0), "do_now": int(source.get("do_now") or 0), "review": int(source.get("review") or 0), "waiting": int(source.get("waiting") or 0), "backlog": int(source.get("backlog") or 0), "not_current": int(source.get("not_current") or 0)} def _database_snapshot(engine: Any) -> tuple[dict[str, str], list[str], dict[str, dict[str, Any]]]: from sqlalchemy import text with engine.connect() as conn: conn.exec_driver_sql("BEGIN READ ONLY") try: identity = conn.execute(text( "SELECT current_database(),current_user,current_setting('transaction_read_only')" )).one() opportunity_ids = [str(value) for value in conn.execute(text( "SELECT id FROM opportunities ORDER BY id" )).scalars()] rows = conn.execute(text(""" SELECT opportunity_id::text, material_process_key, canonical_opportunity_id::text, is_duplicate_representation, business_state, business_next_action, diagnostic_status, confidence, reason_code, reason_text, evidence_refs, flow_version FROM opportunity_flow_state_v2 ORDER BY opportunity_id """)).mappings().all() finally: conn.rollback() return ( {"database": identity[0], "user": identity[1], "transaction_read_only": identity[2]}, opportunity_ids, {str(row["opportunity_id"]): dict(row) for row in rows}, ) def run_audit(*, production_readonly_audit: bool) -> dict[str, Any]: # Imports occur only after argparse, so --help cannot initialize DB code. from app.config import settings from app.db import engine from scripts.simulate_blif_flow_v2 import collect identity, opportunity_ids, persisted = _database_snapshot(engine) mode = str(settings.blif_flow_v2_mode or "").strip().lower() validate_execution( production_readonly_audit=production_readonly_audit, database=identity["database"], transaction_read_only=identity["transaction_read_only"], mode=mode, ) preflight = {**identity, "BLIF_FLOW_V2_MODE": mode, "production_readonly_opt_in": production_readonly_audit} print(json.dumps(preflight, sort_keys=True), flush=True) expected_count = None if production_readonly_audit else 328 projection = collect( expected_database=identity["database"], expected_user=identity["user"], require_read_only=production_readonly_audit, expected_opportunity_count=expected_count, require_opportunities=True, ) records = {row["opportunity_id"]: row for row in projection["opportunities"]} if set(records) != set(opportunity_ids): raise RuntimeError("factual collector did not return the complete opportunity universe") comparisons = [] for opportunity_id in opportunity_ids: record = records[opportunity_id] v1, operational = record["v1"], record["safe_v2"] v2 = persisted.get(opportunity_id) classification, reason = classify_semantic_difference( v1_state=v1.get("commercial_stage"), v1_action=v1.get("current_action"), v2_state=(v2 or {}).get("business_state"), v2_action=(v2 or {}).get("business_next_action"), v2_projection_present=v2 is not None, is_duplicate_representation=bool((v2 or {}).get("is_duplicate_representation")), operational_action=operational.get("effective_operational_action"), operational_precedence=operational.get("precedence"), operational_queue=operational.get("effective_operational_queue"), v2_confidence=(v2 or {}).get("confidence"), v2_diagnostic_status=(v2 or {}).get("diagnostic_status"), ) comparisons.append({ "opportunity_id": opportunity_id, "title": record.get("title"), "customer_name": record.get("customer"), "classification": classification, "classification_reason": reason, "v1": {"state": v1.get("commercial_stage"), "action": v1.get("current_action"), "queue": v1.get("operational_queue"), "reason_code": v1.get("reason")}, "v2": {"business_state": (v2 or {}).get("business_state"), "business_next_action": (v2 or {}).get("business_next_action"), "reason_code": (v2 or {}).get("reason_code"), "diagnostic_status": (v2 or {}).get("diagnostic_status"), "confidence": (v2 or {}).get("confidence")}, "operational_override": {"action": operational.get("effective_operational_action"), "queue": operational.get("effective_operational_queue"), "precedence": operational.get("precedence")}, "material_process_key": (v2 or {}).get("material_process_key"), "canonical_opportunity_id": (v2 or {}).get("canonical_opportunity_id"), "is_duplicate_representation": bool((v2 or {}).get("is_duplicate_representation")), "evidence_refs": (v2 or {}).get("evidence_refs", []), }) counts = Counter(row["classification"] for row in comparisons) for name in CLASSIFICATIONS: counts.setdefault(name, 0) real_conflicts = [row for row in comparisons if row["classification"] == "REAL_CONFLICT"] missing = [row for row in comparisons if row["classification"] == "MISSING_PROJECTION"] semantic_report = { "generated_at": datetime.now(timezone.utc), "preflight": preflight, "opportunity_count": len(opportunity_ids), "projection_count": len(persisted), "classification_counts": dict(sorted(counts.items())), "real_conflicts": real_conflicts, "missing_projections": missing, "comparisons": comparisons, } _write(SEMANTIC, semantic_report) named = {} by_id = {row["opportunity_id"]: row for row in comparisons} for name, opportunity_id in NAMED_IDS.items(): row = by_id[opportunity_id] named[name] = row # Use the complete canonical Operations candidate universe, including # preserved standalone obligations, rather than opportunity cards alone. v1_metrics = _operation_totals(projection["v1_totals"]) safe_metrics = _operation_totals(projection["safe_v2_totals"]) audit = { "generated_at": datetime.now(timezone.utc), "preflight": preflight, "read_only": True, "business_writes": 0, "projection_writes": 0, "opportunity_count": len(opportunity_ids), "projection_count": len(persisted), "canonical_count": sum(not row.get("is_duplicate_representation") for row in persisted.values()), "duplicate_representation_count": sum(bool(row.get("is_duplicate_representation")) for row in persisted.values()), "operations_metrics": {"v1": v1_metrics, "safe_v2": safe_metrics}, "semantic_classification_counts": dict(sorted(counts.items())), "authoritative_cutover_blockers": { "real_conflicts": len(real_conflicts), "missing_projections": len(missing), "blocked": bool(real_conflicts or missing), }, "real_conflicts": real_conflicts, "missing_projections": missing, "named_cases": named, } _write(AUDIT, audit) lines = [ "BLIF FLOW V2 PRODUCTION SHADOW READ-ONLY AUDIT", "", f"database: {identity['database']}", f"user: {identity['user']}", f"transaction_read_only: {identity['transaction_read_only']}", f"BLIF_FLOW_V2_MODE: {mode}", f"production_readonly_opt_in: {production_readonly_audit}", "", f"opportunities: {len(opportunity_ids)}", f"projections: {len(persisted)}", f"canonical: {audit['canonical_count']}", f"duplicate representations: {audit['duplicate_representation_count']}", "", f"V1 metrics: {json.dumps(v1_metrics, sort_keys=True)}", f"SAFE V2 metrics: {json.dumps(safe_metrics, sort_keys=True)}", "", f"semantic classifications: {json.dumps(dict(sorted(counts.items())), sort_keys=True)}", f"REAL_CONFLICT blockers: {len(real_conflicts)}", f"MISSING_PROJECTION blockers: {len(missing)}", "business writes: 0", "projection writes: 0", ] SUMMARY.write_text("\n".join(lines) + "\n", encoding="utf-8") return audit def main(argv: Sequence[str] | None = None) -> int: # --help exits here before application/database imports or connections. args = build_parser().parse_args(argv) result = run_audit(production_readonly_audit=args.production_readonly_audit) print(json.dumps({ "database": result["preflight"]["database"], "opportunities": result["opportunity_count"], "projections": result["projection_count"], "semantic_classifications": result["semantic_classification_counts"], "authoritative_cutover_blockers": result["authoritative_cutover_blockers"], "outputs": [str(AUDIT), str(SEMANTIC), str(SUMMARY)], }, indent=2, sort_keys=True)) return 0 if __name__ == "__main__": raise SystemExit(main())