diff --git a/application/single_app/config.py b/application/single_app/config.py
index d196346e8..6d8303386 100644
--- a/application/single_app/config.py
+++ b/application/single_app/config.py
@@ -101,7 +101,7 @@
 EXECUTOR_TYPE = 'thread'
 EXECUTOR_MAX_WORKERS = 30
 SESSION_TYPE = 'filesystem'
-VERSION = "0.261.350"
+VERSION = "0.261.353"
 IS_DEVELOPMENT = is_development_env_enabled()
 
 # Opt-out for deployments where App Service Easy Auth is active but the platform
diff --git a/application/single_app/functions_appinsights.py b/application/single_app/functions_appinsights.py
index 9e2a53e6f..6c0e8b4c3 100644
--- a/application/single_app/functions_appinsights.py
+++ b/application/single_app/functions_appinsights.py
@@ -12,6 +12,9 @@
 from azure.monitor.opentelemetry import configure_azure_monitor
 import app_settings_cache
 from functions_assist_diagnostics import ASSIST_DIAGNOSTIC_TEXT, safe_assist_diagnostics
+from semantic_kernel_plugins.sql_connection_diagnostics import (
+    SQL_ENGINES, valid_native_error, valid_sqlstate,
+)
 
 # Singleton for the logger and Azure Monitor configuration
 _appinsights_logger = None
@@ -307,6 +310,15 @@ def _build_logger_extra(
             if _is_sensitive_log_key(key):
                 logger_extra[f"{normalized_key}_present"] = value is not None
                 continue
+            diagnostic_key = _normalize_log_key(key)
+            if diagnostic_key in {"sqlstate", "sqlengine", "sqlnativeerror"}:
+                if (
+                    (diagnostic_key == "sqlstate" and valid_sqlstate(value))
+                    or (diagnostic_key == "sqlengine" and type(value) is str and value in SQL_ENGINES)
+                    or (diagnostic_key == "sqlnativeerror" and valid_native_error(value))
+                ):
+                    logger_extra[normalized_key] = value
+                continue
             if isinstance(value, dict):
                 logger_extra[f"{normalized_key}_count"] = len(value)
             elif isinstance(value, (list, tuple, set)):
diff --git a/application/single_app/route_backend_plugins.py b/application/single_app/route_backend_plugins.py
index 19a2be111..8adb74bd8 100644
--- a/application/single_app/route_backend_plugins.py
+++ b/application/single_app/route_backend_plugins.py
@@ -17,6 +17,7 @@
     build_sql_server_odbc_connection_string,
     connect_with_sql_server_odbc_fallback,
 )
+from semantic_kernel_plugins.sql_connection_diagnostics import classify_sql_connection_error
 from semantic_kernel_plugins.rocksdb_plugin import (
     AUTH_SCHEME_API_KEY,
     AUTH_SCHEME_BEARER,
@@ -3417,30 +3418,16 @@ def test_sql_connection():
         else:
             return jsonify({'success': False, 'error': f'Unsupported database type: {database_type}'}), 400
 
-    except ImportError as e:
-        if editor_call:
-            return jsonify({'success': False, 'error': 'The required database driver is not available on the server.'}), 400
-        if database_type == 'sqlserver' and 'libodbc' in str(e):
-            return jsonify({
-                'success': False,
-                'error': 'Database driver not installed: the container image is missing the unixODBC runtime required for SQL Server connections.'
-            }), 400
-        return jsonify({'success': False, 'error': f'Database driver not installed: {str(e)}'}), 400
     except Exception as e:
-        if editor_call:
-            log_event(
-                '[PLUGINS] SQL editor connection test failed.',
-                level=logging.WARNING,
-                extra={'database_type': database_type, 'error_type': type(e).__name__},
-            )
-            return jsonify({'success': False, 'error': 'SQL connection failed. Verify the connection and authentication settings.'}), 400
-        error_msg = str(e)
-        if database_type == 'sqlserver' and "Can't open lib 'ODBC Driver 17 for SQL Server'" in error_msg:
-            error_msg = 'The selected ODBC Driver 17 is not installed in this container image. Select ODBC Driver 18 for SQL Server or rebuild the image with Driver 17.'
-        # Sanitize error message to avoid leaking sensitive details
-        if 'password' in error_msg.lower() or 'pwd' in error_msg.lower():
-            error_msg = 'Authentication failed. Please check your credentials.'
-        return jsonify({'success': False, 'error': f'Connection failed: {error_msg}'}), 400
+        diagnostic = classify_sql_connection_error(e, database_type)
+        log_event(
+            '[SQL_CONNECTION] SQL connection check failed.',
+            level=logging.WARNING,
+            extra={**diagnostic.log_properties(), 'operation': 'sql_connection_test'},
+        )
+        return jsonify({
+            'success': False, 'error': diagnostic.message, 'error_code': diagnostic.code,
+        }), 400
 
 
 @bpap.route('/api/plugins/test-cosmos-connection', methods=['POST'])
diff --git a/application/single_app/semantic_kernel_plugins/sql_connection_diagnostics.py b/application/single_app/semantic_kernel_plugins/sql_connection_diagnostics.py
new file mode 100644
index 000000000..37e9713f9
--- /dev/null
+++ b/application/single_app/semantic_kernel_plugins/sql_connection_diagnostics.py
@@ -0,0 +1,117 @@
+# sql_connection_diagnostics.py
+"""Credential-free SQL connection diagnostics, with no application bootstrap or I/O."""
+
+import re
+from dataclasses import dataclass
+from typing import Optional
+
+
+SQL_ENGINES = frozenset({"sqlserver", "postgresql", "mysql", "sqlite", "unknown"})
+SQLSTATE_PATTERN = re.compile(r"(?:[0-9]{2}|HY|IM|S[0-9]|HV|P0|XX|F0)[0-9A-Z]{3}", re.ASCII)
+MAX_NATIVE_ERROR = 2**31 - 1
+_ODBC_NATIVE_ERROR = re.compile(r"\((-?[0-9]{1,10})\) \((?:SQLDriverConnect|SQLConnect)\)$", re.ASCII)
+_ERROR_TYPES = frozenset({
+    "Error", "DatabaseError", "InterfaceError", "OperationalError", "InternalError",
+    "ProgrammingError", "NotSupportedError", "IntegrityError", "DataError",
+    "ImportError", "ModuleNotFoundError", "TimeoutError", "ConnectionError", "OSError",
+})
+_MESSAGES = {
+    "sql_connection_failed": "SQL connection failed. Verify the connection and authentication settings.",
+    "sql_driver_unavailable": "The required database driver is not available on the server. Contact your administrator.",
+    "sql_connection_timeout": "SQL connection timed out. Check database availability and network access.",
+    "sql_authentication_failed": "SQL authentication was rejected. Verify the selected authentication settings and database access.",
+}
+
+
+def valid_sqlstate(value: object) -> bool:
+    """Accept only bounded SQLSTATE values, not provider messages or arbitrary strings."""
+    return type(value) is str and len(value) == 5 and SQLSTATE_PATTERN.fullmatch(value) is not None
+
+
+def valid_native_error(value: object) -> bool:
+    return type(value) is int and -MAX_NATIVE_ERROR <= value <= MAX_NATIVE_ERROR
+
+
+@dataclass(frozen=True)
+class SqlConnectionDiagnostic:
+    code: str
+    engine: str
+    error_type: str
+    sqlstate: Optional[str] = None
+    native_error: Optional[int] = None
+
+    @property
+    def message(self) -> str:
+        return _MESSAGES[self.code]
+
+    def log_properties(self) -> dict:
+        properties = {
+            "failure_code": self.code,
+            "sql_engine": self.engine,
+            "error_type": self.error_type,
+        }
+        if self.sqlstate is not None:
+            properties["sqlstate"] = self.sqlstate
+        if self.native_error is not None:
+            properties["sql_native_error"] = self.native_error
+        return properties
+
+
+class SqlConnectionError(RuntimeError):
+    """A connection failure safe for existing plugin result and logging handlers."""
+
+    def __init__(self, diagnostic: SqlConnectionDiagnostic):
+        self.diagnostic = diagnostic
+        super().__init__(diagnostic.message)
+
+
+def classify_sql_connection_error(error: Exception, database_type: str) -> SqlConnectionDiagnostic:
+    """Classify structured driver evidence without retaining raw exception messages."""
+    engine = database_type if type(database_type) is str else "unknown"
+    if engine in {"azure_sql", "azuresql"}:
+        engine = "sqlserver"
+    if engine not in SQL_ENGINES:
+        engine = "unknown"
+
+    sqlstate = None
+    native_error = None
+    args = error.args
+    if engine in {"sqlserver", "postgresql"}:
+        for candidate in (getattr(error, "sqlstate", None), getattr(error, "pgcode", None), args[0] if args else None):
+            if valid_sqlstate(candidate):
+                sqlstate = candidate
+                break
+    if engine == "sqlserver" and sqlstate and len(args) > 1:
+        detail = args[1]
+        # pyodbc puts the native code at the end of its SQLDriverConnect diagnostic.
+        # Never collect arbitrary numbers (ports, IDs, or credentials) from provider text.
+        if type(detail) is str and len(detail) <= 8192 and detail.startswith(f"[{sqlstate}]"):
+            match = _ODBC_NATIVE_ERROR.search(detail)
+            if match:
+                candidate = int(match.group(1))
+                if valid_native_error(candidate):
+                    native_error = candidate
+    elif engine == "mysql" and args and valid_native_error(args[0]):
+        native_error = args[0]
+    elif engine == "sqlite":
+        candidate = getattr(error, "sqlite_errorcode", None)
+        if valid_native_error(candidate):
+            native_error = candidate
+
+    code = "sql_connection_failed"
+    if isinstance(error, ImportError) or sqlstate in {"IM002", "IM003"}:
+        code = "sql_driver_unavailable"
+    elif isinstance(error, TimeoutError) or sqlstate in {"HYT00", "HYT01"}:
+        code = "sql_connection_timeout"
+    elif (
+        (sqlstate is not None and sqlstate.startswith("28"))
+        or (engine == "sqlserver" and native_error in {18452, 18456})
+        or (engine == "mysql" and native_error in {1045, 1698})
+    ):
+        code = "sql_authentication_failed"
+
+    error_type = type(error).__name__
+    return SqlConnectionDiagnostic(
+        code, engine, error_type if error_type in _ERROR_TYPES else "Exception",
+        sqlstate, native_error,
+    )
diff --git a/application/single_app/semantic_kernel_plugins/sql_query_plugin.py b/application/single_app/semantic_kernel_plugins/sql_query_plugin.py
index e313aeb99..aee85753f 100644
--- a/application/single_app/semantic_kernel_plugins/sql_query_plugin.py
+++ b/application/single_app/semantic_kernel_plugins/sql_query_plugin.py
@@ -1,3 +1,4 @@
+# sql_query_plugin.py
 """
 SQL Query Plugin for Semantic Kernel
 - Executes SQL queries against various databases (SQL Server, PostgreSQL, MySQL, SQLite)
@@ -19,6 +20,9 @@
     build_sql_server_odbc_connection_string,
     connect_with_sql_server_odbc_fallback,
 )
+from semantic_kernel_plugins.sql_connection_diagnostics import (
+    SqlConnectionError, classify_sql_connection_error,
+)
 
 # Helper class to wrap results with metadata
 class ResultWithMetadata:
@@ -189,16 +193,14 @@ def _create_connection(self):
                 conn.row_factory = sqlite3.Row
                 return conn
                 
-        except ImportError as e:
-            debug_print(f"[SQL_QUERY_PLUGIN] Database driver import failed database_type={self.database_type} error={e}")
-            raise ImportError(f"Required database driver not installed for {self.database_type}: {e}")
         except Exception as e:
-            debug_print(
-                f"[SQL_QUERY_PLUGIN] Connection failed database_type={self.database_type} server={self.server} "
-                f"database={self.database} exception_type={type(e).__name__} message={e}"
+            diagnostic = classify_sql_connection_error(e, self.database_type)
+            log_event(
+                "[SQL_CONNECTION] SQL connection failed.",
+                level=logging.WARNING,
+                extra={**diagnostic.log_properties(), "operation": "sql_query_connect"},
             )
-            log_event(f"[SQL_QUERY_PLUGIN] Connection failed: {e}", extra={"database_type": self.database_type})
-            raise
+            raise SqlConnectionError(diagnostic) from None
 
     @property
     def display_name(self) -> str:
@@ -356,6 +358,8 @@ def execute_query(
                 "data": [],
                 "row_count": 0
             }
+            if isinstance(e, SqlConnectionError):
+                error_result["error_code"] = e.diagnostic.code
             return ResultWithMetadata(error_result, self.metadata)
 
     @kernel_function(description="Execute a query that returns a single scalar value (e.g., COUNT, SUM, MAX, MIN). If the database schema is provided in your instructions, use it directly to construct the query. Otherwise, call get_database_schema from the SQL Schema plugin first to discover table and column names.")
@@ -424,6 +428,8 @@ def execute_scalar(
                 "query": query,
                 "value": None
             }
+            if isinstance(e, SqlConnectionError):
+                error_result["error_code"] = e.diagnostic.code
             return ResultWithMetadata(error_result, self.metadata)
 
     @kernel_function(description="Validate a SQL query for syntax correctness and safety without executing it. Use this to pre-check complex queries before execution, especially when constructing multi-table JOINs or complex WHERE clauses.")
@@ -530,6 +536,8 @@ def query_database(
                 "data": [],
                 "row_count": 0
             }
+            if isinstance(e, SqlConnectionError):
+                error_result["error_code"] = e.diagnostic.code
             return ResultWithMetadata(error_result, self.metadata)
 
     def _clean_query(self, query: str) -> str:
diff --git a/application/single_app/semantic_kernel_plugins/sql_schema_plugin.py b/application/single_app/semantic_kernel_plugins/sql_schema_plugin.py
index 09babe2cb..c74a157fb 100644
--- a/application/single_app/semantic_kernel_plugins/sql_schema_plugin.py
+++ b/application/single_app/semantic_kernel_plugins/sql_schema_plugin.py
@@ -1,3 +1,4 @@
+# sql_schema_plugin.py
 """
 SQL Schema Plugin for Semantic Kernel
 - Connects to various SQL databases (SQL Server, PostgreSQL, MySQL, SQLite)
@@ -17,6 +18,9 @@
     build_sql_server_odbc_connection_string,
     connect_with_sql_server_odbc_fallback,
 )
