mem0ai--mem0/mem0/memory/notices.py

import asyncio
import json
import re
import sys
import threading
from datetime import date, datetime, timedelta, timezone
from typing import Any, Dict, Optional, Tuple

from mem0.memory import telemetry as telemetry_module
from mem0.memory.setup import _load_config, _write_config


FLAG_KEY = "mem0-oss-notices"
NOTICE_ID = "first_run"
TEMPORAL_FEATURE_NOTICE_ID = "temporal_stub"
TEMPORAL_USAGE_NOTICE_ID = "temporal_usage"
DECAY_FEATURE_NOTICE_ID = "decay_stub"
DECAY_USAGE_NOTICE_ID = "decay_usage"
SCALE_THRESHOLD_NOTICE_ID = "scale_threshold"
PERFORMANCE_SLOW_QUERY_NOTICE_ID = "performance_slow_query"
NOTICE_EVENT = "mem0.notice_displayed"
DISPLAYED_VARIANT = "displayed"
HOLDOUT_VARIANT = "holdout"
STATE_SECTION = "notice_state"
STATE_KEY = "first_run"
TEMPORAL_USAGE_STATE_KEY = "temporal_usage"
TEMPORAL_USAGE_CAP = 10
TEMPORAL_USAGE_WINDOW = timedelta(days=7)
DECAY_USAGE_STATE_KEY = "decay_usage"
DECAY_USAGE_CAP = 10
DECAY_USAGE_WINDOW = timedelta(days=7)
DECAY_USAGE_DELETE_THRESHOLD = 5
SCALE_THRESHOLD_STATE_KEY = "scale_threshold"
SCALE_THRESHOLD_CAP = 10
SCALE_THRESHOLD_WINDOW = timedelta(days=7)
SCALE_MEMORY_COUNT_THRESHOLD = 2000
SCALE_MEMORY_COUNT_CHECK_INTERVAL = 100
SCALE_TOP_K_THRESHOLD = 50
PERFORMANCE_SLOW_QUERY_STATE_KEY = "performance_slow_query"
PERFORMANCE_SLOW_QUERY_CAP = 10
PERFORMANCE_SLOW_QUERY_WINDOW = timedelta(days=7)
PERFORMANCE_SLOW_QUERY_THRESHOLD_SECONDS = 2.0
FEATURE_ERROR_CAP = 10
FEATURE_ERROR_WINDOW = timedelta(days=7)
TEMPORAL_FEATURE_ERROR_MESSAGES = {
    "timestamp": "The timestamp parameter is not supported by the OSS Memory SDK.",
    "reference_date": "The reference_date parameter is not supported by the OSS Memory SDK.",
}
DECAY_FEATURE_ERROR_MESSAGE = "The decay parameter is not supported by the OSS Memory SDK."

_ISO_DATE_RE = re.compile(
    r"\b\d{4}-\d{2}-\d{2}(?:[T\s]\d{2}:\d{2}(?::\d{2})?(?:\.\d+)?(?:Z|[+-]\d{2}:?\d{2})?)?\b"
)
_RELATIVE_TIME_RE = re.compile(
    r"\b("
    r"today|yesterday|tomorrow|"
    r"last\s+(?:night|week|month|year)|"
    r"this\s+(?:week|month|year)|"
    r"next\s+(?:week|month|year)|"
    r"(?:past|last)\s+\d+\s+(?:day|days|week|weeks|month|months|year|years)|"
    r"(?:since|before|after|until)\s+(?:today|yesterday|tomorrow|\d{4}-\d{2}-\d{2}|last\s+(?:week|month|year))"
    r")\b",
    re.IGNORECASE,
)
_RANGE_OPERATORS = {"gt", "gte", "lt", "lte"}

_state_lock = threading.Lock()
_first_run_claimed_in_process = False
_decay_usage_successful_delete_count_in_process = 0
_temporal_usage_capacity_reached_in_process = False
_decay_usage_capacity_reached_in_process = False
_scale_threshold_capacity_reached_in_process = False
_performance_slow_query_capacity_reached_in_process = False
_feature_error_capacity_reached_in_process = set()
_scale_memory_count_adds_since_check = 0
_scale_memory_count_checked_in_process = False
_scale_memory_count_threshold_evaluated_in_process = False


def display_first_run_notice(memory_instance, sync_type: str, trigger_function: str) -> None:
    """Best-effort first-run notice check. Never raises or writes unless displayed."""
    if not telemetry_module.MEM0_TELEMETRY:
        return

    if not _claim_first_run_notice(trigger_function):
        return

    variant = None
    try:
        telemetry = telemetry_module._get_oss_telemetry()
        if telemetry is None or telemetry.posthog is None or not telemetry.user_id:
            return

        flags = telemetry.posthog.evaluate_flags(telemetry.user_id, flag_keys=[FLAG_KEY])
        variant = flags.get_flag(FLAG_KEY)
        _update_first_run_variant(variant)

        if variant in (None, False):
            return

        payload = _coerce_mapping(flags.get_flag_payload(FLAG_KEY))
        notices = payload.get("notices", {})
        notice_config = _coerce_mapping(notices.get(NOTICE_ID) if isinstance(notices, dict) else {})
        notice_config_found = bool(notice_config)

        copy = notice_config.get("copy")
        enabled = notice_config.get("enabled", True) if notice_config_found else False
        notice_type = notice_config.get("notice_type", "log_line")

        disabled_reason = None
        bypass_reason = None
        if not notice_config_found:
            bypass_reason = "missing_notice_config"
        elif not enabled:
            disabled_reason = "payload_disabled"
            bypass_reason = disabled_reason
        elif not copy:
            bypass_reason = "missing_copy"
        elif variant != DISPLAYED_VARIANT:
            bypass_reason = "holdout" if variant == HOLDOUT_VARIANT else "not_displayed"

        displayed = variant == DISPLAYED_VARIANT and enabled and bool(copy)

        telemetry.capture_event(
            NOTICE_EVENT,
            {
                "notice_id": NOTICE_ID,
                "notice_type": notice_type,
                "flag_key": FLAG_KEY,
                "variant": variant,
                "displayed": displayed,
                "payload": copy,
                "bypass_reason": bypass_reason,
                "disabled_reason": disabled_reason,
                "notice_config_found": notice_config_found,
                "sync_type": sync_type,
                "trigger_function": trigger_function,
            },
            flags=flags,
        )

        if displayed:
            print(copy, file=sys.stderr)
    except Exception:
        if variant is not None:
            _update_first_run_variant(variant)


