[dabla]

PydanticAIHook.get_conn() resolves its connection (and every connection of a fallback_conn_ids chain) with get_connection() and reads extra_dejson, whose secret masking sends MaskSecret to the supervisor synchronously, and a subclass's _get_provider_kwargs may read more (Snowflake's builds a SnowflakeHook). On the event loop of an async task with another async SDK call in flight, each of those raises DeadlockImminentError, although the pydantic-ai agent the hook builds is itself async-native (await agent.run(...)).

This adds the async counterparts without a second resolution path:

• aget_conn() and acreate_agent() run the unchanged get_conn() / create_agent() on a single-worker executor of the hook module, never on asyncio's default executor: asend() reads its reply on that executor while it holds the comms lock, so lookups filling it would hang the task. From any thread other than the loop's, a synchronous send() waits for the in-flight asend() to release the comms lock instead of raising, so fallback chains, spec_file, model-id qualification and subclass overrides behave exactly as in the sync path; one worker also serializes the hook's sends on 3.0/3.1.
• aget_hook() is the async counterpart of get_hook(), with the same connection type check, so an async operator can dispatch to the pydanticai* subclasses.
• get_hook() now caches the connection it fetched on the hook it returns, so the hook's first get_conn() doesn't fetch it a second time (sync operators included).
• The hook publishes its connection and the connection's extra together, so a second thread calling get_conn() never reads a half-seeded cache (it would otherwise find the connection next to the empty default extra while the first is still masking it), and a failed extra read leaves nothing behind.
• A docs section shows await hook.acreate_agent(...) / await agent.run(...), for Airflow 3.2.1 and later (3.2.0 reads a reply outside the comms lock; DeadlockImminentError exists from 3.3.1, before that the loop blocks).

This is also the first building block for running @task.agent / AgentOperator natively async.

Tests: the sync lookups fail when they run on the event loop. The new tests resolve a model, a fallback chain, a subclass whose provider kwargs read another connection (as primary and as fallback), acreate_agent with every argument forwarded (with and without spec_file), a second acreate_agent that fetches nothing again, get_conn() from two threads on one hook, a retry after a failed extra read, each method completing while the default executor is full, each method carrying the caller's context variables into its worker, two hooks never sending at the same time, and aget_hook with its type check. Existing tests pass unchanged.

---

Was generative AI tooling used to co-author this PR?

• Yes — Claude Opus 5.5

Generated-by: Claude Opus 5.5 following the guidelines

[kaxil]

Thanks for sticking with this through four rounds, LGTM. Two optional test notes inline. It needs a rebase on main for the one-line infer_provider import conflict.

[dabla]

Thanks for sticking with this through four rounds, LGTM. Two optional test notes inline. It needs a rebase on main for the one-line infer_provider import conflict.

Thanks Kaxil for the numerous reviews, really appreciate the time and effort you put into it.