(MOT-4216) fix(database): resolve omitted db to the sole/primary pool

database/src/error.rs

@@ -29,6 +29,10 @@ pub enum DbError {
     #[error("unknown db {db}; available: [{}]", available.join(", "))]
     UnknownDb { db: String, available: Vec<String> },
 
+    #[serde(rename = "MISSING_DB")]
+    #[error("no `db` specified and none of the configured databases is an unambiguous default; available: [{}]", available.join(", "))]
+    MissingDb { available: Vec<String> },
+
     #[serde(rename = "INVALID_PARAM")]
     #[error("invalid parameter at index {index}: {reason}")]
     InvalidParam { index: usize, reason: String },
@@ -105,6 +109,20 @@ mod tests {
         );
     }
 
+    #[test]
+    fn missing_db_serializes_with_stable_code_and_available_handles() {
+        let e = DbError::MissingDb {
+            available: vec!["analytics".into(), "main".into()],
+        };
+        let v: serde_json::Value = serde_json::to_value(&e).unwrap();
+        assert_eq!(v["code"], "MISSING_DB");
+        assert_eq!(v["available"], serde_json::json!(["analytics", "main"]));
+        assert_eq!(
+            e.to_string(),
+            "no `db` specified and none of the configured databases is an unambiguous default; available: [analytics, main]"
+        );
+    }
+
     #[test]
     fn transaction_not_found_serializes_with_stable_code() {
         let e = DbError::TransactionNotFound {

database/src/handlers/begin_transaction.rs

@@ -19,7 +19,10 @@ use std::time::Duration;
 
 #[derive(Deserialize, JsonSchema)]
 pub struct BeginTxReq {
-    pub db: String,
+    /// Logical database name. Optional — omitting it targets the sole
+    /// configured database, or `primary` when several are configured.
+    #[serde(default)]
+    pub db: Option<String>,
     #[serde(default)]
     pub isolation: Option<String>,
     #[serde(default)]
@@ -55,7 +58,8 @@ fn parse_isolation(s: Option<&str>) -> Result<Option<Isolation>, DbError> {
 }
 
 pub async fn handle(state: &AppState, req: BeginTxReq) -> Result<BeginTxResp, String> {
-    let pool = state.pool(&req.db).await.map_err(err_to_str)?;
+    let db = state.resolve_db(req.db).await.map_err(err_to_str)?;
+    let pool = state.pool(&db).await.map_err(err_to_str)?;
     let driver = pool.driver();
     let isolation = parse_isolation(req.isolation.as_deref()).map_err(err_to_str)?;
     let timeout_ms = req
@@ -84,7 +88,7 @@ pub async fn handle(state: &AppState, req: BeginTxReq) -> Result<BeginTxResp, St
             "db_tx_begin_failed",
             Some(json!({
                 "db.system": driver_system(driver),
-                "db.name": req.db,
+                "db.name": db,
                 "isolation": req.isolation,
                 "error": serde_json::to_value(&e).ok(),
             })),
@@ -94,14 +98,14 @@ pub async fn handle(state: &AppState, req: BeginTxReq) -> Result<BeginTxResp, St
 
     let handle = state
         .transactions
-        .insert(req.db.clone(), driver, pinned, timeout)
+        .insert(db.clone(), driver, pinned, timeout)
         .await;
 
     state.log.info(
         "db_tx_started",
         Some(json!({
             "db.system": driver_system(driver),
-            "db.name": req.db,
+            "db.name": db,
             "db.operation": "BEGIN",
             "db.transaction.id": handle.id,
             "isolation": req.isolation,

database/src/handlers/execute.rs

@@ -11,7 +11,10 @@ use serde_json::Value;
 
 #[derive(Deserialize, JsonSchema)]
 pub struct ExecuteReq {
-    pub db: String,
+    /// Logical database name. Optional — omitting it targets the sole
+    /// configured database, or `primary` when several are configured.
+    #[serde(default)]
+    pub db: Option<String>,
     #[serde(alias = "query")]
     pub sql: String,
     #[serde(default, deserialize_with = "crate::handlers::lenient_params")]
@@ -20,15 +23,16 @@ pub struct ExecuteReq {
     pub returning: Vec<String>,
 }
 
-#[derive(Serialize, JsonSchema)]
+#[derive(Debug, Serialize, JsonSchema)]
 pub struct ExecuteResp {
     pub affected_rows: u64,
     pub last_insert_id: Option<String>,
     pub returned_rows: Vec<serde_json::Map<String, Value>>,
 }
 
 pub async fn handle(state: &AppState, req: ExecuteReq) -> Result<ExecuteResp, String> {
-    let pool = state.pool(&req.db).await.map_err(err_to_str)?;
+    let db = state.resolve_db(req.db).await.map_err(err_to_str)?;
+    let pool = state.pool(&db).await.map_err(err_to_str)?;
     // Reject empty SQL uniformly. See the matching guard in query.rs for why
     // this is at the handler boundary rather than per-driver: postgres' driver
     // accepts empty SQL as a no-op success, sqlite/mysql reject — guarding
@@ -41,6 +45,7 @@ pub async fn handle(state: &AppState, req: ExecuteReq) -> Result<ExecuteResp, St
             failed_index: None,
         }));
     }
+    crate::handlers::reject_tx_control_sql(&req.sql).map_err(err_to_str)?;
     let params = JsonParam::from_json_slice(&req.params).map_err(err_to_str)?;
 
     let result = match &pool {
@@ -162,6 +167,23 @@ mod tests {
         assert_eq!(properties["params"]["type"], "array");
     }
 
+    #[tokio::test(flavor = "multi_thread")]
+    async fn execute_rejects_transaction_control_sql_with_pointer() {
+        // rctest5 postmortem: agents run execute("BEGIN") + execute("COMMIT")
+        // expecting a session — but each call draws a fresh pooled
+        // connection, and the leaked BEGIN poisoned the pool for every later
+        // caller. The guard must name the real transactional surfaces.
+        let st = state();
+        for sql in ["BEGIN", "commit;", "  ROLLBACK", "/* tx */ BEGIN IMMEDIATE"] {
+            let err = handle(&st, req(json!({"db":"primary","sql": sql})))
+                .await
+                .unwrap_err();
+            assert!(err.contains("INVALID_PARAM"), "{sql}: {err}");
+            assert!(err.contains("beginTransaction"), "{sql}: {err}");
+            assert!(err.contains("executeBatch"), "{sql}: {err}");
+        }
+    }
+
     #[tokio::test(flavor = "multi_thread")]
     async fn execute_insert_returns_envelope() {
         let st = state();

database/src/handlers/execute_batch.rs

@@ -13,7 +13,10 @@ use serde_json::Value;
 
 #[derive(Deserialize, JsonSchema)]
 pub struct ExecuteBatchReq {
-    pub db: String,
+    /// Logical database name. Optional — omitting it targets the sole
+    /// configured database, or `primary` when several are configured.
+    #[serde(default)]
+    pub db: Option<String>,
     /// Statements to run in order inside one transaction. Each entry is
     /// either a bare SQL string or `{ "sql": "...", "params": [...] }` —
     /// use `params` for dynamic values instead of inlining them into the SQL.

database/src/handlers/mod.rs

@@ -45,6 +45,29 @@ mod tx_sql_guard;
 
 pub(crate) use query::rows_to_objects as query_rows_to_objects;
 
+/// Reject transaction-control SQL on the pooled per-call surfaces. Each call
+/// runs on whichever connection the pool hands out, so BEGIN/COMMIT can never
+/// pair up — worse, a BEGIN that "succeeds" returns its connection to the
+/// pool still inside an open transaction, and every later caller unlucky
+/// enough to draw it fails with `cannot start a transaction within a
+/// transaction` (rctest5 postmortem: three writer agents starved on one
+/// poisoned connection). Point the caller at the real transactional surfaces.
+pub(crate) fn reject_tx_control_sql(sql: &str) -> Result<(), DbError> {
+    if tx_sql_guard::is_transaction_control_sql(sql) {
+        return Err(DbError::InvalidParam {
+            index: 0,
+            reason: "transaction-control SQL (BEGIN/COMMIT/ROLLBACK/SAVEPOINT) is not valid \
+                     here: each call runs on its own pooled connection, so the statements \
+                     cannot pair up and a leaked BEGIN poisons the pool for every later \
+                     caller. Use database::beginTransaction + database::transactionExecute + \
+                     database::commitTransaction, or database::executeBatch for an atomic \
+                     batch."
+                .into(),
+        });
+    }
+    Ok(())
+}
+
 #[derive(Clone)]
 pub struct AppState {
     pub pools: Arc<RwLock<HashMap<String, Pool>>>,
@@ -58,6 +81,30 @@ pub struct AppState {
 }
 
 impl AppState {
+    /// Resolve an optional logical db name to a concrete pool key.
+    ///
+    /// `Some(name)` passes through untouched (unknown names still surface
+    /// `UNKNOWN_DB` from [`AppState::pool`]). `None` falls back to the sole
+    /// configured pool, then to `primary` when several exist; the omission is
+    /// only an error (`MISSING_DB`) when neither rule disambiguates. This
+    /// exists because LLM callers routinely omit `db` on their first call —
+    /// a hard serde failure there is a wasted round-trip on every session.
+    pub async fn resolve_db(&self, db: Option<String>) -> Result<String, DbError> {
+        if let Some(name) = db {
+            return Ok(name);
+        }
+        let pools = self.pools.read().await;
+        if pools.len() == 1 {
+            return Ok(pools.keys().next().expect("len checked above").clone());
+        }
+        if pools.contains_key(crate::config::DEFAULT_DB_NAME) {
+            return Ok(crate::config::DEFAULT_DB_NAME.to_string());
+        }
+        let mut available: Vec<String> = pools.keys().cloned().collect();
+        available.sort();
+        Err(DbError::MissingDb { available })
+    }
+
     pub async fn pool(&self, db: &str) -> Result<Pool, DbError> {
         let pools = self.pools.read().await;
         pools.get(db).cloned().ok_or_else(|| {

database/src/handlers/prepare.rs

@@ -10,7 +10,10 @@ use std::time::Duration;
 
 #[derive(Deserialize, JsonSchema)]
 pub struct PrepareReq {
-    pub db: String,
+    /// Logical database name. Optional — omitting it targets the sole
+    /// configured database, or `primary` when several are configured.
+    #[serde(default)]
+    pub db: Option<String>,
     #[serde(alias = "query")]
     pub sql: String,
     #[serde(default = "default_ttl")]
@@ -30,7 +33,8 @@ const MAX_TTL_SECONDS: u64 = 86_400;
 
 pub async fn handle(state: &AppState, req: PrepareReq) -> Result<PrepareResp, String> {
     let ttl = Duration::from_secs(req.ttl_seconds.min(MAX_TTL_SECONDS));
-    let pool = state.pool(&req.db).await.map_err(err_to_str)?;
+    let db = state.resolve_db(req.db).await.map_err(err_to_str)?;
+    let pool = state.pool(&db).await.map_err(err_to_str)?;
     // Reject empty SQL at the handler boundary, mirroring query.rs / execute.rs.
     // Without this, prepareStatement happily acquires a pool connection and
     // pins it under a UUID handle that can never run successfully — the
@@ -43,6 +47,9 @@ pub async fn handle(state: &AppState, req: PrepareReq) -> Result<PrepareResp, St
             failed_index: None,
         }));
     }
+    // A prepared BEGIN executed via runStatement opens a transaction on the
+    // pinned connection that outlives the handle — same pool poison.
+    crate::handlers::reject_tx_control_sql(&req.sql).map_err(err_to_str)?;
 
     let h = match pool {
         Pool::Sqlite(p) => {

database/src/handlers/query.rs

@@ -11,7 +11,10 @@ use serde_json::Value;
 
 #[derive(Debug, Deserialize, JsonSchema)]
 pub struct QueryReq {
-    pub db: String,
+    /// Logical database name. Optional — omitting it targets the sole
+    /// configured database, or `primary` when several are configured.
+    #[serde(default)]
+    pub db: Option<String>,
     #[serde(alias = "query")]
     pub sql: String,
     #[serde(default, deserialize_with = "crate::handlers::lenient_params")]
@@ -33,7 +36,8 @@ fn default_timeout() -> u64 {
 
 /// Returns a JSON string body suitable to wrap in IIIError on failure.
 pub async fn handle(state: &AppState, req: QueryReq) -> Result<QueryResp, String> {
-    let pool = state.pool(&req.db).await.map_err(err_to_str)?;
+    let db = state.resolve_db(req.db).await.map_err(err_to_str)?;
+    let pool = state.pool(&db).await.map_err(err_to_str)?;
     // Reject empty SQL uniformly. Postgres' tokio-postgres treats `client.query("")`
     // as a valid no-op and returns Ok([]), but sqlite (rusqlite) and mysql
     // (mysql_async) reject it at parse time — without this guard the worker's
@@ -46,6 +50,9 @@ pub async fn handle(state: &AppState, req: QueryReq) -> Result<QueryResp, String
             failed_index: None,
         }));
     }
+    // A `query("BEGIN")` is just as pool-poisoning as an execute — sqlite
+    // happily starts the transaction and returns zero rows.
+    crate::handlers::reject_tx_control_sql(&req.sql).map_err(err_to_str)?;
     let params = JsonParam::from_json_slice(&req.params).map_err(err_to_str)?;
 
     let result = match &pool {
@@ -160,14 +167,69 @@ mod tests {
     }
 
     #[tokio::test(flavor = "multi_thread")]
-    async fn query_missing_db_field_errors() {
-        // Deserialization moved into the SDK once we switched to typed
-        // RegisterFunction::new_async — at the handler boundary the input is
-        // already a QueryReq. This test keeps the missing-field contract by
-        // exercising the deserialization step explicitly.
-        let err = serde_json::from_value::<QueryReq>(json!({"sql":"SELECT 1"}))
-            .unwrap_err()
-            .to_string();
-        assert!(err.contains("missing field"), "got: {err}");
+    async fn query_rejects_transaction_control_sql() {
+        // sqlite happily starts a transaction from a query("BEGIN") and
+        // returns zero rows — the same pool poison as the execute path.
+        let st = state();
+        let err = handle(&st, req(json!({"db":"primary","sql":"BEGIN"})))
+            .await
+            .unwrap_err();
+        assert!(err.contains("INVALID_PARAM"), "got: {err}");
+        assert!(err.contains("beginTransaction"), "got: {err}");
+    }
+
+    #[tokio::test(flavor = "multi_thread")]
+    async fn query_omitted_db_defaults_to_sole_pool() {
+        // LLM callers routinely omit `db` on their first call. With exactly
+        // one configured pool the omission must resolve to it instead of
+        // failing serde — the old `missing field db` error cost a wasted
+        // round-trip in every live session.
+        let st = state();
+        let resp = handle(&st, req(json!({"sql":"SELECT 1 AS one"})))
+            .await
+            .unwrap();
+        assert_eq!(resp.rows[0]["one"], 1);
+    }
+
+    #[tokio::test(flavor = "multi_thread")]
+    async fn omitted_db_prefers_primary_among_many() {
+        let st = state();
+        {
+            let extra = SqlitePool::new("sqlite::memory:", &PoolConfig::default()).unwrap();
+            st.pools
+                .write()
+                .await
+                .insert("analytics".to_string(), Pool::Sqlite(extra));
+        }
+        assert_eq!(st.resolve_db(None).await.unwrap(), "primary");
+    }
+
+    #[tokio::test(flavor = "multi_thread")]
+    async fn omitted_db_without_default_errors_with_available_list() {
+        let st = state();
+        {
+            let mut pools = st.pools.write().await;
+            let p = pools.remove("primary").unwrap();
+            pools.insert("main".to_string(), p);
+            let extra = SqlitePool::new("sqlite::memory:", &PoolConfig::default()).unwrap();
+            pools.insert("analytics".to_string(), Pool::Sqlite(extra));
+        }
+        let err = handle(&st, req(json!({"sql":"SELECT 1"}))).await.unwrap_err();
+        assert!(err.contains("MISSING_DB"), "got: {err}");
+        assert!(
+            err.contains("\"available\":[\"analytics\",\"main\"]"),
+            "got: {err}"
+        );
+    }
+
+    #[test]
+    fn request_schema_marks_db_optional() {
+        let schema = serde_json::to_value(schemars::schema_for!(QueryReq).schema).unwrap();
+        let required = schema["required"].as_array().cloned().unwrap_or_default();
+        assert!(
+            !required.iter().any(|f| f == "db"),
+            "db must not be required in the function schema: {required:?}"
+        );
+        assert!(required.iter().any(|f| f == "sql"));
     }
 }

database/src/handlers/transaction.rs

@@ -12,7 +12,10 @@ use serde_json::{json, Value};
 
 #[derive(Deserialize, JsonSchema)]
 pub struct TxReq {
-    pub db: String,
+    /// Logical database name. Optional — omitting it targets the sole
+    /// configured database, or `primary` when several are configured.
+    #[serde(default)]
+    pub db: Option<String>,
     pub statements: Vec<TxStmtReq>,
     #[serde(default)]
     pub isolation: Option<String>,
@@ -56,7 +59,8 @@ fn failed_index_of(e: &crate::error::DbError) -> Option<usize> {
 }
 
 pub async fn handle(state: &AppState, req: TxReq) -> Result<TxResp, String> {
-    let pool = state.pool(&req.db).await.map_err(err_to_str)?;
+    let db = state.resolve_db(req.db).await.map_err(err_to_str)?;
+    let pool = state.pool(&db).await.map_err(err_to_str)?;
 
     let isolation = match req.isolation.as_deref() {
         Some("read_committed") => Some(Isolation::ReadCommitted),