+from semantic_kernel_plugins.sql_connection_diagnostics import (
+    SqlConnectionError, classify_sql_connection_error,
+)
 
 # Helper class to wrap results with metadata
 class ResultWithMetadata:
@@ -175,16 +179,14 @@ def _create_connection(self):
                 database_path = self.connection_string or self.database
                 return sqlite3.connect(database_path)
                 
-        except ImportError as e:
-            debug_print(f"[SQL_SCHEMA_PLUGIN] Database driver import failed database_type={self.database_type} error={e}")
-            raise ImportError(f"Required database driver not installed for {self.database_type}: {e}")
         except Exception as e:
-            debug_print(
-                f"[SQL_SCHEMA_PLUGIN] Connection failed database_type={self.database_type} server={self.server} "
-                f"database={self.database} exception_type={type(e).__name__} message={e}"
+            diagnostic = classify_sql_connection_error(e, self.database_type)
+            log_event(
+                "[SQL_CONNECTION] SQL connection failed.",
+                level=logging.WARNING,
+                extra={**diagnostic.log_properties(), "operation": "sql_schema_connect"},
             )
-            log_event(f"[SQL_SCHEMA_PLUGIN] Connection failed: {e}", extra={"database_type": self.database_type})
-            raise
+            raise SqlConnectionError(diagnostic) from None
 
     @property
     def display_name(self) -> str:
@@ -338,8 +340,11 @@ def get_database_schema(
                 "database_type": self.database_type,
                 "database": self.database
             })
+            error_result = {"error": error_msg}
+            if isinstance(e, SqlConnectionError):
+                error_result["error_code"] = e.diagnostic.code
             return ResultWithMetadata(
-                {"error": error_msg},
+                error_result,
                 {"source": "sql_schema_plugin", "success": False}
             )
 
