Scenarios and Runbooks — Safeguarded AI Workflows

mindmap
  root((Safeguard<br/>Failure Modes))
    Circuit Breaker Issues
      Stuck Open After Throttling
      Premature Opening
      State Corruption
    Lambda Timeout
      Partial Response on Timeout
      Cold Start Exceeding Budget
      Orphaned Bedrock Call
    IAM Boundaries
      Too Restrictive Blocking Model
      Missing Condition Key
      Cross-Account Access Denied
    Step Functions
      Infinite Loop via Choice State
      Heartbeat Timeout Mismatch
      Retry Multiplication
    Guardrails
      False Positive Blocking Manga Content
      Grounding Score Too Strict
      PII Detection Over-Triggering

flowchart TD
    A["CloudWatch Alarm:<br/>fallback_response_rate > 80%<br/>for > 5 minutes"] --> B{"Check circuit<br/>breaker state"}
    B -->|"State = OPEN"| C["Circuit Stuck Open<br/>Confirmed"]
    B -->|"State = CLOSED"| D["Different issue —<br/>check Bedrock directly"]
    C --> E{"Is Bedrock actually<br/>available? (manual probe)"}
    E -->|"Yes — Bedrock responds<br/>normally"| F["ROOT CAUSE:<br/>Circuit stuck open despite<br/>service recovery"]
    E -->|"No — Bedrock still<br/>failing"| G["Correct behavior —<br/>wait for Bedrock recovery"]
    F --> H{"Check half-open<br/>probe history"}
    H -->|"Probes attempted but<br/>coincidentally failed"| I["ROOT CAUSE:<br/>Fragile single-probe recovery<br/>fails on any transient error"]
    I --> J["Execute Runbook 1"]

"""
Runbook 1: Fix stuck-open circuit breaker with improved recovery logic.
Two fixes: (1) multi-probe minimum even at stage 0, (2) manual override.
"""

import time
import json
import logging
import redis

logger = logging.getLogger("manga_cb_stuck_open")


class CircuitBreakerRecoveryFix:
    """
    Fixes the fragile single-probe recovery problem.

    Changes:
    1. Minimum 3 probes even at stage 0 (was 1)
    2. Tolerate 1 failure per stage (was 0)
    3. Add manual override capability for on-call engineers
    4. Add health check endpoint that bypasses the circuit
    """

    REDIS_PREFIX = "manga:cb:"

    def __init__(self, redis_client: redis.Redis):
        self._redis = redis_client

    def force_close_circuit(self, service_name: str, reason: str) -> dict:
        """
        Manual override: force a circuit breaker to CLOSED state.
        Used by on-call engineers when the circuit is stuck open
        but the service is confirmed healthy.
        """
        key = f"{self.REDIS_PREFIX}{service_name}"

        old_state = self._redis.hgetall(key)
        old_state_str = {
            k.decode(): v.decode()
            for k, v in old_state.items()
        } if old_state else {}

        # Force to closed
        new_state = {
            "state": "CLOSED",
            "failure_count": "0",
            "success_count": "0",
            "total_requests": "0",
            "last_failure": "0",
            "last_state_change": str(time.time()),
            "recovery_stage": "0",
            "probes_in_flight": "0",
        }

        self._redis.hset(key, mapping=new_state)
        self._redis.expire(key, 86400)

        # Log the override with audit trail
        audit_key = f"{self.REDIS_PREFIX}audit:{service_name}:{int(time.time())}"
        self._redis.setex(audit_key, 86400 * 7, json.dumps({
            "action": "FORCE_CLOSE",
            "reason": reason,
            "previous_state": old_state_str,
            "new_state": new_state,
            "timestamp": time.time(),
        }))

        logger.warning(
            "Circuit [%s] FORCE CLOSED by operator: %s (was %s)",
            service_name,
            reason,
            old_state_str.get("state", "UNKNOWN"),
        )

        return {
            "service": service_name,
            "action": "FORCE_CLOSE",
            "previous_state": old_state_str.get("state", "UNKNOWN"),
            "new_state": "CLOSED",
        }

    def health_check_bypass(
        self, service_name: str, check_fn
    ) -> dict:
        """
        Run a health check that bypasses the circuit breaker.
        If the service is healthy, force the circuit closed.
        """
        try:
            start = time.monotonic()
            result = check_fn()
            latency_ms = (time.monotonic() - start) * 1000

            if result.get("healthy", False):
                self.force_close_circuit(
                    service_name,
                    f"Health check passed (latency={latency_ms:.0f}ms)",
                )
                return {
                    "healthy": True,
                    "action": "CIRCUIT_FORCE_CLOSED",
                    "latency_ms": latency_ms,
                }
            else:
                return {
                    "healthy": False,
                    "action": "NONE",
                    "reason": result.get("reason", "Health check failed"),
                }

        except Exception as e:
            return {
                "healthy": False,
                "action": "NONE",
                "reason": str(e),
            }


