Building Self-Healing Antigravity Agents — Detection, Diagnosis, and Recovery in Production

Layer 1: Health signals — what to measure, when to call it abnormal

Detection starts with picking the right signals. The four I currently track:

1. Tool-call success rate over the last 10 minutes — below 80% triggers warning, below 50% triggers critical
2. P95 response latency over the last 10 minutes — three times the baseline triggers warning
3. Empty-response rate over the last 100 calls — the LLM returning "" or null. Above 5% triggers warning
4. Task completion rate over the last hour — proportion of tasks the agent marked "done". Below 70% triggers critical

Numbers 3 and 4 are the ones traditional monitoring misses. Conventional server monitoring assumes "no errors = healthy", but agents fail silently all the time. Without watching task completion, you won't notice that "no errors" actually means "no answers".

Here's the AgentKit 2.0–compatible signal collector I run today. Paste-able if you have Redis available:

# requirements: antigravity-agentkit>=2.0.0, redis>=5.0.0
# env vars: REDIS_URL (e.g. redis://localhost:6379/0)
 
import os
import time
import uuid
from typing import Literal
 
import redis
 
HealthStatus = Literal["healthy", "warning", "critical"]
 
class HealthSignals:
    """Aggregates health signals into Redis and exposes a consolidated state.
 
    Why Redis instead of an in-memory deque:
    - Multi-process agents share state across workers
    - The agent process can crash without losing the recent window
    A small in-memory implementation works fine for single-process setups.
    """
 
    def __init__(self, agent_id: str, redis_url: str | None = None):
        self.agent_id = agent_id
        self.r = redis.from_url(redis_url or os.environ["REDIS_URL"], decode_responses=True)
        self.window_seconds = 600  # rolling 10-minute window
        self.k_calls = f"agent:{agent_id}:tool_calls"
        self.k_empty = f"agent:{agent_id}:empty_responses"
        self.k_base = f"agent:{agent_id}:baseline_p95_ms"
 
    def record_tool_call(self, tool: str, success: bool, latency_ms: float) -> None:
        ts = time.time()
        member = f"{uuid.uuid4().hex}:{int(success)}:{latency_ms:.1f}:{tool}"  # uuid keeps two calls in the same millisecond from collapsing into one ZSET member
        self.r.zadd(self.k_calls, {member: ts})
        self.r.zremrangebyscore(self.k_calls, 0, ts - self.window_seconds)
 
    def record_empty_response(self, was_empty: bool) -> None:
        self.r.lpush(self.k_empty, "1" if was_empty else "0")
        self.r.ltrim(self.k_empty, 0, 99)
 
    def set_baseline_p95(self, ms: float) -> None:
        self.r.set(self.k_base, ms)
 
    def status(self) -> HealthStatus:
        success_rate = self._success_rate()
        p95 = self._p95_latency()
        empty_rate = self._empty_rate()
 
        if success_rate < 0.5 or empty_rate > 0.20:
            return "critical"
        if success_rate < 0.8 or p95 > self._baseline_p95() * 3 or empty_rate > 0.05:
            return "warning"
        return "healthy"
 
    def _window(self) -> list[tuple[bool, float]]:
        now = time.time()
        rows = self.r.zrangebyscore(self.k_calls, now - self.window_seconds, now)
        out = []
        for m in rows:
            _, ok, lat, _tool = m.split(":", 3)  # split only the first three ':' so tool names containing ':' survive
            out.append((ok == "1", float(lat)))
        return out
 
    def _success_rate(self) -> float:
        rows = self._window()
        if not rows:
            return 1.0  # no data yet: do not call it unhealthy
        return sum(1 for ok, _ in rows if ok) / len(rows)
 
    def _p95_latency(self) -> float:
        lats = sorted(lat for _, lat in self._window())
        if not lats:
            return 0.0
        return lats[min(len(lats) - 1, int(len(lats) * 0.95))]
 
    def _baseline_p95(self) -> float:
        v = self.r.get(self.k_base)
        return float(v) if v else 2000.0  # placeholder (ms) until a baseline is registered via set_baseline_p95()
 
    def _empty_rate(self) -> float:
        vals = self.r.lrange(self.k_empty, 0, 99)
        if not vals:
            return 0.0
        return sum(1 for v in vals if v == "1") / len(vals)