async def display_first_run_notice_async(memory_instance, sync_type: str, trigger_function: str) -> None:
    if not telemetry_module.MEM0_TELEMETRY or _first_run_claimed_in_process:
        return
    await asyncio.to_thread(display_first_run_notice, memory_instance, sync_type, trigger_function)


def display_temporal_usage_notice(
    memory_instance,
    sync_type: str,
    trigger_function: str,
    trigger_source: str,
    trigger_reason: str,
) -> None:
    """Best-effort temporal usage notice. Never raises or writes unless displayed."""
    if not telemetry_module.MEM0_TELEMETRY:
        return

    if _temporal_usage_at_capacity():
        return

    try:
        telemetry = telemetry_module._get_oss_telemetry()
        if telemetry is None or telemetry.posthog is None or not telemetry.user_id:
            return

        flags = telemetry.posthog.evaluate_flags(telemetry.user_id, flag_keys=[FLAG_KEY])
        variant = flags.get_flag(FLAG_KEY)
        if variant in (None, False):
            return

        payload = _coerce_mapping(flags.get_flag_payload(FLAG_KEY))
        notices = payload.get("notices", {})
        notice_config = _coerce_mapping(
            notices.get(TEMPORAL_USAGE_NOTICE_ID) if isinstance(notices, dict) else {}
        )
        notice_config_found = bool(notice_config)

        copy = notice_config.get("copy")
        enabled = notice_config.get("enabled", True) if notice_config_found else False
        notice_type = notice_config.get("notice_type", "log_line")

        disabled_reason = None
        bypass_reason = None
        if not notice_config_found:
            bypass_reason = "missing_notice_config"
        elif not enabled:
            disabled_reason = "payload_disabled"
            bypass_reason = disabled_reason
        elif not copy:
            bypass_reason = "missing_copy"
        elif variant != DISPLAYED_VARIANT:
            bypass_reason = "holdout" if variant == HOLDOUT_VARIANT else "not_displayed"

        displayed = variant == DISPLAYED_VARIANT and enabled and bool(copy)

        if not _record_temporal_usage_opportunity(
            variant=variant,
            sync_type=sync_type,
            trigger_function=trigger_function,
            trigger_source=trigger_source,
            trigger_reason=trigger_reason,
        ):
            return

        telemetry.capture_event(
            NOTICE_EVENT,
            {
                "notice_id": TEMPORAL_USAGE_NOTICE_ID,
                "notice_type": notice_type,
                "flag_key": FLAG_KEY,
                "variant": variant,
                "displayed": displayed,
                "payload": copy,
                "bypass_reason": bypass_reason,
                "disabled_reason": disabled_reason,
                "notice_config_found": notice_config_found,
                "sync_type": sync_type,
                "trigger_function": trigger_function,
                "trigger_source": trigger_source,
                "trigger_reason": trigger_reason,
            },
            flags=flags,
        )

        if displayed:
            print(copy, file=sys.stderr)
    except Exception:
        return


async def display_temporal_usage_notice_async(
    memory_instance,
    sync_type: str,
    trigger_function: str,
    trigger_source: str,
    trigger_reason: str,
) -> None:
    await asyncio.to_thread(
        display_temporal_usage_notice,
        memory_instance,
        sync_type,
        trigger_function,
        trigger_source,
        trigger_reason,
    )


def detect_decay_usage_from_delete() -> Optional[Tuple[str, str, Optional[int], Optional[int]]]:
    if not telemetry_module.MEM0_TELEMETRY:
        return None

    global _decay_usage_successful_delete_count_in_process
    try:
        with _state_lock:
            if _decay_usage_capacity_reached_in_process:
                return None
            _decay_usage_successful_delete_count_in_process += 1
            delete_count = _decay_usage_successful_delete_count_in_process

        if delete_count >= DECAY_USAGE_DELETE_THRESHOLD and not _decay_usage_at_capacity():
            return ("delete_count", "repeated_deletes", delete_count, None)
    except Exception:
        return None

    return None


def detect_decay_usage_from_delete_all(deleted_count: Any) -> Optional[Tuple[str, str, Optional[int], Optional[int]]]:
    if not telemetry_module.MEM0_TELEMETRY:
        return None

    deleted_count_value = _coerce_nonnegative_int(deleted_count, 0)
    if deleted_count_value <= 0:
        return None

    return ("delete_all", "bulk_delete", None, deleted_count_value)


