diff --git a/apps/desktop/src/app/settings/custom-endpoints-settings.tsx b/apps/desktop/src/app/settings/custom-endpoints-settings.tsx
index b74b30a2909..b6f7514c513 100644
--- a/apps/desktop/src/app/settings/custom-endpoints-settings.tsx
+++ b/apps/desktop/src/app/settings/custom-endpoints-settings.tsx
@@ -32,6 +32,10 @@ interface EndpointForm {
   makeDefault: boolean
   model: string
   name: string
+  sshHost: string
+  sshKeyPath: string
+  sshPort: string
+  sshUser: string
 }
 
 const EMPTY_FORM: EndpointForm = {
@@ -42,7 +46,11 @@ const EMPTY_FORM: EndpointForm = {
   id: '',
   makeDefault: true,
   model: '',
-  name: ''
+  name: '',
+  sshHost: '',
+  sshKeyPath: '',
+  sshPort: '',
+  sshUser: ''
 }
 
 function formFromEndpoint(endpoint: CustomEndpoint): EndpointForm {
@@ -54,7 +62,11 @@ function formFromEndpoint(endpoint: CustomEndpoint): EndpointForm {
     id: endpoint.id,
     makeDefault: Boolean(endpoint.is_current),
     model: endpoint.model,
-    name: endpoint.name
+    name: endpoint.name,
+    sshHost: endpoint.ssh_tunnel?.host ?? '',
+    sshKeyPath: endpoint.ssh_tunnel?.key_path ?? '',
+    sshPort: endpoint.ssh_tunnel?.port ? String(endpoint.ssh_tunnel.port) : '',
+    sshUser: endpoint.ssh_tunnel?.user ?? ''
   }
 }
 
@@ -70,7 +82,15 @@ function toPayload(form: EndpointForm, models?: string[]): CustomEndpointUpdate
     context_length: Number.isFinite(contextLength) && contextLength > 0 ? contextLength : undefined,
     discover_models: form.discoverModels,
     make_default: form.makeDefault,
-    models: models?.length ? models : undefined
+    models: models?.length ? models : undefined,
+    ssh_tunnel: form.sshHost.trim()
+      ? {
+          host: form.sshHost.trim(),
+          key_path: form.sshKeyPath.trim() || undefined,
+          port: Number.parseInt(form.sshPort, 10) || undefined,
+          user: form.sshUser.trim() || undefined
+        }
+      : {}
   }
 }
 
