How to Build an Agent Pipeline to Prevent Claude Code CLI Cost Spikes and Infinite Loops

Any AI engineer who has deployed LangGraph or AutoGen-based agents to production in an enterprise environment has likely experienced a cold sweat moment: running Claude Code CLI as a background process, only to watch a single code defect cause it to repeatedly call tools on loop—burning through hundreds of dollars in tokens in just 10 minutes.

This isn't a problem you can solve simply by writing better prompts. You need to fix system-level bottlenecks that explode in actual production environments, such as background control failures, infinite node waits, and browser memory crashes. Here are four battle-tested implementation patterns and architectures ready for production deployment.

1. Preventing Token Runaways and Zombie Processes in -p Non-Interactive Mode

When running models like Anthropic's Claude 3 Opus in non-interactive asynchronous mode (using the -p flag), test failures trigger immediate tool re-invocations without user intervention. If an infinite loop occurs here, multi-step reasoning history continually accumulates, consuming massive token volumes with every single turn. This is also when non-terminating processes get left behind in system memory as zombie processes.

To prevent this, you need a wrapper that enforces hard limits at the CLI level and forcefully terminates processes at the script level.

1. Set a per-run spending cap using the --max-budget-usd flag during CLI execution, and restrict tool invocation limits with --max-turns.
2. Restrict allowed tools to read-only operations like Read, Grep, and Glob using the --allowedTools option, and append the --bare flag to strip plugin loading overhead.
3. Attach a Python wrapper class using asyncio.subprocess that kills the process after a specified timeout.

`python
import asyncio
import os
from typing import Any, Dict

class ClaudeCodeWrapper:
def init(self, max_budget_usd: float = 0.50, max_turns: int = 5, timeout_seconds: float = 120.0):
self.max_budget_usd = max_budget_usd
self.max_turns = max_turns
self.timeout_seconds = timeout_seconds

async def execute_validation(self, prompt: str, target_dir: str) -> Dict[str, Any]:
    cmd = [
        "claude", "-p", prompt,
        "--max-budget-usd", str(self.max_budget_usd),
        "--max-turns", str(self.max_turns),
        "--allowedTools", "Read", "Grep", "Glob", "Bash(pytest *)",
        "--add-dir", target_dir,
        "--bare"
    ]
    env = os.environ.copy()
    env["CLAUDE_CODE_SIMPLE"] = "1"

    try:
        process = await asyncio.create_subprocess_exec(
            *cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, cwd=target_dir, env=env
        )
        try:
            stdout_data, stderr_data = await asyncio.wait_for(process.communicate(), timeout=self.timeout_seconds)
        except asyncio.TimeoutError:
            process.kill()
            await process.wait()
            return {"status": "TIMEOUT_EXCEEDED", "exit_code": -1, "error": f"Timeout after {self.timeout_seconds}s"}

        if process.returncode != 0:
            return {"status": "CLI_ERROR", "exit_code": process.returncode, "error": stderr_data.decode("utf-8")}

        return {"status": "SUCCESS", "exit_code": 0, "raw_output": stdout_data.decode("utf-8")}
    except Exception as e:
        return {"status": "WRAPPER_EXCEPTION", "exit_code": -2, "error": str(e)}

`

By implementing this wrapper, costs will not exceed a maximum of $0.50 per validation run, preventing unexpected cost explosions caused by infinite loops.

2. Implementing a Circuit Breaker to Break Infinite Graph Node Retries

LangGraph's default RetryPolicy is designed for network error recovery. When an LLM fails validation due to faulty logic, this default policy fails to handle it properly. Once the retry threshold is crossed, it throws a GraphRecursionError, bringing the entire system to a halt.

You must build a Circuit Breaker inside the State that halts execution and yields control when identical errors repeat continuously.

1. Add a verification_attempts counter and a last_error_signature variable to the State.
2. When entering a second retry after an initial failure, swap the LLM for a lighter model (such as Claude 3.5 Haiku) to reduce routing overhead.
3. If consecutive failures exceed 2 attempts, hand off control to a Human-in-the-loop node to safely route execution to a parent thread.

`python
from typing import TypedDict, Literal
from langgraph.graph import StateGraph, START, END
from langchain_anthropic import ChatAnthropic

class AgentGraphState(TypedDict):
task_prompt: str
verification_attempts: int
last_error_signature: str
verification_status: Literal["PENDING", "PASSED", "FAILED", "CIRCUIT_BROKEN"]

def verification_node(state: AgentGraphState) -> dict:
attempts = state.get("verification_attempts", 0) + 1
model_name = "claude-3-5-haiku-20241022" if attempts >= 2 else "claude-3-opus-20240229"
llm = ChatAnthropic(model=model_name, temperature=0.0)

return {"verification_attempts": attempts, "last_error_signature": "SyntaxError", "verification_status": "FAILED"}

def circuit_breaker_router(state: AgentGraphState) -> str:
if state.get("verification_status") == "PASSED":
return "proceed"
if state.get("verification_attempts", 0) >= 2:
return "trigger_fallback"
return "retry"

def human_in_the_loop_node(state: AgentGraphState) -> dict:
return {"verification_status": "CIRCUIT_BROKEN"}

builder = StateGraph(AgentGraphState)
builder.add_node("verify", verification_node)
builder.add_node("hitl_fallback", human_in_the_loop_node)
builder.add_conditional_edges("verify", circuit_breaker_router, {
"proceed": END, "retry": "verify", "trigger_fallback": "hitl_fallback"
})
builder.add_edge("hitl_fallback", END)
graph_app = builder.compile()

`