def create_improved_recovery_config():
    """
    Improved recovery stages that are more resilient to transient failures.

    Old: [1, 3, 10] with 0 failure tolerance
    New: [3, 5, 15] with 1 failure tolerance per stage
    """
    return {
        "progressive_recovery_stages": [3, 5, 15],
        "failures_tolerated_per_stage": 1,
        "probe_timeout_seconds": 10,
        "stage_evaluation_window_seconds": 30,
    }

flowchart TD
    A["User reports:<br/>'Response cut off mid-sentence'"] --> B["Check Lambda<br/>CloudWatch logs"]
    B --> C{"Lambda execution<br/>duration close to timeout?"}
    C -->|"Duration > 28s<br/>(timeout = 30s)"| D["Lambda Timeout<br/>During Post-Processing"]
    C -->|"Duration < 25s"| E["Different issue —<br/>check response formatting"]
    D --> F{"Check if session<br/>state was saved"}
    F -->|"No DynamoDB write<br/>in logs"| G["ROOT CAUSE:<br/>Lambda timed out after model<br/>response but before session save"]
    G --> H["Execute Runbook 2"]

"""
Runbook 2: Lambda timeout-safe response handling with pre-save checkpointing.
Ensures the model response is saved BEFORE any post-processing.
"""

import json
import time
import logging
import boto3

logger = logging.getLogger("manga_timeout_safe")

dynamodb = boto3.resource("dynamodb", region_name="us-east-1")
response_table = dynamodb.Table("manga_response_checkpoint")


class TimeoutSafeResponseHandler:
    """
    Saves the model response to a checkpoint immediately after receiving it,
    BEFORE any post-processing. If the Lambda times out during post-processing,
    the response is still retrievable from the checkpoint table.
    """

    def __init__(self, context):
        self._context = context
        self._checkpoint_id = None

    def checkpoint_response(
        self, session_id: str, response_text: str, usage: dict
    ) -> str:
        """
        Save the model response to a checkpoint table immediately.
        This is the first thing we do after receiving the Bedrock response.
        Fast write: ~50ms to DynamoDB.
        """
        checkpoint_id = f"{session_id}:{int(time.time() * 1000)}"

        response_table.put_item(Item={
            "checkpoint_id": checkpoint_id,
            "session_id": session_id,
            "response_text": response_text,
            "usage": json.dumps(usage),
            "saved_at": int(time.time()),
            "post_processed": False,
            "ttl": int(time.time()) + 3600,  # 1-hour retention
        })

        self._checkpoint_id = checkpoint_id
        logger.info(
            "Response checkpointed: %s (remaining: %dms)",
            checkpoint_id,
            self._context.get_remaining_time_in_millis(),
        )
        return checkpoint_id

    def mark_post_processed(self) -> None:
        """Mark the checkpoint as fully post-processed."""
        if self._checkpoint_id:
            response_table.update_item(
                Key={"checkpoint_id": self._checkpoint_id},
                UpdateExpression="SET post_processed = :pp",
                ExpressionAttributeValues={":pp": True},
            )

    @staticmethod
    def recover_from_checkpoint(session_id: str) -> dict:
        """
        Recover the last un-post-processed response for a session.
        Called by a Step Functions recovery state after a timeout.
        """
        response = response_table.query(
            IndexName="session_id-saved_at-index",
            KeyConditionExpression="session_id = :sid",
            ExpressionAttributeValues={":sid": session_id},
            ScanIndexForward=False,  # Most recent first
            Limit=1,
        )

        items = response.get("Items", [])
        if not items:
            return {"recovered": False}

        checkpoint = items[0]
        return {
            "recovered": True,
            "response_text": checkpoint["response_text"],
            "checkpoint_id": checkpoint["checkpoint_id"],
            "post_processed": checkpoint.get("post_processed", False),
        }