diff --git a/docs/explanation/fixes/SQL_CONNECTION_DIAGNOSTICS_FIX.md b/docs/explanation/fixes/SQL_CONNECTION_DIAGNOSTICS_FIX.md
new file mode 100644
index 000000000..19cc8b6d9
--- /dev/null
+++ b/docs/explanation/fixes/SQL_CONNECTION_DIAGNOSTICS_FIX.md
@@ -0,0 +1,191 @@
+# SQL connection diagnostics fix - v0.261.353
+
+## Issue and root cause
+
+FED-16 stopped at the action editor's SQL connection check. The server recorded
+HTTP 400 and `InterfaceError`, but those facts did not explain why the database
+connection failed. The editor discarded structured SQL diagnostics, while
+classic callers and runtime plugins could expose raw provider exception text.
+Adding fields at the call site alone was insufficient: the logging boundary
+normally reduces unapproved string properties to their lengths.
+
+**Fixed in version: 0.261.351**, recorded in
+`application/single_app/config.py`.
+
+The base-branch integration preserves its saved-memory diagnostics and advances
+the application from **0.261.350** to **0.261.351**. SQL implementation-version
+references were updated from the initial branch-only **0.261.349**.
+
+This fixes the diagnostic gap, **not the unresolved cause of FED-16's live
+connection failure**. Successful SQL access using an operator's account does not
+qualify the App Service identity's connection. No deployment or live connection
+retry was performed for this fix.
+
+### Cosmos review integration - v0.261.352
+
+PR #1787's SQL route edit changed `route_backend_plugins.py`'s module fingerprint.
+The Cosmos contract check therefore invalidated its three source-only reviews:
+`set_user_plugins`, `edit_plugin`, and `test_cosmos_connection`. Their function
+bodies and Cosmos behavior were unchanged, but the inventory intentionally binds
+reviews to the entire owning module.
+
+**Fixed in version: 0.261.352**, recorded in `application/single_app/config.py`.
+The reviewed inventory now links each owner to its own runtime regression and
+removes only these three stale source-review baseline entries. No gate, baseline
+assurance, or production Cosmos behavior was relaxed.
+
+`functional_tests/test_plugin_routes_cosmos_contract.py` executes the real
+handlers and legacy-update validator with installed-SDK `CosmosDict` point
+responses. It covers successful personal/global updates, exact-ID and personal
+partition misses, global name mismatches, ETag preservation, and lazy Cosmos
+query consumption including empty results and request-charge headers. External
+services are offline seams; the contract runner blocks network access before
+collection.
+
+Validation for **0.261.352**, using a repository-local Python environment with
+the declared application/test pins and without changing shared Python packages:
+
+- `python scripts/check_cosmos_response_boundaries.py --run-contracts`:
+  **1,151 passed and 244 subtests passed**; **1,067 owners, 412 doubles, zero
+  review gaps**. The runner automatically includes all three new owner selectors.
+- New plugin-owner regressions: **8 passed** with Flask **3.1.3** and Cosmos SDK
+  **4.9.0**; SQL diagnostics regressions: **106 passed**.
+- Optimized new owner and SQL classifier selection: **46 passed**. Required
+  operations remain outside assertions; optimized execution is not treated as
+  proof of assertion coverage.
+- Route policy checks: **15/15**, **10/10**, and **3/3** passed; documentation
+  coverage and quality: **7/7** and **6/6** passed.
+
+### CodeQL test-local initialization follow-up - v0.261.353
+
+CodeQL `py/uninitialized-local-variable` flagged `stored` in the personal/global
+SDK response tests (alerts 4846 and 4847). The existing success guards implied
+that a record had been created, but the local bindings were conditional.
+Both tests now initialize `stored` before that condition and assert the SDK
+response type on successful updates before comparing ETags. Missing-record and
+mismatched-owner/name cases retain all refusal assertions.
+
+**Fixed in version: 0.261.353**, recorded in `application/single_app/config.py`.
+No production handler, Cosmos review evidence, test selector, assertion policy,
+or CI configuration changed.
+
+The guarded owner/SQL regression selection passed **114 cases**; its optimized
+owner/classifier selection passed **46 cases**. Cosmos inventory validation
+reported **1,067 owners, 412 doubles, zero review gaps**, and documentation
+coverage/quality checks passed **7/7** and **6/6**. These local results do not
+replace the new commit's CodeQL analysis.
+
+## Technical details
+
+### Shared connection boundary
+
+`semantic_kernel_plugins/sql_connection_diagnostics.py` is a standard-library-only
+classifier. It imports no application configuration, SQL driver, or Azure client
+and performs no I/O.
+
+It normalizes Azure SQL aliases to SQL Server and reads validated SQLSTATE,
+PostgreSQL `pgcode`/`sqlstate`, MySQL numeric error arguments, and SQLite error
+codes when available. ODBC native numbers are accepted only from the bounded,
+SQLSTATE-prefixed `SQLDriverConnect`/`SQLConnect` diagnostic format, not arbitrary
+numbers found in exception messages.
+
+The four application-owned categories are `sql_connection_failed`,
+`sql_driver_unavailable`, `sql_connection_timeout`, and
+`sql_authentication_failed`. See the
+[SQL action troubleshooting reference](../../reference/actions/sql-query.md#connection-failure-categories)
+for their meaning. An `InterfaceError` alone stays general; SQLSTATE class `08`
+does not establish a firewall, TLS, or identity cause.
+
+`route_backend_plugins.py::test_sql_connection` returns static error guidance and
+an additive `error_code`, preserving HTTP 400, `success: false`, and `error`.
+Classic and editor requests use the same provider-error boundary.
+
+`sql_query_plugin.py` and `sql_schema_plugin.py` log the same diagnostic at
+connection creation and raise a safe `SqlConnectionError` with raw exception
+context suppressed in formatted tracebacks. Existing outer handlers therefore
+receive static text rather than raw connection errors. Query, scalar,
+question/query, and complete-schema failure results add `error_code`; the other
+schema methods propagate the safe exception.
+
+### Operator telemetry and privacy
+
+`functions_appinsights.py` validates the new SQL fields before retaining them in
+log records. `[SQL_CONNECTION]` warnings do not require debug logging; exporting
+them still requires Application Insights to be configured.
+
+| Property on the log record | Content |
+| --- | --- |
+| `sc_failure_code` | Static application-owned failure category |
+| `sc_sql_engine` | Allowlisted, normalized engine |
+| `sc_error_type` | Known exception class name, or `Exception` |
+| `sc_operation` | Connection test, query connection, or schema connection |
+| `sc_sqlstate` | Validated five-character provider SQLSTATE, if available |
+| `sc_sql_native_error` | Bounded integer native error number, if available |
+
+Raw provider messages, credentials, connection strings, SQL, server/database
+names, and identity/workspace identifiers are not added to this diagnostic.
+Malformed or unsupported structured fields are omitted. Provider codes stay out
+of browser JSON and plugin connection-failure results. The existing telemetry
+operation correlation is retained; see the
+[logging reference](../../reference/logging-tags.md#sql-connection-diagnostics)
+for a request-correlated Log Analytics query.
+
+### Compatibility and files
+
+Authentication selection, reusable-identity authorization/hydration, driver
+inputs, connection timeouts, successful connection caching, and successful
+response shapes remain unchanged. The existing bounded Driver 17-to-18
+missing-driver fallback is unchanged; no new retries, permissions, TLS changes,
+or connection/authentication repairs were introduced.
+
+The SQL action references and logging-tag inventory document the new behavior.
+`scripts/build_docs_inventory.py` excludes the diagnostic helper as plugin
+infrastructure, alongside the existing ODBC helper. Regenerating the inventory
+leaves the user-facing action count unchanged.
+
+The existing `test_workspace_authoring_backend.py` fixture loads the new
+production dependency for its extracted routes. No federal harness source,
+workflow, selector, acceptance rule, or unrelated fixture defect was repaired.
+Harness findings are retained separately in session artifacts.
+
+## Testing and validation
+
+`functional_tests/test_sql_connection_diagnostics.py` covers evidence-based
+classification, malformed/oversized/nested inputs, conservative unknown errors,
+inventory exclusion, and fresh-process imports with external dependencies and
+network access blocked. Its subprocess probe explicitly checks required
+operations in normal and optimized Python.
+
+`functional_tests/test_sql_connection_diagnostics_runtime.py` runs actual plugin
+classes, ODBC helpers, extracted Flask handlers, and the real logging boundary
+against isolated external I/O seams. Coverage includes provider failure privacy,
+debug on/off, classic/editor and new/saved requests, scoped identities and
+denials, success responses, authentication inputs, timeout caps, connection
+caching, and unchanged fallback bounds.
+
+Validation results:
+
+- Combined focused diagnostics and existing SQL compatibility run: **114 passed**
+  (106 diagnostic tests and eight existing compatibility cases).
+- Optimized classifier run: **38 passed**, including the explicit subprocess
+  probe; optimized execution alone is not treated as proof of assertion coverage.
+- Driver/logging compatibility selection: **16 passed**, with one confirmed
+  pre-existing debug-log expectation failure. Loading the unchanged `HEAD`
+  logger reproduced that failure.
+- Route-policy scripts: **15/15**, **10/10**, and **3/3** passed.
+- Regenerated documentation inventory/coverage: **7/7** passed; documentation
+  site source-quality checks: **6/6** passed.
+- The repository-wide logging-tag check remains failing on unrelated existing
+  mixed-case tags and inventory drift. All tags in modified product files are
+  normalized, and `[SQL_CONNECTION]` is documented.
+
+After integrating base version **0.261.350**, the 114 focused SQL cases, 38
+optimized cases, and 13 documentation checks passed again. Additional inherited
+saved-memory validation reached ten passing cases and five passing subtests, but
+47 cases could not set up because the shared Python installation's OpenSSL dependency lacks
+`lib.GEN_EMAIL`, the environment limitation already noted in the base fix.
+No shared packages, inherited memory code, or assertions were changed to bypass it.
+
+The original live FED-16 acceptance outcome remains unqualified. A separately
+approved deployment and observed connection check are needed to obtain the
+missing provider evidence before selecting a connection fix.
diff --git a/docs/reference/actions/sql-query.md b/docs/reference/actions/sql-query.md
index efac50e58..c11936a9c 100644
--- a/docs/reference/actions/sql-query.md
+++ b/docs/reference/actions/sql-query.md
@@ -46,10 +46,22 @@ Use SQL Database when the answer depends on exact relational data: joins, filter
 
 | Symptom | Likely cause | Fix |
 | --- | --- | --- |
-| Connection test fails | Connection string, server, database, driver, or credentials do not match the target. | Test the same credentials outside SimpleChat, then correct the modal fields. |
+| Connection test fails | The app could not complete the database connection check. The message alone may not identify the cause. | Use the failure categories below and ask an administrator to inspect the correlated SQL connection event. |
 | The agent writes invalid SQL | It does not know the schema. | Pair the agent with SQL Schema or provide table and column names in instructions. |
 | Results are truncated | **Max Rows** is lower than the result size. | Ask for a grouped or filtered result, or raise **Max Rows** within governance limits. |
 
+### Connection failure categories
+
+Since **0.261.351**, connection testing and SQL action runtime share safe error messages. The API and supported plugin result objects include an application-owned `error_code` for connection failures; provider details remain in server telemetry.
+
+| Error code | Meaning and next step |
+| --- | --- |
+| `sql_driver_unavailable` | A driver import or driver-manager lookup failed. Ask the administrator to check the deployed database driver and its dependencies. |
+| `sql_connection_timeout` | A timeout was reported. Check database availability and the app's network path; this does not establish a particular firewall problem. |
+| `sql_authentication_failed` | Structured provider evidence reports authentication rejection. Verify the selected authentication configuration and database access for that identity. |
+| `sql_connection_failed` | The available evidence does not support a more specific category. Inspect the server diagnostic before changing authentication, TLS, or permissions. |
+
+Administrators can find `[SQL_CONNECTION]` warnings with validated SQLSTATE and native error numbers when the provider supplies them. See [SQL connection logging]({{ '/reference/logging-tags/' | relative_url }}#sql-connection-diagnostics). A successful login using an operator's own account does not verify the App Service identity, its network path, or its selected authentication mode.
 
 ## V2 naming and schema discovery
 
diff --git a/docs/reference/actions/sql-schema.md b/docs/reference/actions/sql-schema.md
index e09c82ce4..a4e73a7bd 100644
--- a/docs/reference/actions/sql-schema.md
+++ b/docs/reference/actions/sql-schema.md
@@ -52,13 +52,15 @@ The V2 editor no longer offers ad hoc custom fields; preserved legacy values rem
 
 | Symptom | Likely cause | Fix |
 | --- | --- | --- |
+| Connection fails before metadata discovery | The app cannot establish the configured database connection. | Use the shared [SQL connection failure categories]({{ '/reference/actions/sql-query/' | relative_url }}#connection-failure-categories) and have an administrator inspect the correlated server diagnostic. |
 | Schema output includes too much noise | System tables or broad schemas are included. | Turn off **Include System Tables** and add a **Table Filter**. |
 | The agent cannot see relationships | Metadata permissions are missing or relationships are not defined as foreign keys. | Grant metadata visibility or document the relationship in agent instructions. |
 | The action type is not visible as a separate card | `sql_schema` is hidden from the create-action UI. | Use SQL Database for new actions; existing records still open for compatibility. |
 
+Since **0.261.351**, SQL Schema uses the same credential-safe connection diagnostics as SQL Database. A connection failure in `get_database_schema` includes an application-owned `error_code`; the table-list, table-detail, and relationship methods raise the same safe connection exception. This does not change metadata permissions or the handling of errors after a connection succeeds.
+
 ## Related
 
 - [SQL Query](../sql-query/)
 - [Actions reference index]({{ '/reference/actions/' | relative_url }})
 - [Agents administration]({{ '/admin/agents-actions/' | relative_url }})
-
diff --git a/docs/reference/logging-tags.md b/docs/reference/logging-tags.md
index c76972fd5..fc2673d93 100644
--- a/docs/reference/logging-tags.md
+++ b/docs/reference/logging-tags.md
@@ -42,6 +42,46 @@ and capability codes. A cancelled stream is informational, not an error.
 Orchestration records an outcome only after its saved state is confirmed;
 replaying a cached response does not record another execution.
 
+## SQL connection diagnostics
+
+Since **0.261.351**, `[SQL_CONNECTION]` records connection-check and runtime
+connection failures at WARNING level, independently of debug logging. Application
+Insights must already be configured for these records to reach Azure.
+
+The shared classifier supplies `failure_code`, normalized `sql_engine`, and a
+known `error_type`. Valid provider evidence adds a five-character `sqlstate`
+and/or a bounded integer `sql_native_error`. Missing evidence is omitted, not
+invented. `operation` distinguishes `sql_connection_test`, `sql_query_connect`,
+and `sql_schema_connect`.
+
+The logging boundary preserves these as `sc_failure_code`, `sc_sql_engine`,
+`sc_error_type`, `sc_sqlstate`, `sc_sql_native_error`, and `sc_operation`.
+Use the existing request `OperationId` to correlate the warning; no raw
+identity/workspace identifiers, provider messages, credentials, connection
+strings, server/database names, or SQL are added to this event.
+
+For example, in a workspace-based Log Analytics resource:
+
+```kusto
+AppTraces
+| where TimeGenerated > ago(30m)
+| where OperationId == "PASTE_REQUEST_OPERATION_ID"
+| where tostring(Properties.sc_message) startswith "[SQL_CONNECTION]"
+| project TimeGenerated, OperationId,
+    Operation=tostring(Properties.sc_operation),
+    Code=tostring(Properties.sc_failure_code),
+    Engine=tostring(Properties.sc_sql_engine),
+    ErrorType=tostring(Properties.sc_error_type),
+    SqlState=tostring(Properties.sc_sqlstate),
+    NativeError=toint(Properties.sc_sql_native_error)
+| order by TimeGenerated asc
+| take 100
+```
+
+An `InterfaceError` without structured evidence remains `sql_connection_failed`.
+SQLSTATE class `08` alone does not prove a firewall, TLS, or identity problem.
+See the [SQL action troubleshooting categories]({{ '/reference/actions/sql-query/' | relative_url }}#connection-failure-categories).
+
 ## Request-resolution memory diagnostics
 
 Since **0.261.349**, `[ORCHESTRATION_PLANNER]` records
@@ -304,6 +344,7 @@ for availability states and live-UI validation limits.
 - `[SORT]`
 - `[SOURCE_REVIEW]`
 - `[SPEECH]`
+- `[SQL_CONNECTION]`
 - `[SQL_ODBC]`
 - `[SQL_QUERY_PLUGIN]`
 - `[SQL_SCHEMA_PLUGIN]`
diff --git a/functional_tests/cosmos_response_boundaries.json b/functional_tests/cosmos_response_boundaries.json
index a1f5c315d..0dd7d5cde 100644
--- a/functional_tests/cosmos_response_boundaries.json
+++ b/functional_tests/cosmos_response_boundaries.json
@@ -809,9 +809,6 @@
     "application/single_app/route_backend_groups.py::register_route_backend_groups.api_group_activity": "b68cf13c1361c5df6a62e49dd97426deeacf9389b2889dcb7efc02697d528696",
     "application/single_app/route_backend_groups.py::register_route_backend_groups.api_group_stats": "99baefcd65410b84b8eef2e3c7a2113dc43b3a16444f930e077e4c9311f98cf7",
     "application/single_app/route_backend_m365.py::register_route_backend_m365.list_m365_waiting_requests": "a08dfc34823a0019c09d30bf38f7e25ac1dfd6e81944665f3912f2e4daa7a73f",
-    "application/single_app/route_backend_plugins.py::set_user_plugins": "6521a2fc13713578fcabc57c611d5a73467630fe3b4006082b579cb1dc7db82a",
-    "application/single_app/route_backend_plugins.py::edit_plugin": "e4294468015daedc797eb287aa58c9066c8efe91a9f593a07936b55da9e3e1b1",
-    "application/single_app/route_backend_plugins.py::test_cosmos_connection": "2914add8869a52f46cf5639d26e9dda1a163ab1b980935aa459b85a760b3c9be",
     "application/single_app/route_backend_public_workspaces.py::register_route_backend_public_workspaces.discover_public_workspaces": "062c6c6441e6af64ba3ad877183fbe0c6e5b1b7557e4d98888d37ebb255c2300",
     "application/single_app/route_backend_public_workspaces.py::register_route_backend_public_workspaces.api_public_file_count": "81af253b9f8a788afaa40eed182e0a941eab4c63290cf512516a0ccd8ae29ec8",
     "application/single_app/route_backend_public_workspaces.py::register_route_backend_public_workspaces.api_public_workspace_stats": "0c7b1bca8b55d4380e5fede0efcf6f785f8e4657006c7b65e2e2eb02627c294c",
@@ -23743,14 +23740,14 @@
         ],
         "queries": [],
         "batches": [],
-        "module_sha256": "c0b926e2dd2d7fed9999a9136c944270753cba204c11539937ef49ebe9cad9ed"
+        "module_sha256": "5f329bcdcdf3d94275191de8c5b2e9d4eca3488dd8f280d2e387f9d5467df5ab"
       },
       "classification": "sdk-compatible",
-      "contract": "set_user_plugins: point documents use Mapping field reads/JSON encoding/metadata projections; batch status/body lists are separate, never top-level-normalized as documents. Strict domain sinks remain with their storage owners.",
-      "evidence_scope": "sdk-api-and-source-review",
+      "contract": "Legacy personal updates consume SDK point documents through Mapping-compatible field reads and the real legacy-action validator, preserving the ETag. Reads are bound to the current user partition and exact action ID; missing or other-user records fail before saves.",
       "tests": [
-        "functional_tests/test_m365_cosmos_sdk_contract.py::test_real_sdk_point_documents_are_wrapped_but_nested_json_and_query_rows_are_plain"
-      ]
+        "functional_tests/test_plugin_routes_cosmos_contract.py::test_personal_legacy_update_consumes_sdk_point_response"
+      ],
+      "evidence_scope": "owner-runtime"
     },
     "application/single_app/route_backend_plugins.py::edit_plugin": {
       "boundary": {
@@ -23764,14 +23761,14 @@
         ],
         "queries": [],
         "batches": [],
-        "module_sha256": "c0b926e2dd2d7fed9999a9136c944270753cba204c11539937ef49ebe9cad9ed"
+        "module_sha256": "5f329bcdcdf3d94275191de8c5b2e9d4eca3488dd8f280d2e387f9d5467df5ab"
       },
       "classification": "sdk-compatible",
-      "contract": "edit_plugin: point documents use Mapping field reads/JSON encoding/metadata projections; batch status/body lists are separate, never top-level-normalized as documents. Strict domain sinks remain with their storage owners.",
-      "evidence_scope": "sdk-api-and-source-review",
+      "contract": "Global legacy updates consume SDK point documents through Mapping-compatible ID/name/type reads and the real legacy-action validator, preserving the ETag. Missing exact IDs and mismatched route names fail before saves.",
       "tests": [
-        "functional_tests/test_m365_cosmos_sdk_contract.py::test_real_sdk_point_documents_are_wrapped_but_nested_json_and_query_rows_are_plain"
-      ]
+        "functional_tests/test_plugin_routes_cosmos_contract.py::test_global_legacy_update_consumes_sdk_point_response"
+      ],
+      "evidence_scope": "owner-runtime"
     },
     "application/single_app/route_backend_plugins.py::test_cosmos_connection": {
       "boundary": {
@@ -23785,14 +23782,14 @@
           }
         ],
         "batches": [],
