Scenarios and Runbooks — Skill 1.6.2: Interactive AI Systems

mindmap
  root((Interactive AI Failures))
    ContextWindowOverflow
      TooManyTurnsIntoPrompt
      LongUserMessagesGrowingContext
      SystemPromptPlusHistoryExceedsLimit
    SessionStateErrors
      LostSessionAfterECSRestart
      WrongSessionMergedIntoContext
      ExpiredSessionLoadedForReturningUser
    ConversationCoherence
      CorefResolutionFails
      TopicSwitchDropsOldContext
      BotContradictsPreviousAnswer
    SessionAnomalies
      ContextPoisoningViaHistory
      SessionHijackCrossUserContext
      InfiniteRetryLoopOnDynamoDBError
    OperationalGaps
      NoSessionTTLManagement
      TurnCountNotMonitored
      NoCostPerSessionMetric

flowchart TD
    A["ECS logs:\nValidationException from Bedrock invoke_model"] --> B{"Parse error message:\n'Input is too long'?"}
    B --> C["Check session history\nturn count in DynamoDB"]
    C --> D["Session has > 30 turns?"]
    D -->|Yes| E["CONFIRM: context window overflow\nfrom unbounded history"]
    E --> F["Apply sliding window:\nkeep last 10 turns only"]
    F --> G["Summarize older turns\ninto a single context block"]
    D -->|No| H["Check retrieval context\nsize — too many large chunks?"]

"""
Runbook: Sliding-window conversation history with summarization for MangaAssist.
"""

import boto3
import json

REGION   = "us-east-1"
HAIKU_ID = "anthropic.claude-3-haiku-20240307-v1:0"

bedrock_rt = boto3.client("bedrock-runtime", region_name=REGION)

MAX_HISTORY_TURNS   = 10     # turns to keep verbatim
SUMMARY_TOKEN_LIMIT = 300    # tokens for the conversation summary block
CONTEXT_TOKEN_LIMIT = 2000   # tokens for retrieval chunks


