---
title: "Building Kill-Switches for Autonomous AI Agents"
description: "When a parent agent spawns four subagents, token consumption multiplies. The Antigravity SDK gives you telemetry and hooks."
url: "https://h3manth.com/scribe/zero-overhead-token-budgets/"
canonical_url: "https://h3manth.com/scribe/zero-overhead-token-budgets/"
markdown_url: "https://h3manth.com/scribe/blog/zero-overhead-token-budgets.md"
publishDate: "2026-06-12T10:00:00.000Z"
author: "Hemanth HM"
category: "ai"
tags: ["ai","agents","antigravity","sdk","subagents","governance"]
word_count: 950
reading_time_minutes: 5
x_markdown_tokens: 2243
content_signal: "ai-train=yes, search=yes, ai-input=yes"
---

# Building Kill-Switches for Autonomous AI Agents

> When a parent agent spawns four subagents, token consumption multiplies. The Antigravity SDK gives you telemetry and hooks.

I spent a week inside the Antigravity SDK for the GDE AI Sprint. The topic was Dynamic Subagents — how agents spawn child agents to parallelize work.

The SDK gives you `define_subagent`, `invoke_subagent`, workspace isolation (`branch`, `share`, `inherit`), and a message-passing protocol. Solid primitives. But one thing was missing: **budget enforcement**.

A parent agent spawning four children doesn't cost 30K tokens. It costs 150K+. Each child gets its own context window, its own reasoning trace, its own token bill. And children can spawn children. A three-level tree is 13 concurrent agents, each burning tokens independently.

```
Parent (orchestrator)          30K
├── Research agent             40K
├── Frontend agent             80K
├── Backend agent              60K
└── Test agent                 50K
                              ─────
                              260K tokens — one feature request
```

Without per-child budgets, one rogue subagent — a research agent re-reading the same 50 files — drains the session before the others finish.

So I built it.

## What the SDK gives you

Every model response carries `usage_metadata`:

```python
response = await agent.chat("Refactor the auth module.")
meta = response.usage_metadata

meta.prompt_token_count       # input tokens
meta.candidates_token_count   # output tokens
meta.thinking_token_count     # reasoning tokens (the hidden cost)
meta.total_token_count        # sum
```

`thinking_token_count` is the one that matters. In extended reasoning models, this is 3-5x the visible output. A turn that produces 1,200 tokens of code might burn 8,000 tokens of reasoning. Multiply that across a subagent tree.

The SDK also has lifecycle hooks:

```python
from google.antigravity.hooks import hooks

class MyHook(hooks.AfterModelHook):     # Inspect: runs after model call
    async def run(self, context, data): ...

class MyGate(hooks.BeforeModelHook):    # Decide: runs before, can block
    async def run(self, context, data):
        return hooks.HookResult(allow=False, reason="nope")
```

**Inspect** hooks are read-only. **Decide** hooks can block. That's the entire surface area you need for budget enforcement.

State lives in `context.session.state` — a dict scoped to the agent session. Each subagent gets its own session. Parent can't see children's state. That's the isolation that makes per-agent budgets possible.

## The kill-switch

Two hooks. An accumulator and a gate:

```python
class TokenAccumulator(hooks.AfterModelHook):
    """Count tokens after each model call."""
    async def run(self, context, data):
        usage = data.usage_metadata
        state = context.session.state
        state["total_tokens"] = state.get("total_tokens", 0) + usage.total_token_count
        state["turns"] = state.get("turns", 0) + 1

class BudgetGate(hooks.BeforeModelHook):
    """Block if budget exceeded."""
    def __init__(self, max_tokens: int):
        self.max_tokens = max_tokens

    async def run(self, context, data):
        total = context.session.state.get("total_tokens", 0)
        if total >= self.max_tokens:
            return hooks.HookResult(
                allow=False,
                reason=f"Budget exceeded: {total:,} / {self.max_tokens:,} tokens"
            )
        return hooks.HookResult(allow=True)
```

Wire them into a subagent config:

```python
from google.antigravity import LocalAgentConfig

config = LocalAgentConfig(
    hooks=[TokenAccumulator(), BudgetGate(500_000)]
)
```

That's it. Two hooks. If the agent hits 500K mid-task, `BudgetGate` returns `allow=False` and the session stops.

## Adding circuit breakers

Budget ceilings catch runaway spending. But you also want to catch degenerate behavior early — before the ceiling hits.

**Velocity spikes:**

