diff --git a/CLAUDE.md b/CLAUDE.md
index 5e9f7fd6c..311f01206 100644
--- a/CLAUDE.md
+++ b/CLAUDE.md
@@ -130,6 +130,11 @@ Key CLI flags for control-plane mode:
 - `--process-min-workers N` / `--process-max-workers N`
 - `--process-retire-on-session-end`
 - `--worker-queue-timeout DURATION` / `--worker-idle-timeout DURATION`
+- `--statement-timeout DURATION` — cancels any single statement that runs longer than this; `0`/empty (the default) leaves statements **unbounded**, which is today's behavior. Blast-radius containment only: nothing else in the stack bounds a wedged statement (DuckDB has no statement timeout, client `statement_timeout` is an ignored SET, the cnpg shards run `idle_in_transaction_session_timeout=0`, pgwire clients wait forever), so one engine deadlock cost a tenant 5h33m of their hourly refresh pipeline on 2026-09-16. The deadline rides on the STATEMENT context, never the connection context — a timeout ends one statement, not the session — and surfaces as `57014` with PostgreSQL's exact wording `canceling statement due to statement timeout` (drivers string-match it).
+  - **Every execution path must acquire the statement context through `queryContext`/`queryContextInner`.** That is the ONLY place `RegisterQuery` is called, so a path that skips it is reachable by neither the timeout nor a pgwire `CancelRequest`. Extended-protocol `Execute` used to skip it and called the context-LESS `executor.Query`/`Exec` (which run on `context.Background()` — see `PinnedExecutor`), leaving prepared statements — what pgx, psycopg3 and JDBC send — unbounded and uncancellable. Regression: `TestStatementTimeoutReachesExtendedProtocolExecute`. If you add an execution path, route it through `queryContext` and the `*Context` executor methods.
+  - **Timeout classification is driven by the ERROR and gated on the timeout being configured** (`statementTimedOut(err)`), never by stored per-connection state. An earlier design recorded the live statement context on the connection; it went sticky — once any statement timed out, later unrelated failures were reported as timeouts and their logging skipped. The gate also keeps a deployment with the feature OFF byte-for-byte identical, so internal deadlines (attach, exec, worker gRPC) do not start surfacing as `57014`. With the feature ON an internal deadline racing a statement can still be labelled a statement timeout — an accepted narrowing.
+  - **A cursor is bounded over its whole lifetime, not per `FETCH`**: `queryContextForCursor` takes ONE context at `DECLARE` and holds it across every `FETCH`, so a cursor open longer than the timeout dies — stricter than PostgreSQL. Deliberate (an unbounded cursor also pins a worker), but size the timeout with long-lived cursors in mind.
+  - Caveat: an engine that ignores cancellation (duckdb/duckdb#24961) will not stop work; the client is freed with a clear error but the worker can stay wedged — retiring a timed-out worker is a follow-up that must respect the destroy-before-reuse ordering in `SessionManager.DestroySession`.
 - `--idle-timeout DURATION` — connection idle timeout: a client connection with no traffic for this long is closed and its worker released to hot-idle (in control-plane mode an idle connection otherwise pins a worker forever). The close is preceded by a `FATAL` 57P05 ErrorResponse naming the effective timeout (matching PostgreSQL's `idle_session_timeout`), so clients see a reap, not a network fault. **Control-plane default is `5m`** (`server.DefaultControlPlaneIdleTimeout`; standalone defaults to `24h`); a negative value disables it. `server.New` applies the standalone default, so the control plane sets it explicitly before `InitMinimalServer` (which skips that defaulting).
 - `--memory-budget SIZE` (default 75% RAM) / `--memory-rebalance`
 - `--socket-dir /path` (process backend)
diff --git a/README.md b/README.md
index 7fae0f326..a0fcf9fe6 100644
--- a/README.md
+++ b/README.md
@@ -674,6 +674,8 @@ Options:
   -threads int             DuckDB threads per session
   -process-isolation       Enable process isolation (spawn child process per connection)
   -idle-timeout string     Connection idle timeout (e.g., '30m', '1h', '-1' to disable)
+  -statement-timeout string
+                          Cancel any single statement running longer than this (e.g., '60m'); empty/'0' = unbounded
   -mode string             Run mode: standalone (default), control-plane, duckdb-service, or reshard-runner
   -process-min-workers int Pre-warm process worker count at startup (control-plane mode, default 0)
   -process-max-workers int Max process workers, 0=auto-derived (control-plane mode)
diff --git a/configloader/file_config.go b/configloader/file_config.go
index d06f57f33..9b0d42ac1 100644
--- a/configloader/file_config.go
+++ b/configloader/file_config.go
@@ -25,6 +25,7 @@ type FileConfig struct {
 	ProcessIsolation     bool                `yaml:"process_isolation"`
 	IdleTimeout          string              `yaml:"idle_timeout"`
 	SessionInitTimeout   string              `yaml:"session_init_timeout"`
+	StatementTimeout     string              `yaml:"statement_timeout"`
 	MemoryLimit          string              `yaml:"memory_limit"`
 	Threads              int                 `yaml:"threads"`
 	MemoryBudget         string              `yaml:"memory_budget"`
diff --git a/configresolve/cliflags.go b/configresolve/cliflags.go
index 7d3c2b401..041f4bc4c 100644
--- a/configresolve/cliflags.go
+++ b/configresolve/cliflags.go
@@ -31,6 +31,7 @@ func RegisterCLIInputsFlags(fs *flag.FlagSet) func() CLIInputs {
 	processIsolation := fs.Bool("process-isolation", false, "Enable process isolation (spawn child process per connection)")
 	idleTimeout := fs.String("idle-timeout", "", "Connection idle timeout: close a connection idle (no traffic) this long, freeing its worker (e.g., '30m', '1h', '-1s' to disable). Default 24h standalone, 5m control-plane where idle connections pin a worker (env: DUCKGRES_IDLE_TIMEOUT)")
 	sessionInitTimeout := fs.String("session-init-timeout", "", "Session startup metadata/probe timeout (e.g., '10s', '30s') (env: DUCKGRES_SESSION_INIT_TIMEOUT)")
+	statementTimeout := fs.String("statement-timeout", "", "Cancel any single statement that runs longer than this (e.g., '60m'); empty or '0' leaves statements unbounded (env: DUCKGRES_STATEMENT_TIMEOUT)")
 	memoryLimit := fs.String("memory-limit", "", "DuckDB memory_limit per session (e.g., '4GB') (env: DUCKGRES_MEMORY_LIMIT)")
 	threads := fs.Int("threads", 0, "DuckDB threads per session (env: DUCKGRES_THREADS)")
 	memoryBudget := fs.String("memory-budget", "", "Total memory for all DuckDB sessions (e.g., '24GB') (env: DUCKGRES_MEMORY_BUDGET)")
@@ -85,6 +86,7 @@ func RegisterCLIInputsFlags(fs *flag.FlagSet) func() CLIInputs {
 		cli.ProcessIsolation = *processIsolation
 		cli.IdleTimeout = *idleTimeout
 		cli.SessionInitTimeout = *sessionInitTimeout
+		cli.StatementTimeout = *statementTimeout
 		cli.MemoryLimit = *memoryLimit
 		cli.Threads = *threads
 		cli.MemoryBudget = *memoryBudget
diff --git a/configresolve/cliflags_test.go b/configresolve/cliflags_test.go
index 1af53a807..b633b5629 100644
--- a/configresolve/cliflags_test.go
+++ b/configresolve/cliflags_test.go
@@ -64,6 +64,7 @@ func fieldNameToFlagName(name string) string {
 		{"FilePersistence", "file-persistence"},
 		{"IdleTimeout", "idle-timeout"},
 		{"SessionInitTimeout", "session-init-timeout"},
+		{"StatementTimeout", "statement-timeout"},
 		{"MemoryLimit", "memory-limit"},
 		{"MemoryBudget", "memory-budget"},
 		{"MemoryRebalance", "memory-rebalance"},
diff --git a/configresolve/resolve.go b/configresolve/resolve.go
index 5643118f5..1ffe089fd 100644
--- a/configresolve/resolve.go
+++ b/configresolve/resolve.go
@@ -36,6 +36,7 @@ type CLIInputs struct {
 	ProcessIsolation            bool
 	IdleTimeout                 string
 	SessionInitTimeout          string
+	StatementTimeout            string
 	MemoryLimit                 string
 	Threads                     int
 	MemoryBudget                string
@@ -89,6 +90,7 @@ type Resolved struct {
 	ProcessMaxWorkers               int
 	ProcessRetireOnSessionEnd       bool
 	SessionInitTimeout              time.Duration
+	StatementTimeout                time.Duration
 	WorkerQueueTimeout              time.Duration
 	WorkerIdleTimeout               time.Duration
 	HandoverDrainTimeout            time.Duration
@@ -340,6 +342,13 @@ func ResolveEffective(fileCfg *configloader.FileConfig, cli CLIInputs, getenv fu
 				warn("Invalid idle_timeout duration: " + err.Error())
 			}
 		}
+		if fileCfg.StatementTimeout != "" {
+			if d, err := time.ParseDuration(fileCfg.StatementTimeout); err == nil {
+				cfg.StatementTimeout = d
+			} else {
+				warn("Invalid statement_timeout duration: " + err.Error())
+			}
+		}
 		if fileCfg.SessionInitTimeout != "" {
 			if d, err := time.ParseDuration(fileCfg.SessionInitTimeout); err == nil {
 				cfg.SessionInitTimeout = d
@@ -595,6 +604,13 @@ func ResolveEffective(fileCfg *configloader.FileConfig, cli CLIInputs, getenv fu
 			warn("Invalid DUCKGRES_CLIENT_IDLE_TIMEOUT_MAX: must be a positive duration")
 		}
 	}
+	if v := getenv("DUCKGRES_STATEMENT_TIMEOUT"); v != "" {
+		if d, err := time.ParseDuration(v); err == nil {
+			cfg.StatementTimeout = d
+		} else {
+			warn("Invalid DUCKGRES_STATEMENT_TIMEOUT duration: " + err.Error())
+		}
+	}
 	if v := getenv("DUCKGRES_SESSION_INIT_TIMEOUT"); v != "" {
 		if d, err := time.ParseDuration(v); err == nil {
 			cfg.SessionInitTimeout = d
@@ -942,6 +958,13 @@ func ResolveEffective(fileCfg *configloader.FileConfig, cli CLIInputs, getenv fu
 			warn("Invalid --idle-timeout duration: " + err.Error())
 		}
 	}
+	if cli.Set["statement-timeout"] {
+		if d, err := time.ParseDuration(cli.StatementTimeout); err == nil {
+			cfg.StatementTimeout = d
+		} else {
+			warn("Invalid --statement-timeout duration: " + err.Error())
+		}
+	}
 	if cli.Set["session-init-timeout"] {
 		if d, err := time.ParseDuration(cli.SessionInitTimeout); err == nil {
 			cfg.SessionInitTimeout = d
diff --git a/docs/plans/2026-09-16-warehouse-reliability-plan.md b/docs/plans/2026-09-16-warehouse-reliability-plan.md
new file mode 100644
index 000000000..27406cd06
--- /dev/null
+++ b/docs/plans/2026-09-16-warehouse-reliability-plan.md
@@ -0,0 +1,131 @@
+# Managed-warehouse reliability: root causes and fix plan (2026-09-16)
+
+Three distinct production failures were investigated on `mw-prod-us`. They were
+initially conflated; they are unrelated and have different fixes.
+
+---
+
+## 1. Portola sqlmesh `COMMIT ... SSL error: unexpected eof`  — FIX IN FLIGHT
+
+**Root cause (confirmed by source + measurement).**
+`information_schema.tables` is defined as `... FROM duckdb_tables()`, and
+`duckdb_tables()` calls `TableCatalogEntry::GetStorageInfo` for EVERY table in its
+row loop purely to fill `estimated_size`/`index_count`. In DuckLake that landed on
+`GetTableStats`, whose metadata query joins `ducklake_table_column_stats` and is
+filtered `WHERE table_id = <N>` — one metadata round-trip per table, shipping every
+column's min/max/extra_stats, to extract one integer (`record_count`). The stats
+cache is keyed `<next_file_id, table_id>`, so every commit that writes a file
+invalidates all of it.
+
+**Measured** (portola catalog: 4,497 tables / 516,944 column-stats rows):
+
+| | queries | rows | server-side |
+|---|---|---|---|
+| Current | 4,500 | 516,971 | 185 ms |
+| Batched, cardinality-only | 1 | 4,500 | 1 ms |
+
+Server execution is 185 ms — the cost is **not** database work, it is 4,500
+sequential round-trips. ~1.25 ms/query on a local socket (5.6s floor with no
+network); 4.4–9.1 ms/query through a pooler, which reproduces the observed
+20–41s listing exactly. Scales linearly with table count.
+
+The listing holds a read transaction open on portola's ATTACHed Postgres the
+whole time; their `portola_warehouse` role caps `idle_in_transaction_session_timeout`
+at 30s, so Postgres kills it and DuckDB reports SSL EOF on COMMIT.
+
+**Fix:** PostHog/ducklake#43 (base `posthog/v1.5.5`). Cardinality-only batch path;
+full `GetTableStats` untouched so query plans cannot change.
+
+**Remaining work:**
+- [ ] CI green on #43 (never built locally)
+- [ ] Time a real listing on a large catalog: confirm 20–41s → sub-second
+- [ ] Tag the fork; bump `DUCKLAKE_EXTENSION_TAG` in `Dockerfile` + `Dockerfile.worker`
+- [ ] e2e assertion per the repo testing contract
+- [ ] Promote to prod, then confirm EOFs stop
+
+**Fallback if that slips:** raise the idle-in-transaction window from our side.
+`PostHog/duckdb-postgres` already has `pg_idle_in_transaction_timeout_millis`, but
+it is (a) never set by duckgres and (b) only applied in `PostgresScanConnect` —
+NOT on the `PostgresTransaction` path that catalog listings actually use. The SET
+must be issued *inside* the transaction (PgBouncer transaction pooling discards
+session state between transactions, not within one), and the GUC is `USERSET`, so
+a client may raise it above a role default without touching the customer's DB.
+**It must raise only, never lower, and must preserve `0` (unlimited)** — the
+DuckLake metadata store runs unlimited today and a legitimate posthog data-import
+held an idle metadata transaction for 2h26m. A blanket 120s would have killed it.
+
+Portola's own PR (portolans/backend#9397) does the same thing via
+`ALTER ROLE portola_warehouse SET idle_in_transaction_session_timeout = '120s'`.
+That is a valid stopgap but needs their DBA, and does not survive catalog growth.
+
+---
+
+## 2. Portola 5.5h refresh wedge — in-engine DuckDB deadlock  — NOT FIXED
+
+**Root cause.** Worker `281869` ran
+`INSERT INTO sqlmesh__reports.reports__cohorts_dist_d7_daily...` — `cohorts_dist.sql`,
+"the expensive half: QUANTILE_CONT and percentile aggregations" over **18 GROUPING
+SETS**, a heavy blocking collapse. It did ~6 min of real work (09:44–09:50, 2 cores)
+then sat at **0.075 cores / ~380 B/s (TCP keepalive only) for ~5 hours**. All 62
+threads parked in futex/nanosleep, none in a socket read; gdb recovers only Go
+`runtime.futex` frames. Query duration logged **19,994,187 ms (5h33m)**, ended only
+by the GitHub Actions 6h job timeout.
+
+Worker was 20 min old (fresh credentials) and portola has zero `ExpiredToken` — this
+is not a credential problem. Matches upstream duckdb/duckdb#24961: blocking-collapse
+query hangs, `interrupt()` never honored, all threads park, CPU→0, needs SIGKILL.
+
+**Blast radius.** `sqlmesh-deploy.yml` uses concurrency group `sqlmesh-prod` with
+`cancel-in-progress: false`, so five consecutive hourly refreshes queued behind the
+wedge and were cancelled.
+
+**Plan:**
+- [ ] **Containment first** — a statement timeout in duckgres. Nothing bounds a wedged
+      statement today: no duckgres statement timeout, `idle_in_transaction_session_timeout=0`
+      on the shards, and sqlmesh/psycopg wait forever. This is the single highest-value
+      change for blast radius and is independent of the engine bug.
+- [ ] Track duckdb/duckdb#24961; ship an engine build with the fix
+- [ ] Workaround for portola: reduce grouping sets / split the model
+
+---
+
+## 3. PostHog Metabase failures — expired STS token on a pinned worker  — NOT FIXED
+
+**Root cause.** Worker `279924` started 18:16 and held ONE long-lived Metabase
+connection for ~19h. Per-tenant STS credentials expire (~1h), but worker-side
+refresh (`RefreshS3Secret` via `SessionPool.reuseExistingActivation`) only fires when
+a worker is **reused** for a new session or reclaimed from hot-idle — never
+periodically for a worker continuously busy with one session. The per-connection
+`StartCredentialRefresh` (5-min ticker) is deliberately not started on the
+remote/sharedDB path ("the pool manages it") — but the pool only refreshes on reuse.
+
+Result: 74 failures 00:04→13:35, **each ~256 s then fail**, with
+`HTTP GET ... ExpiredToken: The provided token has expired`. Self-resolved only when
+the worker was finally recycled.
+
+The 256 s is the httpfs retry budget duckgres sets to ride out S3 503 SlowDown
+(`applyHTTPFSRetryBudget`: retries=10, wait=500ms, backoff=2 ≈ 255s cumulative). That
+budget is being spent on a **non-retryable** auth error. The httpfs fork does have
+in-engine `RunWithCredentialRefresh` → `TryRefreshAuthParams`, but it only retries
+once and only if something committed fresh credentials — nothing did.
+
+**Plan:**
+- [ ] Refresh STS credentials for long-lived **active** sessions, not just on reuse
+- [ ] Do not spend the full retry budget on non-retryable auth 400s — fail fast or
+      force a secret refresh
+- [ ] Consider a max worker/session age so a pinned client connection cannot outlive
+      its credentials indefinitely
+
+---
+
+## Cross-cutting
+
+**Nothing bounds a hung or doomed statement anywhere in the stack.** Items 2 and 3
+are different bugs with the same amplifier. A statement timeout plus an
+idle-in-transaction bound would have turned a 5.5h outage into a bounded failure and
+a 256 s stall into an immediate error. Highest-leverage single change after #43.
+
+**Portola-side (their repo, no DBA needed):** 3,987 of 5,940 tables are dead sqlmesh
+dev environments. Invalidating unused envs so the hourly janitor drops them is what
+makes listings cheap permanently — the engine fix removes the per-table round-trip,
+but catalog size still drives everything else.
diff --git a/server/conn.go b/server/conn.go
index 4c316d736..638534f37 100644
--- a/server/conn.go
+++ b/server/conn.go
@@ -134,6 +134,12 @@ type portalExec struct {
 	// finishProfiling runs after rows.Close has allowed a Flight DoGet trailer
 	// to arrive. A suspended portal retains it until its terminal Execute.
 	finishProfiling func()
+	// releaseStmtCtx cancels/unregisters the statement context whose deadline
+	// bounds this portal. A suspended portal's RowSet outlives the Execute that
+	// opened it, so — like finishProfiling — the portal owns the release until
+	// its terminal Execute. Cancelling at the opening Execute's return would
+	// tear down the rowset the next Execute resumes from.
+	releaseStmtCtx func()
 }
 
 // closeExec releases a suspended portal's open rowset (if any). Must be
@@ -147,6 +153,9 @@ func (p *portal) closeExec() {
 	if p.exec.finishProfiling != nil {
 		p.exec.finishProfiling()
 	}
+	if p.exec.releaseStmtCtx != nil {
+		p.exec.releaseStmtCtx()
+	}
 	p.exec = nil
 }
 
@@ -488,7 +497,16 @@ func (c *clientConn) queryContextInner(monitor bool) (context.Context, func()) {
 	// Carry the statement's ID to the engine. The worker stamps it on its own
 	// logs, so a client complaint can be followed from this connection into the
 	// pod that ran the statement.
-	ctx, cancel := context.WithCancel(wire.WithQueryID(c.ctx, c.currentQueryID()))
+	// The statement timeout rides on the statement context, not the connection
+	// context: a timeout must end one statement, not the session. That is also
+	// why statementTimedOut() cannot be answered from c.ctx -- see conn_errors.go.
+	var ctx context.Context
+	var cancel context.CancelFunc
+	if timeout := c.statementTimeout(); timeout > 0 {
+		ctx, cancel = context.WithTimeout(wire.WithQueryID(c.ctx, c.currentQueryID()), timeout)
+	} else {
+		ctx, cancel = context.WithCancel(wire.WithQueryID(c.ctx, c.currentQueryID()))
+	}
 	key := c.backendKey()
 	c.server.RegisterQuery(key, cancel)
 
@@ -1867,7 +1885,7 @@ func (c *clientConn) handleQuery(body []byte) (retErr error) {
 				errCode := classifyErrorCode(err)
 				errMsg := err.Error()
 				if c.isCallerCancellation(err) {
-					errMsg = "canceling statement due to user request"
+					errMsg = c.cancellationMessage(err)
 				} else {
 					c.logQueryError(query, err)
 				}
diff --git a/server/conn_cursor.go b/server/conn_cursor.go
index dccba0e1d..d69e15dab 100644
--- a/server/conn_cursor.go
+++ b/server/conn_cursor.go
@@ -306,7 +306,7 @@ func (c *clientConn) handleFetchCursor(query string, stmt *pg_query.FetchStmt) e
 			errMsg := err.Error()
 			if c.isCallerCancellation(err) {
 				errCode = "57014"
-				errMsg = "canceling statement due to user request"
+				errMsg = c.cancellationMessage(err)
 			}
 			c.sendError("ERROR", errCode, errMsg)
 			c.setTxError()
@@ -381,7 +381,7 @@ func (c *clientConn) handleFetchCursor(query string, stmt *pg_query.FetchStmt) e
 		errMsg := err.Error()
 		if c.isCallerCancellation(err) {
 			errCode = "57014"
-			errMsg = "canceling statement due to user request"
+			errMsg = c.cancellationMessage(err)
 		}
 		c.sendError("ERROR", errCode, errMsg)
 		c.setTxError()
@@ -447,7 +447,7 @@ func (c *clientConn) handleFetchCursorExtended(p *portal) {
 	if cursor.rows == nil {
 		if err := c.openCursor(cursor); err != nil {
 			if c.isCallerCancellation(err) {
-				c.sendError("ERROR", "57014", "canceling statement due to user request")
+				c.sendError("ERROR", "57014", c.cancellationMessage(err))
 			} else {
 				c.sendError("ERROR", "42000", err.Error())
 			}
@@ -506,7 +506,7 @@ func (c *clientConn) handleFetchCursorExtended(p *portal) {
 
 	if err := cursor.rows.Err(); err != nil {
 		if c.isCallerCancellation(err) {
-			c.sendError("ERROR", "57014", "canceling statement due to user request")
+			c.sendError("ERROR", "57014", c.cancellationMessage(err))
 		} else {
 			c.sendError("ERROR", "42000", err.Error())
 		}
diff --git a/server/conn_errors.go b/server/conn_errors.go
index bf80f2b25..084fc2680 100644
--- a/server/conn_errors.go
+++ b/server/conn_errors.go
@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"strings"
+	"time"
 )
 
 // isQueryCancelled checks whether an error string indicates that *some*
@@ -18,6 +19,49 @@ func isQueryCancelled(err error) bool {
 	return err == context.Canceled || (err != nil && strings.Contains(err.Error(), "context canceled"))
 }
 
+// statementTimeout returns the effective per-statement timeout, or 0 when
+// statements are unbounded.
+func (c *clientConn) statementTimeout() time.Duration {
+	if c == nil || c.server == nil {
+		return 0
+	}
+	return c.server.cfg.StatementTimeout
+}
+
+// statementTimedOut reports whether err is this statement hitting the configured
+// statement timeout.
+//
+// Classified from the ERROR, not from stored state. An earlier version recorded
+// the live statement context on the connection and inspected it here, which was
+// sticky: once any statement timed out, a later unrelated failure on a path that
+// did not overwrite the field was reported as a timeout and its logging skipped.
+//
+// Gated on the timeout being configured, so a deployment with the feature off
+// keeps today's classification byte for byte -- internal deadlines (attach,
+// exec, worker gRPC) must not start surfacing as 57014 just because this code
+// exists. When the feature IS on, an internal deadline racing a statement can
+// still be labelled a statement timeout; that is an accepted narrowing, not an
+// oversight.
+func (c *clientConn) statementTimedOut(err error) bool {
+	if c == nil || err == nil || c.statementTimeout() <= 0 {
+		return false
+	}
+	if errors.Is(err, context.DeadlineExceeded) {
+		return true
+	}
+	return strings.Contains(err.Error(), "context deadline exceeded")
+}
+
+// cancellationMessage returns the client-facing 57014 message. The wording
+// mirrors PostgreSQL's exactly, because drivers and ORMs string-match it to
+// decide whether a failure is retryable.
+func (c *clientConn) cancellationMessage(err error) string {
+	if c.statementTimedOut(err) {
+		return "canceling statement due to statement timeout"
+	}
+	return "canceling statement due to user request"
+}
+
 // isCallerCancellation reports whether err is a cancellation that the caller
 // asked for — either through pgwire CancelRequest, an explicit ctx cancel, or
 // a deadline. Distinct from a gRPC Canceled status that bubbles up purely
@@ -27,10 +71,21 @@ func isQueryCancelled(err error) bool {
 // cancelled". This matters for alerting — "Query execution errored." should
 // fire on worker kills, not get silently downgraded to "Worker statement finished.".
 func (c *clientConn) isCallerCancellation(err error) bool {
+	if c == nil {
+		return false
+	}
+	if c.statementTimedOut(err) {
+		return true
+	}
 	if !isQueryCancelled(err) {
 		return false
 	}
-	if c == nil || c.ctx == nil {
+	// A statement timeout is caller-driven even though the connection context is
+	// untouched, so it must not be logged as an infra failure.
+	if c.statementTimedOut(err) {
+		return true
+	}
+	if c.ctx == nil {
 		return false
 	}
 	return c.ctx.Err() != nil
diff --git a/server/conn_extended_query.go b/server/conn_extended_query.go
index 6cd8ca821..57fa8843e 100644
--- a/server/conn_extended_query.go
+++ b/server/conn_extended_query.go
@@ -852,6 +852,28 @@ func (c *clientConn) handleExecute(body []byte) {
 		c.logWorkerStatementFinished(workerStatement, queryStart, queryRowsAff, queryFinalErr)
 	}()
 
+	// Extended-protocol Execute needs a real statement context, like the simple
+	// and batched paths. Without it this path ran on context.Background() (see
+	// PinnedExecutor.Query/Exec), which meant prepared statements — what pgx,
+	// psycopg3 and JDBC actually send — were reachable by NEITHER the statement
+	// timeout NOR a pgwire CancelRequest: RegisterQuery is only ever called from
+	// queryContextInner, and nothing on this path called it.
+	// queryContextInner(false): no disconnect monitor. Like the cursor path, the
+	// monitor's bufio Peek would race the message loop's own reads — the extended
+	// protocol pipelines Parse/Bind/Execute/Sync, so bytes for the next message
+	// are often already buffered. Disconnect still cancels via c.ctx.
+	stmtCtx, stmtCleanup := c.queryContextInner(false)
+	defer func() {
+		// A suspended portal keeps its RowSet open for the NEXT Execute, so the
+		// statement context has to outlive this handler; ownership moves to the
+		// portal and closeExec releases it.
+		if p.exec != nil {
+			p.exec.releaseStmtCtx = stmtCleanup
+			return
+		}
+		stmtCleanup()
+	}()
+
 	if !returnsResults {
 		// Open cursors pin the session's single DuckDB connection — release
 		// them before a transaction-end statement needs it.
@@ -859,12 +881,12 @@ func (c *clientConn) handleExecute(body []byte) {
 
 		// Non-result-returning query: use Exec with converted query
 		runExec := func() (ExecResult, error) {
-			result, err := c.executor.Exec(convertedQuery, args...)
+			result, err := c.executor.ExecContext(stmtCtx, convertedQuery, args...)
 			if err != nil {
 				if fallbackResult, handled, fallbackErr := c.execCompatibilityFallback(convertedQuery, err, func(fallbackQuery string) (ExecResult, error) {
 					return c.runGeneratedWorkerStatement(
 						generatedWorkerStatement(workerOriginRewrite, workerOperationCompatibilityFallback),
-						func() (ExecResult, error) { return c.executor.Exec(fallbackQuery, args...) },
+						func() (ExecResult, error) { return c.executor.ExecContext(stmtCtx, fallbackQuery, args...) },
 					)
 				}); handled {
 					return fallbackResult, fallbackErr
@@ -899,7 +921,7 @@ func (c *clientConn) handleExecute(body []byte) {
 				errCode := classifyErrorCode(err)
 				errMsg := err.Error()
 				if c.isCallerCancellation(err) {
-					errMsg = "canceling statement due to user request"
+					errMsg = c.cancellationMessage(err)
 				} else {
 					c.logQueryError(convertedQuery, err)
 				}
@@ -923,7 +945,7 @@ func (c *clientConn) handleExecute(body []byte) {
 
 	// Result-returning query: use Query with converted query
 	runQuery := func() (RowSet, error) {
-		return c.executor.Query(convertedQuery, args...)
+		return c.executor.QueryContext(stmtCtx, convertedQuery, args...)
 	}
 
 	execStart := time.Now()
@@ -967,7 +989,7 @@ func (c *clientConn) handleExecute(body []byte) {
 		errCode := classifyErrorCode(err)
 		errMsg := err.Error()
 		if c.isCallerCancellation(err) {
-			errMsg = "canceling statement due to user request"
+			errMsg = c.cancellationMessage(err)
 		} else {
 			c.logQueryError(convertedQuery, err)
 		}
@@ -1082,9 +1104,9 @@ func (c *clientConn) handleExecute(body []byte) {
 		queryFinalErr = err
 		errCode := "42000"
 		errMsg := err.Error()
-		if c.isCallerCancellation(err) {
+		if c.statementTimedOut(err) || c.isCallerCancellation(err) {
 			errCode = "57014"
-			errMsg = "canceling statement due to user request"
+			errMsg = c.cancellationMessage(err)
 			c.sendError("ERROR", errCode, errMsg)
 		} else {
 			c.logger().Error("Row iteration error.", "error", err)
@@ -1168,9 +1190,9 @@ func (c *clientConn) resumeSuspendedPortal(p *portal, maxRows int32) {
 		p.closeExec()
 		errCode := "42000"
 		errMsg := err.Error()
-		if c.isCallerCancellation(err) {
+		if c.statementTimedOut(err) || c.isCallerCancellation(err) {
 			errCode = "57014"
-			errMsg = "canceling statement due to user request"
+			errMsg = c.cancellationMessage(err)
 		} else {
 			c.logger().Error("Row iteration error.", "error", err)
 		}
diff --git a/server/conn_query_exec.go b/server/conn_query_exec.go
index 5d148b0a4..813803e9e 100644
--- a/server/conn_query_exec.go
+++ b/server/conn_query_exec.go
@@ -70,7 +70,7 @@ func (c *clientConn) executeQueryDirect(query, cmdType string) error {
 			errCode := classifyErrorCode(err)
 			errMsg := err.Error()
 			if c.isCallerCancellation(err) {
-				errMsg = "canceling statement due to user request"
+				errMsg = c.cancellationMessage(err)
 			} else {
 				c.logQueryError(query, err)
 			}
@@ -246,7 +246,7 @@ func (c *clientConn) executeSelectQuery(query string, cmdType string, workerStat
 		errCode := classifyErrorCode(err)
 		errMsg := err.Error()
 		if c.isCallerCancellation(err) {
-			errMsg = "canceling statement due to user request"
+			errMsg = c.cancellationMessage(err)
 		} else {
 			c.logQueryError(query, err)
 		}
@@ -362,7 +362,7 @@ func (c *clientConn) executeSelectQuery(query string, cmdType string, workerStat
 		errMsg := err.Error()
 		if c.isCallerCancellation(err) {
 			errCode = "57014"
-			errMsg = "canceling statement due to user request"
+			errMsg = c.cancellationMessage(err)
 		} else {
 			c.logQueryError(query, err)
 		}
@@ -569,7 +569,7 @@ func (c *clientConn) executeSingleStatement(query string) (errSent bool, fatalEr
 			if cursor.rows == nil {
 				if err := c.openCursor(cursor); err != nil {
 					if c.isCallerCancellation(err) {
-						c.sendError("ERROR", "57014", "canceling statement due to user request")
+						c.sendError("ERROR", "57014", c.cancellationMessage(err))
 					} else {
 						c.sendError("ERROR", "42000", err.Error())
 					}
@@ -618,7 +618,7 @@ func (c *clientConn) executeSingleStatement(query string) (errSent bool, fatalEr
 			}
 			if err := cursor.rows.Err(); err != nil {
 				if c.isCallerCancellation(err) {
-					c.sendError("ERROR", "57014", "canceling statement due to user request")
+					c.sendError("ERROR", "57014", c.cancellationMessage(err))
 				} else {
 					c.sendError("ERROR", "42000", err.Error())
 				}
@@ -865,7 +865,7 @@ func (c *clientConn) executeSingleStatement(query string) (errSent bool, fatalEr
 				errCode := classifyErrorCode(err)
 				errMsg := err.Error()
 				if c.isCallerCancellation(err) {
-					errMsg = "canceling statement due to user request"
+					errMsg = c.cancellationMessage(err)
 				} else {
 					c.logQueryError(executedQuery, err)
 				}
@@ -917,7 +917,7 @@ func (c *clientConn) executeSingleStatement(query string) (errSent bool, fatalEr
 		errCode := classifyErrorCode(err)
 		errMsg := err.Error()
 		if c.isCallerCancellation(err) {
-			errMsg = "canceling statement due to user request"
+			errMsg = c.cancellationMessage(err)
 		} else {
 			c.logQueryError(executedQuery, err)
 		}
diff --git a/server/conn_querylog_feedback_test.go b/server/conn_querylog_feedback_test.go
index f2947762f..2d4e6d74a 100644
--- a/server/conn_querylog_feedback_test.go
+++ b/server/conn_querylog_feedback_test.go
@@ -16,8 +16,13 @@ type feedbackExecutor struct {
 	queryFn func(query string, args ...any) (RowSet, error)
 }
 
-func (e *feedbackExecutor) QueryContext(_ context.Context, _ string, _ ...any) (RowSet, error) {
-	return nil, errors.New("not implemented")
+// Execution paths must run through the context-aware methods so the statement
+// timeout and CancelRequest reach them, so route the fixture's queryFn here.
+func (e *feedbackExecutor) QueryContext(_ context.Context, query string, args ...any) (RowSet, error) {
+	if e.queryFn == nil {
+		return nil, errors.New("not implemented")
+	}
+	return e.queryFn(query, args...)
 }
 
 func (e *feedbackExecutor) ExecContext(_ context.Context, _ string, _ ...any) (ExecResult, error) {
@@ -168,15 +173,29 @@ func TestHandleExecuteLogsRowsErr(t *testing.T) {
 	// Execute body: empty portal name + maxRows=0
 	c.handleExecute([]byte{0, 0, 0, 0, 0})
 
-	select {
-	case entry := <-ql.ch:
-		if entry.ExceptionCode != "42000" {
-			t.Fatalf("expected exception code 42000, got %q", entry.ExceptionCode)
-		}
-		if entry.Exception != "row iteration failed" {
-			t.Fatalf("expected row iteration error, got %q", entry.Exception)
+	// Extended Execute now acquires a statement context (so the timeout and
+	// CancelRequest reach it), which also emits a QueryStart entry ahead of the
+	// completion entry. Scan for the completion entry rather than assuming the
+	// first one off the channel is it.
+	var found bool
+	for drained := 0; drained < 8 && !found; drained++ {
+		select {
+		case entry := <-ql.ch:
+			if entry.ExceptionCode == "" && entry.Exception == "" {
+				continue // QueryStart
+			}
+			if entry.ExceptionCode != "42000" {
+				t.Fatalf("expected exception code 42000, got %q", entry.ExceptionCode)
+			}
+			if entry.Exception != "row iteration failed" {
+				t.Fatalf("expected row iteration error, got %q", entry.Exception)
+			}
+			found = true
+		default:
+			t.Fatal("expected query log entry for rows.Err path")
 		}
-	default:
-		t.Fatal("expected query log entry for rows.Err path")
+	}
+	if !found {
+		t.Fatal("no completion query log entry for rows.Err path")
 	}
 }
diff --git a/server/conn_querylog_lifecycle_test.go b/server/conn_querylog_lifecycle_test.go
index bd896ab42..4bdfe093c 100644
--- a/server/conn_querylog_lifecycle_test.go
+++ b/server/conn_querylog_lifecycle_test.go
@@ -365,16 +365,29 @@ func TestHandleExecuteLogsProfileAfterResultStreamCloses(t *testing.T) {
 
 	c.handleExecute(append([]byte("p1\x00"), 0, 0, 0, 0))
 
-	select {
-	case entry := <-c.server.queryLogger.ch:
-		if entry.CPUTimeSeconds != 2.5 {
-			t.Fatalf("CPUTimeSeconds = %f, want 2.5", entry.CPUTimeSeconds)
+	// Extended Execute now acquires a statement context, which also emits a
+	// QueryStart entry ahead of the terminal one. Scan for the terminal entry
+	// (the one carrying profiling) rather than taking the first off the channel.
+	var terminal bool
+	for drained := 0; drained < 8 && !terminal; drained++ {
+		select {
+		case entry := <-c.server.queryLogger.ch:
+			if entry.CPUTimeSeconds == 0 && entry.PeakBufferMemoryBytes == 0 {
+				continue // QueryStart
+			}
+			if entry.CPUTimeSeconds != 2.5 {
+				t.Fatalf("CPUTimeSeconds = %f, want 2.5", entry.CPUTimeSeconds)
+			}
+			if entry.PeakBufferMemoryBytes != 8192 {
+				t.Fatalf("PeakBufferMemoryBytes = %d, want 8192", entry.PeakBufferMemoryBytes)
+			}
+			terminal = true
+		default:
+			t.Fatal("expected terminal query-log entry")
 		}
-		if entry.PeakBufferMemoryBytes != 8192 {
-			t.Fatalf("PeakBufferMemoryBytes = %d, want 8192", entry.PeakBufferMemoryBytes)
-		}
-	default:
-		t.Fatal("expected terminal query-log entry")
+	}
+	if !terminal {
+		t.Fatal("no terminal query-log entry carrying profiling")
 	}
 }
 
diff --git a/server/conn_results.go b/server/conn_results.go
index 0e8e74031..eb2d4cad1 100644
--- a/server/conn_results.go
+++ b/server/conn_results.go
@@ -70,8 +70,8 @@ func (c *clientConn) streamRowsToClientExtended(rows RowSet, cmdType string, res
 	}
 
 	if err := rows.Err(); err != nil {
-		if c.isCallerCancellation(err) {
-			c.sendError("ERROR", "57014", "canceling statement due to user request")
+		if c.statementTimedOut(err) || c.isCallerCancellation(err) {
+			c.sendError("ERROR", "57014", c.cancellationMessage(err))
 		} else {
 			c.logger().Error("Row iteration error.", "query", query, "error", err)
 			c.sendError("ERROR", "42000", err.Error())
@@ -145,8 +145,8 @@ func (c *clientConn) streamRowsToClient(rows RowSet, cmdType string, query strin
 	}
 
 	if err := rows.Err(); err != nil {
-		if c.isCallerCancellation(err) {
-			c.sendError("ERROR", "57014", "canceling statement due to user request")
+		if c.statementTimedOut(err) || c.isCallerCancellation(err) {
+			c.sendError("ERROR", "57014", c.cancellationMessage(err))
 		} else {
 			c.logger().Error("Row iteration error.", "query", query, "error", err)
 			c.sendError("ERROR", "42000", err.Error())
diff --git a/server/conn_test.go b/server/conn_test.go
index 2c8b4c5e7..2d061874d 100644
--- a/server/conn_test.go
+++ b/server/conn_test.go
@@ -1470,20 +1470,25 @@ func (e *abortedSelectRecoveryExecutor) PingContext(context.Context) error {
 func (e *abortedSelectRecoveryExecutor) Close() error { return nil }
 
 type abortedExecAlterViewRecoveryExecutor struct {
-	originalQuery string
+	originalAttempts int
+	originalQuery    string
 	noopProfiling
 	rewritten    string
 	execCalls    []string
 	execCtxCalls []string
 }
 
-func (e *abortedExecAlterViewRecoveryExecutor) execResult(query string, callIndex int) (ExecResult, error) {
+func (e *abortedExecAlterViewRecoveryExecutor) execResult(query string, _ int) (ExecResult, error) {
 	trimmed := strings.TrimSpace(query)
 	switch trimmed {
 	case "ROLLBACK":
 		return &fakeExecResult{}, nil
 	case e.originalQuery:
-		if callIndex == 1 {
+		// Count attempts explicitly: statement execution and ROLLBACK now share
+		// ExecContext, so a positional index over the recorded calls is no longer
+		// a reliable way to tell the first attempt from the retry.
+		e.originalAttempts++
+		if e.originalAttempts == 1 {
 			return nil, errors.New("TransactionContext Error: Current transaction is aborted (please ROLLBACK)")
 		}
 		return nil, errors.New("Binder Error: cannot use alter table on a view because this object is not a table; use ALTER VIEW instead")
@@ -1523,6 +1528,7 @@ func (e *abortedExecAlterViewRecoveryExecutor) PingContext(context.Context) erro
 func (e *abortedExecAlterViewRecoveryExecutor) Close() error { return nil }
 
 type abortedAlterViewRecoveryExecutor struct {
+	alterAttempts      int
 	execContextQueries []string
 	noopProfiling
 	execQueries []string
@@ -1538,7 +1544,11 @@ func (e *abortedAlterViewRecoveryExecutor) ExecContext(_ context.Context, query
 	case "ROLLBACK":
 		return &fakeExecResult{}, nil
 	case "ALTER TABLE SOME_VIEW RENAME TO RENAMED_VIEW":
-		if len(e.execContextQueries) == 1 {
+		// Count attempts explicitly: statements and ROLLBACK now share
+		// ExecContext, so a positional index is no longer a reliable
+		// first-attempt signal.
+		e.alterAttempts++
+		if e.alterAttempts == 1 {
 			return nil, errors.New("TransactionContext Error: Current transaction is aborted (please ROLLBACK)")
 		}
 		return nil, errors.New("Binder Error: Cannot use ALTER TABLE statement on object \"some_view\" because it is not a table")
@@ -1899,11 +1909,13 @@ func TestHandleExecuteAbortedRecoveryPreservesAlterViewFallback(t *testing.T) {
 
 	c.handleExecute(body.Bytes())
 
-	expectedExecCalls := []string{originalQuery, originalQuery, rewrittenQuery}
-	if !slices.Equal(executor.execCalls, expectedExecCalls) {
-		t.Fatalf("unexpected Exec calls: got %v want %v", executor.execCalls, expectedExecCalls)
+	// Extended Execute now runs statements through ExecContext so the statement
+	// timeout and CancelRequest reach them; the context-less Exec is no longer
+	// used on this path, and ROLLBACK shares the same method.
+	if len(executor.execCalls) != 0 {
+		t.Fatalf("context-less Exec should no longer be used: got %v", executor.execCalls)
 	}
-	expectedExecContextCalls := []string{"ROLLBACK"}
+	expectedExecContextCalls := []string{originalQuery, "ROLLBACK", originalQuery, rewrittenQuery}
 	if !slices.Equal(executor.execCtxCalls, expectedExecContextCalls) {
 		t.Fatalf("unexpected ExecContext calls: got %v want %v", executor.execCtxCalls, expectedExecContextCalls)
 	}
@@ -1943,15 +1955,19 @@ func TestHandleExecuteRecoversAbortedAutocommitAlterViewFallback(t *testing.T) {
 		"ALTER TABLE some_view RENAME TO renamed_view",
 		"ALTER VIEW some_view RENAME TO renamed_view",
 	}
-	if len(exec.execContextQueries) != 1 || exec.execContextQueries[0] != "ROLLBACK" {
-		t.Fatalf("expected one rollback via ExecContext, got %v", exec.execContextQueries)
+	// Extended Execute now runs statements through ExecContext (so the statement
+	// timeout and CancelRequest reach them), so the recovery sequence and the
+	// ROLLBACK share that method and the context-less Exec goes unused.
+	if len(exec.execQueries) != 0 {
+		t.Fatalf("context-less Exec should no longer be used: got %v", exec.execQueries)
 	}
-	if len(exec.execQueries) != len(wantExec) {
-		t.Fatalf("expected exec sequence %v, got %v", wantExec, exec.execQueries)
+	wantExecContext := []string{wantExec[0], "ROLLBACK", wantExec[1], wantExec[2]}
+	if len(exec.execContextQueries) != len(wantExecContext) {
+		t.Fatalf("expected ExecContext sequence %v, got %v", wantExecContext, exec.execContextQueries)
 	}
-	for i, got := range exec.execQueries {
-		if got != wantExec[i] {
-			t.Fatalf("expected exec query %d to be %q, got %q", i, wantExec[i], got)
+	for i, got := range exec.execContextQueries {
+		if got != wantExecContext[i] {
+			t.Fatalf("expected ExecContext query %d to be %q, got %q", i, wantExecContext[i], got)
 		}
 	}
 }
diff --git a/server/conn_user_secrets.go b/server/conn_user_secrets.go
index 5ff22495c..67015d264 100644
--- a/server/conn_user_secrets.go
+++ b/server/conn_user_secrets.go
@@ -149,7 +149,7 @@ func (c *clientConn) execUserSecretDDL(query string) (handled bool, tag string,
 		}
 		errMsg := execErr.Error()
 		if c.isCallerCancellation(execErr) {
-			errMsg = "canceling statement due to user request"
+			errMsg = c.cancellationMessage(execErr)
 		} else {
 			c.logQueryError(query, execErr)
 		}
diff --git a/server/server.go b/server/server.go
index 55648e9fe..b49a97669 100644
--- a/server/server.go
+++ b/server/server.go
@@ -251,6 +251,30 @@ type Config struct {
 	// Default: 10 seconds.
 	SessionInitTimeout time.Duration
 
+	// StatementTimeout bounds how long a single statement may run before the
+	// server cancels it. 0 (the default) disables it, preserving today's
+	// unbounded behaviour.
+	//
+	// This is blast-radius containment, not a performance knob. Nothing else in
+	// the stack bounds a wedged statement: DuckDB has no statement timeout,
+	// `statement_timeout` from the client is an ignored SET (see
+	// transform.SetShowTransform), the cnpg shards run
+	// idle_in_transaction_session_timeout=0, and pgwire clients wait forever. A
+	// single engine deadlock therefore cost 5h33m of a tenant's hourly refresh
+	// pipeline on 2026-09-16 -- the statement only ended when the CI job hit its
+	// own 6h wall, and every queued refresh behind it was cancelled.
+	//
+	// Size it well above the slowest legitimate statement (that tenant's real
+	// INSERTs peak near 10 minutes) and well below whatever external wall would
+	// otherwise be the only bound.
+	//
+	// NOTE: an engine that ignores cancellation (duckdb/duckdb#24961) will not
+	// stop work when this fires; the client is freed and gets a clear error, but
+	// the worker can stay wedged. Retiring a worker whose statement timed out is
+	// a follow-up -- it has to respect the destroy-before-reuse ordering in
+	// SessionManager.DestroySession.
+	StatementTimeout time.Duration
+
 	// FilePersistence stores DuckDB data in <DataDir>/<username>.duckdb instead of :memory:.
 	// DuckDB memory-maps the file and serves queries from RAM, so performance is similar
 	// to in-memory mode while data persists across connections and restarts.
diff --git a/server/statement_timeout_test.go b/server/statement_timeout_test.go
new file mode 100644
index 000000000..f8c52bcea
--- /dev/null
+++ b/server/statement_timeout_test.go
@@ -0,0 +1,148 @@
+package server
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+)
+
+// deadlineObservingExecutor records whether the context each executor call
+// received actually carried a deadline. That is the property the statement
+// timeout depends on: PinnedExecutor.Query/Exec run on context.Background(), so
+// a path that calls them instead of the *Context variants is silently unbounded.
+type deadlineObservingExecutor struct {
+	selectOneExecutor
+	queryHadDeadline []bool
+	execHadDeadline  []bool
+}
+
+func (e *deadlineObservingExecutor) QueryContext(ctx context.Context, q string, args ...any) (RowSet, error) {
+	_, ok := ctx.Deadline()
+	e.queryHadDeadline = append(e.queryHadDeadline, ok)
+	return e.selectOneExecutor.QueryContext(ctx, q, args...)
+}
+
+func (e *deadlineObservingExecutor) ExecContext(ctx context.Context, q string, args ...any) (ExecResult, error) {
+	_, ok := ctx.Deadline()
+	e.execHadDeadline = append(e.execHadDeadline, ok)
+	return e.selectOneExecutor.ExecContext(ctx, q, args...)
+}
+
+func newTimeoutTestConn(t *testing.T, timeout time.Duration) *clientConn {
+	t.Helper()
+	s := &Server{cfg: Config{StatementTimeout: timeout}}
+	s.activeQueries = make(map[BackendKey]context.CancelFunc)
+	return &clientConn{server: s, ctx: context.Background()}
+}
+
+func TestStatementTimeoutAppliesDeadline(t *testing.T) {
+	c := newTimeoutTestConn(t, 50*time.Millisecond)
+	ctx, cleanup := c.queryContextInner(false)
+	defer cleanup()
+
+	deadline, ok := ctx.Deadline()
+	if !ok {
+		t.Fatal("statement context has no deadline; the timeout was not applied")
+	}
+	if until := time.Until(deadline); until <= 0 || until > time.Second {
+		t.Fatalf("deadline %v out of expected range", until)
+	}
+}
+
+// A zero timeout must preserve today's unbounded behaviour exactly — this knob
+// is opt-in, and a stray deadline would start killing legitimate long queries.
+func TestStatementTimeoutZeroLeavesStatementsUnbounded(t *testing.T) {
+	c := newTimeoutTestConn(t, 0)
+	ctx, cleanup := c.queryContextInner(false)
+	defer cleanup()
+
+	if _, ok := ctx.Deadline(); ok {
+		t.Fatal("statement context has a deadline with StatementTimeout=0")
+	}
+}
+
+// A cursor takes ONE context at DECLARE and holds it across every FETCH, so the
+// timeout bounds the cursor's whole lifetime rather than each FETCH. Stricter
+// than PostgreSQL and deliberate; pin it so it cannot change silently.
+func TestStatementTimeoutBoundsCursorLifetimeNotEachFetch(t *testing.T) {
+	c := newTimeoutTestConn(t, 50*time.Millisecond)
+	ctx, cleanup := c.queryContextForCursor()
+	defer cleanup()
+
+	if _, ok := ctx.Deadline(); !ok {
+		t.Fatal("cursor context has no deadline; the timeout does not reach the cursor path")
+	}
+}
+
+// REGRESSION (review P0): extended-protocol Execute used context-less
+// executor.Query/Exec, so prepared statements — what pgx/psycopg3/JDBC send —
+// were reachable by neither the statement timeout nor a CancelRequest. Drive a
+// real Parse/Bind/Execute and assert the executor saw a deadline.
+func TestStatementTimeoutReachesExtendedProtocolExecute(t *testing.T) {
+	exec := &deadlineObservingExecutor{}
+	c, _ := newBufferedConn(exec)
+	c.server.cfg.StatementTimeout = time.Minute
+	c.stmts = make(map[string]*preparedStmt)
+	c.portals = make(map[string]*portal)
+
+	// Extended-protocol handlers are void; a failure parks on c.fatalErr.
+	c.handleParse(append([]byte("s1\x00SELECT 1\x00"), 0, 0))
+	c.handleBind(append([]byte("p1\x00s1\x00"), 0, 0, 0, 0, 0, 0))
+	c.handleExecute(append([]byte("p1\x00"), 0, 0, 0, 0))
+	if c.fatalErr != nil {
+		t.Fatalf("extended flow failed: %v", c.fatalErr)
+	}
+
+	if len(exec.queryHadDeadline) == 0 {
+		t.Fatal("extended Execute never reached the executor's context-aware path")
+	}
+	for i, had := range exec.queryHadDeadline {
+		if !had {
+			t.Fatalf("extended Execute call %d ran without a deadline: prepared statements are unbounded", i)
+		}
+	}
+}
+
+// The classification is driven by the ERROR and gated on the feature, so it
+// cannot go sticky the way a stored statement context did.
+func TestStatementTimedOutIsErrorDrivenAndFeatureGated(t *testing.T) {
+	deadline := context.DeadlineExceeded
+	wrapped := errors.New("flight execute: context deadline exceeded")
+	other := errors.New("syntax error at or near \"selct\"")
+
+	on := newTimeoutTestConn(t, time.Minute)
+	if !on.statementTimedOut(deadline) || !on.statementTimedOut(wrapped) {
+		t.Fatal("deadline errors not recognised while the timeout is configured")
+	}
+	if on.statementTimedOut(other) || on.statementTimedOut(nil) {
+		t.Fatal("non-deadline error classified as a statement timeout")
+	}
+	if got, want := on.cancellationMessage(deadline), "canceling statement due to statement timeout"; got != want {
+		t.Fatalf("cancellationMessage = %q, want %q", got, want)
+	}
+	if got, want := on.cancellationMessage(context.Canceled), "canceling statement due to user request"; got != want {
+		t.Fatalf("user-cancel wording = %q, want %q", got, want)
+	}
+
+	// Feature off: internal deadlines (attach, exec, worker gRPC) must NOT start
+	// surfacing as 57014 just because this code exists.
+	off := newTimeoutTestConn(t, 0)
+	if off.statementTimedOut(deadline) || off.statementTimedOut(wrapped) {
+		t.Fatal("deadline classified as a statement timeout with the feature disabled")
+	}
+	if got, want := off.cancellationMessage(deadline), "canceling statement due to user request"; got != want {
+		t.Fatalf("disabled-path wording = %q, want %q", got, want)
+	}
+}
+
+func TestIsCallerCancellationCoversStatementTimeout(t *testing.T) {
+	c := newTimeoutTestConn(t, time.Minute)
+	if !c.isCallerCancellation(context.DeadlineExceeded) {
+		t.Fatal("statement timeout not treated as caller cancellation (would log as an infra failure)")
+	}
+	off := newTimeoutTestConn(t, 0)
+	if off.isCallerCancellation(context.DeadlineExceeded) {
+		t.Fatal("deadline treated as caller cancellation with the feature disabled")
+	}
+}
diff --git a/tests/mw-dev/README.md b/tests/mw-dev/README.md
index da41f3f14..970ef02ea 100644
--- a/tests/mw-dev/README.md
+++ b/tests/mw-dev/README.md
@@ -541,6 +541,24 @@ normal `go test ./...` lane.
   and the event stream back, which the Job cannot do for the reason above; the
   key resolution is covered by `TestAnalyticsAPIKeyPrefersDedicatedKey` in
   `internal/cliboot/analytics_test.go`.
+- **`--statement-timeout` (`DUCKGRES_STATEMENT_TIMEOUT`) is unit-only, deliberately.**
+  The knob is server-global and defaults to `0` (unbounded), so asserting it
+  in-Job needs one of two bad options: set a short global timeout on the e2e
+  control plane, which would start killing the harness's own legitimately slow
+  assertions (reshard copies, DuckLake round-trips, concurrent-writer checks);
+  or leave it generous and burn that many minutes on a deliberately slow query
+  just to watch it expire. Neither buys more confidence than the unit tests,
+  which cover the behaviour that can actually regress: the deadline is applied
+  to the STATEMENT context and not the connection context, `0` leaves
+  statements unbounded, an expired deadline classifies as `57014` with
+  PostgreSQL's exact `canceling statement due to statement timeout` wording
+  (drivers string-match it), a user cancel keeps the `due to user request`
+  wording, and `isQueryCancelled` matches `context deadline exceeded` as well
+  as `context canceled`. See `server/statement_timeout_test.go`. If the knob
+  ever becomes per-connection (a client-honored `statement_timeout` GUC is the
+  named follow-up), it becomes cheaply assertable in-Job and should get a
+  harness assertion then.
+
 - **PostHog Logs (OTLP)** — `assert_worker_pod` checks plumbing only: the
   worker must not carry a plaintext `POSTHOG_API_KEY` `value:`, and when the
   CP container's *named* env has a `secretKeyRef` the worker must copy the