The expected behaviour: while the agent runs normally, status() returns "healthy". The moment an external API starts returning rate-limit errors, it shifts to "warning". With sustained tool failures, it becomes "critical".

The interesting design choice is when to read this. I read it at two points: at task start, and immediately before each tool call. If the agent enters a task in critical state, I refuse new write operations and switch to read-only mode until things settle.

Layer 2: Deterministic diagnosis — pattern-match before you reason

Once detection screams, you need to classify what's happening. Don't ask the LLM to do this. There are three reasons:

• The LLM itself may be the unhealthy dependency
• Inference cost balloons exactly when you most need to control cost
• Non-deterministic diagnosis makes the system impossible to improve

I use a deterministic diagnoser instead — regex patterns plus external status-page lookups:

# Diagnosis layer: maps error patterns + external service state to a recovery strategy
import re
import httpx
from dataclasses import dataclass
from enum import Enum
 
class FailurePattern(Enum):
    RATE_LIMIT = "rate_limit"
    SERVICE_DOWN = "service_down"
    TIMEOUT = "timeout"
    EMPTY_LLM_RESPONSE = "empty_llm"
    JSON_PARSE_ERROR = "json_parse"
    UNKNOWN = "unknown"
 
@dataclass
class Diagnosis:
    pattern: FailurePattern
    confidence: float  # 0.0 to 1.0
    recovery_strategy: str
    cooldown_seconds: int
 
class Diagnoser:
    # tool name -> JSON status API URL of the dependency (Atlassian Statuspage style {"status": {"indicator": "none"}} assumed)
    STATUS_API: dict[str, str] = {}
 
    RATE_LIMIT_PATTERNS = [
        re.compile(r"rate.?limit", re.I),
        re.compile(r"too many requests", re.I),
        re.compile(r"429", re.I),
    ]
    TIMEOUT_PATTERNS = [
        re.compile(r"timeout", re.I),
        re.compile(r"deadline exceeded", re.I),
    ]
 
    def diagnose(self, error_message: str, tool_name: str) -> Diagnosis:
        if any(p.search(error_message) for p in self.RATE_LIMIT_PATTERNS):
            return Diagnosis(
                pattern=FailurePattern.RATE_LIMIT,
                confidence=0.95,
                recovery_strategy="exponential_backoff_with_jitter",
                cooldown_seconds=60,
            )
        if any(p.search(error_message) for p in self.TIMEOUT_PATTERNS):
            if self._service_is_degraded(tool_name):
                return Diagnosis(
                    pattern=FailurePattern.SERVICE_DOWN,
                    confidence=0.85,
                    recovery_strategy="failover_to_secondary",
                    cooldown_seconds=300,
                )
            return Diagnosis(
                pattern=FailurePattern.TIMEOUT,
                confidence=0.70,
                recovery_strategy="retry_with_longer_timeout",
                cooldown_seconds=10,
            )
        # No pattern matched — escalate
        return Diagnosis(
            pattern=FailurePattern.UNKNOWN,
            confidence=0.0,
            recovery_strategy="escalate_to_human",
            cooldown_seconds=0,
        )
 
    def _service_is_degraded(self, tool_name: str) -> bool:
        """Ask the dependency's status API. Only URLs registered in STATUS_API are consulted.
        Status page formats differ per vendor, so check the real JSON before registering one."""
        url = self.STATUS_API.get(tool_name)
        if not url:
            return False  # nothing registered: no evidence either way
        try:
            r = httpx.get(url, timeout=2.0)
            return r.json().get("status", {}).get("indicator", "none") != "none"
        except Exception:
            return False  # treat as healthy when the status page itself is unreachable

