refactor(config): remove legacy unauth REDIS_URL fallbacks (#645)

src/backend/services/capacity_manager.py

@@ -159,7 +159,8 @@ def __init__(
         slot_service: Optional[SlotService] = None,
         backlog_service: Optional[BacklogService] = None,
     ):
-        url = redis_url or os.getenv("REDIS_URL", "redis://redis:6379")
+        from config import REDIS_URL as _DEFAULT_REDIS_URL
+        url = redis_url or _DEFAULT_REDIS_URL
         self._redis = redis.from_url(url, decode_responses=True)
         self._slots = slot_service or SlotService(url)
         self._backlog = backlog_service or BacklogService()

src/backend/services/slot_service.py

@@ -63,7 +63,10 @@ class SlotService:
     - ZREMRANGEBYSCORE for cleanup
     """
 
-    def __init__(self, redis_url: str = "redis://redis:6379"):
+    def __init__(self, redis_url: Optional[str] = None):
+        if redis_url is None:
+            from config import REDIS_URL
+            redis_url = REDIS_URL
         self.redis = redis.from_url(redis_url, decode_responses=True)
         self.slots_prefix = "agent:slots:"
         self.metadata_prefix = "agent:slot:"
@@ -398,7 +401,5 @@ def get_slot_service() -> SlotService:
     """Get the global slot service instance."""
     global _slot_service
     if _slot_service is None:
-        import os
-        redis_url = os.getenv("REDIS_URL", "redis://redis:6379")
-        _slot_service = SlotService(redis_url)
+        _slot_service = SlotService()  # resolves REDIS_URL via config
     return _slot_service

src/backend/services/ssh_service.py

@@ -40,7 +40,10 @@
 class SshService:
     """Service for managing ephemeral SSH access to agent containers."""
 
-    def __init__(self, redis_url: str = "redis://redis:6379"):
+    def __init__(self, redis_url: Optional[str] = None):
+        if redis_url is None:
+            from config import REDIS_URL
+            redis_url = REDIS_URL
         self.redis_client = redis.from_url(redis_url, decode_responses=True)
 
     async def inject_ssh_key(self, agent_name: str, public_key: str) -> bool:
@@ -546,6 +549,5 @@ def get_ssh_service() -> SshService:
     """Get the SSH service singleton."""
     global _ssh_service
     if _ssh_service is None:
-        redis_url = os.getenv("REDIS_URL", "redis://redis:6379")
-        _ssh_service = redis_url and SshService(redis_url)
+        _ssh_service = SshService()  # resolves REDIS_URL via config
     return _ssh_service

tests/unit/conftest.py

@@ -4,12 +4,22 @@
 These tests run without a backend connection (no Docker, no API).
 """
 import importlib.util
+import os
 import sys
 import types
 from pathlib import Path
 
 import pytest
 
+# Issue #589 + #645: backend/config.py raises at import if REDIS_URL lacks
+# credentials. Unit tests don't share the parent conftest (the unit/pytest.ini
+# sets `norecursedirs = ..`), so set a creds-bearing dummy here. Any test
+# that wants to assert the fail-fast behavior (e.g. test_config_fail_fast.py)
+# still uses monkeypatch to override.
+os.environ.setdefault("REDIS_URL", "redis://test:test@redis:6379")
+os.environ.setdefault("REDIS_PASSWORD", "test")
+os.environ.setdefault("REDIS_BACKEND_PASSWORD", "test")
+
 # Ensure src/backend is importable before test modules are collected.
 #
 # BACKLOG-001: tests/unit/test_backlog.py does `from models import ...`, which

tests/unit/test_redis_url_no_fallback.py

@@ -0,0 +1,87 @@
+"""Lint-style regression for #645 (#589 follow-up).
+
+`config.py` raises at import if `REDIS_URL` lacks credentials (#589 / PR #643).
+Three backend services bypassed that gate by reading the env var directly
+with an unauthenticated localhost fallback:
+
+    redis_url = os.getenv("REDIS_URL", "redis://redis:6379")
+
+This test enforces the single-source-of-truth principle: every backend
+service must route Redis URL resolution through `config.REDIS_URL`. New
+services that grow a `redis://` literal default — or a fresh
+`os.getenv("REDIS_URL", ...)` fallback — trip these assertions.
+"""
+
+from __future__ import annotations
+
+import os
+import re
+from pathlib import Path
+
+import pytest
+
+pytestmark = pytest.mark.unit
+
+
+def _backend_services_dir() -> Path:
+    """Locate src/backend/services across host and in-container layouts."""
+    candidates = [
+        Path(__file__).resolve().parent.parent.parent / "src" / "backend" / "services",
+        Path("/app/services"),  # trinity-backend container
+    ]
+    env_override = os.environ.get("TRINITY_BACKEND_PATH")
+    if env_override:
+        candidates.insert(0, Path(env_override) / "services")
+    for c in candidates:
+        if c.is_dir():
+            return c
+    raise RuntimeError(
+        "Cannot locate src/backend/services (set TRINITY_BACKEND_PATH)"
+    )
+
+
+_SERVICES = _backend_services_dir()
+_PY_FILES = sorted(p for p in _SERVICES.rglob("*.py") if "__pycache__" not in p.parts)
+
+
+def _grep(pattern: re.Pattern) -> list[tuple[Path, int, str]]:
+    """Return (path, lineno, line) for every match across the services tree."""
+    hits: list[tuple[Path, int, str]] = []
+    for path in _PY_FILES:
+        for lineno, line in enumerate(path.read_text().splitlines(), start=1):
+            if pattern.search(line):
+                hits.append((path, lineno, line.strip()))
+    return hits
+
+
+# Direct env reads of REDIS_URL are forbidden — config.py is the canonical gate.
+_GETENV_REDIS_URL = re.compile(r"""os\.getenv\s*\(\s*['"]REDIS_URL['"]""")
+
+# Unauthenticated localhost literals embedded as defaults are forbidden — they
+# silently bypass #589's fail-fast guard. Constructors must default to None
+# and resolve via config.REDIS_URL when called with no argument.
+_UNAUTH_REDIS_LITERAL = re.compile(r"""['"]redis://redis:6379['"]""")
+
+
+class TestNoOsGetenvRedisUrl:
+    def test_no_os_getenv_redis_url_under_services(self):
+        hits = _grep(_GETENV_REDIS_URL)
+        assert not hits, (
+            "os.getenv(\"REDIS_URL\", ...) must not appear under "
+            "src/backend/services/. Import REDIS_URL from config instead "
+            "(see #645). Offending lines:\n"
+            + "\n".join(f"  {p}:{n}: {s}" for p, n, s in hits)
+        )
+
+
+class TestNoUnauthRedisLiteralDefault:
+    def test_no_unauthenticated_redis_localhost_literal(self):
+        hits = _grep(_UNAUTH_REDIS_LITERAL)
+        assert not hits, (
+            "'redis://redis:6379' literal must not appear under "
+            "src/backend/services/. It silently bypasses #589's fail-fast "
+            "credentials guard (see #645). Use Optional[str] = None + "
+            "lazy import of config.REDIS_URL inside the constructor. "
+            "Offending lines:\n"
+            + "\n".join(f"  {p}:{n}: {s}" for p, n, s in hits)
+        )