By breaking the loop after two consecutive failures and immediately switching to a standby state, you prevent agents from deadlocking and freezing the entire pipeline.

3. Chrome Headless Shell Memory Isolation in CI/CD and Docker

Nodes that capture screenshots or validate DOM structures rely on Chrome Headless Shell. The issue arises when running this code in Docker or GitHub Actions Runner environments, where the Chromium renderer crashes almost immediately. Docker's default shared memory (/dev/shm) is capped at a mere 64MB, leading to Failed to reserve shared memory errors during screen captures.

When spawning browsers in containerized environments, you should apply the following configurations:

1. Increase shared memory in Docker execution using --shm-size=2g, and append the --disable-dev-shm-usage and --no-sandbox flags to Chromium options.
2. Create isolated per-session directories structured as --user-data-dir=/tmp/session_$RUN_ID to avoid cookie or local storage conflicts across runs.
3. Add an artifact storage step in your CI/CD workflow to retain failure logs for post-mortem analysis.

`yaml
name: Agent Background Verification Pipeline
on: [push]

jobs:
headless-browser-validation:
runs-on: ubuntu-latest
container:
image: node:20-buster
options: --shm-size=2g --user root
steps:
- uses: actions/checkout@v4
- name: Install Chrome
run: |
apt-get update && apt-get install -y wget gnupg
wget -q -O - https://dl-ssl.google.com/linux/linux_signing_key.pub | apt-key add -
sh -c 'echo "deb [arch=amd64] http://dl.google.com/linux/chrome/deb/ stable main" >> /etc/apt/sources.list.d/google.list'
apt-get update && apt-get install -y google-chrome-stable --no-install-recommends
- name: Run Agent Verification
env:
ANTHROPIC_API_KEY: ${{ secrets.ANTHROPIC_API_KEY }} SESSION_STORAGE_DIR: "${{ github.workspace }}/artifacts/sessions/${{ github.run_id }}"
run: |
mkdir -p $SESSION_STORAGE_DIR npx ts-node src/cli_validation_entry.ts --user-data-dir="$SESSION_STORAGE_DIR"
- uses: actions/upload-artifact@v4
if: always()
with:
name: validation-artifacts-${{ github.run_id }}
path: ${{ github.workspace }}/artifacts/

`

Expanding shared memory space and isolating sessions ensures browser validation nodes complete reliably without crashing, even in background worker environments.

4. Offloading Log Monitoring to Keep Main State Clean

LangGraph's main State acts as a shared memory pool accessible across all nodes. Dumping raw, multi-megabyte HTML captures or execution logs directly into this State rapidly saturates the context window on subsequent LLM invocations. Database write performance also degrades noticeably.

You should offload bulk data to external storage and maintain only reference paths within the State.

1. Keep only essential metadata in the main State, such as validation_status and an artifact_ref_path for file references.
2. Export multi-megabyte text logs produced by validation nodes as JSON files to local storage or an S3 bucket.
3. Annotate execution flows with LangSmith's @traceable decorator to maintain end-to-end observability.

`python
import json, os, time
from typing import TypedDict, Dict, Any
from langsmith import traceable

class LightMainState(TypedDict):
session_id: str
target_component: str
validation_status: str
artifact_ref_path: str

@traceable(run_type="llm", name="ClaudeCodeValidationSkill", tags=["claude-code-cli", "isolated-node"])
def run_context_free_validation(state: LightMainState) -> Dict[str, Any]:
session_id = state["session_id"]
raw_execution_stdout = "DUMP LOG DATA " * 10000

artifact_dir = f"./artifacts/{session_id}"
os.makedirs(artifact_dir, exist_ok=True)
artifact_full_path = os.path.join(artifact_dir, f"artifact_{int(time.time())}.json")

with open(artifact_full_path, "w", encoding="utf-8") as f:
    json.dump({"session_id": session_id, "stdout": raw_execution_stdout}, f, indent=2)

return {
    "validation_status": "PASSED",
    "artifact_ref_path": artifact_full_path
}

`

This architecture prevents superfluous log data from polluting LLM prompts, and makes troubleshooting significantly easier since developers only need to inspect designated artifact files when issues arise.