Fix buyer-1g4: offload sync CrewAI flow.kickoff() to worker thread

CrewAI's sync Flow.kickoff() internally does
ThreadPoolExecutor.submit(asyncio.run, _run_flow).result() — the
.result() call blocks the calling thread regardless. When invoked
from a FastAPI async handler (BackgroundTasks-driven _run_booking_flow),
that calling thread is the event loop, so every other request hangs
for the duration of the booking.

Wrap the four async-context call sites with asyncio.to_thread() so
the blocking .result() runs on a worker thread and the event loop
stays responsive:

  - api/main.py:574       _run_booking_flow.kickoff
  - api/main.py:588       _run_booking_flow.approve_all (auto_approve)
  - api/main.py:450       approve_all_recommendations endpoint
  - buyer_deal_flow.py:684 run_buyer_deal_flow.kickoff

We deliberately do NOT use ``await flow.kickoff_async()`` here, even
though it looks cleaner. The buyer's Flow @start/@listen step methods
are sync (they call crew.kickoff() directly). Awaiting kickoff_async
from the event loop runs those sync steps in the loop thread anyway,
re-introducing the original block. The to_thread(kickoff) shape puts
the whole flow execution behind a worker thread.

Add regression tests under tests/integration/test_buyer_1g4_async_kickoff.py:
  - Runtime tests assert flow.kickoff / flow.approve_all execute on
    a worker thread (not MainThread).
  - AST source-level tests walk each async def and reject sync
    .kickoff() / .approve_all() calls outside await/to_thread.

Smoke verified: /health stayed responsive (mean 34ms, p95 41ms,
max 43ms) across 82 probes during a 17-min concurrent booking flow.
Before the fix the same probe hung indefinitely (~58 min before kill).

The booking still doesn't reach terminal state end-to-end, but that's
gated on a separate cross-repo contract drift (POST /products/search
returns 405 because the seller exposes only GET /products) — that
bug is already tracked separately and is independent of buyer-1g4.

Author: atc964

src/ad_buyer/flows/buyer_deal_flow.py

