fix: address 9 code-review findings in db timeout, keyring, and diagnostics

- Timeout detection: replace string-matching with pgcode/errno checks
  (psycopg3 pgcode 57014, pymysql errno 3024/1969) to avoid false positives
- Separate set_statement_timeout try-block from query execution try-block
  so config failures are not misclassified as query errors
- Reset statement timeout to 0 in finally block so pooled connections
  don't inherit the limit on subsequent queries
- Log warning when interrupt() fails to stop the SQLite thread
- Improve keyring failure log message; delete old entry before re-saving
- Add _MAX_DIAG_SAMPLE_VALUES constant to replace magic 5 in diagnostics SQL
- Document MySQL MAX_EXECUTION_TIME SELECT-only limitation
- Add 12 new tests covering the above (322 total, all passing)

Author: Adoit

src/open_data_agent/db/connection.py

@@ -7,12 +7,14 @@
 
 from __future__ import annotations
 
+import contextlib
 import logging
 import os
 import re
 from pathlib import Path
 from typing import Any
 
+import keyring
 import yaml
 
 from open_data_agent.config import get_config_dir
@@ -60,43 +62,37 @@ def _load_connections(self) -> dict[str, dict[str, Any]]:
             raise ConfigError(f"Failed to parse connections.yaml: {exc}") from exc
 
     def _save_connections(self, connections: dict[str, dict[str, Any]]) -> None:
-        """Write connections.yaml atomically with 0o600 permissions."""
+        """Write connections.yaml with 0o600 permissions (owner read/write only)."""
         self._config_dir.mkdir(parents=True, exist_ok=True)
         path = str(self._connections_path)
         fd = os.open(path, os.O_CREAT | os.O_WRONLY | os.O_TRUNC, 0o600)
         with os.fdopen(fd, "w") as f:
             yaml.dump(connections, f, default_flow_style=False)
 
     def _keyring_set(self, name: str, password: str) -> bool:
-        """Store password in OS keychain. Returns True on success, False if unavailable."""
+        """Store password in OS keychain. Returns True on success, False on failure."""
         try:
-            import keyring
-            import keyring.errors
-
             keyring.set_password(_KEYRING_SERVICE, name, password)
             return True
         except Exception as exc:  # noqa: BLE001
-            logger.warning("keyring unavailable — storing password in plaintext: %s", exc)
+            logger.warning(
+                "keyring write failed for '%s' — password will be stored in plaintext: %s",
+                name,
+                exc,
+            )
             return False
 
     def _keyring_get(self, name: str) -> str | None:
         """Retrieve password from OS keychain, or None if unavailable/not found."""
         try:
-            import keyring
-
             return keyring.get_password(_KEYRING_SERVICE, name)
         except Exception:  # noqa: BLE001
             return None
 
     def _keyring_delete(self, name: str) -> None:
         """Delete keychain entry; silently no-ops if not found or unavailable."""
-        try:
-            import keyring
-            import keyring.errors
-
+        with contextlib.suppress(Exception):
             keyring.delete_password(_KEYRING_SERVICE, name)
-        except Exception:  # noqa: BLE001
-            pass
 
     def save_connection(self, name: str, params: dict[str, Any]) -> None:
         """Save a named connection. Raises ConfigError on missing/invalid fields."""
@@ -113,7 +109,7 @@ def save_connection(self, name: str, params: dict[str, Any]) -> None:
         stored = dict(params)
         password = str(stored.pop("password", ""))
 
-        # Try keyring first; fall back to plaintext if unavailable.
+        self._keyring_delete(name)
         if self._keyring_set(name, password):
             stored["_keyring"] = True
         else:

src/open_data_agent/db/diagnostics.py

@@ -14,6 +14,11 @@
 
 __all__ = ["DiagnosticEngine"]
 
