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:

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:

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:

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:

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:

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:

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:

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:

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:

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 — zero dependencies, works with OpenAI, Gemini, Claude, Ollama, anything:

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)

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 and PyPI.

Filed a feature request 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.

🔐