Cascading Failures: When One Agent's Mistake Takes Down the Whole System (ASI08)

{
  "Comment": "Agent tool call guarded by a DynamoDB-backed circuit breaker",
  "StartAt": "CheckBreaker",
  "States": {
    "CheckBreaker": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:us-east-1:123456789012:function:circuit-breaker-check",
      "Parameters": { "tool_name.$": "$.tool_name" },
      "ResultPath": "$.breaker",
      "Retry": [
        { "ErrorEquals": ["States.TaskFailed"], "IntervalSeconds": 1, "MaxAttempts": 2, "BackoffRate": 2.0 }
      ],
      "Next": "BreakerDecision"
    },
    "BreakerDecision": {
      "Type": "Choice",
      "Choices": [
        { "Variable": "$.breaker.decision", "StringEquals": "short_circuit", "Next": "GracefulDegradation" }
      ],
      "Default": "AgentToolCall"
    },
    "AgentToolCall": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:us-east-1:123456789012:function:agent-tool-executor",
      "Parameters": { "tool_name.$": "$.tool_name", "args.$": "$.args" },
      "TimeoutSeconds": 30,
      "Retry": [
        { "ErrorEquals": ["ToolTransientError"], "IntervalSeconds": 5, "MaxAttempts": 2, "BackoffRate": 3.0 }
      ],
      "Catch": [
        { "ErrorEquals": ["States.ALL"], "ResultPath": "$.error", "Next": "RecordFailure" }
      ],
      "ResultPath": "$.tool_result",
      "Next": "RecordSuccess"
    },
    "RecordSuccess": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:us-east-1:123456789012:function:circuit-breaker-record",
      "Parameters": { "tool_name.$": "$.tool_name", "outcome": "success" },
      "ResultPath": "$.breaker_update",
      "Retry": [
        { "ErrorEquals": ["States.TaskFailed"], "IntervalSeconds": 1, "MaxAttempts": 2, "BackoffRate": 2.0 }
      ],
      "Next": "Succeeded"
    },
    "RecordFailure": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:us-east-1:123456789012:function:circuit-breaker-record",
      "Parameters": { "tool_name.$": "$.tool_name", "outcome": "failure" },
      "ResultPath": "$.breaker_update",
      "Retry": [
        { "ErrorEquals": ["States.TaskFailed"], "IntervalSeconds": 1, "MaxAttempts": 2, "BackoffRate": 2.0 }
      ],
      "Next": "GracefulDegradation"
    },
    "GracefulDegradation": {
      "Type": "Pass",
      "Result": {
        "status": "degraded",
        "message": "Tool unavailable - returning cached/default response"
      },
      "ResultPath": "$.tool_result",
      "Next": "Succeeded"
    },
    "Succeeded": { "Type": "Succeed" }
  }
}

import os
import time
import boto3
from botocore.exceptions import ClientError

ddb = boto3.resource("dynamodb")
TABLE = ddb.Table(os.environ["BREAKER_TABLE"])

FAILURE_THRESHOLD = int(os.environ.get("FAILURE_THRESHOLD", "5"))
OPEN_SECONDS = int(os.environ.get("OPEN_SECONDS", "30"))  # cooldown + half-open staleness window

def _conditional(update):
    """Run a conditional update. Return True if it applied, False if the condition failed."""
    try:
        update()
        return True
    except ClientError as e:
        if e.response["Error"]["Code"] == "ConditionalCheckFailedException":
            return False
        raise

def check_handler(event, _ctx):
    """Called BEFORE the tool runs. Decides whether to allow the call or short-circuit."""
    tool = event["tool_name"]
    now = int(time.time())
    item = TABLE.get_item(Key={"tool_name": tool}).get("Item") or {}
    state = item.get("state", "CLOSED")

    if state == "CLOSED":
        return {"decision": "allow", "state": "CLOSED"}

    if state == "OPEN":
        opened_at = int(item.get("opened_at", 0))
        if now - opened_at < OPEN_SECONDS:
            return {"decision": "short_circuit", "state": "OPEN"}
        # Cooldown elapsed: exactly one caller wins the single trial slot (OPEN -> HALF_OPEN).
        won = _conditional(lambda: TABLE.update_item(
            Key={"tool_name": tool},
            UpdateExpression="SET #s = :half, half_open_at = :now",
            ConditionExpression="#s = :open AND opened_at = :oa",
            ExpressionAttributeNames={"#s": "state"},
            ExpressionAttributeValues={":half": "HALF_OPEN", ":open": "OPEN", ":oa": opened_at, ":now": now},
        ))
        return {"decision": "allow", "state": "HALF_OPEN"} if won \
            else {"decision": "short_circuit", "state": "OPEN"}

    # HALF_OPEN: a trial is in flight, short-circuit everyone else until it reports back,
    # unless the trial went stale (crashed without recording) -> grab a fresh trial slot.
    half_open_at = int(item.get("half_open_at", now))
    if now - half_open_at < OPEN_SECONDS:
        return {"decision": "short_circuit", "state": "HALF_OPEN"}
    won = _conditional(lambda: TABLE.update_item(
        Key={"tool_name": tool},
        UpdateExpression="SET half_open_at = :now",
        ConditionExpression="half_open_at = :prev",
        ExpressionAttributeValues={":now": now, ":prev": half_open_at},
    ))
    return {"decision": "allow" if won else "short_circuit", "state": "HALF_OPEN"}


