Pydantic AI developer on retainer: asyncpg connection pool bound to application event loop causes RuntimeError in pytest — pydantic-ai 0.0.9 — 100% CI test failure for 7 weeks while production runs correctly — Pydantic AI developer on monthly retainer

pydantic-ai 0.0.9 • Python 3.11 • asyncpg 0.29.0 • PostgreSQL 16 • FastAPI 0.111.0 • module-level pool created at import • pytest new event loop per test • RuntimeError: Task belongs to a different loop • 47 tests fail • production unaffected • fix: lazy pool init with loop check • 2.5h retainer engagement

Halbrook Technology Solutions Ltd. builds a B2B sales intelligence platform. In August 2026 they deployed a pydantic-ai sales agent that answers account manager queries by looking up customer data and order history from their PostgreSQL database. The agent had two tools: get_customer_data() which retrieved account details, contract values, and renewal dates, and get_order_history() which fetched the last 12 months of order records. The production deployment worked correctly from day one. Every CI build since the agent shipped had 47 failing tests with RuntimeError: Task belongs to a different loop. The team added @pytest.mark.asyncio to all test functions. The tests still failed. They added asyncio_mode = auto to pytest.ini. The tests still failed. They opened issues on both the pydantic-ai and asyncpg GitHub repositories. Seven weeks passed without a fix. They engaged a Pydantic AI retainer developer in October 2026.

The agent and tool implementation at the time of the engagement:

# tools/database.py
# Halbrook Technology Solutions Ltd.
# pydantic-ai 0.0.9, Python 3.11, asyncpg 0.29.0

import asyncio
import asyncpg
from pydantic import BaseModel

DATABASE_URL = "postgresql://sales_agent:secret@db.internal:5432/salesdb"

# ← PROBLEM: pool created at module import time
# asyncio.run() creates a temporary event loop to run the coroutine,
# then destroys the event loop after create_pool() returns.
# The pool's internal connections are bound to this temporary event loop.
pool = asyncio.run(asyncpg.create_pool(
    DATABASE_URL,
    min_size=5,
    max_size=20,
    command_timeout=10
))
# After asyncio.run() returns, the event loop is closed.
# The pool.connections[] are asyncpg Connection objects whose _loop attribute
# points to the now-closed temporary event loop.

# agent/sales_agent.py
# Halbrook Technology Solutions Ltd.

import pydantic_ai
from pydantic import BaseModel
from tools.database import pool   # ← imports module with module-level pool creation

class CustomerData(BaseModel):
    account_id: str
    company_name: str
    contract_value_usd: float
    renewal_date: str
    tier: str

class OrderSummary(BaseModel):
    account_id: str
    total_orders_12mo: int
    total_revenue_12mo: float
    last_order_date: str

agent = pydantic_ai.Agent(
    "openai:gpt-4o-mini",
    system_prompt=(
        "You are a sales intelligence assistant. Use the available tools to "
        "look up customer data and answer account manager questions accurately."
    )
)

@agent.tool
async def get_customer_data(ctx: pydantic_ai.RunContext[None], account_id: str) -> CustomerData:
    async with pool.acquire() as conn:   # ← pool is the module-level pool
        row = await conn.fetchrow(
            "SELECT account_id, company_name, contract_value_usd, renewal_date, tier "
            "FROM accounts WHERE account_id = $1",
            account_id
        )
        return CustomerData(**dict(row))

@agent.tool
async def get_order_history(ctx: pydantic_ai.RunContext[None], account_id: str) -> OrderSummary:
    async with pool.acquire() as conn:
        row = await conn.fetchrow(
            "SELECT account_id, COUNT(*) as total_orders_12mo, "
            "SUM(order_value_usd) as total_revenue_12mo, "
            "MAX(order_date)::text as last_order_date "
            "FROM orders WHERE account_id = $1 AND order_date >= NOW() - INTERVAL '12 months' "
            "GROUP BY account_id",
            account_id
        )
        return OrderSummary(**dict(row))

# tests/test_agent.py
# Halbrook Technology Solutions Ltd.
# pytest-asyncio 0.23.6

import pytest
from agent.sales_agent import agent

@pytest.mark.asyncio   # ← Adding this does NOT fix the RuntimeError
async def test_customer_lookup():
    result = await agent.run("What is the contract value for account ACC-001?")
    assert "ACC-001" in result.data
    assert "$" in result.data