def display_decay_usage_notice(
    memory_instance,
    sync_type: str,
    trigger_function: str,
    trigger_source: str,
    trigger_reason: str,
    delete_count: Optional[int] = None,
    deleted_count: Optional[int] = None,
) -> None:
    """Best-effort decay usage notice. Never raises or writes unless displayed."""
    if not telemetry_module.MEM0_TELEMETRY:
        return

    if _decay_usage_at_capacity():
        return

    try:
        telemetry = telemetry_module._get_oss_telemetry()
        if telemetry is None or telemetry.posthog is None or not telemetry.user_id:
            return

        flags = telemetry.posthog.evaluate_flags(telemetry.user_id, flag_keys=[FLAG_KEY])
        variant = flags.get_flag(FLAG_KEY)
        if variant in (None, False):
            return

        payload = _coerce_mapping(flags.get_flag_payload(FLAG_KEY))
        notices = payload.get("notices", {})
        notice_config = _coerce_mapping(
            notices.get(DECAY_USAGE_NOTICE_ID) if isinstance(notices, dict) else {}
        )
        notice_config_found = bool(notice_config)

        copy = notice_config.get("copy")
        enabled = notice_config.get("enabled", True) if notice_config_found else False
        notice_type = notice_config.get("notice_type", "log_line")

        disabled_reason = None
        bypass_reason = None
        if not notice_config_found:
            bypass_reason = "missing_notice_config"
        elif not enabled:
            disabled_reason = "payload_disabled"
            bypass_reason = disabled_reason
        elif not copy:
            bypass_reason = "missing_copy"
        elif variant != DISPLAYED_VARIANT:
            bypass_reason = "holdout" if variant == HOLDOUT_VARIANT else "not_displayed"

        displayed = variant == DISPLAYED_VARIANT and enabled and bool(copy)

        if not _record_decay_usage_opportunity(
            variant=variant,
            sync_type=sync_type,
            trigger_function=trigger_function,
            trigger_source=trigger_source,
            trigger_reason=trigger_reason,
            delete_count=delete_count,
            deleted_count=deleted_count,
        ):
            return

        telemetry.capture_event(
            NOTICE_EVENT,
            {
                "notice_id": DECAY_USAGE_NOTICE_ID,
                "notice_type": notice_type,
                "flag_key": FLAG_KEY,
                "variant": variant,
                "displayed": displayed,
                "payload": copy,
                "bypass_reason": bypass_reason,
                "disabled_reason": disabled_reason,
                "notice_config_found": notice_config_found,
                "sync_type": sync_type,
                "trigger_function": trigger_function,
                "trigger_source": trigger_source,
                "trigger_reason": trigger_reason,
                "delete_count": delete_count,
                "deleted_count": deleted_count,
            },
            flags=flags,
        )

        if displayed:
            print(copy, file=sys.stderr)
    except Exception:
        return


async def display_decay_usage_notice_async(
    memory_instance,
    sync_type: str,
    trigger_function: str,
    trigger_source: str,
    trigger_reason: str,
    delete_count: Optional[int] = None,
    deleted_count: Optional[int] = None,
) -> None:
    await asyncio.to_thread(
        display_decay_usage_notice,
        memory_instance,
        sync_type,
        trigger_function,
        trigger_source,
        trigger_reason,
        delete_count,
        deleted_count,
    )


def detect_scale_threshold_from_top_k(top_k: Any) -> Optional[Tuple[str, str, Optional[int], Optional[int], int]]:
    try:
        top_k_value = int(top_k)
    except (TypeError, ValueError):
        return None

    if top_k_value < SCALE_TOP_K_THRESHOLD:
        return None

    return ("top_k", "high_top_k", top_k_value, None, SCALE_TOP_K_THRESHOLD)


def detect_scale_threshold_from_add_result(
    memory_instance,
    add_result: Any,
) -> Optional[Tuple[str, str, Optional[int], Optional[int], int]]:
    if not telemetry_module.MEM0_TELEMETRY:
        return None

    added_count = _count_added_memories(add_result)
    if added_count == 0:
        return None

    global _scale_memory_count_adds_since_check
    global _scale_memory_count_checked_in_process
    global _scale_memory_count_threshold_evaluated_in_process
    try:
        with _state_lock:
            if _scale_memory_count_threshold_evaluated_in_process:
                return None

            _scale_memory_count_adds_since_check += added_count
            should_check = (
                not _scale_memory_count_checked_in_process
                or _scale_memory_count_adds_since_check >= SCALE_MEMORY_COUNT_CHECK_INTERVAL
            )
            if not should_check:
                return None

            _scale_memory_count_checked_in_process = True
            _scale_memory_count_adds_since_check = 0

            config = _load_config()
            scale_state = _get_notice_state(config, SCALE_THRESHOLD_STATE_KEY)
            if scale_state.get("memory_count_threshold_evaluated"):
                _scale_memory_count_threshold_evaluated_in_process = True
                return None
    except Exception:
        return None

    provider_count = _get_provider_memory_count(memory_instance)
    if provider_count is None or provider_count < SCALE_MEMORY_COUNT_THRESHOLD:
        return None

    if not _mark_scale_memory_count_threshold_evaluated():
        return None

    return (
        "memory_count",
        "memory_count_threshold",
        None,
        provider_count,
        SCALE_MEMORY_COUNT_THRESHOLD,
    )


def display_scale_threshold_notice(
    memory_instance,
    sync_type: str,
    trigger_function: str,
    trigger_source: str,
    trigger_reason: str,
    top_k: Optional[int] = None,
    memory_count: Optional[int] = None,
    threshold: Optional[int] = None,
) -> None:
    """Best-effort scale notice. Never raises or writes unless displayed."""
    if not telemetry_module.MEM0_TELEMETRY:
        return

    if _scale_threshold_at_capacity():
        return

    try:
        telemetry = telemetry_module._get_oss_telemetry()
        if telemetry is None or telemetry.posthog is None or not telemetry.user_id:
            return

        flags = telemetry.posthog.evaluate_flags(telemetry.user_id, flag_keys=[FLAG_KEY])
        variant = flags.get_flag(FLAG_KEY)
        if variant in (None, False):
            return

        payload = _coerce_mapping(flags.get_flag_payload(FLAG_KEY))
        notices = payload.get("notices", {})
        notice_config = _coerce_mapping(
            notices.get(SCALE_THRESHOLD_NOTICE_ID) if isinstance(notices, dict) else {}
        )
        notice_config_found = bool(notice_config)

        copies = _coerce_mapping(notice_config.get("copies")) if notice_config_found else {}
        copy_key = "memory_count" if trigger_source == "memory_count" else "top_k"
        copy = _render_scale_copy(copies.get(copy_key), top_k=top_k, memory_count=memory_count)
        enabled = notice_config.get("enabled", True) if notice_config_found else False
        notice_type = notice_config.get("notice_type", "log_line")

        disabled_reason = None
        bypass_reason = None
        if not notice_config_found:
            bypass_reason = "missing_notice_config"
        elif not enabled:
            disabled_reason = "payload_disabled"
            bypass_reason = disabled_reason
        elif not copy:
            bypass_reason = "missing_copy"
        elif variant != DISPLAYED_VARIANT:
            bypass_reason = "holdout" if variant == HOLDOUT_VARIANT else "not_displayed"

        displayed = variant == DISPLAYED_VARIANT and enabled and bool(copy)

        if not _record_scale_threshold_opportunity(
            variant=variant,
            sync_type=sync_type,
            trigger_function=trigger_function,
            trigger_source=trigger_source,
            trigger_reason=trigger_reason,
            top_k=top_k,
            memory_count=memory_count,
            threshold=threshold,
        ):
            return

        telemetry.capture_event(
            NOTICE_EVENT,
            {
                "notice_id": SCALE_THRESHOLD_NOTICE_ID,
                "notice_type": notice_type,
                "flag_key": FLAG_KEY,
                "variant": variant,
                "displayed": displayed,
                "payload": copy,
                "bypass_reason": bypass_reason,
                "disabled_reason": disabled_reason,
                "notice_config_found": notice_config_found,
                "sync_type": sync_type,
                "trigger_function": trigger_function,
                "trigger_source": trigger_source,
                "trigger_reason": trigger_reason,
                "top_k": top_k,
                "memory_count": memory_count,
                "threshold": threshold,
            },
            flags=flags,
        )

        if displayed:
            print(copy, file=sys.stderr)
    except Exception:
        return


