diff --git a/nodedb-types/src/error/sqlstate.rs b/nodedb-types/src/error/sqlstate.rs
index af330a98a..8f03dcace 100644
--- a/nodedb-types/src/error/sqlstate.rs
+++ b/nodedb-types/src/error/sqlstate.rs
@@ -90,6 +90,11 @@ pub const TYPE_GUARD_VIOLATION: &str = "23608";
 /// `28000` — `invalid_authorization_specification` (no valid credentials)
 pub const INVALID_AUTHORIZATION: &str = "28000";
 
+// ── Class 3D — Invalid Catalog Name ──────────────────────────────────────────
+
+/// `3D000` — `invalid_catalog_name` (the selected database does not exist)
+pub const INVALID_CATALOG_NAME: &str = "3D000";
+
 // ── Class 40 — Transaction Rollback ──────────────────────────────────────────
 
 /// `40001` — `serialization_failure` (write conflict; client should retry)
diff --git a/nodedb/src/control/server/native/dispatch/conversion.rs b/nodedb/src/control/server/native/dispatch/conversion.rs
index 4bf8eeacb..895bedd87 100644
--- a/nodedb/src/control/server/native/dispatch/conversion.rs
+++ b/nodedb/src/control/server/native/dispatch/conversion.rs
@@ -7,6 +7,7 @@ use nodedb_types::conversion::json_to_value_ref;
 use nodedb_types::protocol::NativeResponse;
 
 use crate::bridge::envelope::Response;
+use crate::control::server::native::sqlstate_code::sqlstate_error;
 use crate::control::server::response_shape::types::ShapedRows;
 use crate::control::server::shared::ddl::sqlstate::error_code_to_sqlstate;
 use crate::control::server::shared::ddl::{DdlError, DdlResult};
@@ -52,6 +53,29 @@ pub(crate) fn error_to_native(seq: u64, e: &crate::Error) -> NativeResponse {
     NativeResponse::error_with_code(seq, code, message, ndb_code)
 }
 
+/// Convert a Control-Plane error into a native error frame under a SQLSTATE
+/// the call site chooses.
+///
+/// Same classification as [`error_to_native`] — the numeric code comes from
+/// the one `Error` mapping table — but for the guards that render a more
+/// specific SQLSTATE than the error's own variant implies: a plan that cannot
+/// be built is `42601` to a SQL client whatever its internal cause, and an
+/// RLS injection failure is `42501`. Those sites still hold the classified
+/// error, so the code must come from it rather than be inferred back out of
+/// the SQLSTATE they just chose.
+pub(crate) fn error_to_native_with_sqlstate(
+    seq: u64,
+    sqlstate: impl Into<String>,
+    e: &crate::Error,
+) -> NativeResponse {
+    NativeResponse::error_with_code(
+        seq,
+        sqlstate,
+        e.to_string(),
+        crate::error_classify::classify(e).code().0,
+    )
+}
+
 /// Convert a `NodeDbError` produced while shaping a response into a
 /// NativeResponse error frame.
 ///
