AI Conversation Engine

# LangGraph offers granular control over flow
graph = StateGraph(
    state_schema=ConversationState,  # State typed
    context_schema=ContextSchema,    # Injected context
)

# Conditional edges based on state
graph.add_conditional_edges(
    "semantic_routing",
    lambda state: route_by_intent(state),  # Pure function
    {
        "pricing": "pricing_fast_lane",
        "appointment": "booking_flow_router",
        "information": "agent_node",
        "greeting": "format_response",
    }
)

# Differentiated RetryPolicy by error type
retry_policy = RetryPolicy(
    max_attempts=3,
    initial_interval=0.5,
    backoff_factor=2.0,
    retry_on=TRANSIENT_EXCEPTIONS,  # Only transient!
)

async def semantic_routing_node(state: dict, service: Any, config: Any):
    """
    Semantic Routing (Pre-LLM).
    
    1. Check Priority-0 flags (Booking/Handover) → If yes, ignore semantic
    2. Call SemanticRouter.classify() via embeddings
    3. If High Confidence:
       a. Create synthetic structured_analysis
       b. Validate via RoutingDecisionEngine (Policies/Circuit Breaker)
       c. Set routing_decision + fast_path_taken=True
    4. If Low Confidence: add hints, proceed to LLM
    """
    
    # Priority-0: deterministic states have absolute priority
    pending_action = state.get("pending_action", "")
    if pending_action.startswith("confirm_"):
        # User is confirming something - don't reclassify
        return {"skip_semantic": True, "reason": "pending_confirmation"}
    
    # Classification via embeddings (~7ms p95)
    result = await semantic_router.classify(
        text=user_message,
        hybrid_alpha=0.7,        # 70% embedding, 30% keyword
        abstain_threshold=0.55,  # Below this → goes to LLM
        bias_weight=0.03,        # Boost for priority intents
    )
    
    if result.confidence >= 0.85:
        # High confidence: bypass LLM
        return {
            "routing_decision": result.route,
            "fast_path_taken": True,
            "semantic_confidence": result.confidence,
        }
    else:
        # Low confidence: hints for the LLM
        return {
            "semantic_hints": result.top_routes,
            "fast_path_taken": False,
        }

# Single Source of Truth for transitions
FSM_TRANSITIONS = {
    "detect": {
        "discover_booking",   # Search existing appointment
        "discover_slots",     # Search available slots
        "await_confirmation", # Direct confirmation
        "clarification",      # Ask for more details
        "handover",           # Escalate to human
    },
    "discover_booking": {
        "await_selection",    # Multiple found
        "discover_slots",     # Needs slots
        "await_confirmation", # Single found
        "clarification",      # Needs details
        "handover",
    },
    "discover_slots": {
        "await_selection",    # Multiple slots
        "await_confirmation", # Single slot
        "clarification",
        "handover",
    },
    "await_selection": {
        "await_confirmation", # User selected
        "discover_slots",     # New search
        "discover_booking",   # Change booking
        "handover",
    },
    "await_confirmation": {
        "completed",          # ✅ ERP executed
        "discover_slots",     # User changed mind
        "discover_booking",
        "handover",
    },
    # Terminal states
    "completed": {"completed", "handover"},
    "handover": {"handover"},
}

def is_valid_transition(from_state: str, to_state: str) -> bool:
    """Validate if transition is allowed."""
    return to_state in FSM_TRANSITIONS.get(from_state, set())

class ContextComposer:
    """
    Composes unified context from multiple sources.
    
    Sources:
    - Memory Engine (handover state, customer memory)
    - Tenant Config (identity, rules)
    - Conversation History
    
    Features:
    - Budget management (CONTEXT_BUDGET_TOKENS=1200)
    - Priority-based truncation
    - Block-based format (never cuts IDENTITY, RULES, INPUT)
    """
    
    def __init__(self):
        self.budget_tokens = int(os.getenv("CONTEXT_BUDGET_TOKENS", "1200"))
        self.memory_engine_url = os.getenv("MEMORY_ENGINE_URL")
    
    async def compose(
        self,
        tenant_id: str,
        conversation_id: str,
        user_id: str,
        message: str,
        tenant_config: dict,
        existing_history: list | None = None,
    ) -> ContextContractV1:
        """
        Compose full context with budget enforcement.
        
        Block Priority (never truncated):
        1. IDENTITY - who the assistant is
        2. RULES - tenant business rules  
        3. INPUT - user message
        
        Truncatable (in priority order):
        4. MEMORY - customer memory
        5. HANDOVER - handover context
        6. FOCUS - focused context (RAG results)
        """
        # Fetch parallel (asyncio.gather)
        memory_ctx, handover_ctx = await asyncio.gather(
            self._fetch_memory(tenant_id, user_id),
            self._fetch_handover(tenant_id, conversation_id),
        )
        
        # Build identity from tenant config
        identity = self._build_identity(tenant_config)
        
        # Apply budget with priority truncation
        return self._apply_budget(
            identity=identity,
            memory=memory_ctx,
            handover=handover_ctx,
            message=message,
        )