async def display_scale_threshold_notice_async(
    memory_instance,
    sync_type: str,
    trigger_function: str,
    trigger_source: str,
    trigger_reason: str,
    top_k: Optional[int] = None,
    memory_count: Optional[int] = None,
    threshold: Optional[int] = None,
) -> None:
    await asyncio.to_thread(
        display_scale_threshold_notice,
        memory_instance,
        sync_type,
        trigger_function,
        trigger_source,
        trigger_reason,
        top_k,
        memory_count,
        threshold,
    )


def display_performance_slow_query_notice(
    memory_instance,
    sync_type: str,
    trigger_function: str,
    elapsed_seconds: float,
    top_k: int,
    result_count: int,
) -> None:
    """Best-effort slow-query notice. Never raises or writes unless displayed."""
    if not telemetry_module.MEM0_TELEMETRY:
        return

    if _performance_slow_query_at_capacity():
        return

    try:
        telemetry = telemetry_module._get_oss_telemetry()
        if telemetry is None or telemetry.posthog is None or not telemetry.user_id:
            return

        flags = telemetry.posthog.evaluate_flags(telemetry.user_id, flag_keys=[FLAG_KEY])
        variant = flags.get_flag(FLAG_KEY)
        if variant in (None, False):
            return

        payload = _coerce_mapping(flags.get_flag_payload(FLAG_KEY))
        notices = payload.get("notices", {})
        notice_config = _coerce_mapping(
            notices.get(PERFORMANCE_SLOW_QUERY_NOTICE_ID) if isinstance(notices, dict) else {}
        )
        notice_config_found = bool(notice_config)

        copy = notice_config.get("copy")
        enabled = notice_config.get("enabled", True) if notice_config_found else False
        notice_type = notice_config.get("notice_type", "log_line")

        disabled_reason = None
        bypass_reason = None
        if not notice_config_found:
            bypass_reason = "missing_notice_config"
        elif not enabled:
            disabled_reason = "payload_disabled"
            bypass_reason = disabled_reason
        elif not copy:
            bypass_reason = "missing_copy"
        elif variant != DISPLAYED_VARIANT:
            bypass_reason = "holdout" if variant == HOLDOUT_VARIANT else "not_displayed"

        displayed = variant == DISPLAYED_VARIANT and enabled and bool(copy)
        trigger_reason = "slow_query"

        if not _record_performance_slow_query_opportunity(
            variant=variant,
            sync_type=sync_type,
            trigger_function=trigger_function,
            trigger_reason=trigger_reason,
        ):
            return

        telemetry.capture_event(
            NOTICE_EVENT,
            {
                "notice_id": PERFORMANCE_SLOW_QUERY_NOTICE_ID,
                "notice_type": notice_type,
                "flag_key": FLAG_KEY,
                "variant": variant,
                "displayed": displayed,
                "payload": copy,
                "bypass_reason": bypass_reason,
                "disabled_reason": disabled_reason,
                "notice_config_found": notice_config_found,
                "sync_type": sync_type,
                "trigger_function": trigger_function,
                "trigger_reason": trigger_reason,
                "elapsed_ms": round(elapsed_seconds * 1000),
                "threshold_ms": round(PERFORMANCE_SLOW_QUERY_THRESHOLD_SECONDS * 1000),
                "top_k": top_k,
                "result_count": result_count,
            },
            flags=flags,
        )

        if displayed:
            print(copy, file=sys.stderr)
    except Exception:
        return


async def display_performance_slow_query_notice_async(
    memory_instance,
    sync_type: str,
    trigger_function: str,
    elapsed_seconds: float,
    top_k: int,
    result_count: int,
) -> None:
    await asyncio.to_thread(
        display_performance_slow_query_notice,
        memory_instance,
        sync_type,
        trigger_function,
        elapsed_seconds,
        top_k,
        result_count,
    )


def get_temporal_feature_error_message(sync_type: str, trigger_function: str, trigger_parameter: str) -> str:
    """Return the temporal feature error copy and capture event when available."""
    return _get_feature_error_message(
        TEMPORAL_FEATURE_NOTICE_ID,
        TEMPORAL_FEATURE_ERROR_MESSAGES[trigger_parameter],
        sync_type,
        trigger_function,
        trigger_parameter,
    )


async def get_temporal_feature_error_message_async(
    sync_type: str,
    trigger_function: str,
    trigger_parameter: str,
) -> str:
    return await asyncio.to_thread(
        get_temporal_feature_error_message,
        sync_type,
        trigger_function,
        trigger_parameter,
    )


def get_decay_feature_error_message(sync_type: str, trigger_function: str, trigger_parameter: str) -> str:
    """Return the decay feature error copy and capture event when available."""
    return _get_feature_error_message(
        DECAY_FEATURE_NOTICE_ID,
        DECAY_FEATURE_ERROR_MESSAGE,
        sync_type,
        trigger_function,
        trigger_parameter,
    )


async def get_decay_feature_error_message_async(
    sync_type: str,
    trigger_function: str,
    trigger_parameter: str,
) -> str:
    return await asyncio.to_thread(
        get_decay_feature_error_message,
        sync_type,
        trigger_function,
        trigger_parameter,
    )