@@ -93,7 +117,7 @@ pub(crate) fn error_code_to_native(
     code: Option<&crate::bridge::envelope::ErrorCode>,
 ) -> NativeResponse {
     let Some(code) = code else {
-        return NativeResponse::error(seq, "XX000", "unknown data plane error");
+        return sqlstate_error(seq, "XX000", "unknown data plane error");
     };
     let (_, sqlstate, message) = error_code_to_sqlstate(code);
     let public = nodedb_types::NodeDbError::from(crate::Error::DataPlane(code.clone()));
@@ -108,12 +132,18 @@ pub(crate) fn error_code_to_native(
 /// row-returning / status / empty result determines the response (a status tag
 /// becomes a single-column status row, a row result becomes a columns+rows
 /// frame, an empty result or an empty vec becomes a bare OK).
+///
+/// `DdlError` is authored as a SQLSTATE and a message and never holds a
+/// classified `Error`, so the numeric code the client rebuilds its typed error
+/// from comes from `sqlstate_code`. Without it every DDL refusal — a `DROP
+/// TABLE` naming a collection that does not exist, a denied `GRANT` — reaches
+/// the client as a generic internal failure.
 pub(crate) fn ddl_result_to_native(
     seq: u64,
     result: Result<Vec<DdlResult>, DdlError>,
 ) -> NativeResponse {
     match result {
-        Err(DdlError { sqlstate, message }) => NativeResponse::error(seq, sqlstate, message),
+        Err(DdlError { sqlstate, message }) => sqlstate_error(seq, sqlstate, message),
         // Unknown pgwire response variants are dropped during translation, so
         // the first element is the first meaningful result — mirroring the
         // previous bridge, which returned on the first known variant.
@@ -313,6 +343,57 @@ mod tests {
         );
     }
 
+    /// A DDL refusal is authored as a SQLSTATE with no `Error` behind it, so
+    /// its numeric code comes from the SQLSTATE table. Without it the frame
+    /// ships `ndb_code == 0` and a `DROP TABLE` naming an absent collection
+    /// arrives as a generic internal failure while the identical `SELECT`
+    /// arrives typed.
+    #[test]
+    fn ddl_refusals_carry_their_numeric_code() {
+        let response = ddl_result_to_native(
+            1,
+            Err(DdlError {
+                sqlstate: "42P01".to_owned(),
+                message: "collection 'missing' does not exist".to_owned(),
+            }),
+        );
+
+        let error = response
+            .error
+            .expect("error responses must carry a payload");
+        assert_eq!(error.code, "42P01");
+        assert_eq!(
+            error.ndb_code,
+            nodedb_types::error::ErrorCode::COLLECTION_NOT_FOUND.0
+        );
+        assert_eq!(error.message, "collection 'missing' does not exist");
+    }
+
+    /// A site that renders a more specific SQLSTATE than the error implies
+    /// must still take the classification from the error rather than from the
+    /// SQLSTATE it just chose: `42601` is shared by several conditions, while
+    /// the error in hand names exactly one.
+    #[test]
+    fn a_site_chosen_sqlstate_keeps_the_errors_classification() {
+        let response = error_to_native_with_sqlstate(
+            1,
+            "42601",
+            &crate::Error::PlanError {
+                detail: "no such column".to_owned(),
+            },
+        );
+
+        let error = response
+            .error
+            .expect("error responses must carry a payload");
+        assert_eq!(error.code, "42601");
+        assert_eq!(
+            error.ndb_code,
+            nodedb_types::error::ErrorCode::PLAN_ERROR.0,
+            "the classification must come from the error, not from the SQLSTATE"
+        );
+    }
+
     /// The numeric code is populated for every variant, including the ones
     /// whose SQLSTATE falls through to `XX000` — otherwise the fix would be a
     /// per-variant special case rather than one classification.
diff --git a/nodedb/src/control/server/native/dispatch/direct_ops.rs b/nodedb/src/control/server/native/dispatch/direct_ops.rs
index a3a9588d4..906c9c81f 100644
--- a/nodedb/src/control/server/native/dispatch/direct_ops.rs
+++ b/nodedb/src/control/server/native/dispatch/direct_ops.rs
@@ -16,7 +16,8 @@ use nodedb_physical::physical_task::{PhysicalTask, PostSetOp};
 
 use super::response::data_plane_response_to_native;
 use super::single_task::dispatch_single_task;
-use super::{DispatchCtx, error_to_native};
+use super::{DispatchCtx, error_to_native, error_to_native_with_sqlstate};
+use crate::control::server::native::sqlstate_code::sqlstate_error;
 
 /// Dispatch a direct Data Plane operation by opcode.
 pub(crate) async fn handle_direct_op(
@@ -55,7 +56,7 @@ pub(crate) async fn handle_direct_op(
 
     // Per-operation cap enforcement (vector dim, top_k, batch size, etc.).
     if let Err(e) = super::limits::check_op_limits(ctx.state, fields) {
-        return NativeResponse::error(seq, "0A000", e.to_string());
+        return error_to_native_with_sqlstate(seq, "0A000", &e);
     }
 
     // Quota enforcement — reject before planning or dispatch.
@@ -65,7 +66,7 @@ pub(crate) async fn handle_direct_op(
 
     let mut plan = match super::plan_builder::build_plan(ctx, op, fields, &collection) {
         Ok(p) => p,
-        Err(e) => return NativeResponse::error(seq, "42601", e.to_string()),
+        Err(e) => return error_to_native_with_sqlstate(seq, "42601", &e),
     };
 
     // Apply RLS before any special Control-Plane orchestration can observe the plan.
@@ -75,7 +76,7 @@ pub(crate) async fn handle_direct_op(
         &ctx.state.rls,
         ctx.auth_context(),
     ) {
-        return NativeResponse::error(seq, "42501", e.to_string());
+        return error_to_native_with_sqlstate(seq, "42501", &e);
     }
 
     // Refuse what column redaction cannot cover (an aggregate over a redacted
@@ -86,7 +87,7 @@ pub(crate) async fn handle_direct_op(
         ctx.auth_context(),
         &ctx.state.redaction,
     ) {
-        return NativeResponse::error(seq, "0A000", e.to_string());
+        return error_to_native_with_sqlstate(seq, "0A000", &e);
     }
 
     // Extracted before `plan` is moved/cloned into any of the branches below
@@ -107,7 +108,7 @@ pub(crate) async fn handle_direct_op(
     if let Some(info) = &plan_metering_info
         && let Err(e) = admit_quota_for_dispatch(ctx.state, &ctx.scope, info)
     {
-        return NativeResponse::error(seq, "53400", e.to_string());
+        return error_to_native_with_sqlstate(seq, "53400", &e);
     }
 
     // Whether the blanket metering call below (after the block) still needs
@@ -328,7 +329,7 @@ pub(crate) async fn handle_direct_op(
             let task = match authorized_tasks.into_tasks().into_iter().next() {
                 Some(task) => task,
                 None => {
-                    return NativeResponse::error(
+                    return sqlstate_error(
                         seq,
                         "XX000",
                         "authorization returned no task capability",
diff --git a/nodedb/src/control/server/native/dispatch/graph_match.rs b/nodedb/src/control/server/native/dispatch/graph_match.rs
index 35ec0de31..49f9d92ab 100644
--- a/nodedb/src/control/server/native/dispatch/graph_match.rs
+++ b/nodedb/src/control/server/native/dispatch/graph_match.rs
@@ -10,7 +10,7 @@ use crate::control::server::shared::quota_admission::admit_quota_for_dispatch;
 
 use super::raw_dispatch::dispatch_authorized_single_task;
 use super::response::data_plane_response_to_native;
-use super::{DispatchCtx, error_to_native};
+use super::{DispatchCtx, error_to_native, error_to_native_with_sqlstate};
 
 /// Dispatch a native `GraphMatch` op, unwrapping the DP `{rows, frontier}`
 /// envelope into a bare rows array before native conversion.
@@ -36,7 +36,7 @@ pub(crate) async fn handle_graph_match(
     let tenant_id = ctx.tenant_id();
 
     if let Err(error) = super::limits::check_op_limits(ctx.state, fields) {
-        return NativeResponse::error(seq, "0A000", error.to_string());
+        return error_to_native_with_sqlstate(seq, "0A000", &error);
     }
     if let Err(error) = ctx.state.check_tenant_quota(tenant_id) {
         return error_to_native(seq, &error);
@@ -45,7 +45,7 @@ pub(crate) async fn handle_graph_match(
     let mut plan =
         match super::plan_builder::build_plan(ctx, OpCode::GraphMatch, fields, &collection) {
             Ok(plan) => plan,
-            Err(error) => return NativeResponse::error(seq, "42601", error.to_string()),
+            Err(error) => return error_to_native_with_sqlstate(seq, "42601", &error),
         };
     if let Err(error) = crate::control::planner::rls_injection::inject_rls_for_single_plan(
         tenant_id.as_u64(),
@@ -53,7 +53,7 @@ pub(crate) async fn handle_graph_match(
         &ctx.state.rls,
         ctx.auth_context(),
     ) {
-        return NativeResponse::error(seq, "42501", error.to_string());
+        return error_to_native_with_sqlstate(seq, "42501", &error);
     }
     // Refuse what column redaction cannot cover: a MATCH returns graph
     // topology, which the result-path masking hook has no columns to rewrite.
@@ -63,7 +63,7 @@ pub(crate) async fn handle_graph_match(
         ctx.auth_context(),
         &ctx.state.redaction,
     ) {
-        return NativeResponse::error(seq, "0A000", error.to_string());
+        return error_to_native_with_sqlstate(seq, "0A000", &error);
     }
 
     // Stamp the active transaction id so MATCH reads resolve this connection's
@@ -84,7 +84,7 @@ pub(crate) async fn handle_graph_match(
     if let Some(info) = &plan_metering_info
         && let Err(e) = admit_quota_for_dispatch(ctx.state, &ctx.scope, info)
     {
-        return NativeResponse::error(seq, "53400", e.to_string());
+        return error_to_native_with_sqlstate(seq, "53400", &e);
     }
     let _request = ctx.state.tenant_request_guard(tenant_id);
     let raw = dispatch_authorized_single_task(ctx, tenant_id, vshard_id, plan, txn_id).await;
diff --git a/nodedb/src/control/server/native/dispatch/mod.rs b/nodedb/src/control/server/native/dispatch/mod.rs
index 2bb6f10ac..279bc151a 100644
--- a/nodedb/src/control/server/native/dispatch/mod.rs
+++ b/nodedb/src/control/server/native/dispatch/mod.rs
@@ -28,7 +28,7 @@ pub(crate) use admission_op::admission_operation;
 pub(crate) use auth::{NativeAuthOutcome, handle_auth, handle_ping};
 pub(crate) use conversion::{
     ddl_result_to_native, error_code_to_native, error_response_to_native, error_to_native,
-    shape_error_to_native, to_native_columns_rows,
+    error_to_native_with_sqlstate, shape_error_to_native, to_native_columns_rows,
 };
 pub(crate) use ctx::DispatchCtx;
 pub(crate) use direct_ops::handle_direct_op;
diff --git a/nodedb/src/control/server/native/dispatch/single_task.rs b/nodedb/src/control/server/native/dispatch/single_task.rs
index 846f8afb7..5f592a108 100644
--- a/nodedb/src/control/server/native/dispatch/single_task.rs
+++ b/nodedb/src/control/server/native/dispatch/single_task.rs
@@ -16,7 +16,7 @@ use crate::types::{Lsn, RequestId};
 
 use super::raw_dispatch::dispatch_authorized_single_task;
 use super::response::data_plane_response_to_native;
-use super::{DispatchCtx, error_code_to_native, error_to_native};
+use super::{DispatchCtx, error_code_to_native, error_to_native, error_to_native_with_sqlstate};
 
 /// Dispatch one plan via the gateway (when wired) or the local SPSC path,
 /// converting the Data-Plane response into a `NativeResponse`.
@@ -76,7 +76,7 @@ pub(super) async fn dispatch_single_task(
     if let Some(info) = &plan_metering_info
         && let Err(e) = admit_quota_for_dispatch(ctx.state, &ctx.scope, info)
     {
-        return NativeResponse::error(seq, "53400", e.to_string());
+        return error_to_native_with_sqlstate(seq, "53400", &e);
     }
 
     let task = match route_in_tx_write(
diff --git a/nodedb/src/control/server/native/dispatch/sql.rs b/nodedb/src/control/server/native/dispatch/sql.rs
index 6871df24c..8aead8a32 100644
--- a/nodedb/src/control/server/native/dispatch/sql.rs
+++ b/nodedb/src/control/server/native/dispatch/sql.rs
@@ -27,6 +27,7 @@ use super::transaction_savepoint::{
     handle_release_savepoint, handle_rollback_to_savepoint, handle_savepoint,
 };
 use super::{DispatchCtx, error_to_native};
+use crate::control::server::native::sqlstate_code::sqlstate_error;
 
 /// Handle a SQL statement: transaction control, SET/SHOW, DDL, or DataFusion.
 ///
@@ -110,7 +111,7 @@ async fn handle_sql_inner(
     }
 
     if ctx.sessions.transaction_state(ctx.peer_addr) == TransactionState::Failed {
-        return resp(NativeResponse::error(
+        return resp(sqlstate_error(
             seq,
             "25P02",
             "current transaction is aborted, commands ignored until end of transaction block",
@@ -361,7 +362,7 @@ async fn execute_planned(
         match try_open_sql_stream(ctx, seq, &tasks, database_id, Some(&output_schema)).await {
             Ok(Some(mut stream)) => {
                 let Some(scope) = lease_scope.take() else {
-                    return resp(NativeResponse::error(
+                    return resp(sqlstate_error(
                         seq,
                         "XX000",
                         "internal error: query lease scope missing before SQL stream dispatch",
@@ -382,7 +383,7 @@ async fn execute_planned(
     // after this local owner is dropped. Lazy streams above retain the raw
     // scope directly in their stream owner.
     let Some(lease_scope) = lease_scope.take() else {
-        return resp(NativeResponse::error(
+        return resp(sqlstate_error(
             seq,
             "XX000",
             "internal error: query lease scope missing before materialized SQL dispatch",
diff --git a/nodedb/src/control/server/native/dispatch/sql_admin.rs b/nodedb/src/control/server/native/dispatch/sql_admin.rs
index 9f78638b6..ac3df3a72 100644
--- a/nodedb/src/control/server/native/dispatch/sql_admin.rs
+++ b/nodedb/src/control/server/native/dispatch/sql_admin.rs
@@ -9,6 +9,7 @@ use nodedb_types::protocol::NativeResponse;
 use nodedb_types::value::Value;
 
 use super::{DispatchCtx, error_to_native};
+use crate::control::server::native::sqlstate_code::sqlstate_error;
 
 // ─── SET / SHOW / RESET (SQL form) ─────────────────────────────────
 
@@ -36,7 +37,7 @@ pub(super) fn handle_set_sql(ctx: &DispatchCtx<'_>, seq: u64, sql: &str) -> Nati
                 .to_string(),
         )
     } else {
-        return NativeResponse::error(seq, "42601", "invalid SET syntax");
+        return sqlstate_error(seq, "42601", "invalid SET syntax");
     };
 
     ctx.sessions.set_parameter(ctx.peer_addr, key, value);
diff --git a/nodedb/src/control/server/native/dispatch/sql_loop.rs b/nodedb/src/control/server/native/dispatch/sql_loop.rs
index 9700d4025..e1f4483e0 100644
--- a/nodedb/src/control/server/native/dispatch/sql_loop.rs
+++ b/nodedb/src/control/server/native/dispatch/sql_loop.rs
@@ -32,8 +32,9 @@ use super::sql_dispatch_task::dispatch_task;
 use super::streaming::SqlOutcome;
 use super::{
     DispatchCtx, error_code_to_native, error_response_to_native, error_to_native,
-    shape_error_to_native, to_native_columns_rows,
+    error_to_native_with_sqlstate, shape_error_to_native, to_native_columns_rows,
 };
+use crate::control::server::native::sqlstate_code::sqlstate_error;
 
 /// Wrap a materialized response as a non-streaming [`SqlOutcome`].
 #[inline]
@@ -69,11 +70,7 @@ pub(super) async fn run_dispatch_loop(
 
     for task in tasks {
         if task.tenant_id != ctx.tenant_id() {
-            return resp(NativeResponse::error(
-                seq,
-                "42501",
-                "tenant isolation violation",
-            ));
+            return resp(sqlstate_error(seq, "42501", "tenant isolation violation"));
         }
 
         // Cloned before `route_in_tx_write` consumes `task`, so a staged
@@ -101,7 +98,7 @@ pub(super) async fn run_dispatch_loop(
         if let Some(info) = &plan_metering_info
             && let Err(e) = admit_quota_for_dispatch(ctx.state, &ctx.scope, info)
         {
-            return resp(NativeResponse::error(seq, "53400", e.to_string()));
+            return resp(error_to_native_with_sqlstate(seq, "53400", &e));
         }
 
         // In transaction: route through the protocol-neutral staging gate.
@@ -154,7 +151,7 @@ pub(super) async fn run_dispatch_loop(
                 Arc::clone(&plan_lease_scope),
             )
         {
-            return resp(NativeResponse::error(
+            return resp(sqlstate_error(
                 seq,
                 "XX000",
                 "internal error: failed to retain descriptor leases for buffered transaction tasks",
diff --git a/nodedb/src/control/server/native/dispatch/streaming.rs b/nodedb/src/control/server/native/dispatch/streaming.rs
index b3c8bf329..ea67d6c75 100644
--- a/nodedb/src/control/server/native/dispatch/streaming.rs
+++ b/nodedb/src/control/server/native/dispatch/streaming.rs
@@ -47,7 +47,7 @@ impl SqlOutcome {
     pub(crate) fn into_response(self) -> NativeResponse {
         match self {
             SqlOutcome::Response(r) => *r,
-            SqlOutcome::Stream(s) => NativeResponse::error(
+            SqlOutcome::Stream(s) => crate::control::server::native::sqlstate_code::sqlstate_error(
                 s.seq,
                 "XX000",
                 "internal error: SQL stream produced on a non-streaming path",
diff --git a/nodedb/src/control/server/native/dispatch/transaction.rs b/nodedb/src/control/server/native/dispatch/transaction.rs
index a01e587ac..415cb8be3 100644
--- a/nodedb/src/control/server/native/dispatch/transaction.rs
+++ b/nodedb/src/control/server/native/dispatch/transaction.rs
@@ -81,7 +81,12 @@ pub(crate) fn handle_begin(ctx: &DispatchCtx<'_>, seq: u64) -> NativeResponse {
                 crate::Error::BadRequest { detail } => detail.clone(),
                 other => other.to_string(),
             };
-            NativeResponse::error(seq, "25P02", message)
+            NativeResponse::error_with_code(
+                seq,
+                "25P02",
+                message,
+                crate::error_classify::classify(&e).code().0,
+            )
         }
     }
 }
diff --git a/nodedb/src/control/server/native/dispatch/transaction_savepoint.rs b/nodedb/src/control/server/native/dispatch/transaction_savepoint.rs
index e1eb43113..4a304cd92 100644
--- a/nodedb/src/control/server/native/dispatch/transaction_savepoint.rs
+++ b/nodedb/src/control/server/native/dispatch/transaction_savepoint.rs
@@ -15,18 +15,17 @@ use crate::control::server::shared::session::savepoint_ops::{self, SavepointErro
 
 use super::DispatchCtx;
 use super::transaction::NativeTxnDp;
+use crate::control::server::native::sqlstate_code::sqlstate_error;
 
 /// Map a neutral savepoint error to a native error frame.
 fn savepoint_error_to_native(seq: u64, e: &SavepointError) -> NativeResponse {
     match e {
-        SavepointError::NoActiveTransaction => NativeResponse::error(
+        SavepointError::NoActiveTransaction => sqlstate_error(
             seq,
             "25P01",
             "SAVEPOINT can only be used in transaction blocks",
         ),
-        SavepointError::NotFound { message } => {
-            NativeResponse::error(seq, "3B001", message.clone())
-        }
+        SavepointError::NotFound { message } => sqlstate_error(seq, "3B001", message.clone()),
     }
 }
 
diff --git a/nodedb/src/control/server/native/mod.rs b/nodedb/src/control/server/native/mod.rs
index 9d423721f..a641e529c 100644
--- a/nodedb/src/control/server/native/mod.rs
+++ b/nodedb/src/control/server/native/mod.rs
@@ -4,3 +4,4 @@ pub mod codec;
 pub mod dispatch;
 pub mod handshake;
 pub mod session;
+pub mod sqlstate_code;
diff --git a/nodedb/src/control/server/native/session/auth.rs b/nodedb/src/control/server/native/session/auth.rs
index 27e034a82..591972581 100644
--- a/nodedb/src/control/server/native/session/auth.rs
+++ b/nodedb/src/control/server/native/session/auth.rs
@@ -12,6 +12,8 @@ use crate::control::server::shared::authorization::authorize_database;
 
 use super::NativeSession;
 use super::dispatch;
+use crate::control::server::native::dispatch::error_to_native_with_sqlstate;
+use crate::control::server::native::sqlstate_code::sqlstate_error;
 
 impl NativeSession {
     /// Handle authentication request.
@@ -22,7 +24,7 @@ impl NativeSession {
         // a client silently swap to a different (database, tenant) scope while
         // still holding the original scope's connection slots.
         if self.identity.is_some() || self.connection_permit.is_some() {
-            return NativeResponse::error(
+            return sqlstate_error(
                 seq,
                 "0A000",
                 "already authenticated; reconnect to switch identity",
@@ -33,11 +35,11 @@ impl NativeSession {
             RequestFields::Text(f) => match &f.auth {
                 Some(a) => a,
                 None => {
-                    return NativeResponse::error(seq, "28000", "missing 'auth' field");
+                    return sqlstate_error(seq, "28000", "missing 'auth' field");
                 }
             },
             _ => {
-                return NativeResponse::error(seq, "0A000", "unsupported request fields variant");
+                return sqlstate_error(seq, "0A000", "unsupported request fields variant");
             }
         };
 
@@ -65,7 +67,7 @@ impl NativeSession {
                     self.transport,
                     &self.peer_addr.to_string(),
                 ) {
-                    return NativeResponse::error(seq, "28000", format!("{e}"));
+                    return error_to_native_with_sqlstate(seq, "28000", &e);
                 }
 
                 // Bind the requested database before acquiring any scoped
@@ -80,18 +82,14 @@ impl NativeSession {
                     Some(name) => match catalog.get_database_id_by_name(name) {
                         Ok(Some(db_id)) => db_id,
                         Ok(None) => {
-                            return NativeResponse::error(
+                            return sqlstate_error(
                                 seq,
                                 "3D000",
                                 "selected database does not exist",
                             );
                         }
                         Err(_) => {
-                            return NativeResponse::error(
-                                seq,
-                                "XX000",
-                                "database catalog lookup failed",
-                            );
+                            return sqlstate_error(seq, "XX000", "database catalog lookup failed");
                         }
                     },
                     None => identity
@@ -105,18 +103,10 @@ impl NativeSession {
                 match catalog.get_database(db_id) {
                     Ok(Some(_)) => {}
                     Ok(None) => {
-                        return NativeResponse::error(
-                            seq,
-                            "3D000",
-                            "selected database does not exist",
-                        );
+                        return sqlstate_error(seq, "3D000", "selected database does not exist");
                     }
                     Err(_) => {
-                        return NativeResponse::error(
-                            seq,
-                            "XX000",
-                            "database catalog lookup failed",
-                        );
+                        return sqlstate_error(seq, "XX000", "database catalog lookup failed");
                     }
                 }
 
@@ -124,7 +114,7 @@ impl NativeSession {
                 // database- or tenant-scoped capacity or mutate session state.
                 let audit = ArcAuditEmitter(std::sync::Arc::clone(&self.state.audit));
                 if authorize_database(&identity, db_id, &audit).is_err() {
-                    return NativeResponse::error(seq, "42501", "permission denied for database");
+                    return sqlstate_error(seq, "42501", "permission denied for database");
                 }
 
                 // Phase 2 admission: acquire per-database and per-tenant permits
@@ -134,10 +124,11 @@ impl NativeSession {
                 let db_permit = match self.admission_registry.try_acquire_database(db_id) {
                     Ok(p) => p,
                     Err(e) => {
-                        return NativeResponse::error(
+                        return NativeResponse::error_with_code(
                             seq,
                             nodedb_types::error::sqlstate::QUOTA_EXCEEDED,
                             format!("{e}"),
+                            nodedb_types::error::ErrorCode::DATABASE_QUOTA_EXCEEDED.0,
                         );
                     }
                 };
@@ -147,10 +138,11 @@ impl NativeSession {
                         Err(e) => {
                             // db_permit is dropped here, releasing the DB slot.
                             drop(db_permit);
-                            return NativeResponse::error(
+                            return NativeResponse::error_with_code(
                                 seq,
                                 nodedb_types::error::sqlstate::QUOTA_EXCEEDED,
                                 format!("{e}"),
+                                nodedb_types::error::ErrorCode::TENANT_QUOTA_EXCEEDED.0,
                             );
                         }
                     };
@@ -165,7 +157,7 @@ impl NativeSession {
                     // don't leak slots into the per-DB / per-tenant pools.
                     drop(tenant_permit);
                     drop(db_permit);
-                    return NativeResponse::error(
+                    return sqlstate_error(
                         seq,
                         "XX000",
                         "internal error: global admission permit missing during auth assembly",
@@ -229,12 +221,12 @@ impl NativeSession {
             // other auth error (wrong password, lockout, unknown user) stays
             // collapsed into the generic invalid-password 28P01 so none can be
             // distinguished from the others.
-            Err(e @ crate::Error::RateExceeded { .. }) => NativeResponse::error(
+            Err(e @ crate::Error::RateExceeded { .. }) => error_to_native_with_sqlstate(
                 seq,
                 nodedb_types::error::sqlstate::TOO_MANY_CONNECTIONS,
-                format!("{e}"),
+                &e,
             ),
-            Err(e) => NativeResponse::error(seq, "28P01", format!("{e}")),
+            Err(e) => sqlstate_error(seq, "28P01", format!("{e}")),
         }
     }
 }
diff --git a/nodedb/src/control/server/native/session/request.rs b/nodedb/src/control/server/native/session/request.rs
index 994f16cac..ad81dfadd 100644
--- a/nodedb/src/control/server/native/session/request.rs
+++ b/nodedb/src/control/server/native/session/request.rs
@@ -8,6 +8,7 @@ use nodedb_types::protocol::{AuthMethod, NativeResponse, OpCode, RequestFields,
 use super::NativeSession;
 use super::dispatch::{self, DispatchCtx};
 use crate::config::auth::AuthMode;
+use crate::control::server::native::sqlstate_code::sqlstate_error;
 
 impl NativeSession {
     /// Route a decoded request to the appropriate handler.
@@ -62,7 +63,7 @@ impl NativeSession {
                 let Some(trust_id) =
                     super::super::super::session_auth::configured_trust_identity(&self.state)
                 else {
-                    return SqlOutcome::Response(Box::new(NativeResponse::error(
+                    return SqlOutcome::Response(Box::new(sqlstate_error(
                         seq,
                         "28000",
                         "configured trust identity is unavailable",
@@ -88,7 +89,7 @@ impl NativeSession {
                     return SqlOutcome::Response(Box::new(auth_response));
                 }
             } else {
-                return SqlOutcome::Response(Box::new(NativeResponse::error(
+                return SqlOutcome::Response(Box::new(sqlstate_error(
                     seq,
                     "28000",
                     "not authenticated. Send Auth request first.",
@@ -99,7 +100,7 @@ impl NativeSession {
         let identity = match self.identity.as_ref() {
             Some(id) => id,
             None => {
-                return SqlOutcome::Response(Box::new(NativeResponse::error(
+                return SqlOutcome::Response(Box::new(sqlstate_error(
                     seq,
                     "28000",
                     "not authenticated",
@@ -199,7 +200,7 @@ impl NativeSession {
         let fields = match &req.fields {
             RequestFields::Text(f) => f,
             _ => {
-                return SqlOutcome::Response(Box::new(NativeResponse::error(
+                return SqlOutcome::Response(Box::new(sqlstate_error(
                     seq,
                     "0A000",
                     "unsupported request field format for this server version",
@@ -213,7 +214,7 @@ impl NativeSession {
             let sql = match &fields.sql {
                 Some(s) => s.as_str(),
                 None => {
-                    return SqlOutcome::Response(Box::new(NativeResponse::error(
+                    return SqlOutcome::Response(Box::new(sqlstate_error(
                         seq,
                         "42601",
                         "missing 'sql' field",
@@ -239,7 +240,7 @@ impl NativeSession {
                                 dispatch::handle_sql(&ctx, seq, sql, None).await,
                             ));
                         }
-                        return SqlOutcome::Response(Box::new(NativeResponse::error(
+                        return SqlOutcome::Response(Box::new(sqlstate_error(
                             seq,
                             "42601",
                             "missing 'key' field",
@@ -258,7 +259,7 @@ impl NativeSession {
                                 dispatch::handle_sql(&ctx, seq, sql, None).await,
                             ));
                         }
-                        return SqlOutcome::Response(Box::new(NativeResponse::error(
+                        return SqlOutcome::Response(Box::new(sqlstate_error(
                             seq,
                             "42601",
                             "missing 'key' field",
@@ -271,7 +272,7 @@ impl NativeSession {
                 let key = match &fields.key {
                     Some(k) => k.as_str(),
                     None => {
-                        return SqlOutcome::Response(Box::new(NativeResponse::error(
+                        return SqlOutcome::Response(Box::new(sqlstate_error(
                             seq,
                             "42601",
                             "missing 'key' field",
@@ -291,7 +292,7 @@ impl NativeSession {
                 let sql = match &fields.sql {
                     Some(s) => s.as_str(),
                     None => {
-                        return SqlOutcome::Response(Box::new(NativeResponse::error(
+                        return SqlOutcome::Response(Box::new(sqlstate_error(
                             seq,
                             "42601",
                             "missing 'sql' field",
@@ -380,7 +381,7 @@ impl NativeSession {
                 let sql = match &fields.sql {
                     Some(s) => s.as_str(),
                     None => {
-                        return SqlOutcome::Response(Box::new(NativeResponse::error(
+                        return SqlOutcome::Response(Box::new(sqlstate_error(
                             seq,
                             "42601",
                             "missing 'sql' field",
@@ -394,7 +395,7 @@ impl NativeSession {
             OpCode::Auth | OpCode::Ping | OpCode::Status => unreachable!(),
             // OpCode is #[non_exhaustive]; future opcodes that reach this
             // handler before session.rs is updated return a typed error.
-            _ => NativeResponse::error(seq, "0A000", "opcode not supported by this server version"),
+            _ => sqlstate_error(seq, "0A000", "opcode not supported by this server version"),
         };
 
         SqlOutcome::Response(Box::new(response))
diff --git a/nodedb/src/control/server/native/session/run.rs b/nodedb/src/control/server/native/session/run.rs
index 712d63b57..90ad7a339 100644
--- a/nodedb/src/control/server/native/session/run.rs
+++ b/nodedb/src/control/server/native/session/run.rs
@@ -18,6 +18,7 @@ use super::NativeSession;
 use super::codec::{self, FrameFormat};
 use super::dispatch;
 use super::session_chunk::chunk_large_response;
+use crate::control::server::native::sqlstate_code::sqlstate_error;
 
 /// Rollback dependencies retained independently from the connection future.
 ///
@@ -187,11 +188,8 @@ impl NativeSession {
                     "session absolute timeout ({}s), closing connection",
                     absolute_timeout_secs
                 );
-                let shutdown_resp = NativeResponse::error(
-                    0,
-                    "57P01",
-                    "session timeout: absolute lifetime exceeded",
-                );
+                let shutdown_resp =
+                    sqlstate_error(0, "57P01", "session timeout: absolute lifetime exceeded");
                 if let Ok(bytes) = super::codec::encode_response(
                     &shutdown_resp,
                     self.format.unwrap_or(FrameFormat::MessagePack),
@@ -224,8 +222,12 @@ impl NativeSession {
                 Ok(None) => return Ok(()), // clean EOF
                 Err(crate::Error::BadRequest { detail }) => {
                     // Send a typed error before closing so the client knows why.
-                    let err_resp =
-                        NativeResponse::error(0, "54000", format!("frame rejected: {detail}"));
+                    let err_resp = NativeResponse::error_with_code(
+                        0,
+                        "54000",
+                        format!("frame rejected: {detail}"),
+                        nodedb_types::error::ErrorCode::BAD_REQUEST.0,
+                    );
                     let format = self.format.unwrap_or(FrameFormat::MessagePack);
                     if let Ok(bytes) = codec::encode_response(&err_resp, format) {
                         let _ = codec::write_frame(&mut self.stream, &bytes).await;
@@ -248,7 +250,7 @@ impl NativeSession {
             // Decode and handle.
             let outcome = match codec::decode_request(&payload, format) {
                 Ok(req) => self.handle_request(req).await,
-                Err(e) => dispatch::SqlOutcome::Response(Box::new(NativeResponse::error(
+                Err(e) => dispatch::SqlOutcome::Response(Box::new(sqlstate_error(
                     0,
                     "42601",
                     format!("{e}"),
diff --git a/nodedb/src/control/server/native/sqlstate_code.rs b/nodedb/src/control/server/native/sqlstate_code.rs
new file mode 100644
index 000000000..e22ac5196
--- /dev/null
+++ b/nodedb/src/control/server/native/sqlstate_code.rs
@@ -0,0 +1,158 @@
+// SPDX-License-Identifier: BUSL-1.1
+
+//! Numeric NodeDB codes for native error frames authored as a bare SQLSTATE.
+//!
+//! A native error frame carries both a SQLSTATE and the stable numeric NodeDB
+//! code, and the client rebuilds its typed error from the number: a frame that
+//! ships `ndb_code == 0` collapses on arrival into a generic internal failure,
+//! so `is_not_found()`, `is_auth_denied()` and `is_retriable()` all answer
+//! wrongly for it.
+//!
+//! Most frames get their number from [`crate::error_classify::classify`],
+//! which is the one internal-`Error`-to-public mapping the crate owns. This
+//! module exists for the frames that never held an `Error` to classify: a DDL
+//! refusal ([`DdlError`](crate::control::server::shared::ddl::DdlError) is
+//! authored as a SQLSTATE plus a message, in ~600 places, and has no numeric
+//! code to carry), and the session/dispatch guards that reject a request with
+//! a literal SQLSTATE and a static message. For those the SQLSTATE *is* the
+//! only classification the server ever produced, so reading it back is a
+//! lookup rather than a guess.
+//!
+//! This is the inverse of the client-side rule in
+//! `NodeDbError::from_wire`, and deliberately so. There, every SQLSTATE the
+//! server can emit arrives through one funnel, so a reverse mapping would have
+//! to resolve `23505` into either a unique violation or a duplicate
+//! idempotency key with no way to tell them apart. Here the lookup happens at
+//! the site that chose the SQLSTATE, and the table only carries SQLSTATEs
+//! whose NodeDB classification is unambiguous *whatever* site emitted them.
+//!
+//! Everything else maps to `0`, which is exactly the frame today's code ships,
+//! so an unmapped SQLSTATE is never worse off than before this table existed.
+//! Three groups stay unmapped on purpose:
+//!
+//! - **Overloaded SQLSTATEs.** `53400` is `QUOTA_OVERCOMMIT`,
+//!   `TENANT_QUOTA_EXCEEDED`, `DATABASE_QUOTA_EXCEEDED` and `SERVER_OVERLOAD`;
+//!   `0A000` is `SQL_NOT_ENABLED`, `CANNOT_DROP_DEFAULT_DATABASE` and
+//!   `CANNOT_CLONE_MIRROR`. A caller that knows which one it is passes the
+//!   code explicitly instead of routing through this table.
+//! - **SQLSTATEs with no NodeDB variant.** `42P07` (duplicate table), `42704`
+//!   (undefined object), `25P02` (aborted transaction), `3B001` (no such
+//!   savepoint). These need new `ErrorCode`/`ErrorDetails` variants to type at
+//!   all, which is a public-API change tracked separately.
+//! - **SQLSTATEs that are deliberately undistinguished.** Every credential
+//!   failure renders as `28P01` and every ILP auth failure as a single code
+//!   with one message, precisely so a caller cannot tell a wrong password from
+//!   an unknown user. Typing them would rebuild the oracle that collapsing
+//!   removed.
+
+use nodedb_types::error::{ErrorCode, sqlstate};
+use nodedb_types::protocol::NativeResponse;
+
+/// The numeric NodeDB code a bare `sqlstate` classifies to, or `0` when it
+/// carries no unambiguous classification.
+pub(crate) fn ndb_code_for_sqlstate(sqlstate_str: &str) -> u16 {
+    let code = match sqlstate_str {
+        sqlstate::UNDEFINED_TABLE => ErrorCode::COLLECTION_NOT_FOUND,
+        sqlstate::INVALID_CATALOG_NAME => ErrorCode::DATABASE_NOT_FOUND,
+        sqlstate::INSUFFICIENT_PRIVILEGE => ErrorCode::AUTHORIZATION_DENIED,
+        sqlstate::UNDEFINED_FUNCTION => ErrorCode::UNDEFINED_FUNCTION,
+        // Both a malformed request and a plan that cannot be built render as
+        // `42601`, so this cannot say which. It does not have to: the two
+        // differ in which side wrote the bad statement, not in how a client
+        // must react, and both `BadRequest` and `PlanError` are client errors
+        // that no caller should retry.
+        sqlstate::SYNTAX_ERROR => ErrorCode::BAD_REQUEST,
+        // A cross-shard OCC abort and a retryable refusal both mean "nothing
+        // applied, retry the whole thing" — the same contract `WriteConflict`
+        // states, and the classification a retry loop reads.
+        sqlstate::SERIALIZATION_FAILURE => ErrorCode::WRITE_CONFLICT,
+        sqlstate::QUERY_CANCELED => ErrorCode::DEADLINE_EXCEEDED,
+        sqlstate::TOO_MANY_CONNECTIONS => ErrorCode::RATE_EXCEEDED,
+        sqlstate::INTERNAL_ERROR => ErrorCode::INTERNAL,
+        _ => return 0,
+    };
+    code.0
+}
+
+/// Build a native error frame from a bare SQLSTATE, classifying it through
+/// [`ndb_code_for_sqlstate`].
+///
+/// Use this wherever a site rejects a request with a literal SQLSTATE and no
+/// `Error` value. A site that holds an `Error` must use
+/// `error_to_native` / `error_to_native_with_sqlstate` instead: those read the
+/// classification the error already carries rather than inferring one.
+pub(crate) fn sqlstate_error(
+    seq: u64,
+    sqlstate_str: impl Into<String>,
+    message: impl Into<String>,
+) -> NativeResponse {
+    let sqlstate_str = sqlstate_str.into();
+    let ndb_code = ndb_code_for_sqlstate(&sqlstate_str);
+    NativeResponse::error_with_code(seq, sqlstate_str, message, ndb_code)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn classified_sqlstates_carry_their_code() {
+        assert_eq!(
+            ndb_code_for_sqlstate("42P01"),
+            ErrorCode::COLLECTION_NOT_FOUND.0
+        );
+        assert_eq!(
+            ndb_code_for_sqlstate("42501"),
+            ErrorCode::AUTHORIZATION_DENIED.0
+        );
+        assert_eq!(ndb_code_for_sqlstate("42601"), ErrorCode::BAD_REQUEST.0);
+        assert_eq!(
+            ndb_code_for_sqlstate("3D000"),
+            ErrorCode::DATABASE_NOT_FOUND.0
+        );
+        assert_eq!(ndb_code_for_sqlstate("XX000"), ErrorCode::INTERNAL.0);
+    }
+
+    /// A retry loop reads the numeric code, so the SQLSTATE the server sends
+    /// precisely to get a transaction retried must not arrive unclassified.
+    #[test]
+    fn serialization_failure_stays_retriable() {
+        let frame = sqlstate_error(1, "40001", "OCC abort");
+        let payload = frame.error.expect("error frames carry a payload");
+        assert_eq!(payload.ndb_code, ErrorCode::WRITE_CONFLICT.0);
+        assert!(
+            nodedb_types::NodeDbError::from_wire(ErrorCode(payload.ndb_code), payload.message)
+                .is_retriable()
+        );
+    }
+
+    /// An overloaded or unmapped SQLSTATE must fall through to `0` rather than
+    /// pick a side: `0` is what the frame ships today, so an unknown SQLSTATE
+    /// is no worse off, while a wrong guess would misreport retriability.
+    #[test]
+    fn ambiguous_and_unknown_sqlstates_stay_unclassified() {
+        // Overloaded across several NodeDB variants.
+        assert_eq!(ndb_code_for_sqlstate("53400"), 0);
+        assert_eq!(ndb_code_for_sqlstate("0A000"), 0);
+        // No NodeDB variant exists to map onto.
+        assert_eq!(ndb_code_for_sqlstate("42P07"), 0);
+        assert_eq!(ndb_code_for_sqlstate("42704"), 0);
+        // Deliberately undistinguished so credential failures stay opaque.
+        assert_eq!(ndb_code_for_sqlstate("28P01"), 0);
+        assert_eq!(ndb_code_for_sqlstate("28000"), 0);
+        // Not a SQLSTATE this server emits.
+        assert_eq!(ndb_code_for_sqlstate("99999"), 0);
+    }
+
+    /// An unclassified frame must still reach the client exactly as it does
+    /// today — same SQLSTATE, same message, `ndb_code == 0` — so adding the
+    /// table cannot regress a path it does not cover.
+    #[test]
+    fn unclassified_frame_is_unchanged() {
+        let frame = sqlstate_error(7, "42P07", "table 'repro_t' already exists");
+        let payload = frame.error.expect("error frames carry a payload");
+        assert_eq!(payload.code, "42P07");
+        assert_eq!(payload.message, "table 'repro_t' already exists");
+        assert_eq!(payload.ndb_code, 0);
+    }
+}
diff --git a/nodedb/tests/native_error_code_classification.rs b/nodedb/tests/native_error_code_classification.rs
index 88f8c18a5..ed79c30d2 100644
--- a/nodedb/tests/native_error_code_classification.rs
+++ b/nodedb/tests/native_error_code_classification.rs
@@ -304,3 +304,71 @@ async fn native_permission_denial_carries_authorization_code() {
         err.message
     );
 }
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn native_ddl_refusal_carries_collection_not_found_code() {
+    let server = TestServer::start().await;
+
+    let mut stream = native_session(&server).await;
+    // DDL is refused by the DDL router, whose neutral error is authored as a
+    // SQLSTATE and a message and never holds a classified `Error` — so this
+    // frame reached the client with `ndb_code == 0` after the SQL and
+    // direct-op paths above were already carrying their codes. The same
+    // statement over the SQL path (`SELECT * FROM ...`) types correctly, so
+    // only a client that happened to run DDL saw the flattening.
+    let resp = send_sql(&mut stream, 1, "DROP TABLE native_err_absent_ddl").await;
+
+    assert_eq!(
+        resp.status,
+        ResponseStatus::Error,
+        "dropping an absent collection must be refused"
+    );
+    let err = resp.error.expect("error payload expected");
+    assert_eq!(
+        err.code,
+        sqlstate::UNDEFINED_TABLE,
+        "an absent collection must map to its own SQLSTATE, got {}: {}",
+        err.code,
+        err.message
+    );
+    assert_eq!(
+        err.ndb_code,
+        ErrorCode::COLLECTION_NOT_FOUND.0,
+        "a DDL refusal must carry the numeric code too, got {} ({})",
+        err.ndb_code,
+        err.message
+    );
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn native_duplicate_ddl_keeps_its_sqlstate_and_message() {
+    let server = TestServer::start().await;
+
+    let mut stream = native_session(&server).await;
+    let create = send_sql(&mut stream, 1, "CREATE COLLECTION native_err_dup").await;
+    assert_eq!(create.status, ResponseStatus::Ok, "create the target");
+
+    // `42P07` has no `ErrorDetails` variant to map onto, so it stays
+    // unclassified rather than being forced onto an approximate one. This
+    // pins that the fallback is still exactly what shipped before — same
+    // SQLSTATE, same message, `ndb_code == 0` — so an unmapped SQLSTATE is
+    // never made worse by the table that classifies the mapped ones.
+    let resp = send_sql(&mut stream, 2, "CREATE COLLECTION native_err_dup").await;
+
+    assert_eq!(
+        resp.status,
+        ResponseStatus::Error,
+        "creating an existing collection must be refused"
+    );
+    let err = resp.error.expect("error payload expected");
+    assert_eq!(err.code, "42P07", "duplicate object keeps its own SQLSTATE");
+    assert!(
+        err.message.contains("native_err_dup"),
+        "the server's message must survive verbatim, got {}",
+        err.message
+    );
+    assert_eq!(
+        err.ndb_code, 0,
+        "a SQLSTATE with no NodeDB variant must stay unclassified, not be guessed at"
+    );
+}