The output of diagnose() is a recovery_strategy string consumed by the next layer.

You might wonder: why regex and status-page checks instead of letting the LLM classify? Because diagnosis should complete in a few hundred milliseconds, for fractions of a cent, and produce the same answer every time. Those are exactly the properties LLMs are bad at when they're already unhealthy. I keep LLM-based reasoning for the final UNKNOWN escalation only.

Layer 3: Recovery strategies — graduated fallback

The recovery layer executes whichever strategy diagnosis chose. I currently support six:

• exponential_backoff_with_jitter — for rate limits. 1s, 2s, 4s, 8s ... with up to 30% random jitter
• failover_to_secondary — switch providers (Anthropic → Gemini, or vice versa)
• retry_with_longer_timeout — for slow but not dead services. Doubles the timeout, retries up to 3x
• degrade_to_readonly — stop writing, keep answering read-only queries
• circuit_break — stop calling for 15 minutes, then half-open a single probe
• escalate_to_human — Slack me, agent stays silent

The non-negotiable is that every strategy must be idempotent and side-effect-safe. If recovery fires while the agent is creating a Stripe invoice, you must not double-charge the user. I require an idempotency_key on every write tool call.

import asyncio
import os
import random
from typing import Any, Callable, Awaitable
 
import httpx
 
class RecoveryExecutor:
    def __init__(self, primary_provider: str = "anthropic"):
        self.primary = primary_provider
        self.secondary = "gemini" if primary_provider == "anthropic" else "anthropic"
 
    async def execute(
        self,
        strategy: str,
        operation: Callable[..., Awaitable[Any]],
        *args, **kwargs
    ) -> Any:
        if strategy == "exponential_backoff_with_jitter":
            return await self._exp_backoff(operation, *args, **kwargs)
        if strategy == "failover_to_secondary":
            kwargs["provider"] = self.secondary
            return await operation(*args, **kwargs)
        if strategy == "retry_with_longer_timeout":
            kwargs["timeout"] = kwargs.get("timeout", 30) * 2
            return await operation(*args, **kwargs)
        if strategy == "degrade_to_readonly":
            kwargs["mode"] = "readonly"
            return await operation(*args, **kwargs)
        if strategy == "circuit_break":
            raise CircuitBreakerOpen("Circuit breaker open. Probe again in 15 minutes.")
        if strategy == "escalate_to_human":
            await self._notify_slack(operation.__name__, args, kwargs)
            raise HumanEscalationRequired("Manual intervention required.")
        raise ValueError(f"Unknown recovery strategy: {strategy}")
 
    async def _notify_slack(self, name: str, args: tuple, kwargs: dict) -> None:
        """Post a hand-off notice to a Slack Incoming Webhook. URL comes from SLACK_WEBHOOK_URL."""
        url = os.environ.get("SLACK_WEBHOOK_URL")
        if not url:
            return
        safe = {k: v for k, v in kwargs.items() if k not in ("api_key", "token")}  # never post secrets
        async with httpx.AsyncClient(timeout=5.0) as client:
            await client.post(url, json={"text": f"[self-healing] handing {name} to a human: {safe}"})
 
    async def _exp_backoff(
        self,
        operation: Callable[..., Awaitable[Any]],
        *args,
        max_attempts: int = 5,
        **kwargs,
    ) -> Any:
        for attempt in range(max_attempts):
            try:
                return await operation(*args, **kwargs)
            except Exception:
                if attempt == max_attempts - 1:
                    raise
                base = 2 ** attempt  # 1, 2, 4, 8 s (no sleep after the last attempt)
                jitter = base * random.uniform(0, 0.3)
                await asyncio.sleep(base + jitter)
 
class CircuitBreakerOpen(Exception):
    pass
 
class HumanEscalationRequired(Exception):
    pass
 