def _get_feature_error_message(
    notice_id: str,
    plain_error: str,
    sync_type: str,
    trigger_function: str,
    trigger_parameter: str,
) -> str:
    if not telemetry_module.MEM0_TELEMETRY:
        return plain_error

    if _feature_error_at_capacity(notice_id):
        return plain_error

    try:
        telemetry = telemetry_module._get_oss_telemetry()
        if telemetry is None or telemetry.posthog is None or not telemetry.user_id:
            return plain_error

        flags = telemetry.posthog.evaluate_flags(telemetry.user_id, flag_keys=[FLAG_KEY])
        variant = flags.get_flag(FLAG_KEY)
        if variant in (None, False):
            return plain_error

        payload = _coerce_mapping(flags.get_flag_payload(FLAG_KEY))
        notices = payload.get("notices", {})
        notice_config = _coerce_mapping(
            notices.get(notice_id) if isinstance(notices, dict) else {}
        )
        notice_config_found = bool(notice_config)

        copy = notice_config.get("copy")
        enabled = notice_config.get("enabled", True) if notice_config_found else False
        notice_type = notice_config.get("notice_type", "error")

        disabled_reason = None
        bypass_reason = None
        if not notice_config_found:
            bypass_reason = "missing_notice_config"
        elif not enabled:
            disabled_reason = "payload_disabled"
            bypass_reason = disabled_reason
        elif not copy:
            bypass_reason = "missing_copy"
        elif variant not in (DISPLAYED_VARIANT, HOLDOUT_VARIANT):
            bypass_reason = "not_displayed"

        displayed = variant in (DISPLAYED_VARIANT, HOLDOUT_VARIANT) and enabled and bool(copy)

        if not _record_feature_error_opportunity(
            notice_id=notice_id,
            variant=variant,
            sync_type=sync_type,
            trigger_function=trigger_function,
            trigger_parameter=trigger_parameter,
        ):
            return plain_error

        telemetry.capture_event(
            NOTICE_EVENT,
            {
                "notice_id": notice_id,
                "notice_type": notice_type,
                "flag_key": FLAG_KEY,
                "variant": variant,
                "displayed": displayed,
                "payload": copy,
                "bypass_reason": bypass_reason,
                "disabled_reason": disabled_reason,
                "notice_config_found": notice_config_found,
                "sync_type": sync_type,
                "trigger_function": trigger_function,
                "trigger_parameter": trigger_parameter,
            },
            flags=flags,
        )

        if displayed:
            return copy
    except Exception:
        return plain_error

    return plain_error


def detect_temporal_usage_from_metadata(metadata: Optional[Dict[str, Any]]) -> Optional[Tuple[str, str]]:
    try:
        if not isinstance(metadata, dict):
            return None

        for key, value in _walk_mapping(metadata):
            temporal_key = _is_temporal_key(key)
            if temporal_key and _looks_temporal_value(value, allow_epoch=True):
                return ("metadata", "date_like_metadata")
    except Exception:
        return None
    return None


def detect_temporal_usage_from_search(
    query: Any,
    filters: Optional[Dict[str, Any]],
) -> Optional[Tuple[str, str]]:
    try:
        if isinstance(query, str):
            if _RELATIVE_TIME_RE.search(query):
                return ("query", "relative_phrase")
            if _ISO_DATE_RE.search(query):
                return ("query", "date_like_query")

        if _has_temporal_filter(filters):
            return ("filter", "date_range_filter")
    except Exception:
        return None
    return None


def _claim_first_run_notice(trigger_function: str) -> bool:
    global _first_run_claimed_in_process

    with _state_lock:
        if _first_run_claimed_in_process:
            return False

        config = _load_config()
        state = config.get(STATE_SECTION)
        if isinstance(state, dict):
            first_run = state.get(STATE_KEY)
            if isinstance(first_run, dict) and first_run.get("consumed"):
                _first_run_claimed_in_process = True
                return False

        if not isinstance(state, dict):
            state = {}

        state[STATE_KEY] = {
            "consumed": True,
            "consumed_at": datetime.now(timezone.utc).isoformat(),
            "trigger_function": trigger_function,
            "variant": None,
        }
        config[STATE_SECTION] = state
        _write_config(config)
        _first_run_claimed_in_process = True
        return True


def _update_first_run_variant(variant) -> None:
    try:
        with _state_lock:
            config = _load_config()
            state = config.get(STATE_SECTION)
            if not isinstance(state, dict):
                state = {}
            first_run = state.get(STATE_KEY)
            if not isinstance(first_run, dict):
                first_run = {"consumed": True}
            first_run["variant"] = variant
            state[STATE_KEY] = first_run
            config[STATE_SECTION] = state
            _write_config(config)
    except Exception:
        return


def _feature_error_at_capacity(notice_id: str) -> bool:
    if notice_id in _feature_error_capacity_reached_in_process:
        return True

    try:
        with _state_lock:
            config = _load_config()
            entries = _recent_feature_error_entries(config, notice_id, datetime.now(timezone.utc))
            at_capacity = len(entries) >= FEATURE_ERROR_CAP
            if at_capacity:
                _feature_error_capacity_reached_in_process.add(notice_id)
            return at_capacity
    except Exception:
        return True


def _record_feature_error_opportunity(
    *,
    notice_id: str,
    variant: str,
    sync_type: str,
    trigger_function: str,
    trigger_parameter: str,
) -> bool:
    try:
        with _state_lock:
            now = datetime.now(timezone.utc)
            config = _load_config()
            entries = _recent_feature_error_entries(config, notice_id, now)
            if len(entries) >= FEATURE_ERROR_CAP:
                _feature_error_capacity_reached_in_process.add(notice_id)
                return False

            entries.append(
                {
                    "evaluated_at": now.isoformat(),
                    "variant": variant,
                    "sync_type": sync_type,
                    "trigger_function": trigger_function,
                    "trigger_parameter": trigger_parameter,
                }
            )

            state = config.get(STATE_SECTION)
            if not isinstance(state, dict):
                state = {}
            feature_state = state.get(notice_id)
            if not isinstance(feature_state, dict):
                feature_state = {}
            feature_state["events"] = entries
            state[notice_id] = feature_state
            config[STATE_SECTION] = state
            _write_config(config)
            if len(entries) >= FEATURE_ERROR_CAP:
                _feature_error_capacity_reached_in_process.add(notice_id)
            return True
    except Exception:
        return False