def lambda_handler_improved(event: dict, context) -> dict:
    """
    Improved Lambda handler with checkpoint-first pattern.
    """
    handler = TimeoutSafeResponseHandler(context)
    session_id = event.get("sessionId", "")

    # Phase 1-3: Load context, build prompt, invoke Bedrock (as before)
    response_text = _invoke_bedrock(event)
    usage = {"input_tokens": 500, "output_tokens": 300}

    # CRITICAL: Checkpoint the response IMMEDIATELY
    checkpoint_id = handler.checkpoint_response(
        session_id, response_text, usage
    )

    # Phase 4: Post-processing (may timeout — that is OK now)
    remaining_ms = context.get_remaining_time_in_millis()
    if remaining_ms > 3000:
        # Enough time for full post-processing
        _save_session_state(session_id, response_text)
        _update_metrics(usage)
        handler.mark_post_processed()
    else:
        # Not enough time — skip post-processing; it will be retried
        logger.warning(
            "Insufficient time for post-processing (%dms) — "
            "checkpoint %s will be recovered",
            remaining_ms, checkpoint_id,
        )

    return {
        "statusCode": 200,
        "answer": response_text,
        "checkpoint_id": checkpoint_id,
    }


def _invoke_bedrock(event: dict) -> str:
    """Placeholder for Bedrock invocation."""
    return "Model response text"


def _save_session_state(session_id: str, response: str) -> None:
    """Placeholder for session state save."""
    pass


def _update_metrics(usage: dict) -> None:
    """Placeholder for metrics update."""
    pass

flowchart TD
    A["CloudWatch Alarm:<br/>Bedrock InvokeModel errors = 100%"] --> B{"Check error type"}
    B -->|"AccessDeniedException"| C["IAM Permission<br/>Issue Confirmed"]
    B -->|"ThrottlingException"| D["Rate limiting —<br/>different runbook"]
    C --> E{"When did this start?"}
    E -->|"Correlates with<br/>deployment timestamp"| F["ROOT CAUSE:<br/>IAM policy change in<br/>latest deployment"]
    E -->|"No recent deployment"| G["Check for SCP change<br/>or permission boundary update"]
    F --> H{"Inspect the<br/>IAM policy diff"}
    H --> I["Region mismatch:<br/>Policy says us-east-1<br/>Lambda runs in ap-northeast-1"]
    I --> J["Execute Runbook 3"]

arn:aws:bedrock:us-east-1::foundation-model/anthropic.claude-3-sonnet-*

arn:aws:bedrock:ap-northeast-1::foundation-model/anthropic.claude-3-sonnet-*

"""
Runbook 3: IAM policy validation and automated region-aware policy generation.
Prevents region mismatch in Bedrock model ARNs.
"""

import json
import logging
import boto3

logger = logging.getLogger("manga_iam_validator")

iam_client = boto3.client("iam")