+# Each (col, table) pair fires 2 live queries; cap combinations to bound diagnostic overhead.
+_MAX_DIAG_COLUMNS = 4
+_MAX_DIAG_TABLES = 2
+_MAX_DIAG_SAMPLE_VALUES = 5  # DISTINCT sample values fetched per (col, table) pair
+
 logger = logging.getLogger("open_data_agent.db.diagnostics")
 
 _TABLE_PATTERN = re.compile(
@@ -91,14 +96,14 @@ def diagnose(self, sql: str, result: QueryResult) -> str:
                     lines.append(f"[diagnostic]   Table '{table}': could not count rows.")
 
             filter_cols = _extract_filter_columns(sql)
-            for col in filter_cols[:4]:  # limit to 4 filter columns
-                for table in table_names[:2]:
+            for col in filter_cols[:_MAX_DIAG_COLUMNS]:
+                for table in table_names[:_MAX_DIAG_TABLES]:
                     try:
                         quoted_table = self._adapter.quote_identifier(table)
                         quoted_col = self._adapter.quote_identifier(col)
                         cursor = self._conn.execute(
                             f"SELECT DISTINCT {quoted_col} FROM {quoted_table} "
-                            f"WHERE {quoted_col} IS NOT NULL LIMIT 5"
+                            f"WHERE {quoted_col} IS NOT NULL LIMIT {_MAX_DIAG_SAMPLE_VALUES}"
                         )
                         sample_vals = [str(r[0]) for r in cursor.fetchall()]
                         cursor2 = self._conn.execute(

src/open_data_agent/db/dialect.py

@@ -65,6 +65,14 @@ def introspect_columns_sql(self, schema: str, table: str) -> str:
     def normalize_column_row(self, row: dict[str, Any]) -> NormalizedColumn:
         """Map a dialect-specific introspection row to NormalizedColumn."""
 
+    @property
+    def supports_server_timeout(self) -> bool:
+        """True if the dialect supports a DB-level statement timeout (no thread needed)."""
+        return False
+
+    def set_statement_timeout(self, conn: Any, timeout_seconds: int) -> None:  # noqa: B027
+        """Set a DB-level statement timeout on *conn*. No-op for dialects that don't support it."""
+
 
 # ─────────────────────────────────────────────────────────────────────────────
 # SQLiteAdapter
@@ -211,6 +219,14 @@ def dangerous_patterns(self) -> list[str]:
             r"ALTER\s+SYSTEM",
         ]
 
+    @property
+    def supports_server_timeout(self) -> bool:
+        return True
+
+    def set_statement_timeout(self, conn: Any, timeout_seconds: int) -> None:
+        """Set PostgreSQL statement_timeout (milliseconds)."""
+        conn.execute(f"SET statement_timeout = {timeout_seconds * 1000}")
+
     def introspect_schemas_sql(self) -> str:
         return (
             "SELECT schema_name "
@@ -322,6 +338,14 @@ def dangerous_patterns(self) -> list[str]:
             r"\bGRANT\b",
         ]
 
+    @property
+    def supports_server_timeout(self) -> bool:
+        return True
+
+    def set_statement_timeout(self, conn: Any, timeout_seconds: int) -> None:
+        """Set MySQL MAX_EXECUTION_TIME (ms). NOTE: applies to SELECT only; 0 disables."""
+        conn.execute(f"SET SESSION MAX_EXECUTION_TIME = {timeout_seconds * 1000}")
+
     def introspect_schemas_sql(self) -> str:
         return (
             "SELECT schema_name "

src/open_data_agent/db/query.py

@@ -89,42 +89,14 @@ def execute(self, sql: str, question: str | None = None) -> QueryResult:
         columns: list[str] = []
         rows: list[tuple[object, ...]] = []
 
-        result_holder: dict[str, Any] = {}
-        exc_holder: list[Exception] = []
-
-        def _run() -> None:
-            try:
-                cursor = self._conn.execute(sql_with_limit)
-                result_holder["columns"] = (
-                    [d[0] for d in cursor.description] if cursor.description else []
-                )
-                result_holder["rows"] = [tuple(r) for r in cursor.fetchall()]
-            except Exception as exc:
-                exc_holder.append(exc)
-
-        thread = threading.Thread(target=_run, daemon=True)
-        thread.start()
-        thread.join(timeout=timeout_seconds)
-
-        if thread.is_alive():
-            # SQLite: interrupt(); psycopg v3: cancel(); pymysql: no-op (daemon reaped at exit).
-            with contextlib.suppress(AttributeError):
-                self._conn.interrupt()
-            try:
-                self._conn.cancel()
-            except AttributeError:
-                pass
-            except Exception as _cancel_exc:  # noqa: BLE001
-                logger.debug("conn.cancel() raised: %s", _cancel_exc)
-            thread.join(timeout=1.0)
-            error = f"Query timed out after {timeout_seconds}s"
-            logger.error("Query timed out: %s", sql_with_limit[:200])
-        elif exc_holder:
-            error = str(exc_holder[0])
-            logger.error("Query execution failed: %s", exc_holder[0])
+        if self._adapter.supports_server_timeout:
+            columns, rows, error = self._execute_with_server_timeout(
+                sql_with_limit, timeout_seconds
+            )
         else:
-            columns = result_holder.get("columns", [])
-            rows = result_holder.get("rows", [])
+            columns, rows, error = self._execute_with_thread_timeout(
+                sql_with_limit, timeout_seconds
+            )
 
         duration_ms = (time.monotonic() - t0) * 1000
 
@@ -152,6 +124,87 @@ def _run() -> None:
 
         return result
 
+    def _execute_with_server_timeout(
+        self, sql: str, timeout_seconds: int
+    ) -> tuple[list[str], list[tuple[object, ...]], str | None]:
+        """Execute using a DB-level statement timeout (PostgreSQL, MySQL)."""
+        try:
+            self._adapter.set_statement_timeout(self._conn, timeout_seconds)
+        except Exception as exc:  # noqa: BLE001
+            logger.warning(
+                "Failed to set DB-level statement timeout (%ss): %s — "
+                "query will proceed without server-side timeout",
+                timeout_seconds,
+                exc,
+            )
+
+        try:
+            cursor = self._conn.execute(sql)
+            cols = [d[0] for d in cursor.description] if cursor.description else []
+            fetched = [tuple(r) for r in cursor.fetchall()]
+            return cols, fetched, None
+        except Exception as exc:  # noqa: BLE001
+            if self._is_timeout_exception(exc):
+                logger.error("Query timed out: %s", sql[:200])
+                return [], [], f"Query timed out after {timeout_seconds}s"
+            logger.error("Query execution failed: %s", exc)
+            return [], [], str(exc)
+        finally:
+            with contextlib.suppress(Exception):
+                self._adapter.set_statement_timeout(self._conn, 0)
+
+    @staticmethod
+    def _is_timeout_exception(exc: Exception) -> bool:
+        """True if *exc* is a DB-level statement timeout (pgcode 57014 or MySQL errno 3024/1969)."""
+        pgcode = getattr(exc, "pgcode", None) or getattr(exc, "sqlstate", None)
+        if pgcode == "57014":
+            return True
+        errno = getattr(exc, "args", (None,))[0] if exc.args else None
+        return errno in (3024, 1969)
+
+    def _execute_with_thread_timeout(
+        self, sql: str, timeout_seconds: int
+    ) -> tuple[list[str], list[tuple[object, ...]], str | None]:
+        """Execute in a daemon thread with a join timeout (SQLite fallback)."""
+        result_holder: dict[str, Any] = {}
+        exc_holder: list[Exception] = []
+
+        def _run() -> None:
+            try:
+                cursor = self._conn.execute(sql)
+                result_holder["columns"] = (
+                    [d[0] for d in cursor.description] if cursor.description else []
+                )
+                result_holder["rows"] = [tuple(r) for r in cursor.fetchall()]
+            except Exception as exc:
+                exc_holder.append(exc)
+
+        thread = threading.Thread(target=_run, daemon=True)
+        thread.start()
+        thread.join(timeout=timeout_seconds)
+
+        if thread.is_alive():
+            with contextlib.suppress(AttributeError):
+                self._conn.interrupt()
+            thread.join(timeout=1.0)
+            if thread.is_alive():
+                logger.warning(
+                    "Query thread did not stop after interrupt — thread may be leaked: %s",
+                    sql[:200],
+                )
+            logger.error("Query timed out: %s", sql[:200])
+            return [], [], f"Query timed out after {timeout_seconds}s"
+
+        if exc_holder:
+            logger.error("Query execution failed: %s", exc_holder[0])
+            return [], [], str(exc_holder[0])
+
+        return (
+            result_holder.get("columns", []),
+            result_holder.get("rows", []),
+            None,
+        )
+
     def _inject_limit(self, sql: str) -> tuple[str, int, bool]:
         """Inject or clamp LIMIT; returns (sql_with_limit, effective_limit, was_truncated)."""
         config = self._config