# Failure output for all 47 tests:
# RuntimeError: Task belongs to a different loop
# File "tools/database.py", line 31, in get_customer_data
#   async with pool.acquire() as conn:
# File "asyncpg/pool.py", line 617, in acquire
#   return await asyncpg.pool.PoolConnectionProxy.__aenter__(holder)
# RuntimeError: Task belongs to a different loop
#
# The test's asyncio event loop ≠ the pool's creation event loop.
# asyncpg detects this when pool.acquire() tries to schedule
# connection work on the current (test) event loop while the pool
# holds connections bound to the (import-time) event loop.

Why production works but tests fail

The key difference between the production FastAPI application and the pytest test runner is event loop lifecycle. In production, the ASGI server (Uvicorn) creates a single long-lived event loop that runs for the entire lifetime of the process. The database connection pool used by the production agent is created inside FastAPI’s lifespan() function, which runs within this event loop. The pool’s connections are bound to the Uvicorn event loop. All subsequent pool.acquire() calls happen in the same Uvicorn event loop. Everything is consistent.

But the pool imported by agent/sales_agent.py via from tools.database import pool was created by a different mechanism: asyncio.run(asyncpg.create_pool(...)) at module import time. This pool’s connections are bound to the event loop that asyncio.run() created during import — a temporary event loop that is destroyed immediately after create_pool() returns. The production application creates its own pool separately in the lifespan() function; the module-level pool in tools/database.py is never used in production. It only affects tests, which import tools/database.py and use the module-level pool directly.

# Event loop timeline — production vs. test

# PRODUCTION (Uvicorn)
# ─────────────────────────────────────────────────────────────────────
# t=0: Uvicorn creates event loop L1 (long-lived, runs forever)
# t=1: FastAPI lifespan() starts in L1
# t=2: asyncpg pool P1 created in L1 (connections bound to L1)
# t=3..∞: requests handled in L1; pool.acquire() in L1 → works correctly
# Result: P1 and all requests share L1 → no event loop mismatch

# TEST (pytest + pytest-asyncio)
# ─────────────────────────────────────────────────────────────────────
# t=0: pytest collection starts
# t=1: pytest imports agent/sales_agent.py
# t=2: agent/sales_agent.py imports tools/database.py
# t=3: asyncio.run(create_pool()) creates temp event loop L0, creates pool P0,
#       P0.connections[].loop = L0, asyncio.run() closes L0 and returns P0
# t=4: P0 is now module-level `pool` in tools/database.py (bound to closed L0)
# t=5: pytest begins test_customer_lookup()
# t=6: pytest-asyncio creates event loop L1 for this test function
# t=7: test awaits agent.run("What is the contract value for account ACC-001?")
# t=8: agent calls get_customer_data() tool
# t=9: tool executes: async with pool.acquire() as conn:
#       pool = P0 (bound to L0, which is closed)
#       current loop = L1 (test function loop)
#       asyncpg: current_loop != pool._loop → RuntimeError: Task belongs to a different loop
# t=10: test fails — same for all 47 tests

# WHY @pytest.mark.asyncio DOESN'T HELP:
# The mark changes how pytest runs the test coroutine (wraps it in an event loop)
# but does not change when tools/database.py was imported (t=2, during collection)
# or when asyncio.run(create_pool()) was called (t=3, also during collection).
# By the time any test runs, P0 is already bound to the closed L0.

# WHY asyncio_mode = auto DOESN'T HELP:
# Same reason — asyncio_mode controls test event loop creation policy,
# not module import order or module-level asyncio.run() behavior.

Discovery: reading the traceback and tracing module imports

The retainer developer’s first step was to read the full traceback carefully rather than focusing only on the RuntimeError message:

# Full traceback from test failure (condensed)
# RuntimeError: Task belongs to a different loop
#
# asyncpg/pool.py:617: in acquire
#   async with asyncpg.pool.PoolConnectionProxy(holder, loop=loop) as conn:
# asyncpg/pool.py:298: in __aenter__
#   conn = await holder._acquire(timeout)
# asyncpg/pool.py:314: in _acquire
#   self._con = await self._pool._get_new_connection()  # OR from pool cache
# asyncpg/connection.py:2281: in _check_loop
#   if self._loop is not loop:
#       raise RuntimeError("Task belongs to a different loop")
#
# asyncpg.connection._check_loop() compares:
#   self._loop = the event loop the Connection object was created in
#   loop = asyncio.get_running_loop() at the time of the pool.acquire() call
# If they differ → RuntimeError

# Diagnostic: find out which event loop the pool was created in
import asyncio
import tools.database as db