class IAMPolicyValidator:
    """
    Validates IAM policies for MangaAssist Lambda roles.
    Catches common mistakes like region mismatches before deployment.
    """

    EXPECTED_REGIONS = ["ap-northeast-1", "us-east-1"]  # Primary + DR

    REQUIRED_BEDROCK_MODELS = [
        "anthropic.claude-3-sonnet-20240229-v1:0",
        "anthropic.claude-3-haiku-20240307-v1:0",
    ]

    def validate_bedrock_access(
        self, policy_document: dict, expected_region: str
    ) -> list[str]:
        """
        Validate that the IAM policy grants Bedrock access in the correct region.
        Returns a list of issues found (empty = valid).
        """
        issues = []

        for statement in policy_document.get("Statement", []):
            if statement.get("Effect") != "Allow":
                continue

            actions = statement.get("Action", [])
            if isinstance(actions, str):
                actions = [actions]

            bedrock_actions = [a for a in actions if a.startswith("bedrock:")]
            if not bedrock_actions:
                continue

            resources = statement.get("Resource", [])
            if isinstance(resources, str):
                resources = [resources]

            for resource in resources:
                if resource == "*":
                    issues.append(
                        "WARN: Wildcard resource for Bedrock — should restrict to specific models"
                    )
                    continue

                if "foundation-model/" in resource:
                    # Check region in ARN
                    arn_parts = resource.split(":")
                    if len(arn_parts) >= 4:
                        arn_region = arn_parts[3]
                        if arn_region != expected_region:
                            issues.append(
                                f"ERROR: Region mismatch — policy ARN has "
                                f"'{arn_region}' but Lambda runs in "
                                f"'{expected_region}'. Fix the ARN to: "
                                f"arn:aws:bedrock:{expected_region}::foundation-model/..."
                            )

            # Check that required models are covered
            for model in self.REQUIRED_BEDROCK_MODELS:
                model_covered = any(
                    model in r or "*" in r for r in resources
                )
                if not model_covered:
                    issues.append(
                        f"WARN: Model '{model}' not explicitly covered in policy"
                    )

        return issues

    def generate_correct_policy(self, region: str, account_id: str) -> dict:
        """
        Generate a correct Bedrock access policy for the given region.
        """
        model_arns = [
            f"arn:aws:bedrock:{region}::foundation-model/{model}"
            for model in self.REQUIRED_BEDROCK_MODELS
        ]

        return {
            "Version": "2012-10-17",
            "Statement": [
                {
                    "Sid": "AllowBedrockInvoke",
                    "Effect": "Allow",
                    "Action": [
                        "bedrock:InvokeModel",
                        "bedrock:InvokeModelWithResponseStream",
                    ],
                    "Resource": model_arns,
                    "Condition": {
                        "StringEquals": {
                            "aws:RequestedRegion": region,
                        },
                    },
                },
                {
                    "Sid": "AllowGuardrails",
                    "Effect": "Allow",
                    "Action": ["bedrock:ApplyGuardrail"],
                    "Resource": [
                        f"arn:aws:bedrock:{region}:{account_id}:guardrail/*"
                    ],
                },
            ],
        }


def pre_deployment_iam_check(
    role_name: str, expected_region: str
) -> dict:
    """
    Run as a pre-deployment check in the CI/CD pipeline.
    Validates the IAM policy before deploying Lambda changes.
    """
    validator = IAMPolicyValidator()

    # Get current role policies
    response = iam_client.list_attached_role_policies(RoleName=role_name)
    all_issues = []

    for policy in response.get("AttachedPolicies", []):
        policy_arn = policy["PolicyArn"]
        policy_version = iam_client.get_policy(
            PolicyArn=policy_arn
        )["Policy"]["DefaultVersionId"]

        policy_doc = iam_client.get_policy_version(
            PolicyArn=policy_arn,
            VersionId=policy_version,
        )["PolicyVersion"]["Document"]

        issues = validator.validate_bedrock_access(policy_doc, expected_region)
        if issues:
            all_issues.extend(issues)

    return {
        "role": role_name,
        "issues": all_issues,
        "valid": len([i for i in all_issues if i.startswith("ERROR")]) == 0,
    }

