feat: add BLIF Flow v2 shadow compare cutover layer

app/admin_ui/pages/opportunities.py

@@ -1544,20 +1544,14 @@ def _parse_opportunity_dt(value: object):
 def _opportunity_lifecycle_state(opp: dict) -> str:
     state = str(opp.get("lifecycle_state") or "active").strip().lower() or "active"
     now = datetime.now(timezone.utc)
-    pending_call = str(opp.get("pending_follow_up_action_code") or "").strip().upper() == "CALL_CUSTOMER"
-    pending_call_due = _parse_opportunity_dt(opp.get("pending_follow_up_due_at"))
-    if pending_call:
-        return "follow_up_due" if pending_call_due and pending_call_due <= now else "scheduled_follow_up"
-    # Compatibility only: a stale denormalized lifecycle marker cannot invent
-    # a scheduled call when no active CALL_CUSTOMER task exists.
-    if state == "scheduled_follow_up":
+    pending_followup = str(opp.get("pending_follow_up_action_code") or "").strip().upper()
+    pending_due = _parse_opportunity_dt(opp.get("pending_follow_up_due_at"))
+    if pending_followup:
+        return "follow_up_due" if pending_due and pending_due <= now else "scheduled_follow_up"
+    # Compatibility-only timestamps/lifecycle markers cannot create work. An
+    # active pending follow-up task and its due_at are the operational source.
+    if state in {"scheduled_follow_up", "follow_up_due"}:
         return "active"
-    nurture_until = _parse_opportunity_dt(opp.get("nurture_until"))
-    next_follow_up = _parse_opportunity_dt(opp.get("next_follow_up_at"))
-    if state == "nurture" and nurture_until and nurture_until <= now:
-        return "follow_up_due"
-    if state in {"awaiting_customer", "active", "scheduled_follow_up"} and next_follow_up and next_follow_up <= now:
-        return "follow_up_due"
     return state
 
 

app/blif_flow_v2_projection_service.py

@@ -102,14 +102,15 @@ def rebuild_blif_flow_v2_projection(
 ) -> dict[str, Any]:
     """Idempotently rebuild projection rows and state-change transitions.
 
-    ``off`` is a no-op. ``shadow`` writes only additive projection tables.
-    Compare/authoritative behavior is deliberately not implemented.
+    ``off`` is a no-op. ``shadow`` and ``compare`` write only the same additive
+    projection tables. Compare affects observation at the read boundary, never
+    the persisted business rows. Authoritative remains deliberately disabled.
     """
     selected_mode = str(mode or settings.blif_flow_v2_mode or "off").strip().lower()
     if selected_mode == "off":
         return {"mode": "off", "projection_count": 0, "transitions_written": 0, "disabled": True}
-    if selected_mode != "shadow":
-        raise RuntimeError(f"BLIF Flow v2 mode {selected_mode!r} is not implemented; only off/shadow are safe")
+    if selected_mode not in {"shadow", "compare"}:
+        raise RuntimeError(f"BLIF Flow v2 mode {selected_mode!r} is disabled; only off/shadow/compare are safe")
     if not re.fullmatch(r"[a-z_][a-z0-9_]*", target_schema):
         raise ValueError("invalid target_schema")
 
