bq wraps every processor call in db.begin_nested() (a SAVEPOINT) on the session it hands you. Two consequences for processors that wait on long subprocesses (e.g. a 20-minute eval/agent run):
db.commit()inside a processor breaks both prod (commits inside bq's begin_nested context manager -> 'Can't operate on closed transaction inside context manager') and connection-bound test fixtures.- Leaving the tape transaction idle through the wait trips Postgres
idle_in_transaction_session_timeout(if configured on the engine): the connection is killed mid-processor, every later session use raises PendingRollbackError, and bq requeues the task forever — re-running paid work each cycle.
Working pattern (three moves):
- Poll-wait the subprocess in a loop (
proc.wait(timeout=30)cycles) and tick a keepalive each cycle:db.execute(select(1))resets the idle-in-transaction timer, so the strict leak detector can stay ON. - Do checkpoint-committed work in dedicated short sessions on the caller's bind:
with Session(db.get_bind(), expire_on_commit=False) as work_db: ...; work_db.commit(). In prod that's a fresh pooled connection with real commits; under a connection-bound pytest fixture (externalconnection.begin()+Session(bind=connection)) SQLAlchemy 2.0's default join_transaction_mode='conditional_savepoint' turns the inner commit into a savepoint release, so test isolation is preserved with NO harness changes. - Read every ORM value the long leg needs into plain locals before the wait; never touch the tape session's ORM objects mid-wait (attribute access on expired objects silently reopens a transaction).
Also add pool_pre_ping=True to the worker engine so a dead pooled connection is replaced at checkout instead of exploding the next task.