flowchart TD
    A["CloudWatch Alarm:<br/>StepFunctions ExecutionDuration<br/>> 55s (p95)"] --> B{"Check execution<br/>event history"}
    B -->|"Same state visited > 10 times"| C["Infinite Loop<br/>Confirmed"]
    B -->|"Normal progression"| D["Slow execution —<br/>different issue"]
    C --> E{"Check the Choice state<br/>condition variable"}
    E -->|"Variable never reaches<br/>expected value"| F["ROOT CAUSE:<br/>Condition variable not<br/>updated by Lambda handler"]
    E -->|"Variable oscillates"| G["ROOT CAUSE:<br/>Lambda handler alternates<br/>the variable value"]
    F --> H["Execute Runbook 4"]
    G --> H

"""
Runbook 4: Step Functions loop guard with external iteration tracking
and automatic execution termination.
"""

import time
import json
import logging
import boto3

logger = logging.getLogger("manga_loop_guard")

sfn_client = boto3.client("stepfunctions", region_name="us-east-1")
cloudwatch = boto3.client("cloudwatch", region_name="us-east-1")


class StepFunctionsLoopGuard:
    """
    External loop guard for Step Functions workflows.
    Tracks state visit counts in Redis and terminates executions
    that exceed the maximum iteration count.
    """

    MAX_STATE_VISITS = 10  # No single state should be visited more than 10 times
    REDIS_PREFIX = "manga:sfn:guard:"

    def __init__(self, redis_client):
        self._redis = redis_client

    def record_state_visit(
        self, execution_id: str, state_name: str
    ) -> dict:
        """
        Record a state visit. Called by a Lambda at the start of each state.
        Returns whether the execution should continue.
        """
        key = f"{self.REDIS_PREFIX}{execution_id}"
        field = f"state:{state_name}"

        visit_count = self._redis.hincrby(key, field, 1)
        self._redis.expire(key, 300)  # 5-minute TTL

        total_visits = sum(
            int(v) for v in self._redis.hvals(key) if v
        )

        if visit_count > self.MAX_STATE_VISITS:
            logger.error(
                "Loop detected: execution=%s state=%s visits=%d",
                execution_id, state_name, visit_count,
            )
            return {
                "continue": False,
                "reason": f"State '{state_name}' visited {visit_count} times (max: {self.MAX_STATE_VISITS})",
                "total_visits": total_visits,
            }

        return {"continue": True, "visit_count": visit_count}

    def terminate_stuck_execution(
        self, execution_arn: str, reason: str
    ) -> None:
        """Terminate a stuck Step Functions execution."""
        try:
            sfn_client.stop_execution(
                executionArn=execution_arn,
                error="LoopGuardTermination",
                cause=reason,
            )
            logger.warning(
                "Terminated stuck execution: %s — %s",
                execution_arn, reason,
            )
        except Exception as e:
            logger.error(
                "Failed to terminate execution %s: %s",
                execution_arn, str(e),
            )


def create_loop_safe_increment_handler():
    """
    Fixed Lambda handler that properly increments the iteration counter.
    Includes a guard check that prevents the loop from running away.
    """
    def handler(event, context):
        ctx = event.get("reasoning_context", {})

        # BUG FIX: Properly increment the counter
        current = int(ctx.get("current_iteration", 0))
        ctx["current_iteration"] = current + 1  # Was: current (no increment)

        # Guard check: raise error if iteration is unreasonably high
        if ctx["current_iteration"] > 20:
            raise Exception(
                f"Safety guard: iteration count {ctx['current_iteration']} "
                f"exceeds maximum. Possible infinite loop."
            )

        return ctx

    return handler

flowchart TD
    A["Support ticket:<br/>'Chatbot won't answer my<br/>question about manga'"] --> B["Check session trace<br/>for guardrail block"]
    B -->|"GUARDRAIL_INTERVENED<br/>in trace"| C["Guardrail Block<br/>Confirmed"]
    C --> D{"Review the blocked<br/>content"}
    D -->|"Content is a legitimate<br/>manga product query"| E["False Positive<br/>Confirmed"]
    D -->|"Content violates policy"| F["Correct block —<br/>no action needed"]
    E --> G{"Check which filter<br/>triggered"}
    G -->|"Violence filter with<br/>'graphic' keyword"| H["ROOT CAUSE:<br/>Guardrail violence filter<br/>triggered by manga-appropriate<br/>vocabulary"]
    H --> I["Execute Runbook 5"]