# ==============================================================================
# 🔧 NOTE ON TOOL BUDGET (Tech Lead Review)
# ==============================================================================
# This module (pricing_fast_lane) is NOT subject to tool_budget because:
# 1. It's an OPTIMIZED path that executes tool directly
# 2. It's DETERMINISTIC - doesn't enter iteration loop
# 3. tool_budget is for controlling agent ITERATIONS
# 4. If fast lane fails, terminates without handover
# ==============================================================================

async def pricing_fast_lane_node(state: dict, service: Any, config: Any):
    """
    Optimized path for price queries.
    
    Flow:
    1. Check cache (intelligent_cache_service)
    2. If miss: RAG search via docling_semantic_search_tool
    3. Format response directly (no LLM)
    4. Return with fast_lane_executed=True
    """
    # Intelligent cache with coalescing
    cache_key = build_intelligent_cache_key(
        tenant_id=config.tenant_id,
        query_normalized=normalize_query(user_message),
    )
    
    cached = await get_or_compute_with_coalescing(
        key=cache_key,
        compute_fn=lambda: _do_rag_search(state, service),
        ttl_seconds=300,
    )
    
    if cached:
        return {
            "pricing_response": cached,
            "fast_lane_executed": True,
            "cache_hit": True,
        }
    
    # RAG search (deterministic)
    results = await docling_semantic_search(
        query=user_message,
        tenant_id=config.tenant_id,
        k=5,
        threshold=0.7,  # Adaptive threshold
    )
    
    # Format response (template-based, no LLM)
    response = format_pricing_response(results)
    
    return {
        "pricing_response": response,
        "fast_lane_executed": True,
    }

# Whitelist of allowed sources
ALLOWED_UPDATERS = {
    "booking_flow_router",      # Main state machine router
    "discover_booking_node",    # Booking discovery (IO task)
    "discover_slots_node",      # Slot discovery (IO task)
    "confirm_booking",          # Confirm booking node
    "confirm_reschedule",       # Confirm reschedule node
    "confirm_cancel",           # Confirm cancel node
    "task_create_appointment",  # ERP mutation task
    "memory_engine_sync",       # Memory Engine persistence
    "workflow_hygiene",         # Workflow hygiene reset
}

class BookingFlowUpdate:
    """
    Validated booking_flow update with all constraints checked.
    
    Constraints:
    1. Whitelist validation (source must be in ALLOWED_UPDATERS)
    2. Monotonicity check (turn_seq >= current)
    3. Immutability guards (stage transitions must be valid)
    4. Automatic normalization (version++, updated_at)
    """
    
    def __init__(self, update: dict, current: dict, source: str):
        # Whitelist check
        if source not in ALLOWED_UPDATERS:
            booking_flow_rejections_total.labels(reason="unauthorized_source").inc()
            raise BookingFlowUpdateError(
                f"Source '{source}' not authorized to update booking_flow"
            )
        
        # Monotonicity check
        new_seq = update.get("turn_seq", 0)
        current_seq = current.get("turn_seq", 0)
        if new_seq < current_seq:
            booking_flow_rejections_total.labels(reason="out_of_order").inc()
            raise BookingFlowUpdateError(
                f"Out-of-order update: {new_seq} < {current_seq}"
            )
        
        # Transition validation
        new_stage = update.get("stage")
        current_stage = current.get("stage", "detect")
        if new_stage and not is_valid_transition(current_stage, new_stage):
            booking_flow_rejections_total.labels(reason="invalid_transition").inc()
            raise BookingFlowUpdateError(
                f"Invalid transition: {current_stage} → {new_stage}"
            )

# SOTA FIX (December 2025): Transient vs Permanent Exceptions

TRANSIENT_EXCEPTIONS = (
    httpx.TimeoutException,
    httpx.ConnectError,
    httpx.ReadTimeout,
    openai.APIConnectionError,
    openai.RateLimitError,       # Retriable with backoff
    openai.APITimeoutError,
    ConnectionError,
    TimeoutError,
    asyncio.TimeoutError,
)

# Permanent exceptions should NOT be retried
PERMANENT_EXCEPTIONS = (
    openai.AuthenticationError,   # Invalid API key
    openai.BadRequestError,       # Malformed prompt
    openai.PermissionDeniedError, # No permission
    ValueError,
    TypeError,
    KeyError,
)

# RetryPolicy uses whitelist (retry_on), not blacklist
retry_policy = RetryPolicy(
    max_attempts=3,
    initial_interval=0.5,         # 500ms
    backoff_factor=2.0,           # 500ms → 1s → 2s
    retry_on=TRANSIENT_EXCEPTIONS,
)