refactor(db): use semantic pooler probe identifiers

backend/app/db.py

@@ -66,7 +66,9 @@ async def _probe_pooler_admin_console(admin_db: str) -> str | None:
     simple query protocol.
     """
 
-    dsn, password = build_admin_console_dsn(settings.database_url, admin_db)
+    pooler_dsn, pooler_password = build_admin_console_dsn(
+        settings.database_url, admin_db
+    )
 
     raw_timeout = float(settings.db_pooler_probe_timeout_seconds)
     if raw_timeout <= 0.0:
@@ -76,22 +78,22 @@ async def _probe_pooler_admin_console(admin_db: str) -> str | None:
     # Note: some PostgreSQL/libpq versions effectively treat values < 2 as 2.
     timeout_seconds = max(2, math.ceil(raw_timeout))
 
-    def _run() -> str | None:
+    def _run_pooler_probe() -> str | None:
         with psycopg.connect(
-            dsn,
-            password=password,
+            pooler_dsn,
+            password=pooler_password,
             connect_timeout=timeout_seconds,
-        ) as conn:
-            with conn.cursor() as cur:
-                cur.execute("SHOW VERSION;")
-                row = cur.fetchone()
-                if not row or row[0] is None:
+        ) as pooler_connection:
+            with pooler_connection.cursor() as pooler_cursor:
+                pooler_cursor.execute("SHOW VERSION;")
+                version_row = pooler_cursor.fetchone()
+                if not version_row or version_row[0] is None:
                     return None
-                return str(row[0])
+                return str(version_row[0])
 
     try:
         return await asyncio.wait_for(
-            asyncio.to_thread(_run), timeout=float(timeout_seconds) + 0.2
+            asyncio.to_thread(_run_pooler_probe), timeout=float(timeout_seconds) + 0.2
         )
     except Exception:  # noqa: BLE001
         return None

backend/tests/test_db_pooler_identifier_naming.py

@@ -0,0 +1,43 @@
+"""Naming-contract regressions for the database pooler probe boundary."""
+
+from __future__ import annotations
+
+import ast
+from pathlib import Path
+
+
+_DB_SOURCE = Path(__file__).parents[1] / "app" / "db.py"
+
+
+def _pooler_probe_function() -> ast.AsyncFunctionDef:
+    """Return the pooler-probe AST from the repository-owned database module."""
+    module_tree = ast.parse(_DB_SOURCE.read_text(encoding="utf-8"))
+    for module_node in module_tree.body:
+        if isinstance(module_node, ast.AsyncFunctionDef) and module_node.name == "_probe_pooler_admin_console":
+            return module_node
+    raise AssertionError("pooler probe function is missing")
+
+
+def test_pooler_probe_uses_bounded_context_identifiers() -> None:
+    """Require semantic names for the private pooler connection/query boundary."""
+    probe_function = _pooler_probe_function()
+    nested_functions = [
+        function_node
+        for function_node in probe_function.body
+        if isinstance(function_node, ast.FunctionDef)
+    ]
+    assert [function_node.name for function_node in nested_functions] == ["_run_pooler_probe"]
+
+    owned_names = {
+        name_node.id
+        for name_node in ast.walk(probe_function)
+        if isinstance(name_node, ast.Name)
+    }
+    assert {"dsn", "password", "conn", "cur", "row"}.isdisjoint(owned_names)
+    assert {
+        "pooler_dsn",
+        "pooler_password",
+        "pooler_connection",
+        "pooler_cursor",
+        "version_row",
+    }.issubset(owned_names)