"""
Runbook 5: Guardrail false positive management with domain-specific
word lists and pre-screening bypass for product queries.
"""

import json
import re
import logging
import boto3

logger = logging.getLogger("manga_guardrail_fp")

bedrock = boto3.client("bedrock", region_name="us-east-1")


class GuardrailFalsePositiveManager:
    """
    Manages guardrail false positives for manga-domain queries.

    Strategy:
    1. Pre-screen queries for manga-domain vocabulary
    2. If the query is product-related, apply guardrail with elevated thresholds
    3. Track false positive patterns for guardrail config updates
    """

    # Manga-domain words that trigger guardrail violence filters but are
    # legitimate product vocabulary
    MANGA_DOMAIN_VOCABULARY = {
        "graphic",       # "graphic novel" not "graphic violence"
        "battle",        # "battle manga" / "battle scenes"
        "fight",         # "fight choreography"
        "death",         # "Death Note" (manga title)
        "attack",        # "Attack on Titan" (manga title)
        "kill",          # "kill la kill" (anime/manga title)
        "blood",         # "Blood+" (manga title) / common genre element
        "demon",         # "Demon Slayer" (manga title)
        "chainsaw",      # "Chainsaw Man" (manga title)
        "assassination", # "Assassination Classroom" (manga title)
        "hell",          # "Hellsing" (manga title)
        "gun",           # "Gundam" (franchise)
        "sword",         # common manga element
        "dark",          # "darker" tone preferences
        "violent",       # "violent manga" as genre search
    }

    # Patterns that indicate a product query vs. harmful content
    PRODUCT_QUERY_PATTERNS = [
        re.compile(r"volume[s]?\s+\d+", re.IGNORECASE),
        re.compile(r"(which|what|best|favorite)\s+manga", re.IGNORECASE),
        re.compile(r"(recommend|suggest|similar)\s+", re.IGNORECASE),
        re.compile(r"(price|cost|buy|order|stock)", re.IGNORECASE),
        re.compile(r"(genre|shounen|seinen|shojo)", re.IGNORECASE),
        re.compile(r"(chapter|arc|series|author)", re.IGNORECASE),
    ]

    def is_likely_product_query(self, query: str) -> bool:
        """
        Determine if a query is about manga products (vs. harmful content).
        If it matches product patterns and uses domain vocabulary,
        it is likely a false positive candidate.
        """
        has_domain_word = any(
            word in query.lower() for word in self.MANGA_DOMAIN_VOCABULARY
        )
        has_product_pattern = any(
            pattern.search(query) for pattern in self.PRODUCT_QUERY_PATTERNS
        )
        return has_domain_word and has_product_pattern

    def apply_guardrail_with_context(
        self,
        content: str,
        source: str = "INPUT",
        is_product_query: bool = False,
    ) -> dict:
        """
        Apply Bedrock Guardrail with domain-aware context.
        For product queries, use a higher threshold guardrail version.
        """
        # Use different guardrail versions based on query type
        if is_product_query:
            guardrail_id = "manga-content-guardrail"
            guardrail_version = "2"  # Version 2: Higher violence threshold for manga domain
        else:
            guardrail_id = "manga-content-guardrail"
            guardrail_version = "1"  # Version 1: Standard thresholds

        bedrock_runtime = boto3.client("bedrock-runtime", region_name="us-east-1")

        try:
            response = bedrock_runtime.apply_guardrail(
                guardrailIdentifier=guardrail_id,
                guardrailVersion=guardrail_version,
                source=source,
                content=[{"text": {"text": content}}],
            )

            action = response.get("action", "NONE")

            if action == "GUARDRAIL_INTERVENED" and is_product_query:
                # Log as potential false positive
                logger.info(
                    "Guardrail blocked product query (potential FP): '%s...'",
                    content[:100],
                )
                self._log_false_positive(content, response)

            return {
                "action": action,
                "guardrail_version": guardrail_version,
                "is_product_query": is_product_query,
                "response": response,
            }

        except Exception as e:
            logger.error("Guardrail call failed: %s", e)
            return {"action": "NONE", "error": str(e)}

    def _log_false_positive(self, content: str, response: dict) -> None:
        """Log a potential false positive for review and guardrail tuning."""
        dynamodb = boto3.resource("dynamodb", region_name="us-east-1")
        table = dynamodb.Table("manga_guardrail_fp_log")

        import time
        table.put_item(Item={
            "log_id": f"fp-{int(time.time() * 1000)}",
            "content_preview": content[:500],
            "guardrail_response": json.dumps(response, default=str)[:2000],
            "timestamp": int(time.time()),
            "reviewed": False,
            "ttl": int(time.time()) + 86400 * 30,  # 30-day retention
        })


