LLM agent resource cleanup cost patterns: the leaks that silently multiply your bill

import asyncio
import time
from contextlib import asynccontextmanager
from typing import Any, Callable, Awaitable
import runguard

rg = runguard.init(api_key="rg_live_...")

class ResourceScope:
    """
    Async context manager that tracks acquired resources
    and releases them in LIFO order on __aexit__.
    """
    def __init__(self, scope_name: str):
        self._scope_name = scope_name
        self._closers: list[tuple[str, Callable[[], Awaitable[None] | None]]] = []

    def register(self, name: str, closer: Callable) -> None:
        """Register a resource closer. Called automatically by acquire()."""
        self._closers.append((name, closer))

    async def __aenter__(self) -> "ResourceScope":
        return self

    @rg.resource_guard(cleanup_timeout_ms=500)
    async def __aexit__(self, *_) -> None:
        errors = []
        for name, closer in reversed(self._closers):
            t0 = time.monotonic()
            try:
                result = closer()
                if asyncio.iscoroutine(result):
                    await result
                elapsed_ms = (time.monotonic() - t0) * 1000
                if elapsed_ms > 500:
                    rg.alert(
                        f"slow_cleanup",
                        detail=f"{self._scope_name}/{name} took {elapsed_ms:.0f}ms to close",
                        severity="warning"
                    )
            except Exception as exc:
                errors.append(f"{name}: {exc}")
        if errors:
            raise RuntimeError(
                f"ResourceScope({self._scope_name}) cleanup errors: {errors}"
            )

async def run_agent_task(task_id: str, task_input: dict) -> dict:
    async with ResourceScope(f"task-{task_id}") as scope:

        # HTTP client — always closed on scope exit
        import httpx
        client = httpx.AsyncClient(timeout=30.0)
        scope.register("http_client", client.aclose)

        # Database pool — always released on scope exit
        import asyncpg
        pool = await asyncpg.create_pool(
            "postgresql://user:pass@localhost/agentdb",
            min_size=1, max_size=3
        )
        scope.register("db_pool", pool.close)

        # FAISS index — always unloaded on scope exit
        import faiss
        index = faiss.read_index("/data/embeddings.index")
        scope.register("faiss_index", lambda: None)  # FAISS releases on GC,
                                                      # but explicit del ensures timing

        # ... agent logic runs here ...
        result = await _execute_agent_steps(client, pool, index, task_input)

        return result
    # All three resources released here, in reverse order

@rg.connection_guard(max_open=5, alert_on_exceed=True)
async def run_agent_task(task_id: str, task_input: dict) -> dict:
    """
    RunGuard tracks HTTP + DB connections opened inside this function.
    If the count exceeds max_open=5, it fires a Slack/PagerDuty alert
    and (optionally) raises a BudgetExceededError to abort the task.
    """
    async with ResourceScope(f"task-{task_id}") as scope:
        # ... same setup as above ...
        pass

import Anthropic from "@anthropic-ai/sdk";
import { Pool } from "pg";
import RunGuard from "@runguard/sdk";

const rg = new RunGuard({ apiKey: process.env.RUNGUARD_API_KEY! });

class ManagedHttpClient {
  readonly client: Anthropic;

  constructor() {
    this.client = new Anthropic();
  }

  async [Symbol.asyncDispose](): Promise<void> {
    // Anthropic SDK v0.28+ exposes destroy()
    // for explicit connection teardown
    await (this.client as any).destroy?.();
  }
}

class ManagedDbPool {
  readonly pool: Pool;

  constructor(connectionString: string) {
    this.pool = new Pool({ connectionString, max: 3 });
  }

  async [Symbol.asyncDispose](): Promise<void> {
    await this.pool.end();
  }
}

async function runAgentTask(
  taskId: string,
  taskInput: Record<string, unknown>
): Promise<Record<string, unknown>> {
  // Both resources are automatically disposed when the block exits,
  // whether by normal return or by thrown exception.
  await using httpClient = new ManagedHttpClient();
  await using dbPool = new ManagedDbPool(process.env.DATABASE_URL!);

  // RunGuard wraps the task to track resource duration + cost
  return await rg.withTask(
    { taskId, tags: ["agent", "production"] },
    async (guard) => {
      const conn = await dbPool.pool.connect();
      try {
        // ... agent logic using httpClient.client and conn ...
        const result = await executeAgentSteps(
          httpClient.client,
          conn,
          taskInput,
          guard
        );
        return result;
      } finally {
        conn.release(); // always release individual connection
      }
    }
  );
  // dbPool.pool.end() and httpClient destroy() called automatically here
}

