Scope boundary

Owns externally interruptible, bounded agent control-loop transitions and structured audit evidence for proposed actions, policy decisions, approvals, results, counters, breakers, and containment actions. It does not require raw chain-of-thought.

AID-H-017.003: AID-H-017.001 ensures the orchestrator can inspect, pause, approve, break, or terminate at bounded state transitions; AID-H-017.003 separates non-side-effecting planning from deterministic action selection and dispatch. An interruptible loop can use many planning architectures, while plan-then-execute still needs interruption points.

AID-H-037.001: AID-H-017.001 records typed state, actions, decisions, and outcomes; AID-H-037.001 controls access to confidential hidden reasoning traces and exceptional raw-reasoning storage. Operational auditability should not depend on retaining chain-of-thought.

Implementation guidance

• AID-H-017.001-G001: Implement a step-wise agent executor that yields control for external validation.
• AID-H-017.001-G002: Implement circuit breakers in the agent control loop that detect runaway execution patterns and force the agent into a fail-closed state.
• AID-H-017.001-G003: Emit auditable breaker events that record trigger condition, loop counters, tool-call trace, tenant or session context, and resulting containment action so recursive abuse can be reconstructed during forensics.

AID-H-017.001-G001

Concept:

A monolithic agent that runs its entire control loop in a single, opaque function call is difficult to secure or monitor. A secure architecture breaks the control loop into discrete, observable steps. This allows an external system to inspect the agent plan before execution and potentially interrupt or require approval for high-risk steps.

Implement a step-wise agent executor

Instead of a single run() method, design the executor as a generator or state machine that yields after each plan or tool action. The orchestrator can then insert human approval, policy checks, or risk scoring between steps.

# File: agent_arch/interruptible_agent.py
import hashlib
import json

def action_bytes(action):
    return json.dumps(action, sort_keys=True, separators=(",", ":"),
                      allow_nan=False).encode("utf-8")

class InterruptibleAgentExecutor:
    def __init__(self, agent, execution_broker):
        self.agent = agent
        self.execution_broker = execution_broker

    def run_step(self, inputs):
        # Freeze the exact plan; the broker retrieves its own verified receipt.
        frozen = action_bytes(self.agent.plan(inputs))
        digest = hashlib.sha256(frozen).hexdigest()
        receipt_id = yield {"type": "plan", "action": json.loads(frozen),
                            "action_sha256": digest}
        if not isinstance(receipt_id, str) or not receipt_id.strip():
            raise PermissionError("an execution authorization receipt is required")
        observation = self.execution_broker.execute_approved(
            receipt_id=receipt_id, expected_action_sha256=digest)
        yield {"type": "observation", "result": observation}

# The orchestrator sends a receipt ID with stepper.send(receipt_id).
# Plain next(stepper) cannot execute the plan. The H-018 broker independently
# verifies its stored receipt, current authorization, scope, expiry, one-use
# claim and action digest, then executes the stored action, never this plan.
# Routine actions may receive a policy authorization; high-impact actions
# require H-018.003 approval. A bare human "yes" is not a broker receipt.

Action: Build the main loop as an interruptible execution primitive rather than a monolithic black box. This becomes the control point that all later breaker, policy, and approval checks depend on.

Keep the action safely bounded

Reconcile the complete population of every proposed step and resulting dispatch attempt in each eligible task; reject expired, revoked, or stale policy and asset bindings under the policy-defined freshness window.

Persist a canonical proposal and digest, yield control, require a fresh independently issued decision bound to that exact proposal, and execute only after atomic decision consumption; never let generator resumption imply approval.

Verify safely

A separately credentialed verifier independently replays proposal-to-decision binding, denial, replay, mutation, expiry, crash/restart, and dispatch fixtures from the durable step ledger, reads back effective state from the enforcement system, exercises positive, negative, boundary, stale-evidence, and dependency-failure fixtures, and reconciles every eligible item.

Deploy the existing H-018 execution broker as the executor dependency; its execute_approved adapter must retrieve and verify the action-bound receipt and perform execution-time authorization. This interruption mechanism neither issues approvals nor re-scores the broker. Test ordinary resume, denied approval, wrong-action and expired receipts: all must produce zero tool effects.

AID-H-017.001-G002

Concept:

A circuit breaker is one enforcement control with two inseparable parts: detect that the control loop is no longer behaving safely, then move the agent into a predefined fail-closed state. If you only count iterations without a containment policy, the loop stops ambiguously. If you define safe mode or HITL review without a trip policy, that response never fires. Build both together as one module.