def _recent_feature_error_entries(config: Dict[str, Any], notice_id: str, now: datetime):
    state = config.get(STATE_SECTION)
    if not isinstance(state, dict):
        return []

    feature_state = state.get(notice_id)
    if not isinstance(feature_state, dict):
        return []

    entries = feature_state.get("events")
    if not isinstance(entries, list):
        return []

    cutoff = now - FEATURE_ERROR_WINDOW
    recent = []
    for entry in entries:
        if not isinstance(entry, dict):
            continue
        evaluated_at = _parse_datetime(entry.get("evaluated_at"))
        if evaluated_at is not None and evaluated_at >= cutoff:
            recent.append(entry)
    return recent


def _temporal_usage_at_capacity() -> bool:
    global _temporal_usage_capacity_reached_in_process
    if _temporal_usage_capacity_reached_in_process:
        return True

    try:
        with _state_lock:
            config = _load_config()
            entries = _recent_temporal_usage_entries(config, datetime.now(timezone.utc))
            at_capacity = len(entries) >= TEMPORAL_USAGE_CAP
            if at_capacity:
                _temporal_usage_capacity_reached_in_process = True
            return at_capacity
    except Exception:
        return True


def _record_temporal_usage_opportunity(
    *,
    variant: str,
    sync_type: str,
    trigger_function: str,
    trigger_source: str,
    trigger_reason: str,
) -> bool:
    global _temporal_usage_capacity_reached_in_process
    try:
        with _state_lock:
            now = datetime.now(timezone.utc)
            config = _load_config()
            entries = _recent_temporal_usage_entries(config, now)
            if len(entries) >= TEMPORAL_USAGE_CAP:
                _temporal_usage_capacity_reached_in_process = True
                return False

            entries.append(
                {
                    "evaluated_at": now.isoformat(),
                    "variant": variant,
                    "sync_type": sync_type,
                    "trigger_function": trigger_function,
                    "trigger_source": trigger_source,
                    "trigger_reason": trigger_reason,
                }
            )

            state = config.get(STATE_SECTION)
            if not isinstance(state, dict):
                state = {}
            temporal_state = state.get(TEMPORAL_USAGE_STATE_KEY)
            if not isinstance(temporal_state, dict):
                temporal_state = {}
            temporal_state["events"] = entries
            state[TEMPORAL_USAGE_STATE_KEY] = temporal_state
            config[STATE_SECTION] = state
            _write_config(config)
            if len(entries) >= TEMPORAL_USAGE_CAP:
                _temporal_usage_capacity_reached_in_process = True
            return True
    except Exception:
        return False


def _recent_temporal_usage_entries(config: Dict[str, Any], now: datetime):
    state = config.get(STATE_SECTION)
    if not isinstance(state, dict):
        return []

    temporal_state = state.get(TEMPORAL_USAGE_STATE_KEY)
    if not isinstance(temporal_state, dict):
        return []

    entries = temporal_state.get("events")
    if not isinstance(entries, list):
        return []

    cutoff = now - TEMPORAL_USAGE_WINDOW
    recent = []
    for entry in entries:
        if not isinstance(entry, dict):
            continue
        evaluated_at = _parse_datetime(entry.get("evaluated_at"))
        if evaluated_at is not None and evaluated_at >= cutoff:
            recent.append(entry)
    return recent


def _decay_usage_at_capacity() -> bool:
    global _decay_usage_capacity_reached_in_process
    if _decay_usage_capacity_reached_in_process:
        return True

    try:
        with _state_lock:
            config = _load_config()
            entries = _recent_decay_usage_entries(config, datetime.now(timezone.utc))
            at_capacity = len(entries) >= DECAY_USAGE_CAP
            if at_capacity:
                _decay_usage_capacity_reached_in_process = True
            return at_capacity
    except Exception:
        return True


def _record_decay_usage_opportunity(
    *,
    variant: str,
    sync_type: str,
    trigger_function: str,
    trigger_source: str,
    trigger_reason: str,
    delete_count: Optional[int],
    deleted_count: Optional[int],
) -> bool:
    global _decay_usage_capacity_reached_in_process
    try:
        with _state_lock:
            now = datetime.now(timezone.utc)
            config = _load_config()
            entries = _recent_decay_usage_entries(config, now)
            if len(entries) >= DECAY_USAGE_CAP:
                _decay_usage_capacity_reached_in_process = True
                return False

            entry = {
                "evaluated_at": now.isoformat(),
                "variant": variant,
                "sync_type": sync_type,
                "trigger_function": trigger_function,
                "trigger_source": trigger_source,
                "trigger_reason": trigger_reason,
            }
            if delete_count is not None:
                entry["delete_count"] = delete_count
            if deleted_count is not None:
                entry["deleted_count"] = deleted_count
            entries.append(entry)

            state = config.get(STATE_SECTION)
            if not isinstance(state, dict):
                state = {}
            decay_state = state.get(DECAY_USAGE_STATE_KEY)
            if not isinstance(decay_state, dict):
                decay_state = {}
            decay_state["events"] = entries
            state[DECAY_USAGE_STATE_KEY] = decay_state
            config[STATE_SECTION] = state
            _write_config(config)
            if len(entries) >= DECAY_USAGE_CAP:
                _decay_usage_capacity_reached_in_process = True
            return True
    except Exception:
        return False


def _recent_decay_usage_entries(config: Dict[str, Any], now: datetime):
    state = config.get(STATE_SECTION)
    if not isinstance(state, dict):
        return []

    decay_state = state.get(DECAY_USAGE_STATE_KEY)
    if not isinstance(decay_state, dict):
        return []

    entries = decay_state.get("events")
    if not isinstance(entries, list):
        return []

    cutoff = now - DECAY_USAGE_WINDOW
    recent = []
    for entry in entries:
        if not isinstance(entry, dict):
            continue
        evaluated_at = _parse_datetime(entry.get("evaluated_at"))
        if evaluated_at is not None and evaluated_at >= cutoff:
            recent.append(entry)
    return recent