class HealthCheckFailed(Exception):
    pass

Expected runtime behaviour: rate-limit error → wait 1s → 2s → 4s → 8s → give up on the fifth failure. Service-down → instant failover to the secondary provider. Sustained failure across both → circuit breaker opens, half-opens 15 minutes later for a single probe.

Retrying writes safely — a timeout is not a "failure"

My first design treated a timeout as just another failure. But a timeout only tells you that no answer came back; it says nothing about whether the other side already did the work. Re-running a read is harmless. Re-running a write sits one step away from doing it twice.

A timeout is not a failure — it is an unknown outcome. Once you accept that, each strategy gets a clear answer to "may I use this on a write?"

If the API accepts idempotency keys — Stripe does, via the Idempotency-Key header — build the key from the task ID plus the step number, and a resent write will only be executed once. A random key per attempt defeats the whole protection, so reuse the same value on every retry.

One more thing: don't hard-code the wait. 429 and 503 responses may carry a Retry-After header, either in seconds or as an HTTP date, telling you when the server would like you back (RFC 9110). Compare it with your backoff delay and take the longer of the two.

from datetime import datetime, timezone
from email.utils import parsedate_to_datetime
 
def retry_after_seconds(headers: dict[str, str], default: float) -> float:
    """Turn Retry-After (seconds or HTTP date) into seconds. Falls back to default."""
    raw = headers.get("Retry-After")
    if not raw:
        return default
    if raw.isdigit():
        return float(raw)
    try:
        when = parsedate_to_datetime(raw)
        return max(0.0, (when - datetime.now(timezone.utc)).total_seconds())
    except (TypeError, ValueError):
        return default
 
# Inside _exp_backoff (assumes an HTTP client whose exception exposes headers):
# wait = max(base + jitter, retry_after_seconds(headers, 0.0))
# if wait > 120:           # a wait beyond the cap goes to a human instead
#     raise HumanEscalationRequired("Retry-After is too long")

The 120-second cap is my own rule of thumb. If you decide how long your task can afford to wait before you need it, a long Retry-After stops being a dilemma and becomes a clean hand-off to escalation.

Wiring it up: the AgentKit 2.0 middleware

Now we bind the three layers into a single middleware that sits around every tool call. The method and attribute names on ToolMiddleware and ToolCallContext below are a sketch of the shape, not a documented API — check which hooks your AgentKit version actually exposes and rename accordingly:

from typing import Any
 
from antigravity.agentkit import Agent, ToolMiddleware, ToolCallContext
 
class SelfHealingMiddleware(ToolMiddleware):
    def __init__(
        self,
        health_signals: HealthSignals,
        diagnoser: Diagnoser,
        executor: RecoveryExecutor,
    ):
        self.health = health_signals
        self.diag = diagnoser
        self.exec = executor
 
    async def before_tool_call(self, ctx: ToolCallContext) -> None:
        status = self.health.status()
        if status == "critical":
            if not ctx.tool.is_read_only:
                raise HealthCheckFailed(
                    f"Agent is critical. Write tool {ctx.tool.name} blocked."
                )
 
    async def after_tool_call(self, ctx: ToolCallContext, result: Any) -> None:
        """Record successes too. If only failures are logged, the success rate is skewed and always looks low."""
        self.health.record_tool_call(tool=ctx.tool.name, success=True, latency_ms=ctx.elapsed_ms)
        self.health.record_empty_response(result in ("", None))
 
    async def on_tool_error(self, ctx: ToolCallContext, error: Exception) -> Any:
        self.health.record_tool_call(
            tool=ctx.tool.name,
            success=False,
            latency_ms=ctx.elapsed_ms,
        )
        diagnosis = self.diag.diagnose(str(error), ctx.tool.name)
        if diagnosis.confidence < 0.5:
            raise error
        return await self.exec.execute(
            diagnosis.recovery_strategy,
            ctx.tool.invoke,
            **ctx.tool_args,
        )
 