```python
class VelocityMonitor(hooks.AfterModelHook):
    async def run(self, context, data):
        state = context.session.state
        history = state.setdefault("turn_history", [])
        history.append(data.usage_metadata.total_token_count)

        if len(history) >= 5:
            recent_avg = sum(history[-5:]) / 5
            baseline_avg = sum(history[:-5]) / max(len(history) - 5, 1)
            if recent_avg > baseline_avg * 2.5:
                state["velocity_warning"] = True
```

If a subagent was averaging 15K/turn and jumps to 45K, it's probably re-reading the codebase in a loop.

**Degenerate loops:**

```python
class LoopDetector(hooks.PostToolCallHook):
    async def run(self, context, data):
        state = context.session.state
        log = state.setdefault("tool_log", [])
        log.append(f"{data.tool_call.name}:{data.tool_call.args}")

        if len(log) >= 3 and len(set(log[-3:])) == 1:
            state["loop_detected"] = True
```

Same tool, same args, three times in a row. The agent is stuck.

**Reasoning runaway:**

```python
class ReasoningGuard(hooks.AfterModelHook):
    async def run(self, context, data):
        meta = data.usage_metadata
        if meta.total_token_count == 0:
            return
        ratio = meta.thinking_token_count / meta.total_token_count
        state = context.session.state
        streak = state.get("reasoning_streak", 0)

        if ratio > 0.8:
            state["reasoning_streak"] = streak + 1
            if streak + 1 >= 3:
                state["reasoning_runaway"] = True
        else:
            state["reasoning_streak"] = 0
```

An agent spending 40K thinking to produce 500 output is going in circles.

## Composing it

One Decide hook reads all the signals:

```python
class KillSwitch(hooks.BeforeModelHook):
    def __init__(self, max_tokens: int):
        self.max_tokens = max_tokens

    async def run(self, context, data):
        state = context.session.state
        total = state.get("total_tokens", 0)

        if total >= self.max_tokens:
            return hooks.HookResult(allow=False, reason=f"Token budget: {total:,}/{self.max_tokens:,}")
        if state.get("loop_detected"):
            return hooks.HookResult(allow=False, reason="Degenerate tool loop")
        if state.get("reasoning_runaway"):
            return hooks.HookResult(allow=False, reason="Reasoning runaway")
        return hooks.HookResult(allow=True)
```

Wire everything per subagent:

```python
def governed_agent(role, max_tokens):
    return LocalAgentConfig(
        system_prompt=f"You are a {role} agent.",
        hooks=[
            TokenAccumulator(),
            VelocityMonitor(),
            LoopDetector(),
            ReasoningGuard(),
            KillSwitch(max_tokens),
        ]
    )

# Each child gets its own budget
research_config = governed_agent("research", 50_000)
frontend_config = governed_agent("frontend", 100_000)
backend_config  = governed_agent("backend", 80_000)
```

Five hooks per agent. Inspect hooks accumulate, Decide hook blocks. The SDK handles graceful shutdown.

## Extracting it

The patterns aren't SDK-specific. Token counting, budget ceilings, velocity detection, loop detection — these work with any LLM provider.

So I extracted them into [token-limiter](https://github.com/hemanth/token-limiter) — zero dependencies, works with OpenAI, Gemini, Claude, Ollama, anything:

```python
from token_limiter import token_limiter, from_openai

budget = token_limiter(max_tokens=500_000, max_cost=5.00)
budget.record(from_openai(response))

if not budget.ok:
    print(budget.reason)
```

```js
import { tokenLimiter, fromGemini } from 'token-limiter';

const budget = tokenLimiter({ maxTokens: 500_000, maxCost: 5.00 });
budget.record(fromGemini(response));

if (!budget.ok) console.log(budget.reason);
```

Available on [npm](https://www.npmjs.com/package/token-limiter) and [PyPI](https://pypi.org/project/token-limiter/).

Filed a [feature request](https://github.com/google-antigravity/antigravity-sdk-python/issues/59) on the Antigravity SDK proposing built-in budget hooks following the existing `deny()` / `allow()` pattern. Happy to do a PR if they open contributions.

## What I learned

The SDK's hook system is well-designed. `usage_metadata` + `AfterModelHook` + `BeforeModelHook` + `SessionContext.state` gives you everything you need to build governance. The fact that each subagent gets its own session state is the key — it means per-agent budgets work without any plumbing.

What's missing is the policy layer. `deny()` controls what an agent can do. Budget hooks would control how much. Together they're the governance stack that makes autonomous subagent trees safe for production.

The agents that make it to production won't be the most capable. They'll be the most governable.

🔐