Step 1: Define the trip policy

# File: agent_runtime/circuit_breaker.py
from dataclasses import dataclass
from enum import Enum
from typing import Optional


class ContainmentAction(str, Enum):
    HITL_REVIEW = "hitl_review"
    SAFE_MODE = "safe_mode"
    TERMINATE = "terminate"


@dataclass(frozen=True)
class CircuitBreakerPolicy:
    max_iterations: int
    max_consecutive_tool_calls: int
    max_recursion_depth: int
    max_identical_action_repeats: int
    max_error_retries: int
    max_no_progress_steps: int
    max_cumulative_tokens: int
    max_cumulative_cost_microunits: int

    def __post_init__(self) -> None:
        limits = {
            "max_iterations": self.max_iterations,
            "max_consecutive_tool_calls": self.max_consecutive_tool_calls,
            "max_recursion_depth": self.max_recursion_depth,
            "max_identical_action_repeats": self.max_identical_action_repeats,
            "max_error_retries": self.max_error_retries,
            "max_no_progress_steps": self.max_no_progress_steps,
            "max_cumulative_tokens": self.max_cumulative_tokens,
            "max_cumulative_cost_microunits": self.max_cumulative_cost_microunits,
        }
        invalid = [
            name for name, value in limits.items()
            if isinstance(value, bool) or not isinstance(value, int) or value <= 0
        ]
        if invalid:
            raise ValueError(f"circuit-breaker limits must be positive integers: {invalid}")


@dataclass(frozen=True)
class VerifiedUsageDelta:
    input_tokens: int
    output_tokens: int
    cost_microunits: int

    def __post_init__(self) -> None:
        values = (self.input_tokens, self.output_tokens, self.cost_microunits)
        if any(isinstance(value, bool) or not isinstance(value, int) or value < 0 for value in values):
            raise ValueError("verified usage deltas must be nonnegative integers")

    @property
    def total_tokens(self) -> int:
        return self.input_tokens + self.output_tokens


@dataclass
class CircuitBreakerState:
    iteration_count: int = 0
    consecutive_tool_calls: int = 0
    recursion_depth: int = 0
    consecutive_errors: int = 0
    identical_action_count: int = 0
    no_progress_steps: int = 0
    cumulative_tokens: int = 0
    cumulative_cost_microunits: int = 0
    last_action_fingerprint: Optional[str] = None
    trip_reason: Optional[str] = None


def update_breaker(
    state: CircuitBreakerState,
    policy: CircuitBreakerPolicy,
    action_fingerprint: str,
    made_progress: bool,
    had_error: bool,
    was_tool_call: bool,
    recursion_depth: int,
    verified_usage_delta: VerifiedUsageDelta,
) -> CircuitBreakerState:
    if not action_fingerprint:
        raise ValueError("action_fingerprint must be non-empty")
    if not all(isinstance(value, bool) for value in (made_progress, had_error, was_tool_call)):
        raise TypeError("made_progress, had_error and was_tool_call must be booleans")
    if isinstance(recursion_depth, bool) or not isinstance(recursion_depth, int) or recursion_depth < 0:
        raise ValueError("recursion_depth must be a nonnegative integer")

    if not isinstance(verified_usage_delta, VerifiedUsageDelta):
        raise TypeError("trusted usage-meter delta is required")

    state.iteration_count += 1
    state.consecutive_tool_calls = state.consecutive_tool_calls + 1 if was_tool_call else 0
    state.recursion_depth = recursion_depth
    state.consecutive_errors = state.consecutive_errors + 1 if had_error else 0
    state.no_progress_steps = 0 if made_progress else state.no_progress_steps + 1
    state.cumulative_tokens += verified_usage_delta.total_tokens
    state.cumulative_cost_microunits += verified_usage_delta.cost_microunits

    if action_fingerprint == state.last_action_fingerprint:
        state.identical_action_count += 1
    else:
        state.identical_action_count = 1
    state.last_action_fingerprint = action_fingerprint

    checks = [
        (state.iteration_count > policy.max_iterations, "max_iterations_exceeded"),
        (state.consecutive_tool_calls > policy.max_consecutive_tool_calls, "max_consecutive_tool_calls_exceeded"),
        (state.recursion_depth > policy.max_recursion_depth, "max_recursion_depth_exceeded"),
        (state.identical_action_count > policy.max_identical_action_repeats, "identical_action_repeat_loop"),
        (state.consecutive_errors > policy.max_error_retries, "max_error_retries_exceeded"),
        (state.no_progress_steps > policy.max_no_progress_steps, "no_forward_progress"),
        (state.cumulative_tokens >= policy.max_cumulative_tokens, "token_budget_exhausted"),
        (state.cumulative_cost_microunits >= policy.max_cumulative_cost_microunits, "cost_budget_exhausted"),
    ]

    for condition, reason in checks:
        if condition:
            state.trip_reason = reason
            break

    return state