def estimate_tokens(text: str) -> int:
    return max(1, len(text) // 4)


def summarize_old_turns(turns: list[dict]) -> str:
    """Summarize turns beyond the sliding window using Claude 3 Haiku."""
    if not turns:
        return ""
    history_text = "\n".join(
        f"{'User' if t['role'] == 'user' else 'Assistant'}: {t['content']}"
        for t in turns
    )
    resp = bedrock_rt.invoke_model(
        modelId=HAIKU_ID,
        body=json.dumps({
            "anthropic_version": "bedrock-2023-05-31",
            "max_tokens": SUMMARY_TOKEN_LIMIT,
            "messages": [{
                "role": "user",
                "content": (
                    f"Summarize this manga store conversation in 2-3 sentences, "
                    f"focusing on what manga and topics were discussed:\n\n{history_text}"
                ),
            }],
        }),
        accept="application/json", contentType="application/json",
    )
    return json.loads(resp["body"].read())["content"][0]["text"].strip()


def build_messages_with_sliding_window(
    conversation_history: list[dict],
    current_user_message: str,
    context_chunks: list[str],
    system_prompt: str,
) -> tuple[list, int]:
    """
    Build Bedrock messages array with sliding window and context budget.
    Returns (messages, estimated_tokens).
    """
    # Split history into old (summarized) and recent (verbatim)
    old_turns    = conversation_history[:-MAX_HISTORY_TURNS] if len(conversation_history) > MAX_HISTORY_TURNS else []
    recent_turns = conversation_history[-MAX_HISTORY_TURNS:]

    # Summarize old turns
    summary_block = ""
    if old_turns:
        summary = summarize_old_turns(old_turns)
        summary_block = f"[Earlier conversation summary]\n{summary}\n\n"

    # Trim retrieval context to token budget
    context_parts = []
    used_tokens   = 0
    for chunk in context_chunks:
        chunk_tokens = estimate_tokens(chunk)
        if used_tokens + chunk_tokens > CONTEXT_TOKEN_LIMIT:
            break
        context_parts.append(chunk)
        used_tokens += chunk_tokens
    context_block = "\n\n".join(f"[Catalog {i+1}]\n{c}"
                                for i, c in enumerate(context_parts))

    # Assemble the final user message
    current_message_content = (
        f"{summary_block}"
        f"Catalog context:\n{context_block}\n\n"
        f"Question: {current_user_message}"
    )

    messages = [
        *[{"role": t["role"], "content": t["content"]} for t in recent_turns],
        {"role": "user", "content": current_message_content},
    ]

    total_tokens = (estimate_tokens(system_prompt)
                    + sum(estimate_tokens(m["content"]) for m in messages))
    return messages, total_tokens


def safe_invoke(conversation_history, current_user_message, context_chunks, system_prompt):
    messages, total_tokens = build_messages_with_sliding_window(
        conversation_history, current_user_message, context_chunks, system_prompt
    )
    print(f"[CONTEXT] Estimated tokens: {total_tokens}")
    resp = bedrock_rt.invoke_model(
        modelId="anthropic.claude-3-sonnet-20240229-v1:0",
        body=json.dumps({
            "anthropic_version": "bedrock-2023-05-31",
            "max_tokens": 512,
            "system": system_prompt,
            "messages": messages,
        }),
        accept="application/json", contentType="application/json",
    )
    return json.loads(resp["body"].read())["content"][0]["text"]

flowchart TD
    A["Users report: chatbot\n'forgot' past conversation"] --> B{"Check ECS deployment events:\nwas a task restart recent?"}
    B --> C["Restart timestamp matches\nreport timing?"]
    C -->|Yes| D["CONFIRM: in-memory session state\nlost due to ECS restart"]
    D --> E["Migrate session storage from\nin-memory to DynamoDB + Redis"]
    E --> F["Test: restart ECS task and\nverify session survives"]
    D --> G["Add connection draining:\nwait 30s for active sessions\nbefore replacing task"]
    C -->|No| H["Check Redis connection timeout\nor session TTL expiry"]

"""
Runbook: Persistent session management using DynamoDB + Redis for MangaAssist.
"""

import boto3
import redis
import json
import time

REGION  = "us-east-1"
TABLE   = boto3.resource("dynamodb", region_name=REGION).Table("mangaassist-sessions")
r       = redis.Redis(
    host="mangaassist-cache.abc.ng.0001.use1.cache.amazonaws.com",
    port=6379, ssl=True, decode_responses=True,
)

SESSION_CACHE_TTL = 1800   # 30 min in Redis
DYNAMO_SESSION_TTL = 86400  # 24 hr in DynamoDB (ttl field for auto-expiry)


def load_session(session_id: str) -> list[dict]:
    """Load conversation history: Redis first, then DynamoDB."""
    # L1 cache: Redis
    cached = r.get(f"session:{session_id}:history")
    if cached:
        return json.loads(cached)

    # L2 persistent: DynamoDB
    resp   = TABLE.get_item(Key={"pk": f"SESSION#{session_id}"})
    item   = resp.get("Item")
    if item:
        history = json.loads(item.get("history", "[]"))
        r.setex(f"session:{session_id}:history", SESSION_CACHE_TTL, json.dumps(history))
        return history
    return []


def save_session(session_id: str, history: list[dict]):
    """Persist conversation history to DynamoDB and update Redis."""
    ttl_ts = int(time.time()) + DYNAMO_SESSION_TTL
    TABLE.put_item(Item={
        "pk":      f"SESSION#{session_id}",
        "history": json.dumps(history),
        "ttl":     ttl_ts,
    })
    r.setex(f"session:{session_id}:history", SESSION_CACHE_TTL, json.dumps(history))


def append_turn(session_id: str, role: str, content: str, max_turns: int = 20):
    """Append a turn to the session and persist immediately."""
    history = load_session(session_id)
    history.append({"role": role, "content": content})
    if len(history) > max_turns * 2:   # keep max_turns user+assistant pairs
        history = history[-(max_turns * 2):]
    save_session(session_id, history)
    return history


# ── ECS task shutdown handler ─────────────────────────────────────────────────
import signal
import sys

def graceful_shutdown(signum, frame):
    """Flush all pending session writes on SIGTERM (ECS graceful stop)."""
    print("[SHUTDOWN] Received SIGTERM — flushing sessions before exit")
    # In production: iterate active_sessions dict and save_session() for each
    sys.exit(0)

signal.signal(signal.SIGTERM, graceful_shutdown)

flowchart TD
    A["User reports wrong answer\nfor 'this one' / 'the first one' style queries"] --> B{"Check conversation history:\nis the referent in the visible window?"}
    B --> C["Referent introduced before\nthe sliding window cutoff?"]
    C -->|Yes| D["CONFIRM: coreference fails\ndue to truncated history"]
    D --> E["Expand window or \nadd entity memory to conversation summary"]
    E --> F["Summarize old turn entities:\ntrack 'first = Vinland Saga'"]
    D --> G["Add clarification prompt:\nrequest entity name if resolution ambiguous"]
    C -->|No| H["Check if retrieval context\nis providing a confusing first result"]

"""
Runbook: Entity-preserving conversation summary for coreference continuity.
"""

import boto3
import json
import re

REGION   = "us-east-1"
HAIKU_ID = "anthropic.claude-3-haiku-20240307-v1:0"

bedrock_rt = boto3.client("bedrock-runtime", region_name=REGION)


def extract_manga_entity_map(turns: list[dict]) -> dict[str, str]:
    """
    Use Claude 3 Haiku to extract ordinal-to-title mappings from old turns.
    Returns a dict like: {"the first one": "Vinland Saga", "second": "Berserk"}
    """
    if not turns:
        return {}
    history = "\n".join(
        f"{'User' if t['role']=='user' else 'Bot'}: {t['content']}"
        for t in turns
    )
    resp = bedrock_rt.invoke_model(
        modelId=HAIKU_ID,
        body=json.dumps({
            "anthropic_version": "bedrock-2023-05-31",
            "max_tokens": 150,
            "messages": [{
                "role": "user",
                "content": (
                    f"From this conversation, extract any ordinal or pronoun references "
                    f"to manga titles. Return as JSON where keys are the reference phrase "
                    f"and values are the manga title. Example: {{\"the first one\": \"Vinland Saga\"}}.\n\n"
                    f"Conversation:\n{history}"
                ),
            }],
        }),
        accept="application/json", contentType="application/json",
    )
    raw = json.loads(resp["body"].read())["content"][0]["text"].strip()
    try:
        match = re.search(r"\{.*\}", raw, re.DOTALL)
        return json.loads(match.group(0)) if match else {}
    except Exception:
        return {}


def build_entity_aware_summary(old_turns: list[dict]) -> str:
    """
    Create a summary block that preserves entity-reference mappings.
    """
    entity_map = extract_manga_entity_map(old_turns)
    if not entity_map:
        return ""

    mapping_lines = "\n".join(f"  - '{ref}' = {title}"
                              for ref, title in entity_map.items())
    return (
        f"[Conversation entity memory — resolve references using this map]\n"
        f"{mapping_lines}\n"
    )


def build_messages_with_entity_memory(
    conversation_history: list[dict],
    current_message: str,
    context_chunks: list[str],
    max_recent_turns: int = 10,
) -> list[dict]:
    old_turns    = conversation_history[:-max_recent_turns]
    recent_turns = conversation_history[-max_recent_turns:]

    entity_block  = build_entity_aware_summary(old_turns)
    context_block = "\n\n".join(f"[Catalog {i+1}]\n{c}"
                                for i, c in enumerate(context_chunks[:3]))

    full_message = f"{entity_block}Catalog context:\n{context_block}\n\nQuestion: {current_message}"
    return [
        *[{"role": t["role"], "content": t["content"]} for t in recent_turns],
        {"role": "user", "content": full_message},
    ]

flowchart TD
    A["Customer claims AI 'previously said'\na price that no product has"] --> B{"Pull session history from DynamoDB:\nis there an assistant turn with ¥100?"}
    B --> C["Assistant turn exists with\nfabricated price not in catalog?"]
    C -->|Yes| D["CONFIRM: context poisoning\nvia session manipulation"]
    D --> E["Revoke session; invalidate\ncorresponding Redis keys"]
    E --> F["Sign session history entries:\nHMAC of each turn"]
    D --> G["Log security incident;\nreview other sessions from same user"]
    C -->|No| H["Check for hallucinated price\nin current response — grounding failure?"]

"""
Runbook: Session integrity protection with signed history entries.
"""

import boto3
import json
import hmac
import hashlib
import os

REGION       = "us-east-1"
SIGNING_KEY  = os.environ.get("SESSION_SIGNING_KEY", "PLACEHOLDER_REPLACE_IN_PRODUCTION")
TABLE        = boto3.resource("dynamodb", region_name=REGION).Table("mangaassist-sessions")


def sign_turn(role: str, content: str, turn_index: int) -> str:
    """Compute HMAC-SHA256 signature for a conversation turn."""
    message = f"{turn_index}:{role}:{content}".encode()
    return hmac.new(SIGNING_KEY.encode(), message, hashlib.sha256).hexdigest()


def verify_turn(turn: dict, turn_index: int) -> bool:
    """Verify the integrity of a stored conversation turn."""
    expected = sign_turn(turn["role"], turn["content"], turn_index)
    stored   = turn.get("sig", "")
    return hmac.compare_digest(expected, stored)


def save_turn(session_id: str, role: str, content: str) -> dict:
    """Append and persist a turn with an integrity signature."""
    resp    = TABLE.get_item(Key={"pk": f"SESSION#{session_id}"})
    history = json.loads(resp.get("Item", {}).get("history", "[]"))
    idx     = len(history)
    turn    = {"role": role, "content": content, "sig": sign_turn(role, content, idx)}
    history.append(turn)
    TABLE.put_item(Item={
        "pk":      f"SESSION#{session_id}",
        "history": json.dumps(history),
    })
    return turn


def load_verified_history(session_id: str) -> list[dict]:
    """Load history, discarding any turn that fails integrity verification."""
    resp    = TABLE.get_item(Key={"pk": f"SESSION#{session_id}"})
    history = json.loads(resp.get("Item", {}).get("history", "[]"))
    verified = []
    for idx, turn in enumerate(history):
        if verify_turn(turn, idx):
            verified.append({"role": turn["role"], "content": turn["content"]})
        else:
            print(f"[SECURITY] Turn {idx} in session {session_id} failed verification — discarded")
    if len(verified) < len(history):
        print(f"[SECURITY] Session {session_id}: {len(history)-len(verified)} turns discarded")
    return verified


# ── Session ID validation: ensure session belongs to the caller ───────────────
import re

def validate_session_ownership(session_id: str, user_id: str) -> bool:
    """Verify the session_id is bound to the authenticated user_id."""
    # In production: session_id should encode user_id (e.g., JWT sub claim)
    # Simple check: reject session IDs that don't start with the user's prefix
    if not re.match(r'^[a-zA-Z0-9_-]{20,}$', session_id):
        return False
    resp = TABLE.get_item(Key={"pk": f"SESSION#{session_id}"})
    item = resp.get("Item", {})
    return item.get("user_id") == user_id

flowchart TD
    A["DynamoDB billing alarm:\ncost > $200/day"] --> B{"Check DynamoDB item count\nand size for sessions table"}
    B --> C["Item count > 100K with\nno TTL attribute?"]
    C -->|Yes| D["CONFIRM: no TTL configured;\nstalediscarded sessions accumulating"]
    D --> E["Enable DynamoDB TTL\non 'ttl' attribute immediately"]
    E --> F["Backfill ttl attribute\nfor existing items > 30 days old"]
    D --> G["Scan and delete sessions\nolder than threshold (long-running job)"]
    C -->|No| H["Check read capacity units\nor analytics scan queries"]

"""
Runbook: DynamoDB TTL enforcement and backfill for session table.
"""

import boto3
import time

REGION     = "us-east-1"
TABLE_NAME = "mangaassist-sessions"
SESSION_LIFE_SECONDS = 30 * 24 * 3600  # 30 days

dynamo   = boto3.client("dynamodb", region_name=REGION)
dynamo_r = boto3.resource("dynamodb", region_name=REGION)
table    = dynamo_r.Table(TABLE_NAME)


# ── Step 1: Enable TTL on the table ──────────────────────────────────────────
def enable_dynamodb_ttl():
    dynamo.update_time_to_live(
        TableName=TABLE_NAME,
        TimeToLiveSpecification={"Enabled": True, "AttributeName": "ttl"},
    )
    print(f"[TTL] DynamoDB TTL enabled on '{TABLE_NAME}' using attribute 'ttl'")


# ── Step 2: Write sessions with TTL ──────────────────────────────────────────
def create_session(session_id: str, user_id: str) -> dict:
    """Always write a TTL timestamp when creating a session."""
    now = int(time.time())
    item = {
        "pk":      f"SESSION#{session_id}",
        "user_id": user_id,
        "history": "[]",
        "created_at": now,
        "ttl":     now + SESSION_LIFE_SECONDS,
    }
    table.put_item(Item=item)
    return item


# ── Step 3: Backfill TTL for existing items without ttl attribute ─────────────
def backfill_session_ttl(dry_run: bool = True):
    """
    Scan the sessions table and add 'ttl' attribute to items missing it.
    Run with dry_run=True first to estimate item count.
    """
    paginator = dynamo.get_paginator("scan")
    stale_cutoff = int(time.time()) - SESSION_LIFE_SECONDS
    count_missing = 0
    count_expired = 0

    for page in paginator.paginate(TableName=TABLE_NAME, ProjectionExpression="pk,created_at,#ttl",
                                   ExpressionAttributeNames={"#ttl": "ttl"}):
        for item in page["Items"]:
            pk = item["pk"]["S"]
            if "ttl" in item:
                continue  # already has TTL
            created_at = int(item.get("created_at", {}).get("N", 0))
            is_expired = created_at > 0 and created_at < stale_cutoff
            count_missing += 1
            if is_expired:
                count_expired += 1
            if not dry_run:
                new_ttl = max(created_at + SESSION_LIFE_SECONDS, int(time.time()) + 3600)
                table.update_item(
                    Key={"pk": pk},
                    UpdateExpression="SET #ttl = :ttl",
                    ExpressionAttributeNames={"#ttl": "ttl"},
                    ExpressionAttributeValues={":ttl": new_ttl},
                )

    print(f"[BACKFILL] {'DRY RUN: ' if dry_run else ''}Items missing TTL: {count_missing}, "
          f"likely expired (> 30 days): {count_expired}")