-        "module_sha256": "c0b926e2dd2d7fed9999a9136c944270753cba204c11539937ef49ebe9cad9ed"
+        "module_sha256": "5f329bcdcdf3d94275191de8c5b2e9d4eca3488dd8f280d2e387f9d5467df5ab"
       },
       "classification": "sdk-compatible",
-      "contract": "test_cosmos_connection: query rows/pages are plain JSON, not Cosmos point wrappers. Nested domain shape checks therefore need no point-response repair.",
-      "evidence_scope": "sdk-api-and-source-review",
+      "contract": "The connection check consumes query rows only to verify connectivity and collect lazy response-hook charge headers. Empty and scalar-ID query results are supported; no point wrapper enters a strict domain sink.",
       "tests": [
-        "functional_tests/test_m365_cosmos_sdk_contract.py::test_real_sdk_point_documents_are_wrapped_but_nested_json_and_query_rows_are_plain"
-      ]
+        "functional_tests/test_plugin_routes_cosmos_contract.py::test_cosmos_connection_consumes_lazy_query_results"
+      ],
+      "evidence_scope": "owner-runtime"
     },
     "application/single_app/route_backend_public_documents.py::register_route_backend_public_documents.api_list_public_documents": {
       "boundary": {
diff --git a/functional_tests/test_plugin_routes_cosmos_contract.py b/functional_tests/test_plugin_routes_cosmos_contract.py
new file mode 100644
index 000000000..9eb316b7f
--- /dev/null
+++ b/functional_tests/test_plugin_routes_cosmos_contract.py
@@ -0,0 +1,201 @@
+# test_plugin_routes_cosmos_contract.py
+"""Runtime evidence for the plugin route owners' Cosmos response contracts.
+
+Version: 0.261.353
+Implemented in: 0.261.352
+
+Execute unchanged route and legacy-validator definitions with the installed SDK's
+CosmosDict point responses and offline external services. This covers the owners
+whose module fingerprints changed with PR #1787's SQL diagnostics.
+"""
+
+import ast
+import builtins
+from copy import deepcopy
+import logging
+import os
+import re
+from types import SimpleNamespace
+from unittest.mock import Mock
+
+from azure.cosmos import exceptions
+from azure.cosmos._cosmos_responses import CosmosDict
+from flask import Blueprint, Flask, current_app, jsonify, request
+import pytest
+
+from functions_action_manifest import McpConfigurationError, is_retired_mcp_stdio, resolve_action_type
+from test_support.agent_delegation import APP_ROOT, execute_functions
+from test_support.m365 import CosmosContainer
+from test_support.offline_bootstrap import OfflineCosmos
+
+
+@pytest.fixture
+def plugin_routes(monkeypatch):
+    app = Flask(__name__)
+    app.config["TESTING"] = True
+    namespace = {
+        "__name__": __name__,
+        "bpap": Blueprint("plugin_cosmos_contract", __name__),
+        "jsonify": jsonify, "request": request, "current_app": current_app,
+        "logging": logging, "os": os, "re": re, "builtins": builtins,
+        "login_required": lambda function: function,
+        "user_required": lambda function: function,
+        "admin_required": lambda function: function,
+        "enabled_required": lambda *args: lambda function: function,
+        "swagger_route": lambda **kwargs: lambda function: function,
+        "get_auth_security": lambda: [],
+        "get_current_user_id": lambda: "actor",
+        "log_event": Mock(),
+        "log_action_update": Mock(),
+        "log_action_creation": Mock(),
+        "log_action_deletion": Mock(),
+        "azure_cosmos": SimpleNamespace(exceptions=exceptions),
+        "CosmosHttpResponseError": exceptions.CosmosHttpResponseError,
+        "McpConfigurationError": McpConfigurationError,
+        "is_retired_mcp_stdio": is_retired_mcp_stdio,
+        "resolve_action_type": resolve_action_type,
+    }
+    # Load the real validator and its constants without unrelated app bootstrap.
+    tree = ast.parse((APP_ROOT / "json_schema_validation.py").read_text(encoding="utf-8"))
+    validator_names = {
+        "LegacyActionCreationError", "LEGACY_ACTION_CREATION_MESSAGE",
+        "ACTION_MIGRATION_ID_PREFIX", "is_legacy_msgraph_type", "validate_legacy_action_update",
+    }
+    definitions = [
+        node for node in tree.body
+        if getattr(node, "name", None) in validator_names
+        or (
+            isinstance(node, ast.Assign)
+            and any(isinstance(target, ast.Name) and target.id in validator_names for target in node.targets)
+        )
+    ]
+    exec(compile(ast.Module(body=definitions, type_ignores=[]), "json_schema_validation.py", "exec"), namespace)
+    namespace["validate_legacy_action_update"] = Mock(wraps=namespace["validate_legacy_action_update"])
+    execute_functions(
+        "route_backend_plugins.py", {"set_user_plugins", "edit_plugin", "test_cosmos_connection"}, namespace,
+    )
+    monkeypatch.setattr(builtins, "kernel_reload_needed", False, raising=False)
+    app.register_blueprint(namespace["bpap"])
+    return app.test_client(), namespace
+
+
+@pytest.mark.parametrize("owner", ["actor", "another-user", None])
+def test_personal_legacy_update_consumes_sdk_point_response(plugin_routes, owner):
+    client, namespace = plugin_routes
+    container = CosmosContainer(partition_field="user_id")
+    action = {"id": "action-id", "name": "historical", "type": "msgraph", "user_id": "actor"}
+    stored = None
+    if owner is not None:
+        stored = container.create_item({**action, "user_id": owner})
+        assert isinstance(stored, CosmosDict)
+    saved = Mock()
+    namespace.update({
+        "cosmos_personal_actions_container": container,
+        "get_global_actions": lambda: [],
+        "get_personal_actions": lambda *args, **kwargs: [deepcopy(action)],
+        "list_legacy_personal_actions": lambda *args: [],
+        "SecretReturnType": SimpleNamespace(NAME="name"),
+        "LEGACY_ACTION_PREFIX": "legacy-action-",
+        "_prepare_personal_action_payload": lambda *args, **kwargs: (deepcopy(action), None),
+        "save_personal_action": saved,
+    })
+
+    response = client.post("/api/user/plugins", json=[action])
+    existing = namespace["validate_legacy_action_update"].call_args.args[1]
+
+    if owner == "actor":
+        assert response.status_code == 200
+        assert response.json == {"success": True}
+        assert isinstance(stored, CosmosDict)
+        assert isinstance(existing, CosmosDict)
+        assert existing["_etag"] == stored["_etag"]
+        assert type(existing["user_id"]) is str
+        saved.assert_called_once_with("actor", action)
+    else:
+        assert response.status_code == 400
+        assert existing is None
+        saved.assert_not_called()
+
+
+@pytest.mark.parametrize("stored_name", ["historical", "different-name", None])
+def test_global_legacy_update_consumes_sdk_point_response(plugin_routes, stored_name):
+    client, namespace = plugin_routes
+    container = CosmosContainer(partition_field="id")
+    action = {"id": "action-id", "name": "historical", "type": "msgraph"}
+    stored = None
+    if stored_name is not None:
+        stored = container.create_item({**action, "name": stored_name})
+        assert isinstance(stored, CosmosDict)
+    saved = Mock()
+    namespace.update({
+        "cosmos_global_actions_container": container,
+        "get_global_actions": lambda **kwargs: [deepcopy(action)],
+        "_apply_plugin_runtime_defaults": lambda action: None,
+        "apply_plugin_validation_defaults": lambda action: action,
+        "discover_plugin_types": lambda **kwargs: ["msgraph"],
+        "validate_plugin": lambda action: None,
+        "PluginHealthChecker": SimpleNamespace(validate_plugin_manifest=lambda *args: (True, [])),
+        "_enforce_mcp_destination_policy": lambda *args, **kwargs: None,
+        "WORKSPACE_IDENTITY_SCOPE_GLOBAL": "global",
+        "get_merged_plugin_settings": lambda *args: {},
+        "_validate_action_identity_for_scope": lambda *args: None,
+        "save_global_action": saved,
+        "_redact_plugin_for_logging": lambda action: {"type": action["type"]},
+    })
+
+    response = client.put("/api/admin/plugins/historical", json=action)
+    existing = namespace["validate_legacy_action_update"].call_args.args[1]
+
+    if stored_name == "historical":
+        assert response.status_code == 200
+        assert response.json == {"success": True}
+        assert isinstance(stored, CosmosDict)
+        assert isinstance(existing, CosmosDict)
+        assert existing["_etag"] == stored["_etag"]
+        saved.assert_called_once_with({**action, "metadata": {}, "additionalFields": {}}, user_id="actor")
+    else:
+        assert response.status_code == (404 if stored_name else 400)
+        if stored_name:
+            assert isinstance(existing, CosmosDict)
+        else:
+            assert existing is None
+        saved.assert_not_called()
+
+
+@pytest.mark.parametrize("rows", [[], ["existing-document-id"]])
+def test_cosmos_connection_consumes_lazy_query_results(plugin_routes, monkeypatch, rows):
+    client, namespace = plugin_routes
+    cosmos = OfflineCosmos()
+    container = cosmos.database.get_container_client("records")
+    consumed = []
+
+    def query_items(*, query, enable_cross_partition_query, max_item_count, response_hook):
+        assert query == "SELECT TOP 1 VALUE c.id FROM c"
+        assert enable_cross_partition_query is True
+        assert max_item_count == 1
+        yield from rows
+        response_hook({"x-ms-request-charge": "1.5"}, None)
+        consumed.append(True)
+
+    monkeypatch.setattr(container, "query_items", query_items)
+    constructor = Mock(return_value=cosmos)
+    credential = object()
+    namespace.update({
+        "azure_cosmos": SimpleNamespace(CosmosClient=constructor, exceptions=exceptions),
+        "DefaultAzureCredential": lambda: credential,
+        "validate_azure_cosmos_endpoint": lambda endpoint: endpoint,
+        "_load_existing_plugin_for_test": lambda *args, **kwargs: None,
+    })
+
+    response = client.post("/api/plugins/test-cosmos-connection", json={
+        "endpoint": "https://offline.documents.azure.com",
+        "database_name": "database", "container_name": "records", "timeout": 100,
+    })
+
+    assert response.status_code == 200
+    assert response.json["success"] is True
+    assert consumed == [True]
+    constructor.assert_called_once_with(
+        "https://offline.documents.azure.com", credential=credential, timeout=30, connection_timeout=30,
+    )
+    assert namespace["log_event"].call_args.kwargs["extra"]["request_charge"] == "1.5"
diff --git a/functional_tests/test_sql_connection_diagnostics.py b/functional_tests/test_sql_connection_diagnostics.py
new file mode 100644
index 000000000..26c102c3e
--- /dev/null
+++ b/functional_tests/test_sql_connection_diagnostics.py
@@ -0,0 +1,159 @@
+# test_sql_connection_diagnostics.py
+"""
+Functional tests for credential-safe SQL connection diagnostics.
+Version: 0.261.353
+Implemented in: 0.261.351
+
+Validate structured evidence, conservative classification, privacy, and cold import
+behavior without SQL drivers, application bootstrap, Azure access, or network I/O.
+"""
+
+import importlib
+import json
+import os
+from pathlib import Path
+import subprocess
+import sys
+
+import pytest
+
+
+APP_ROOT = Path(__file__).resolve().parents[1] / "application" / "single_app"
+SECRET = "private-credential-and-host.example.test"
+
+
+@pytest.fixture
+def diagnostics(monkeypatch):
+    monkeypatch.syspath_prepend(str(APP_ROOT))
+    return importlib.import_module("semantic_kernel_plugins.sql_connection_diagnostics")
+
+
+@pytest.mark.parametrize("engine,state,detail,expected,native", [
+    ("sqlserver", "28000", f"[28000] {SECRET} (18456) (SQLDriverConnect)", "sql_authentication_failed", 18456),
+    ("azure_sql", "HYT00", SECRET, "sql_connection_timeout", None),
+    ("azuresql", "IM002", SECRET, "sql_driver_unavailable", None),
+    ("sqlserver", "IM003", SECRET, "sql_driver_unavailable", None),
+    ("sqlserver", "08001", f"[08001] {SECRET} (10060) (SQLDriverConnect)", "sql_connection_failed", 10060),
+    ("sqlserver", "08001", f"[08001] {SECRET} (18456) (SQLDriverConnect)", "sql_authentication_failed", 18456),
+    ("sqlserver", "HY000", SECRET, "sql_connection_failed", None),
+])
+def test_interface_error_uses_evidence_not_class_name(diagnostics, engine, state, detail, expected, native):
+    error = type("InterfaceError", (Exception,), {})(state, detail)
+    result = diagnostics.classify_sql_connection_error(error, engine)
+    assert result.code == expected
+    assert result.engine == "sqlserver"
+    assert result.sqlstate == state
+    assert result.native_error == native
+    assert result.error_type == "InterfaceError"
+    assert SECRET not in json.dumps(result.log_properties())
+    assert SECRET not in str(diagnostics.SqlConnectionError(result))
+
+
+@pytest.mark.parametrize("engine,attributes,args,expected,native,state", [
+    ("postgresql", {"pgcode": "28P01"}, (SECRET,), "sql_authentication_failed", None, "28P01"),
+    ("postgresql", {"sqlstate": "08006"}, (SECRET,), "sql_connection_failed", None, "08006"),
+    ("mysql", {}, (1045, SECRET), "sql_authentication_failed", 1045, None),
+    ("mysql", {}, (2003, SECRET), "sql_connection_failed", 2003, None),
+    ("sqlite", {"sqlite_errorcode": 14}, (SECRET,), "sql_connection_failed", 14, None),
+    ("sqlite", {"sqlite_errorcode": 2062}, (SECRET,), "sql_connection_failed", 2062, None),
+])
+def test_other_provider_structured_fields(diagnostics, engine, attributes, args, expected, native, state):
+    error = type("OperationalError", (Exception,), attributes)(*args)
+    result = diagnostics.classify_sql_connection_error(error, engine)
+    assert (result.code, result.native_error, result.sqlstate) == (expected, native, state)
+    assert SECRET not in repr(result)
+
+
+@pytest.mark.parametrize("error,code", [
+    (ImportError(SECRET), "sql_driver_unavailable"),
+    (ModuleNotFoundError(SECRET), "sql_driver_unavailable"),
+    (TimeoutError(SECRET), "sql_connection_timeout"),
+    (ConnectionError(SECRET), "sql_connection_failed"),
+    (type("InterfaceError", (Exception,), {})(SECRET), "sql_connection_failed"),
+])
+def test_python_errors_are_safely_classified(diagnostics, error, code):
+    result = diagnostics.classify_sql_connection_error(error, "sqlserver")
+    assert result.code == code
+    assert result.sqlstate is None
+    assert result.native_error is None
+    assert SECRET not in result.message
+
+
+@pytest.mark.parametrize("args,attributes", [
+    ((), {}),
+    ((["28000", SECRET],), {}),
+    (({"sqlstate": "28000", "password": SECRET},), {}),
+    ((SECRET * 10000,), {}),
+    (("TOKEN",), {"pgcode": "TOKEN", "sqlstate": ["28000"]}),
+    (("28000", f"[28000] {SECRET}:1433 id=18456"), {}),
+    (("28000", f"{SECRET} (18456) (SQLDriverConnect)"), {}),
+    (("28000", f"[28000] {SECRET} (99999999999) (SQLDriverConnect)"), {}),
+    (("28000", f"[28000] {SECRET * 1000} (18456) (SQLDriverConnect)"), {}),
+])
+def test_unstructured_or_malformed_arguments_are_not_retained(diagnostics, args, attributes):
+    error = type("UntrustedProviderError", (Exception,), attributes)(*args)
+    result = diagnostics.classify_sql_connection_error(error, "sqlserver")
+    assert result.native_error is None
+    assert result.error_type == "Exception"
+    assert SECRET not in repr(result)
+    assert "TOKEN" not in repr(result)
+
+
+@pytest.mark.parametrize("value", [True, False, 2**40, -(2**40), "1045", [1045], {"errno": 1045}, None])
+def test_invalid_native_codes_are_rejected(diagnostics, value):
+    error = Exception(value, SECRET)
+    error.sqlite_errorcode = value
+    mysql = diagnostics.classify_sql_connection_error(error, "mysql")
+    sqlite = diagnostics.classify_sql_connection_error(error, "sqlite")
+    assert mysql.native_error is None
+    assert sqlite.native_error is None
+
+
+def test_unknown_engine_cannot_supply_provider_evidence(diagnostics):
+    result = diagnostics.classify_sql_connection_error(Exception("28000", SECRET), SECRET)
+    assert result.code == "sql_connection_failed"
+    assert result.engine == "unknown"
+    assert result.sqlstate is None
+
+
+def test_diagnostic_helper_is_not_a_user_facing_action(monkeypatch):
+    monkeypatch.syspath_prepend(str(APP_ROOT.parents[1] / "scripts"))
+    inventory = importlib.import_module("build_docs_inventory")
+    actions = inventory.extract_actions()
+    assert all(action["module"] != "sql_connection_diagnostics.py" for action in actions)
+
+
+def test_cold_import_and_required_operations_survive_optimized_python():
+    probe = r'''
+import importlib.abc
+import socket
+import sys
+
+class BlockExternalImports(importlib.abc.MetaPathFinder):
+    def find_spec(self, fullname, path=None, target=None):
+        if fullname.split(".")[0] in {
+            "config", "functions_settings", "functions_appinsights", "azure",
+            "pyodbc", "psycopg2", "pymysql",
+        }:
+            raise RuntimeError("Diagnostic module crossed bootstrap boundary: " + fullname)
+
+sys.meta_path.insert(0, BlockExternalImports())
+def no_network(*args, **kwargs):
+    raise RuntimeError("Network access forbidden")
+socket.socket = no_network
+socket.getaddrinfo = no_network
+from semantic_kernel_plugins.sql_connection_diagnostics import classify_sql_connection_error
+result = classify_sql_connection_error(Exception("HYT00", "private data"), "azure_sql")
+if result.code != "sql_connection_timeout" or result.engine != "sqlserver":
+    raise AssertionError("Required diagnostic operation failed")
+if result.log_properties()["sqlstate"] != "HYT00":
+    raise AssertionError("Required diagnostic evidence was lost")
+'''
+    env = {**os.environ, "PYTHONPATH": str(APP_ROOT), "PYTHONDONTWRITEBYTECODE": "1"}
+    for flags in ([], ["-O"]):
+        completed = subprocess.run(
+            [sys.executable, *flags, "-c", probe], env=env, capture_output=True,
+            text=True, timeout=30, check=False,
+        )
+        if completed.returncode != 0:
+            raise AssertionError(completed.stdout + completed.stderr)
diff --git a/functional_tests/test_sql_connection_diagnostics_runtime.py b/functional_tests/test_sql_connection_diagnostics_runtime.py
new file mode 100644
index 000000000..1928b7e71
--- /dev/null
+++ b/functional_tests/test_sql_connection_diagnostics_runtime.py
@@ -0,0 +1,462 @@
+# test_sql_connection_diagnostics_runtime.py
+"""
+Functional tests for SQL connection diagnostics across routes, plugins, and logging.
+Version: 0.261.353
+Implemented in: 0.261.351
+
+Run real plugin classes, connection builders, Flask handlers, and log_event against
+isolated provider seams. No SQL, Cosmos, Azure Monitor, or network calls are allowed.
+"""
+
+import builtins
+from copy import deepcopy
+import importlib.util
+import json
+import logging
+from pathlib import Path
+import socket
+import sys
+from types import SimpleNamespace
+import traceback
+from unittest.mock import Mock
+
+import pytest
+
+from test_support.agent_delegation import module_stub
+from test_workspace_authoring_backend import environment  # noqa: F401
+
+
+APP_ROOT = Path(__file__).resolve().parents[1] / "application" / "single_app"
+SECRET = "private-provider-credential"
+
+
+def load_module(monkeypatch, name, filename):
+    spec = importlib.util.spec_from_file_location(name, APP_ROOT / filename)
+    module = importlib.util.module_from_spec(spec)
+    monkeypatch.setitem(sys.modules, name, module)
+    spec.loader.exec_module(module)
+    return module
+
+
+@pytest.fixture(autouse=True)
+def block_network(monkeypatch):
+    def blocked(*args, **kwargs):
+        raise AssertionError("SQL diagnostic tests must not contact live services.")
+
+    monkeypatch.setattr(socket.socket, "connect", blocked)
+    monkeypatch.setattr(socket, "getaddrinfo", blocked)
+
+
+@pytest.fixture
+def runtime(monkeypatch):
+    monkeypatch.syspath_prepend(str(APP_ROOT))
+    monkeypatch.setitem(sys.modules, "azure.monitor.opentelemetry", module_stub(
+        "azure.monitor.opentelemetry", configure_azure_monitor=Mock(
+            side_effect=AssertionError("Exporter configuration is not allowed in these tests."),
+        ),
+    ))
+    settings = {"enable_debug_logging": False}
+    monkeypatch.setitem(sys.modules, "app_settings_cache", module_stub(
+        "app_settings_cache", get_settings_cache=lambda: settings,
+    ))
+    diagnostics = load_module(
+        monkeypatch, "semantic_kernel_plugins.sql_connection_diagnostics",
+        r"semantic_kernel_plugins\sql_connection_diagnostics.py",
+    )
+    logger_module = load_module(monkeypatch, "functions_appinsights", "functions_appinsights.py")
+    records = []
+
+    class Capture(logging.Handler):
+        def emit(self, record):
+            records.append(record)
+
+    logger = logging.Logger("sql-connection-test", level=logging.DEBUG)
+    logger.addHandler(Capture())
+    logger_module._appinsights_logger = logger
+    monkeypatch.setitem(sys.modules, "functions_debug", module_stub(
+        "functions_debug", debug_print=logger_module.debug_print,
+    ))
+    monkeypatch.setitem(sys.modules, "semantic_kernel.functions", module_stub(
+        "semantic_kernel.functions", kernel_function=lambda **kwargs: lambda function: function,
+    ))
+    monkeypatch.setitem(sys.modules, "semantic_kernel_plugins.plugin_invocation_logger", module_stub(
+        "semantic_kernel_plugins.plugin_invocation_logger",
+        plugin_function_logger=lambda *args: lambda function: function,
+    ))
+    odbc = load_module(
+        monkeypatch, "semantic_kernel_plugins.sql_odbc_utils",
+        r"semantic_kernel_plugins\sql_odbc_utils.py",
+    )
+    query = load_module(
+        monkeypatch, "semantic_kernel_plugins.sql_query_plugin",
+        r"semantic_kernel_plugins\sql_query_plugin.py",
+    )
+    schema = load_module(
+        monkeypatch, "semantic_kernel_plugins.sql_schema_plugin",
+        r"semantic_kernel_plugins\sql_schema_plugin.py",
+    )
+    return SimpleNamespace(
+        query=query.SQLQueryPlugin, schema=schema.SQLSchemaPlugin,
+        diagnostics=diagnostics, logging=logger_module, records=records,
+        settings=settings, odbc=odbc,
+    )
+
+
+def plugin_manifest(engine="sqlserver", auth_type="managed_identity", connection_string=None):
+    return {
+        "name": "sql_test", "auth": {"type": auth_type},
+        "additionalFields": {
+            "database_type": engine, "server": "db.example.test", "database": "test-db",
+            "username": "sql-user" if auth_type == "username_password" else None,
+            "password": SECRET if auth_type == "username_password" else None,
+            "auth_type": auth_type, "connection_string": connection_string, "timeout": 7,
+        },
+    }
+
+
+@pytest.mark.parametrize("kind", ["query", "schema"])
+@pytest.mark.parametrize("engine,module,error,expected", [
+    ("sqlserver", "pyodbc", Exception("28000", f"[28000] {SECRET} (18456) (SQLDriverConnect)"), "sql_authentication_failed"),
+    ("azure_sql", "pyodbc", Exception("HYT00", SECRET), "sql_connection_timeout"),
+    ("postgresql", "psycopg2", type("OperationalError", (Exception,), {"pgcode": "28P01"})(SECRET), "sql_authentication_failed"),
+    ("mysql", "pymysql", Exception(1045, SECRET), "sql_authentication_failed"),
+    ("sqlite", "sqlite3", Exception(SECRET), "sql_connection_failed"),
+])
+@pytest.mark.parametrize("debug", [False, True])
+def test_plugins_log_safe_structured_connection_failures(runtime, monkeypatch, capsys, kind, engine, module, error, expected, debug):
+    runtime.settings["enable_debug_logging"] = debug
+    connect = Mock(side_effect=error)
+    monkeypatch.setitem(sys.modules, module, module_stub(module, connect=connect))
+    plugin = getattr(runtime, kind)(plugin_manifest(engine))
+    runtime.records.clear()
+    capsys.readouterr()
+    with pytest.raises(runtime.diagnostics.SqlConnectionError) as caught:
+        plugin._get_connection()
+    formatted_trace = "".join(traceback.format_exception(caught.value))
+    assert SECRET not in formatted_trace
+    assert caught.value.diagnostic.code == expected
+    assert plugin._connection is None
+    connect.assert_called_once()
+    events = [record for record in runtime.records if getattr(record, "sc_message", "").startswith("[SQL_CONNECTION]")]
+    assert len(events) == 1
+    event = events[0]
+    assert event.levelno == logging.WARNING
+    assert event.sc_failure_code == expected
+    assert event.sc_sql_engine == ("sqlserver" if engine == "azure_sql" else engine)
+    assert event.sc_operation == f"sql_{kind}_connect"
+    assert SECRET not in json.dumps([record.__dict__ for record in runtime.records], default=str)
+    assert SECRET not in capsys.readouterr().out
+
+
+@pytest.mark.parametrize("method,args", [
+    ("execute_query", ("SELECT 1",)),
+    ("execute_scalar", ("SELECT 1",)),
+    ("query_database", ("Read a value", "SELECT 1")),
+])
+def test_query_failure_results_have_safe_code_without_provider_text(runtime, monkeypatch, method, args):
+    monkeypatch.setitem(sys.modules, "pyodbc", module_stub(
+        "pyodbc", connect=Mock(side_effect=Exception("28000", SECRET)),
+    ))
+    plugin = runtime.query(plugin_manifest())
+    result = getattr(plugin, method)(*args)
+    assert result.data["error_code"] == "sql_authentication_failed"
+    assert "error" in result.data
+    assert SECRET not in json.dumps(result.data)
+    assert SECRET not in json.dumps(result.metadata)
+    assert "sqlstate" not in result.data
+
+
+def test_schema_failure_results_and_raised_errors_are_safe(runtime, monkeypatch):
+    connect = Mock(side_effect=Exception("IM002", SECRET))
+    monkeypatch.setitem(sys.modules, "pyodbc", module_stub("pyodbc", connect=connect))
+    plugin = runtime.schema(plugin_manifest())
+    result = plugin.get_database_schema()
+    assert result.data["error_code"] == "sql_driver_unavailable"
+    assert result.metadata == {"source": "sql_schema_plugin", "success": False}
+    assert SECRET not in json.dumps(result.data)
+    for method, args in [
+        ("get_table_list", ()), ("get_table_schema", ("Example",)), ("get_relationships", ()),
+    ]:
+        with pytest.raises(runtime.diagnostics.SqlConnectionError) as caught:
+            getattr(plugin, method)(*args)
+        assert SECRET not in str(caught.value)
+
+
+@pytest.mark.parametrize("kind", ["query", "schema"])
+def test_driver_import_failures_do_not_escape_as_raw_import_errors(runtime, monkeypatch, kind):
+    original_import = builtins.__import__
+
+    def missing_driver(name, *args, **kwargs):
+        if name == "pyodbc":
+            raise ImportError(SECRET)
+        return original_import(name, *args, **kwargs)
+
+    plugin = getattr(runtime, kind)(plugin_manifest())
+    with monkeypatch.context() as scope:
+        scope.setattr(builtins, "__import__", missing_driver)
+        with pytest.raises(runtime.diagnostics.SqlConnectionError) as caught:
+            plugin._get_connection()
+    assert caught.value.diagnostic.code == "sql_driver_unavailable"
+    assert SECRET not in str(caught.value)
+
+
+@pytest.mark.parametrize("kind", ["query", "schema"])
+@pytest.mark.parametrize("auth,connection_string,expected_suffix", [
+    ("managed_identity", None, ";Authentication=ActiveDirectoryMsi"),
+    ("integrated", None, ";Trusted_Connection=yes"),
+    ("username_password", None, f";UID=sql-user;PWD={SECRET}"),
+    ("connection_string", f"DRIVER={{ODBC Driver 18 for SQL Server}};SERVER=other;PWD={SECRET}", None),
+])
+def test_auth_inputs_and_timeout_are_unchanged(runtime, monkeypatch, kind, auth, connection_string, expected_suffix):
+    connection = Mock()
+    connect = Mock(return_value=connection)
+    monkeypatch.setitem(sys.modules, "pyodbc", module_stub("pyodbc", connect=connect))
+    plugin = getattr(runtime, kind)(plugin_manifest(auth_type=auth, connection_string=connection_string))
+    result = plugin._get_connection()
+    again = plugin._get_connection()
+    assert result is connection and again is connection
+    connect.assert_called_once()
+    actual = connect.call_args.args[0]
+    if connection_string:
+        assert actual == connection_string
+    else:
+        assert actual == "DRIVER={ODBC Driver 18 for SQL Server};SERVER=db.example.test;DATABASE=test-db" + expected_suffix
+    assert connect.call_args.kwargs == ({"timeout": 7} if kind == "query" else {})
+
+
+@pytest.mark.parametrize("kind", ["query", "schema"])
+def test_driver_fallback_remains_bounded_and_non_driver_errors_do_not_retry(runtime, monkeypatch, kind):
+    saved = "DRIVER={ODBC Driver 17 for SQL Server};SERVER=db.example.test;DATABASE=test-db"
+    connect = Mock(side_effect=[
+        Exception("01000", "[unixODBC][Driver Manager]Can't open lib 'ODBC Driver 17 for SQL Server' : file not found"),
+        Exception("28000", SECRET),
+    ])
+    monkeypatch.setitem(sys.modules, "pyodbc", module_stub("pyodbc", connect=connect))
+    plugin = getattr(runtime, kind)(plugin_manifest(connection_string=saved))
+    with pytest.raises(runtime.diagnostics.SqlConnectionError) as caught:
+        plugin._get_connection()
+    assert caught.value.diagnostic.code == "sql_authentication_failed"
+    assert connect.call_count == 2
+    assert "ODBC Driver 17" in connect.call_args_list[0].args[0]
+    assert "ODBC Driver 18" in connect.call_args_list[1].args[0]
+    assert connect.call_args_list[0].kwargs == connect.call_args_list[1].kwargs
+    connect.reset_mock(side_effect=True)
+    connect.side_effect = Exception("28000", SECRET)
+    with pytest.raises(runtime.diagnostics.SqlConnectionError):
+        plugin._get_connection()
+    connect.assert_called_once()
+
+
+def test_successful_query_scalar_and_schema_shapes_are_unchanged(runtime, monkeypatch):
+    cursor = Mock()
+    cursor.description = [("number",)]
+    cursor.fetchmany.return_value = [(1,)]
+    cursor.fetchone.return_value = (1,)
+    cursor.fetchall.return_value = []
+    connection = Mock()
+    connection.cursor.return_value = cursor
+    monkeypatch.setitem(sys.modules, "pyodbc", module_stub("pyodbc", connect=Mock(return_value=connection)))
+    query = runtime.query(plugin_manifest())
+    rows = query.execute_query("SELECT 1")
+    scalar = query.execute_scalar("SELECT 1")
+    answer = query.query_database("Read a value", "SELECT 1")
+    schema = runtime.schema(plugin_manifest()).get_database_schema()
+    assert rows.data == {
+        "columns": ["number"], "data": [{"number": 1}], "row_count": 1,
+        "is_truncated": False, "query": "SELECT 1;",
+    }
+    assert scalar.data == {"value": 1, "query": "SELECT 1;"}
+    assert answer.data == {**rows.data, "question": "Read a value"}
+    assert schema.data == {
+        "database_type": "sqlserver", "database_name": "test-db", "tables": {}, "relationships": [],
+    }
+
+
+@pytest.mark.parametrize("debug", [False, True])
+def test_real_logger_preserves_valid_evidence_and_rejects_invalid_values(runtime, capsys, debug):
+    runtime.settings["enable_debug_logging"] = debug
+    diagnostic = runtime.diagnostics.classify_sql_connection_error(
+        Exception("28000", f"[28000] {SECRET} (18456) (SQLDriverConnect)"), "azure_sql",
+    )
+    runtime.logging.log_event(
+        "[SQL_CONNECTION] SQL connection failed.", level=logging.WARNING,
+        extra={**diagnostic.log_properties(), "operation": "sql_connection_test"},
+    )
+    event = runtime.records[-1]
+    assert event.sc_sqlstate == "28000"
+    assert event.sc_sql_native_error == 18456
+    assert event.sc_sql_engine == "sqlserver"
+    assert event.sc_failure_code == "sql_authentication_failed"
+    assert SECRET not in json.dumps(event.__dict__, default=str)
+    assert SECRET not in capsys.readouterr().out
+    for state, engine, native in [
+        (SECRET, SECRET, SECRET), ("TOKEN", {}, True), ("28000\n", ["sqlserver"], 2**40),
+    ]:
+        properties = runtime.logging._build_logger_extra("[SQL_CONNECTION]", {
+            "sqlstate": state, "sql_engine": engine, "sql_native_error": native,
+        })
+        assert properties == {"sc_message": "[SQL_CONNECTION]", "sc_message_length": 16}
+
+
+@pytest.mark.parametrize("editor", [False, True])
+@pytest.mark.parametrize("saved", [False, True])
+@pytest.mark.parametrize("engine,module,error,expected", [
+    ("azure_sql", "pyodbc", Exception("28000", SECRET), "sql_authentication_failed"),
+    ("postgresql", "psycopg2", TimeoutError(SECRET), "sql_connection_timeout"),
+    ("mysql", "pymysql", Exception(1045, SECRET), "sql_authentication_failed"),
+    ("sqlite", "sqlite3", ImportError(SECRET), "sql_driver_unavailable"),
+])
+def test_sql_endpoint_has_shared_safe_errors_for_classic_and_editor(environment, monkeypatch, editor, saved, engine, module, error, expected):
+    env = environment
+    fields = {
+        "database_type": engine, "server": "db.example.test", "database": "test-db",
+        "auth_type": "managed_identity",
+    }
+    payload = {**fields, "auth_type": "managed_identity", "action_scope": "personal"}
+    if editor:
+        payload.update({"type": "sql_query", "auth": {"type": "identity", "identity": "managed_identity"}, "additionalFields": fields})
+    if saved:
+        env.services.add("actions", "personal", "actor", {
+            "id": "saved-sql", "name": "saved_sql", "user_id": "actor", "type": "sql_query",
+            "auth": {"type": "identity", "identity": "managed_identity"}, "additionalFields": fields,
+        })
+        payload["existing_plugin"] = {"scope": "personal", "id": "saved-sql", "name": "saved_sql"}
+    original = deepcopy(env.services.records)
+    connect = Mock(side_effect=error)
+    monkeypatch.setitem(sys.modules, module, module_stub(module, connect=connect))
+    response = env.client.post("/api/plugins/test-sql-connection", json=payload)
+    assert response.status_code == 400
+    body = response.get_json()
+    assert body["success"] is False
+    assert body["error_code"] == expected
+    assert SECRET not in json.dumps(body)
+    assert "sqlstate" not in body and "sql_native_error" not in body
+    connect.assert_called_once()
+    events = [call for call in env.services.appinsights.log_event.call_args_list if call.args[0].startswith("[SQL_CONNECTION]")]
+    assert len(events) == 1
+    assert events[0].kwargs["level"] == logging.WARNING
+    assert events[0].kwargs["extra"]["failure_code"] == expected
+    assert env.services.records == original
+    assert env.services.writes == []
+
+
+@pytest.mark.parametrize("editor", [False, True])
+@pytest.mark.parametrize("engine,module,message", [
+    ("sqlserver", "pyodbc", "Successfully connected to test-db on db.example.test."),
+    ("postgresql", "psycopg2", "Successfully connected to PostgreSQL database test-db."),
+    ("mysql", "pymysql", "Successfully connected to MySQL database test-db."),
+    ("sqlite", "sqlite3", "Successfully connected to SQLite database."),
+])
+def test_sql_endpoint_success_shape_and_provider_inputs_are_unchanged(environment, monkeypatch, editor, engine, module, message):
+    env = environment
+    fields = {
+        "database_type": engine, "server": "db.example.test", "database": "test-db",
+        "auth_type": "username_password", "username": "sql-user", "password": SECRET,
+        "timeout": 99,
+    }
+    payload = {**fields, "action_scope": "personal"}
+    if editor:
+        payload.update({"type": "sql_query", "auth": {"type": "user"}, "additionalFields": fields})
+    connection = Mock()
+    connect = Mock(return_value=connection)
+    monkeypatch.setitem(sys.modules, module, module_stub(module, connect=connect))
+    response = env.client.post("/api/plugins/test-sql-connection", json=payload)
+    assert response.status_code == 200
+    assert response.get_json() == {"success": True, "message": message}
+    connect.assert_called_once()
+    if engine == "sqlserver":
+        assert connect.call_args.args == (
+            f"DRIVER={{ODBC Driver 18 for SQL Server}};SERVER=db.example.test;DATABASE=test-db;UID=sql-user;PWD={SECRET}",
+        )
+        assert connect.call_args.kwargs == {"timeout": 15}
+    elif engine == "sqlite":
+        assert connect.call_args.args == ("test-db",)
+        assert connect.call_args.kwargs == {"timeout": 15}
+    else:
+        assert connect.call_args.args == ()
+        assert connect.call_args.kwargs == {
+            "host": "db.example.test", "database": "test-db", "connect_timeout": 15,
+            "user": "sql-user", "password": SECRET,
+        }
+    connection.cursor.return_value.execute.assert_called_once_with("SELECT 1")
+    connection.cursor.return_value.close.assert_called_once()
+    connection.close.assert_called_once()
+    assert env.services.writes == []
+
+
+def test_sql_endpoint_reusable_identity_and_real_logger_share_contract(environment, runtime, monkeypatch):
+    env = environment
+
+    def hydrate(manifest, scope_type, scope_id, *, return_type):
+        assert (scope_type, scope_id) == ("personal", "actor")
+        result = deepcopy(manifest)
+        result["auth"] = {"type": "identity", "identity": "managed_identity"}
+        result["additionalFields"]["auth_type"] = "managed_identity"
+        return result
+
+    env.identities.hydrate_action_identity_reference.side_effect = hydrate
+    monkeypatch.setitem(env.namespace, "log_event", runtime.logging.log_event)
+    connect = Mock(side_effect=Exception("28000", f"[28000] {SECRET} (18456) (SQLDriverConnect)"))
+    monkeypatch.setitem(sys.modules, "pyodbc", module_stub("pyodbc", connect=connect))
+    response = env.client.post("/api/plugins/test-sql-connection", json={
+        "type": "sql_query", "identity_id": "owned-identity",
+        "auth": {"type": "identity", "identity": "owned-identity"}, "action_scope": "personal",
+        "additionalFields": {
+            "database_type": "azure_sql", "server": "db.example.test", "database": "test-db",
+        },
+    })
+    assert response.status_code == 400
+    assert response.get_json()["error_code"] == "sql_authentication_failed"
+    env.identities.hydrate_action_identity_reference.assert_called_once()
+    assert "Authentication=ActiveDirectoryMsi" in connect.call_args.args[0]
+    event = runtime.records[-1]
+    assert (event.sc_sqlstate, event.sc_sql_native_error, event.sc_sql_engine) == ("28000", 18456, "sqlserver")
+    assert SECRET not in json.dumps(event.__dict__, default=str)
+    assert "owned-identity" not in json.dumps(event.__dict__, default=str)
+
+
+@pytest.mark.parametrize("scope,admin,expected", [
+    ("group", False, 400), ("global", True, 400), ("global", False, 403),
+])
+def test_scoped_identity_authorization_is_unchanged(environment, monkeypatch, scope, admin, expected):
+    env = environment
+    monkeypatch.setitem(env.namespace, "WORKSPACE_IDENTITY_SCOPE_GROUP", "group")
+    monkeypatch.setitem(env.namespace, "WORKSPACE_IDENTITY_SCOPE_GLOBAL", "global")
+    monkeypatch.setitem(env.namespace, "require_active_group", env.services.groups.require_active_group)
+    monkeypatch.setitem(env.namespace, "assert_group_role", env.services.groups.assert_group_role)
+    with env.client.session_transaction() as state:
+        state["user"] = {**state["user"], "roles": ["User", "Admin"] if admin else ["User"]}
+    env.identities.hydrate_action_identity_reference.return_value = {
+        "auth": {"type": "identity", "identity": "managed_identity"},
+        "additionalFields": {"auth_type": "managed_identity"},
+    }
+    connect = Mock(side_effect=Exception("HYT00", SECRET))
+    monkeypatch.setitem(sys.modules, "pyodbc", module_stub("pyodbc", connect=connect))
+    response = env.client.post("/api/plugins/test-sql-connection", json={
+        "action_scope": scope, "identity_id": "scoped-identity", "database_type": "sqlserver",
+        "server": "db.example.test", "database": "test-db",
+    })
+    assert response.status_code == expected
+    if expected == 403:
+        connect.assert_not_called()
+        env.identities.hydrate_action_identity_reference.assert_not_called()
+    else:
+        assert response.get_json()["error_code"] == "sql_connection_timeout"
+        connect.assert_called_once()
+        call = env.identities.hydrate_action_identity_reference.call_args
+        assert call.args[1:3] == (scope, "group-a" if scope == "group" else "global")
+
+
+def test_foreign_personal_identity_is_denied_before_connect(environment, monkeypatch):
+    env = environment
+    env.identities.hydrate_action_identity_reference.side_effect = PermissionError(SECRET)
+    connect = Mock()
+    monkeypatch.setitem(sys.modules, "pyodbc", module_stub("pyodbc", connect=connect))
+    response = env.client.post("/api/plugins/test-sql-connection", json={
+        "identity_id": "foreign", "auth": {"type": "identity", "identity": "foreign"},
+        "additionalFields": {"database_type": "azure_sql", "server": "db.example.test", "database": "test-db"},
+    })
+    assert response.status_code == 403
+    assert SECRET not in json.dumps(response.get_json())
+    connect.assert_not_called()
+    assert env.services.writes == []
diff --git a/functional_tests/test_workspace_authoring_backend.py b/functional_tests/test_workspace_authoring_backend.py
index 482f0627f..79fedbed6 100644
--- a/functional_tests/test_workspace_authoring_backend.py
+++ b/functional_tests/test_workspace_authoring_backend.py
@@ -1,7 +1,7 @@
 # test_workspace_authoring_backend.py
 """Executable contracts for native personal agent/action authoring.
 
-Version: 0.261.122
+Version: 0.261.353
 Implemented in: 0.261.096
 
 Real Flask route definitions, validators, payload normalizers, Key Vault helpers,
@@ -253,6 +253,10 @@ def update_secret_properties(self, name, expires_on=None):
         rocks = _load_module(stack, "semantic_kernel_plugins.rocksdb_plugin", r"semantic_kernel_plugins\rocksdb_plugin.py")
         modules.append(rocks)
         modules.append(_load_module(stack, "semantic_kernel_plugins.sql_odbc_utils", r"semantic_kernel_plugins\sql_odbc_utils.py"))
+        modules.append(_load_module(
+            stack, "semantic_kernel_plugins.sql_connection_diagnostics",
+            r"semantic_kernel_plugins\sql_connection_diagnostics.py",
+        ))
         keyvault = _load_module(stack, "functions_keyvault")
         schema = _load_module(stack, "json_schema_validation")
         modules.extend([
diff --git a/scripts/build_docs_inventory.py b/scripts/build_docs_inventory.py
index ad2b0f442..01f4b13cb 100644
--- a/scripts/build_docs_inventory.py
+++ b/scripts/build_docs_inventory.py
@@ -39,6 +39,7 @@
     "plugin_invocation_thoughts.py",
     "plugin_loader.py",
     "plugin_logging_examples.py",
+    "sql_connection_diagnostics.py",
     "sql_odbc_utils.py",
     "ui_test_plugin.py",
 }