@@ -320,6 +340,46 @@ export function CustomEndpointsSettings({ onConfigSaved, onMainModelChanged }: C
                 value={form.baseUrl}
               />
             </label>
+            <div className="grid gap-3 rounded-md border border-border/40 p-3">
+              <div className="text-xs text-muted-foreground">
+                SSH tunnel (optional) — the endpoint URL is resolved from the remote host; Hermes chooses a private local port.
+              </div>
+              <div className="grid gap-3 sm:grid-cols-2">
+                <label className="grid gap-1.5 text-xs text-muted-foreground">
+                  SSH host
+                  <Input
+                    onChange={event => setForm(current => ({ ...current, sshHost: event.target.value }))}
+                    placeholder="user@host or host alias"
+                    value={form.sshHost}
+                  />
+                </label>
+                <label className="grid gap-1.5 text-xs text-muted-foreground">
+                  SSH user
+                  <Input
+                    onChange={event => setForm(current => ({ ...current, sshUser: event.target.value }))}
+                    placeholder="from SSH config"
+                    value={form.sshUser}
+                  />
+                </label>
+                <label className="grid gap-1.5 text-xs text-muted-foreground">
+                  Identity file
+                  <Input
+                    onChange={event => setForm(current => ({ ...current, sshKeyPath: event.target.value }))}
+                    placeholder="~/.ssh/id_ed25519"
+                    value={form.sshKeyPath}
+                  />
+                </label>
+                <label className="grid gap-1.5 text-xs text-muted-foreground">
+                  SSH port
+                  <Input
+                    inputMode="numeric"
+                    onChange={event => setForm(current => ({ ...current, sshPort: event.target.value }))}
+                    placeholder="22"
+                    value={form.sshPort}
+                  />
+                </label>
+              </div>
+            </div>
             <div className="grid gap-3 sm:grid-cols-[minmax(0,1fr)_12rem]">
               <label className="grid gap-1.5 text-xs text-muted-foreground">
                 Default Model
diff --git a/apps/desktop/src/types/hermes.ts b/apps/desktop/src/types/hermes.ts
index accbe39d999..ffbb16b69f3 100644
--- a/apps/desktop/src/types/hermes.ts
+++ b/apps/desktop/src/types/hermes.ts
@@ -161,6 +161,14 @@ export interface CustomEndpoint {
   models: string[]
   name: string
   source?: string
+  ssh_tunnel?: null | SshTunnelConfig
+}
+
+export interface SshTunnelConfig {
+  host: string
+  key_path?: string
+  port?: number
+  user?: string
 }
 
 export interface CustomEndpointsResponse {
@@ -184,6 +192,7 @@ export interface CustomEndpointUpdate {
   model: string
   models?: string[]
   name: string
+  ssh_tunnel?: Record<string, never> | SshTunnelConfig
 }
 
 export interface CustomEndpointValidationResponse {
diff --git a/hermes_cli/config.py b/hermes_cli/config.py
index 2a37a6d63fb..0972119dffc 100644
--- a/hermes_cli/config.py
+++ b/hermes_cli/config.py
@@ -5333,6 +5333,7 @@ def _normalize_custom_provider_entry(
         "context_length", "rate_limit_delay",
         "request_timeout_seconds", "stale_timeout_seconds",
         "discover_models", "extra_body", "extra_headers",
+        "ssh_tunnel",
         "ssl_ca_cert", "ssl_verify",
     }
     for camel, snake in _CAMEL_ALIASES.items():
@@ -5454,6 +5455,10 @@ def _normalize_custom_provider_entry(
     if isinstance(discover_models, bool):
         normalized["discover_models"] = discover_models
 
+    ssh_tunnel = entry.get("ssh_tunnel")
+    if isinstance(ssh_tunnel, dict):
+        normalized["ssh_tunnel"] = dict(ssh_tunnel)
+
     extra_body = entry.get("extra_body")
     if isinstance(extra_body, dict):
         normalized["extra_body"] = dict(extra_body)
@@ -5501,6 +5506,7 @@ def _custom_provider_entry_to_provider_config(
         "context_length",
         "rate_limit_delay",
         "discover_models",
+        "ssh_tunnel",
         "extra_body",
         "extra_headers",
         "ssl_ca_cert",
diff --git a/hermes_cli/runtime_provider.py b/hermes_cli/runtime_provider.py
index 3b3d0eafb8e..cbba9c5acc2 100644
--- a/hermes_cli/runtime_provider.py
+++ b/hermes_cli/runtime_provider.py
@@ -63,6 +63,15 @@ def _normalize_custom_provider_name(value: str) -> str:
     return value.strip().lower().replace(" ", "-")
 
 
+def _apply_ssh_tunnel(runtime: Dict[str, Any], ssh_tunnel: Any) -> Dict[str, Any]:
+    """Rewrite an SSH-backed custom endpoint to its managed loopback URL."""
+    if ssh_tunnel:
+        from hermes_cli.ssh_tunnel import resolve_ssh_tunnel_url
+
+        runtime["base_url"] = resolve_ssh_tunnel_url(runtime["base_url"], ssh_tunnel)
+    return runtime
+
+
 def _loopback_hostname(host: str) -> bool:
     h = (host or "").lower().rstrip(".")
     return h in {"localhost", "127.0.0.1", "::1", "0.0.0.0"}
@@ -719,6 +728,8 @@ def _get_named_custom_provider(requested_provider: str) -> Optional[Dict[str, An
                     api_mode = _parse_api_mode(entry.get("api_mode") or entry.get("transport"))
                     if api_mode:
                         result["api_mode"] = api_mode
+                    if isinstance(entry.get("ssh_tunnel"), dict):
+                        result["ssh_tunnel"] = dict(entry["ssh_tunnel"])
                     _lift_max_output_tokens(entry, result)
                     return result
             # Also check the 'name' field if present
@@ -742,6 +753,8 @@ def _get_named_custom_provider(requested_provider: str) -> Optional[Dict[str, An
                         api_mode = _parse_api_mode(entry.get("api_mode") or entry.get("transport"))
                         if api_mode:
                             result["api_mode"] = api_mode
+                        if isinstance(entry.get("ssh_tunnel"), dict):
+                            result["ssh_tunnel"] = dict(entry["ssh_tunnel"])
                         _lift_max_output_tokens(entry, result)
                         return result
 
@@ -1050,7 +1063,7 @@ def _resolve_named_custom_runtime(
         pool_result = _try_resolve_from_custom_pool(base_url, "custom", None)
         if pool_result:
             pool_result["source"] = "direct-alias"
-            return pool_result
+            return _apply_ssh_tunnel(pool_result, _get_model_config().get("ssh_tunnel"))
         _da_is_openai_url   = base_url_host_matches(base_url, "openai.com") or base_url_host_matches(base_url, "openai.azure.com")
         _da_is_openrouter   = base_url_host_matches(base_url, "openrouter.ai")
         api_key_candidates = [
@@ -1067,7 +1080,7 @@ def _resolve_named_custom_runtime(
             (c for c in api_key_candidates if has_usable_secret(c)),
             "",
         ) or "no-key-required"
-        return {
+        result = {
             "provider": "custom",
             "api_mode": _detect_api_mode_for_url(base_url) or "chat_completions",
             "base_url": base_url,
@@ -1075,6 +1088,7 @@ def _resolve_named_custom_runtime(
             "source": "direct-alias",
             "requested_provider": requested_provider,
         }
+        return _apply_ssh_tunnel(result, _get_model_config().get("ssh_tunnel"))
 
     custom_provider = _get_named_custom_provider(requested_provider)
     if not custom_provider:
@@ -1108,7 +1122,7 @@ def _resolve_named_custom_runtime(
         # credentials. NEVER log the values.
         if custom_provider.get("extra_headers"):
             pool_result["extra_headers"] = dict(custom_provider["extra_headers"])
-        return pool_result
+        return _apply_ssh_tunnel(pool_result, custom_provider.get("ssh_tunnel"))
 
     _cp_is_openai_url   = base_url_host_matches(base_url, "openai.com") or base_url_host_matches(base_url, "openai.azure.com")
     _cp_is_openrouter   = base_url_host_matches(base_url, "openrouter.ai")
@@ -1148,7 +1162,7 @@ def _resolve_named_custom_runtime(
     request_overrides = _custom_provider_request_overrides(custom_provider)
     if request_overrides:
         result["request_overrides"] = request_overrides
-    return result
+    return _apply_ssh_tunnel(result, custom_provider.get("ssh_tunnel"))
 
 
 def _resolve_openrouter_runtime(
@@ -1275,12 +1289,12 @@ def _resolve_openrouter_runtime(
             provider_name=requested_provider if requested_norm != "custom" else None,
         )
         if pool_result:
-            return pool_result
+            return _apply_ssh_tunnel(pool_result, model_cfg.get("ssh_tunnel"))
 
     if effective_provider == "custom" and not api_key and not _is_openrouter_url:
         api_key = "no-key-required"
 
-    return {
+    result = {
         "provider": effective_provider,
         "api_mode": _resolve_plain_custom_api_mode(model_cfg, base_url)
         if effective_provider == "custom"
@@ -1291,6 +1305,7 @@ def _resolve_openrouter_runtime(
         "api_key": api_key,
         "source": source,
     }
+    return _apply_ssh_tunnel(result, model_cfg.get("ssh_tunnel"))
 
 
 def _resolve_azure_foundry_runtime(
diff --git a/hermes_cli/ssh_tunnel.py b/hermes_cli/ssh_tunnel.py
new file mode 100644
index 00000000000..cee0d3975ad
--- /dev/null
+++ b/hermes_cli/ssh_tunnel.py
@@ -0,0 +1,191 @@
+"""Managed SSH tunnels for configured OpenAI-compatible endpoints.
+
+A tunnel is process-local and reused by every agent in the same Hermes process.
+The persisted provider URL remains the address as seen from the SSH host; only
+the runtime URL is rewritten to a loopback port selected by the kernel.
+"""
+
+from __future__ import annotations
+
+import atexit
+import hashlib
+import logging
+import os
+import shutil
+import socket
+import subprocess
+import threading
+import time
+from dataclasses import dataclass
+from typing import Any
+from urllib.parse import urlparse, urlunparse
+
+logger = logging.getLogger(__name__)
+
+_CONNECT_TIMEOUT_SECONDS = 15
+_START_TIMEOUT_SECONDS = 5
+_TUNNELS: dict[str, "ManagedSshTunnel"] = {}
+_TUNNELS_LOCK = threading.Lock()
+_CONTROL_CHARS = frozenset(chr(i) for i in (*range(32), 127))
+
+
+def _has_control_chars(value: str) -> bool:
+    return any(char in _CONTROL_CHARS for char in value)
+
+
+def _validate_ssh_value(name: str, value: str, *, required: bool = False) -> str:
+    clean = str(value or "").strip()
+    if required and not clean:
+        raise ValueError(f"SSH tunnel requires {name}.")
+    if clean and (_has_control_chars(clean) or clean.startswith("-")):
+        raise ValueError(f"Unsafe SSH tunnel {name}.")
+    return clean
+
+
+@dataclass(frozen=True)
+class SshTunnelConfig:
+    host: str
+    user: str = ""
+    port: int = 22
+    key_path: str = ""
+
+    @classmethod
+    def from_dict(cls, raw: Any) -> "SshTunnelConfig | None":
+        if not isinstance(raw, dict):
+            return None
+        host = _validate_ssh_value("host", raw.get("host", ""), required=True)
+        user = _validate_ssh_value("user", raw.get("user", ""))
+        if "@" in host and not user:
+            user, host = host.split("@", 1)
+            user = _validate_ssh_value("user", user, required=True)
+            host = _validate_ssh_value("host", host, required=True)
+        elif "@" in host:
+            raise ValueError("SSH tunnel host must not include a user when SSH user is set separately.")
+        key_path = _validate_ssh_value("key path", raw.get("key_path", ""))
+        try:
+            port = int(raw.get("port") or 22)
+        except (TypeError, ValueError) as exc:
+            raise ValueError("SSH tunnel port must be an integer between 1 and 65535.") from exc
+        if not 1 <= port <= 65535:
+            raise ValueError("SSH tunnel port must be between 1 and 65535.")
+        return cls(host=host, user=user, port=port, key_path=key_path)
+
+
+def _pick_local_port() -> int:
+    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
+        sock.bind(("127.0.0.1", 0))
+        return int(sock.getsockname()[1])
+
+
+def _endpoint_target(base_url: str) -> tuple[str, int, str]:
+    parsed = urlparse(str(base_url or "").strip())
+    if parsed.scheme not in {"http", "https"} or not parsed.hostname:
+        raise ValueError("SSH-tunneled endpoint URL must include http(s) scheme and host.")
+    try:
+        remote_port = parsed.port or (443 if parsed.scheme == "https" else 80)
+    except ValueError as exc:
+        raise ValueError("SSH-tunneled endpoint URL has an invalid port.") from exc
+    return parsed.hostname, remote_port, parsed.geturl().rstrip("/")
+
+
+class ManagedSshTunnel:
+    def __init__(self, config: SshTunnelConfig, remote_host: str, remote_port: int):
+        self.config = config
+        self.remote_host = remote_host
+        self.remote_port = remote_port
+        self.local_port: int | None = None
+        self.process: subprocess.Popen[bytes] | None = None
+
+    @property
+    def target(self) -> str:
+        return f"{self.config.user + '@' if self.config.user else ''}{self.config.host}"
+
+    def _args(self, local_port: int) -> list[str]:
+        args = [
+            "ssh",
+            "-N",
+            "-o", "BatchMode=yes",
+            "-o", "StrictHostKeyChecking=accept-new",
+            "-o", "ExitOnForwardFailure=yes",
+            "-o", f"ConnectTimeout={_CONNECT_TIMEOUT_SECONDS}",
+        ]
+        if self.config.port != 22:
+            args.extend(["-p", str(self.config.port)])
+        if self.config.key_path:
+            args.extend(["-i", self.config.key_path])
+        args.extend(["-L", f"127.0.0.1:{local_port}:{self.remote_host}:{self.remote_port}", "--", self.target])
+        return args
+
+    def start(self) -> int:
+        if self.local_port and self.process and self.process.poll() is None:
+            return self.local_port
+        if not shutil.which("ssh"):
+            raise RuntimeError("SSH is not installed or not in PATH. Install an OpenSSH client first.")
+        if self.config.key_path and not os.path.isfile(os.path.expanduser(self.config.key_path)):
+            raise RuntimeError(f"SSH identity file does not exist: {self.config.key_path}")
+
+        for _ in range(3):
+            local_port = _pick_local_port()
+            process = subprocess.Popen(
+                self._args(local_port),
+                stdin=subprocess.DEVNULL,
+                stdout=subprocess.DEVNULL,
+                stderr=subprocess.PIPE,
+            )
+            deadline = time.monotonic() + _START_TIMEOUT_SECONDS
+            while time.monotonic() < deadline:
+                if process.poll() is not None:
+                    process.communicate()
+                    raise RuntimeError("SSH tunnel failed to start. Check the SSH connection and credentials.")
+                try:
+                    with socket.create_connection(("127.0.0.1", local_port), timeout=0.15):
+                        self.process = process
+                        self.local_port = local_port
+                        logger.info("SSH tunnel ready on loopback port %s", local_port)
+                        return local_port
+                except OSError:
+                    time.sleep(0.05)
+            process.terminate()
+            try:
+                process.wait(timeout=1)
+            except subprocess.TimeoutExpired:
+                process.kill()
+        raise RuntimeError("SSH tunnel could not bind an automatically assigned local port.")
+
+    def close(self) -> None:
+        if self.process and self.process.poll() is None:
+            self.process.terminate()
+            try:
+                self.process.wait(timeout=3)
+            except subprocess.TimeoutExpired:
+                self.process.kill()
+        self.process = None
+        self.local_port = None
+
+
+def resolve_ssh_tunnel_url(base_url: str, ssh_tunnel: Any) -> str:
+    """Start/reuse a configured tunnel and return its loopback runtime URL."""
+    config = SshTunnelConfig.from_dict(ssh_tunnel)
+    if config is None:
+        return base_url.rstrip("/")
+    remote_host, remote_port, normalized_url = _endpoint_target(base_url)
+    identity = hashlib.sha256(repr((config, remote_host, remote_port)).encode()).hexdigest()
+    with _TUNNELS_LOCK:
+        tunnel = _TUNNELS.get(identity)
+        if tunnel is None:
+            tunnel = ManagedSshTunnel(config, remote_host, remote_port)
+            _TUNNELS[identity] = tunnel
+        local_port = tunnel.start()
+    parsed = urlparse(normalized_url)
+    return urlunparse(parsed._replace(netloc=f"127.0.0.1:{local_port}")).rstrip("/")
+
+
+def close_ssh_tunnels() -> None:
+    with _TUNNELS_LOCK:
+        tunnels = list(_TUNNELS.values())
+        _TUNNELS.clear()
+    for tunnel in tunnels:
+        tunnel.close()
+
+
+atexit.register(close_ssh_tunnels)
diff --git a/hermes_cli/web_server.py b/hermes_cli/web_server.py
index 497d740e1c1..4bdb033d5ea 100644
--- a/hermes_cli/web_server.py
+++ b/hermes_cli/web_server.py
@@ -1273,6 +1273,7 @@ class CustomEndpointUpdate(BaseModel):
     context_length: Optional[int] = None
     discover_models: bool = True
     make_default: bool = False
+    ssh_tunnel: Optional[Dict[str, Any]] = None
     models: Optional[List[str]] = None
 
 
@@ -7659,6 +7660,7 @@ def _custom_endpoint_response(cfg: Dict[str, Any]) -> Dict[str, Any]:
                 "models": models,
                 "context_length": raw_entry.get("context_length"),
                 "discover_models": bool(raw_entry.get("discover_models", True)),
+                "ssh_tunnel": raw_entry.get("ssh_tunnel") if isinstance(raw_entry.get("ssh_tunnel"), dict) else None,
                 "has_api_key": has_api_key,
                 "api_key_preview": api_key_preview,
                 "is_current": endpoint_id == current_provider,
@@ -7675,6 +7677,7 @@ def _custom_endpoint_response(cfg: Dict[str, Any]) -> Dict[str, Any]:
             "models": [current_model] if current_model else [],
             "context_length": model_cfg.get("context_length"),
             "discover_models": True,
+            "ssh_tunnel": model_cfg.get("ssh_tunnel") if isinstance(model_cfg.get("ssh_tunnel"), dict) else None,
             "has_api_key": has_api_key,
             "api_key_preview": api_key_preview,
             "is_current": True,
@@ -7751,6 +7754,16 @@ def _write_custom_endpoint(cfg: Dict[str, Any], body: CustomEndpointUpdate) -> T
         "model": model,
         "discover_models": bool(body.discover_models),
     })
+    if body.ssh_tunnel:
+        try:
+            from hermes_cli.ssh_tunnel import SshTunnelConfig
+
+            SshTunnelConfig.from_dict(body.ssh_tunnel)
+        except ValueError as exc:
+            raise HTTPException(status_code=400, detail=str(exc)) from exc
+        entry["ssh_tunnel"] = dict(body.ssh_tunnel)
+    elif body.ssh_tunnel == {}:
+        entry.pop("ssh_tunnel", None)
     # Same for the model map: merge rather than replace, so existing models
     # keep their context lengths. ``body.models`` is the catalogue the panel's
     # Test button already discovered — without it only the one hand-typed
@@ -7901,6 +7914,13 @@ async def validate_custom_endpoint(body: CustomEndpointUpdate):
     if not base_url:
         return {"ok": False, "reachable": True, "message": "Enter an endpoint URL first.", "models": []}
 
+    try:
+        from hermes_cli.ssh_tunnel import resolve_ssh_tunnel_url
+
+        base_url = resolve_ssh_tunnel_url(base_url, body.ssh_tunnel)
+    except (RuntimeError, ValueError) as exc:
+        return {"ok": False, "reachable": False, "message": str(exc), "models": []}
+
     url = base_url + "/models"
     headers = {"Accept": "application/json"}
     if body.api_key and body.api_key.strip():
diff --git a/tests/hermes_cli/test_runtime_provider_resolution.py b/tests/hermes_cli/test_runtime_provider_resolution.py
index f6e1347bd4e..32cf4d3b990 100644
--- a/tests/hermes_cli/test_runtime_provider_resolution.py
+++ b/tests/hermes_cli/test_runtime_provider_resolution.py
@@ -2084,6 +2084,29 @@ def test_named_custom_runtime_propagates_extra_body_direct_path(monkeypatch):
     }
 
 
+def test_named_custom_runtime_uses_managed_ssh_tunnel(monkeypatch):
+    monkeypatch.setattr(rp, "resolve_provider", lambda *a, **k: "remote-vllm")
+    monkeypatch.setattr(
+        rp,
+        "_get_named_custom_provider",
+        lambda _p: {
+            "name": "remote-vllm",
+            "base_url": "http://127.0.0.1:30090/v1",
+            "api_key": "test-key",
+            "ssh_tunnel": {"host": "gpu.example", "key_path": "/tmp/key"},
+        },
+    )
+    monkeypatch.setattr(rp, "_try_resolve_from_custom_pool", lambda *a, **k: None)
+    monkeypatch.setattr(
+        "hermes_cli.ssh_tunnel.resolve_ssh_tunnel_url",
+        lambda base_url, tunnel: "http://127.0.0.1:49152/v1",
+    )
+
+    resolved = rp.resolve_runtime_provider(requested="remote-vllm")
+
+    assert resolved["base_url"] == "http://127.0.0.1:49152/v1"
+
+
 def test_named_custom_runtime_propagates_model_pool_path(monkeypatch):
     """Model should propagate even when credential pool handles credentials."""
     monkeypatch.setattr(rp, "resolve_provider", lambda *a, **k: "my-server")
diff --git a/tests/hermes_cli/test_ssh_tunnel.py b/tests/hermes_cli/test_ssh_tunnel.py
new file mode 100644
index 00000000000..21d9133d5ae
--- /dev/null
+++ b/tests/hermes_cli/test_ssh_tunnel.py
@@ -0,0 +1,73 @@
+from hermes_cli import ssh_tunnel
+
+
+def test_tunnel_config_accepts_user_at_host_and_explicit_key():
+    config = ssh_tunnel.SshTunnelConfig.from_dict({
+        "host": "ari@example.test",
+        "port": 2222,
+        "key_path": "/tmp/key",
+    })
+
+    assert config == ssh_tunnel.SshTunnelConfig(
+        host="example.test", user="ari", port=2222, key_path="/tmp/key"
+    )
+    agent_config = ssh_tunnel.SshTunnelConfig.from_dict({"host": "example.test"})
+    assert agent_config is not None
+    assert agent_config.key_path == ""
+
+
+def test_tunnel_url_rewrites_only_the_runtime_host(monkeypatch):
+    ssh_tunnel.close_ssh_tunnels()
+    started = []
+
+    class LiveProcess:
+        def __init__(self):
+            self.alive = True
+
+        def poll(self):
+            return None if self.alive else 0
+
+        def terminate(self):
+            self.alive = False
+
+        def wait(self, timeout=None):
+            return 0
+
+        def kill(self):
+            self.alive = False
+
+    def fake_start(self):
+        if self.local_port and self.process and self.process.poll() is None:
+            return self.local_port
+        started.append(self)
+        self.local_port = 45678
+        self.process = LiveProcess()
+        return self.local_port
+
+    monkeypatch.setattr(ssh_tunnel.ManagedSshTunnel, "start", fake_start)
+
+    url = ssh_tunnel.resolve_ssh_tunnel_url(
+        "http://127.0.0.1:30090/v1",
+        {"host": "remote.example", "user": "ari", "key_path": "/tmp/key"},
+    )
+
+    assert url == "http://127.0.0.1:45678/v1"
+    assert len(started) == 1
+    assert started[0].remote_host == "127.0.0.1"
+    assert started[0].remote_port == 30090
+
+    assert ssh_tunnel.resolve_ssh_tunnel_url(
+        "http://127.0.0.1:30090/v1",
+        {"host": "remote.example", "user": "ari", "key_path": "/tmp/key"},
+    ) == url
+    assert len(started) == 1
+    ssh_tunnel.close_ssh_tunnels()
+
+
+def test_tunnel_config_rejects_user_twice():
+    try:
+        ssh_tunnel.SshTunnelConfig.from_dict({"host": "ari@example.test", "user": "other"})
+    except ValueError as exc:
+        assert "must not include a user" in str(exc)
+    else:
+        raise AssertionError("unsafe SSH target accepted")