AI agent context poisoning: four types of context corruption and how to detect each before it causes a runaway

• Structural isolation: Wrap all tool results in a structured container that signals their origin: <tool_result name="search_web" source="external">...</tool_result>. Instruct the model explicitly that content inside tool_result tags is untrusted external data that may attempt to override instructions. This does not prevent all injection but reduces naive cases by 60–80% in practice.
• Keyword scanning before appending: Before adding a tool result to the context, scan it for high-signal injection patterns: “ignore previous instructions”, “system prompt:”, “new task:”, role-play setup strings. If found, sanitize the result by escaping or stripping the detected pattern and logging an alert. This is not foolproof but catches the most common naive attacks.
• RunGuard budget cap as a secondary layer: Even if a prompt injection succeeds in redirecting the agent, the dollar budget cap limits the maximum cost the injected task can incur. An injection that redirects the agent to make 100 API calls to an exfiltration endpoint still trips the budget cap and halts the run. The guard doesn’t prevent the injection from entering the context; it limits the blast radius of successful injections.

• Structured output with grounding requirements: Require sub-agents to return structured outputs that include a confidence field and a sources list. A response with confidence: "low" or an empty sources list should be treated as unreliable and not appended to the calling agent’s context as fact. Use it as a “best effort” signal only.
• Dual-agent verification for high-stakes results: For facts that downstream decisions depend on, call a second independent sub-agent with the same question. Only append the result to the main context if both sub-agents agree. Disagreement triggers a human-in-the-loop escalation or a task abort. The cost of running two sub-agents is small compared to the cost of acting on a hallucinated fact at scale.
• RunGuard budget cap on sub-agent tool calls: If your orchestrator calls sub-agents as tools, wrap each sub-agent call with a per-call budget cap. A sub-agent that tries to ground its answer by making many downstream calls (to APIs, databases, or further sub-agents) is limited in how much hallucination it can accumulate through its own context before its cap fires. This doesn’t prevent a single-call hallucination, but it prevents a hallucination-chasing loop within the sub-agent from contaminating the calling agent’s context with a deeply-researched, highly-confident wrong answer.

from runguard import guard_async, ContextOverflowError, LoopDetectedError, BudgetExceededError
from openai import AsyncOpenAI
import re

client = AsyncOpenAI()

INJECTION_PATTERNS = [
    r"ignore (?:all |your |previous )?instructions",
    r"new task:",
    r"system prompt:",
    r"you are now in (?:developer|jailbreak|admin)",
    r"disregard (?:all |your |previous )?instructions",
]

def scan_for_injection(content: str) -> str | None:
    """Return the matched pattern string if injection is detected, else None."""
    lower = content.lower()
    for pattern in INJECTION_PATTERNS:
        if re.search(pattern, lower):
            return pattern
    return None

def build_safe_tool_result(tool_name: str, raw_content: str) -> dict:
    """Wrap tool result in structural isolation and scan for injection."""
    injection = scan_for_injection(raw_content)
    if injection:
        # Log and sanitize — don't silently accept the injection
        import logging
        logging.warning(f"Potential injection in {tool_name} result: pattern={injection!r}")
        safe_content = f"[Tool result sanitized: injection pattern detected. Original content suppressed.]"
    else:
        safe_content = raw_content
    return {
        "role": "tool",
        "content": f"<tool_result name=\"{tool_name}\" source=\"external\">{safe_content}</tool_result>",
    }


async def _call_model(messages: list) -> dict:
    response = await client.chat.completions.create(
        model="gpt-4o",
        messages=messages,
    )
    choice = response.choices[0]
    usage = response.usage
    usd = ((usage.prompt_tokens * 2.5 + usage.completion_tokens * 10.0) / 1_000_000) if usage else 0.0
    tool_calls = getattr(choice.message, "tool_calls", None) or []
    sig = tool_calls[0].function.name if tool_calls else "end_turn"
    return {"choice": choice, "usd": usd, "sig": sig}

from gpt_tokenizer import encode as gpt_encode  # type: ignore

guarded = guard_async(
    _call_model,
    budget={"max_usd": 2.0},
    loop={"repeats": 3, "max_cycle_len": 4},
    context={"maxContextTokens": 128_000, "headroom": 8_000},
    tokens=lambda msgs: sum(
        len(gpt_encode(m.get("content", "") or "")) + 4 for m in msgs
    ) + 3,
)


async def run_guarded_agent(system_prompt: str, user_query: str) -> str:
    messages = [
        {"role": "system", "content": system_prompt},
        {"role": "user", "content": user_query},
    ]

    while True:
        try:
            result = await guarded(messages)
        except LoopDetectedError as e:
            return f"Loop detected (pattern: {e.pattern!r}). Context likely contains noise pollution — compact and retry."
        except BudgetExceededError as e:
            return f"Budget cap (${e.spent:.4f}) reached. Run aborted."
        except ContextOverflowError as e:
            return f"Context guard fired at {e.projected_tokens} tokens. Compact context before retrying."

        choice = result["choice"]
        messages.append(choice.message)

        tool_calls = getattr(choice.message, "tool_calls", None) or []
        if not tool_calls:
            return choice.message.content or ""

        for tc in tool_calls:
            raw_result = await execute_tool(tc)
            # Apply injection scan + structural isolation before appending
            safe_msg = build_safe_tool_result(tc.function.name, raw_result)
            safe_msg["tool_call_id"] = tc.id
            messages.append(safe_msg)


async def execute_tool(tc) -> str:
    # Your tool dispatch logic
    ...