@@ -208,7 +209,7 @@ def rebuild_blif_flow_v2_projection(
             conn.close()
 
     return {
-        "mode": "shadow", "projection_count": len(values),
+        "mode": selected_mode, "projection_count": len(values),
         "canonical_count": sum(not value["is_duplicate_representation"] for value in values),
         "duplicate_count": sum(value["is_duplicate_representation"] for value in values),
         "transitions_written": transitions_written, "disabled": False,

app/config.py

@@ -21,8 +21,9 @@ class Settings(BaseSettings):
     # TIMESTAMPTZ nem os índices usados pelo schema core.
     database_url: str
     clientflow_persist: bool = True
-    # Flow v2 is additive shadow scaffolding only. Compare/authoritative values
-    # are reserved for future work and are not activated by this implementation.
+    # Flow v2 remains non-authoritative. Compare observes V1/V2 differences
+    # while returning V1; authoritative is accepted by config only to fail
+    # closed at the switching boundary.
     blif_flow_v2_mode: Literal["off", "shadow", "compare", "authoritative"] = "off"
 
     clientflow_webhook_secret: str = ""

app/opportunity_next_action_service.py

@@ -7,11 +7,14 @@ from __future__ import annotations
 
 from dataclasses import asdict, dataclass
 from collections import defaultdict
+import json
+import logging
 from typing import Any, Dict, Iterable, Optional
 
 from sqlalchemy import bindparam, text
 
 from app.db import engine
+from app.config import settings
 # Backward-compatible static anchors from v1.5.59: quotation_doc, invoice_doc, confirmar pagamento antes de emitir fatura, Pagamento confirmado com base em, Criar/enviar fatura.
 from app.domain.opportunity_flow import (
     OpportunityEvidence,
@@ -21,6 +24,9 @@ from app.domain.opportunity_flow import (
 )
 
 
+logger = logging.getLogger(__name__)
+
+
 @dataclass
 class OpportunityNextAction:
     action_code: str
@@ -37,6 +43,67 @@ class OpportunityNextAction:
         return asdict(self)
 
 
+def _flow_v2_mode() -> str:
+    return str(settings.blif_flow_v2_mode or "off").strip().lower()
+
+
+def _load_v2_comparison_rows(opportunity_ids: list[str]) -> dict[str, dict[str, Any]]:
+    """Read the persisted projection only; comparison must never derive writes."""
+    if not opportunity_ids:
+        return {}
+    with engine.connect() as conn:
+        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, flow_version
+            FROM opportunity_flow_state_v2
+            WHERE opportunity_id::text IN :opportunity_ids
+        """).bindparams(bindparam("opportunity_ids", expanding=True)),
+            {"opportunity_ids": opportunity_ids}).mappings().all()
+    return {str(row["opportunity_id"]): dict(row) for row in rows}
+
+
+def compare_v1_v2_decisions(
+    v1_decisions: Dict[str, Dict[str, Any]],
+    v2_rows: dict[str, dict[str, Any]],
+) -> list[dict[str, Any]]:
+    """Return payload-safe structured comparisons without message content."""
+    comparisons = []
+    for opportunity_id, v1 in v1_decisions.items():
+        v2 = v2_rows.get(opportunity_id)
+        v1_action = str(v1.get("action_code") or "") or None
+        v2_action = (v2 or {}).get("business_next_action")
+        comparisons.append({
+            "opportunity_id": opportunity_id,
+            "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")),
+            "v1_action": v1_action,
+            "v1_reason_code": v1.get("reason_if_blocked") or v1.get("decision_version"),
+            "v2_business_state": (v2 or {}).get("business_state"),
+            "v2_business_next_action": v2_action,
+            "v2_reason_code": (v2 or {}).get("reason_code"),
+            "v2_diagnostic_status": (v2 or {}).get("diagnostic_status"),
+            "v2_confidence": (v2 or {}).get("confidence"),
+            "projection_present": v2 is not None,
+            "action_agrees": v2 is not None and v1_action == v2_action,
+            "returned_source": "v1",
+        })
+    return comparisons
+
+
+def _observe_flow_v2(v1_decisions: Dict[str, Dict[str, Any]]) -> None:
+    mode = _flow_v2_mode()
+    if mode == "authoritative":
+        raise RuntimeError("BLIF Flow v2 authoritative mode is disabled; cutover boundary is fail-closed")
+    if mode != "compare" or not v1_decisions:
+        return
+    rows = _load_v2_comparison_rows(list(v1_decisions))
+    for comparison in compare_v1_v2_decisions(v1_decisions, rows):
+        logger.info("blif_flow_v2_compare %s", json.dumps(comparison, sort_keys=True, default=str))
+
+
 def _first_row(conn: Any, sql: str, params: Dict[str, Any]) -> Optional[Dict[str, Any]]:
     row = conn.execute(text(sql), params).mappings().first()
     return dict(row) if row else None
@@ -160,6 +227,8 @@ def get_opportunity_next_action(
     decision is now produced by the company workflow engine.
     """
 
+    if _flow_v2_mode() == "authoritative":
+        raise RuntimeError("BLIF Flow v2 authoritative mode is disabled; cutover boundary is fail-closed")
     evidence = (_build_db_evidence(opportunity_id, preloaded=preloaded)
                 if preloaded is not None else _build_db_evidence(opportunity_id))
     if evidence is None:
@@ -172,8 +241,9 @@ def get_opportunity_next_action(
             reason_if_blocked="opportunity_not_found",
         ).to_dict()
 
-    decision = decide_opportunity_next_action(evidence, load_company_profile(evidence.company_profile))
-    return decision.to_dict()
+    decision = decide_opportunity_next_action(evidence, load_company_profile(evidence.company_profile)).to_dict()
+    _observe_flow_v2({opportunity_id: decision})
+    return decision
 
 
 def _bulk_statement(sql: str):
@@ -298,6 +368,8 @@ def _bulk_operation_snapshots(
 
 def get_opportunity_next_actions(opportunity_ids: Iterable[str]) -> Dict[str, Dict[str, Any]]:
     """Return the same decisions as the single-item API with a fixed query count."""
+    if _flow_v2_mode() == "authoritative":
+        raise RuntimeError("BLIF Flow v2 authoritative mode is disabled; cutover boundary is fail-closed")
     ids = list(dict.fromkeys(str(value).strip() for value in opportunity_ids if str(value).strip()))
     if not ids:
         return {}
@@ -376,4 +448,5 @@ def get_opportunity_next_actions(opportunity_ids: Iterable[str]) -> Dict[str, Di
             company_profile="blif",
         )
         decisions[oid] = decide_opportunity_next_action(evidence, profile).to_dict()
+    _observe_flow_v2(decisions)
     return decisions