Intelligent Caching — Scenarios and Runbooks

graph TD
    A[User submits negative feedback<br/>'This isn't what I asked for'] --> B{Feedback classifier}
    B -->|Cache-related| C[Check if response<br/>was served from cache]
    C -->|source = L2_SEMANTIC| D[Log: semantic_cache_false_positive]
    D --> E[CloudWatch Metric:<br/>cache.false_positive_count]
    E --> F{Rate > 0.5%<br/>of L2 hits?}
    F -->|Yes| G[CloudWatch Alarm:<br/>CacheFalsePositiveRate]
    G --> H[PagerDuty / SNS Alert]

    B -->|Not cache-related| I[Route to standard<br/>feedback pipeline]
    C -->|source = BEDROCK| J[Not a cache issue —<br/>model quality problem]

    style G fill:#e76f51,stroke:#f4a261,color:#fff

graph TD
    FP[False Positive<br/>Detected] --> Q1{Was the cached query<br/>actually different intent?}
    Q1 -->|Yes| RC1[Root Cause: Intent classifier<br/>assigned same intent to<br/>semantically different queries]
    Q1 -->|No, same intent| Q2{Is cosine similarity<br/>barely above threshold?}
    Q2 -->|Yes, sim = 0.920–0.930| RC2[Root Cause: Threshold too low<br/>for this intent category]
    Q2 -->|No, sim > 0.93| Q3{Are the queries about<br/>related but distinct titles?}
    Q3 -->|Yes| RC3[Root Cause: Franchise titles<br/>cluster too tightly in<br/>embedding space]
    Q3 -->|No| RC4[Root Cause: Embedding model<br/>lacks domain specificity<br/>for manga titles]

    RC1 --> FIX1[Add intent sub-classification<br/>dragon_ball vs dragon_ball_super]
    RC2 --> FIX2[Raise threshold for<br/>recommendation intent to 0.96]
    RC3 --> FIX3[Include title entity<br/>as hard filter in cache key]
    RC4 --> FIX4[Fine-tune embedding model<br/>on manga title pairs]

    style FP fill:#e76f51,stroke:#f4a261,color:#fff
    style RC3 fill:#264653,stroke:#2a9d8f,color:#fff

# Emergency: raise recommendation threshold to reject borderline matches
# Deploy via environment variable — no code deploy needed
# ECS task definition environment variable:
CACHE_THRESHOLD_RECOMMENDATION = "0.97"

import redis
import logging

logger = logging.getLogger(__name__)


def invalidate_recommendation_cache(
    redis_client: redis.Redis,
    index_name: str = "mangaassist_cache_idx",
) -> int:
    """
    Remove all recommendation-intent entries from the semantic cache.
    Called as emergency response to false positive spike.
    """
    cursor = 0
    removed = 0
    while True:
        cursor, keys = redis_client.scan(cursor, match="sc:*", count=500)
        for key in keys:
            intent = redis_client.hget(key, "intent")
            if intent and intent.decode() == "recommendation":
                redis_client.delete(key)
                removed += 1
        if cursor == 0:
            break
    logger.info("Invalidated %d recommendation cache entries", removed)
    return removed

def recommendation_cache_search_with_entity_filter(
    redis_client: redis.Redis,
    embedding: list[float],
    primary_entity: str,
    intent: str = "recommendation",
) -> dict | None:
    """
    Enhanced search that requires the primary manga title entity
    to match exactly, in addition to vector similarity.
    This prevents 'Dragon Ball' from matching 'Dragon Ball Super'.
    """
    import numpy as np

    vec_bytes = np.array(embedding, dtype=np.float32).tobytes()

    # Hard filter: intent must match AND query text must contain the exact title
    # Using TEXT search on the query field as a secondary filter
    filter_expr = f"@intent:{{{intent}}} @query:{primary_entity}"
    query = f"({filter_expr})=>[KNN 3 @vec $blob AS dist]"

    results = redis_client.execute_command(
        "FT.SEARCH", "mangaassist_cache_idx", query,
        "PARAMS", "2", "blob", vec_bytes,
        "SORTBY", "dist", "ASC",
        "LIMIT", "0", "1",
        "RETURN", "3", "response", "query", "dist",
        "DIALECT", "2",
    )

    if results[0] == 0:
        return None

    # Parse and apply threshold check
    # ... (standard threshold logic)
    return None  # Placeholder

graph TD
    A[Bedrock ThrottlingException<br/>rate > 100/min] --> ALARM1[CloudWatch Alarm:<br/>BedrockThrottleRate]
    B[Cache miss rate spikes<br/>to > 95% for 5 min] --> ALARM2[CloudWatch Alarm:<br/>CacheMissRateSpike]
    C[OpenSearch latency<br/>p99 > 5000ms] --> ALARM3[CloudWatch Alarm:<br/>RAGLatencyHigh]
    D[ECS task CPU > 90%<br/>for > 2 min] --> ALARM4[CloudWatch Alarm:<br/>ECSCPUHigh]

    ALARM1 --> COMP[CloudWatch Composite Alarm:<br/>CacheStampedeDetected]
    ALARM2 --> COMP
    ALARM3 --> COMP
    ALARM4 --> COMP

    COMP --> SNS[SNS → PagerDuty<br/>P1 Incident]
    COMP --> AUTO[Auto-Remediation Lambda]

    style COMP fill:#e76f51,stroke:#f4a261,color:#fff

graph TD
    STAMP[Cache Stampede<br/>Detected] --> Q1{Is this a<br/>new product/event?}
    Q1 -->|Yes| RC1[Root Cause: No cache warming<br/>for the new release.<br/>Cold cache + sudden demand spike.]
    Q1 -->|No| Q2{Did cache recently<br/>get invalidated?}
    Q2 -->|Yes| RC2[Root Cause: Mass invalidation<br/>flushed entries that were<br/>immediately re-requested.]
    Q2 -->|No| Q3{Did TTLs expire<br/>simultaneously?}
    Q3 -->|Yes| RC3[Root Cause: Synchronized TTL<br/>expiry. All entries for<br/>an intent expired at once.]
    Q3 -->|No| RC4[Root Cause: Infrastructure issue<br/>Redis restart or connectivity<br/>loss emptied the cache.]

    RC1 --> FIX1[Implement pre-release<br/>cache warming pipeline]
    RC2 --> FIX2[Add stampede lock<br/>single-flight pattern]
    RC3 --> FIX3[Add TTL jitter<br/>±10% randomization]
    RC4 --> FIX4[Redis cluster mode<br/>+ AOF persistence]

    style STAMP fill:#e76f51,stroke:#f4a261,color:#fff
    style RC1 fill:#264653,stroke:#2a9d8f,color:#fff

import asyncio
import hashlib
import logging
import time
from typing import Optional

logger = logging.getLogger(__name__)


class StampedeProtection:
    """
    Single-flight / request coalescing pattern.
    When N concurrent requests arrive for the same cache key,
    only 1 invokes Bedrock. The other N-1 wait for the result.

    Uses Redis distributed locks to coordinate across ECS containers.
    """

    LOCK_PREFIX = "lock:"
    LOCK_TTL = 30  # seconds — max time to wait for Bedrock response

    def __init__(self, redis_client, semantic_cache, bedrock_invoker):
        self.redis = redis_client
        self.cache = semantic_cache
        self.invoker = bedrock_invoker
        self._local_waiters: dict[str, asyncio.Event] = {}

    async def get_or_invoke(
        self,
        query: str,
        intent: str,
        entities: dict,
        rag_context: str,
        session_history: list,
    ) -> dict:
        """
        Try cache → if miss, acquire lock → invoke Bedrock → store → release.
        Concurrent requests for the same query wait on the lock.
        """
        # Step 1: Try cache
        cached = self.cache.lookup(query=query, intent=intent)
        if cached:
            return cached

        # Step 2: Compute coalescing key
        normalized = self.cache._normalize(query)
        coalesce_key = hashlib.md5(f"{normalized}|{intent}".encode()).hexdigest()
        lock_key = f"{self.LOCK_PREFIX}{coalesce_key}"

        # Step 3: Try to acquire distributed lock
        acquired = self.redis.set(lock_key, "1", nx=True, ex=self.LOCK_TTL)

        if acquired:
            # This container is the leader — invoke Bedrock
            try:
                response = self.invoker.invoke(
                    user_message=query,
                    rag_context=rag_context,
                    session_history=session_history,
                )
                # Store in cache for all waiters
                self.cache.store(
                    query=query,
                    response=response["response_text"],
                    intent=intent,
                    entities=entities,
                )
                return {
                    "response": response["response_text"],
                    "source": "BEDROCK_LEADER",
                    "similarity": 1.0,
                }
            finally:
                self.redis.delete(lock_key)
        else:
            # Another container is the leader — poll cache until result appears
            logger.info("Waiting for leader to populate cache for key=%s", coalesce_key)
            for _ in range(60):  # Wait up to 30 seconds (60 x 0.5s)
                await asyncio.sleep(0.5)
                cached = self.cache.lookup(query=query, intent=intent)
                if cached:
                    cached["source"] = "COALESCED_WAIT"
                    return cached

            # Timeout — fall through to direct invocation
            logger.warning("Coalescing timeout for key=%s, invoking directly", coalesce_key)
            response = self.invoker.invoke(
                user_message=query,
                rag_context=rag_context,
                session_history=session_history,
            )
            return {
                "response": response["response_text"],
                "source": "BEDROCK_TIMEOUT_FALLBACK",
                "similarity": 1.0,
            }

async def warm_new_release_cache(
    title: str,
    volume: str,
    cache_warmer,  # CacheWarmer instance
) -> dict:
    """
    Triggered by EventBridge when a new manga release is added to the catalog.
    Pre-populates cache 1 hour before the release goes live.
    """
    result = cache_warmer.warm_new_release(title=title, volume=volume)
    logger.info("Pre-warmed cache for %s vol %s: %s", title, volume, result)
    return result

import random

def ttl_with_jitter(base_ttl: int, jitter_pct: float = 0.10) -> int:
    """
    Add ±10% random jitter to TTL to prevent synchronized expiry.
    base_ttl=3600 → returns 3240–3960.
    """
    jitter = int(base_ttl * jitter_pct)
    return base_ttl + random.randint(-jitter, jitter)

graph TD
    A[Customer complaint:<br/>'Chatbot says 12800 but<br/>website says 8980'] --> B[Support ticket created]
    B --> C{Check cache source<br/>in response metadata}
    C -->|source = L2_SEMANTIC| D[Confirm stale cache]

    E[EventBridge event:<br/>price_changed for<br/>Demon Slayer] --> F{Was invalidation<br/>Lambda invoked?}
    F -->|No invocation in<br/>CloudWatch Logs| G[Confirm: EventBridge<br/>rule did not trigger]

    H[Automated price<br/>consistency check] --> I{Chatbot response price<br/>== catalog API price?}
    I -->|Mismatch| J[CloudWatch Metric:<br/>cache.price_consistency_error]
    J --> K[CloudWatch Alarm:<br/>StalePriceDetected]
    K --> L[PagerDuty P1 Alert]

    style K fill:#e76f51,stroke:#f4a261,color:#fff
    style G fill:#e76f51,stroke:#f4a261,color:#fff

graph TD
    STALE[Stale Price in<br/>Cached Response] --> Q1{Was price_changed<br/>event emitted?}
    Q1 -->|No| RC1[Root Cause: Catalog CMS<br/>did not emit event.<br/>Manual price update bypassed<br/>event pipeline.]
    Q1 -->|Yes| Q2{Did EventBridge<br/>rule match the event?}
    Q2 -->|No| RC2[Root Cause: EventBridge rule<br/>filter pattern mismatch.<br/>Event schema changed but<br/>rule was not updated.]
    Q2 -->|Yes| Q3{Did Lambda execute<br/>successfully?}
    Q3 -->|No| RC3[Root Cause: Lambda timeout<br/>or permission error.<br/>Could not connect to Redis<br/>or CloudFront.]
    Q3 -->|Yes| Q4{Were correct Redis<br/>keys deleted?}
    Q4 -->|No| RC4[Root Cause: Key scan pattern<br/>did not match. Entity extraction<br/>for 'Demon Slayer' returned<br/>'demon slayer' (case mismatch).]
    Q4 -->|Yes| RC5[Root Cause: Race condition.<br/>Cache entry re-populated<br/>before CloudFront invalidation<br/>completed.]

    RC2 --> FIX[Fix EventBridge rule filter<br/>to match current event schema]

    style STALE fill:#e76f51,stroke:#f4a261,color:#fff
    style RC2 fill:#264653,stroke:#2a9d8f,color:#fff

import redis
import boto3
import logging

logger = logging.getLogger(__name__)


def emergency_price_cache_flush(
    redis_url: str,
    title: str,
    cloudfront_distribution_id: str,
) -> dict:
    """
    Emergency flush of all cache entries for a specific product.
    Called manually by on-call engineer when price staleness is detected.
    """
    r = redis.Redis.from_url(redis_url, decode_responses=True)
    cf = boto3.client("cloudfront")

    # Flush Redis (L2)
    removed = 0
    cursor = 0
    while True:
        cursor, keys = r.scan(cursor, match="sc:*", count=500)
        for key in keys:
            query_text = r.hget(key, "query")
            entities_json = r.hget(key, "ents")
            if query_text and title.lower() in query_text.lower():
                r.delete(key)
                removed += 1
            elif entities_json and title.lower() in entities_json.lower():
                r.delete(key)
                removed += 1
        if cursor == 0:
            break

    # Flush CloudFront (L4)
    invalidation = cf.create_invalidation(
        DistributionId=cloudfront_distribution_id,
        InvalidationBatch={
            "Paths": {
                "Quantity": 1,
                "Items": [f"/api/products/*demon-slayer*"],
            },
            "CallerReference": f"emergency-{title}-{__import__('time').time()}",
        },
    )

    result = {
        "redis_entries_removed": removed,
        "cloudfront_invalidation_id": invalidation["Invalidation"]["Id"],
    }
    logger.info("Emergency cache flush for '%s': %s", title, result)
    return result

{
    "source": ["com.mangaassist.catalog"],
    "detail-type": ["product.price.changed"],
    "detail": {
        "change_type": ["price_update", "flash_sale_start", "flash_sale_end"]
    }
}

import asyncio
import logging
import random

logger = logging.getLogger(__name__)


class PriceConsistencyChecker:
    """
    Periodically samples cached product_info responses and compares
    the price in the response against the live catalog API.
    Runs as a background task in each ECS container.
    """

    def __init__(self, semantic_cache, catalog_api_client, check_interval: int = 60):
        self.cache = semantic_cache
        self.catalog = catalog_api_client
        self.interval = check_interval
        self._inconsistencies = 0

    async def run(self) -> None:
        """Run continuous price consistency checks."""
        while True:
            await asyncio.sleep(self.interval)
            try:
                await self._check_sample()
            except Exception as e:
                logger.error("Price consistency check failed: %s", e)

    async def _check_sample(self) -> None:
        """Sample 10 random product_info cache entries and verify prices."""
        # Scan for product_info entries
        cursor = 0
        candidates = []
        while len(candidates) < 50:
            cursor, keys = self.cache.redis.scan(cursor, match="sc:*", count=100)
            for key in keys:
                intent = self.cache.redis.hget(key, "intent")
                if intent and intent.decode() == "product_info":
                    candidates.append(key)
            if cursor == 0:
                break

        # Sample 10
        sample = random.sample(candidates, min(10, len(candidates)))
        for key in sample:
            response_text = self.cache.redis.hget(key, "response")
            entities_json = self.cache.redis.hget(key, "ents")
            if not response_text or not entities_json:
                continue

            response_text = response_text.decode()
            entities = __import__("json").loads(entities_json.decode())
            title = entities.get("title")
            if not title:
                continue

            # Get live price from catalog
            live_price = await self.catalog.get_price(title)
            if live_price is None:
                continue

            # Check if cached response contains the correct price
            if str(live_price) not in response_text:
                self._inconsistencies += 1
                logger.warning(
                    "Price inconsistency: title='%s', live_price=%s, "
                    "cached response does not contain live price. Key=%s",
                    title, live_price, key.decode(),
                )
                # Auto-invalidate the stale entry
                self.cache.redis.delete(key)
                logger.info("Auto-invalidated stale cache entry: %s", key.decode())

graph TD
    A[Bedrock usage metrics] --> B{cacheReadInputTokens<br/>dropped to 0?}
    B -->|Yes| C[CloudWatch Metric:<br/>bedrock.prompt_cache_miss_rate = 100%]
    C --> D[CloudWatch Alarm:<br/>PromptCacheMissSpike]
    D --> E{Correlate with<br/>deployment event}
    E -->|Deployment within<br/>last 10 min| F[Diagnosis: Prompt change<br/>invalidated Bedrock cache]
    E -->|No deployment| G[Diagnosis: Cache eviction<br/>due to 5-min inactivity<br/>or provider-side issue]

    style D fill:#e76f51,stroke:#f4a261,color:#fff

graph TD
    MISS[Prompt Cache Miss<br/>Rate = 100%] --> Q1{Was system prompt<br/>text changed?}
    Q1 -->|Yes| RC1[Root Cause: Any change to<br/>cached prefix invalidates<br/>the Bedrock prompt cache.<br/>Even a single character.]
    Q1 -->|No| Q2{Was model ID<br/>changed?}
    Q2 -->|Yes| RC2[Root Cause: Different model =<br/>different cache namespace.<br/>Cache is per-model.]
    Q2 -->|No| Q3{Was there a gap<br/>> 5 min with no traffic?}
    Q3 -->|Yes| RC3[Root Cause: Bedrock prompt<br/>cache evicts after 5 min<br/>of inactivity.]
    Q3 -->|No| RC4[Root Cause: Provider-side<br/>cache eviction. Not<br/>controllable by customer.]

    RC1 --> FIX1[Schedule prompt changes<br/>during low-traffic windows.<br/>Monitor TTFT recovery.]

    style MISS fill:#e76f51,stroke:#f4a261,color:#fff
    style RC1 fill:#264653,stroke:#2a9d8f,color:#fff

import boto3
from datetime import datetime, timedelta


def check_prompt_cache_status(region: str = "us-east-1") -> dict:
    """
    Query recent Bedrock invocations to check prompt cache behavior.
    Compares cache_read vs cache_write token counts.
    """
    cw = boto3.client("cloudwatch", region_name=region)

    # Check custom metric for prompt cache hits
    response = cw.get_metric_statistics(
        Namespace="MangaAssist/Cache",
        MetricName="PromptCacheReadTokens",
        StartTime=datetime.utcnow() - timedelta(minutes=30),
        EndTime=datetime.utcnow(),
        Period=60,
        Statistics=["Sum", "SampleCount"],
    )

    datapoints = sorted(response["Datapoints"], key=lambda x: x["Timestamp"])
    if not datapoints:
        return {"status": "NO_DATA"}

    recent = datapoints[-1]
    earlier = datapoints[0] if len(datapoints) > 1 else None

    return {
        "current_cache_read_tokens": recent["Sum"],
        "current_sample_count": recent["SampleCount"],
        "earlier_cache_read_tokens": earlier["Sum"] if earlier else None,
        "cache_active": recent["Sum"] > 0,
        "diagnosis": (
            "Prompt cache is active"
            if recent["Sum"] > 0
            else "Prompt cache is NOT active — likely invalidated by prompt change"
        ),
    }

# deployment_config.yaml — schedule prompt changes during low-traffic
prompt_deployment:
    preferred_window: "03:00-05:00 JST"  # 3-5 AM JST — lowest traffic
    pre_warm_requests: 5  # Send 5 dummy requests to warm the cache
    monitoring:
        watch_metric: "bedrock.prompt_cache_hit_rate"
        recovery_threshold: 0.90  # Alert if not recovered within 5 min
        alert_channel: "#mangaassist-ops"

graph TD
    A[CloudWatch Metric:<br/>ElastiCache BytesUsedForCache] --> B{> 85% of<br/>maxmemory?}
    B -->|Yes| C[CloudWatch Alarm:<br/>RedisCacheMemoryHigh]
    C --> D[SNS → Ops Channel]

    E[Application Metric:<br/>cache.hit_rate] --> F{Dropped > 5% from<br/>7-day moving average?}
    F -->|Yes| G[CloudWatch Alarm:<br/>CacheHitRateDegradation]
    G --> D

    H[ElastiCache Metric:<br/>Evictions] --> I{Evictions > 0?}
    I -->|Yes| J[CloudWatch Alarm:<br/>RedisCacheEvictions]
    J --> D

    D --> K[Composite Alarm:<br/>CacheMemoryExhaustion]
    K --> L[PagerDuty P2 Alert]

    style K fill:#e76f51,stroke:#f4a261,color:#fff

graph TD
    MEM[Redis Memory<br/>Exhaustion] --> Q1{Are TTLs being<br/>set on all entries?}
    Q1 -->|No| RC1[Root Cause: Some code paths<br/>store entries without TTL.<br/>These entries never expire.]
    Q1 -->|Yes| Q2{Is a single intent<br/>consuming > 40% memory?}
    Q2 -->|Yes| RC2[Root Cause: Low hit-rate intents<br/>consuming disproportionate memory.<br/>Recommendations: 40% memory,<br/>8% hit rate.]
    Q2 -->|No| Q3{Are entries being<br/>re-created after TTL expiry?}
    Q3 -->|Yes| RC3[Root Cause: Cache churn —<br/>entries expire, get re-created,<br/>and accumulate unique keys<br/>due to timestamp in key.]
    Q3 -->|No| RC4[Root Cause: Traffic growth<br/>exceeded original memory<br/>sizing. Need to scale<br/>the cluster.]

    RC2 --> FIX[Implement memory budgets<br/>per intent category +<br/>reduce recommendation TTL]

    style MEM fill:#e76f51,stroke:#f4a261,color:#fff
    style RC2 fill:#264653,stroke:#2a9d8f,color:#fff

import redis
import json
import logging

logger = logging.getLogger(__name__)


def emergency_memory_cleanup(
    redis_url: str,
    low_value_intents: list[str] = None,
    max_entries_to_remove: int = 50_000,
) -> dict:
    """
    Emergency memory cleanup: remove low-value cache entries.
    Targets entries with low hit counts and low-value intents.
    """
    if low_value_intents is None:
        low_value_intents = ["greeting", "recommendation"]

    r = redis.Redis.from_url(redis_url, decode_responses=True)

    removed = 0
    cursor = 0
    while removed < max_entries_to_remove:
        cursor, keys = r.scan(cursor, match="sc:*", count=500)
        for key in keys:
            intent = r.hget(key, "intent")
            hits = r.hget(key, "hits")

            should_remove = False
            # Remove low-value intents
            if intent in low_value_intents:
                should_remove = True
            # Remove entries with 0 hits that are older than 1 hour
            elif hits and int(hits) == 0:
                ts = r.hget(key, "ts")
                if ts and ((__import__("time").time() - float(ts)) > 3600):
                    should_remove = True

            if should_remove:
                r.delete(key)
                removed += 1

        if cursor == 0:
            break

    result = {
        "entries_removed": removed,
        "memory_before": r.info("memory")["used_memory_human"],
    }
    logger.info("Emergency cleanup: %s", result)
    return result

import logging
import time

logger = logging.getLogger(__name__)

# Memory budget: percentage of total Redis memory per intent category
INTENT_MEMORY_BUDGET = {
    "faq": 0.25,               # 25% — high hit rate, high value
    "product_info": 0.25,      # 25% — high hit rate, business-critical
    "shipping_info": 0.15,     # 15% — moderate volume, stable answers
    "manga_release_date": 0.15,# 15% — moderate volume, time-sensitive
    "recommendation": 0.10,    # 10% — low hit rate, reduce from 40%
    "greeting": 0.02,          # 2% — minimal variation
    "manga_search": 0.05,      # 5% — moderate hit rate
    "_other": 0.03,            # 3% — catch-all
}


class MemoryBudgetEnforcer:
    """
    Enforces per-intent memory budgets in the Redis cache.
    When an intent exceeds its budget, the least-recently-hit
    entries for that intent are evicted.
    """

    def __init__(self, redis_client, max_memory_bytes: int):
        self.redis = redis_client
        self.max_memory = max_memory_bytes

    def enforce(self) -> dict:
        """
        Scan all cache entries, compute per-intent memory usage,
        and evict entries from over-budget intents.
        """
        # Phase 1: Inventory
        intent_entries: dict[str, list[dict]] = {}
        cursor = 0
        while True:
            cursor, keys = self.redis.scan(cursor, match="sc:*", count=500)
            for key in keys:
                intent = self.redis.hget(key, "intent")
                hits = self.redis.hget(key, "hits")
                ts = self.redis.hget(key, "ts")
                mem = self.redis.memory_usage(key) or 0

                intent_str = intent.decode() if intent else "_other"
                if intent_str not in intent_entries:
                    intent_entries[intent_str] = []
                intent_entries[intent_str].append({
                    "key": key,
                    "hits": int(hits) if hits else 0,
                    "ts": float(ts) if ts else 0,
                    "memory": mem,
                })
            if cursor == 0:
                break

        # Phase 2: Check budgets and evict
        eviction_report = {}
        for intent, entries in intent_entries.items():
            budget_pct = INTENT_MEMORY_BUDGET.get(intent, INTENT_MEMORY_BUDGET["_other"])
            budget_bytes = int(self.max_memory * budget_pct)
            current_usage = sum(e["memory"] for e in entries)

            if current_usage <= budget_bytes:
                eviction_report[intent] = {
                    "status": "within_budget",
                    "usage_bytes": current_usage,
                    "budget_bytes": budget_bytes,
                    "evicted": 0,
                }
                continue

            # Sort by hit count (ascending) then timestamp (ascending)
            # → evict least-hit, oldest entries first
            entries.sort(key=lambda e: (e["hits"], e["ts"]))

            evicted = 0
            for entry in entries:
                if current_usage <= budget_bytes:
                    break
                self.redis.delete(entry["key"])
                current_usage -= entry["memory"]
                evicted += 1

            eviction_report[intent] = {
                "status": "evicted",
                "usage_bytes": current_usage,
                "budget_bytes": budget_bytes,
                "evicted": evicted,
            }

        logger.info("Memory budget enforcement: %s", eviction_report)
        return eviction_report

# EventBridge scheduled rule — runs every 6 hours
{
    "schedule": "rate(6 hours)",
    "target": {
        "arn": "arn:aws:ecs:us-east-1:123456789:cluster/manga-cluster",
        "taskDefinition": "cache-maintenance:latest",
        "overrides": {
            "containerOverrides": [{
                "name": "cache-maintenance",
                "command": ["python", "-m", "cache_maintenance.enforce_budgets"]
            }]
        }
    }
}