def record_handler(event, _ctx):
    """Called AFTER the tool runs (or fails). Updates the breaker state."""
    tool = event["tool_name"]
    now = int(time.time())

    if event["outcome"] == "success":
        # Success closes the breaker and clears the counter.
        TABLE.update_item(
            Key={"tool_name": tool},
            UpdateExpression="SET #s = :closed, failure_count = :zero REMOVE opened_at, half_open_at",
            ExpressionAttributeNames={"#s": "state"},
            ExpressionAttributeValues={":closed": "CLOSED", ":zero": 0},
        )
        return {"state": "CLOSED", "failure_count": 0}

    # Failure: atomically bump the counter.
    resp = TABLE.update_item(
        Key={"tool_name": tool},
        UpdateExpression="SET failure_count = if_not_exists(failure_count, :zero) + :one",
        ExpressionAttributeValues={":zero": 0, ":one": 1},
        ReturnValues="UPDATED_NEW",
    )
    failures = int(resp["Attributes"]["failure_count"])

    # Trip open at the threshold (or when a HALF_OPEN trial fails). Only stamp opened_at on the
    # transition INTO open, so repeated failures don't keep resetting the cooldown.
    if failures >= FAILURE_THRESHOLD:
        _conditional(lambda: TABLE.update_item(
            Key={"tool_name": tool},
            UpdateExpression="SET #s = :open, opened_at = :now REMOVE half_open_at",
            ConditionExpression="attribute_not_exists(#s) OR #s <> :open",
            ExpressionAttributeNames={"#s": "state"},
            ExpressionAttributeValues={":open": "OPEN", ":now": now},
        ))
        return {"state": "OPEN", "failure_count": failures}

    return {"state": "CLOSED", "failure_count": failures}

aws dynamodb create-table \
  --table-name agent-circuit-breakers \
  --attribute-definitions AttributeName=tool_name,AttributeType=S \
  --key-schema AttributeName=tool_name,KeyType=HASH \
  --billing-mode PAY_PER_REQUEST

import boto3

  cloudwatch = boto3.client("cloudwatch")

  AGENTS = ["market-analysis", "position-sizing", "execution", "compliance"]
  ERROR_THRESHOLD = 5      # > 5 errors in a 60s period => that agent is "erroring"
  CASCADE_THRESHOLD = 2    # >= 2 agents failing at once => cascade


  def build_metrics(agents):
      """Build the metric-math graph: a per-agent 'failing' flag, summed into a count."""
      metrics = []
      breach_ids = []
      for idx, name in enumerate(agents, start=1):
          err_id, beat_id, breach_id = f"e{idx}", f"h{idx}", f"b{idx}"
          breach_ids.append(breach_id)

          # Raw error count for this agent (feeds the expression, not returned)
          metrics.append({
              "Id": err_id,
              "MetricStat": {
                  "Metric": {"Namespace": "AgentPlatform", "MetricName": "ErrorCount",
                             "Dimensions": [{"Name": "AgentName", "Value": name}]},
                  "Period": 60, "Stat": "Sum"},
              "ReturnData": False})

          # Heartbeat the agent emits on a timer. Absent/0 => the agent has gone silent.
          metrics.append({
              "Id": beat_id,
              "MetricStat": {
                  "Metric": {"Namespace": "AgentPlatform", "MetricName": "Heartbeat",
                             "Dimensions": [{"Name": "AgentName", "Value": name}]},
                  "Period": 60, "Stat": "Sum"},
              "ReturnData": False})

          # An agent is "failing" if it is erroring OR it has gone silent.
          # FILL(...,0) turns missing data into 0, so a DEAD agent scores as a failure
          # instead of quietly sitting in INSUFFICIENT_DATA and being ignored.
          metrics.append({
              "Id": breach_id,
              "Expression": f"(FILL({err_id},0) > {ERROR_THRESHOLD}) OR (FILL({beat_id},0) < 1)",
              "Label": f"{name}-failing", "ReturnData": False})

      # cascade = number of agents currently failing. This is the series the alarm watches.
      metrics.append({
          "Id": "cascade",
          "Expression": " + ".join(breach_ids),
          "Label": "AgentsInFailureState", "ReturnData": True})
      return metrics


  cloudwatch.put_metric_alarm(
      AlarmName="cascade-detection-trading-agents",
      AlarmDescription="Fires when 2 or more agents are erroring or silent at the same time.",
      Metrics=build_metrics(AGENTS),
      EvaluationPeriods=1,
      DatapointsToAlarm=1,
      Threshold=CASCADE_THRESHOLD,
      ComparisonOperator="GreaterThanOrEqualToThreshold",
      TreatMissingData="notBreaching",
      AlarmActions=[
          "arn:aws:lambda:us-east-1:123456789012:function:emergency-circuit-breaker"
      ],
  )

