fix: keep GCP knowledge reads on Cloud SQL

hermes-agent/leoclean-bin/teleo-kb

@@ -1,12 +1,11 @@
 #!/usr/bin/env bash
 set -euo pipefail
 
-PROFILE_DIR=/home/teleo/.hermes/profiles/leoclean
+PROFILE_DIR=${HERMES_HOME:-/home/teleo/.hermes/profiles/leoclean}
 KB_TOOL="$PROFILE_DIR/bin/kb_tool.py"
 CLOUDSQL_TOOL="$PROFILE_DIR/bin/cloudsql_memory_tool.py"
 MODE=${TELEO_KB_MODE:-auto}
 COMMAND=${1:-}
-CLOUDSQL_STATUS_TIMEOUT=${TELEO_CLOUDSQL_STATUS_TIMEOUT_SECONDS:-30}
 
 cloudsql_supported() {
   case "$COMMAND" in
@@ -45,13 +44,12 @@ case "$MODE" in
     exec python3 "$KB_TOOL" "$@"
     ;;
   auto)
-    if [ -x "$CLOUDSQL_TOOL" ] && command -v psql >/dev/null 2>&1 && command -v gcloud >/dev/null 2>&1; then
+    # Canonical commands must fail closed on Cloud SQL errors. Falling through
-      if cloudsql_supported; then
+    # to a different local database would make source-of-truth behavior depend
-        if timeout "${CLOUDSQL_STATUS_TIMEOUT}s" python3 "$CLOUDSQL_TOOL" status --format json --redacted >/dev/null 2>&1; then
+    # on health-check latency.
+    if [ -x "$CLOUDSQL_TOOL" ] && command -v psql >/dev/null 2>&1 && command -v gcloud >/dev/null 2>&1 && cloudsql_supported; then
       run_cloudsql "$@"
     fi
-      fi
-    fi
     if command -v docker >/dev/null 2>&1; then
       exec python3 "$KB_TOOL" --local "$@"
     fi

tests/test_hermes_leoclean_kb_bridge_source.py

@@ -7,6 +7,7 @@ import contextlib
 import importlib.util
 import io
 import json
+import os
 import re
 import subprocess
 from pathlib import Path
@@ -66,6 +67,76 @@ def test_cloudsql_wrapper_supports_expected_review_gated_commands() -> None:
     assert "canonical apply" in cloudsql_text.lower()
 
 
+def _install_wrapper_fixture(tmp_path: Path) -> tuple[Path, dict[str, str]]:
+    profile = tmp_path / "profile"
+    bin_dir = profile / "bin"
+    bin_dir.mkdir(parents=True)
+    wrapper = bin_dir / "teleo-kb"
+    wrapper.write_bytes((BRIDGE_DIR / "teleo-kb").read_bytes())
+    wrapper.chmod(0o700)
+    cloudsql = bin_dir / "cloudsql_memory_tool.py"
+    cloudsql.write_text(
+        "import json, sys\nprint(json.dumps({'backend': 'cloudsql', 'argv': sys.argv[1:]}))\n",
+        encoding="utf-8",
+    )
+    cloudsql.chmod(0o700)
+    local = bin_dir / "kb_tool.py"
+    local.write_text(
+        "import json, sys\nprint(json.dumps({'backend': 'local', 'argv': sys.argv[1:]}))\n",
+        encoding="utf-8",
+    )
+    local.chmod(0o700)
+    fake_bin = tmp_path / "fake-bin"
+    fake_bin.mkdir()
+    for name in ("psql", "gcloud"):
+        executable = fake_bin / name
+        executable.write_text("#!/usr/bin/env bash\nexit 0\n", encoding="utf-8")
+        executable.chmod(0o700)
+    env = {
+        **os.environ,
+        "HERMES_HOME": str(profile),
+        "TELEO_KB_MODE": "auto",
+        "PATH": f"{fake_bin}:{os.environ['PATH']}",
+    }
+    return wrapper, env
+
+
+def test_cloudsql_wrapper_auto_mode_routes_supported_reads_without_slow_probe(tmp_path: Path) -> None:
+    wrapper, env = _install_wrapper_fixture(tmp_path)
+
+    completed = subprocess.run(
+        [str(wrapper), "search", "AI sandbagging M&A liability", "--format", "json"],
+        text=True,
+        capture_output=True,
+        check=False,
+        env=env,
+    )
+
+    assert completed.returncode == 0, completed.stderr
+    result = json.loads(completed.stdout)
+    assert result == {
+        "backend": "cloudsql",
+        "argv": ["search", "AI sandbagging M&A liability", "--format", "json"],
+    }
+
+
+def test_cloudsql_wrapper_auto_mode_never_falls_back_after_supported_command_failure(tmp_path: Path) -> None:
+    wrapper, env = _install_wrapper_fixture(tmp_path)
+    cloudsql = Path(env["HERMES_HOME"]) / "bin" / "cloudsql_memory_tool.py"
+    cloudsql.write_text("raise SystemExit(23)\n", encoding="utf-8")
+
+    completed = subprocess.run(
+        [str(wrapper), "status", "--format", "json"],
+        text=True,
+        capture_output=True,
+        check=False,
+        env=env,
+    )
+
+    assert completed.returncode == 23
+    assert "local" not in completed.stdout
+
+
 def test_bridge_edge_proposals_stage_strict_apply_payloads() -> None:
     cloudsql_text = (BRIDGE_DIR / "cloudsql_memory_tool.py").read_text()
     vps_text = (BRIDGE_DIR / "kb_tool.py").read_text()