agent = Agent(
    name="payment-support-agent",
    model="claude-sonnet-4-5",
    tools=[stripe_tools, slack_tools],
    middleware=[
        SelfHealingMiddleware(
            health_signals=HealthSignals(agent_id="payment-support"),
            diagnoser=Diagnoser(),
            executor=RecoveryExecutor(primary_provider="anthropic"),
        ),
    ],
)

Replaying the original Stripe rate-limit incident with this middleware in place: the agent doesn't go silent. It enters exponential_backoff_with_jitter, eventually fails over to Gemini if Stripe stays unhealthy, and only pages me when the circuit breaker has been open long enough that human attention is warranted.

Since adding it, the number of nights I get woken by Slack has dropped noticeably.

Tracing recoveries so the agent gets smarter over time

The system above survives outages — but it stops there unless you record what happened. I emit a structured trace event for every recovery, with:

• timestamp
• failed tool name
• raw error message
• diagnosis result (pattern + confidence)
• recovery strategy executed
• recovery duration
• final outcome (success / escalation)

Reviewing this weekly is where the real improvement happens. Tools with a high UNKNOWN rate are candidates for new regex patterns. Strategies that consistently fail to recover within their cooldown window are candidates for tuning.

I pipe these traces into Looker Studio and watch failure rate, recovery success, and mean recovery time per tool. Operating an agent fleet starts to feel a lot like SRE error-budget thinking. For more on the SLO side, see Designing SRE-style SLOs and error budgets for Antigravity AI agents.

A scenario: how the design absorbs a partial upstream outage

I would love to print a real incident record here, but I'd rather not invent numbers or dates, so this is a scenario you can reproduce with failure injection: about 60% of requests time out and the rest succeed, slowly. A partial outage is the hardest kind to judge because it never fully stops. Picture it hitting the code-review agent.

The row to look at is the one where the status page still says everything is fine. Status pages can lag behind the actual outage, which is why the diagnoser doesn't rely on an external declaration alone: a run of local timeouts is enough evidence to switch paths.

A failure that keeps half-working takes longer to reach a human than one that stops cleanly. Re-measure the timings above with failure injection in your own environment, and write the measured numbers into your runbook.

Hooking into existing observability stacks

Most teams already have Datadog, New Relic, Honeycomb, or Grafana. Don't replace them — extend them. The recovery traces are valuable precisely because they slot into what your team already watches.

The key spans I emit per recovery event:

• agent.tool_call — top-level span for every tool invocation
• agent.tool_call.error — child span when the call fails, with the raw error and tool metadata
• agent.diagnosis — child span carrying the pattern, confidence, and chosen strategy
• agent.recovery — child span for the actual recovery execution, with attempt count, total wait time, and final outcome

In OpenTelemetry-compatible format, this lets ops engineers filter incidents by recovery strategy, count escalate_to_human events per service, and correlate agent behaviour with upstream provider outages. I keep a Grafana dashboard with three panels: "incidents per hour by tool", "recovery strategies invoked (stacked)", and "MTTR distribution". When something looks wrong, it usually shows up here first.

Honeycomb in particular is a good fit because the high cardinality of tool_name and recovery_strategy plays to its strengths. Slicing by recovery_strategy:circuit_break for the last 24 hours is one click. Doing the same in a metric-only system requires pre-aggregation that you probably did not set up in advance.

A note on cost during recovery

Self-healing is not free. Every retry costs tokens, every failover routes to a different (potentially more expensive) provider, and every unnecessary circuit-break leaves user requests unanswered. The instinct of "just retry harder" is the failure mode I see most often when teams adopt this pattern.

I track three cost-side metrics alongside the reliability ones:

• Cost per incident — total tokens (and dollars) spent during recovery for each failed task. If this spikes, your strategies are too aggressive
• Wasted retries — calls that succeeded after retry but where the original error was a real bug, not a transient. These should approach zero. If they don't, you're masking real bugs
• Failover cost differential — how much more expensive your secondary provider is. If you fail over often to a 2x-priced model, the math may argue for fixing your primary instead

