fix(query): preserve structured backend errors

crates/dbx-core/src/agent_recovery.rs

@@ -0,0 +1,87 @@
+//! Recovery decisions derived from typed JDBC Agent failures.
+
+use crate::db::agent_driver::{AgentCallError, AgentErrorCategory, AgentSessionDisposition};
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum RecoveryScope {
+    UserOperation,
+    ReadOnlyMetadata { retried: bool },
+    Keepalive,
+    ConnectionOpen,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum RecoveryDecision {
+    KeepSession,
+    RetryReadOnlyMetadata,
+    QuarantineSession,
+    ReplaceRuntime,
+}
+
+impl RecoveryDecision {
+    pub fn discards_session(self) -> bool {
+        !matches!(self, Self::KeepSession)
+    }
+
+    pub fn replaces_runtime(self) -> bool {
+        matches!(self, Self::ReplaceRuntime)
+    }
+}
+
+pub struct RecoveryPolicy;
+
+impl RecoveryPolicy {
+    pub fn decide(error: &AgentCallError, scope: RecoveryScope) -> RecoveryDecision {
+        match error {
+            AgentCallError::ContractViolation { .. } => RecoveryDecision::QuarantineSession,
+            AgentCallError::Transport { .. } => RecoveryDecision::ReplaceRuntime,
+            AgentCallError::Timeout { .. } | AgentCallError::Canceled { .. } => RecoveryDecision::QuarantineSession,
+            AgentCallError::Structured { context, .. } => {
+                if matches!(context.category, AgentErrorCategory::Timeout | AgentErrorCategory::Canceled) {
+                    return RecoveryDecision::QuarantineSession;
+                }
+                Self::from_disposition(Some(context.session_disposition), Some(context.category), scope)
+            }
+            AgentCallError::Legacy { hints, .. } => {
+                Self::from_disposition(hints.session_disposition, hints.category, scope)
+            }
+        }
+    }
+
+    fn from_disposition(
+        disposition: Option<AgentSessionDisposition>,
+        category: Option<AgentErrorCategory>,
+        scope: RecoveryScope,
+    ) -> RecoveryDecision {
+        match disposition {
+            Some(AgentSessionDisposition::ReplaceRuntime) => RecoveryDecision::ReplaceRuntime,
+            Some(AgentSessionDisposition::Quarantine)
+                if category == Some(AgentErrorCategory::Connection)
+                    && matches!(scope, RecoveryScope::ReadOnlyMetadata { retried: false }) =>
+            {
+                RecoveryDecision::RetryReadOnlyMetadata
+            }
+            Some(AgentSessionDisposition::Quarantine) => RecoveryDecision::QuarantineSession,
+            Some(AgentSessionDisposition::Keep) | None => RecoveryDecision::KeepSession,
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    #[test]
+    fn business_modules_do_not_consume_legacy_marker_fields() {
+        let business_sources = [
+            ("query.rs", include_str!("query.rs")),
+            ("schema.rs", include_str!("schema.rs")),
+            ("connection.rs", include_str!("connection.rs")),
+        ];
+        let retired_consumers = ["agent_rpc_error_category", "agent_rpc_error_session_id", "agent_session_disposition"];
+
+        for (name, source) in business_sources {
+            for consumer in retired_consumers {
+                assert!(!source.contains(consumer), "{name} still consumes {consumer}");
+            }
+        }
+    }
+}

crates/dbx-core/src/agent_runtime.rs

@@ -2,10 +2,9 @@ use serde::de::DeserializeOwned;
 use std::time::Duration;
 
 use crate::agent_manager::{AgentManager, DEFAULT_JRE_KEY};
+use crate::agent_recovery::{RecoveryPolicy, RecoveryScope};
 use crate::database_capabilities;
-use crate::db::agent_driver::{
-    agent_session_disposition, AgentDriverClient, AgentMethod, AgentRuntimeClient, AgentSessionDisposition,
-};
+use crate::db::agent_driver::{AgentCallError, AgentDriverClient, AgentMethod, AgentRuntimeClient};
 use crate::models::connection::DatabaseType;
 
 pub struct SharedConnectionOpenError {
@@ -113,7 +112,7 @@ pub async fn spawn_shared_connection_client(
         }
     };
     if let Err(err) = runtime
-        .call::<serde_json::Value>(AgentMethod::OpenSession.as_str(), session_params, Some(connect_timeout), None)
+        .call_typed::<serde_json::Value>(AgentMethod::OpenSession.as_str(), session_params, Some(connect_timeout), None)
         .await
     {
         let open_error = shared_connection_open_error(err, runtime.clone());
@@ -124,12 +123,11 @@ pub async fn spawn_shared_connection_client(
 }
 
 fn shared_connection_open_error(
-    message: String,
+    error: AgentCallError,
     runtime: std::sync::Arc<AgentRuntimeClient>,
 ) -> SharedConnectionOpenError {
-    let runtime =
-        (agent_session_disposition(&message) == Some(AgentSessionDisposition::ReplaceRuntime)).then_some(runtime);
-    SharedConnectionOpenError { message, runtime }
+    let runtime = RecoveryPolicy::decide(&error, RecoveryScope::ConnectionOpen).replaces_runtime().then_some(runtime);
+    SharedConnectionOpenError { message: error.into_legacy_string(), runtime }
 }
 
 async fn forget_unused_runtime_after_failed_open(
@@ -412,9 +410,17 @@ for line in sys.stdin:
     #[tokio::test]
     async fn replace_runtime_open_error_reports_runtime_without_killing_it_directly() {
         let (manager, _cell, runtime, script_path) = test_shared_runtime("replace-runtime-open-error").await;
-        let error = "Agent RPC error (-1): capacity exhausted\nDBX_AGENT_ERROR_DATA:{\"category\":\"resource\",\"sessionDisposition\":\"replace_runtime\"}";
+        let error = AgentCallError::Legacy {
+            rpc_code: Some(-1),
+            message: "capacity exhausted".to_string(),
+            hints: crate::db::agent_driver::LegacyAgentHints {
+                category: Some(crate::db::agent_driver::AgentErrorCategory::Resource),
+                session_disposition: Some(crate::db::agent_driver::AgentSessionDisposition::ReplaceRuntime),
+                ..Default::default()
+            },
+        };
 
-        let open_error = shared_connection_open_error(error.to_string(), runtime.clone());
+        let open_error = shared_connection_open_error(error, runtime.clone());
 
         assert!(open_error.runtime.as_ref().is_some_and(|failed| std::sync::Arc::ptr_eq(failed, &runtime)));
         assert!(!runtime.is_failed());

crates/dbx-core/src/backend_error.rs

@@ -0,0 +1,803 @@
+//! Stable, safe backend error envelopes and the v1 JDBC Agent catalog.
+
+use std::collections::BTreeMap;
+
+use serde::{Deserialize, Serialize};
+
+use crate::db::agent_driver::{
+    category_name, stage_name, valid_agent_error_combination, AgentCallError, AgentErrorCategory, AgentErrorContext,
+    AgentErrorStage, AgentOperationOutcome, AgentSessionDisposition,
+};
+
+const MAX_DETAIL_BYTES: usize = 512;
+
+/// The origin of a backend error as exposed on the wire.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub enum BackendErrorSource {
+    JdbcAgent,
+    JdbcAgentLegacy,
+    LegacyBackend,
+}
+
+/// 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")]
+pub enum BackendOperationOutcome {
+    NotStarted,
+    Unknown,
+}
+
+/// Scalar values permitted in catalog-owned message parameters.
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+#[serde(untagged)]
+pub enum BackendMessageParam {
+    String(String),
+    Integer(i64),
+    Boolean(bool),
+}
+
+/// Allowlisted diagnostics safe for a public error envelope.
+#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct BackendErrorDiagnostics {
+    #[serde(skip_serializing_if = "Option::is_none")]
+    category: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    stage: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    sql_state: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    vendor_code: Option<i32>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    exception_class: Option<String>,
+}
+
+/// Public v1 backend error envelope.
+///
+/// The fields are private on purpose: callers create envelopes through the
+/// catalog adapters, so code/messageKey/params cannot drift independently.
+#[derive(Debug, Clone, PartialEq, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct BackendError {
+    version: u8,
+    code: String,
+    message_key: String,
+    message_params: BTreeMap<String, BackendMessageParam>,
+    source: BackendErrorSource,
+    operation_outcome: BackendOperationOutcome,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    detail: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    diagnostics: Option<BackendErrorDiagnostics>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    help_url: Option<String>,
+}
+
+impl BackendError {
+    /// Convert a typed Agent error into the stable v1 JDBC catalog envelope.
+    pub fn from_agent_call_error(error: &AgentCallError) -> Self {
+        match error {
+            AgentCallError::Structured { message, context, .. } => {
+                let (entry, outcome) = structured_entry(context);
+                Self::new(
+                    entry,
+                    BackendErrorSource::JdbcAgent,
+                    outcome,
+                    stage_param(entry, context.stage),
+                    safe_detail(message),
+                    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::ContractViolation { message, .. } => Self::new(
+                catalog_entry(CatalogCode::ContractInvalid),
+                BackendErrorSource::JdbcAgent,
+                BackendOperationOutcome::Unknown,
+                BTreeMap::new(),
+                safe_detail(message),
+                None,
+            ),
+            AgentCallError::Transport { message } => Self::new(
+                catalog_entry(CatalogCode::ProtocolFailed),
+                BackendErrorSource::JdbcAgent,
+                BackendOperationOutcome::Unknown,
+                BTreeMap::new(),
+                safe_detail(message),
+                None,
+            ),
+            AgentCallError::Timeout { stage, operation_outcome } => {
+                let entry = catalog_entry(match operation_outcome {
+                    AgentOperationOutcome::NotStarted => CatalogCode::TimeoutNotStarted,
+                    AgentOperationOutcome::Unknown => CatalogCode::TimeoutUnknown,
+                });
+                Self::new(
+                    entry,
+                    BackendErrorSource::JdbcAgent,
+                    map_outcome(*operation_outcome),
+                    stage_param(entry, *stage),
+                    None,
+                    Some(diagnostics_for_local("timeout", *stage)),
+                )
+            }
+            AgentCallError::Canceled { stage, operation_outcome } => Self::new(
+                catalog_entry(CatalogCode::Canceled),
+                BackendErrorSource::JdbcAgent,
+                map_outcome(*operation_outcome),
+                stage_param(catalog_entry(CatalogCode::Canceled), *stage),
+                None,
+                Some(diagnostics_for_local("canceled", *stage)),
+            ),
+        }
+    }
+
+    /// Adapt a non-Agent legacy backend string without attempting text classification.
+    pub fn from_legacy_backend(message: &str) -> Self {
+        Self::new(
+            catalog_entry(CatalogCode::LegacyBackend),
+            BackendErrorSource::LegacyBackend,
+            BackendOperationOutcome::Unknown,
+            BTreeMap::new(),
+            safe_detail(message),
+            None,
+        )
+    }
+
+    /// Create a timeout envelope while retaining the bounded diagnostic detail.
+    pub fn from_timeout_detail(message: &str) -> Self {
+        let entry = catalog_entry(CatalogCode::TimeoutUnknown);
+        Self::new(
+            entry,
+            BackendErrorSource::JdbcAgent,
+            BackendOperationOutcome::Unknown,
+            stage_param(entry, AgentErrorStage::Execute),
+            safe_detail(message),
+            Some(diagnostics_for_local("timeout", AgentErrorStage::Execute)),
+        )
+    }
+
+    /// Create a SQL failure envelope while retaining the bounded diagnostic detail.
+    pub fn from_sql_detail(message: &str) -> Self {
+        let entry = catalog_entry(CatalogCode::SqlFailed);
+        Self::new(
+            entry,
+            BackendErrorSource::JdbcAgent,
+            BackendOperationOutcome::Unknown,
+            stage_param(entry, AgentErrorStage::Execute),
+            safe_detail(message),
+            Some(diagnostics_for_local("sql", AgentErrorStage::Execute)),
+        )
+    }
+
+    /// 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,
+            map_outcome(operation_outcome),
+            stage_param(entry, stage),
+            None,
+            Some(diagnostics_for_local("canceled", stage)),
+        )
+    }
+
+    /// Convert a legacy boundary string while preserving Agent data when the
+    /// compatibility adapter can prove that the string came from an Agent.
+    pub fn from_legacy_string(message: &str) -> Self {
+        crate::db::agent_driver::try_agent_error_from_legacy(message)
+            .map_or_else(|| Self::from_legacy_backend(message), |error| Self::from_agent_call_error(&error))
+    }
+
+    pub fn version(&self) -> u8 {
+        self.version
+    }
+
+    pub fn code(&self) -> &str {
+        &self.code
+    }
+
+    pub fn message_key(&self) -> &str {
+        &self.message_key
+    }
+
+    pub fn message_params(&self) -> &BTreeMap<String, BackendMessageParam> {
+        &self.message_params
+    }
+
+    pub fn source(&self) -> BackendErrorSource {
+        self.source
+    }
+
+    pub fn operation_outcome(&self) -> BackendOperationOutcome {
+        self.operation_outcome
+    }
+
+    pub fn detail(&self) -> Option<&str> {
+        self.detail.as_deref()
+    }
+
+    pub fn without_detail(mut self) -> Self {
+        self.detail = None;
+        self
+    }
+
+    pub fn diagnostics(&self) -> Option<&BackendErrorDiagnostics> {
+        self.diagnostics.as_ref()
+    }
+
+    fn new(
+        entry: &'static CatalogEntry,
+        source: BackendErrorSource,
+        operation_outcome: BackendOperationOutcome,
+        message_params: BTreeMap<String, BackendMessageParam>,
+        detail: Option<String>,
+        diagnostics: Option<BackendErrorDiagnostics>,
+    ) -> Self {
+        debug_assert_eq!(message_params.len(), entry.params.len());
+        debug_assert!(message_params.iter().all(|(key, value)| entry.allows_value(key, value)));
+        Self {
+            version: 1,
+            code: entry.code.to_string(),
+            message_key: entry.message_key.to_string(),
+            message_params,
+            source,
+            operation_outcome,
+            detail,
+            diagnostics,
+            help_url: entry.help_url.map(str::to_string),
+        }
+    }
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+enum CatalogCode {
+    ConnectionFailed,
+    ConnectionInterrupted,
+    TimeoutNotStarted,
+    TimeoutUnknown,
+    Canceled,
+    BusyRetryLater,
+    RuntimeReplaced,
+    SqlFailed,
+    ProtocolFailed,
+    ContractInvalid,
+    JdbcLegacyFailure,
+    LegacyBackend,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+enum ParamKind {
+    String,
+    Integer,
+    Boolean,
+}
+
+#[derive(Debug, Clone, Copy)]
+struct ParamSpec {
+    name: &'static str,
+    kind: ParamKind,
+}
+
+#[derive(Debug, Clone, Copy)]
+struct CatalogEntry {
+    code: &'static str,
+    message_key: &'static str,
+    params: &'static [ParamSpec],
+    help_url: Option<&'static str>,
+}
+
+const STAGE_PARAM: &[ParamSpec] = &[ParamSpec { name: "stage", kind: ParamKind::String }];
+const NO_PARAMS: &[ParamSpec] = &[];
+
+fn catalog_entry(code: CatalogCode) -> &'static CatalogEntry {
+    const ENTRIES: &[(CatalogCode, CatalogEntry)] = &[
+        (
+            CatalogCode::ConnectionFailed,
+            CatalogEntry {
+                code: "DBX-JDBC-1001",
+                message_key: "backendErrors.jdbc.connectionFailed",
+                params: STAGE_PARAM,
+                help_url: None,
+            },
+        ),
+        (
+            CatalogCode::ConnectionInterrupted,
+            CatalogEntry {
+                code: "DBX-JDBC-1002",
+                message_key: "backendErrors.jdbc.connectionInterrupted",
+                params: STAGE_PARAM,
+                help_url: None,
+            },
+        ),
+        (
+            CatalogCode::TimeoutNotStarted,
+            CatalogEntry {
+                code: "DBX-JDBC-2001",
+                message_key: "backendErrors.jdbc.operationTimedOut",
+                params: STAGE_PARAM,
+                help_url: None,
+            },
+        ),
+        (
+            CatalogCode::TimeoutUnknown,
+            CatalogEntry {
+                code: "DBX-JDBC-2002",
+                message_key: "backendErrors.jdbc.operationTimedOut",
+                params: STAGE_PARAM,
+                help_url: None,
+            },
+        ),
+        (
+            CatalogCode::Canceled,
+            CatalogEntry {
+                code: "DBX-JDBC-2003",
+                message_key: "backendErrors.jdbc.operationCanceled",
+                params: STAGE_PARAM,
+                help_url: None,
+            },
+        ),
+        (
+            CatalogCode::BusyRetryLater,
+            CatalogEntry {
+                code: "DBX-JDBC-3001",
+                message_key: "backendErrors.jdbc.busyRetryLater",
+                params: STAGE_PARAM,
+                help_url: None,
+            },
+        ),
+        (
+            CatalogCode::RuntimeReplaced,
+            CatalogEntry {
+                code: "DBX-JDBC-3002",
+                message_key: "backendErrors.jdbc.runtimeReplaced",
+                params: STAGE_PARAM,
+                help_url: None,
+            },
+        ),
+        (
+            CatalogCode::SqlFailed,
+            CatalogEntry {
+                code: "DBX-JDBC-4001",
+                message_key: "backendErrors.jdbc.sqlFailed",
+                params: STAGE_PARAM,
+                help_url: None,
+            },
+        ),
+        (
+            CatalogCode::ProtocolFailed,
+            CatalogEntry {
+                code: "DBX-JDBC-5001",
+                message_key: "backendErrors.jdbc.protocolFailed",
+                params: NO_PARAMS,
+                help_url: None,
+            },
+        ),
+        (
+            CatalogCode::ContractInvalid,
+            CatalogEntry {
+                code: "DBX-JDBC-5002",
+                message_key: "backendErrors.jdbc.contractInvalid",
+                params: NO_PARAMS,
+                help_url: None,
+            },
+        ),
+        (
+            CatalogCode::JdbcLegacyFailure,
+            CatalogEntry {
+                code: "DBX-JDBC-9001",
+                message_key: "backendErrors.jdbc.legacyFailure",
+                params: NO_PARAMS,
+                help_url: None,
+            },
+        ),
+        (
+            CatalogCode::LegacyBackend,
+            CatalogEntry {
+                code: "DBX-LEGACY-0001",
+                message_key: "backendErrors.legacy",
+                params: NO_PARAMS,
+                help_url: None,
+            },
+        ),
+    ];
+    ENTRIES.iter().find(|(kind, _)| *kind == code).map(|(_, entry)| entry).expect("catalog entry missing")
+}
+
+fn structured_entry(context: &AgentErrorContext) -> (&'static CatalogEntry, BackendOperationOutcome) {
+    if !valid_agent_error_combination(context) {
+        return (catalog_entry(CatalogCode::ContractInvalid), BackendOperationOutcome::Unknown);
+    }
+
+    let code = match context.category {
+        AgentErrorCategory::Connection => match (context.stage, context.operation_outcome) {
+            (
+                AgentErrorStage::Request
+                | AgentErrorStage::Checkout
+                | AgentErrorStage::Connect
+                | AgentErrorStage::Validate,
+                AgentOperationOutcome::NotStarted,
+            ) => CatalogCode::ConnectionFailed,
+            (
+                AgentErrorStage::Execute | AgentErrorStage::Fetch | AgentErrorStage::Cancel | AgentErrorStage::Close,
+                AgentOperationOutcome::Unknown,
+            ) => CatalogCode::ConnectionInterrupted,
+            _ => CatalogCode::ContractInvalid,
+        },
+        AgentErrorCategory::Timeout => match context.operation_outcome {
+            AgentOperationOutcome::NotStarted => CatalogCode::TimeoutNotStarted,
+            AgentOperationOutcome::Unknown => CatalogCode::TimeoutUnknown,
+        },
+        AgentErrorCategory::Canceled => CatalogCode::Canceled,
+        AgentErrorCategory::Resource => {
+            if context.session_disposition == AgentSessionDisposition::ReplaceRuntime {
+                CatalogCode::RuntimeReplaced
+            } else if context.operation_outcome == AgentOperationOutcome::NotStarted {
+                CatalogCode::BusyRetryLater
+            } else {
+                CatalogCode::ContractInvalid
+            }
+        }
+        AgentErrorCategory::Sql => {
+            if matches!(
+                context.stage,
+                AgentErrorStage::Execute | AgentErrorStage::Fetch | AgentErrorStage::Cancel | AgentErrorStage::Close
+            ) && context.operation_outcome == AgentOperationOutcome::Unknown
+            {
+                CatalogCode::SqlFailed
+            } else {
+                CatalogCode::ContractInvalid
+            }
+        }
+        AgentErrorCategory::Protocol => CatalogCode::ProtocolFailed,
+    };
+    let entry = catalog_entry(code);
+    let outcome = if matches!(code, CatalogCode::ContractInvalid) {
+        BackendOperationOutcome::Unknown
+    } else {
+        map_outcome(context.operation_outcome)
+    };
+    (entry, outcome)
+}
+
+fn map_outcome(outcome: AgentOperationOutcome) -> BackendOperationOutcome {
+    match outcome {
+        AgentOperationOutcome::NotStarted => BackendOperationOutcome::NotStarted,
+        AgentOperationOutcome::Unknown => BackendOperationOutcome::Unknown,
+    }
+}
+
+fn stage_param(entry: &'static CatalogEntry, stage: AgentErrorStage) -> BTreeMap<String, BackendMessageParam> {
+    let mut params = BTreeMap::new();
+    let value = BackendMessageParam::String(stage_name(stage).to_string());
+    if entry.allows_value("stage", &value) {
+        params.insert("stage".to_string(), value);
+    }
+    params
+}
+
+impl CatalogEntry {
+    fn allows_value(self, name: &str, value: &BackendMessageParam) -> bool {
+        self.params.iter().any(|param| param.name == name && param.kind == value.kind())
+    }
+}
+
+impl BackendMessageParam {
+    fn kind(&self) -> ParamKind {
+        match self {
+            Self::String(_) => ParamKind::String,
+            Self::Integer(_) => ParamKind::Integer,
+            Self::Boolean(_) => ParamKind::Boolean,
+        }
+    }
+}
+
+fn diagnostics_from_context(context: &AgentErrorContext) -> BackendErrorDiagnostics {
+    BackendErrorDiagnostics {
+        category: Some(category_name(context.category).to_string()),
+        stage: Some(stage_name(context.stage).to_string()),
+        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)),
+    }
+}
+
+fn diagnostics_for_local(category: &str, stage: AgentErrorStage) -> BackendErrorDiagnostics {
+    BackendErrorDiagnostics {
+        category: Some(category.to_string()),
+        stage: Some(stage_name(stage).to_string()),
+        ..Default::default()
+    }
+}
+
+fn bounded_ascii(value: &str, max: usize) -> String {
+    value.chars().filter(|ch| ch.is_ascii_graphic() || *ch == ' ').take(max).collect()
+}
+
+fn bounded_text(value: &str, max_bytes: usize) -> String {
+    let end = value
+        .char_indices()
+        .map(|(index, ch)| (index, index + ch.len_utf8()))
+        .take_while(|(_, next)| *next <= max_bytes)
+        .map(|(_, next)| next)
+        .last()
+        .unwrap_or(0);
+    value[..end].to_string()
+}
+
+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)) {
+        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)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::db::agent_driver::{AgentErrorStage, AgentSessionDisposition};
+    use std::collections::BTreeSet;
+
+    fn context(
+        category: AgentErrorCategory,
+        stage: AgentErrorStage,
+        outcome: AgentOperationOutcome,
+    ) -> AgentErrorContext {
+        AgentErrorContext {
+            contract_version: 1,
+            category,
+            retryable: true,
+            session_disposition: if matches!(category, AgentErrorCategory::Timeout | AgentErrorCategory::Canceled) {
+                AgentSessionDisposition::Quarantine
+            } else {
+                AgentSessionDisposition::Keep
+            },
+            stage,
+            operation_outcome: outcome,
+            agent_session_id: Some("private-session".to_string()),
+            sql_state: Some("08006".to_string()),
+            vendor_code: Some(-7),
+            exception_class: Some("java.sql.SQLException".to_string()),
+        }
+    }
+
+    #[test]
+    fn catalog_owns_exact_code_and_message_key_pairs() {
+        let cases = [
+            (CatalogCode::ConnectionFailed, "DBX-JDBC-1001", "backendErrors.jdbc.connectionFailed"),
+            (CatalogCode::ConnectionInterrupted, "DBX-JDBC-1002", "backendErrors.jdbc.connectionInterrupted"),
+            (CatalogCode::TimeoutNotStarted, "DBX-JDBC-2001", "backendErrors.jdbc.operationTimedOut"),
+            (CatalogCode::TimeoutUnknown, "DBX-JDBC-2002", "backendErrors.jdbc.operationTimedOut"),
+            (CatalogCode::Canceled, "DBX-JDBC-2003", "backendErrors.jdbc.operationCanceled"),
+            (CatalogCode::BusyRetryLater, "DBX-JDBC-3001", "backendErrors.jdbc.busyRetryLater"),
+            (CatalogCode::RuntimeReplaced, "DBX-JDBC-3002", "backendErrors.jdbc.runtimeReplaced"),
+            (CatalogCode::SqlFailed, "DBX-JDBC-4001", "backendErrors.jdbc.sqlFailed"),
+            (CatalogCode::ProtocolFailed, "DBX-JDBC-5001", "backendErrors.jdbc.protocolFailed"),
+            (CatalogCode::ContractInvalid, "DBX-JDBC-5002", "backendErrors.jdbc.contractInvalid"),
+            (CatalogCode::JdbcLegacyFailure, "DBX-JDBC-9001", "backendErrors.jdbc.legacyFailure"),
+            (CatalogCode::LegacyBackend, "DBX-LEGACY-0001", "backendErrors.legacy"),
+        ];
+        let mut codes = BTreeSet::new();
+        for (kind, expected_code, expected_key) in cases {
+            let entry = catalog_entry(kind);
+            assert_eq!(entry.code, expected_code);
+            assert_eq!(entry.message_key, expected_key);
+            assert!(codes.insert(entry.code), "duplicate catalog code: {}", entry.code);
+        }
+    }
+
+    #[test]
+    fn catalog_maps_valid_variants_and_emits_allowlisted_params() {
+        let cases = [
+            (
+                AgentErrorCategory::Connection,
+                AgentErrorStage::Request,
+                AgentOperationOutcome::NotStarted,
+                "DBX-JDBC-1001",
+            ),
+            (AgentErrorCategory::Connection, AgentErrorStage::Execute, AgentOperationOutcome::Unknown, "DBX-JDBC-1002"),
+            (AgentErrorCategory::Timeout, AgentErrorStage::Connect, AgentOperationOutcome::NotStarted, "DBX-JDBC-2001"),
+            (AgentErrorCategory::Timeout, AgentErrorStage::Execute, AgentOperationOutcome::Unknown, "DBX-JDBC-2002"),
+            (AgentErrorCategory::Canceled, AgentErrorStage::Cancel, AgentOperationOutcome::Unknown, "DBX-JDBC-2003"),
+            (
+                AgentErrorCategory::Resource,
+                AgentErrorStage::Request,
+                AgentOperationOutcome::NotStarted,
+                "DBX-JDBC-3001",
+            ),
+            (AgentErrorCategory::Resource, AgentErrorStage::Execute, AgentOperationOutcome::Unknown, "DBX-JDBC-5002"),
+            (AgentErrorCategory::Sql, AgentErrorStage::Execute, AgentOperationOutcome::Unknown, "DBX-JDBC-4001"),
+            (AgentErrorCategory::Protocol, AgentErrorStage::Request, AgentOperationOutcome::Unknown, "DBX-JDBC-5002"),
+        ];
+        for (category, stage, outcome, expected_code) in cases {
+            let error = AgentCallError::Structured {
+                rpc_code: -1,
+                message: "safe failure".to_string(),
+                context: context(category, stage, outcome),
+            };
+            let envelope = BackendError::from_agent_call_error(&error);
+            assert_eq!(envelope.code(), expected_code);
+            assert_eq!(envelope.version(), 1);
+            if expected_code != "DBX-JDBC-5002" {
+                assert_eq!(
+                    envelope.message_params().get("stage"),
+                    Some(&BackendMessageParam::String(stage_name(stage).to_string()))
+                );
+            }
+        }
+    }
+
+    #[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",
+        ] {
+            assert!(safe_detail(message).is_none(), "detail leaked: {message}");
+        }
+        let error = BackendError::from_agent_call_error(&AgentCallError::Structured {
+            rpc_code: -1,
+            message: "jdbc:postgresql://host/db?password=secret".to_string(),
+            context: context(
+                AgentErrorCategory::Connection,
+                AgentErrorStage::Request,
+                AgentOperationOutcome::NotStarted,
+            ),
+        });
+        let value = serde_json::to_value(error).unwrap();
+        assert_eq!(value["source"], "jdbcAgent");
+        assert!(value.get("retryable").is_none());
+        assert!(value.get("sessionDisposition").is_none());
+        assert!(value.get("agentSessionId").is_none());
+        assert!(value.get("detail").is_none());
+    }
+
+    #[test]
+    fn invalid_structured_combinations_use_contract_catalog_code() {
+        let invalid = [
+            (
+                AgentErrorCategory::Timeout,
+                AgentErrorStage::Request,
+                AgentOperationOutcome::Unknown,
+                AgentSessionDisposition::Keep,
+            ),
+            (
+                AgentErrorCategory::Protocol,
+                AgentErrorStage::Request,
+                AgentOperationOutcome::NotStarted,
+                AgentSessionDisposition::ReplaceRuntime,
+            ),
+            (
+                AgentErrorCategory::Resource,
+                AgentErrorStage::Execute,
+                AgentOperationOutcome::Unknown,
+                AgentSessionDisposition::Keep,
+            ),
+            (
+                AgentErrorCategory::Sql,
+                AgentErrorStage::Connect,
+                AgentOperationOutcome::NotStarted,
+                AgentSessionDisposition::Keep,
+            ),
+        ];
+        for (category, stage, outcome, disposition) in invalid {
+            let mut ctx = context(category, stage, outcome);
+            ctx.session_disposition = disposition;
+            let envelope = BackendError::from_agent_call_error(&AgentCallError::Structured {
+                rpc_code: -1,
+                message: "invalid combination".to_string(),
+                context: ctx,
+            });
+            assert_eq!(envelope.code(), "DBX-JDBC-5002");
+            assert_eq!(envelope.operation_outcome(), BackendOperationOutcome::Unknown);
+        }
+    }
+
+    #[test]
+    fn legacy_adapters_use_stable_sources() {
+        let legacy = BackendError::from_agent_call_error(&AgentCallError::Legacy {
+            rpc_code: None,
+            message: "old agent".to_string(),
+            hints: Default::default(),
+        });
+        assert_eq!(legacy.code(), "DBX-JDBC-9001");
+        assert_eq!(legacy.source, BackendErrorSource::JdbcAgentLegacy);
+        assert_eq!(BackendError::from_legacy_backend("driver failed").code(), "DBX-LEGACY-0001");
+    }
+
+    #[test]
+    fn legacy_agent_envelope_keeps_multiline_database_detail() {
+        let legacy = AgentCallError::Legacy {
+            rpc_code: Some(-1),
+            message: "ERROR: relation \"dbx_table_that_does_not_exist\" does not exist\n  Position: 15".to_string(),
+            hints: Default::default(),
+        }
+        .into_legacy_string();
+        let error = BackendError::from_legacy_string(&legacy);
+
+        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")
+        );
+    }
+
+    #[test]
+    fn legacy_agent_envelope_keeps_dm_chinese_database_detail() {
+        let legacy = AgentCallError::Legacy {
+            rpc_code: Some(-1),
+            message: "无效的表或视图名\n错误码: -2106".to_string(),
+            hints: Default::default(),
+        }
+        .into_legacy_string();
+        let error = BackendError::from_legacy_string(&legacy);
+
+        assert_eq!(error.code(), "DBX-JDBC-9001");
+        assert_eq!(error.detail(), Some("无效的表或视图名 错误码: -2106"));
+    }
+
+    #[test]
+    fn strict_marker_round_trip_keeps_catalog_code() {
+        let error = AgentCallError::Structured {
+            rpc_code: -1,
+            message: "connection lost".to_string(),
+            context: context(AgentErrorCategory::Connection, AgentErrorStage::Execute, AgentOperationOutcome::Unknown),
+        };
+        let legacy =
+            crate::db::agent_driver::append_legacy_error_context(&error.into_legacy_string(), "SQL text omitted");
+
+        assert_eq!(BackendError::from_legacy_string(&legacy).code(), "DBX-JDBC-1002");
+    }
+}

