[NOTICKET] fix: F-5 — isolate DB threads; stop treating session hardening as best-effort

`asyncio.wait_for` cancels the awaiting COROUTINE. The `to_thread` worker underneath
is not cancellable — it runs to completion, holding a thread and a connection on the
CUSTOMER's database long after we returned "timed out" to the user. Degrading their
production database is our liability, and it is invisible from our side because we
already answered.

(a) Dedicated bounded pool. DB work moves off `asyncio.to_thread` (which always uses
the shared default executor) onto its own ThreadPoolExecutor(50, "dbexec") via
`run_in_executor`. Abandoned workers previously accumulated in a pool of
min(32, cpu_count+4) shared with every other to_thread caller in the process — notably
the tabular Parquet loader — so a handful of slow customer queries could stall
unrelated work service-wide. Sized to the engine cache's own ceiling (_MAX_ENGINES=50
x _POOL_SIZE=1): more threads than that cannot make progress anyway. Timeout semantics
are unchanged.

(b) Session hardening is no longer best-effort — and reading it turned up a latent bug
worse than the finding. Both SETs shared one `try`:

cur.execute("SET statement_timeout = ...")
cur.execute("SET default_transaction_read_only = on")

so a statement_timeout failure SKIPPED THE READ-ONLY SET ENTIRELY, and the connection
then served queries in a writable session — behind a logger.warning, with the code
commenting that this was fine because "the authoritative guarantee is the compiler +
sqlglot guard". Defense in depth means this layer holds too.

They are now independent:
- `default_transaction_read_only` FAILS THE CONNECTION if it cannot be set. A
writable session against a customer database is not something to degrade into.
- `statement_timeout` logs at ERROR with `degraded_seam` but does not refuse service.
It is the customer's I/O at risk, not our correctness, so refusing outright is the
worse trade — but it is no longer a warning described as best-effort, because with
(a) established it is the ONLY real server-side bound.

The old comment claiming statement_timeout is "backed by the executor's
asyncio.wait_for" was circular and is corrected in place — that is the whole point of
this finding.

RISK TO WATCH: if any deployment currently fails the read-only SET silently, its
sources now fail loudly instead of running unhardened. Believed impossible — Neon
accepts it as a SET, which the existing comment documents — but it is the one
judgement call in this change and it is deliberate.

7 new tests. Suite 456 passed / 0 failed / 7 skipped. Ruff clean on touched paths.
`import main` OK. Readiness eval 17/17.

Docs: DEV_PLAN #45/#48 closed, #49 added; contract documents PII redaction on both
the traceability preview and report evidence tables.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

API_CONTRACT_BE_PYTHON.md