Step 2: Map trip reasons to fail-closed outcomes

# File: agent_runtime/fail_closed.py
from agent_runtime.circuit_breaker import ContainmentAction


def choose_containment(trip_reason: str, risk_tier: str) -> ContainmentAction:
    if risk_tier == "critical":
        return ContainmentAction.TERMINATE
    if trip_reason in {
        "max_recursion_depth_exceeded", "max_iterations_exceeded",
        "token_budget_exhausted", "cost_budget_exhausted",
    }:
        return ContainmentAction.TERMINATE
    if trip_reason in {"identical_action_repeat_loop", "max_error_retries_exceeded"}:
        return ContainmentAction.HITL_REVIEW
    return ContainmentAction.SAFE_MODE

Step 3: Enforce the breaker inside the orchestrator

# File: agent_runtime/orchestrator.py
from typing import Protocol

from agent_runtime.circuit_breaker import (
    CircuitBreakerPolicy,
    CircuitBreakerState,
    ContainmentAction,
    VerifiedUsageDelta,
    update_breaker,
)
from agent_runtime.fail_closed import choose_containment


class PlannedAction(Protocol):
    advances_goal: bool
    previous_step_failed: bool
    is_tool_call: bool
    recursion_depth: int
    verified_usage_delta: VerifiedUsageDelta

    def fingerprint(self) -> str: ...


class ContainableAgent(Protocol):
    risk_tier: str

    def apply_fail_closed_state(self, containment: ContainmentAction) -> None: ...


def enforce_circuit_breaker(
    *,
    state: CircuitBreakerState,
    policy: CircuitBreakerPolicy,
    planned_action: PlannedAction,
    agent: ContainableAgent,
) -> CircuitBreakerState:
    next_state = update_breaker(
        state=state,
        policy=policy,
        action_fingerprint=planned_action.fingerprint(),
        made_progress=planned_action.advances_goal,
        had_error=planned_action.previous_step_failed,
        was_tool_call=planned_action.is_tool_call,
        recursion_depth=planned_action.recursion_depth,
        verified_usage_delta=planned_action.verified_usage_delta,
    )

    if not next_state.trip_reason:
        return next_state

    containment = choose_containment(next_state.trip_reason, risk_tier=agent.risk_tier)
    agent.apply_fail_closed_state(containment)
    raise RuntimeError(
        f"Circuit breaker tripped: {next_state.trip_reason} -> {containment.value}"
    )

Obtain token and cost deltas from a trusted model gateway or provider billing meter bound to the exact run; never accept a model-supplied counter. If policy requires either meter and its verified delta is missing or invalid, do not dispatch another action. These cumulative values are breaker trip inputs only; AID-H-037.002 remains the owner of reasoning-budget policy and this guidance creates no second budget-control PASS.

Action: Evaluate the breaker before every action, never silently restart after a breaker trip, and require explicit policy or human approval before the agent can leave the fail-closed state. Keep breaker telemetry as a separate sibling control so observability and enforcement remain independently trackable.

Keep the action safely bounded

Reconcile the complete population of all loop steps, retries, recursive entries, tool calls, and progress signals for each eligible run; reject expired, revoked, or stale policy and asset bindings under the policy-defined freshness window.

AID-H-017.001-G003

Concept:

Circuit breaker events are first-class security events, not just operational logs. They must be structured, correlated with the agent session timeline (AID-D-005.004), and include enough context for forensic reconstruction: what the agent was trying to do, what tools it called, how many times, and what triggered the trip. This is how you separate a buggy agent from an adversarially manipulated one.

Structured breaker event schema

# File: agent/breaker_telemetry.py
import json
from datetime import datetime, timezone