import RunGuard from "@runguard/sdk";

type Disposer = () => Promise<void> | void;

class AsyncResourceManager {
  private readonly disposers: Array<{ name: string; fn: Disposer }> = [];
  private readonly guard: ReturnType<typeof RunGuard.prototype.withTask>;

  constructor(private readonly scopeName: string) {}

  register(name: string, disposer: Disposer): void {
    this.disposers.push({ name, fn: disposer });
  }

  async run<T>(fn: (mgr: AsyncResourceManager) => Promise<T>): Promise<T> {
    try {
      return await fn(this);
    } finally {
      await this.disposeAll();
    }
  }

  private async disposeAll(): Promise<void> {
    const errors: string[] = [];
    for (const { name, fn } of [...this.disposers].reverse()) {
      const t0 = performance.now();
      try {
        await fn();
        const elapsed = performance.now() - t0;
        if (elapsed > 500) {
          // RunGuard cleanup-timing alert
          rg.alert("slow_cleanup", {
            scope: this.scopeName,
            resource: name,
            elapsedMs: Math.round(elapsed),
            severity: "warning",
          });
        }
      } catch (err) {
        errors.push(`${name}: ${err}`);
      }
    }
    if (errors.length) {
      throw new Error(
        `AsyncResourceManager(${this.scopeName}) cleanup errors: ${errors.join("; ")}`
      );
    }
  }
}

// Usage
async function runAgentTask(taskId: string, input: Record<string, unknown>) {
  const mgr = new AsyncResourceManager(`task-${taskId}`);
  return mgr.run(async (m) => {
    const client = new Anthropic();
    m.register("anthropic_client", async () => (client as any).destroy?.());

    const pool = new Pool({ connectionString: process.env.DATABASE_URL, max: 3 });
    m.register("db_pool", () => pool.end());

    // ... agent logic ...
    return executeAgentSteps(client, pool, input);
  });
}

from collections import OrderedDict
from typing import Optional
import numpy as np
import runguard

rg = runguard.init(api_key="rg_live_...")

class BoundedEmbeddingCache:
    """
    LRU cache for document embeddings.
    Evicts the least-recently-used entry when max_size is reached.
    Reports hit rate to RunGuard so cache efficiency appears alongside
    token-cost metrics.
    """
    def __init__(self, max_size: int = 10_000):
        self._cache: OrderedDict[str, np.ndarray] = OrderedDict()
        self._max_size = max_size
        self._hits = 0
        self._misses = 0
        self._evictions = 0

    def get(self, key: str) -> Optional[np.ndarray]:
        if key in self._cache:
            self._cache.move_to_end(key)
            self._hits += 1
            return self._cache[key]
        self._misses += 1
        return None

    def put(self, key: str, embedding: np.ndarray) -> None:
        if key in self._cache:
            self._cache.move_to_end(key)
        else:
            if len(self._cache) >= self._max_size:
                evicted_key, _ = self._cache.popitem(last=False)
                self._evictions += 1
                # Notify RunGuard — high eviction rate means cache is too small
                # and embeddings are being recomputed (costing API calls)
                rg.metric(
                    "embedding_cache_eviction",
                    value=1,
                    tags={
                        "evicted_key_prefix": evicted_key[:8],
                        "cache_size": self._max_size,
                        "total_evictions": self._evictions,
                    }
                )
            self._cache[key] = embedding

    @property
    def hit_rate(self) -> float:
        total = self._hits + self._misses
        return self._hits / total if total > 0 else 0.0

    def report_to_runguard(self) -> None:
        """Call this periodically (e.g. every 60s) to push cache stats."""
        rg.metric("embedding_cache_hit_rate", value=self.hit_rate)
        rg.metric("embedding_cache_size", value=len(self._cache))

# Shared singleton — 10,000 entries × 1,536 float32 = ~59 MB max
_embedding_cache = BoundedEmbeddingCache(max_size=10_000)

async def get_embedding(text: str, client) -> np.ndarray:
    key = hash(text).to_bytes(8, "big").hex()
    cached = _embedding_cache.get(key)
    if cached is not None:
        return cached
    response = await client.embeddings.create(
        model="text-embedding-3-small",
        input=text
    )
    embedding = np.array(response.data[0].embedding, dtype=np.float32)
    _embedding_cache.put(key, embedding)
    return embedding