fix(errors): unify backend error protocol and preserve db detail

crates/dbx-core/src/backend_error.rs

@@ -1,4 +1,4 @@
-//! Stable, safe backend error envelopes and the v1 JDBC Agent catalog.
+//! Stable backend error envelopes and the v1 JDBC Agent catalog.
 
 use std::collections::BTreeMap;
 
@@ -9,9 +9,12 @@ use crate::db::agent_driver::{
     AgentErrorStage, AgentOperationOutcome, AgentSessionDisposition,
 };
 
-const MAX_DETAIL_BYTES: usize = 512;
+const MAX_DETAIL_BYTES: usize = 64 * 1024;
 
-/// The origin of a backend error as exposed on the wire.
+/// The v1 compatibility source of a backend error.
+///
+/// New code should use `origin` for subsystem/adapter information. This field
+/// remains unchanged so older clients can continue to consume v1 envelopes.
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
 #[serde(rename_all = "camelCase")]
 pub enum BackendErrorSource {
@@ -20,6 +23,53 @@ pub enum BackendErrorSource {
     LegacyBackend,
 }
 
+/// Extensible subsystem metadata for backend errors.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub enum BackendSubsystem {
+    Database,
+    Tunnel,
+    Extension,
+    Ai,
+    MessageQueue,
+    Backend,
+}
+
+/// The adapter that produced the error within its subsystem.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub enum BackendErrorAdapter {
+    JdbcAgent,
+    JdbcAgentLegacy,
+    Native,
+    Plugin,
+    Http,
+    Legacy,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct BackendErrorOrigin {
+    subsystem: BackendSubsystem,
+    adapter: BackendErrorAdapter,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    driver: Option<&'static str>,
+}
+
+impl BackendErrorOrigin {
+    const fn database(adapter: BackendErrorAdapter) -> Self {
+        Self { subsystem: BackendSubsystem::Database, adapter, driver: None }
+    }
+
+    const fn database_driver(adapter: BackendErrorAdapter, driver: &'static str) -> Self {
+        Self { subsystem: BackendSubsystem::Database, adapter, driver: Some(driver) }
+    }
+
+    const fn backend() -> Self {
+        Self { subsystem: BackendSubsystem::Backend, adapter: BackendErrorAdapter::Legacy, driver: None }
+    }
+}
+
 /// Whether the operation definitely had not started or its result is unknown.
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
 #[serde(rename_all = "snake_case")]
@@ -51,6 +101,8 @@ pub struct BackendErrorDiagnostics {
     vendor_code: Option<i32>,
     #[serde(skip_serializing_if = "Option::is_none")]
     exception_class: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    adapter_code: Option<String>,
 }
 
 /// Public v1 backend error envelope.