@@ -371,7 +371,7 @@ Report v2 fields (added 2026-07-09; all default-empty, so older stored reports r
 - `bq_answers` — one entry per business question. `status` is `answered` | `partial` | `unanswered`; `record_ids` cite the backing analyses. Written in the analysis's language (Indonesian objective → Indonesian answers).
 - `unresolved` — runs that were attempted but produced no usable evidence (every `analyze_*` step failed). Not part of the findings body.
 - `excluded` — runs the caller excluded via `exclude_record_ids`.
-- `evidence_tables` — `record_id` → small result tables copied from the run's stored outputs (max 3 tables per record, max 10 rows each; `truncated: true` when rows were capped). Rendered as markdown tables under the matching Key Findings group in `rendered_markdown`.
+- `evidence_tables` — `record_id` → small result tables copied from the run's stored outputs (max 3 tables per record, max 10 rows each; `truncated: true` when rows were capped). Rendered as markdown tables under the matching Key Findings group in `rendered_markdown`. **PII redaction (added 2026-07-24):** cells in a column the catalog flags as PII are replaced with `"[redacted]"`; column headers are kept. A report is permanent and versioned, so this is where an unmasked customer name would have lasted longest. The narrative (`executive_summary`, findings, `bq_answers`) is generated from the real values and is unaffected — only the raw evidence dump is redacted. Reports generated before this date are unchanged.
 - `charts` *(added 2026-07-14)* — `record_id` → `dataeyond.chart.v1` envelopes (see §Charts) copied verbatim from the run's stored outputs (max 3 per record). `rendered_markdown` gains an `## EDA` section where each chart appears as a fenced block the frontend renders with plotly.js:
 
   ````text
@@ -509,6 +509,7 @@ Field rules:
 - `thinking`, `filters[].description`, `tool_calls[].summary` are built from fixed templates, never an LLM — traceability adds no latency or token cost and cannot hallucinate.
 - The payload also carries an internal `user_id` (ownership); the frontend may ignore it.
 - Truncation: `preview` ≤ 5 rows; any string inside `input`/`output`/`preview`/`snippet` ≤ 300 chars (executed `query` ≤ 2000); rows beyond the preview are dropped (`row_count` is preserved).
+- **PII redaction in `preview` (added 2026-07-24).** Cells belonging to a column the catalog flags as PII are replaced with the literal string `"[redacted]"`. When any cell was redacted the output also carries `pii_masked`: the list of column names affected — render that so the user understands the blanks are deliberate, not missing data. The **column headers are never redacted**, only the values. This affects the stored provenance record only: the chat answer itself is generated from the real values, so a question like "list our top customers" still answers normally. Records written before this date have no `pii_masked` key and are returned unchanged.
 
 Response `200` for `structured_flow`:
 

DEV_PLAN.md

@@ -246,10 +246,11 @@ plus the live report bug on analysis `966224d4…`. Same status legend as §0.
 | 42 | **F-20 observability** — `degraded_seam=<name>` on every never-throw / silent-drop path | Rifqi | 🔄 | The 2026-07-23 report bug was invisible by construction: the record was dropped with zero logging. **Partially shipped 2026-07-23** — the 10 seams where silent degradation is user-visible now emit a stable `degraded_seam` field (+ `repr(e)` instead of `str(e)`, so an empty-`str()` Fernet error is no longer a blank log): `input_guard_fail_open`, `analysis_catalog_read`, `report_floor_record_read`, `traceability_persist`, `traceability_flush`, `chart_persist` (×2), `report_input_persist` (×2), `analysis_state_ensure`. **Remaining:** the other ~76 `except Exception` sites, most of which are in unwired routers (`db_client`, `data_catalog`, `users`) or non-live paths — deliberately not swept, since a blanket edit across unwired code is exactly the drive-by §7A forbids. Control flow unchanged throughout (§5.4) |
 | 43 | **Go identity contract** — what does Go forward, and when? | Rifqi ↔ Harry | ⬜ new | Needed to replace #37's shared secret with real per-user authorization. Until then the store predicates in #38 are defensive only |
 | 44 | **F-17 / F-18 / F-25 compiler-parity batch** | Rifqi | ✅ new | Shipped 2026-07-23. Three execution-verified review findings that had **no task row** — the tracker jumped from #39 to #40 and lost them. **F-17 (High):** the bare-select check was gated on `if ir.group_by`, so a mixed select with `group_by=[]` passed; Postgres then failed loudly but the pandas path silently DROPPED the column while `output_columns` still advertised it → a real-looking table with a fabricated all-null column. Fixed in `validator.py` (fires whenever any agg is present — no false positives possible) + a presence backstop in `TabularExecutor`. **F-18:** `astype(str)` ran before `na=False`, so `NULL LIKE '%an%'` matched the literal `"nan"`/`"None"`. **F-25:** `SqlCompiler` raised on an empty `in`/`not_in` where pandas and `_column_values`' own docstring implement the empty-set semantics — a two-step plan whose first step legitimately returned zero rows hard-failed instead of answering. 16 new parity tests; one stale test updated (it pinned the old F-25 raise) |
-| 45 | **F-9 PII in persisted artifacts** — mask the traceability preview + report evidence tables | Rifqi | ⬜ new | **Decided 2026-07-23** (CODE_REVIEW §7) but never tracked. `pii_flag` is an ingestion-time control only: nothing stops the planner SELECTing a flagged column, and the real values then land in `message_traceability.data` (served by a GET that is only now becoming scoped, #40) and are frozen permanently into `reports.content`. Decision: **the assembler still sees values** (otherwise "list our top customers" stops being answerable); mask only the persisted artifacts. Carry `pii_flag` onto `retrieve_data`'s output meta — the catalog is already in scope at `data_access.py:232` |
+| 45 | **F-9 PII in persisted artifacts** — mask the traceability preview + report evidence tables | Rifqi | ✅ | Shipped 2026-07-24. `pii_flag` was an **ingestion-time control only**: it nulls `sample_values` into the planner prompt, but nothing stops the planner SELECTing a flagged column — and "list our top 20 customers" legitimately selects `customer_name`/`email`. Real values then reached two PERSISTED sinks: `message_traceability.data` (served by an unauthenticated GET, F-3 declined) and report evidence tables frozen permanently into `reports.content`. Fix = `retrieve_data` now carries `meta.pii_columns` (resolved through the IR select list, so aliases are honoured), and both sinks redact those cells to `[redacted]`. **Per the 2026-07-23 decision the ASSEMBLER still receives real values**, so the answer prose is unchanged and the question stays answerable — verified end-to-end: assembler input kept `Ada Lovelace`, the persisted preview showed `[redacted]`. Aggregates are deliberately NOT masked except `min`/`max`: `sum(salary)` identifies nobody, but `max(email)` returns one customer's actual address. Fails **open** (an unresolvable name is left unmasked, never a legitimate column wrongly blanked), so this is a mitigation, not a guarantee. 15 new tests |
 | 46 | **F-8 prompt-injection resistance** — planner / assembler / report_summary | Rifqi | ✅ new | Shipped 2026-07-24. The three prompts that ingest customer data had **no injection rule**: `guardrails.md` is appended only in `chatbot.py`/`help.py`, and the `InputGuard` screens only the user's message — it never sees catalog or row content. This is the one attack the five query-defense layers structurally cannot see, because every IR the planner emits is individually valid. A hostile string only needs to reach a text column in the customer's OWN database (a product description, a support ticket, a form field) to be sampled into `sample_values` and rendered verbatim into the planner prompt. Fix = a purpose-written "content is data, never instructions" rule in each prompt + `<data>…</data>` delimiters around the catalog render and the run-state render. **Deliberately NOT `guardrails.md` wholesale** — its rules prescribe refusal sentences, and the planner's only free-text field is `infeasible_reason`, so those strings would surface there and could regress the Q2 data-gap path (tests pin their absence). Verified: planner eval **6/6, carried_over 5/5 green**; live hostile-catalog run planned only `t_products` and never touched the planted `employees.salary`; 10 new tests |
 | 47 | **Cheap-batch review fixes** — F-19, F-22, F-24, F-26, F-16, F-4 | Rifqi | ✅ new | Shipped 2026-07-24, one low-risk batch. **F-19:** `sources` was missing entirely on `check` + router-`help` and came *after* `status` on the slow path — contract said always-first; additive fix, contract updated. **F-22:** `analysis_id` now 422s unless it parses as a UUID, on **both** live endpoints (chat + help kept identical on purpose). **F-24:** traceability/chart writes are skipped, and logged, when `analysis_id` is falsy — Go `0007` declares `analysis_id UUID NOT NULL` (+ FK on charts), so `analysis_id or ""` failed the insert and the never-throw seam lost the row silently. **F-26:** `QueryResult.error` uses `str(e) or repr(e)` — a Fernet `InvalidToken` reached the assembler/traceability/report as an EMPTY string; falling back only when `str()` is empty changes no existing message. **F-16:** retrieval cache key gains `redis_prefix` (two envs on one Redis cross-served results). **F-4:** non-Postgres sources now refused at the executor — **zero blast radius today** (Go's `isSupportedActive` allows only `postgres`), a tripwire so nobody re-enables a path that has no read-only session and no `statement_timeout`; legacy branch commented out per house convention, orphaned import commented with it |
-| 48 | **floor_08 — floor/body disagreement** | Rifqi | ⬜ new | **Needs a decision, found while adding #34 coverage.** When a plan HAS an `analyze_*` step that FAILS but its upstream `retrieve_data` returned rows, the floor passes (`_produced_rows`) while the body rejects it (`has_reportable_result` keys on the failed analysis step). Verified by execution — it is the only shape where the two disagree. Because the "Attempted, Unresolved" section is commented out (`generator.py:527`), such a run leaves **no trace anywhere**: if it is the session's only run, the report generates but comes out empty and the business question renders "Unanswered" — exactly the #33 bug through a different door. Suggested fix: apply the floor's second arm only when the plan has NO analysis step, mirroring the body predicate; `floor_08` then flips to `expected_ready: false`. **Not changed — Rifqi owns the floor** |
+| 48 | **floor_08 — floor/body disagreement** | Rifqi | ✅ | **Decided + shipped 2026-07-24 (lead).** #34's row-producing arm was unconditional, so a plan that HAS an `analyze_*` step which FAILED still cleared the floor on the strength of its upstream fetch — while the body rejected it. Because the "Attempted, Unresolved" section is commented out, such a run left no trace: as a session's only run it produced an empty report with the business question "Unanswered" (the #33 bug via another door). Fix = the arm now applies **only when the plan has no analysis step**, mirroring `has_reportable_result`; the shared `_plan_has_analysis` helper means the two predicates can no longer drift on that question. `floor_08` flips to `expected_ready: false`. The INTENDED asymmetry is preserved and re-verified: a zero-row retrieval still fails the floor but passes the body (floor_03). Readiness eval **17/17** |
+| 49 | **F-5 timeout does not stop the customer's query** | Rifqi | ✅ new | Shipped 2026-07-24. `asyncio.wait_for` cancels the awaiting **coroutine**; the `to_thread` worker is not cancellable and runs to completion, holding a thread and a connection on the **customer's** database after we already answered "timed out". Two fixes. **(a) Dedicated bounded pool** — DB work moves to its own `ThreadPoolExecutor(50, 'dbexec')` via `run_in_executor`; abandoned workers previously accumulated in the shared default pool (`min(32, cpu_count+4)`) alongside every other `to_thread` caller, notably the tabular Parquet loader, so a few slow customer queries could stall unrelated work process-wide. **(b) Session hardening is no longer best-effort** — and a **latent bug** was found while reading it: both SETs shared one `try`, so a `statement_timeout` failure **skipped `default_transaction_read_only` entirely** and the connection served queries in a WRITABLE session behind a `logger.warning`. Now independent: read-only **fails the connection** if it cannot be set (a writable session against a customer DB is not something to degrade into); `statement_timeout` logs at **error** with `degraded_seam` but does not refuse service, since it is their I/O at risk rather than our correctness. ⚠️ **Blast radius to watch:** if any deployment currently fails that SET silently, its sources now fail loudly instead. Believed impossible (Neon accepts it as a SET — the existing comment documents this), but it is the one judgement call here. 7 new tests |
 
 **Reading `eval/readiness/results/` (note for future sessions).** Four files are dated
 2026-07-23. `…_150632.json` scores **4/15 (26.7%)** — that is **not** a product

eval/readiness/results/readiness_result_2026-07-24_094133.json

@@ -0,0 +1,278 @@
+{
+  "run": {
+    "timestamp": "2026-07-24T09:41:33",
+    "dataset": "readiness_dataset.json",
+    "target": "src/agents/report/readiness.is_report_ready",
+    "total": 17,
+    "passed": 17,
+    "accuracy": 1.0,
+    "runtime_avg_ms": 0.0
+  },
+  "alignment_gap": {
+    "count": 2,
+    "ids": [
+      "align_01",
+      "align_02"
+    ]
+  },
+  "by_group": {
+    "floor": {
+      "n": 8,
+      "passed": 8,
+      "accuracy": 1.0
+    },
+    "delta": {
+      "n": 5,
+      "passed": 5,
+      "accuracy": 1.0
+    },
+    "edge": {
+      "n": 1,
+      "passed": 1,
+      "accuracy": 1.0
+    },
+    "alignment": {
+      "n": 3,
+      "passed": 3,
+      "accuracy": 1.0
+    }
+  },
+  "cases": [
+    {
+      "id": "floor_01",
+      "group": "floor",
+      "expected_ready": false,
+      "got_ready": false,
+      "expected_missing": [
+        "at least one completed analysis"
+      ],
+      "got_missing": [
+        "at least one completed analysis"
+      ],
+      "correct": true,
+      "aligned": false,
+      "gap": false,
+      "latency_ms": 0.0
+    },
+    {
+      "id": "floor_02",
+      "group": "floor",
+      "expected_ready": false,
+      "got_ready": false,
+      "expected_missing": [
+        "at least one completed analysis"
+      ],
+      "got_missing": [
+        "at least one completed analysis"
+      ],
+      "correct": true,
+      "aligned": false,
+      "gap": false,
+      "latency_ms": 0.0
+    },
+    {
+      "id": "floor_03",
+      "group": "floor",
+      "expected_ready": false,
+      "got_ready": false,
+      "expected_missing": [
+        "at least one completed analysis"
+      ],
+      "got_missing": [
+        "at least one completed analysis"
+      ],
+      "correct": true,
+      "aligned": false,
+      "gap": false,
+      "latency_ms": 0.0
+    },
+    {
+      "id": "floor_04",
+      "group": "floor",
+      "expected_ready": true,
+      "got_ready": true,
+      "expected_missing": [],
+      "got_missing": [],
+      "correct": true,
+      "aligned": true,
+      "gap": false,
+      "latency_ms": 0.0
+    },
+    {
+      "id": "floor_05",
+      "group": "floor",
+      "expected_ready": true,
+      "got_ready": true,
+      "expected_missing": [],
+      "got_missing": [],
+      "correct": true,
+      "aligned": true,
+      "gap": false,
+      "latency_ms": 0.0
+    },
+    {
+      "id": "floor_06",
+      "group": "floor",
+      "expected_ready": true,
+      "got_ready": true,
+      "expected_missing": [],
+      "got_missing": [],
+      "correct": true,
+      "aligned": true,
+      "gap": false,
+      "latency_ms": 0.0
+    },
+    {
+      "id": "floor_07",
+      "group": "floor",
+      "expected_ready": true,
+      "got_ready": true,
+      "expected_missing": [],
+      "got_missing": [],
+      "correct": true,
+      "aligned": true,
+      "gap": false,
+      "latency_ms": 0.0
+    },
+    {
+      "id": "floor_08",
+      "group": "floor",
+      "expected_ready": false,
+      "got_ready": false,
+      "expected_missing": [
+        "at least one completed analysis"
+      ],
+      "got_missing": [
+        "at least one completed analysis"
+      ],
+      "correct": true,
+      "aligned": true,
+      "gap": false,
+      "latency_ms": 0.0
+    },
+    {
+      "id": "delta_01",
+      "group": "delta",
+      "expected_ready": false,
+      "got_ready": false,
+      "expected_missing": [
+        "a new analysis since the last report"
+      ],
+      "got_missing": [
+        "a new analysis since the last report"
+      ],
+      "correct": true,
+      "aligned": true,
+      "gap": false,
+      "latency_ms": 0.0
+    },
+    {
+      "id": "delta_02",
+      "group": "delta",
+      "expected_ready": true,
+      "got_ready": true,
+      "expected_missing": [],
+      "got_missing": [],
+      "correct": true,
+      "aligned": true,
+      "gap": false,
+      "latency_ms": 0.0
+    },
+    {
+      "id": "delta_03",
+      "group": "delta",
+      "expected_ready": true,
+      "got_ready": true,
+      "expected_missing": [],
+      "got_missing": [],
+      "correct": true,
+      "aligned": true,
+      "gap": false,
+      "latency_ms": 0.0
+    },
+    {
+      "id": "delta_04",
+      "group": "delta",
+      "expected_ready": false,
+      "got_ready": false,
+      "expected_missing": [
+        "a new analysis since the last report"
+      ],
+      "got_missing": [
+        "a new analysis since the last report"
+      ],
+      "correct": true,
+      "aligned": true,
+      "gap": false,
+      "latency_ms": 0.0
+    },
+    {
+      "id": "delta_05",
+      "group": "delta",
+      "expected_ready": false,
+      "got_ready": false,
+      "expected_missing": [
+        "a new analysis since the last report"
+      ],
+      "got_missing": [
+        "a new analysis since the last report"
+      ],
+      "correct": true,
+      "aligned": true,
+      "gap": false,
+      "latency_ms": 0.0
+    },
+    {
+      "id": "edge_01",
+      "group": "edge",
+      "expected_ready": false,
+      "got_ready": false,
+      "expected_missing": [
+        "at least one completed analysis"
+      ],
+      "got_missing": [
+        "at least one completed analysis"
+      ],
+      "correct": true,
+      "aligned": false,
+      "gap": false,
+      "latency_ms": 0.0
+    },
+    {
+      "id": "align_01",
+      "group": "alignment",
+      "expected_ready": true,
+      "got_ready": true,
+      "expected_missing": [],
+      "got_missing": [],
+      "correct": true,
+      "aligned": false,
+      "gap": true,
+      "latency_ms": 0.0
+    },
+    {
+      "id": "align_02",
+      "group": "alignment",
+      "expected_ready": true,
+      "got_ready": true,
+      "expected_missing": [],
+      "got_missing": [],
+      "correct": true,
+      "aligned": false,
+      "gap": true,
+      "latency_ms": 0.0
+    },
+    {
+      "id": "align_03",
+      "group": "alignment",
+      "expected_ready": true,
+      "got_ready": true,
+      "expected_missing": [],
+      "got_missing": [],
+      "correct": true,
+      "aligned": true,
+      "gap": false,
+      "latency_ms": 0.0
+    }
+  ]
+}

src/database_client/engine.py

@@ -136,19 +136,47 @@ class UserEngineCache:
         # connect event (not per query, so the pooling latency win stays). These are
         # ordinary SET commands, NOT libpq startup `options` — Neon's transaction
         # pooler rejects `default_transaction_read_only` as a startup parameter but
-        # accepts it as a SET. Best-effort: the authoritative read-only guarantee is
-        # the compiler (SELECT-only) + the sqlglot DML guard; statement_timeout is
-        # backed by the executor's asyncio.wait_for. So a failure here must not break
-        # the connection.
+        # accepts it as a SET.
+        #
+        # REWORKED 2026-07-24 (F-5). Three things were wrong here:
+        #
+        #  1. Both SETs shared one `try`, so a `statement_timeout` failure skipped
+        #     `default_transaction_read_only` entirely — the connection then served
+        #     queries in a WRITABLE session, silently, behind a `logger.warning`. They
+        #     are now independent.
+        #  2. The old comment said statement_timeout is "backed by the executor's
+        #     asyncio.wait_for". It is not. `wait_for` cancels the awaiting coroutine;
+        #     the `to_thread` worker underneath is NOT cancellable and runs to
+        #     completion, holding a connection on the customer's database. This SET is
+        #     therefore the ONLY real bound on how long a query burns their I/O.
+        #  3. Consequently neither SET is "best-effort" any more.
+        #
+        # `default_transaction_read_only` now FAILS THE CONNECTION if it cannot be set:
+        # a writable session against a customer database is not something to degrade
+        # into. `statement_timeout` logs at error but does not fail the connect — the
+        # query still runs bounded by LIMIT and the caller still stops waiting; it is
+        # their I/O at risk, not correctness, so refusing service outright would be a
+        # worse trade. Both are visible as `degraded_seam`.
         @event.listens_for(engine, "connect")
         def _init_session(dbapi_conn, _record):  # noqa: ANN001
+            cur = dbapi_conn.cursor()
             try:
-                cur = dbapi_conn.cursor()
-                cur.execute(f"SET statement_timeout = {_STATEMENT_TIMEOUT_MS}")
+                try:
+                    cur.execute(f"SET statement_timeout = {_STATEMENT_TIMEOUT_MS}")
+                except Exception as exc:  # noqa: BLE001
+                    # Not fatal, but the customer's server now has no cap on how long
+                    # our query runs after we have given up waiting for it.
+                    logger.error(
+                        "statement_timeout SET failed — customer queries are unbounded "
+                        "server-side on this connection",
+                        degraded_seam="db_statement_timeout_unset",
+                        error=repr(exc),
+                    )
+                # Deliberately NOT wrapped: if this raises, SQLAlchemy discards the
+                # connection and the query fails loudly. That is the intended outcome.
                 cur.execute("SET default_transaction_read_only = on")
+            finally:
                 cur.close()
-            except Exception as exc:  # noqa: BLE001 — best-effort session hardening
-                logger.warning("session init SET failed", error=str(exc))
 
         return engine
 

src/query/executor/db.py

@@ -21,6 +21,7 @@ from __future__ import annotations
 
 import asyncio
 import time
+from concurrent.futures import ThreadPoolExecutor
 from typing import Any
 
 import sqlglot
@@ -45,6 +46,25 @@ from .base import BaseExecutor, QueryResult
 logger = get_logger("db_executor")
 
 _QUERY_TIMEOUT_SECONDS = 30
+
+# Dedicated pool for blocking customer-DB work (F-5, 2026-07-24).
+#
+# `asyncio.wait_for` cancels the awaiting COROUTINE; the worker thread underneath is
+# not cancellable and runs to completion regardless. So every timed-out query leaves a
+# thread occupied until the customer's server gives up (bounded by the connection's
+# `statement_timeout`, which is best-effort — see engine.py). On the DEFAULT executor
+# those abandoned workers accumulate in a pool of `min(32, cpu_count + 4)` that is
+# shared with every other `to_thread` caller in the process — notably the tabular
+# Parquet loader. A handful of slow customer queries could therefore stall unrelated
+# work across the whole service.
+#
+# Isolating them means the blast radius of a slow customer database is queries against
+# THAT class of work, not the entire process. Sized to the engine cache's own ceiling
+# (_MAX_ENGINES=50 x _POOL_SIZE=1) — more threads than that cannot make progress
+# anyway, since each needs a pooled connection.
+_DB_THREAD_POOL = ThreadPoolExecutor(
+    max_workers=50, thread_name_prefix="dbexec"
+)
 _DBCLIENT_PREFIX = "dbclient://"
 
 
@@ -88,9 +108,20 @@ class DbExecutor(BaseExecutor):
                 )
             creds = decrypt_credentials_dict(client.credentials)
 
+            # `run_in_executor` on the dedicated pool, not `asyncio.to_thread` (which
+            # always uses the shared default executor). The timeout semantics are
+            # unchanged — `wait_for` still stops US waiting after 30s — but a worker
+            # abandoned by that timeout now occupies a DB-only thread. See
+            # _DB_THREAD_POOL. (F-5)
+            loop = asyncio.get_running_loop()
             columns, rows = await asyncio.wait_for(
-                asyncio.to_thread(
-                    self._run_sync, client_id, client.db_type, creds, compiled
+                loop.run_in_executor(
+                    _DB_THREAD_POOL,
+                    self._run_sync,
+                    client_id,
+                    client.db_type,
+                    creds,
+                    compiled,
                 ),
                 timeout=_QUERY_TIMEOUT_SECONDS,
             )