print(f"Current running loop: {id(asyncio.get_event_loop())}")
print(f"Pool connection loop: {id(db.pool._holders[0]._con._loop)}")
# Output (collected during pytest test):
# Current running loop: 140234567890112   ← test function event loop
# Pool connection loop: 140234512345678   ← import-time asyncio.run() loop (now closed)
# These IDs differ → RuntimeError confirmed

# Find where the pool is created: grep tools/database.py
# → Line 12: pool = asyncio.run(asyncpg.create_pool(DATABASE_URL, ...))
# This is a module-level asyncio.run() call → created in import-time temp loop.

The diagnosis was clear within the first 30 minutes. The fix required changing the pool initialization from module-level to lazy (on-demand, within the calling event loop).

The fix: lazy pool initialization with event loop identity check

# tools/database.py — FIXED
# Halbrook Technology Solutions Ltd.
# pydantic-ai 0.0.9, Python 3.11, asyncpg 0.29.0

import asyncio
import asyncpg
from typing import Optional

DATABASE_URL = "postgresql://sales_agent:secret@db.internal:5432/salesdb"

# Module-level cache variables — initialized to None, populated lazily
_pool: Optional[asyncpg.Pool] = None
_pool_loop: Optional[asyncio.AbstractEventLoop] = None

async def get_pool() -> asyncpg.Pool:
    """
    Return the asyncpg connection pool, creating or recreating it as needed.

    Creates the pool within the currently running event loop. If the pool
    already exists and was created in the same loop, reuses it. If the pool
    was created in a different loop (happens in pytest where each test gets
    a new event loop), closes the old pool and creates a new one.

    This is safe for production (one long-lived event loop → pool created
    once on first request, reused for all subsequent requests) and for tests
    (new event loop per test → pool created fresh for each test, closed on
    loop change detection).
    """
    global _pool, _pool_loop

    current_loop = asyncio.get_running_loop()

    if _pool is not None and _pool_loop is current_loop:
        # Pool exists and was created in the current event loop — reuse it
        return _pool

    if _pool is not None:
        # Pool exists but was created in a different event loop (e.g., previous test)
        # Close it before creating a new one in the current loop
        try:
            await _pool.close()
        except Exception:
            pass  # Pool may already be unusable if the old loop was closed

    # Create new pool in the current event loop
    _pool = await asyncpg.create_pool(
        DATABASE_URL,
        min_size=5,
        max_size=20,
        command_timeout=10
    )
    _pool_loop = current_loop
    return _pool

# agent/sales_agent.py — FIXED
# Halbrook Technology Solutions Ltd.

import pydantic_ai
from pydantic import BaseModel
from tools.database import get_pool   # ← import function, not module-level pool instance

class CustomerData(BaseModel):
    account_id: str
    company_name: str
    contract_value_usd: float
    renewal_date: str
    tier: str

class OrderSummary(BaseModel):
    account_id: str
    total_orders_12mo: int
    total_revenue_12mo: float
    last_order_date: str

agent = pydantic_ai.Agent(
    "openai:gpt-4o-mini",
    system_prompt=(
        "You are a sales intelligence assistant. Use the available tools to "
        "look up customer data and answer account manager questions accurately."
    )
)

@agent.tool
async def get_customer_data(ctx: pydantic_ai.RunContext[None], account_id: str) -> CustomerData:
    pool = await get_pool()   # ← lazy init within current event loop
    async with pool.acquire() as conn:
        row = await conn.fetchrow(
            "SELECT account_id, company_name, contract_value_usd, renewal_date, tier "
            "FROM accounts WHERE account_id = $1",
            account_id
        )
        return CustomerData(**dict(row))

@agent.tool
async def get_order_history(ctx: pydantic_ai.RunContext[None], account_id: str) -> OrderSummary:
    pool = await get_pool()   # ← lazy init within current event loop
    async with pool.acquire() as conn:
        row = await conn.fetchrow(
            "SELECT account_id, COUNT(*) as total_orders_12mo, "
            "SUM(order_value_usd) as total_revenue_12mo, "
            "MAX(order_date)::text as last_order_date "
            "FROM orders WHERE account_id = $1 AND order_date >= NOW() - INTERVAL '12 months' "
            "GROUP BY account_id",
            account_id
        )
        return OrderSummary(**dict(row))

The retainer developer also added a conftest.py fixture to close the pool after each test, ensuring clean resource teardown between tests and preventing pool connection exhaustion when running the full test suite:

# tests/conftest.py — ADDED
# Halbrook Technology Solutions Ltd.

import pytest
import asyncio
import tools.database as db