import os
import boto3
from botocore.config import Config
from botocore.exceptions import ClientError

# Adaptive retries add client-side rate limiting so a burst of StopExecution calls
# rides out throttling instead of aborting the halt.
_CFG = Config(retries={"max_attempts": 10, "mode": "adaptive"})
sfn = boto3.client("stepfunctions", config=_CFG)
sns = boto3.client("sns")

# Comma-separated list, so "halt all agents" can span more than one workflow.
STATE_MACHINE_ARNS = [a.strip() for a in os.environ["STATE_MACHINE_ARNS"].split(",") if a.strip()]
INCIDENT_TOPIC_ARN = os.environ["INCIDENT_TOPIC_ARN"]
# Safety valve: DRY_RUN=true logs what WOULD be stopped without stopping anything.
DRY_RUN = os.environ.get("DRY_RUN", "false").lower() == "true"

def emergency_circuit_breaker(event, context):
    """Triggered by the cascade alarm. Stop EVERY running execution of the workflow(s),
    then notify. Paginates so it truly reaches all of them, and keeps going if an
    individual stop fails."""
    found = stopped = failed = 0

    paginator = sfn.get_paginator("list_executions")
    for sm_arn in STATE_MACHINE_ARNS:
        for page in paginator.paginate(stateMachineArn=sm_arn, statusFilter="RUNNING"):
            for execution in page["executions"]:
                found += 1
                arn = execution["executionArn"]
                if DRY_RUN:
                    print(f"[DRY_RUN] would stop {arn}")
                    continue
                try:
                    sfn.stop_execution(
                        executionArn=arn,
                        error="CascadeDetected",
                        cause="Cascade alarm triggered - correlated failures across agents",
                    )
                    stopped += 1
                except ClientError as e:
                    # Race (execution just finished) or throttling that exhausted retries.
                    # Log and keep going - one bad stop must not abandon the rest.
                    failed += 1
                    code = e.response["Error"]["Code"]
                    print(f"[ERROR] could not stop {arn}: {code} - {e}")

    result = {"found": found, "stopped": stopped, "failed": failed, "dry_run": DRY_RUN}
    _notify(result)
    return result

def _notify(r):
    if r["dry_run"]:
        subject = f"[DRY RUN] Agent cascade: {r['found']} executions would be halted"
    elif r["failed"]:
        subject = f"AGENT CASCADE - stopped {r['stopped']}/{r['found']}, {r['failed']} FAILED"
    else:
        subject = f"AGENT CASCADE - stopped all {r['stopped']} executions"

    sns.publish(
        TopicArn=INCIDENT_TOPIC_ARN,
        Subject=subject[:100],  # SNS Subject hard limit is 100 chars
        Message=(
            "Cascade alarm fired.\n"
            f"Running executions found: {r['found']}\n"
            f"Successfully stopped:     {r['stopped']}\n"
            f"Failed to stop:           {r['failed']}\n"
            f"Dry run:                  {r['dry_run']}\n\n"
            "Note: StopExecution halts the state machine but does not force-kill a Lambda "
            "task that is already running - those finish or hit their own timeout."
        ),
    )

# In your agent's tool execution layer
import threading
import time


class AgentRateLimiter:
    """External limits for agent tool calls, enforced OUTSIDE the LLM's loop.
    Caps calls-per-minute per tool, and retries per operation. Thread-safe.

    Scope: one process. If the agent runs across several instances, back this with
    shared state (Redis/DynamoDB) so the limit is global instead of per-instance.
    """

    def __init__(self, max_calls_per_minute=10, max_retries=2):
        self.max_calls = max_calls_per_minute
        self.max_retries = max_retries
        self.calls = {}      # tool_name -> recent call timestamps
        self.attempts = {}   # operation_id -> attempts so far
        self._lock = threading.Lock()

    def can_call(self, tool_name: str) -> bool:
        """True if this tool is under its per-minute limit (and records the call)."""
        now = time.monotonic()
        with self._lock:
            recent = [t for t in self.calls.get(tool_name, []) if t > now - 60]
            if len(recent) >= self.max_calls:
                self.calls[tool_name] = recent
                return False
            recent.append(now)
            self.calls[tool_name] = recent
            return True

    def can_retry(self, operation_id: str) -> bool:
        """True while an operation still has retries left. The agent doesn't get to
        decide 'I should retry' - this does. max_retries=2 allows 3 attempts total."""
        with self._lock:
            self.attempts[operation_id] = self.attempts.get(operation_id, 0) + 1
            return self.attempts[operation_id] <= self.max_retries + 1