runtime_context.py

thread_context = {}


def set_thread_id(thread_id: str):
    thread_context["thread_id"] = thread_id


def get_thread_id():
    return thread_context.get("thread_id", "unknown")


AI_ENABLED_CACHE = True


def set_ai_enabled(value: bool):
    global AI_ENABLED_CACHE
    AI_ENABLED_CACHE = bool(value)


def is_ai_enabled():
    return AI_ENABLED_CACHE


CUSTOMER_AI_ENABLED_CACHE: dict[str, bool] = {}


def normalize_phone(phone: str | None) -> str:
    return (phone or "").strip()


def set_customer_ai_enabled(phone: str | None, value: bool):
    normalized_phone = normalize_phone(phone)

    if not normalized_phone:
        return

    CUSTOMER_AI_ENABLED_CACHE[normalized_phone] = bool(value)


def set_customer_ai_enabled_for_phone(phone: str | None, value: bool):
    set_customer_ai_enabled(phone, value)


def is_customer_ai_enabled(phone: str | None) -> bool:
    normalized_phone = normalize_phone(phone)

    if not normalized_phone:
        return True

    return CUSTOMER_AI_ENABLED_CACHE.get(normalized_phone, True)


def get_customer_ai_enabled(phone: str | None) -> bool:
    return is_customer_ai_enabled(phone)


def remove_customer_ai_enabled(phone: str | None):
    normalized_phone = normalize_phone(phone)

    if normalized_phone:
        CUSTOMER_AI_ENABLED_CACHE.pop(normalized_phone, None)


def load_customer_ai_enabled_cache(rows: list[dict]):
    CUSTOMER_AI_ENABLED_CACHE.clear()

    for row in rows or []:
        phone = normalize_phone(row.get("phone") or row.get("thread_id"))
        if not phone:
            continue

        CUSTOMER_AI_ENABLED_CACHE[phone] = bool(row.get("ai_enabled", True))


def get_customer_ai_enabled_cache():
    return CUSTOMER_AI_ENABLED_CACHE.copy()

SYSTEM_PROMPT_VARIABLE_CACHE = ""


def set_system_prompt_variable(value: str | None):
    global SYSTEM_PROMPT_VARIABLE_CACHE
    SYSTEM_PROMPT_VARIABLE_CACHE = (value or "").strip()


def get_system_prompt_variable() -> str:
    return SYSTEM_PROMPT_VARIABLE_CACHE