@pytest.fixture(autouse=True)
async def close_pool_after_test():
    """Close the asyncpg pool after each test to prevent connection exhaustion."""
    yield
    # After the test: close the pool if one was created for this test's event loop
    if db._pool is not None:
        try:
            await db._pool.close()
        except Exception:
            pass
        db._pool = None
        db._pool_loop = None

Validation

# Test suite results after fix

# Before fix (7 weeks August–October 2026):
# Tests: 47 total  |  0 passed  |  47 FAILED
# Failure: RuntimeError: Task belongs to a different loop (100% of tests)
# CI status: FAILED on every commit for 7 weeks
# Team action: skipped all agent tests by adding @pytest.mark.skip to 47 functions
# Production: running correctly (different pool instance in lifespan())

# After fix (October 24 2026):
# Tests: 47 total  |  47 passed  |  0 FAILED
# Failure: none
# CI status: GREEN
# Removed: all 47 @pytest.mark.skip decorators (re-enabled all tests)

# Production performance check (load test, 200 requests):
# p50 agent run latency:   1.82s   (was 1.80s — negligible change)
# p95 agent run latency:   3.21s   (was 3.18s — negligible change)
# Database pool:           created on first request, reused for all subsequent
# Pool size at steady state: 5 connections (min_size=5, as before)

# Additional finding:
# The module-level asyncio.run(create_pool()) in the original code was creating
# a dead pool that was never used in production (production used the lifespan pool)
# and was always broken in tests. Removing it had zero production impact.
# The production-correct pattern for asyncpg with FastAPI is:
#   lifespan() function creates pool in the ASGI event loop
#   pool injected via pydantic-ai dependency injection (RunContext)
# The lazy get_pool() approach is an acceptable alternative when dependency
# injection requires larger refactoring scope than the retainer budget allows.

A secondary recommendation from the retainer developer: for future pydantic-ai agents that require database access, use pydantic-ai’s built-in dependency injection rather than module-level pool references. The RunContext[Deps] pattern passes a pre-created pool (from the ASGI lifespan) into tool functions without any module-level state, eliminating the event loop binding issue entirely:

# Preferred long-term pattern: pydantic-ai dependency injection
# Eliminates module-level state and event loop binding concerns

from dataclasses import dataclass
import asyncpg
import pydantic_ai

@dataclass
class AgentDeps:
    pool: asyncpg.Pool   # created in ASGI lifespan, injected per request

agent = pydantic_ai.Agent(
    "openai:gpt-4o-mini",
    deps_type=AgentDeps,
    system_prompt="You are a sales intelligence assistant."
)

@agent.tool
async def get_customer_data(ctx: pydantic_ai.RunContext[AgentDeps], account_id: str) -> CustomerData:
    # ctx.deps.pool was created in the ASGI lifespan event loop
    # It is always in the correct event loop for this request
    async with ctx.deps.pool.acquire() as conn:
        row = await conn.fetchrow("SELECT ... FROM accounts WHERE account_id = $1", account_id)
        return CustomerData(**dict(row))

# In tests: create a test pool fixture scoped to the test event loop
# and pass it as deps when calling agent.run():
# result = await agent.run("query", deps=AgentDeps(pool=test_pool))

Total retainer engagement: 2.5 hours (0.5h triage: read full traceback; identify asyncpg.connection._check_loop(); add event loop ID diagnostic prints; confirm pool loop != test loop; locate module-level asyncio.run(create_pool()) in tools/database.py; 1.0h root cause: trace import order; confirm pool created during pytest collection in temporary event loop; confirm why @pytest.mark.asyncio and asyncio_mode=auto don’t help — pool pre-dated any test event loop; verify production uses separate lifespan pool; confirm module-level pool never used in production; 0.5h fix: implement lazy get_pool() with event loop identity check; update both tool functions; add conftest.py pool teardown fixture; 0.5h validation: run full 47-test suite; confirm 47/47 pass; run production load test; confirm p50 unchanged; present dependency injection pattern as long-term alternative).

Related retainer posts

• DSPy developer on retainer — compiled ChainOfThought program loses few-shot demos after program.load() — production RAG performs same as uncompiled baseline for 3 months
• Instructor developer on retainer — Claude returns JSON string for Optional[List[str]] instead of JSON array — anyOf schema triggers string serialization — 15% extraction failure rate for 11 weeks
• Portkey developer on retainer — fallback from GPT-3.5-turbo to Claude Haiku passes OpenAI-only name field — Anthropic rejects with 400 — 23% failure rate for 6 weeks