def _scale_threshold_at_capacity() -> bool:
    global _scale_threshold_capacity_reached_in_process
    if _scale_threshold_capacity_reached_in_process:
        return True

    try:
        with _state_lock:
            config = _load_config()
            entries = _recent_scale_threshold_entries(config, datetime.now(timezone.utc))
            at_capacity = len(entries) >= SCALE_THRESHOLD_CAP
            if at_capacity:
                _scale_threshold_capacity_reached_in_process = True
            return at_capacity
    except Exception:
        return True


def _record_scale_threshold_opportunity(
    *,
    variant: str,
    sync_type: str,
    trigger_function: str,
    trigger_source: str,
    trigger_reason: str,
    top_k: Optional[int],
    memory_count: Optional[int],
    threshold: Optional[int],
) -> bool:
    global _scale_threshold_capacity_reached_in_process
    try:
        with _state_lock:
            now = datetime.now(timezone.utc)
            config = _load_config()
            entries = _recent_scale_threshold_entries(config, now)
            if len(entries) >= SCALE_THRESHOLD_CAP:
                _scale_threshold_capacity_reached_in_process = True
                return False

            entry = {
                "evaluated_at": now.isoformat(),
                "variant": variant,
                "sync_type": sync_type,
                "trigger_function": trigger_function,
                "trigger_source": trigger_source,
                "trigger_reason": trigger_reason,
            }
            if top_k is not None:
                entry["top_k"] = top_k
            if memory_count is not None:
                entry["memory_count"] = memory_count
            if threshold is not None:
                entry["threshold"] = threshold
            entries.append(entry)

            state = config.get(STATE_SECTION)
            if not isinstance(state, dict):
                state = {}
            scale_state = state.get(SCALE_THRESHOLD_STATE_KEY)
            if not isinstance(scale_state, dict):
                scale_state = {}
            scale_state["events"] = entries
            if trigger_source == "memory_count":
                scale_state["memory_count_threshold_evaluated"] = True
            state[SCALE_THRESHOLD_STATE_KEY] = scale_state
            config[STATE_SECTION] = state
            _write_config(config)
            if len(entries) >= SCALE_THRESHOLD_CAP:
                _scale_threshold_capacity_reached_in_process = True
            return True
    except Exception:
        return False


def _mark_scale_memory_count_threshold_evaluated() -> bool:
    global _scale_memory_count_threshold_evaluated_in_process
    try:
        with _state_lock:
            config = _load_config()
            state = config.get(STATE_SECTION)
            if not isinstance(state, dict):
                state = {}
            scale_state = state.get(SCALE_THRESHOLD_STATE_KEY)
            if not isinstance(scale_state, dict):
                scale_state = {}
            if scale_state.get("memory_count_threshold_evaluated"):
                _scale_memory_count_threshold_evaluated_in_process = True
                return False

            scale_state["memory_count_threshold_evaluated"] = True
            state[SCALE_THRESHOLD_STATE_KEY] = scale_state
            config[STATE_SECTION] = state
            _write_config(config)
            _scale_memory_count_threshold_evaluated_in_process = True
            return True
    except Exception:
        return False


def _recent_scale_threshold_entries(config: Dict[str, Any], now: datetime):
    state = config.get(STATE_SECTION)
    if not isinstance(state, dict):
        return []

    scale_state = state.get(SCALE_THRESHOLD_STATE_KEY)
    if not isinstance(scale_state, dict):
        return []

    entries = scale_state.get("events")
    if not isinstance(entries, list):
        return []

    cutoff = now - SCALE_THRESHOLD_WINDOW
    recent = []
    for entry in entries:
        if not isinstance(entry, dict):
            continue
        evaluated_at = _parse_datetime(entry.get("evaluated_at"))
        if evaluated_at is not None and evaluated_at >= cutoff:
            recent.append(entry)
    return recent


def _performance_slow_query_at_capacity() -> bool:
    global _performance_slow_query_capacity_reached_in_process
    if _performance_slow_query_capacity_reached_in_process:
        return True

    try:
        with _state_lock:
            config = _load_config()
            entries = _recent_performance_slow_query_entries(config, datetime.now(timezone.utc))
            at_capacity = len(entries) >= PERFORMANCE_SLOW_QUERY_CAP
            if at_capacity:
                _performance_slow_query_capacity_reached_in_process = True
            return at_capacity
    except Exception:
        return True


def _record_performance_slow_query_opportunity(
    *,
    variant: str,
    sync_type: str,
    trigger_function: str,
    trigger_reason: str,
) -> bool:
    global _performance_slow_query_capacity_reached_in_process
    try:
        with _state_lock:
            now = datetime.now(timezone.utc)
            config = _load_config()
            entries = _recent_performance_slow_query_entries(config, now)
            if len(entries) >= PERFORMANCE_SLOW_QUERY_CAP:
                _performance_slow_query_capacity_reached_in_process = True
                return False

            entries.append(
                {
                    "evaluated_at": now.isoformat(),
                    "variant": variant,
                    "sync_type": sync_type,
                    "trigger_function": trigger_function,
                    "trigger_reason": trigger_reason,
                }
            )

            state = config.get(STATE_SECTION)
            if not isinstance(state, dict):
                state = {}
            performance_state = state.get(PERFORMANCE_SLOW_QUERY_STATE_KEY)
            if not isinstance(performance_state, dict):
                performance_state = {}
            performance_state["events"] = entries
            state[PERFORMANCE_SLOW_QUERY_STATE_KEY] = performance_state
            config[STATE_SECTION] = state
            _write_config(config)
            if len(entries) >= PERFORMANCE_SLOW_QUERY_CAP:
                _performance_slow_query_capacity_reached_in_process = True
            return True
    except Exception:
        return False


def _recent_performance_slow_query_entries(config: Dict[str, Any], now: datetime):
    state = config.get(STATE_SECTION)
    if not isinstance(state, dict):
        return []

    performance_state = state.get(PERFORMANCE_SLOW_QUERY_STATE_KEY)
    if not isinstance(performance_state, dict):
        return []

    entries = performance_state.get("events")
    if not isinstance(entries, list):
        return []

    cutoff = now - PERFORMANCE_SLOW_QUERY_WINDOW
    recent = []
    for entry in entries:
        if not isinstance(entry, dict):
            continue
        evaluated_at = _parse_datetime(entry.get("evaluated_at"))
        if evaluated_at is not None and evaluated_at >= cutoff:
            recent.append(entry)
    return recent