@@ -65,6 +117,7 @@ pub struct BackendError {
     message_key: String,
     message_params: BTreeMap<String, BackendMessageParam>,
     source: BackendErrorSource,
+    origin: BackendErrorOrigin,
     operation_outcome: BackendOperationOutcome,
     #[serde(skip_serializing_if = "Option::is_none")]
     detail: Option<String>,
@@ -80,37 +133,51 @@ impl BackendError {
         match error {
             AgentCallError::Structured { message, context, .. } => {
                 let (entry, outcome) = structured_entry(context);
+                let detail = match context.category {
+                    AgentErrorCategory::Sql => bounded_native_detail(message),
+                    _ => bounded_detail(message),
+                };
                 Self::new(
                     entry,
                     BackendErrorSource::JdbcAgent,
+                    BackendErrorOrigin::database(BackendErrorAdapter::JdbcAgent),
                     outcome,
                     stage_param(entry, context.stage),
-                    safe_detail(message),
+                    detail,
                     Some(diagnostics_from_context(context)),
                 )
             }
-            AgentCallError::Legacy { message, .. } => Self::new(
-                catalog_entry(CatalogCode::JdbcLegacyFailure),
-                BackendErrorSource::JdbcAgentLegacy,
-                BackendOperationOutcome::Unknown,
-                BTreeMap::new(),
-                safe_detail(message),
-                None,
-            ),
+            AgentCallError::Legacy { message, hints, .. } => {
+                let detail = match hints.category {
+                    Some(AgentErrorCategory::Sql) => bounded_native_detail(message),
+                    _ => bounded_detail(message),
+                };
+                Self::new(
+                    catalog_entry(CatalogCode::JdbcLegacyFailure),
+                    BackendErrorSource::JdbcAgentLegacy,
+                    BackendErrorOrigin::database(BackendErrorAdapter::JdbcAgentLegacy),
+                    BackendOperationOutcome::Unknown,
+                    BTreeMap::new(),
+                    detail,
+                    None,
+                )
+            }
             AgentCallError::ContractViolation { message, .. } => Self::new(
                 catalog_entry(CatalogCode::ContractInvalid),
                 BackendErrorSource::JdbcAgent,
+                BackendErrorOrigin::database(BackendErrorAdapter::JdbcAgent),
                 BackendOperationOutcome::Unknown,
                 BTreeMap::new(),
-                safe_detail(message),
+                bounded_detail(message),
                 None,
             ),
             AgentCallError::Transport { message } => Self::new(
                 catalog_entry(CatalogCode::ProtocolFailed),
                 BackendErrorSource::JdbcAgent,
+                BackendErrorOrigin::database(BackendErrorAdapter::JdbcAgent),
                 BackendOperationOutcome::Unknown,
                 BTreeMap::new(),
-                safe_detail(message),
+                bounded_detail(message),
                 None,
             ),
             AgentCallError::Timeout { stage, operation_outcome } => {
@@ -121,6 +188,7 @@ impl BackendError {
                 Self::new(
                     entry,
                     BackendErrorSource::JdbcAgent,
+                    BackendErrorOrigin::database(BackendErrorAdapter::JdbcAgent),
                     map_outcome(*operation_outcome),
                     stage_param(entry, *stage),
                     None,
@@ -130,6 +198,7 @@ impl BackendError {
             AgentCallError::Canceled { stage, operation_outcome } => Self::new(
                 catalog_entry(CatalogCode::Canceled),
                 BackendErrorSource::JdbcAgent,
+                BackendErrorOrigin::database(BackendErrorAdapter::JdbcAgent),
                 map_outcome(*operation_outcome),
                 stage_param(catalog_entry(CatalogCode::Canceled), *stage),
                 None,
@@ -143,9 +212,10 @@ impl BackendError {
         Self::new(
             catalog_entry(CatalogCode::LegacyBackend),
             BackendErrorSource::LegacyBackend,
+            BackendErrorOrigin::backend(),
             BackendOperationOutcome::Unknown,
             BTreeMap::new(),
-            safe_detail(message),
+            bounded_detail(message),
             None,
         )
     }
@@ -156,32 +226,57 @@ impl BackendError {
         Self::new(
             entry,
             BackendErrorSource::JdbcAgent,
+            BackendErrorOrigin::database(BackendErrorAdapter::Native),
             BackendOperationOutcome::Unknown,
             stage_param(entry, AgentErrorStage::Execute),
-            safe_detail(message),
+            bounded_detail(message),
             Some(diagnostics_for_local("timeout", AgentErrorStage::Execute)),
         )
     }
 
-    /// Create a SQL failure envelope while retaining the bounded diagnostic detail.
+    /// Create a SQL failure envelope while retaining bounded native driver detail.
+    ///
+    /// The caller must pass a typed SQL execution error, not a connection or
+    /// transport diagnostic. Native SQL text is intentionally not parsed or
+    /// rewritten because the database dialect owns its format.
     pub fn from_sql_detail(message: &str) -> Self {
         let entry = catalog_entry(CatalogCode::SqlFailed);
         Self::new(
             entry,
             BackendErrorSource::JdbcAgent,
+            BackendErrorOrigin::database(BackendErrorAdapter::Native),
             BackendOperationOutcome::Unknown,
             stage_param(entry, AgentErrorStage::Execute),
-            safe_detail(message),
+            bounded_native_detail(message),
             Some(diagnostics_for_local("sql", AgentErrorStage::Execute)),
         )
     }
 
+    /// Adapt a DuckDB worker error while retaining both the native detail and
+    /// the worker protocol code for diagnostics at the public boundary.
+    pub fn from_duckdb_worker_error(code: &str, message: &str) -> Self {
+        let is_sql_error = matches!(code, "duckdb_execute_failed" | "duckdb_worker_poisoned");
+        let entry = catalog_entry(if is_sql_error { CatalogCode::SqlFailed } else { CatalogCode::LegacyBackend });
+        let source = if is_sql_error { BackendErrorSource::JdbcAgent } else { BackendErrorSource::LegacyBackend };
+        let category = if is_sql_error { "sql" } else { "backend" };
+        Self::new(
+            entry,
+            source,
+            BackendErrorOrigin::database_driver(BackendErrorAdapter::Native, "duckdb"),
+            BackendOperationOutcome::Unknown,
+            stage_param(entry, AgentErrorStage::Execute),
+            if is_sql_error { bounded_native_detail(message) } else { bounded_detail(message) },
+            Some(diagnostics_for_local_with_adapter_code(category, AgentErrorStage::Execute, code)),
+        )
+    }
+
     /// Create a cancellation envelope for work canceled by the Rust executor.
     pub fn from_canceled(stage: AgentErrorStage, operation_outcome: AgentOperationOutcome) -> Self {
         let entry = catalog_entry(CatalogCode::Canceled);
         Self::new(
             entry,
             BackendErrorSource::LegacyBackend,
+            BackendErrorOrigin::database(BackendErrorAdapter::Native),
             map_outcome(operation_outcome),
             stage_param(entry, stage),
             None,
@@ -216,6 +311,10 @@ impl BackendError {
         self.source
     }
 
+    pub fn origin(&self) -> BackendErrorOrigin {
+        self.origin
+    }
+
     pub fn operation_outcome(&self) -> BackendOperationOutcome {
         self.operation_outcome
     }
@@ -236,6 +335,7 @@ impl BackendError {
     fn new(
         entry: &'static CatalogEntry,
         source: BackendErrorSource,
+        origin: BackendErrorOrigin,
         operation_outcome: BackendOperationOutcome,
         message_params: BTreeMap<String, BackendMessageParam>,
         detail: Option<String>,
@@ -249,6 +349,7 @@ impl BackendError {
             message_key: entry.message_key.to_string(),
             message_params,
             source,
+            origin,
             operation_outcome,
             detail,
             diagnostics,
@@ -506,13 +607,23 @@ fn diagnostics_from_context(context: &AgentErrorContext) -> BackendErrorDiagnost
         sql_state: context.sql_state.as_deref().map(|value| bounded_ascii(value, 32)),
         vendor_code: context.vendor_code,
         exception_class: context.exception_class.as_deref().map(|value| bounded_ascii(value, 128)),
+        ..Default::default()
     }
 }
 
 fn diagnostics_for_local(category: &str, stage: AgentErrorStage) -> BackendErrorDiagnostics {
+    diagnostics_for_local_with_adapter_code(category, stage, "")
+}
+
+fn diagnostics_for_local_with_adapter_code(
+    category: &str,
+    stage: AgentErrorStage,
+    adapter_code: &str,
+) -> BackendErrorDiagnostics {
     BackendErrorDiagnostics {
         category: Some(category.to_string()),
         stage: Some(stage_name(stage).to_string()),
+        adapter_code: (!adapter_code.is_empty()).then(|| bounded_ascii(adapter_code, 64)),
         ..Default::default()
     }
 }
@@ -532,53 +643,358 @@ fn bounded_text(value: &str, max_bytes: usize) -> String {
     value[..end].to_string()
 }
 
+fn bounded_detail(message: &str) -> Option<String> {
+    if message.trim().is_empty() {
+        return None;
+    }
+    let detail = safe_detail(message)?;
+    let detail = bounded_text(&detail, MAX_DETAIL_BYTES);
+    (!detail.is_empty()).then_some(detail)
+}
+
+fn bounded_native_detail(message: &str) -> Option<String> {
+    // This path is only for typed native SQL failures. Do not infer or rewrite
+    // SQL content here; preserving the driver's diagnostic is the contract.
+    let trimmed = message.trim();
+    if trimmed.is_empty() {
+        return None;
+    }
+    let detail = bounded_text(trimmed, MAX_DETAIL_BYTES);
+    (!detail.is_empty()).then_some(detail)
+}
+
 fn safe_detail(message: &str) -> Option<String> {
     let trimmed = message.trim();
     if trimmed.is_empty() {
         return None;
     }
-    let normalized = trimmed.split_ascii_whitespace().collect::<Vec<_>>().join(" ");
-    let lowered = normalized.to_ascii_lowercase();
-    let sensitive_markers = [
-        "://",
-        "jdbc:",
-        "password",
-        "passwd",
-        "pwd",
-        "token",
-        "secret",
-        "authorization",
-        "bearer",
-        "api_key",
-        "apikey",
-        "credential",
-        "auth=",
-        "key=",
-        "user=",
-        "username=",
-        "uid=",
-        "access_key",
-        "private_key",
-        "agent session",
-        "agentsessionid",
-        "session id",
-        "session_id",
-        "session=",
-    ];
-    if sensitive_markers.iter().any(|marker| lowered.contains(marker)) {
+
+    let detail = redact_session_identifier(&redact_sensitive_fragments(trimmed));
+    if contains_only_redacted_sensitive_tokens(&detail) {
         return None;
     }
-    let sql_verbs = [
-        "select", "insert", "update", "delete", "drop", "create", "alter", "truncate", "merge", "call", "with",
-        "grant", "revoke", "comment", "explain", "begin", "commit", "rollback",
-    ];
-    if lowered.split(|ch: char| !ch.is_ascii_alphabetic()).any(|word| sql_verbs.contains(&word)) {
-        return None;
-    }
-    let detail = bounded_text(&normalized, MAX_DETAIL_BYTES);
     (!detail.is_empty()).then_some(detail)
 }
 
+fn contains_only_redacted_sensitive_tokens(value: &str) -> bool {
+    let mut has_token = false;
+    let all_sensitive = value.split_whitespace().all(|token| {
+        has_token = true;
+        if token == "[redacted]" || matches!(token, "=" | ":") {
+            return true;
+        }
+        let normalized = token.trim_matches(|ch: char| !ch.is_ascii_alphanumeric() && ch != '_' && ch != '-');
+        if sensitive_key_name(normalized) {
+            return true;
+        }
+        ['=', ':'].iter().any(|separator| {
+            token.split_once(*separator).is_some_and(|(key, value)| {
+                sensitive_key_name(key)
+                    && value.trim_matches(|ch| matches!(ch, ';' | ',' | ']' | ')' | '}')) == "[redacted]"
+            })
+        })
+    });
+    has_token && all_sensitive
+}
+
+fn redact_sensitive_fragments(value: &str) -> String {
+    let redacted_url = redact_url_userinfo(value);
+    let mut redacted = String::with_capacity(redacted_url.len());
+    let mut cursor = 0;
+    while let Some((value_start, value_end)) = next_sensitive_assignment(&redacted_url, cursor) {
+        redacted.push_str(&redacted_url[cursor..value_start]);
+        redacted.push_str("[redacted]");
+        cursor = value_end;
+    }
+    redacted.push_str(&redacted_url[cursor..]);
+    let source_tokens = redacted.split_whitespace().collect::<Vec<_>>();
+    let mut tokens = Vec::with_capacity(source_tokens.len());
+    let mut redact_next = false;
+    let mut changed = false;
+    let mut index = 0;
+    while index < source_tokens.len() {
+        let token = source_tokens[index];
+        if redact_next {
+            tokens.push("[redacted]");
+            redact_next = false;
+            changed = true;
+        } else if token.eq_ignore_ascii_case("bearer") {
+            tokens.push("[redacted]");
+            redact_next = true;
+            changed = true;
+        } else if token.eq_ignore_ascii_case("authorization:") && source_tokens.get(index + 1) == Some(&"[redacted]") {
+            tokens.push("[redacted]");
+            index += 1;
+            changed = true;
+        } else {
+            tokens.push(token);
+        }
+        index += 1;
+    }
+    if changed {
+        tokens.join(" ")
+    } else {
+        redacted
+    }
+}
+
+fn redact_url_userinfo(value: &str) -> String {
+    let mut ranges = Vec::new();
+    let mut search_from = 0;
+    while let Some(relative_scheme_end) = value[search_from..].find("://") {
+        let authority_start = search_from + relative_scheme_end + "://".len();
+        let authority_end = value[authority_start..]
+            .char_indices()
+            .find(|(_, ch)| matches!(ch, '/' | '?' | '#') || ch.is_ascii_whitespace())
+            .map(|(offset, _)| authority_start + offset)
+            .unwrap_or(value.len());
+        let authority = &value[authority_start..authority_end];
+
+        if let Some(user_info_end) = authority.rfind('@') {
+            let user_info = &authority[..user_info_end];
+            if let Some(password_separator) = user_info.find(':') {
+                let password_start = authority_start + password_separator + 1;
+                let password_end = authority_start + user_info_end;
+                if password_start < password_end {
+                    ranges.push((password_start, password_end));
+                }
+            }
+        }
+
+        if authority_end == value.len() {
+            break;
+        }
+        search_from = authority_end;
+    }
+
+    if ranges.is_empty() {
+        return value.to_string();
+    }
+
+    let mut redacted = String::with_capacity(value.len());
+    let mut cursor = 0;
+    for (start, end) in ranges {
+        redacted.push_str(&value[cursor..start]);
+        redacted.push_str("[redacted]");
+        cursor = end;
+    }
+    redacted.push_str(&value[cursor..]);
+    redacted
+}
+
+fn sensitive_key_name(key: &str) -> bool {
+    let key = key.chars().filter(|ch| ch.is_ascii_alphanumeric()).map(|ch| ch.to_ascii_lowercase()).collect::<String>();
+    matches!(
+        key.as_str(),
+        "password"
+            | "passwd"
+            | "pwd"
+            | "token"
+            | "accesstoken"
+            | "refreshtoken"
+            | "secret"
+            | "authorization"
+            | "apikey"
+            | "credential"
+            | "auth"
+            | "key"
+            | "user"
+            | "username"
+            | "uid"
+            | "accesskey"
+            | "privatekey"
+            | "session"
+            | "sessionid"
+            | "agentsessionid"
+            | "jwt"
+            | "cookie"
+    )
+}
+
+fn next_sensitive_assignment(value: &str, from: usize) -> Option<(usize, usize)> {
+    let mut index = from;
+    while index < value.len() {
+        let ch = value[index..].chars().next()?;
+        if !is_sensitive_key_char(ch)
+            || (index > 0 && value[..index].chars().next_back().is_some_and(is_sensitive_key_char))
+        {
+            index += ch.len_utf8();
+            continue;
+        }
+
+        let key_start = index;
+        let mut key_end = index;
+        while key_end < value.len() {
+            let key_char = value[key_end..].chars().next()?;
+            if !is_sensitive_key_char(key_char) {
+                break;
+            }
+            key_end += key_char.len_utf8();
+        }
+        if !sensitive_key_name(&value[key_start..key_end]) {
+            index = key_end;
+            continue;
+        }
+
+        let mut separator_start = key_end;
+        if let Some(quote) = value[..key_start].chars().next_back().filter(|ch| matches!(ch, '\'' | '"')) {
+            if value[separator_start..].starts_with(quote) {
+                separator_start += quote.len_utf8();
+            }
+        }
+        while separator_start < value.len()
+            && value[separator_start..].chars().next().is_some_and(|ch| ch.is_ascii_whitespace())
+        {
+            separator_start += value[separator_start..].chars().next()?.len_utf8();
+        }
+        let separator = value[separator_start..].chars().next()?;
+        if !matches!(separator, '=' | ':') {
+            index = key_end;
+            continue;
+        }
+
+        let mut value_start = separator_start + separator.len_utf8();
+        while value_start < value.len()
+            && value[value_start..].chars().next().is_some_and(|ch| ch.is_ascii_whitespace())
+        {
+            value_start += value[value_start..].chars().next()?.len_utf8();
+        }
+        let value_end = consume_sensitive_value(value, value_start, &value[key_start..key_end]);
+        return Some((value_start, value_end));
+    }
+    None
+}
+
+fn is_sensitive_key_char(ch: char) -> bool {
+    ch.is_ascii_alphanumeric() || matches!(ch, '_' | '-')
+}
+
+fn consume_sensitive_value(value: &str, start: usize, key: &str) -> usize {
+    if start >= value.len() {
+        return start;
+    }
+    let first = value[start..].chars().next().unwrap_or_default();
+    if first == '{' {
+        return consume_braced_value(value, start);
+    }
+    if matches!(first, '\'' | '"') {
+        let mut escaped = false;
+        let mut iter = value[start + first.len_utf8()..].char_indices();
+        while let Some((offset, ch)) = iter.next() {
+            if escaped {
+                escaped = false;
+                continue;
+            }
+            if ch == '\\' {
+                escaped = true;
+                continue;
+            }
+            if ch == first {
+                let end = start + first.len_utf8() + offset + ch.len_utf8();
+                if value[end..].starts_with(first) {
+                    iter.next();
+                    continue;
+                }
+                return end;
+            }
+        }
+        return value.len();
+    }
+
+    let mut end = start;
+    while end < value.len() {
+        let ch = value[end..].chars().next().unwrap_or_default();
+        if ch.is_ascii_whitespace() || matches!(ch, '&' | ';' | ',' | ']' | ')' | '}') {
+            break;
+        }
+        end += ch.len_utf8();
+    }
+
+    let normalized_key =
+        key.chars().filter(|ch| ch.is_ascii_alphanumeric()).map(|ch| ch.to_ascii_lowercase()).collect::<String>();
+    if normalized_key == "authorization" {
+        let scheme = value[start..end].to_ascii_lowercase();
+        if matches!(scheme.as_str(), "bearer" | "basic" | "digest") {
+            let mut token_start = end;
+            while token_start < value.len()
+                && value[token_start..].chars().next().is_some_and(|ch| ch.is_ascii_whitespace())
+            {
+                token_start += value[token_start..].chars().next().unwrap().len_utf8();
+            }
+            let mut token_end = token_start;
+            while token_end < value.len() {
+                let ch = value[token_end..].chars().next().unwrap_or_default();
+                if ch.is_ascii_whitespace() || matches!(ch, '&' | ';' | ',' | ']' | ')' | '}') {
+                    break;
+                }
+                token_end += ch.len_utf8();
+            }
+            return token_end;
+        }
+    }
+    end
+}
+
+fn consume_braced_value(value: &str, start: usize) -> usize {
+    let mut index = start + '{'.len_utf8();
+    while index < value.len() {
+        let ch = value[index..].chars().next().unwrap_or_default();
+        if ch == '}' {
+            let next = index + ch.len_utf8();
+            if value[next..].starts_with('}') {
+                index = next + '}'.len_utf8();
+                continue;
+            }
+            return next;
+        }
+        index += ch.len_utf8();
+    }
+    value.len()
+}
+
+fn redact_session_identifier(value: &str) -> String {
+    let lowered = value.to_ascii_lowercase();
+    for marker in ["session id", "session_id", "agentsessionid"] {
+        if let Some(marker_start) = lowered.find(marker) {
+            let value_start = value[marker_start + marker.len()..]
+                .char_indices()
+                .find(|(_, ch)| !ch.is_ascii_whitespace() && *ch != ':' && *ch != '=')
+                .map(|(offset, _)| marker_start + marker.len() + offset);
+            if let Some(value_start) = value_start {
+                let value_end = value[value_start..]
+                    .char_indices()
+                    .find(|(_, ch)| ch.is_ascii_whitespace())
+                    .map(|(offset, _)| value_start + offset)
+                    .unwrap_or(value.len());
+                let mut result = value.to_string();
+                result.replace_range(value_start..value_end, "[redacted]");
+                return result;
+            }
+        }
+    }
+    let Some(colon) = value.find(':') else {
+        return value.to_string();
+    };
+    if !lowered[..colon].contains("session") {
+        return value.to_string();
+    }
+    let value_start = value[colon + 1..]
+        .char_indices()
+        .find(|(_, ch)| !ch.is_ascii_whitespace())
+        .map(|(offset, _)| colon + 1 + offset);
+    let Some(value_start) = value_start else {
+        return value.to_string();
+    };
+    let value_end = value[value_start..]
+        .char_indices()
+        .find(|(_, ch)| ch.is_ascii_whitespace())
+        .map(|(offset, _)| value_start + offset)
+        .unwrap_or(value.len());
+    let mut result = value.to_string();
+    result.replace_range(value_start..value_end, "[redacted]");
+    result
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -675,17 +1091,65 @@ mod tests {
     }
 
     #[test]
-    fn unsafe_detail_is_omitted_and_internal_fields_are_not_serialized() {
-        for message in [
-            "DROP TABLE users",
-            "SELECT\tpassword FROM users",
-            "postgresql://host/db?user=alice&pwd=secret",
-            "Bearer token-value",
-            "CALL refresh_cache()",
-            "syntax error in statement [UPDATE customers SET ssn='123']",
-            "Agent session not found: 7f51e7f4-7cee-42db-bfeb-76d1199d1afe",
+    fn sql_errors_expose_generic_database_origin_without_changing_v1_source() {
+        let payload =
+            serde_json::to_value(BackendError::from_sql_detail("relation missing_table does not exist")).unwrap();
+        assert_eq!(payload["source"], "jdbcAgent");
+        assert_eq!(payload["origin"]["subsystem"], "database");
+        assert_eq!(payload["origin"]["adapter"], "native");
+    }
+
+    #[test]
+    fn duckdb_worker_sql_error_preserves_worker_code_detail_and_driver_origin() {
+        let payload = serde_json::to_value(BackendError::from_duckdb_worker_error(
+            "duckdb_execute_failed",
+            "Parser Error: syntax error at or near SELECT",
+        ))
+        .unwrap();
+
+        assert_eq!(payload["code"], "DBX-JDBC-4001");
+        assert_eq!(payload["detail"], "Parser Error: syntax error at or near SELECT");
+        assert_eq!(payload["origin"]["subsystem"], "database");
+        assert_eq!(payload["origin"]["adapter"], "native");
+        assert_eq!(payload["origin"]["driver"], "duckdb");
+        assert_eq!(payload["diagnostics"]["adapterCode"], "duckdb_execute_failed");
+    }
+
+    #[test]
+    fn public_detail_redacts_credentials_and_preserves_native_sql_text() {
+        for (message, expected) in [
+            ("Incorrect syntax near SELECT", "Incorrect syntax near SELECT"),
+            ("ERROR: relation missing_table does not exist", "ERROR: relation missing_table does not exist"),
+            ("ORA-00942: table or view does not exist", "ORA-00942: table or view does not exist"),
+            (
+                "syntax error in statement [UPDATE customers SET ssn='123']",
+                "syntax error in statement [UPDATE customers SET ssn='123']",
+            ),
+            (
+                "syntax error at or near 'SELECT email FROM customers WHERE ssn = 123'",
+                "syntax error at or near 'SELECT email FROM customers WHERE ssn = 123'",
+            ),
+            (
+                "failed executing SELECT * FROM [Users] WHERE email='literal-secret@example.com'",
+                "failed executing SELECT * FROM [Users] WHERE email='literal-secret@example.com'",
+            ),
+            (
+                "connection failed: jdbc:postgresql://host/db?user=alice&password=secret",
+                "connection failed: jdbc:postgresql://host/db?user=[redacted]&password=[redacted]",
+            ),
+            ("connection failed: Authorization: Bearer abc123", "connection failed: [redacted]"),
+            ("Agent session not found: 7f51e7f4-7cee-42db-bfeb-76d1199d1afe", "Agent session not found: [redacted]"),
+            ("Agent session id private-session", "Agent session id [redacted]"),
         ] {
-            assert!(safe_detail(message).is_none(), "detail leaked: {message}");
+            assert_eq!(bounded_detail(message).as_deref(), Some(expected), "detail changed unexpectedly: {message}");
+        }
+        for message in ["DROP TABLE users", "SELECT password FROM users", "CALL refresh_cache()"] {
+            assert_eq!(bounded_detail(message).as_deref(), Some(message));
+        }
+        for message in
+            ["Bearer token-value", "Authorization: Bearer token-value", "password = secret", "password: secret"]
+        {
+            assert!(bounded_detail(message).is_none(), "credential detail leaked: {message}");
         }
         let error = BackendError::from_agent_call_error(&AgentCallError::Structured {
             rpc_code: -1,
@@ -701,7 +1165,104 @@ mod tests {
         assert!(value.get("retryable").is_none());
         assert!(value.get("sessionDisposition").is_none());
         assert!(value.get("agentSessionId").is_none());
-        assert!(value.get("detail").is_none());
+        assert_eq!(value["detail"], "jdbc:postgresql://host/db?password=[redacted]");
+    }
+
+    #[test]
+    fn public_detail_redacts_sensitive_key_value_variants() {
+        let message = concat!(
+            "driver failed: PASSWORD:secret-a passwd = 'secret-b' PWD : \"secret-c\" ",
+            "refresh_token=secret-d access-token : secret-e api_key='secret-f' ",
+            "apikey : secret-g credential=secret-h private-key:secret-i ",
+            "authorization: Bearer secret-j cookie = \"session=secret-k\" ",
+            "sessionid:secret-l jwt = secret-m secret:secret-n token = secret-o ",
+            "access_token:secret-p session=secret-q private_key='secret-r'"
+        );
+
+        let detail = bounded_detail(message).expect("non-sensitive context should remain");
+        for secret in [
+            "secret-a", "secret-b", "secret-c", "secret-d", "secret-e", "secret-f", "secret-g", "secret-h", "secret-i",
+            "secret-j", "secret-k", "secret-l", "secret-m", "secret-n", "secret-o", "secret-p", "secret-q", "secret-r",
+        ] {
+            assert!(!detail.contains(secret), "sensitive value leaked: {secret}; detail={detail}");
+        }
+        assert!(detail.contains("driver failed"));
+    }
+
+    #[test]
+    fn public_detail_redacts_mixed_url_and_dsn_sensitive_fields() {
+        let message = concat!(
+            "connection failed for jdbc:postgresql://db.example/app?user=alice&password : url-secret&sslmode=require ",
+            "host=db.example port=5432 password:\"dsn secret\" session = 'session-secret'"
+        );
+
+        let detail = bounded_detail(message).expect("non-sensitive context should remain");
+        for secret in ["alice", "url-secret", "dsn secret", "session-secret"] {
+            assert!(!detail.contains(secret), "sensitive value leaked: {secret}; detail={detail}");
+        }
+        assert!(detail.contains("connection failed"));
+    }
+
+    #[test]
+    fn serialized_native_sql_detail_preserves_url_like_text_and_braced_literals() {
+        let cases = [
+            (
+                "syntax error in statement [SELECT 'jdbc:postgresql://alice:url-secret@db.example/app']",
+                "syntax error in statement [SELECT 'jdbc:postgresql://alice:url-secret@db.example/app']",
+            ),
+            (
+                "syntax error in statement [SELECT 'Driver={PostgreSQL};PWD={dsn;secret};SERVER=db.example']",
+                "syntax error in statement [SELECT 'Driver={PostgreSQL};PWD={dsn;secret};SERVER=db.example']",
+            ),
+        ];
+
+        for (message, expected) in cases {
+            let payload = serde_json::to_value(BackendError::from_sql_detail(message)).unwrap();
+            assert_eq!(payload["detail"].as_str(), Some(expected), "detail changed unexpectedly: {message}");
+        }
+    }
+
+    #[test]
+    fn serialized_public_detail_preserves_nested_sql_and_literals() {
+        let message = "syntax error in statement [SELECT * FROM [Users] WHERE email='literal-secret@example.com']";
+        let payload = serde_json::to_value(BackendError::from_sql_detail(message)).unwrap();
+
+        assert_eq!(payload["detail"].as_str(), Some(message));
+    }
+
+    #[test]
+    fn structured_connection_detail_still_redacts_credentials() {
+        let error = BackendError::from_agent_call_error(&AgentCallError::Structured {
+            rpc_code: -1,
+            message: "connection failed: jdbc:postgresql://alice:url-secret@db.example/app".to_string(),
+            context: context(
+                AgentErrorCategory::Connection,
+                AgentErrorStage::Connect,
+                AgentOperationOutcome::NotStarted,
+            ),
+        });
+
+        assert_eq!(error.detail(), Some("connection failed: jdbc:postgresql://alice:[redacted]@db.example/app"));
+    }
+
+    #[test]
+    fn structured_sql_detail_preserves_native_text() {
+        let message = "syntax error in statement [UPDATE users SET password='user-input']";
+        let error = BackendError::from_agent_call_error(&AgentCallError::Structured {
+            rpc_code: -1,
+            message: message.to_string(),
+            context: context(AgentErrorCategory::Sql, AgentErrorStage::Execute, AgentOperationOutcome::Unknown),
+        });
+
+        assert_eq!(error.detail(), Some(message));
+    }
+
+    #[test]
+    fn bounded_detail_remains_bounded_after_redaction() {
+        let detail = bounded_detail(&format!("ERROR: {}", "x".repeat(MAX_DETAIL_BYTES + 100))).unwrap();
+        assert!(detail.len() > 512);
+        assert!(detail.len() <= MAX_DETAIL_BYTES);
+        assert!(detail.is_char_boundary(MAX_DETAIL_BYTES));
     }
 
     #[test]
@@ -770,7 +1331,7 @@ mod tests {
         assert_eq!(error.code(), "DBX-JDBC-9001");
         assert_eq!(
             error.detail(),
-            Some("ERROR: relation \"dbx_table_that_does_not_exist\" does not exist Position: 15")
+            Some("ERROR: relation \"dbx_table_that_does_not_exist\" does not exist\n  Position: 15")
         );
     }
 
@@ -785,7 +1346,7 @@ mod tests {
         let error = BackendError::from_legacy_string(&legacy);
 
         assert_eq!(error.code(), "DBX-JDBC-9001");
-        assert_eq!(error.detail(), Some("无效的表或视图名 错误码: -2106"));
+        assert_eq!(error.detail(), Some("无效的表或视图名\n错误码: -2106"));
     }
 
     #[test]

crates/dbx-core/src/db/duckdb_worker_process.rs

@@ -195,6 +195,17 @@ impl DuckDbWorkerClient {
         cancel_token: Option<CancellationToken>,
         query_timeout: Option<Duration>,
     ) -> Result<db::QueryResult, String> {
+        self.execute_typed(database, sql, max_rows, cancel_token, query_timeout).await.map_err(|error| error.message)
+    }
+
+    pub async fn execute_typed(
+        &self,
+        database: Option<String>,
+        sql: String,
+        max_rows: Option<usize>,
+        cancel_token: Option<CancellationToken>,
+        query_timeout: Option<Duration>,
+    ) -> Result<db::QueryResult, DuckDbWorkerError> {
         let _query_guard = self.inner.query_lock.lock().await;
         let client = self.clone();
         // Cancellation and timeout restart the worker via cancel_or_kill below. An ordinary
@@ -206,7 +217,10 @@ impl DuckDbWorkerClient {
         // than letting the worker self-exit) avoids racing our own next request against a dying
         // worker, and OS-level kill never runs the destructor that would abort the process.
         let future = async move {
-            client.ensure_connected().await?;
+            client
+                .ensure_connected()
+                .await
+                .map_err(|message| DuckDbWorkerError::new("duckdb_worker_connect_failed", message))?;
             match client
                 .send_request_structured::<db::QueryResult>(
                     DuckDbWorkerMethod::Execute,
@@ -220,7 +234,7 @@ impl DuckDbWorkerClient {
                     if error.code == DUCKDB_WORKER_POISONED_CODE {
                         client.kill().await;
                     }
-                    Err(error.message)
+                    Err(error)
                 }
             }
         };
@@ -252,9 +266,9 @@ impl DuckDbWorkerClient {
         }
     }
 
-    async fn cancel_or_kill(&self, final_error: String) -> Result<db::QueryResult, String> {
+    async fn cancel_or_kill(&self, final_error: String) -> Result<db::QueryResult, DuckDbWorkerError> {
         let _ = self.cancel().await;
-        Err(final_error)
+        Err(DuckDbWorkerError::from(final_error))
     }
 
     pub async fn list_databases(&self) -> Result<Vec<db::DatabaseInfo>, String> {

crates/dbx-core/src/query.rs

@@ -36,7 +36,7 @@ pub const QUERY_CANCELED: &str = "Query canceled";
 /// (desktop/CLI diagnose first; this is only the Rust SQL-executor backstop).
 const MONGO_SHELL_COMMAND_HINT: &str = "Use MongoDB shell-style commands, for example: db.collection.find({}).limit(100), db.collection.aggregate([]), db.collection.aggregate([], { explain: true }), db.version(), db.collection.countDocuments({}), db.collection.distinct(\"field\"), db.collection.getIndexes(), db.collection.createIndex({...}), or db.collection.insertOne({...}).";
 const SQL_OMITTED_ERROR_CONTEXT: &str =
-    "SQL text omitted from user-facing error; enable debug SQL diagnostics for a redacted statement.";
+    "SQL text omitted from user-facing error; enable debug SQL diagnostics to inspect the original statement.";
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub enum PoolErrorAction {
@@ -48,6 +48,7 @@ pub enum PoolErrorAction {
 #[derive(Debug, Clone)]
 pub enum QueryExecutionError {
     Agent(AgentCallError),
+    DuckDb { code: String, message: String },
     Canceled { stage: AgentErrorStage, operation_outcome: AgentOperationOutcome },
     Timeout(String),
     Sql(String),
@@ -58,6 +59,7 @@ impl QueryExecutionError {
     pub fn into_legacy_string(self) -> String {
         match self {
             Self::Agent(error) => error.into_legacy_string(),
+            Self::DuckDb { message, .. } => message,
             Self::Canceled { .. } => canceled_error(),
             Self::Timeout(error) => error,
             Self::Sql(error) => error,
@@ -68,6 +70,9 @@ impl QueryExecutionError {
     pub fn into_backend_error(self) -> crate::backend_error::BackendError {
         match self {
             Self::Agent(error) => crate::backend_error::BackendError::from_agent_call_error(&error),
+            Self::DuckDb { code, message } => {
+                crate::backend_error::BackendError::from_duckdb_worker_error(&code, &message)
+            }
             Self::Canceled { stage, operation_outcome } => {
                 crate::backend_error::BackendError::from_canceled(stage, operation_outcome)
             }
@@ -80,9 +85,12 @@ impl QueryExecutionError {
     fn with_omitted_sql_context(self, sql: &str) -> Self {
         match self {
             Self::Agent(error) => Self::Agent(error),
+            Self::DuckDb { code, message } => {
+                Self::DuckDb { code, message: query_error_with_omitted_sql_context(&message, sql) }
+            }
             canceled @ Self::Canceled { .. } => canceled,
             Self::Timeout(error) => Self::Timeout(query_error_with_omitted_sql_context(&error, sql)),
-            Self::Sql(error) => Self::Sql(query_error_with_omitted_sql_context(&error, sql)),
+            Self::Sql(error) => Self::Sql(append_typed_sql_error_context(&error, sql)),
             Self::Legacy(error) => Self::Legacy(query_error_with_omitted_sql_context(&error, sql)),
         }
     }
@@ -90,6 +98,7 @@ impl QueryExecutionError {
     fn with_context(self, context: &str) -> Self {
         match self {
             Self::Agent(error) => Self::Agent(error),
+            Self::DuckDb { code, message } => Self::DuckDb { code, message: format!("{message}; {context}") },
             canceled @ Self::Canceled { .. } => canceled,
             Self::Timeout(error) => Self::Timeout(format!("{error}; {context}")),
             Self::Sql(error) => Self::Sql(format!("{error}; {context}")),
@@ -100,7 +109,7 @@ impl QueryExecutionError {
     fn as_agent_error(&self) -> Option<&AgentCallError> {
         match self {
             Self::Agent(error) => Some(error),
-            Self::Canceled { .. } | Self::Timeout(_) | Self::Sql(_) | Self::Legacy(_) => None,
+            Self::DuckDb { .. } | Self::Canceled { .. } | Self::Timeout(_) | Self::Sql(_) | Self::Legacy(_) => None,
         }
     }
 }
@@ -109,6 +118,7 @@ impl std::fmt::Display for QueryExecutionError {
     fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         match self {
             Self::Agent(error) => error.fmt(formatter),
+            Self::DuckDb { message, .. } => formatter.write_str(message),
             Self::Canceled { .. } => formatter.write_str(QUERY_CANCELED),
             Self::Timeout(error) => formatter.write_str(error),
             Self::Sql(error) => formatter.write_str(error),
@@ -139,6 +149,14 @@ fn query_error_with_omitted_sql_context(error: &str, _sql: &str) -> String {
     crate::db::agent_driver::append_legacy_error_context(error, SQL_OMITTED_ERROR_CONTEXT)
 }
 
+fn append_typed_sql_error_context(error: &str, _sql: &str) -> String {
+    if error.contains(SQL_OMITTED_ERROR_CONTEXT) {
+        return error.to_string();
+    }
+    let separator = if error.trim_start().starts_with("Server error:") { " " } else { "\n" };
+    format!("{error}{separator}{SQL_OMITTED_ERROR_CONTEXT}")
+}
+
 /// A multi-statement result with metadata intended for query clients.
 ///
 /// `execution_error` is emitted for synthesized per-statement errors so clients
@@ -900,6 +918,7 @@ fn query_execution_error_action(
     }
     match error {
         QueryExecutionError::Canceled { .. } => PoolErrorAction::Keep,
+        QueryExecutionError::DuckDb { message, .. } => query_pool_error_action(db_type, sql, message),
         QueryExecutionError::Timeout(message)
         | QueryExecutionError::Sql(message)
         | QueryExecutionError::Legacy(message) => query_pool_error_action(db_type, sql, message),
@@ -1252,6 +1271,8 @@ async fn do_execute_typed(
     let pool = connections.get(pool_key).ok_or("Connection not found")?;
 
     let mut typed_agent_error = None;
+    #[cfg(feature = "duckdb-sidecar")]
+    let mut typed_duckdb_error = None;
     let result: Result<db::QueryResult, String> = match pool {
         #[cfg(feature = "duckdb-sidecar")]
         PoolKind::DuckDbWorker(client) => {
@@ -1271,7 +1292,17 @@ async fn do_execute_typed(
             let database = database.map(str::to_string);
             let max_rows = options.max_rows;
             drop(connections);
-            client.execute(database, sql, max_rows, cancel_token, query_timeout).await
+            match client.execute_typed(database, sql, max_rows, cancel_token, query_timeout).await {
+                Ok(result) => Ok(result),
+                Err(error) => {
+                    let is_control_error = error.message == QUERY_CANCELED
+                        || is_dbx_query_timeout_error(&error.message.to_ascii_lowercase());
+                    if !is_control_error {
+                        typed_duckdb_error = Some(error.clone());
+                    }
+                    Err(error.message)
+                }
+            }
         }
         #[cfg(not(feature = "duckdb-sidecar"))]
         PoolKind::DuckDbWorker(_) => {
@@ -1625,6 +1656,10 @@ async fn do_execute_typed(
     result
         .map(normalize_query_result_for_js)
         .map_err(|error| {
+            #[cfg(feature = "duckdb-sidecar")]
+            if let Some(duckdb_error) = typed_duckdb_error {
+                return QueryExecutionError::DuckDb { code: duckdb_error.code, message: duckdb_error.message };
+            }
             typed_agent_error.map_or_else(|| QueryExecutionError::Legacy(error), QueryExecutionError::Agent)
         })
         .map_err(|error| classify_query_error(pool_db_type, error))
@@ -5044,6 +5079,18 @@ for line in sys.stdin:
         assert_eq!(error.into_backend_error().code(), "DBX-JDBC-1002");
     }
 
+    #[test]
+    fn duckdb_worker_error_preserves_catalog_identity_and_detail() {
+        let error = QueryExecutionError::DuckDb {
+            code: "duckdb_execute_failed".to_string(),
+            message: "Catalog Error: Table missing_table does not exist".to_string(),
+        };
+        let backend_error = error.into_backend_error();
+
+        assert_eq!(backend_error.code(), "DBX-JDBC-4001");
+        assert_eq!(backend_error.detail(), Some("Catalog Error: Table missing_table does not exist"));
+    }
+
     #[test]
     fn query_timeout_preserves_timeout_catalog_identity_and_detail() {
         let error = classify_query_error(
@@ -5060,7 +5107,7 @@ for line in sys.stdin:
         );
         assert_eq!(
             backend_error.detail(),
-            Some("Query timed out after 1 seconds SQL text omitted from user-facing error; enable debug SQL diagnostics for a redacted statement.")
+            Some("Query timed out after 1 seconds\nSQL text omitted from user-facing error; enable debug SQL diagnostics to inspect the original statement.")
         );
     }
 
@@ -5081,7 +5128,7 @@ for line in sys.stdin:
         assert_eq!(
             backend_error.detail(),
             Some(
-                "ERROR: relation \"dbx_table_that_does_not_exist\" does not exist SQL text omitted from user-facing error; enable debug SQL diagnostics for a redacted statement."
+                "ERROR: relation \"dbx_table_that_does_not_exist\" does not exist\nSQL text omitted from user-facing error; enable debug SQL diagnostics to inspect the original statement."
             )
         );
     }
@@ -5106,7 +5153,7 @@ for line in sys.stdin:
         assert_eq!(
             backend_error.detail(),
             Some(
-                "Server error: `ERROR 1064 (42000): You have an error in your SQL syntax` SQL text omitted from user-facing error; enable debug SQL diagnostics for a redacted statement."
+                "Server error: `ERROR 1064 (42000): You have an error in your SQL syntax` SQL text omitted from user-facing error; enable debug SQL diagnostics to inspect the original statement."
             )
         );
     }
@@ -5127,7 +5174,7 @@ for line in sys.stdin:
         assert_eq!(
             backend_error.detail(),
             Some(
-                "ERROR: relation \"dbx_table_that_does_not_exist\" does not exist SQL text omitted from user-facing error; enable debug SQL diagnostics for a redacted statement."
+                "ERROR: relation \"dbx_table_that_does_not_exist\" does not exist\nSQL text omitted from user-facing error; enable debug SQL diagnostics to inspect the original statement."
             )
         );
     }
@@ -5147,7 +5194,7 @@ for line in sys.stdin:
         assert_eq!(
             backend_error.detail(),
             Some(
-                "Statement 1 failed: ERROR: relation \"dbx_table_that_does_not_exist\" does not exist SQL text omitted from user-facing error; enable debug SQL diagnostics for a redacted statement."
+                "Statement 1 failed: ERROR: relation \"dbx_table_that_does_not_exist\" does not exist\nSQL text omitted from user-facing error; enable debug SQL diagnostics to inspect the original statement."
             )
         );
     }
@@ -5511,6 +5558,18 @@ for line in sys.stdin:
         assert_eq!(repeated.matches(SQL_OMITTED_ERROR_CONTEXT).count(), 1);
     }
 
+    #[test]
+    fn typed_sql_error_context_keeps_driver_text_on_one_line() {
+        let error = QueryExecutionError::Sql("Server error: `ERROR 1064 (42000): syntax error`".to_string())
+            .with_omitted_sql_context("SELECT * FROM users")
+            .into_backend_error();
+
+        assert_eq!(
+            error.detail(),
+            Some("Server error: `ERROR 1064 (42000): syntax error` SQL text omitted from user-facing error; enable debug SQL diagnostics to inspect the original statement.")
+        );
+    }
+
     #[test]
     fn reconnect_retry_error_context_omits_raw_sql() {
         let sql = "select 'secret-123' as token";