perf(ci): reuse a session-scoped engine for the per-test DB reset

The autouse integration teardown created a fresh SQLAlchemy engine + Postgres
connection for EVERY test, then disposed it — --durations showed the 15
slowest ops in both long shards were all ~1.5-2s teardowns (the connect+SCRAM
handshake, not test logic). Hoist the truncate engine to a session-scoped,
pool_pre_ping'd fixture so the pooled connection is reused across teardowns;
the TRUNCATE+restore still runs per test, so isolation is unchanged. Lazy
create_engine means the no-DB unit job instantiates but never connects.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

tests/conftest.py

@@ -104,32 +104,51 @@ async def client(app):
 _SEED_SNAPSHOT: dict[str, list[dict]] | None = None
 
 
+@pytest.fixture(scope="session")
+def _truncate_engine():
+    """Session-scoped sync engine reused by the per-test DB-reset teardown.
+
+    Creating a fresh engine + Postgres connection for EVERY test's teardown was
+    the integration suite's dominant cost — `--durations` showed the 15 slowest
+    operations in both long shards were all ~1.5-2s teardowns, i.e. the
+    connect+SCRAM handshake, not test logic. A single pooled connection reused
+    across teardowns cuts that to the TRUNCATE itself. `pool_pre_ping` guards
+    against Postgres reaping the idle connection mid-suite (a stale-connection
+    failure would be a nasty flaky bounce; the ping is sub-millisecond).
+
+    `create_engine` is lazy — it opens no connection until first `.begin()` —
+    so the no-DB unit job instantiates this object but never connects.
+    """
+    eng = create_engine(_sync_database_url(), future=True, pool_pre_ping=True)
+    yield eng
+    eng.dispose()
+
+
 @pytest.fixture(autouse=True)
-def _reset_db_after_integration(request):
+def _reset_db_after_integration(request, _truncate_engine):
     """Integration tests exercise app/celery code that commits on its own
     connection, which the rollback fixtures above can't undo — so data
     would leak between tests. Capture the seeded baseline at the first
     integration test, then after each integration-marked test truncate
-    every model table and restore that baseline. Connects to the DB ONLY
-    for integration-marked tests, so the no-DB fast unit job is untouched.
+    every model table and restore that baseline. Touches the DB ONLY for
+    integration-marked tests, so the no-DB fast unit job is untouched.
+
+    Uses the session-scoped `_truncate_engine` (pooled, reused) rather than
+    building a fresh engine per test — see that fixture's docstring.
     """
     global _SEED_SNAPSHOT
     is_integration = request.node.get_closest_marker("integration") is not None
 
     if is_integration and _SEED_SNAPSHOT is None:
-        eng = create_engine(_sync_database_url(), future=True)
         snap: dict[str, list[dict]] = {}
-        try:
-            with eng.connect() as conn:
-                for t in Base.metadata.sorted_tables:
-                    rows = [
-                        dict(m)
-                        for m in conn.execute(t.select()).mappings().all()
-                    ]
-                    if rows:
-                        snap[t.name] = rows
-        finally:
-            eng.dispose()
+        with _truncate_engine.connect() as conn:
+            for t in Base.metadata.sorted_tables:
+                rows = [
+                    dict(m)
+                    for m in conn.execute(t.select()).mappings().all()
+                ]
+                if rows:
+                    snap[t.name] = rows
         _SEED_SNAPSHOT = snap
 
     yield
@@ -139,18 +158,14 @@ def _reset_db_after_integration(request):
     tables = ", ".join(t.name for t in Base.metadata.sorted_tables)
     if not tables:
         return
-    eng = create_engine(_sync_database_url(), future=True)
-    try:
-        with eng.begin() as conn:
-            conn.exec_driver_sql(
-                f"TRUNCATE {tables} RESTART IDENTITY CASCADE"
-            )
-            for t in Base.metadata.sorted_tables:
-                rows = (_SEED_SNAPSHOT or {}).get(t.name)
-                if rows:
-                    conn.execute(t.insert(), rows)
-    finally:
-        eng.dispose()
+    with _truncate_engine.begin() as conn:
+        conn.exec_driver_sql(
+            f"TRUNCATE {tables} RESTART IDENTITY CASCADE"
+        )
+        for t in Base.metadata.sorted_tables:
+            rows = (_SEED_SNAPSHOT or {}).get(t.name)
+            if rows:
+                conn.execute(t.insert(), rows)
 
 
 @pytest_asyncio.fixture(autouse=True)