def update_guardrail_for_manga_domain():
    """
    Create Guardrail Version 2 with manga-domain-appropriate thresholds.

    Changes from Version 1:
    - Violence filter: LOW → MEDIUM (allows manga battle discussion)
    - Add contextual grounding check (prevents actual harmful content)
    - Add word filter exemptions for manga titles containing violent words
    """
    return {
        "guardrailId": "manga-content-guardrail",
        "version": "2",
        "contentPolicyConfig": {
            "filtersConfig": [
                {
                    "type": "VIOLENCE",
                    "inputStrength": "MEDIUM",   # Was LOW
                    "outputStrength": "MEDIUM",  # Was LOW
                },
                {
                    "type": "HATE",
                    "inputStrength": "LOW",
                    "outputStrength": "LOW",
                },
                {
                    "type": "SEXUAL",
                    "inputStrength": "LOW",
                    "outputStrength": "LOW",
                },
                {
                    "type": "INSULTS",
                    "inputStrength": "HIGH",
                    "outputStrength": "HIGH",
                },
            ],
        },
        "wordPolicyConfig": {
            "managedWordListsConfig": [
                {"type": "PROFANITY"},
            ],
            # Do NOT add manga titles to the word filter blocklist
        },
        "contextualGroundingPolicyConfig": {
            "filtersConfig": [
                {
                    "type": "GROUNDING",
                    "threshold": 0.7,
                },
                {
                    "type": "RELEVANCE",
                    "threshold": 0.6,
                },
            ],
        },
    }

flowchart TD
    START["Safeguard System Alert"] --> Q1{"What is the<br/>symptom?"}
    Q1 -->|"High fallback rate<br/>with OPEN circuit"| S1["Scenario 1:<br/>Stuck Circuit Breaker"]
    Q1 -->|"Truncated user response"| S2["Scenario 2:<br/>Lambda Timeout"]
    Q1 -->|"AccessDeniedException<br/>on all requests"| S3["Scenario 3:<br/>IAM Misconfiguration"]
    Q1 -->|"Workflow exceeding<br/>60s regularly"| S4["Scenario 4:<br/>Infinite Loop"]
    Q1 -->|"Users report legitimate<br/>queries blocked"| S5["Scenario 5:<br/>Guardrail False Positive"]
    S1 --> R1["Health check bypass<br/>→ Force close circuit<br/>→ Fix probe logic"]
    S2 --> R2["Checkpoint response before post-processing<br/>→ Recovery state in Step Functions"]
    S3 --> R3["Fix region in ARN<br/>→ Add CI/CD IAM validation<br/>→ Canary test post-deploy"]
    S4 --> R4["Fix increment handler<br/>→ External loop guard<br/>→ Alarm on duration > 50s"]
    S5 --> R5["Create domain-specific guardrail version<br/>→ Pre-screen product queries<br/>→ FP review pipeline"]