@@ -11,6 +11,7 @@
 match each ref against package capabilities (proposal §5.1).
 """
 
+import asyncio
 import logging
 import sqlite3
 from datetime import datetime
@@ -680,8 +681,12 @@ async def run_buyer_deal_flow(
         if audience_plan is not None:
             flow.state.audience_plan = audience_plan
 
-        # Run flow
-        result = flow.kickoff()
+        # Run flow.
+        # buyer-1g4: offload sync flow.kickoff() to a worker thread so
+        # the caller's event loop stays responsive. See api/main.py
+        # _run_booking_flow for the full rationale (sync Flow steps
+        # block the loop if we use ``await flow.kickoff_async()``).
+        result = await asyncio.to_thread(flow.kickoff)
 
         return {
             "result": result,

src/ad_buyer/interfaces/api/main.py

@@ -3,6 +3,7 @@
 
 """FastAPI server for the Ad Buyer System."""
 
+import asyncio
 import json
 import logging
 import sqlite3
@@ -447,7 +448,10 @@ async def approve_all_recommendations(job_id: str) -> dict[str, Any]:
             detail="Flow state not available. Job may have expired.",
         )
 
-    result = flow.approve_all()
+    # buyer-1g4: approve_all() runs sync CrewAI work that holds the
+    # calling thread; offload to a worker thread so the event loop
+    # stays responsive.
+    result = await asyncio.to_thread(flow.approve_all)
 
     job["status"] = "completed" if result.get("status") == "success" else "failed"
     job["booked_lines"] = [b.model_dump() for b in flow.state.booked_lines]
@@ -571,7 +575,22 @@ async def _run_booking_flow(job_id: str, request: BookingRequest) -> None:
         job["_flow"] = flow
 
         job["progress"] = 0.2
-        _result = flow.kickoff()
+        # buyer-1g4: offload the whole sync flow.kickoff() to a worker
+        # thread so the FastAPI event loop stays responsive while the
+        # crew agents (which block on real LLM HTTP calls) run.
+        #
+        # We do NOT use ``await flow.kickoff_async()`` here even though
+        # it looks cleaner — the buyer's Flow ``@start`` / ``@listen``
+        # step methods are sync and themselves call ``crew.kickoff()``
+        # directly. Awaited from the event loop, those sync steps would
+        # run IN the event loop thread, re-introducing the block.
+        #
+        # CrewAI's sync ``Flow.kickoff()`` already does
+        # ``ThreadPoolExecutor.submit(asyncio.run, _run_flow).result()``
+        # internally — the ``.result()`` is what blocks the calling
+        # thread. ``asyncio.to_thread`` puts that ``.result()`` on a
+        # worker thread (two threads deep, but the event loop is free).
+        _result = await asyncio.to_thread(flow.kickoff)
 
         # Propagate any errors captured by flow steps into the job response so
         # the client can see what went wrong instead of silent-success (ar-jbod).
@@ -585,7 +604,8 @@ async def _run_booking_flow(job_id: str, request: BookingRequest) -> None:
         job["recommendations"] = [r.model_dump() for r in flow.state.pending_approvals]
 
         if request.auto_approve:
-            flow.approve_all()
+            # buyer-1g4: same reason as above — offload sync work.
+            await asyncio.to_thread(flow.approve_all)
             job["booked_lines"] = [b.model_dump() for b in flow.state.booked_lines]
             job["status"] = "completed"
         else:

tests/integration/test_buyer_1g4_async_kickoff.py

@@ -0,0 +1,222 @@
+# Author: Green Mountain Systems AI Inc.
+# Donated to IAB Tech Lab
+
+"""Regression tests for buyer-1g4: sync CrewAI Flow.kickoff() must not
+block the FastAPI event loop.
+
+CrewAI's sync ``Flow.kickoff()`` internally does::
+
+    with ThreadPoolExecutor(max_workers=1) as pool:
+        return pool.submit(ctx.run, asyncio.run, _run_flow()).result()
+
+The final ``.result()`` blocks the calling thread until the inner thread
+finishes — so calling ``flow.kickoff()`` directly from a FastAPI
+``async def`` handler blocks the entire event loop for the duration of
+the run.
+
+The fix is to wrap the call with ``asyncio.to_thread`` so the
+``.result()`` runs on a worker thread instead of the event loop thread.
+We deliberately do NOT use ``await flow.kickoff_async()`` — the buyer's
+Flow ``@start`` / ``@listen`` step methods are themselves sync (they
+call ``crew.kickoff()`` directly), so awaiting kickoff_async from the
+event loop just runs those sync steps in the loop thread anyway.
+
+These tests guard against anyone reverting to an un-offloaded sync
+``kickoff()`` / ``approve_all()`` in async contexts by asserting the
+calls execute on a worker thread, not on ``MainThread``.
+"""
+
+import pytest
+
+from ad_buyer.flows.deal_booking_flow import DealBookingFlow
+from ad_buyer.interfaces.api.main import (
+    BookingRequest,
+    CampaignBrief,
+    _run_booking_flow,
+    jobs,
+)
+from ad_buyer.time_utils import utc_now
+
+
+def _seed_job(job_id: str) -> None:
+    jobs[job_id] = {
+        "job_id": job_id,
+        "status": "pending",
+        "progress": 0.0,
+        "errors": [],
+        "budget_allocations": {},
+        "recommendations": [],
+        "booked_lines": [],
+        "updated_at": utc_now().isoformat(),
+        "created_at": utc_now().isoformat(),
+    }
+
+
+def _sample_brief() -> CampaignBrief:
+    return CampaignBrief(
+        name="buyer-1g4 regression",
+        objectives=["awareness"],
+        budget=50000,
+        start_date="2026-07-01",
+        end_date="2026-07-31",
+        target_audience={"geo": ["US"]},
+    )
+
+
+class TestRunBookingFlowOffloadsKickoff:
+    """``api.main._run_booking_flow`` must offload sync CrewAI work so
+    the FastAPI event loop stays responsive."""
+
+    @pytest.mark.asyncio
+    async def test_kickoff_runs_on_worker_thread(self, monkeypatch):
+        """``flow.kickoff()`` (sync) must execute on a worker thread,
+        never on ``MainThread``. ``asyncio.to_thread`` provides this
+        offload; calling ``flow.kickoff()`` directly from the async
+        background task would run it on the event-loop thread and
+        block every other request (buyer-1g4)."""
+
+        import threading
+
+        kickoff_threads: list[str] = []
+
+        def fake_kickoff(self, *args, **kwargs):
+            kickoff_threads.append(threading.current_thread().name)
+            return {}
+
+        monkeypatch.setattr(DealBookingFlow, "kickoff", fake_kickoff)
+
+        job_id = "test-buyer-1g4-kickoff-thread"
+        _seed_job(job_id)
+        try:
+            await _run_booking_flow(
+                job_id,
+                BookingRequest(brief=_sample_brief(), auto_approve=False),
+            )
+
+            assert kickoff_threads, "flow.kickoff should have been called"
+            assert "MainThread" not in kickoff_threads[0], (
+                f"flow.kickoff ran on {kickoff_threads[0]} — buyer-1g4 "
+                "regression: should be offloaded via asyncio.to_thread"
+            )
+            assert jobs[job_id]["status"] != "failed", (
+                f"Job failed unexpectedly: errors={jobs[job_id]['errors']}"
+            )
+        finally:
+            jobs.pop(job_id, None)
+
+    @pytest.mark.asyncio
+    async def test_approve_all_runs_on_worker_thread(self, monkeypatch):
+        """When ``auto_approve=True``, ``approve_all`` is sync CrewAI
+        work and must be offloaded via ``asyncio.to_thread`` for the
+        same reason as ``kickoff``."""
+
+        import threading
+
+        approve_threads: list[str] = []
+
+        def fake_kickoff(self, *args, **kwargs):
+            return {}
+
+        def fake_approve_all(self):
+            approve_threads.append(threading.current_thread().name)
+            return {"status": "success", "booked_lines": []}
+
+        monkeypatch.setattr(DealBookingFlow, "kickoff", fake_kickoff)
+        monkeypatch.setattr(DealBookingFlow, "approve_all", fake_approve_all)
+
+        job_id = "test-buyer-1g4-approve-all-thread"
+        _seed_job(job_id)
+        try:
+            await _run_booking_flow(
+                job_id,
+                BookingRequest(brief=_sample_brief(), auto_approve=True),
+            )
+
+            assert approve_threads, "approve_all should have been called"
+            assert "MainThread" not in approve_threads[0], (
+                f"approve_all ran on {approve_threads[0]} — buyer-1g4 "
+                "regression: should be offloaded via asyncio.to_thread"
+            )
+        finally:
+            jobs.pop(job_id, None)
+
+
+class TestSourceLevelAsyncContract:
+    """Source-level assertions for async-context call sites.
+
+    These guard against silent reverts to sync ``kickoff()`` /
+    unwrapped ``approve_all()`` inside ``async def`` functions. They are
+    cheaper than full mocking for sites whose runtime invocation needs
+    extensive client/flow stubbing (e.g. ``run_buyer_deal_flow``).
+    """
+
+    @staticmethod
+    def _calls_in_async_functions(filename: str) -> dict[str, list[str]]:
+        """Walk a source file and return, for each ``async def`` function,
+        the list of attribute-call names invoked synchronously (i.e.
+        not via ``await ...`` and not via ``asyncio.to_thread(...)``).
+        """
+        import ast
+        import pathlib
+
+        path = pathlib.Path(filename)
+        tree = ast.parse(path.read_text())
+
+        results: dict[str, list[str]] = {}
+
+        class _Visitor(ast.NodeVisitor):
+            def __init__(self):
+                self._async_stack: list[str] = []
+
+            def visit_AsyncFunctionDef(self, node):
+                self._async_stack.append(node.name)
+                results.setdefault(node.name, [])
+                self.generic_visit(node)
+                self._async_stack.pop()
+
+            def visit_FunctionDef(self, node):
+                # Skip nested sync defs — they have their own loop semantics.
+                pass
+
+            def visit_Await(self, node):
+                # Anything under an Await is fine — don't descend.
+                pass
+
+            def visit_Call(self, node):
+                if self._async_stack and isinstance(node.func, ast.Attribute):
+                    results[self._async_stack[-1]].append(node.func.attr)
+                self.generic_visit(node)
+
+        _Visitor().visit(tree)
+        return results
+
+    def test_run_booking_flow_no_sync_kickoff(self):
+        from ad_buyer.interfaces.api import main as api_main
+
+        calls = self._calls_in_async_functions(api_main.__file__)
+        booking_calls = calls.get("_run_booking_flow", [])
+        assert "kickoff" not in booking_calls, (
+            "_run_booking_flow contains a sync .kickoff() call — buyer-1g4 "
+            "regression. Wrap with ``await asyncio.to_thread(flow.kickoff)``."
+        )
+
+    def test_approve_all_recommendations_endpoint_no_sync_approve(self):
+        from ad_buyer.interfaces.api import main as api_main
+
+        calls = self._calls_in_async_functions(api_main.__file__)
+        endpoint_calls = calls.get("approve_all_recommendations", [])
+        assert "approve_all" not in endpoint_calls, (
+            "approve_all_recommendations endpoint contains a sync "
+            ".approve_all() call — buyer-1g4 regression. Wrap with "
+            "``await asyncio.to_thread(flow.approve_all)``."
+        )
+
+    def test_run_buyer_deal_flow_no_sync_kickoff(self):
+        from ad_buyer.flows import buyer_deal_flow
+
+        calls = self._calls_in_async_functions(buyer_deal_flow.__file__)
+        run_calls = calls.get("run_buyer_deal_flow", [])
+        assert "kickoff" not in run_calls, (
+            "run_buyer_deal_flow contains a sync .kickoff() call — "
+            "buyer-1g4 regression. Use ``await flow.kickoff_async()``."
+        )