A practical guardrail: set a per-task recovery budget. If recovery has already consumed, say, 4x the original task's expected cost, give up and escalate. Without this, a flapping dependency can run up a serious bill before anyone notices.

# Budgeted executor — caps how much cash we spend recovering from a single task
class BudgetedExecutor(RecoveryExecutor):
    def __init__(self, primary_provider: str, max_recovery_cost_usd: float = 0.10):
        super().__init__(primary_provider)
        self.max_cost = max_recovery_cost_usd
 
    async def execute(self, strategy, operation, *args, **kwargs):
        spent = kwargs.pop("_recovery_spent_usd", 0.0)
        if spent >= self.max_cost:
            await self._notify_slack(
                f"Recovery budget exhausted: ${spent:.4f} / ${self.max_cost}",
                args, kwargs,
            )
            raise HumanEscalationRequired("Recovery budget exhausted.")
        return await super().execute(strategy, operation, *args, **kwargs)

This kind of budget is the difference between "self-healing" and "self-bankrupting".

Testing self-healing logic without breaking production

The most awkward part of this work is testing it. You cannot easily reproduce a Stripe rate-limit on a sandbox key, and "wait for it to happen in production" is exactly the wrong feedback loop.

What I do instead is simulate failures at the middleware boundary. I have a FailureInjector that I enable only in staging — and a small set of integration tests that walk through every diagnosis path:

# Failure injection harness for testing recovery paths
import pytest
import asyncio
 
class FailureInjector:
    # Wraps any tool to inject specific failure modes during testing
    def __init__(self, tool, failure_mode=None, fail_count=1):
        self.tool = tool
        self.failure_mode = failure_mode
        self.fail_count = fail_count
        self.calls = 0
 
    async def invoke(self, *args, **kwargs):
        self.calls += 1
        if self.failure_mode and self.calls <= self.fail_count:
            if self.failure_mode == "rate_limit":
                raise Exception("429: rate limit exceeded")
            if self.failure_mode == "timeout":
                raise asyncio.TimeoutError("deadline exceeded")
            if self.failure_mode == "empty":
                return ""
        return await self.tool.invoke(*args, **kwargs)
 
class FakeInvoiceTool:
    async def invoke(self, *args, **kwargs):
        return {"ok": True}
 
@pytest.mark.asyncio
async def test_rate_limit_recovers_via_backoff(monkeypatch):
    async def no_sleep(_seconds):  # keep the test fast
        return None
    monkeypatch.setattr(asyncio, "sleep", no_sleep)
 
    tool = FailureInjector(FakeInvoiceTool(), failure_mode="rate_limit", fail_count=2)
    executor = RecoveryExecutor()
    result = await executor.execute("exponential_backoff_with_jitter", tool.invoke, customer_id="cus_123")
    assert result is not None
    assert tool.calls == 3  # failed twice, succeeded on third attempt

What I gained from this discipline: confidence to ship recovery changes without staging an actual rate-limit incident. What I lost: a few hours building the harness. The trade was very much worth it.

A subtler benefit is that writing the failure-injection tests forces you to enumerate failure modes you had not thought about. The first time I wrote tests for empty-LLM responses, I realised my code did not even have a code path for that — it just propagated None and broke downstream. That bug never reached production because the test made it visible.

Choosing what to monitor first

If you are starting fresh, do not try to wire up all four health signals on day one. Start with the one that maps most cleanly to your agent's purpose:

• For agents that produce content (reports, summaries, code), watch empty-response rate first. Quality regressions hide there
• For agents that orchestrate workflows (multi-step tasks, long-running jobs), watch task completion rate first. Silent stalls hide there
• For agents that integrate with external APIs (Stripe, Slack, GitHub), watch tool-call success rate first. Rate limits and outages hide there
• For agents users interact with synchronously, watch P95 latency first. The user experience hides there

