fix: route all agent loopback calls through internal_api_base() helper (#3322)

#2753 made the agent loopback base port-configurable but only for
_COOKBOOK_BASE in tool_implementations. Several other in-process loopback
calls still hardcoded http://localhost:7000 and broke off port 7000:
cookbook_serve_lifecycle (model-endpoints x2, shell/exec), builtin_actions
(model/serve), task_routes (calendar x3), and the gallery/email calls in
tool_implementations.

Extract the resolution (ODYSSEUS_INTERNAL_BASE / APP_PORT / 7000 fallback,
127.0.0.1 to avoid IPv6 ambiguity) into core.constants.internal_api_base()
and route every call site through it. Rename the now-misnamed _COOKBOOK_BASE
to _INTERNAL_BASE since it serves gallery/email/calendar/serve too. Adds a
test for the resolver plus a regression guard against reintroducing the
literal.

Part of #2752.

Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>

core/constants.py

@@ -38,3 +38,22 @@ CLEANUP_INTERVAL_HOURS = int(os.getenv("CLEANUP_INTERVAL_HOURS", "24"))
 # Default parameters
 DEFAULT_TEMPERATURE = 1.0
 DEFAULT_MAX_TOKENS = 0
+
+
+def internal_api_base() -> str:
+    """Base URL for in-process loopback calls to Odysseus's own API.
+
+    Agent tools and background jobs reach admin-gated routes by calling the
+    running server over HTTP. Resolution order:
+      1. ODYSSEUS_INTERNAL_BASE  - explicit override (e.g. behind a TLS proxy).
+      2. APP_PORT                - http://127.0.0.1:$APP_PORT (docker-compose).
+      3. Fallback http://127.0.0.1:7000 - legacy default.
+
+    127.0.0.1 (not "localhost") avoids IPv6/DNS ambiguity for a strictly-local
+    call. Without this, loopback tools fail with "All connection attempts
+    failed" whenever the server is not on port 7000.
+    """
+    override = os.environ.get("ODYSSEUS_INTERNAL_BASE")
+    if override:
+        return override.rstrip("/")
+    return f"http://127.0.0.1:{os.environ.get('APP_PORT', '7000')}"

routes/task_routes.py

@@ -11,6 +11,7 @@ from fastapi import APIRouter, HTTPException, Request
 from pydantic import BaseModel
 
 from core.database import SessionLocal, ScheduledTask, TaskRun
+from core.constants import internal_api_base
 from src.auth_helpers import get_current_user
 from src.task_scheduler import compute_next_run, HOUSEKEEPING_DEFAULTS
 from routes.prefs_routes import _load_for_user, _save_for_user
@@ -56,7 +57,7 @@ def _maybe_cascade_calendar_event(task) -> None:
         try:
             with httpx.Client(timeout=10) as client:
                 r = client.delete(
-                    f"http://localhost:7000/api/calendar/events/{uid}",
+                    f"{internal_api_base()}/api/calendar/events/{uid}",
                     headers=headers,
                 )
                 if r.status_code >= 400:
@@ -81,7 +82,7 @@ def _maybe_cascade_calendar_event(task) -> None:
     try:
         with httpx.Client(timeout=10) as client:
             # Find the Cookbook calendar.
-            cal_r = client.get("http://localhost:7000/api/calendar/calendars", headers=headers)
+            cal_r = client.get(f"{internal_api_base()}/api/calendar/calendars", headers=headers)
             if cal_r.status_code >= 400:
                 return
             cals = (cal_r.json() or {}).get("calendars", [])
@@ -98,7 +99,7 @@ def _maybe_cascade_calendar_event(task) -> None:
             start = (now - _td(days=30)).isoformat()
             end = (now + _td(days=365)).isoformat()
             ev_r = client.get(
-                "http://localhost:7000/api/calendar/events",
+                f"{internal_api_base()}/api/calendar/events",
                 params={"start": start, "end": end, "calendar": cal_href},
                 headers=headers,
             )

src/builtin_actions.py

@@ -12,6 +12,7 @@ from typing import Tuple
 
 from src.auth_helpers import owner_filter
 from core.platform_compat import IS_WINDOWS, find_bash
+from core.constants import internal_api_base
 
 logger = logging.getLogger(__name__)
 
@@ -2118,7 +2119,7 @@ async def action_cookbook_serve(
 
     try:
         async with httpx.AsyncClient(timeout=30) as client:
-            r = await client.post("http://localhost:7000/api/model/serve",
+            r = await client.post(f"{internal_api_base()}/api/model/serve",
                                   json=body, headers=headers)
             data = r.json() if r.content else {}
     except Exception as e:

src/cookbook_serve_lifecycle.py

@@ -19,6 +19,7 @@ import time
 from pathlib import Path
 
 import httpx
+from core.constants import internal_api_base
 
 logger = logging.getLogger(__name__)
 
@@ -58,7 +59,7 @@ async def _delete_endpoint_for_task(task: dict) -> None:
     try:
         async with httpx.AsyncClient(timeout=8) as client:
             r = await client.get(
-                "http://localhost:7000/api/model-endpoints",
+                f"{internal_api_base()}/api/model-endpoints",
                 headers=_internal_headers(),
             )
             if r.status_code >= 400:
@@ -73,7 +74,7 @@ async def _delete_endpoint_for_task(task: dict) -> None:
                 ep = next((e for e in eps if hostport in (e.get("base_url") or "")), None)
             if ep:
                 await client.delete(
-                    f"http://localhost:7000/api/model-endpoints/{ep['id']}",
+                    f"{internal_api_base()}/api/model-endpoints/{ep['id']}",
                     headers=_internal_headers(),
                 )
                 logger.info(
@@ -108,7 +109,7 @@ async def _stop_serve(session_id: str, remote_host: str = "", ssh_port: str = ""
     try:
         async with httpx.AsyncClient(timeout=15) as client:
             r = await client.post(
-                "http://localhost:7000/api/shell/exec",
+                f"{internal_api_base()}/api/shell/exec",
                 json={"command": cmd},
                 headers=_internal_headers(),
             )

src/tool_implementations.py

@@ -13,6 +13,7 @@ import re
 from typing import Any, Dict, List, Optional
 
 from src.constants import MAX_OUTPUT_CHARS, MAX_READ_CHARS
+from core.constants import internal_api_base
 
 
 def get_mcp_manager():
@@ -2492,24 +2493,12 @@ async def do_manage_calendar(content: str, owner: Optional[str] = None) -> Dict:
 
 # ── Cookbook tools ──
 
-# Cookbook routes loopback. The agent's tool calls run in-process but
+# In-process loopback base for agent tools that call Odysseus's own API
-# need to reach admin-gated cookbook routes; we ride the per-process
+# (cookbook state, model serve, gallery, email, calendar). We ride the
-# internal token so require_admin lets us through. See core/middleware.py.
+# per-process internal token so require_admin lets us through. See
-#
+# core/middleware.py. Resolution (override / APP_PORT / 7000) lives in
-# Resolution order:
+# core.constants.internal_api_base().
-#   1. ODYSSEUS_INTERNAL_BASE — explicit override (e.g. behind a TLS proxy).
+_INTERNAL_BASE = internal_api_base()
-#   2. APP_PORT — derive http://127.0.0.1:$APP_PORT (matches docker-compose).
-#   3. Fallback http://127.0.0.1:7000 — preserves legacy default.
-#
-# 127.0.0.1 (not "localhost") avoids IPv6/DNS ambiguity for a strictly-local
-# call. Without this, tools that loop back (app_api, trigger_research,
-# cookbook state read/write) fail with "All connection attempts failed"
-# whenever the running uvicorn isn't on 7000 — which is most non-default
-# deployments and any side-by-side multi-instance setup.
-_COOKBOOK_BASE = os.environ.get(
-    "ODYSSEUS_INTERNAL_BASE",
-    f"http://127.0.0.1:{os.environ.get('APP_PORT', '7000')}",
-)
 
 
 def _internal_headers(owner: Optional[str] = None) -> Dict[str, str]:
@@ -2528,7 +2517,7 @@ async def _cookbook_servers() -> Dict[str, Any]:
     import httpx
     try:
         async with httpx.AsyncClient(timeout=10) as client:
-            r = await client.get(f"{_COOKBOOK_BASE}/api/cookbook/state", headers=_internal_headers())
+            r = await client.get(f"{_INTERNAL_BASE}/api/cookbook/state", headers=_internal_headers())
             state = r.json() if r.headers.get("content-type", "").startswith("application/json") else {}
     except Exception:
         return {"default_host": "", "hosts": []}
@@ -2594,7 +2583,7 @@ async def _cookbook_env_for_host(host: str) -> Dict[str, Any]:
     state: Dict[str, Any] = {}
     try:
         async with httpx.AsyncClient(timeout=10) as client:
-            r = await client.get(f"{_COOKBOOK_BASE}/api/cookbook/state", headers=headers)
+            r = await client.get(f"{_INTERNAL_BASE}/api/cookbook/state", headers=headers)
             state = r.json() if r.headers.get("content-type", "").startswith("application/json") else {}
     except Exception as e:
         logger.debug(f"cookbook env lookup failed for host={host!r}: {e}")
@@ -2654,7 +2643,7 @@ async def _cookbook_register_task(session_id: str, model: str, host: str,
     headers = _internal_headers()
     try:
         async with httpx.AsyncClient(timeout=10) as client:
-            r = await client.get(f"{_COOKBOOK_BASE}/api/cookbook/state", headers=headers)
+            r = await client.get(f"{_INTERNAL_BASE}/api/cookbook/state", headers=headers)
             state = r.json() if r.headers.get("content-type", "").startswith("application/json") else {}
     except Exception as e:
         logger.debug(f"cookbook state read failed: {e}")
@@ -2698,7 +2687,7 @@ async def _cookbook_register_task(session_id: str, model: str, host: str,
     state["tasks"] = tasks
     try:
         async with httpx.AsyncClient(timeout=10) as client:
-            r = await client.post(f"{_COOKBOOK_BASE}/api/cookbook/state",
+            r = await client.post(f"{_INTERNAL_BASE}/api/cookbook/state",
                                   json=state, headers=headers)
         return r.status_code < 400
     except Exception as e:
@@ -2781,7 +2770,7 @@ async def do_app_api(content: str, owner: Optional[str] = None) -> Dict:
         return {"error": "Invalid JSON arguments", "exit_code": 1}
 
     action = (args.get("action") or "call").lower()
-    base = _COOKBOOK_BASE
+    base = _INTERNAL_BASE
 
     if action == "endpoints":
         # Fetch FastAPI's OpenAPI schema so the agent can discover any
@@ -3042,7 +3031,7 @@ async def do_download_model(content: str, owner: Optional[str] = None) -> Dict:
     if env_cfg.get("ssh_port"):   payload["ssh_port"]   = env_cfg["ssh_port"]
     try:
         async with httpx.AsyncClient(timeout=30) as client:
-            resp = await client.post(f"{_COOKBOOK_BASE}/api/model/download",
+            resp = await client.post(f"{_INTERNAL_BASE}/api/model/download",
                                      json=payload, headers=_internal_headers())
             data = resp.json()
         if data.get("ok"):
@@ -3118,7 +3107,7 @@ async def do_serve_model(content: str, owner: Optional[str] = None) -> Dict:
     if env_cfg.get("ssh_port"):   payload["ssh_port"]   = env_cfg["ssh_port"]
     try:
         async with httpx.AsyncClient(timeout=30) as client:
-            resp = await client.post(f"{_COOKBOOK_BASE}/api/model/serve",
+            resp = await client.post(f"{_INTERNAL_BASE}/api/model/serve",
                                      json=payload, headers=_internal_headers())
             data = resp.json()
         if data.get("ok"):
@@ -3158,7 +3147,7 @@ async def do_list_served_models(content: str, owner: Optional[str] = None) -> Di
     cookbook_tasks: List[Dict[str, Any]] = []
     try:
         async with httpx.AsyncClient(timeout=15) as client:
-            resp = await client.get(f"{_COOKBOOK_BASE}/api/cookbook/tasks/status",
+            resp = await client.get(f"{_INTERNAL_BASE}/api/cookbook/tasks/status",
                                     headers=_internal_headers())
             cookbook_tasks = (resp.json() or {}).get("tasks") or []
     except Exception as e:
@@ -3277,7 +3266,7 @@ async def _cookbook_kill_session(session_id: str, *, remote_host: str = "",
     state: Dict[str, Any] = {}
     try:
         async with httpx.AsyncClient(timeout=10) as client:
-            resp = await client.get(f"{_COOKBOOK_BASE}/api/cookbook/state", headers=headers)
+            resp = await client.get(f"{_INTERNAL_BASE}/api/cookbook/state", headers=headers)
             state = resp.json() or {}
     except Exception as e:
         logger.debug(f"cookbook state lookup failed for {session_id}: {e}")
@@ -3306,7 +3295,7 @@ async def _cookbook_kill_session(session_id: str, *, remote_host: str = "",
 
     try:
         async with httpx.AsyncClient(timeout=15) as client:
-            resp = await client.post(f"{_COOKBOOK_BASE}/api/shell/exec",
+            resp = await client.post(f"{_INTERNAL_BASE}/api/shell/exec",
                                      json={"command": cmd}, headers=headers)
         if resp.status_code >= 400:
             return {"error": f"shell/exec returned HTTP {resp.status_code}: {resp.text[:200]}", "exit_code": 1}
@@ -3327,7 +3316,7 @@ async def _cookbook_kill_session(session_id: str, *, remote_host: str = "",
             try:
                 matched["status"] = "stopped"
                 async with httpx.AsyncClient(timeout=10) as client:
-                    await client.post(f"{_COOKBOOK_BASE}/api/cookbook/state",
+                    await client.post(f"{_INTERNAL_BASE}/api/cookbook/state",
                                       json=state, headers=headers)
             except Exception as e:
                 logger.debug(f"failed to mark {session_id} stopped in state: {e}")
@@ -3390,7 +3379,7 @@ async def do_tail_serve_output(content: str, owner: Optional[str] = None) -> Dic
         state: Dict[str, Any] = {}
         try:
             async with httpx.AsyncClient(timeout=10) as client:
-                resp = await client.get(f"{_COOKBOOK_BASE}/api/cookbook/state", headers=headers)
+                resp = await client.get(f"{_INTERNAL_BASE}/api/cookbook/state", headers=headers)
                 state = resp.json() or {}
         except Exception as e:
             logger.debug(f"cookbook state lookup failed for {session_id}: {e}")
@@ -3428,7 +3417,7 @@ async def do_tail_serve_output(content: str, owner: Optional[str] = None) -> Dic
         host_label = "local"
     try:
         async with httpx.AsyncClient(timeout=20) as client:
-            resp = await client.post(f"{_COOKBOOK_BASE}/api/shell/exec",
+            resp = await client.post(f"{_INTERNAL_BASE}/api/shell/exec",
                                      json={"command": cmd}, headers=headers)
         if resp.status_code >= 400:
             return {"error": f"shell/exec returned HTTP {resp.status_code}: {resp.text[:200]}", "exit_code": 1}
@@ -3479,7 +3468,7 @@ async def do_list_downloads(content: str, owner: Optional[str] = None) -> Dict:
     import httpx
     try:
         async with httpx.AsyncClient(timeout=15) as client:
-            resp = await client.get(f"{_COOKBOOK_BASE}/api/cookbook/tasks/status",
+            resp = await client.get(f"{_INTERNAL_BASE}/api/cookbook/tasks/status",
                                     headers=_internal_headers())
             data = resp.json()
         tasks = [t for t in data.get("tasks", []) if (t.get("type") or "").lower() == "download"]
@@ -3530,7 +3519,7 @@ async def do_search_hf_models(content: str, owner: Optional[str] = None) -> Dict
         params["limit"] = str(limit)
     try:
         async with httpx.AsyncClient(timeout=30) as client:
-            resp = await client.get(f"{_COOKBOOK_BASE}/api/cookbook/hf-latest",
+            resp = await client.get(f"{_INTERNAL_BASE}/api/cookbook/hf-latest",
                                     params=params, headers=_internal_headers())
             data = resp.json()
         models = data.get("models") if isinstance(data, dict) else data
@@ -3596,7 +3585,7 @@ async def do_adopt_served_model(content: str, owner: Optional[str] = None) -> Di
         check = f"tmux has-session -t {shlex.quote(sess)} 2>&1"
     try:
         async with httpx.AsyncClient(timeout=10) as client:
-            r = await client.post(f"{_COOKBOOK_BASE}/api/shell/exec",
+            r = await client.post(f"{_INTERNAL_BASE}/api/shell/exec",
                                   json={"command": check}, headers=headers)
             data = r.json() if r.headers.get("content-type", "").startswith("application/json") else {}
         if r.status_code >= 400 or (data.get("exit_code") not in (None, 0)):
@@ -3613,7 +3602,7 @@ async def do_adopt_served_model(content: str, owner: Optional[str] = None) -> Di
     server_up = False
     try:
         async with httpx.AsyncClient(timeout=10) as client:
-            r = await client.post(f"{_COOKBOOK_BASE}/api/shell/exec",
+            r = await client.post(f"{_INTERNAL_BASE}/api/shell/exec",
                                   json={"command": health_cmd}, headers=headers)
             body = (r.json() or {}).get("stdout", "") if r.headers.get("content-type", "").startswith("application/json") else ""
             server_up = '"data"' in body or '"object"' in body
@@ -3624,7 +3613,7 @@ async def do_adopt_served_model(content: str, owner: Optional[str] = None) -> Di
     # overwrite the whole file (that'd nuke presets).
     try:
         async with httpx.AsyncClient(timeout=10) as client:
-            r = await client.get(f"{_COOKBOOK_BASE}/api/cookbook/state", headers=headers)
+            r = await client.get(f"{_INTERNAL_BASE}/api/cookbook/state", headers=headers)
             state = r.json() if r.headers.get("content-type", "").startswith("application/json") else {}
     except Exception as e:
         return {"error": f"could not read cookbook state: {e}", "exit_code": 1}
@@ -3660,7 +3649,7 @@ async def do_adopt_served_model(content: str, owner: Optional[str] = None) -> Di
         state["tasks"] = tasks
         try:
             async with httpx.AsyncClient(timeout=10) as client:
-                await client.post(f"{_COOKBOOK_BASE}/api/cookbook/state",
+                await client.post(f"{_INTERNAL_BASE}/api/cookbook/state",
                                   json=state, headers=headers)
         except Exception as e:
             return {"error": f"could not save cookbook state: {e}", "exit_code": 1}
@@ -3737,7 +3726,7 @@ async def do_list_serve_presets(content: str, owner: Optional[str] = None) -> Di
     import httpx
     try:
         async with httpx.AsyncClient(timeout=10) as client:
-            resp = await client.get(f"{_COOKBOOK_BASE}/api/cookbook/state",
+            resp = await client.get(f"{_INTERNAL_BASE}/api/cookbook/state",
                                     headers=_internal_headers())
             state = resp.json() or {}
     except Exception as e:
@@ -3785,7 +3774,7 @@ async def do_serve_preset(content: str, owner: Optional[str] = None) -> Dict:
 
     try:
         async with httpx.AsyncClient(timeout=10) as client:
-            resp = await client.get(f"{_COOKBOOK_BASE}/api/cookbook/state",
+            resp = await client.get(f"{_INTERNAL_BASE}/api/cookbook/state",
                                     headers=_internal_headers())
             state = resp.json() or {}
     except Exception as e:
@@ -3829,7 +3818,7 @@ async def do_serve_preset(content: str, owner: Optional[str] = None) -> Dict:
 
     try:
         async with httpx.AsyncClient(timeout=30) as client:
-            resp = await client.post(f"{_COOKBOOK_BASE}/api/model/serve",
+            resp = await client.post(f"{_INTERNAL_BASE}/api/model/serve",
                                      json=payload, headers=_internal_headers())
             data = resp.json()
         if data.get("ok"):
@@ -3881,7 +3870,7 @@ async def do_list_cached_models(content: str, owner: Optional[str] = None) -> Di
             p["platform"] = args["platform"]
         try:
             async with httpx.AsyncClient(timeout=60) as client:
-                resp = await client.get(f"{_COOKBOOK_BASE}/api/model/cached",
+                resp = await client.get(f"{_INTERNAL_BASE}/api/model/cached",
                                         params=p, headers=headers)
                 data = resp.json()
             ms = data.get("models", []) if isinstance(data, dict) else (data or [])
@@ -3901,7 +3890,7 @@ async def do_list_cached_models(content: str, owner: Optional[str] = None) -> Di
         servers: list = []
         try:
             async with httpx.AsyncClient(timeout=10) as client:
-                st = await client.get(f"{_COOKBOOK_BASE}/api/cookbook/state", headers=headers)
+                st = await client.get(f"{_INTERNAL_BASE}/api/cookbook/state", headers=headers)
                 st_data = st.json() if st.headers.get("content-type", "").startswith("application/json") else {}
             servers = (st_data.get("env", {}) or {}).get("servers") or []
         except Exception as e:
@@ -3972,7 +3961,7 @@ async def do_list_cached_models(content: str, owner: Optional[str] = None) -> Di
             downloaded = []
             try:
                 async with httpx.AsyncClient(timeout=10) as client:
-                    st = await client.get(f"{_COOKBOOK_BASE}/api/cookbook/state", headers=headers)
+                    st = await client.get(f"{_INTERNAL_BASE}/api/cookbook/state", headers=headers)
                     state = st.json() if st.headers.get("content-type", "").startswith("application/json") else {}
                 for t in (state.get("tasks") or []):
                     if not isinstance(t, dict) or t.get("type") != "download":
@@ -4043,7 +4032,7 @@ async def do_edit_image(content: str, owner: Optional[str] = None) -> Dict:
         payload["scale"] = args["scale"]
     try:
         async with httpx.AsyncClient(timeout=120) as client:
-            resp = await client.post(f"http://localhost:7000/api/gallery/{action}", json=payload)
+            resp = await client.post(f"{_INTERNAL_BASE}/api/gallery/{action}", json=payload)
             data = resp.json()
         if data.get("success") or data.get("id"):
             return {"output": f"Image edited ({action}). New image ID: {data.get('id', '?')}", "exit_code": 0}
@@ -4159,7 +4148,7 @@ async def do_trigger_research(content: str, owner: Optional[str] = None) -> Dict
         payload["search_provider"] = args["search_provider"]
     try:
         async with httpx.AsyncClient(timeout=30) as client:
-            resp = await client.post(f"{_COOKBOOK_BASE}/api/research/start",
+            resp = await client.post(f"{_INTERNAL_BASE}/api/research/start",
                                      json=payload, headers=_internal_headers(owner))
         if resp.status_code >= 400:
             return {"error": f"research/start returned HTTP {resp.status_code}: {resp.text[:200]}", "exit_code": 1}
@@ -4219,7 +4208,7 @@ async def do_resolve_contact(content: str, owner: Optional[str] = None) -> Dict:
     async with httpx.AsyncClient(timeout=30) as client:
         # 2. Email history (sent/received)
         try:
-            resp = await client.get("http://localhost:7000/api/email/resolve-contact", params={"name": name})
+            resp = await client.get(f"{_INTERNAL_BASE}/api/email/resolve-contact", params={"name": name})
             if resp.status_code == 200:
                 for c in (resp.json().get("contacts") or []):
                     email = (c.get("email") or "").strip().lower()

tests/test_internal_api_base.py

@@ -0,0 +1,52 @@
+"""internal_api_base() resolution + a guard that loopback call sites use it."""
+import importlib
+import pathlib
+
+import pytest
+
+import core.constants as cc
+
+
+def _base(monkeypatch, **env):
+    for k in ("ODYSSEUS_INTERNAL_BASE", "APP_PORT"):
+        monkeypatch.delenv(k, raising=False)
+    for k, v in env.items():
+        monkeypatch.setenv(k, v)
+    return cc.internal_api_base()
+
+
+def test_default_is_legacy_7000(monkeypatch):
+    assert _base(monkeypatch) == "http://127.0.0.1:7000"
+
+
+def test_app_port_is_honored(monkeypatch):
+    assert _base(monkeypatch, APP_PORT="7860") == "http://127.0.0.1:7860"
+
+
+def test_explicit_override_wins_and_is_stripped(monkeypatch):
+    # Override beats APP_PORT and trailing slash is trimmed.
+    assert _base(monkeypatch, APP_PORT="7860",
+                 ODYSSEUS_INTERNAL_BASE="https://proxy.example/") == "https://proxy.example"
+
+
+def test_uses_127_not_localhost(monkeypatch):
+    # 127.0.0.1 avoids IPv6/DNS ambiguity for the strictly-local loopback.
+    assert "localhost" not in _base(monkeypatch)
+
+
+def test_no_hardcoded_loopback_left_in_call_sites():
+    # Regression guard: the converted files must not reintroduce the literal.
+    root = pathlib.Path(__file__).resolve().parent.parent
+    for rel in (
+        "src/tool_implementations.py",
+        "src/cookbook_serve_lifecycle.py",
+        "src/builtin_actions.py",
+        "routes/task_routes.py",
+    ):
+        text = (root / rel).read_text(encoding="utf-8")
+        # Allow it only inside comments; flag any code occurrence.
+        for ln in text.splitlines():
+            stripped = ln.strip()
+            if stripped.startswith("#"):
+                continue
+            assert "localhost:7000" not in ln, f"{rel}: hardcoded loopback URL: {ln.strip()}"