def _parse_datetime(value: Any):
    if not isinstance(value, str):
        return None
    try:
        parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
    except ValueError:
        return None
    if parsed.tzinfo is None:
        parsed = parsed.replace(tzinfo=timezone.utc)
    return parsed.astimezone(timezone.utc)


def _get_notice_state(config: Dict[str, Any], state_key: str) -> Dict[str, Any]:
    state = config.get(STATE_SECTION)
    if not isinstance(state, dict):
        return {}

    notice_state = state.get(state_key)
    return notice_state if isinstance(notice_state, dict) else {}


def _count_added_memories(add_result: Any) -> int:
    results = add_result.get("results") if isinstance(add_result, dict) else add_result
    if not isinstance(results, list):
        return 0

    count = 0
    for item in results:
        if isinstance(item, dict) and item.get("event") == "ADD":
            count += 1
    return count


def _get_provider_memory_count(memory_instance) -> Optional[int]:
    vector_store = getattr(memory_instance, "vector_store", None)
    if vector_store is None:
        return None

    try:
        count = getattr(vector_store, "count", None)
        if callable(count):
            value = _coerce_nonnegative_int(count(), None)
            if value is not None:
                return value
    except Exception:
        pass

    try:
        col_info = getattr(vector_store, "col_info", None)
        if callable(col_info):
            collection_name = getattr(vector_store, "collection_name", None)
            if collection_name is None:
                schema = getattr(vector_store, "schema", None)
                if isinstance(schema, dict):
                    index = schema.get("index")
                    if isinstance(index, dict):
                        collection_name = index.get("name")
            if collection_name is not None:
                try:
                    info = col_info(collection_name)
                except TypeError:
                    info = col_info()
            else:
                info = col_info()
            value = _extract_count(info)
            if value is not None:
                return value

            client = getattr(vector_store, "client", None)
            client_count = (
                getattr(client, "count", None) if client is not None and collection_name is not None else None
            )
            if callable(client_count):
                return _extract_count(client_count(index=collection_name))
    except Exception:
        return None

    return None


def _extract_count(info: Any) -> Optional[int]:
    if info is None:
        return None

    if isinstance(info, dict):
        for key in ("count", "points_count", "vectors_count", "indexed_vectors_count", "num_docs"):
            value = _coerce_nonnegative_int(info.get(key), None)
            if value is not None:
                return value
        return None

    model_dump = getattr(info, "model_dump", None)
    if callable(model_dump):
        try:
            value = _extract_count(model_dump())
            if value is not None:
                return value
        except Exception:
            return None

    for attr in ("count", "points_count", "vectors_count", "indexed_vectors_count", "num_docs"):
        value = _coerce_nonnegative_int(getattr(info, attr, None), None)
        if value is not None:
            return value

    return None


def _render_scale_copy(copy_template: Any, *, top_k: Optional[int], memory_count: Optional[int]) -> Optional[str]:
    if not isinstance(copy_template, str) or not copy_template.strip():
        return None
    try:
        return copy_template.format(top_k=top_k, memory_count=memory_count)
    except Exception:
        return copy_template


def _coerce_nonnegative_int(value: Any, default: Optional[int]) -> Optional[int]:
    if isinstance(value, bool):
        return default
    try:
        parsed = int(value)
    except (TypeError, ValueError):
        return default
    return parsed if parsed >= 0 else default


def _coerce_mapping(payload: Any) -> Dict[str, Any]:
    if isinstance(payload, dict):
        return payload
    if isinstance(payload, str):
        try:
            parsed = json.loads(payload)
        except Exception:
            return {}
        return parsed if isinstance(parsed, dict) else {}
    return {}


def _walk_mapping(value: Any, parent_key: str = ""):
    if isinstance(value, dict):
        for key, child in value.items():
            key_text = str(key)
            yield key_text, child
            yield from _walk_mapping(child, key_text)
    elif isinstance(value, (list, tuple, set)):
        for child in value:
            yield from _walk_mapping(child, parent_key)


def _is_temporal_key(key: Any) -> bool:
    key_text = str(key).lower()
    exact_keys = {
        "date",
        "time",
        "timestamp",
        "datetime",
        "event_date",
        "reference_date",
        "created_at",
        "updated_at",
        "started_at",
        "ended_at",
        "expires_at",
    }
    return (
        key_text in exact_keys
        or key_text.endswith("_date")
        or key_text.endswith("_time")
        or key_text.endswith("_at")
        or "timestamp" in key_text
    )


def _looks_temporal_value(value: Any, allow_epoch: bool) -> bool:
    if isinstance(value, datetime):
        return True
    if isinstance(value, date):
        return True
    if isinstance(value, str):
        return bool(_ISO_DATE_RE.search(value) or _RELATIVE_TIME_RE.search(value))
    if allow_epoch and isinstance(value, (int, float)) and not isinstance(value, bool):
        return 946684800 <= value <= 4102444800 or 946684800000 <= value <= 4102444800000
    return False


def _has_temporal_filter(filters: Any) -> bool:
    if not isinstance(filters, dict):
        return False

    for key, value in filters.items():
        if key in {"AND", "OR", "NOT", "$and", "$or", "$not"}:
            if isinstance(value, list) and any(_has_temporal_filter(item) for item in value):
                return True
            if isinstance(value, dict) and _has_temporal_filter(value):
                return True
            continue

        temporal_key = _is_temporal_key(key)
        if isinstance(value, dict):
            range_values = [item for op, item in value.items() if op in _RANGE_OPERATORS]
            if range_values and (
                temporal_key
                or any(_looks_temporal_value(item, allow_epoch=temporal_key) for item in range_values)
            ):
                return True
            if _has_temporal_filter(value):
                return True
        elif temporal_key and _looks_temporal_value(value, allow_epoch=True):
            return True

    return False