Pick one signal, instrument it cleanly, and run for two weeks before adding the next. Adding all four at once means you will tune none of them well — and tuning is where the value comes from. The threshold values I cited (80%, 50%, 5%, 70%) work for my workload, not yours. Watch the actual distribution before fixing thresholds.

The same principle applies to recovery strategies. Implement exponential_backoff_with_jitter first, since it covers more than half of all incidents. Add failover_to_secondary second once you have a working secondary integration. Defer circuit_break until you have actual traces showing repeated failure cascades; without that data, you will set the wrong cooldown windows and either trip too eagerly or never trip at all.

When NOT to add self-healing

I want to argue against self-healing for a moment, because the worst case is to layer all of this on an agent that does not need it.

You do not need this design when:

• The agent runs on demand, with a human watching the result. If you are using Antigravity interactively for code edits, the human is the recovery loop
• The agent's failure mode is a bad answer rather than no answer. Self-healing will not help quality issues — that is an evaluation problem
• The agent's tasks are cheap to retry from scratch. If the worst case is "redo the entire task", retry-from-scratch is simpler than mid-task recovery
• You have not run the agent in production long enough to know what actually breaks. The biggest mistake is building self-healing in advance against imagined failures. Build it after you have observed real ones for at least 30 days

Self-healing is a maintenance burden. The diagnosis layer needs updating when error formats drift. The recovery code needs care when adding new providers. If your agent does not need 24/7 reliability, simpler designs win.

The clearest signal that you do need it: someone is being paged outside business hours.

Three traps I walked into in production

A clean design diagram doesn't survive contact with production. Here are the three traps that bit me:

Trap 1: the health check made the agent slower

My first version queried Redis on every tool call. A few milliseconds per call sounds fine until you realise complex agent loops make dozens of tool calls. The overhead added up to something users could feel. The fix: cache the health status for 5 seconds. A 5-second delay in transitioning to critical is acceptable; a 500ms tax on every successful loop is not.

Trap 2: failover targets had cold-start latency

The first time we failed over from Anthropic to Gemini, the Gemini side hadn't been hit in hours. The first call was visibly slower, thanks to reconnecting from scratch. Users complained that "the failover is slower than the failure". The fix: send a small warm-up probe to both providers every 15 minutes so neither connection goes cold.

Trap 3: my regexes stopped matching when a dependency reworded its errors

When a vendor rewords an error, pattern matching misses silently. Say "Quota exceeded" becomes "Resource has been exhausted": my rate-limit regex no longer matches, and everything lands in UNKNOWN. The fix: measure the diagnoser itself. I alert when the share of UNKNOWN diagnoses crosses 5% of recent failures — that catches drifted wording before users do.

How to know it's working

If you can't measure the change, this is just over-engineering. Four metrics actually move:

• MTTR — time from first error to recovered response. Compare the manual-intervention era with the automatic one, using the same definition from the same traces
• Auto-recovery rate — share of incidents that resolved without escalate_to_human. Start with 90% as a target
• False-positive rate — circuit breakers tripped on healthy traffic. Aim for one a week or fewer
• Pages outside business hours — the human cost. Put the before and after side by side

That last one is the one I actually care about. If your night-time Slack notifications aren't trending toward zero, the design isn't paying off yet. For the broader observability picture, see Designing trace and observability for AI agents in Antigravity.

One concrete next step

Don't try to ship all three layers at once. The single most useful first step is this:

Open the error logs of an agent you're already running. List the top five failure patterns over the last 30 days. Give each one a name.

That alone tells you where your agent actually breaks, which determines whether you start with detection or diagnosis. I started with diagnosis because just three patterns accounted for most of my incidents — handling those three visibly cut my night-time pages before I'd built any of the rest.

Your agent will only start getting smarter once it stops dying silently. The three-layer design is the first loop that makes that possible.