def emit_breaker_event(
    agent_id: str,
    session_id: str,
    tenant_id: str,
    trace_id: str,
    trip_reason: str,
    breaker_config: dict,
    breaker_state: dict,
    recent_tool_calls: list[dict],
    containment_action: str,
    evidence_writer,
):
    event = {
        "event_type": "circuit_breaker_trip",
        "ts": datetime.now(timezone.utc).isoformat(),
        "severity": "high",
        "agent_id": agent_id,
        "session_id": session_id,
        "tenant_id": tenant_id,
        "trace_id": trace_id,
        "trip_reason": trip_reason,
        "breaker_config": breaker_config,
        "breaker_state": {
            "iteration_count": breaker_state.get("iteration_count"),
            "consecutive_tool_calls": breaker_state.get("consecutive_tool_calls"),
            "identical_action_count": breaker_state.get("identical_action_count"),
            "consecutive_errors": breaker_state.get("consecutive_errors"),
            "recursion_depth": breaker_state.get("recursion_depth"),
            "cumulative_tokens": breaker_state.get("cumulative_tokens"),
            "cumulative_cost_microunits": breaker_state.get("cumulative_cost_microunits"),
        },
        "recent_tool_calls": recent_tool_calls[-20:],
        "containment_action": containment_action,
    }
    receipt = evidence_writer.append(json.loads(json.dumps(event)))
    if not receipt:
        raise RuntimeError("breaker evidence sink did not acknowledge the event")
    return receipt

Action: Emit a circuit_breaker_trip event every time a breaker fires, include the trace_id that links back to the main agent session log, and keep enough recent tool-call history in the event to reconstruct loop behavior without a second query.

Keep the action safely bounded

Reconcile the complete population of every breaker trip and associated containment transition; reject expired, revoked, or stale policy and asset bindings under the policy-defined freshness window.

Verify safely

A separately credentialed verifier independently replays trip ledger against event-sink records, event schema and hashes, and reads back the containment outcome, reads back effective state from the enforcement system, exercises positive, negative, exact token- and cost-budget boundary, missing-meter, stale-evidence, and dependency-failure fixtures, confirms that no later action is dispatched after exhaustion, and reconciles every eligible item.

Open-source tools

• LangChain
• LlamaIndex
• Microsoft AutoGen
• Temporal
• RabbitMQ
• Valkey
• Redis 8+ (AGPL-3.0)
• Flagsmith
• Unleash
• LangGraph (interrupt and checkpoint substrate; external authorization required)

Commercial / hosted tools

• Temporal Cloud
• Cortex XSOAR
• Splunk SOAR
• LaunchDarkly
• Harness Feature Management & Experimentation (FME)

Threat mappings

MITRE ATLAS

• AML.T0050 Command and Scripting Interpreter
• AML.T0053 AI Agent Tool Invocation (interruptible loops allow inspection before tool invocation)
• AML.T0117 Autonomous Attack-Path Adaptation (bounded state transitions let the orchestrator stop a replanned path before a high-risk side effect)

MAESTRO

• Agent Tool Misuse (L7)

OWASP LLM Top 10 2026

• LLM03:2026 Excessive Agency
• LLM06:2026 Unbounded Consumption

OWASP ML Top 10 2023

• N/A

OWASP Top 10 for Agentic Applications 2026

• ASI02:2026 Tool Misuse and Exploitation
• ASI08:2026 Cascading Failures (interruptible loops break cascade propagation)
• ASI10:2026 Rogue Agents

NIST Adversarial Machine Learning 2025

• NISTAML.039 Compromising connected resources (interruptible loops prevent cascading resource compromise)

Cisco Integrated AI Security and Safety Framework

• AISubtech-13.1.1 Compute Exhaustion
• AISubtech-13.1.2 Memory Flooding (bounds excessive tool-call or delegation growth in agent processing pipelines)
• AISubtech-13.1.3 Model Denial of Service
• AISubtech-13.1.4 Application Denial of Service
• AISubtech-13.1.5 Decision Paralysis Attacks
• AISubtech-13.2.1 Service Misuse for Cost Inflation
• AISubtech-2.1.1 Execution Approval Bypass
• AISubtech-4.3.4 Replay Exploitation
• AITech-12.1 Tool Exploitation
• AITech-13.1 Disruption of Availability
• AITech-13.2 Cost Harvesting / Repurposing
• AITech-2.1 Excessive Agency

Google Secure AI Framework 2.0 - Risks

• RA: Rogue Actions (interruptible loops catch rogue actions before execution)

Databricks AI Security Framework 3.0

• Agents - Core 13.4: Resource Overload (circuit breakers in loops prevent resource exhaustion)
• Agents - Core 13.2: Tool Misuse