crates/dbx-core/src/connection.rs

@@ -15,9 +15,10 @@ use crate::agent_connection::{
     trino_like_jdbc_connection_string, AgentSessionRole,
 };
 use crate::agent_manager::{JavaRuntimeMode, DEFAULT_JRE_KEY};
+use crate::agent_recovery::{RecoveryDecision, RecoveryPolicy, RecoveryScope};
 use crate::database_capabilities;
 use crate::db;
-use crate::db::agent_driver::AgentMethod;
+use crate::db::agent_driver::{AgentCallError, AgentMethod};
 use crate::db::http_tunnel::HttpTunnelManager;
 use crate::db::proxy_tunnel::ProxyTunnelManager;
 use crate::db::ssh_tunnel::TunnelManager;
@@ -550,8 +551,7 @@ fn fail_stop_removed_agent_pool(pool_key: &str, pool: &PoolKind) {
 }
 
 fn should_replace_agent_runtime(error: &str) -> bool {
-    crate::db::agent_driver::agent_session_disposition(error)
-        == Some(crate::db::agent_driver::AgentSessionDisposition::ReplaceRuntime)
+    crate::db::agent_driver::agent_recovery_decision(error, RecoveryScope::ConnectionOpen).replaces_runtime()
 }
 
 pub fn metadata_connection_config(config: &ConnectionConfig) -> ConnectionConfig {
@@ -681,10 +681,10 @@ pub fn sqlserver_uses_legacy_driver(config: &ConnectionConfig) -> bool {
 }
 
 pub fn sqlserver_legacy_driver_error(agent_error: &str) -> String {
-    // AgentManager currently returns launch failures as strings. This exact marker is generated
-    // internally and only adds guidance for an explicitly selected compatibility driver.
+    // This mapper handles both AgentManager launch strings and Agent call errors, so context
+    // must remain before any structured-error compatibility marker.
     if agent_error.contains("driver is not installed") {
-        format!("{agent_error}\n\n{SQLSERVER_LEGACY_DRIVER_INSTALL_HINT}")
+        crate::db::agent_driver::append_legacy_error_context(agent_error, SQLSERVER_LEGACY_DRIVER_INSTALL_HINT)
     } else {
         agent_error.to_string()
     }
@@ -1462,12 +1462,14 @@ impl AppState {
                         Ok(Ok(())) => {}
                         Ok(Err(err)) => {
                             log::warn!("Connection keepalive failed for '{key}': {err}; invalidating pool");
+                            let replace_runtime =
+                                err.recovery_decision().is_some_and(RecoveryDecision::replaces_runtime);
                             if !detach_keepalive_target_if_current(
                                 &routing,
                                 &connections,
                                 &key,
                                 target,
-                                should_replace_agent_runtime(&err),
+                                replace_runtime,
                             )
                             .await
                             {
@@ -2904,15 +2906,15 @@ impl AppState {
                         return false;
                     };
                     let timeout = crate::db::connection_timeout();
-                    match agent.validate_connection(Some(timeout)).await {
+                    match agent.validate_connection_typed(Some(timeout)).await {
                         Ok(_) => false,
-                        Err(err) if is_agent_validate_connection_unsupported(&err) => {
+                        Err(err) if is_agent_validate_connection_unsupported(&err.to_string()) => {
                             log::debug!(
                                 "Agent connection pool '{pool_key}' does not support validate_connection; keeping pool"
                             );
                             false
                         }
-                        Err(err) if should_replace_agent_runtime(&err) => {
+                        Err(err) if RecoveryPolicy::decide(&err, RecoveryScope::Keepalive).replaces_runtime() => {
                             log::warn!(
                                 "Agent connection pool '{pool_key}' requested runtime replacement during health probe: {err}"
                             );
@@ -3143,12 +3145,12 @@ impl AppState {
         self.detach_pool_by_key(&pool_key, true).await
     }
 
-    pub(crate) async fn detach_metadata_pool_after_error(
+    pub(crate) async fn detach_metadata_pool_after_recovery(
         &self,
         connection_id: &str,
         database: Option<&str>,
         client_session_id: Option<&str>,
-        error: &str,
+        agent_session_id: Option<&str>,
         replace_agent_runtime: bool,
     ) -> bool {
         let config = {
@@ -3159,11 +3161,11 @@ impl AppState {
         let base_pool_key = base_pool_key_for(db_type, connection_id, database, false);
         let pool_key =
             pool_key_for_session_role(config.as_ref(), base_pool_key, client_session_id, AgentSessionRole::Metadata);
-        if let Some(session_id) = crate::db::agent_driver::agent_rpc_error_session_id(error) {
+        if let Some(session_id) = agent_session_id {
             let expected_client = {
                 let connections = self.connections.read().await;
                 match connections.get(&pool_key) {
-                    Some(PoolKind::Agent(client)) if client.matches_session_id(&session_id) => Some(client.clone()),
+                    Some(PoolKind::Agent(client)) if client.matches_session_id(session_id) => Some(client.clone()),
                     Some(PoolKind::Agent(_)) => return false,
                     _ => None,
                 }
@@ -3786,9 +3788,9 @@ impl AppState {
                         log::debug!("Agent connection pool '{key}' is busy; skipping resume health probe");
                         continue;
                     };
-                    match agent.validate_connection(Some(timeout)).await {
+                    match agent.validate_connection_typed(Some(timeout)).await {
                         Ok(_) => true,
-                        Err(err) if is_agent_validate_connection_unsupported(&err) => {
+                        Err(err) if is_agent_validate_connection_unsupported(&err.to_string()) => {
                             log::debug!(
                                 "Agent connection pool '{key}' does not support validate_connection; keeping pool"
                             );
@@ -3796,7 +3798,11 @@ impl AppState {
                         }
                         Err(e) => {
                             log::warn!("Agent connection pool '{key}' is unhealthy: {e}");
-                            failed_agent_checks.push((key.clone(), client.clone(), should_replace_agent_runtime(&e)));
+                            failed_agent_checks.push((
+                                key.clone(),
+                                client.clone(),
+                                RecoveryPolicy::decide(&e, RecoveryScope::Keepalive).replaces_runtime(),
+                            ));
                             false
                         }
                     }
@@ -4030,6 +4036,36 @@ enum KeepaliveTarget {
     Agent(Arc<db::agent_driver::PooledAgentClient>),
 }
 
+#[derive(Debug)]
+enum KeepaliveError {
+    Agent(AgentCallError),
+    Legacy(String),
+}
+
+impl KeepaliveError {
+    fn recovery_decision(&self) -> Option<RecoveryDecision> {
+        match self {
+            Self::Agent(error) => Some(RecoveryPolicy::decide(error, RecoveryScope::Keepalive)),
+            Self::Legacy(_) => None,
+        }
+    }
+}
+
+impl std::fmt::Display for KeepaliveError {
+    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            Self::Agent(error) => error.fmt(formatter),
+            Self::Legacy(error) => formatter.write_str(error),
+        }
+    }
+}
+
+impl From<String> for KeepaliveError {
+    fn from(error: String) -> Self {
+        Self::Legacy(error)
+    }
+}
+
 impl KeepaliveTarget {
     fn matches_pool(&self, pool: &PoolKind) -> bool {
         match (self, pool) {
@@ -4095,41 +4131,55 @@ fn keepalive_target_from_pool(pool: &PoolKind, config: &ConnectionConfig) -> Opt
     }
 }
 
-async fn ping_keepalive_target(target: &mut KeepaliveTarget, timeout: Duration) -> Result<(), String> {
+async fn ping_keepalive_target(target: &mut KeepaliveTarget, timeout: Duration) -> Result<(), KeepaliveError> {
     match target {
         KeepaliveTarget::Mysql(pool) => {
             let mut conn = db::mysql::get_conn_with_health_check(pool).await?;
-            conn.ping().await.map_err(|e| e.to_string())
+            conn.ping().await.map_err(|error| KeepaliveError::Legacy(error.to_string()))
         }
         KeepaliveTarget::Postgres(pool) => {
             let client = pool.get().await.map_err(|e| format!("PostgreSQL pool error: {e}"))?;
-            client.simple_query("SELECT 1").await.map(|_| ()).map_err(|e| e.to_string())
+            client.simple_query("SELECT 1").await.map(|_| ()).map_err(|error| KeepaliveError::Legacy(error.to_string()))
         }
-        KeepaliveTarget::Rqlite(client) => db::rqlite_driver::test_connection(client, timeout).await,
-        KeepaliveTarget::Turso(client) => db::turso_driver::test_connection(client, timeout).await,
+        KeepaliveTarget::Rqlite(client) => {
+            db::rqlite_driver::test_connection(client, timeout).await.map_err(Into::into)
+        }
+        KeepaliveTarget::Turso(client) => db::turso_driver::test_connection(client, timeout).await.map_err(Into::into),
         KeepaliveTarget::MongoDb { client, database } => {
-            db::mongo_driver::test_connection(client, timeout, database.as_deref()).await
+            db::mongo_driver::test_connection(client, timeout, database.as_deref()).await.map_err(Into::into)
+        }
+        KeepaliveTarget::ClickHouse(client) => {
+            db::clickhouse_driver::test_connection(client, timeout).await.map_err(Into::into)
         }
-        KeepaliveTarget::ClickHouse(client) => db::clickhouse_driver::test_connection(client, timeout).await,
         KeepaliveTarget::SqlServer(client) => {
             let Ok(mut client) = client.try_lock() else {
                 return Ok(());
             };
-            db::sqlserver::test_connection(&mut client).await
+            db::sqlserver::test_connection(&mut client).await.map_err(Into::into)
+        }
+        KeepaliveTarget::Elasticsearch(client) => {
+            db::elasticsearch_driver::test_connection(client, timeout).await.map_err(Into::into)
+        }
+        KeepaliveTarget::Easysearch(client) => {
+            db::easysearch_driver::test_connection(client, timeout).await.map_err(Into::into)
+        }
+        KeepaliveTarget::HBase(client) => {
+            db::hbase_driver::test_connection(client, timeout).await.map(|_| ()).map_err(Into::into)
+        }
+        KeepaliveTarget::VectorDb(client) => {
+            db::vector_driver::test_connection(client, timeout).await.map_err(Into::into)
+        }
+        KeepaliveTarget::InfluxDb(client) => {
+            db::influxdb_driver::test_connection(client, timeout).await.map_err(Into::into)
         }
-        KeepaliveTarget::Elasticsearch(client) => db::elasticsearch_driver::test_connection(client, timeout).await,
-        KeepaliveTarget::Easysearch(client) => db::easysearch_driver::test_connection(client, timeout).await,
-        KeepaliveTarget::HBase(client) => db::hbase_driver::test_connection(client, timeout).await.map(|_| ()),
-        KeepaliveTarget::VectorDb(client) => db::vector_driver::test_connection(client, timeout).await,
-        KeepaliveTarget::InfluxDb(client) => db::influxdb_driver::test_connection(client, timeout).await,
         KeepaliveTarget::Agent(client) => {
             let Ok(mut client) = client.try_lock() else {
                 return Ok(());
             };
-            match client.validate_connection(Some(timeout)).await {
+            match client.validate_connection_typed(Some(timeout)).await {
                 Ok(_) => Ok(()),
-                Err(err) if is_agent_validate_connection_unsupported(&err) => Ok(()),
-                Err(err) => Err(err),
+                Err(error) if is_agent_validate_connection_unsupported(&error.to_string()) => Ok(()),
+                Err(error) => Err(KeepaliveError::Agent(error)),
             }
         }
     }
@@ -4949,6 +4999,34 @@ mod tests {
         assert!(message.contains("enable SQL Server legacy compatibility mode again"));
     }
 
+    #[test]
+    fn sqlserver_legacy_driver_hint_preserves_structured_agent_error() {
+        let error = crate::db::agent_driver::AgentCallError::Structured {
+            rpc_code: -6007,
+            message: "driver is not installed".to_string(),
+            context: crate::db::agent_driver::AgentErrorContext {
+                contract_version: 1,
+                category: crate::db::agent_driver::AgentErrorCategory::Connection,
+                retryable: false,
+                session_disposition: crate::db::agent_driver::AgentSessionDisposition::Quarantine,
+                stage: crate::db::agent_driver::AgentErrorStage::Connect,
+                operation_outcome: crate::db::agent_driver::AgentOperationOutcome::NotStarted,
+                agent_session_id: None,
+                sql_state: None,
+                vendor_code: None,
+                exception_class: None,
+            },
+        }
+        .into_legacy_string();
+
+        let message = sqlserver_legacy_driver_error(&error);
+
+        assert!(matches!(
+            crate::db::agent_driver::agent_error_from_legacy(&message, None),
+            crate::db::agent_driver::AgentCallError::Structured { .. }
+        ));
+    }
+
     #[test]
     fn agent_connect_params_include_url_params() {
         let mut config = mysql_config(Some("testdb"));
@@ -6788,13 +6866,12 @@ for line in sys.stdin:
         state.configs.write().await.insert(config.id.clone(), config);
         let pool_key = "conn:analytics:role:metadata";
         state.connections.write().await.insert(pool_key.to_string(), PoolKind::Agent(current_client.clone()));
-        let stale_error = concat!(
-            "Agent RPC error (-1): stale metadata failure\nDBX_AGENT_ERROR_DATA:",
-            r#"{"category":"resource","sessionDisposition":"replace_runtime","agentSessionId":"stale-session"}"#
+        assert!(
+            !state
+                .detach_metadata_pool_after_recovery("conn", Some("analytics"), None, Some("stale-session"), true,)
+                .await
         );
 
-        assert!(!state.detach_metadata_pool_after_error("conn", Some("analytics"), None, stale_error, true).await);
-
         let connections = state.connections.read().await;
         let PoolKind::Agent(routed_client) = connections.get(pool_key).expect("new metadata generation must remain")
         else {
@@ -6816,10 +6893,7 @@ for line in sys.stdin:
 
         let error = super::ping_keepalive_target(&mut target, Duration::from_secs(1)).await.unwrap_err();
 
-        assert_eq!(
-            crate::db::agent_driver::agent_session_disposition(&error),
-            Some(crate::db::agent_driver::AgentSessionDisposition::ReplaceRuntime)
-        );
+        assert!(matches!(error.recovery_decision(), Some(crate::agent_recovery::RecoveryDecision::ReplaceRuntime)));
         assert!(!runtime.is_failed());
         runtime.kill();
         let _ = std::fs::remove_dir_all(dir);

crates/dbx-core/src/lib.rs

@@ -5,6 +5,7 @@ pub mod agent_explain;
 pub mod agent_kv;
 pub mod agent_loop;
 pub mod agent_manager;
+pub mod agent_recovery;
 pub mod agent_runtime;
 pub mod agent_service;
 pub mod agent_tools;
@@ -15,6 +16,7 @@ pub mod ai_codex_cli;
 pub mod ai_effort;
 mod ai_model_filter;
 pub mod ai_pi_agent_cli;
+pub mod backend_error;
 pub mod changelog;
 pub mod cloud_sync;
 pub mod config;

crates/dbx-core/src/plugins.rs

@@ -337,7 +337,7 @@ impl PluginDriverSession {
         process.child.id()
     }
 
-    #[cfg(test)]
+    #[cfg(all(test, unix))]
     pub(crate) async fn start_for_test(
         plugin: InstalledPlugin,
         driver_id: String,
@@ -513,10 +513,9 @@ fn resolve_plugin_executable(plugin_dir: &Path, executable: &str) -> PathBuf {
 
 #[cfg(test)]
 mod tests {
-    use super::{
-        read_decoded_plugin_response, read_plugin_line, InstalledPlugin, PluginDriverManifest, PluginDriverSession,
-        PluginManifest, PluginRuntimeEnv,
-    };
+    use super::{read_decoded_plugin_response, read_plugin_line, InstalledPlugin, PluginManifest};
+    #[cfg(unix)]
+    use super::{PluginDriverManifest, PluginDriverSession, PluginRuntimeEnv};
     use std::path::PathBuf;
     use tokio::io::BufReader;
 

crates/dbx-core/src/schema.rs

@@ -1,3 +1,4 @@
+use crate::agent_recovery::{RecoveryDecision, RecoveryPolicy, RecoveryScope};
 use crate::connection::{
     connection_url_for_endpoint, database_connection_config, gaussdb_uses_m_jdbc_driver, task_client_session_id,
     AppState, MysqlMode, PoolKind,
@@ -793,8 +794,9 @@ async fn list_schemas_once(
                                 .await
                                 .map(|schemas| filter_visible_schema_names(schemas, visible_schema_filter.as_deref()))
                                 .map_err(|fallback_error| {
-                                    format!(
-                                        "{agent_error}\n\nNative PostgreSQL metadata fallback failed: {fallback_error}"
+                                    crate::db::agent_driver::append_legacy_error_context(
+                                        &agent_error,
+                                        &format!("Native PostgreSQL metadata fallback failed: {fallback_error}"),
                                     )
                                 });
                         }
@@ -2043,7 +2045,10 @@ async fn list_tables_once(
                                 db::postgres::list_tables_filtered(&pool, schema, filter, limit, offset).await
                             };
                             return result.map_err(|fallback_error| {
-                                format!("{agent_error}\n\nNative PostgreSQL metadata fallback failed: {fallback_error}")
+                                crate::db::agent_driver::append_legacy_error_context(
+                                    &agent_error,
+                                    &format!("Native PostgreSQL metadata fallback failed: {fallback_error}"),
+                                )
                             });
                         }
                     }
@@ -3052,7 +3057,10 @@ for line in sys.stdin:
 
         let error = super::get_table_ddl_core(&state, "conn", "analytics", "APP", "EVENTS", None).await.unwrap_err();
 
-        assert_eq!(crate::db::agent_driver::agent_rpc_error_category(&error).as_deref(), Some("timeout"));
+        assert_eq!(
+            crate::db::agent_driver::try_agent_error_from_legacy(&error).and_then(|error| error.category()),
+            Some(crate::db::agent_driver::AgentErrorCategory::Timeout)
+        );
         assert_eq!(std::fs::read_to_string(call_count_path).unwrap(), "1");
         assert!(!state.connections.read().await.contains_key(pool_key));
         runtime.kill();
@@ -4686,8 +4694,9 @@ async fn list_objects_once(
                         {
                             return db::postgres::list_objects(&pool, schema).await.map(unpaged_object_list).map_err(
                                 |fallback_error| {
-                                    format!(
-                                        "{agent_error}\n\nNative PostgreSQL metadata fallback failed: {fallback_error}"
+                                    crate::db::agent_driver::append_legacy_error_context(
+                                        &agent_error,
+                                        &format!("Native PostgreSQL metadata fallback failed: {fallback_error}"),
                                     )
                                 },
                             );
@@ -4819,8 +4828,9 @@ async fn list_completion_objects_once(
                                 .await
                                 .map(filter_completion_objects)
                                 .map_err(|fallback_error| {
-                                    format!(
-                                        "{agent_error}\n\nNative PostgreSQL metadata fallback failed: {fallback_error}"
+                                    crate::db::agent_driver::append_legacy_error_context(
+                                        &agent_error,
+                                        &format!("Native PostgreSQL metadata fallback failed: {fallback_error}"),
                                     )
                                 });
                         }
@@ -4924,19 +4934,16 @@ where
     let mut retried = false;
     loop {
         let result = operation().await;
-        let action = result
-            .as_ref()
-            .err()
-            .map(|error| metadata_error_action(db_type, error, retried))
-            .unwrap_or(MetadataErrorAction::Return);
-        match action {
+        let recovery =
+            result.as_ref().err().map(|error| metadata_recovery(db_type, error, retried)).unwrap_or_default();
+        match recovery.action {
             MetadataErrorAction::ReplaceRuntime => {
                 state
-                    .detach_metadata_pool_after_error(
+                    .detach_metadata_pool_after_recovery(
                         connection_id,
                         database,
                         client_session_id,
-                        result.as_ref().err().expect("replace-runtime action requires an error"),
+                        recovery.agent_session_id.as_deref(),
                         true,
                     )
                     .await;
@@ -4944,11 +4951,11 @@ where
             }
             MetadataErrorAction::Discard => {
                 state
-                    .detach_metadata_pool_after_error(
+                    .detach_metadata_pool_after_recovery(
                         connection_id,
                         database,
                         client_session_id,
-                        result.as_ref().err().expect("discard action requires an error"),
+                        recovery.agent_session_id.as_deref(),
                         false,
                     )
                     .await;
@@ -4959,25 +4966,26 @@ where
                 if let Err(error) =
                     state.reconnect_metadata_pool_for_session(connection_id, database, client_session_id).await
                 {
-                    match metadata_error_action(db_type, &error, true) {
+                    let reconnect_recovery = metadata_recovery(db_type, &error, true);
+                    match reconnect_recovery.action {
                         MetadataErrorAction::ReplaceRuntime => {
                             state
-                                .detach_metadata_pool_after_error(
+                                .detach_metadata_pool_after_recovery(
                                     connection_id,
                                     database,
                                     client_session_id,
-                                    &error,
+                                    reconnect_recovery.agent_session_id.as_deref(),
                                     true,
                                 )
                                 .await;
                         }
                         MetadataErrorAction::Retry | MetadataErrorAction::Discard => {
                             state
-                                .detach_metadata_pool_after_error(
+                                .detach_metadata_pool_after_recovery(
                                     connection_id,
                                     database,
                                     client_session_id,
-                                    &error,
+                                    reconnect_recovery.agent_session_id.as_deref(),
                                     false,
                                 )
                                 .await;
@@ -4992,26 +5000,48 @@ where
     }
 }
 
-#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
 enum MetadataErrorAction {
     Retry,
     Discard,
     ReplaceRuntime,
+    #[default]
     Return,
 }
 
+#[derive(Debug, Default)]
+struct MetadataRecovery {
+    action: MetadataErrorAction,
+    agent_session_id: Option<String>,
+}
+
+#[cfg(test)]
 fn metadata_error_action(db_type: Option<DatabaseType>, error: &str, retried: bool) -> MetadataErrorAction {
-    if crate::db::agent_driver::agent_session_disposition(error)
-        == Some(crate::db::agent_driver::AgentSessionDisposition::ReplaceRuntime)
-    {
-        MetadataErrorAction::ReplaceRuntime
-    } else if !retried && is_retryable_metadata_error(error) {
+    metadata_recovery(db_type, error, retried).action
+}
+
+fn metadata_recovery(db_type: Option<DatabaseType>, error: &str, retried: bool) -> MetadataRecovery {
+    if db_type.is_some_and(|db_type| crate::database_capabilities::is_agent_type(&db_type)) {
+        if let Some(error) = crate::db::agent_driver::try_agent_error_from_legacy(error) {
+            let agent_session_id = error.session_id().map(str::to_string);
+            let action = match RecoveryPolicy::decide(&error, RecoveryScope::ReadOnlyMetadata { retried }) {
+                RecoveryDecision::RetryReadOnlyMetadata => MetadataErrorAction::Retry,
+                RecoveryDecision::QuarantineSession => MetadataErrorAction::Discard,
+                RecoveryDecision::ReplaceRuntime => MetadataErrorAction::ReplaceRuntime,
+                RecoveryDecision::KeepSession => MetadataErrorAction::Return,
+            };
+            return MetadataRecovery { action, agent_session_id };
+        }
+    }
+
+    let action = if !retried && is_retryable_metadata_error(error) {
         MetadataErrorAction::Retry
     } else if should_discard_pool_after_error(db_type, error) {
         MetadataErrorAction::Discard
     } else {
         MetadataErrorAction::Return
-    }
+    };
+    MetadataRecovery { action, agent_session_id: None }
 }
 
 #[cfg(test)]
@@ -5025,11 +5055,9 @@ async fn replace_metadata_runtime(
 }
 
 fn is_retryable_metadata_error(error: &str) -> bool {
-    let category = crate::db::agent_driver::agent_rpc_error_category(error);
-    if let Some(category) = category {
-        return category == "connection"
-            && crate::db::agent_driver::agent_session_disposition(error)
-                == Some(crate::db::agent_driver::AgentSessionDisposition::Quarantine);
+    if let Some(error) = crate::db::agent_driver::try_agent_error_from_legacy(error) {
+        return RecoveryPolicy::decide(&error, RecoveryScope::ReadOnlyMetadata { retried: false })
+            == RecoveryDecision::RetryReadOnlyMetadata;
     }
     error == "Pool not found" || crate::query::is_connection_error(error)
 }
@@ -5294,8 +5322,9 @@ async fn get_columns_core_for_session_inner(
                                     .await
                                     .map(deduplicate_column_infos)
                                     .map_err(|fallback_error| {
-                                        format!(
-                                            "{agent_error}\n\nNative PostgreSQL metadata fallback failed: {fallback_error}"
+                                        crate::db::agent_driver::append_legacy_error_context(
+                                            &agent_error,
+                                            &format!("Native PostgreSQL metadata fallback failed: {fallback_error}"),
                                         )
                                     });
                             }