diff --git a/agent/memory_provider.py b/agent/memory_provider.py
index a73a9a475e..559fc3df6c 100644
--- a/agent/memory_provider.py
+++ b/agent/memory_provider.py
@@ -291,6 +291,10 @@ class MemoryProvider(ABC):
           required:    True if required (default: False)
           default:     default value (optional)
           choices:     list of valid values (optional)
+          type:        text, integer, number, or boolean (optional)
+          minimum:     numeric lower bound for integer/number fields (optional)
+          maximum:     numeric upper bound for integer/number fields (optional)
+          step:        numeric input step for Dashboard rendering (optional)
           url:         URL where user can get this credential (optional)
           env_var:     explicit env var name for secrets (default: auto-generated)
 
diff --git a/hermes_cli/web_server.py b/hermes_cli/web_server.py
index a6678d04f2..d3d3ca2249 100644
--- a/hermes_cli/web_server.py
+++ b/hermes_cli/web_server.py
@@ -5573,6 +5573,12 @@ def _normalize_memory_provider_schema(name: str, provider: Any) -> List[Dict[str
             kind = "select"
         elif explicit_kind in {"bool", "boolean"} or isinstance(raw.get("default"), bool):
             kind = "boolean"
+        elif explicit_kind in {"int", "integer"} or (
+            isinstance(raw.get("default"), int) and not isinstance(raw.get("default"), bool)
+        ):
+            kind = "integer"
+        elif explicit_kind in {"float", "number"} or isinstance(raw.get("default"), float):
+            kind = "number"
         else:
             kind = "text"
 
@@ -5593,6 +5599,9 @@ def _normalize_memory_provider_schema(name: str, provider: Any) -> List[Dict[str
             "options": options,
             "url": str(raw.get("url") or ""),
             "when": raw.get("when") if isinstance(raw.get("when"), dict) else None,
+            "minimum": raw.get("minimum"),
+            "maximum": raw.get("maximum"),
+            "step": raw.get("step"),
             "_env_key": str(raw.get("env_var") or "") or None,
         })
 
@@ -5736,6 +5745,9 @@ def _public_memory_provider_field(field: Dict[str, Any], data: Dict[str, Any]) -
         "options": field.get("options", []),
         "url": field.get("url", ""),
         "when": field.get("when"),
+        "minimum": field.get("minimum"),
+        "maximum": field.get("maximum"),
+        "step": field.get("step"),
     }
     return entry
 
@@ -5758,6 +5770,31 @@ def _coerce_schema_field(field: Dict[str, Any], raw: Any) -> Any:
     if field["kind"] == "boolean":
         return _coerce_bool(raw, default=_coerce_bool(_field_default(field), default=False))
 
+    if field["kind"] in {"integer", "number"}:
+        value = raw if raw is not None and raw != "" else _field_default(field)
+        try:
+            if isinstance(value, bool):
+                raise ValueError
+            parsed = float(value)
+            if not math.isfinite(parsed):
+                raise ValueError
+            if field["kind"] == "integer":
+                if not parsed.is_integer():
+                    raise ValueError
+                result: int | float = int(parsed)
+            else:
+                result = parsed
+        except (TypeError, ValueError, OverflowError) as exc:
+            raise ValueError(f"Invalid numeric value for '{field['key']}'") from exc
+
+        minimum = field.get("minimum")
+        maximum = field.get("maximum")
+        if minimum is not None and result < minimum:
+            raise ValueError(f"'{field['key']}' must be at least {minimum}")
+        if maximum is not None and result > maximum:
+            raise ValueError(f"'{field['key']}' must be at most {maximum}")
+        return result
+
     value = str(raw if raw is not None else "").strip()
     if field["kind"] == "select":
         if not value:
diff --git a/plugins/memory/openviking/__init__.py b/plugins/memory/openviking/__init__.py
index 57dc4cbe92..ff3793a0b8 100644
--- a/plugins/memory/openviking/__init__.py
+++ b/plugins/memory/openviking/__init__.py
@@ -42,6 +42,7 @@ import time
 import uuid
 import zipfile
 from dataclasses import dataclass, replace
+from functools import lru_cache
 from pathlib import Path
 from typing import Any, Callable, Dict, List, Optional, Set
 from urllib.parse import quote, unquote, urlparse
@@ -131,6 +132,9 @@ _LOCAL_OPENVIKING_AUTOSTART_TIMEOUT = 60.0
 # is refused in well under this; it exists only so a wedged listener cannot
 # block the autostart path.
 _LOCAL_OPENVIKING_PROBE_TIMEOUT = 2.0
+_LOCAL_SERVER_STARTED = "started"
+_LOCAL_SERVER_OCCUPIED = "occupied"
+_LOCAL_SERVER_FAILED = "failed"
 # After a refresh attempt fails for a given (unchanged) config, skip re-probing
 # for this long. Keeps "unavailable endpoints reconnect on a later access"
 # true while preventing every provider access from paying a 3s health probe
@@ -143,6 +147,8 @@ _RUN_LOCKS_RELATIVE_DIR = Path("openviking") / "runs"
 _LEGACY_RECOVERY_LOCK_FILENAME = "legacy-recovery.lock"
 _LOCK_BUSY_ERRNOS = {errno.EWOULDBLOCK, errno.EACCES, errno.EAGAIN}
 _SETUP_CANCELLED = object()
+_INVALID_SETTING_WARNINGS: Set[tuple[str, str]] = set()
+_INVALID_SETTING_WARNINGS_LOCK = threading.Lock()
 
 
 @dataclass(frozen=True)
@@ -161,6 +167,10 @@ class _OpenVikingHTTPError(RuntimeError):
         self.status_code = status_code
 
 
+class _OpenVikingEndpointError(ValueError):
+    """Raised when a configured endpoint cannot be used safely."""
+
+
 def _sanitize_openviking_error_message(message: str, status_code: Optional[int] = None) -> str:
     text = (message or "").strip()
     status = f"HTTP {status_code}" if status_code else "HTTP error"
@@ -414,10 +424,7 @@ class _VikingClient:
 
     def health(self) -> bool:
         try:
-            resp = self._httpx.get(
-                self._url("/health"), headers=self._headers(), timeout=3.0
-            )
-            return resp.status_code == 200
+            return _is_openviking_health_payload(self.health_payload())
         except Exception:
             return False
 
@@ -724,6 +731,27 @@ def _clean_config_value(value: Any) -> str:
     return value.strip() if isinstance(value, str) else ""
 
 
+def _openviking_endpoint_label(value: Any) -> str:
+    """Return a credential-free endpoint label suitable for logs and UI."""
+    raw = _clean_config_value(value)
+    if not raw:
+        return "<empty endpoint>"
+    try:
+        parsed = urlparse(raw if "://" in raw else f"//{raw}")
+        host = parsed.hostname
+        if not host:
+            return "<configured endpoint>"
+        display_host = f"[{host}]" if ":" in host and not host.startswith("[") else host
+        try:
+            port = parsed.port
+        except ValueError:
+            port = None
+        scheme = f"{parsed.scheme}://" if parsed.scheme else ""
+        return f"{scheme}{display_host}{f':{port}' if port is not None else ''}"
+    except Exception:
+        return "<configured endpoint>"
+
+
 def _default_ovcli_config_path() -> Path:
     return Path.home() / _OVCLI_DEFAULT_RELATIVE_PATH
 
@@ -753,13 +781,16 @@ def _load_ovcli_config(path: Optional[Path] = None) -> dict:
 
 
 def _connection_values_from_ovcli(data: dict) -> dict:
+    endpoint_value = _clean_config_value(data.get("url"))
     api_key = _clean_config_value(data.get("api_key")) or _clean_config_value(data.get("root_api_key"))
     root_api_key = _clean_config_value(data.get("root_api_key"))
     send_identity = not api_key or api_key == root_api_key
     account = _clean_config_value(data.get("account") or data.get("account_id"))
     user = _clean_config_value(data.get("user") or data.get("user_id"))
     return {
-        "endpoint": _normalize_openviking_url(data.get("url")),
+        # A linked profile with no URL contributes no endpoint; the resolver
+        # can then continue to config.yaml and finally the built-in default.
+        "endpoint": _normalize_openviking_url(endpoint_value) if endpoint_value else "",
         "api_key": api_key,
         "root_api_key": root_api_key,
         "account": account if send_identity else "",
@@ -793,53 +824,88 @@ def _validate_openviking_identity_value(value: str, *, field: str) -> tuple[bool
     return True, "", trimmed
 
 
+@lru_cache(maxsize=128)
+def _openviking_endpoint_is_always_blocked(candidate: str) -> bool:
+    """Check the safety floor once per configured endpoint value.
+
+    Endpoint resolution is configuration work, but the live provider resolves
+    its settings on every access so Dashboard and ``/reload`` changes take
+    effect without a restart. Caching by the complete endpoint keeps that hot
+    path from repeating potentially slow DNS lookups; changing the configured
+    URL still produces a fresh validation.
+    """
+    from tools.url_safety import is_always_blocked_url
+
+    return is_always_blocked_url(candidate)
+
+
 def _normalize_openviking_url(url: str) -> str:
     trimmed = _clean_config_value(url).rstrip("/")
     if not trimmed:
         return _DEFAULT_ENDPOINT
     lower = trimmed.lower()
     if lower in {"::1", "[::1]"}:
-        return "http://[::1]:1933"
-    if lower.startswith("[::1]:"):
-        return f"http://[::1]:{trimmed.rsplit(':', 1)[1]}"
-    if lower.startswith("::1:"):
-        return f"http://[::1]:{trimmed.rsplit(':', 1)[1]}"
-    if "://" in trimmed:
+        candidate = "http://[::1]:1933"
+    elif lower.startswith("[::1]:") or lower.startswith("::1:"):
+        candidate = f"http://[::1]:{trimmed.rsplit(':', 1)[1]}"
+    elif "://" in trimmed:
         candidate = trimmed
     else:
-        host, _sep, port = trimmed.partition(":")
-        if host.lower() in {"localhost", "127.0.0.1"}:
-            candidate = f"http://{host}:{port or '1933'}"
-        else:
-            candidate = trimmed
+        candidate = f"http://{trimmed}"
+
+    try:
+        parsed = urlparse(candidate)
+        if parsed.scheme.lower() not in {"http", "https"} or not parsed.hostname:
+            raise ValueError("OpenViking endpoints must use http:// or https:// with a host.")
+        # Force validation of malformed ports (``urlparse`` defers it).
+        parsed.port
+        if parsed.username or parsed.password or parsed.query or parsed.fragment:
+            raise ValueError(
+                "OpenViking endpoints cannot contain user info, query parameters, or fragments."
+            )
+    except ValueError as exc:
+        raise _OpenVikingEndpointError(
+            f"Invalid OpenViking endpoint {_openviking_endpoint_label(candidate)}: {exc}"
+        ) from exc
 
     # Local / LAN self-host remains allowed; reject cloud-metadata and other
     # always-blocked floors so a poisoned endpoint cannot SSRF via memory sync.
+    # Never silently replace an explicitly unsafe endpoint with localhost: that
+    # could attach Hermes to an unrelated deployment and forward credentials to
+    # a destination the user did not configure.
     try:
-        from tools.url_safety import is_always_blocked_url
-
         check_url = candidate if "://" in candidate else f"http://{candidate}"
-        if is_always_blocked_url(check_url):
-            logger.warning(
-                "OpenViking endpoint '%s' targets an always-blocked address; "
-                "falling back to the default local endpoint.",
-                candidate,
+        if _openviking_endpoint_is_always_blocked(check_url):
+            raise _OpenVikingEndpointError(
+                "OpenViking endpoint "
+                f"{_openviking_endpoint_label(candidate)} targets a blocked metadata address."
             )
-            return _DEFAULT_ENDPOINT
+    except _OpenVikingEndpointError:
+        raise
     except Exception as exc:
-        logger.warning(
-            "OpenViking always-blocked endpoint check failed; "
-            "falling back to the default local endpoint: %s",
-            exc,
-        )
-        return _DEFAULT_ENDPOINT
+        logger.debug("OpenViking endpoint safety validation failed", exc_info=True)
+        raise _OpenVikingEndpointError(
+            "OpenViking endpoint safety validation failed; Hermes refused the connection."
+        ) from exc
 
     return candidate
 
 
+def _is_openviking_health_payload(payload: Any) -> bool:
+    """Match OpenViking's documented ``GET /health`` response contract."""
+    return (
+        isinstance(payload, dict)
+        and payload.get("status") == "ok"
+        and payload.get("healthy") is True
+        and isinstance(payload.get("version"), str)
+        and bool(payload["version"].strip())
+    )
+
+
 def _load_profile(path: Path, *, source: str, name: str) -> Optional[_OvcliProfile]:
     try:
         data = _load_ovcli_config(path)
+        values = _connection_values_from_ovcli(data)
     except Exception as e:
         logger.debug("Skipping invalid OpenViking CLI config %s: %s", path, e)
         return None
@@ -848,7 +914,7 @@ def _load_profile(path: Path, *, source: str, name: str) -> Optional[_OvcliProfi
         name=name,
         path=path,
         data=data,
-        values=_connection_values_from_ovcli(data),
+        values=values,
     )
 
 
@@ -914,7 +980,10 @@ def _discover_ovcli_profiles() -> list[_OvcliProfile]:
 
 
 def _is_local_openviking_url(value: str) -> bool:
-    candidate = _normalize_openviking_url(value)
+    try:
+        candidate = _normalize_openviking_url(value)
+    except _OpenVikingEndpointError:
+        return False
     if not candidate:
         return False
     if "://" not in candidate:
@@ -1110,10 +1179,11 @@ def _validate_openviking_reachability(endpoint: str) -> tuple[bool, str]:
         client = _VikingClient(endpoint)
         if hasattr(client, "health_payload"):
             payload = client.health_payload()
-            if payload.get("healthy") is False:
+            if isinstance(payload, dict) and payload.get("healthy") is False:
                 return False, "OpenViking server responded but reported unhealthy status."
-            if payload:
+            if _is_openviking_health_payload(payload):
                 return True, ""
+            return False, "OpenViking server responded, but its /health response is not valid OpenViking."
         elif client.health():
             return True, ""
     except Exception as e:
@@ -1205,8 +1275,10 @@ def _validate_openviking_setup_values(
             agent=_clean_config_value(values.get("agent")) or _DEFAULT_AGENT,
         )
         health = client.health_payload()
-        if health.get("healthy") is False:
+        if isinstance(health, dict) and health.get("healthy") is False:
             return False, "OpenViking server responded but reported unhealthy status.", None
+        if not _is_openviking_health_payload(health):
+            return False, "Server /health response is not valid OpenViking.", None
         if _should_probe_openviking_auth(
             health,
             require_api_key=require_api_key,
@@ -1278,21 +1350,81 @@ def _local_openviking_port_is_open(host: str, port: int) -> bool:
         return False
 
 
-def _start_local_openviking_server(endpoint: str) -> tuple[bool, str]:
+def _describe_local_port_listener(host: str, port: int) -> str:
+    """Best-effort process identity for an occupied local TCP port."""
+    try:
+        import psutil
+
+        wildcard_hosts = {"0.0.0.0", "::", "::0"}
+        aliases = {host.lower()}
+        if host.lower() == "localhost":
+            aliases.update({"127.0.0.1", "::1"})
+        for conn in psutil.net_connections(kind="inet"):
+            if conn.status != psutil.CONN_LISTEN or not conn.laddr:
+                continue
+            listener_host = str(
+                conn.laddr.ip if hasattr(conn.laddr, "ip") else conn.laddr[0]
+            ).lower()
+            listener_port = int(
+                conn.laddr.port if hasattr(conn.laddr, "port") else conn.laddr[1]
+            )
+            if listener_port != port:
+                continue
+            if listener_host not in wildcard_hosts and listener_host not in aliases:
+                continue
+            if conn.pid is None:
+                break
+            try:
+                process_name = psutil.Process(conn.pid).name()
+            except (psutil.Error, OSError):
+                process_name = "unknown process"
+            process_name = re.sub(r"[^\w .+-]", "?", str(process_name))[:80]
+            return f"{process_name or 'unknown process'} (PID {conn.pid})"
+    except Exception:
+        logger.debug(
+            "Could not identify the process listening on %s:%s",
+            host,
+            port,
+            exc_info=True,
+        )
+    return "an unidentified process"
+
+
+def _local_listener_suffix(endpoint: str) -> str:
+    if not _is_local_openviking_url(endpoint):
+        return ""
+    try:
+        host, port = _local_openviking_bind(endpoint)
+    except ValueError:
+        return ""
+    if not _local_openviking_port_is_open(host, port):
+        return ""
+    return f" The listener on {host}:{port} is {_describe_local_port_listener(host, port)}."
+
+
+def _start_local_openviking_server(endpoint: str) -> tuple[str, str]:
     try:
         host, port = _local_openviking_bind(endpoint)
     except ValueError as e:
-        return False, f"Could not parse local OpenViking URL: {e}"
+        return _LOCAL_SERVER_FAILED, f"Could not parse local OpenViking URL: {e}"
     # Health probes can time out client-side while the server is up and well.
     # Spawning on that signal alone produces a process that immediately dies on
     # DataDirectoryLocked, and — because the probe keeps timing out — repeats
-    # every cooldown window. Treat an occupied port as "already started": both
-    # callers only need the server running, not started by us.
+    # every cooldown window. Treat an occupied port only as a spawn-prevention
+    # signal, never as proof that the listener is OpenViking.
     if _local_openviking_port_is_open(host, port):
-        return True, f"openviking-server is already running on {host}:{port}."
+        listener = _describe_local_port_listener(host, port)
+        return (
+            _LOCAL_SERVER_OCCUPIED,
+            f"Port {host}:{port} is occupied by {listener}. Hermes did not start "
+            "openviking-server because the listener has not passed OpenViking's /health check.",
+        )
     server_cmd = shutil.which("openviking-server")
     if not server_cmd:
-        return False, "openviking-server was not found on PATH. Start it manually, then retry."
+        return (
+            _LOCAL_SERVER_FAILED,
+            "openviking-server was not found on PATH. Start it manually, then retry.",
+        )
     log_path = _openviking_server_log_path()
     try:
         log_path.parent.mkdir(parents=True, exist_ok=True)
@@ -1305,8 +1437,11 @@ def _start_local_openviking_server(endpoint: str) -> tuple[bool, str]:
                 start_new_session=True,
             )
     except Exception as e:
-        return False, f"Could not start openviking-server: {e}"
-    return True, f"Started openviking-server on {host}:{port} in the background. Logs: {log_path}"
+        return _LOCAL_SERVER_FAILED, f"Could not start openviking-server: {e}"
+    return (
+        _LOCAL_SERVER_STARTED,
+        f"Started openviking-server on {host}:{port} in the background. Logs: {log_path}",
+    )
 
 
 def _wait_for_openviking_health(
@@ -1355,9 +1490,9 @@ def _handle_unreachable_endpoint(
             cancel_returns=cancelled,
         )
         if choice == 0:
-            started, start_message = _start_local_openviking_server(endpoint)
+            start_state, start_message = _start_local_openviking_server(endpoint)
             print(f"  {start_message}")
-            if not started:
+            if start_state != _LOCAL_SERVER_STARTED:
                 return False
             print("  Waiting for OpenViking server to become reachable...", flush=True)
             if _wait_for_openviking_health(
@@ -1403,7 +1538,7 @@ def _runtime_openviking_timeout_message(endpoint: str) -> str:
         f"Local OpenViking server at {endpoint} is not reachable. "
         "Tried to start openviking-server, but it did not become reachable "
         f"within {_LOCAL_OPENVIKING_AUTOSTART_TIMEOUT:.0f} seconds. "
-        "OpenViking memory disabled; will retry on a later access or when "
+        "OpenViking memory is temporarily unavailable; Hermes will retry on a later access or when "
         "the config changes."
     )
 
@@ -1413,18 +1548,26 @@ def _classify_runtime_openviking_health(client: _VikingClient, endpoint: str) ->
     try:
         if hasattr(client, "health_payload"):
             payload = client.health_payload()
-            if payload.get("healthy") is False:
+            if isinstance(payload, dict) and payload.get("healthy") is False:
                 return (
                     "responded",
-                    f"OpenViking server at {endpoint} responded but reported unhealthy status.",
+                    f"Service at {endpoint} responded but reported unhealthy OpenViking status."
+                    f"{_local_listener_suffix(endpoint)}",
                 )
-            return "healthy", ""
+            if _is_openviking_health_payload(payload):
+                return "healthy", ""
+            return (
+                "responded",
+                f"Service at {endpoint} responded, but its /health response is not valid OpenViking."
+                f"{_local_listener_suffix(endpoint)}",
+            )
         if client.health():
             return "healthy", ""
     except _OpenVikingHTTPError as e:
         return (
             "responded",
-            f"OpenViking server at {endpoint} responded with {_format_openviking_exception(e)}.",
+            f"Service at {endpoint} responded with {_format_openviking_exception(e)}."
+            f"{_local_listener_suffix(endpoint)}",
         )
     except Exception:
         return "unreachable", ""
@@ -1478,7 +1621,20 @@ def _prompt_manual_connection_values(prompt, select, cancelled, *, service: bool
         print(f"  OpenViking Service endpoint: {endpoint}")
     else:
         while True:
-            endpoint = _normalize_openviking_url(prompt("OpenViking server URL", default=_DEFAULT_ENDPOINT))
+            try:
+                endpoint = _normalize_openviking_url(
+                    prompt("OpenViking server URL", default=_DEFAULT_ENDPOINT)
+                )
+            except _OpenVikingEndpointError as exc:
+                retry = _retry_or_cancel_manual_setup(
+                    select,
+                    "  Invalid OpenViking endpoint",
+                    str(exc),
+                    cancelled,
+                )
+                if retry is _SETUP_CANCELLED:
+                    return _SETUP_CANCELLED
+                continue
             _print_validation_progress("Checking OpenViking server...")
             reachable, message = _validate_openviking_reachability(endpoint)
             if reachable:
@@ -1513,16 +1669,16 @@ def _prompt_manual_connection_values(prompt, select, cancelled, *, service: bool
             credential_choice = select(
                 "  OpenViking credential",
                 [
-                    ("No API key", "local dev mode"),
-                    ("User API key", "server derives account/user automatically"),
+                    ("User API key", "recommended; server derives account/user automatically"),
                     ("Root API key", "requires account and user IDs"),
+                    ("No API key", "only for explicitly unauthenticated local development"),
                 ],
                 default=0,
                 cancel_returns=cancelled,
             )
             if credential_choice == cancelled:
                 return _SETUP_CANCELLED
-            if credential_choice == 0:
+            if credential_choice == 2:
                 values["agent"] = _clean_config_value(
                     prompt(_AGENT_PROMPT_LABEL, default=_DEFAULT_AGENT)
                 ) or _DEFAULT_AGENT
@@ -1540,7 +1696,7 @@ def _prompt_manual_connection_values(prompt, select, cancelled, *, service: bool
                 if retry is _SETUP_CANCELLED:
                     return _SETUP_CANCELLED
                 continue
-            api_key_type = "root" if credential_choice == 2 else "user"
+            api_key_type = "root" if credential_choice == 1 else "user"
         elif not api_key_type:
             credential_choice = select(
                 "  OpenViking API key type",
@@ -2024,18 +2180,21 @@ class OpenVikingMemoryProvider(MemoryProvider):
             },
             {
                 "key": "api_key",
-                "description": "OpenViking API key (leave blank for local dev mode)",
+                "description": (
+                    "OpenViking API key (recommended; only leave blank for an explicitly "
+                    "unauthenticated local development server)"
+                ),
                 "secret": True,
                 "env_var": "OPENVIKING_API_KEY",
             },
             {
                 "key": "account",
-                "description": "OpenViking tenant account ID (blank for user API keys)",
+                "description": "Advanced local identity override (leave blank for user API keys)",
                 "env_var": "OPENVIKING_ACCOUNT",
             },
             {
                 "key": "user",
-                "description": "OpenViking user ID within the account (blank for user API keys)",
+                "description": "Advanced local user override (leave blank for user API keys)",
                 "env_var": "OPENVIKING_USER",
             },
             {
@@ -2050,59 +2209,108 @@ class OpenVikingMemoryProvider(MemoryProvider):
             {
                 "key": "recall_limit",
                 "description": "Maximum memories injected by automatic recall",
+                "type": "integer",
+                "minimum": 1,
+                "maximum": 100,
                 "default": _DEFAULT_RECALL_LIMIT,
                 "env_var": "OPENVIKING_RECALL_LIMIT",
             },
             {
                 "key": "recall_score_threshold",
                 "description": "Minimum relevance score for automatic recall",
+                "type": "number",
+                "minimum": 0.0,
+                "maximum": 1.0,
+                "step": 0.01,
                 "default": _DEFAULT_RECALL_SCORE_THRESHOLD,
                 "env_var": "OPENVIKING_RECALL_SCORE_THRESHOLD",
             },
             {
                 "key": "recall_max_injected_chars",
                 "description": "Maximum total characters injected by recall",
+                "type": "integer",
+                "minimum": 100,
+                "maximum": 50000,
                 "default": _DEFAULT_RECALL_MAX_INJECTED_CHARS,
                 "env_var": "OPENVIKING_RECALL_MAX_INJECTED_CHARS",
             },
             {
                 "key": "profile_token_budget",
                 "description": "Maximum session-start memory tokens injected",
+                "type": "integer",
+                "minimum": 500,
+                "maximum": 50000,
                 "default": _DEFAULT_PROFILE_TOKEN_BUDGET,
                 "env_var": "OPENVIKING_PROFILE_TOKEN_BUDGET",
             },
             {
                 "key": "recall_timeout_seconds",
                 "description": "Total timeout for recall (seconds)",
+                "type": "number",
+                "minimum": 0.25,
+                "maximum": 60.0,
+                "step": 0.25,
                 "default": _DEFAULT_RECALL_TIMEOUT_SECONDS,
                 "env_var": "OPENVIKING_RECALL_TIMEOUT_SECONDS",
             },
             {
                 "key": "recall_request_timeout_seconds",
                 "description": "Per-request timeout for recall (seconds)",
+                "type": "number",
+                "minimum": 0.25,
+                "maximum": 60.0,
+                "step": 0.25,
                 "default": _DEFAULT_RECALL_REQUEST_TIMEOUT_SECONDS,
                 "env_var": "OPENVIKING_RECALL_REQUEST_TIMEOUT_SECONDS",
             },
             {
                 "key": "recall_full_read_limit",
                 "description": "Max full L2 content reads per recall",
+                "type": "integer",
+                "minimum": 0,
+                "maximum": 100,
                 "default": _DEFAULT_RECALL_FULL_READ_LIMIT,
                 "env_var": "OPENVIKING_RECALL_FULL_READ_LIMIT",
             },
             {
                 "key": "recall_prefer_abstract",
                 "description": "Use abstracts instead of full L2 reads",
+                "type": "boolean",
                 "default": False,
                 "env_var": "OPENVIKING_RECALL_PREFER_ABSTRACT",
             },
             {
                 "key": "recall_resources",
                 "description": "Include resources in recall",
+                "type": "boolean",
                 "default": False,
                 "env_var": "OPENVIKING_RECALL_RESOURCES",
             },
         ]
 
+    def save_config(self, values: Dict[str, Any], hermes_home: str) -> None:
+        """Validate and persist Dashboard configuration for the active profile."""
+        normalized = dict(values or {})
+        normalized.pop("api_key", None)
+        normalized.pop("root_api_key", None)
+        endpoint = _clean_config_value(normalized.get("endpoint"))
+        if endpoint:
+            normalized["endpoint"] = _normalize_openviking_url(endpoint)
+
+        from hermes_cli.config import load_config, save_config
+
+        config = load_config()
+        memory_config = config.get("memory")
+        if not isinstance(memory_config, dict):
+            memory_config = {}
+            config["memory"] = memory_config
+        provider_config = memory_config.get("openviking")
+        if not isinstance(provider_config, dict):
+            provider_config = {}
+        provider_config.update(normalized)
+        memory_config["openviking"] = provider_config
+        save_config(config)
+
     def get_status_config(self, provider_config: dict) -> dict:
         provider_config = dict(provider_config or {})
         if provider_config.get("use_ovcli_config"):
@@ -2264,7 +2472,7 @@ class OpenVikingMemoryProvider(MemoryProvider):
                 if not healthy:
                     warning_message = (
                         f"OpenViking server at {endpoint} is still not reachable after auto-start. "
-                        "OpenViking memory disabled; will retry on a later access or when "
+                        "OpenViking memory is temporarily unavailable; Hermes will retry on a later access or when "
                         "the config changes."
                     )
                 else:
@@ -2283,7 +2491,7 @@ class OpenVikingMemoryProvider(MemoryProvider):
             except Exception as e:
                 warning_message = (
                     f"OpenViking server at {endpoint} could not be attached after auto-start: {e}. "
-                    "OpenViking memory disabled; will retry on a later access or when "
+                    "OpenViking memory is temporarily unavailable; Hermes will retry on a later access or when "
                     "the config changes."
                 )
 
@@ -2309,7 +2517,7 @@ class OpenVikingMemoryProvider(MemoryProvider):
         if not _is_local_openviking_url(endpoint):
             _emit_runtime_warning(
                 f"Remote OpenViking server at {endpoint} is not reachable. "
-                "OpenViking memory disabled; will retry on a later access or when "
+                "OpenViking memory is temporarily unavailable; Hermes will retry on a later access or when "
                 "the config changes. "
                 "Check the configured endpoint and network connectivity.",
                 warning_callback,
@@ -2330,12 +2538,12 @@ class OpenVikingMemoryProvider(MemoryProvider):
                 return
 
             self._runtime_start_pending = True
-            started, start_message = _start_local_openviking_server(endpoint)
-            if not started:
+            start_state, start_message = _start_local_openviking_server(endpoint)
+            if start_state != _LOCAL_SERVER_STARTED:
                 self._runtime_start_pending = False
                 warning_message = (
                     f"Local OpenViking server at {endpoint} is not reachable. {start_message} "
-                    "OpenViking memory disabled; will retry on a later access or when "
+                    "OpenViking memory is temporarily unavailable; Hermes will retry on a later access or when "
                     "the config changes."
                 )
                 self._client = None
@@ -2366,7 +2574,28 @@ class OpenVikingMemoryProvider(MemoryProvider):
                 )
 
     def initialize(self, session_id: str, **kwargs) -> None:
-        settings = _resolve_connection_settings(_load_hermes_openviking_config())
+        warning_callback = (
+            kwargs.get("warning_callback")
+            if kwargs.get("platform") == "cli"
+            else None
+        )
+        status_callback = (
+            kwargs.get("status_callback")
+            if kwargs.get("platform") == "cli"
+            else None
+        )
+        connection_error = ""
+        try:
+            settings = _resolve_connection_settings(_load_hermes_openviking_config())
+        except _OpenVikingEndpointError as exc:
+            connection_error = str(exc)
+            settings = {
+                "endpoint": "",
+                "api_key": "",
+                "account": "",
+                "user": "",
+                "agent": _DEFAULT_AGENT,
+            }
         self._endpoint = settings["endpoint"]
         self._api_key = settings["api_key"]
         self._account = settings["account"]
@@ -2389,38 +2618,43 @@ class OpenVikingMemoryProvider(MemoryProvider):
         self._hermes_home = hermes_home
         self._acquire_run_lock()
         self._profile_prefetched_sessions.clear()
-        warning_callback = (
-            kwargs.get("warning_callback")
-            if kwargs.get("platform") == "cli"
-            else None
-        )
-        status_callback = (
-            kwargs.get("status_callback")
-            if kwargs.get("platform") == "cli"
-            else None
-        )
 
-        try:
-            self._client = _VikingClient(
-                self._endpoint, self._api_key,
-                account=self._account, user=self._user, agent=self._agent,
+        if connection_error:
+            self._failed_refresh = (
+                ("invalid-endpoint", connection_error),
+                time.monotonic(),
+            )
+            _emit_runtime_warning(
+                f"{connection_error} OpenViking memory is temporarily unavailable; "
+                "correct the endpoint and reload the configuration.",
+                warning_callback,
             )
-            health_state, health_message = _classify_runtime_openviking_health(self._client, self._endpoint)
-            if health_state == "unreachable":
-                self._handle_runtime_openviking_unreachable(
-                    status_callback=status_callback,
-                    warning_callback=warning_callback,
-                )
-            elif health_state != "healthy":
-                _emit_runtime_warning(
-                    f"{health_message} OpenViking memory disabled; will retry on a "
-                    "later access or when the config changes.",
-                    warning_callback,
-                )
-                self._client = None
-        except ImportError:
-            logger.warning("httpx not installed — OpenViking plugin disabled")
             self._client = None
+        else:
+            try:
+                self._client = _VikingClient(
+                    self._endpoint, self._api_key,
+                    account=self._account, user=self._user, agent=self._agent,
+                )
+                health_state, health_message = _classify_runtime_openviking_health(
+                    self._client,
+                    self._endpoint,
+                )
+                if health_state == "unreachable":
+                    self._handle_runtime_openviking_unreachable(
+                        status_callback=status_callback,
+                        warning_callback=warning_callback,
+                    )
+                elif health_state != "healthy":
+                    _emit_runtime_warning(
+                        f"{health_message} OpenViking memory is temporarily unavailable; "
+                        "Hermes will retry on a later access or when the config changes.",
+                        warning_callback,
+                    )
+                    self._client = None
+            except ImportError:
+                logger.warning("httpx not installed — OpenViking plugin disabled")
+                self._client = None
 
         if self._client:
             self._conn_snapshot = (
@@ -2459,7 +2693,25 @@ class OpenVikingMemoryProvider(MemoryProvider):
             self._client = None
             return None
 
-        settings = _resolve_connection_settings(_load_hermes_openviking_config())
+        try:
+            settings = _resolve_connection_settings(_load_hermes_openviking_config())
+        except _OpenVikingEndpointError as exc:
+            failed_key = ("invalid-endpoint", str(exc))
+            failed = self._failed_refresh
+            should_warn = not (
+                failed is not None
+                and failed[0] == failed_key
+                and time.monotonic() - failed[1] < _FAILED_CONFIG_RETRY_COOLDOWN_SECONDS
+            )
+            self._failed_refresh = (failed_key, time.monotonic())
+            self._client = None
+            if should_warn:
+                logger.warning(
+                    "%s OpenViking memory is temporarily unavailable; correct the endpoint "
+                    "and reload the configuration.",
+                    exc,
+                )
+            return None
         endpoint = settings["endpoint"]
         api_key = settings["api_key"]
         account = settings["account"]
@@ -2520,8 +2772,8 @@ class OpenVikingMemoryProvider(MemoryProvider):
         self._failed_refresh = (settings_key, time.monotonic())
         if health_state == "responded":
             logger.warning(
-                "%s OpenViking memory disabled; will retry on a later access "
-                "(after cooldown) or when the config changes.",
+                "%s OpenViking memory is temporarily unavailable; Hermes will retry on a "
+                "later access (after cooldown) or when the config changes.",
                 health_message,
             )
         else:  # unreachable
@@ -3238,29 +3490,86 @@ class OpenVikingMemoryProvider(MemoryProvider):
             return ""
 
     @staticmethod
-    def _env_bool(name: str, default: bool = False) -> bool:
-        raw = os.environ.get(name)
-        if raw is None or raw == "":
-            return default
-        return raw.strip().lower() in {"1", "true", "yes", "on"}
+    def _warn_invalid_setting_once(source: str, value: Any, default: Any) -> None:
+        warning_key = (source, repr(value))
+        with _INVALID_SETTING_WARNINGS_LOCK:
+            if warning_key in _INVALID_SETTING_WARNINGS:
+                return
+            _INVALID_SETTING_WARNINGS.add(warning_key)
+        logger.warning("Invalid %s value %r; using default %r.", source, value, default)
 
     @staticmethod
-    def _env_int(name: str, default: int, *, minimum: int, maximum: int) -> int:
-        raw = os.environ.get(name)
-        try:
-            value = int(float(raw)) if raw not in {None, ""} else default
-        except (TypeError, ValueError):
-            value = default
-        return max(minimum, min(maximum, value))
+    def _setting_value(env_name: str, config_value: Any) -> tuple[Any, str]:
+        env_value = os.environ.get(env_name)
+        if env_value is not None and env_value.strip():
+            return env_value, env_name
+        config_key = env_name.removeprefix("OPENVIKING_").lower()
+        return config_value, f"memory.openviking.{config_key}"
 
-    @staticmethod
-    def _env_float(name: str, default: float, *, minimum: float, maximum: float) -> float:
-        raw = os.environ.get(name)
+    @classmethod
+    def _setting_bool(
+        cls,
+        env_name: str,
+        config_value: Any,
+        *,
+        default: bool,
+    ) -> bool:
+        value, source = cls._setting_value(env_name, config_value)
+        if isinstance(value, bool):
+            return value
+        if isinstance(value, str):
+            normalized = value.strip().lower()
+            if normalized in {"1", "true", "yes", "on"}:
+                return True
+            if normalized in {"0", "false", "no", "off"}:
+                return False
+        cls._warn_invalid_setting_once(source, value, default)
+        return default
+
+    @classmethod
+    def _setting_int(
+        cls,
+        env_name: str,
+        config_value: Any,
+        *,
+        default: int,
+        minimum: int,
+        maximum: int,
+    ) -> int:
+        value, source = cls._setting_value(env_name, config_value)
         try:
-            value = float(raw) if raw not in {None, ""} else default
-        except (TypeError, ValueError):
-            value = default
-        return max(minimum, min(maximum, value))
+            if isinstance(value, bool):
+                raise ValueError
+            numeric = float(value)
+            if not numeric.is_integer() or not float("-inf") < numeric < float("inf"):
+                raise ValueError
+            parsed = int(numeric)
+        except (TypeError, ValueError, OverflowError):
+            cls._warn_invalid_setting_once(source, value, default)
+            parsed = default
+        return max(minimum, min(maximum, parsed))
+
+    @classmethod
+    def _setting_float(
+        cls,
+        env_name: str,
+        config_value: Any,
+        *,
+        default: float,
+        minimum: float,
+        maximum: float,
+    ) -> float:
+        value, source = cls._setting_value(env_name, config_value)
+        try:
+            if isinstance(value, bool):
+                raise ValueError
+            parsed = float(value)
+            if not float("-inf") < parsed < float("inf"):
+                raise ValueError
+        except (TypeError, ValueError, OverflowError):
+            cls._warn_invalid_setting_once(source, value, default)
+            parsed = default
+        return max(minimum, min(maximum, parsed))
 
     def _recall_config(self) -> Dict[str, Any]:
         # Read from config.yaml → memory.openviking as primary source, env vars
@@ -3269,46 +3578,60 @@ class OpenVikingMemoryProvider(MemoryProvider):
         cfg = provider_config
 
         return {
-            "limit": self._env_int(
+            "limit": self._setting_int(
                 "OPENVIKING_RECALL_LIMIT",
                 cfg.get("recall_limit", _DEFAULT_RECALL_LIMIT),
+                default=_DEFAULT_RECALL_LIMIT,
                 minimum=1, maximum=100,
             ),
-            "score_threshold": self._env_float(
+            "score_threshold": self._setting_float(
                 "OPENVIKING_RECALL_SCORE_THRESHOLD",
                 cfg.get("recall_score_threshold", _DEFAULT_RECALL_SCORE_THRESHOLD),
+                default=_DEFAULT_RECALL_SCORE_THRESHOLD,
                 minimum=0.0, maximum=1.0,
             ),
-            "max_injected_chars": self._env_int(
+            "max_injected_chars": self._setting_int(
                 "OPENVIKING_RECALL_MAX_INJECTED_CHARS",
                 cfg.get("recall_max_injected_chars", _DEFAULT_RECALL_MAX_INJECTED_CHARS),
+                default=_DEFAULT_RECALL_MAX_INJECTED_CHARS,
                 minimum=100, maximum=50000,
             ),
-            "timeout_seconds": self._env_float(
+            "timeout_seconds": self._setting_float(
                 "OPENVIKING_RECALL_TIMEOUT_SECONDS",
                 cfg.get("recall_timeout_seconds", _DEFAULT_RECALL_TIMEOUT_SECONDS),
+                default=_DEFAULT_RECALL_TIMEOUT_SECONDS,
                 minimum=0.25, maximum=60.0,
             ),
-            "request_timeout_seconds": self._env_float(
+            "request_timeout_seconds": self._setting_float(
                 "OPENVIKING_RECALL_REQUEST_TIMEOUT_SECONDS",
                 cfg.get("recall_request_timeout_seconds", _DEFAULT_RECALL_REQUEST_TIMEOUT_SECONDS),
+                default=_DEFAULT_RECALL_REQUEST_TIMEOUT_SECONDS,
                 minimum=0.25, maximum=60.0,
             ),
-            "full_read_limit": self._env_int(
+            "full_read_limit": self._setting_int(
                 "OPENVIKING_RECALL_FULL_READ_LIMIT",
                 cfg.get("recall_full_read_limit", _DEFAULT_RECALL_FULL_READ_LIMIT),
+                default=_DEFAULT_RECALL_FULL_READ_LIMIT,
                 minimum=0, maximum=100,
             ),
-            "prefer_abstract": self._env_bool("OPENVIKING_RECALL_PREFER_ABSTRACT",
-                cfg.get("recall_prefer_abstract", False)),
-            "resources": self._env_bool("OPENVIKING_RECALL_RESOURCES",
-                cfg.get("recall_resources", False)),
+            "prefer_abstract": self._setting_bool(
+                "OPENVIKING_RECALL_PREFER_ABSTRACT",
+                cfg.get("recall_prefer_abstract", False),
+                default=False,
+            ),
+            "resources": self._setting_bool(
+                "OPENVIKING_RECALL_RESOURCES",
+                cfg.get("recall_resources", False),
+                default=False,
+            ),
         }
 
     def _profile_token_budget(self) -> int:
-        return self._env_int(
+        cfg = _load_hermes_openviking_config()
+        return self._setting_int(
             "OPENVIKING_PROFILE_TOKEN_BUDGET",
-            _DEFAULT_PROFILE_TOKEN_BUDGET,
+            cfg.get("profile_token_budget", _DEFAULT_PROFILE_TOKEN_BUDGET),
+            default=_DEFAULT_PROFILE_TOKEN_BUDGET,
             minimum=500,
             maximum=50000,
         )
@@ -4267,6 +4590,12 @@ class OpenVikingMemoryProvider(MemoryProvider):
             if rotate:
                 self._session_id = new_id
                 self._turn_count = 0
+            elif compression:
+                # commit_memory_session() has already extracted every turn up
+                # to this boundary. Keep the same sid, but start the live
+                # session's turn accounting again at zero so an immediate
+                # session end cannot duplicate the just-finished extraction.
+                self._turn_count = 0
 
         if compression:
             # Discard both old and new session IDs so the profile is re-injected
@@ -4291,8 +4620,8 @@ class OpenVikingMemoryProvider(MemoryProvider):
                 self._clear_session_committed(old_session_id)
 
         if not rotate:
-            # Same-session rewind (/undo) or no-op rotation: no commit and no
-            # counter reset.
+            # Same-session rewind (/undo) or no-op rotation: no new commit.
+            # Compression already reset the extracted-turn count above.
             logger.debug(
                 "OpenViking on_session_switch skipped rotation: session=%s rewound=%s",
                 old_session_id, rewound,
diff --git a/plugins/memory/retaindb/__init__.py b/plugins/memory/retaindb/__init__.py
index c83a283a40..9c24d4d334 100644
--- a/plugins/memory/retaindb/__init__.py
+++ b/plugins/memory/retaindb/__init__.py
@@ -52,9 +52,9 @@ def _load_retaindb_config() -> Dict[str, Any]:
     unset. The secret ``api_key`` continues to come from the environment.
     """
     try:
-        from hermes_cli.config import load_config
+        from hermes_cli.config import load_config_readonly
 
-        config = load_config()
+        config = load_config_readonly()
         memory_config = config.get("memory", {}) if isinstance(config, dict) else {}
         provider_config = memory_config.get("retaindb", {}) if isinstance(memory_config, dict) else {}
         return dict(provider_config) if isinstance(provider_config, dict) else {}
diff --git a/tests/hermes_cli/test_web_server.py b/tests/hermes_cli/test_web_server.py
index fc4a41c4c0..4c0b5500f1 100644
--- a/tests/hermes_cli/test_web_server.py
+++ b/tests/hermes_cli/test_web_server.py
@@ -142,7 +142,7 @@ class TestReloadEnv:
     def test_adds_new_vars(self, tmp_path):
         """reload_env() adds vars from .env that are not in os.environ."""
         env_file = tmp_path / ".env"
-        env_file.write_text("TEST_RELOAD_VAR=hello123\n")
+        env_file.write_text("TEST_RELOAD_VAR=hello123\n", encoding="utf-8")
         with patch.dict(reload_env.__globals__, {"get_env_path": lambda: env_file}):
             os.environ.pop("TEST_RELOAD_VAR", None)
             count = reload_env()
@@ -523,6 +523,79 @@ class TestWebServerEndpoints:
     def _provider_field_map(payload):
         return {field["key"]: field for field in payload["fields"]}
 
+    def test_openviking_recall_fields_are_numeric_dashboard_controls(self):
+        resp = self.client.get("/api/memory/providers/openviking/config")
+
+        assert resp.status_code == 200
+        fields = self._provider_field_map(resp.json())
+        assert fields["recall_limit"]["kind"] == "integer"
+        assert fields["recall_limit"]["minimum"] == 1
+        assert fields["recall_limit"]["maximum"] == 100
+        assert fields["recall_score_threshold"]["kind"] == "number"
+        assert fields["recall_score_threshold"]["step"] == 0.01
+        assert fields["recall_resources"]["kind"] == "boolean"
+
+    def test_openviking_dashboard_persists_typed_recall_values(self):
+        from hermes_cli.config import load_config
+
+        resp = self.client.put(
+            "/api/memory/providers/openviking/config",
+            json={
+                "values": {
+                    "endpoint": "http://127.0.0.1:1933",
+                    "recall_limit": "12",
+                    "recall_score_threshold": "0.42",
+                    "recall_max_injected_chars": "8000",
+                    "profile_token_budget": "7000",
+                    "recall_timeout_seconds": "2.5",
+                    "recall_request_timeout_seconds": "1.5",
+                    "recall_full_read_limit": "5",
+                    "recall_prefer_abstract": True,
+                    "recall_resources": False,
+                }
+            },
+        )
+
+        assert resp.status_code == 200
+        config = load_config()["memory"]["openviking"]
+        assert config["recall_limit"] == 12
+        assert config["recall_score_threshold"] == 0.42
+        assert config["profile_token_budget"] == 7000
+        assert config["recall_prefer_abstract"] is True
+        assert config["recall_resources"] is False
+
+    def test_openviking_dashboard_rejects_out_of_range_recall_value(self):
+        resp = self.client.put(
+            "/api/memory/providers/openviking/config",
+            json={
+                "values": {
+                    "endpoint": "http://127.0.0.1:1933",
+                    "recall_limit": 101,
+                }
+            },
+        )
+
+        assert resp.status_code == 400
+        assert "must be at most 100" in resp.json()["detail"]
+
+    def test_openviking_dashboard_rejects_blocked_endpoint_before_saving(self):
+        from hermes_cli.config import load_config
+
+        resp = self.client.put(
+            "/api/memory/providers/openviking/config",
+            json={
+                "values": {
+                    "endpoint": "http://169.254.169.254/latest/meta-data/credential",
+                }
+            },
+        )
+
+        assert resp.status_code == 400
+        assert "blocked metadata address" in resp.json()["detail"]
+        assert "credential" not in resp.json()["detail"]
+        memory_config = load_config().get("memory", {})
+        assert "openviking" not in memory_config
+
 
 
 
@@ -2764,7 +2837,7 @@ class TestDiscoverUserThemes:
         monkeypatch.setenv("HERMES_HOME", str(tmp_path))
         themes_dir = tmp_path / "dashboard-themes"
         themes_dir.mkdir()
-        (themes_dir / "mine.yaml").write_text("name: mine\n")
+        (themes_dir / "mine.yaml").write_text("name: mine\n", encoding="utf-8")
 
         other = tmp_path / "other-profile"
         other.mkdir()
@@ -3307,7 +3380,7 @@ class TestDashboardPluginManifestExtensions:
         import json
         plug_dir = tmp_path / "plugins" / name / "dashboard"
         plug_dir.mkdir(parents=True)
-        (plug_dir / "manifest.json").write_text(json.dumps(manifest))
+        (plug_dir / "manifest.json").write_text(json.dumps(manifest), encoding="utf-8")
         return plug_dir
 
     def test_override_and_hidden_carried_through(self, tmp_path, monkeypatch):
diff --git a/tests/openviking_plugin/test_openviking.py b/tests/openviking_plugin/test_openviking.py
index 8c0e39ad4a..7bdd525a72 100644
--- a/tests/openviking_plugin/test_openviking.py
+++ b/tests/openviking_plugin/test_openviking.py
@@ -16,7 +16,8 @@ def _write_skill(skills_dir, name, body="Do the thing."):
     skill_dir = skills_dir / name
     skill_dir.mkdir(parents=True, exist_ok=True)
     (skill_dir / "SKILL.md").write_text(
-        f"---\nname: {name}\ndescription: Description for {name}\n---\n\n# {name}\n\n{body}\n"
+        f"---\nname: {name}\ndescription: Description for {name}\n---\n\n# {name}\n\n{body}\n",
+        encoding="utf-8",
     )
     return skill_dir
 
@@ -25,7 +26,10 @@ def _write_bundle(bundles_dir, slug, skills):
     bundles_dir.mkdir(parents=True, exist_ok=True)
     lines = [f"name: {slug}", "skills:"]
     lines.extend(f"  - {skill}" for skill in skills)
-    (bundles_dir / f"{slug}.yaml").write_text("\n".join(lines) + "\n")
+    (bundles_dir / f"{slug}.yaml").write_text(
+        "\n".join(lines) + "\n",
+        encoding="utf-8",
+    )
 
 
 class FakeVikingClient:
@@ -266,6 +270,7 @@ class TestOpenVikingConfigSchema:
         provider = OpenVikingMemoryProvider()
 
         schema = provider.get_config_schema()
+        fields = {entry["key"]: entry for entry in schema}
         env_vars = {entry.get("env_var") for entry in schema}
 
         assert "OPENVIKING_RECALL_LIMIT" in env_vars
@@ -276,6 +281,11 @@ class TestOpenVikingConfigSchema:
         assert "OPENVIKING_RECALL_FULL_READ_LIMIT" in env_vars
         assert "OPENVIKING_RECALL_PREFER_ABSTRACT" in env_vars
         assert "OPENVIKING_RECALL_RESOURCES" in env_vars
+        assert fields["recall_limit"]["type"] == "integer"
+        assert fields["recall_limit"]["minimum"] == 1
+        assert fields["recall_limit"]["maximum"] == 100
+        assert fields["recall_score_threshold"]["type"] == "number"
+        assert fields["recall_prefer_abstract"]["type"] == "boolean"
         assert provider._recall_config() == {
             "limit": 6,
             "score_threshold": 0.15,
@@ -294,19 +304,23 @@ class TestOpenVikingConfigSchema:
         hermes_home = tmp_path / "hermes_test"
         hermes_home.mkdir(exist_ok=True)
         config_yaml = hermes_home / "config.yaml"
-        config_yaml.write_text("""\
+        config_yaml.write_text(
+            """\
 memory:
   provider: openviking
   openviking:
     recall_limit: 12
     recall_score_threshold: 0.42
     recall_max_injected_chars: 8000
+    profile_token_budget: 7000
     recall_timeout_seconds: 2.0
     recall_request_timeout_seconds: 1.5
     recall_full_read_limit: 5
     recall_prefer_abstract: true
     recall_resources: true
-""")
+""",
+            encoding="utf-8",
+        )
         monkeypatch.setenv("HERMES_HOME", str(hermes_home))
         # Clear any OPENVIKING_RECALL_* env vars so config.yaml prevails
         for key in list(os.environ):
@@ -324,6 +338,7 @@ memory:
         assert cfg["full_read_limit"] == 5
         assert cfg["prefer_abstract"] is True
         assert cfg["resources"] is True
+        assert provider._profile_token_budget() == 7000
 
     def test_recall_config_env_overrides_config_yaml(self, monkeypatch, tmp_path):
         """Env vars OPENVIKING_RECALL_* take precedence over config.yaml values
@@ -331,13 +346,16 @@ memory:
         hermes_home = tmp_path / "hermes_test"
         hermes_home.mkdir(exist_ok=True)
         config_yaml = hermes_home / "config.yaml"
-        config_yaml.write_text("""\
+        config_yaml.write_text(
+            """\
 memory:
   provider: openviking
   openviking:
     recall_limit: 12
     recall_resources: true
-""")
+""",
+            encoding="utf-8",
+        )
         monkeypatch.setenv("HERMES_HOME", str(hermes_home))
         # Override config.yaml via env
         monkeypatch.setenv("OPENVIKING_RECALL_LIMIT", "6")
@@ -355,13 +373,16 @@ memory:
         hermes_home = tmp_path / "hermes_test"
         hermes_home.mkdir(exist_ok=True)
         config_yaml = hermes_home / "config.yaml"
-        config_yaml.write_text("""\
+        config_yaml.write_text(
+            """\
 memory:
   provider: openviking
   openviking:
     recall_limit: 3
     # No recall_resources set — should use default (False)
-""")
+""",
+            encoding="utf-8",
+        )
         monkeypatch.setenv("HERMES_HOME", str(hermes_home))
         for key in list(os.environ):
             if key.startswith("OPENVIKING_RECALL_"):
@@ -374,6 +395,68 @@ memory:
         assert cfg["resources"] is False, "omitted key should use default"
         assert cfg["timeout_seconds"] == 4.0, "omitted key should use built-in default"
 
+    def test_dashboard_shaped_string_values_are_typed(self, monkeypatch):
+        for key in list(os.environ):
+            if key.startswith("OPENVIKING_RECALL_") or key == "OPENVIKING_PROFILE_TOKEN_BUDGET":
+                monkeypatch.delenv(key, raising=False)
+        monkeypatch.setattr(
+            openviking_plugin,
+            "_load_hermes_openviking_config",
+            lambda: {
+                "recall_limit": "12",
+                "recall_score_threshold": "0.42",
+                "recall_prefer_abstract": "false",
+                "recall_resources": "true",
+                "profile_token_budget": "7500",
+            },
+        )
+        provider = OpenVikingMemoryProvider()
+
+        cfg = provider._recall_config()
+
+        assert cfg["limit"] == 12
+        assert cfg["score_threshold"] == 0.42
+        assert cfg["prefer_abstract"] is False
+        assert cfg["resources"] is True
+        assert provider._profile_token_budget() == 7500
+
+    def test_invalid_recall_values_fall_back_without_type_errors(self, monkeypatch):
+        for key in list(os.environ):
+            if key.startswith("OPENVIKING_RECALL_") or key == "OPENVIKING_PROFILE_TOKEN_BUDGET":
+                monkeypatch.delenv(key, raising=False)
+        monkeypatch.setattr(
+            openviking_plugin,
+            "_load_hermes_openviking_config",
+            lambda: {
+                "recall_limit": "many",
+                "recall_score_threshold": True,
+                "recall_prefer_abstract": "sometimes",
+                "profile_token_budget": "7.5",
+            },
+        )
+        provider = OpenVikingMemoryProvider()
+
+        cfg = provider._recall_config()
+
+        assert cfg["limit"] == 6
+        assert cfg["score_threshold"] == 0.15
+        assert cfg["prefer_abstract"] is False
+        assert provider._profile_token_budget() == 6000
+
+    def test_recall_env_overrides_string_config_with_native_types(self, monkeypatch):
+        monkeypatch.setattr(
+            openviking_plugin,
+            "_load_hermes_openviking_config",
+            lambda: {"recall_limit": "12", "recall_resources": "false"},
+        )
+        monkeypatch.setenv("OPENVIKING_RECALL_LIMIT", "4")
+        monkeypatch.setenv("OPENVIKING_RECALL_RESOURCES", "true")
+
+        cfg = OpenVikingMemoryProvider()._recall_config()
+
+        assert cfg["limit"] == 4
+        assert cfg["resources"] is True
+
 
 class TestOpenVikingTurnConversion:
     def test_extract_current_turn_anchors_on_latest_matching_user_and_assistant(self):
@@ -569,7 +652,7 @@ class TestOpenVikingAutoRecallPrefetch:
             def do_GET(self):
                 parsed = urlparse(self.path)
                 if parsed.path == "/health":
-                    self._send_json({"healthy": True})
+                    self._send_json({"status": "ok", "healthy": True, "version": "test"})
                     return
                 if parsed.path == "/api/v1/content/read":
                     query = parse_qs(parsed.query)
@@ -933,7 +1016,7 @@ class TestEnsureClientReloadsEnv:
                 start_calls.append(endpoint)
             first_start_entered.set()
             release_start.wait(timeout=2)
-            return True, "started"
+            return openviking_plugin._LOCAL_SERVER_STARTED, "started"
 
         monkeypatch.setattr(openviking_plugin, "_start_local_openviking_server", start_local)
         monkeypatch.setattr(
@@ -1085,7 +1168,10 @@ class TestUnavailableWarningsPromiseRetry:
         monkeypatch.setattr(
             openviking_plugin,
             "_start_local_openviking_server",
-            lambda endpoint: (False, "openviking-server was not found on PATH."),
+            lambda endpoint: (
+                openviking_plugin._LOCAL_SERVER_FAILED,
+                "openviking-server was not found on PATH.",
+            ),
         )
         provider = OpenVikingMemoryProvider()
         provider._endpoint = "http://127.0.0.1:1934"
diff --git a/tests/plugins/memory/test_openviking_endpoint_always_blocked.py b/tests/plugins/memory/test_openviking_endpoint_always_blocked.py
index 271d4c1cc5..25b48ec820 100644
--- a/tests/plugins/memory/test_openviking_endpoint_always_blocked.py
+++ b/tests/plugins/memory/test_openviking_endpoint_always_blocked.py
@@ -1,10 +1,17 @@
 """OpenViking endpoint always-blocked floor."""
 
-from plugins.memory.openviking import _DEFAULT_ENDPOINT, _normalize_openviking_url
+import pytest
+
+from plugins.memory.openviking import (
+    _OpenVikingEndpointError,
+    _normalize_openviking_url,
+    _openviking_endpoint_is_always_blocked,
+)
 
 
 def test_openviking_blocks_metadata_endpoint():
-    assert _normalize_openviking_url("http://169.254.169.254/") == _DEFAULT_ENDPOINT
+    with pytest.raises(_OpenVikingEndpointError, match="blocked metadata address"):
+        _normalize_openviking_url("http://169.254.169.254/")
 
 
 def test_openviking_keeps_default_loopback():
@@ -12,7 +19,37 @@ def test_openviking_keeps_default_loopback():
 
 
 def test_openviking_blocks_ecs_metadata_hostname():
-    assert (
+    with pytest.raises(_OpenVikingEndpointError, match="blocked metadata address"):
         _normalize_openviking_url("http://metadata.google.internal/computeMetadata/v1/")
-        == _DEFAULT_ENDPOINT
+
+
+def test_openviking_rejects_endpoint_credentials_and_query():
+    with pytest.raises(_OpenVikingEndpointError, match="cannot contain user info"):
+        _normalize_openviking_url("https://user:secret@example.com?api_key=secret")
+
+
+def test_openviking_validates_shorthand_ipv6_port():
+    assert _normalize_openviking_url("::1:1934") == "http://[::1]:1934"
+    with pytest.raises(_OpenVikingEndpointError, match="Port could not be cast"):
+        _normalize_openviking_url("::1:not-a-port")
+
+
+def test_openviking_caches_safety_check_for_unchanged_endpoint(monkeypatch):
+    import tools.url_safety as url_safety
+
+    calls = []
+    _openviking_endpoint_is_always_blocked.cache_clear()
+    monkeypatch.setattr(
+        url_safety,
+        "is_always_blocked_url",
+        lambda value: calls.append(value) or False,
     )
+
+    assert _normalize_openviking_url("https://openviking.example.test") == (
+        "https://openviking.example.test"
+    )
+    assert _normalize_openviking_url("https://openviking.example.test") == (
+        "https://openviking.example.test"
+    )
+    assert calls == ["https://openviking.example.test"]
+    _openviking_endpoint_is_always_blocked.cache_clear()
diff --git a/tests/plugins/memory/test_openviking_provider.py b/tests/plugins/memory/test_openviking_provider.py
index b73a93b2a8..52700c7a4a 100644
--- a/tests/plugins/memory/test_openviking_provider.py
+++ b/tests/plugins/memory/test_openviking_provider.py
@@ -117,12 +117,41 @@ def test_openviking_provider_config_loader_uses_readonly_config(monkeypatch):
     assert config is not backing_config["memory"]["openviking"]
 
 
+def test_connection_settings_read_dashboard_config_file(tmp_path, monkeypatch):
+    _clear_openviking_env(monkeypatch)
+    hermes_home = tmp_path / "hermes"
+    hermes_home.mkdir()
+    (hermes_home / "config.yaml").write_text(
+        """\
+memory:
+  provider: openviking
+  openviking:
+    endpoint: http://saved.test:1933
+    account: saved-account
+    user: saved-user
+    agent: saved-agent
+""",
+        encoding="utf-8",
+    )
+    monkeypatch.setenv("HERMES_HOME", str(hermes_home))
+
+    settings = openviking_module._resolve_connection_settings(
+        openviking_module._load_hermes_openviking_config()
+    )
+
+    assert settings["endpoint"] == "http://saved.test:1933"
+    assert settings["account"] == "saved-account"
+    assert settings["user"] == "saved-user"
+    assert settings["agent"] == "saved-agent"
+    assert settings["api_key"] == ""
+
+
 def test_linked_ovcli_config_is_read_at_runtime(tmp_path, monkeypatch):
     _clear_openviking_env(monkeypatch)
     ovcli_path = tmp_path / "ovcli.conf"
     ovcli_path.write_text(
         json.dumps({
-            "url": "http://openviking-one.local",
+            "url": "http://openviking-one.test",
             "api_key": "key-one",
             "account": "acct-one",
             "user": "alice",
@@ -135,7 +164,7 @@ def test_linked_ovcli_config_is_read_at_runtime(tmp_path, monkeypatch):
     settings = openviking_module._resolve_connection_settings(provider_config)
 
     assert settings == {
-        "endpoint": "http://openviking-one.local",
+        "endpoint": "http://openviking-one.test",
         "api_key": "key-one",
         "account": "",
         "user": "",
@@ -144,7 +173,7 @@ def test_linked_ovcli_config_is_read_at_runtime(tmp_path, monkeypatch):
 
     ovcli_path.write_text(
         json.dumps({
-            "url": "http://openviking-two.local",
+            "url": "http://openviking-two.test",
             "api_key": "key-two",
             "agent_id": "agent-two",
         }),
@@ -154,7 +183,7 @@ def test_linked_ovcli_config_is_read_at_runtime(tmp_path, monkeypatch):
     settings = openviking_module._resolve_connection_settings(provider_config)
 
     assert settings == {
-        "endpoint": "http://openviking-two.local",
+        "endpoint": "http://openviking-two.test",
         "api_key": "key-two",
         "account": "",
         "user": "",
@@ -162,6 +191,38 @@ def test_linked_ovcli_config_is_read_at_runtime(tmp_path, monkeypatch):
     }
 
 
+def test_linked_ovcli_without_url_falls_through_to_dashboard_endpoint(tmp_path, monkeypatch):
+    _clear_openviking_env(monkeypatch)
+    ovcli_path = tmp_path / "ovcli.conf"
+    ovcli_path.write_text(json.dumps({"api_key": "linked-key"}), encoding="utf-8")
+
+    settings = openviking_module._resolve_connection_settings({
+        "use_ovcli_config": True,
+        "ovcli_config_path": str(ovcli_path),
+        "endpoint": "http://saved.test:1933",
+    })
+
+    assert settings["endpoint"] == "http://saved.test:1933"
+    assert settings["api_key"] == "linked-key"
+
+
+def test_profile_discovery_skips_unsafe_ovcli_endpoint(tmp_path):
+    profile_path = tmp_path / "ovcli.conf.blocked"
+    profile_path.write_text(
+        json.dumps({"url": "http://169.254.169.254/latest/meta-data"}),
+        encoding="utf-8",
+    )
+
+    assert (
+        openviking_module._load_profile(
+            profile_path,
+            source="saved",
+            name="blocked",
+        )
+        is None
+    )
+
+
 def test_connection_values_omit_stale_identity_for_user_key_with_root_key():
     values = openviking_module._connection_values_from_ovcli({
         "url": "https://openviking.example",
@@ -178,11 +239,11 @@ def test_connection_values_omit_stale_identity_for_user_key_with_root_key():
 
 def test_link_ovcli_profile_removes_stale_inline_config(tmp_path):
     env_path = tmp_path / ".env"
-    env_path.write_text("OPENVIKING_ENDPOINT=http://old.local\nOTHER_KEY=keep\n", encoding="utf-8")
+    env_path.write_text("OPENVIKING_ENDPOINT=http://old.test\nOTHER_KEY=keep\n", encoding="utf-8")
     config = {"memory": {}}
     provider_config = {
         "use_ovcli_config": False,
-        "endpoint": "http://stale.local",
+        "endpoint": "http://stale.test",
         "api_key": "stale-key",
         "account": "default",
         "user": "default",
@@ -211,12 +272,12 @@ def test_post_setup_existing_profile_picker_validates_and_links_saved_profile(tm
     hermes_home = tmp_path / "hermes"
     hermes_home.mkdir()
     env_path = hermes_home / ".env"
-    env_path.write_text("OPENVIKING_ENDPOINT=http://old.local\nOTHER_KEY=keep\n", encoding="utf-8")
+    env_path.write_text("OPENVIKING_ENDPOINT=http://old.test\nOTHER_KEY=keep\n", encoding="utf-8")
     openviking_home = tmp_path / ".openviking"
     openviking_home.mkdir()
     active_path = openviking_home / "ovcli.conf"
     saved_path = openviking_home / "ovcli.conf.VPS"
-    active_path.write_text(json.dumps({"url": "http://active.local"}), encoding="utf-8")
+    active_path.write_text(json.dumps({"url": "http://active.test"}), encoding="utf-8")
     saved_path.write_text(
         json.dumps({"url": "https://vps.example", "api_key": "user-key"}),
         encoding="utf-8",
@@ -262,6 +323,51 @@ def test_post_setup_existing_profile_picker_validates_and_links_saved_profile(tm
     assert "OTHER_KEY=keep" in env_text
 
 
+def test_local_setup_recommends_user_api_key_before_unauthenticated_mode(monkeypatch):
+    monkeypatch.setattr(
+        openviking_module,
+        "_validate_openviking_reachability",
+        lambda endpoint: (True, ""),
+    )
+    monkeypatch.setattr(
+        openviking_module,
+        "_validate_openviking_setup_values",
+        lambda values, *, require_api_key=False: (True, "", "user"),
+    )
+    credential_menu = {}
+
+    def select(title, options, *, default=0, cancel_returns=None):
+        assert title == "  OpenViking credential"
+        credential_menu["options"] = options
+        credential_menu["default"] = default
+        return 0
+
+    def prompt(label, default=None, secret=False):
+        if label == "OpenViking server URL":
+            return default
+        if label == "OpenViking user API key":
+            assert secret is True
+            return "user-key"
+        if label == openviking_module._AGENT_PROMPT_LABEL:
+            return default
+        raise AssertionError(f"Unexpected prompt: {label}")
+
+    values = openviking_module._prompt_manual_connection_values(
+        prompt,
+        select,
+        -1,
+    )
+
+    assert [label for label, _description in credential_menu["options"]] == [
+        "User API key",
+        "Root API key",
+        "No API key",
+    ]
+    assert credential_menu["default"] == 0
+    assert values["api_key"] == "user-key"
+    assert values["api_key_type"] == "user"
+
+
 def test_start_local_openviking_server_uses_endpoint_host_and_port(monkeypatch):
     popen_calls = []
 
@@ -273,9 +379,9 @@ def test_start_local_openviking_server_uses_endpoint_host_and_port(monkeypatch):
     monkeypatch.setattr(openviking_module.shutil, "which", lambda name: "/usr/local/bin/openviking-server")
     monkeypatch.setattr(openviking_module.subprocess, "Popen", fake_popen)
 
-    started, message = openviking_module._start_local_openviking_server("http://127.0.0.1:1934")
+    state, message = openviking_module._start_local_openviking_server("http://127.0.0.1:1934")
 
-    assert started is True
+    assert state == openviking_module._LOCAL_SERVER_STARTED
     assert "127.0.0.1:1934" in message
     args, kwargs = popen_calls[0]
     assert args == ["/usr/local/bin/openviking-server", "--host", "127.0.0.1", "--port", "1934"]
@@ -291,6 +397,11 @@ def test_start_local_openviking_server_does_not_spawn_when_port_already_open(mon
         return True
 
     monkeypatch.setattr(openviking_module, "_local_openviking_port_is_open", fake_probe)
+    monkeypatch.setattr(
+        openviking_module,
+        "_describe_local_port_listener",
+        lambda host, port: "python-test-server (PID 4242)",
+    )
     monkeypatch.setattr(openviking_module.shutil, "which", lambda name: "/usr/local/bin/openviking-server")
     monkeypatch.setattr(
         openviking_module.subprocess,
@@ -298,16 +409,23 @@ def test_start_local_openviking_server_does_not_spawn_when_port_already_open(mon
         MagicMock(side_effect=AssertionError("must not spawn while a server is already listening")),
     )
 
-    started, message = openviking_module._start_local_openviking_server("http://127.0.0.1:1934")
+    state, message = openviking_module._start_local_openviking_server("http://127.0.0.1:1934")
 
-    assert started is True
-    assert "already running" in message
+    assert state == openviking_module._LOCAL_SERVER_OCCUPIED
+    assert "python-test-server (PID 4242)" in message
+    assert "not passed OpenViking's /health check" in message
+    assert "already running" not in message
     assert probed == [("127.0.0.1", 1934)]
 
 
-def test_start_local_openviking_server_reports_running_server_without_cli_on_path(monkeypatch):
-    """The port probe outranks PATH: a reachable server is started, whoever launched it."""
+def test_start_local_openviking_server_reports_occupied_port_without_cli_on_path(monkeypatch):
+    """The port probe outranks PATH but never claims the listener is OpenViking."""
     monkeypatch.setattr(openviking_module, "_local_openviking_port_is_open", lambda host, port: True)
+    monkeypatch.setattr(
+        openviking_module,
+        "_describe_local_port_listener",
+        lambda host, port: "an unidentified process",
+    )
     monkeypatch.setattr(openviking_module.shutil, "which", lambda name: None)
     monkeypatch.setattr(
         openviking_module.subprocess,
@@ -315,10 +433,10 @@ def test_start_local_openviking_server_reports_running_server_without_cli_on_pat
         MagicMock(side_effect=AssertionError("must not spawn")),
     )
 
-    started, message = openviking_module._start_local_openviking_server("http://127.0.0.1:1934")
+    state, message = openviking_module._start_local_openviking_server("http://127.0.0.1:1934")
 
-    assert started is True
-    assert "already running" in message
+    assert state == openviking_module._LOCAL_SERVER_OCCUPIED
+    assert "unidentified process" in message
 
 
 def test_start_local_openviking_server_rejects_unparseable_url_before_probing(monkeypatch):
@@ -328,9 +446,9 @@ def test_start_local_openviking_server_rejects_unparseable_url_before_probing(mo
         MagicMock(side_effect=AssertionError("must not probe an unparseable endpoint")),
     )
 
-    started, message = openviking_module._start_local_openviking_server("http://127.0.0.1:not-a-port")
+    state, message = openviking_module._start_local_openviking_server("http://127.0.0.1:not-a-port")
 
-    assert started is False
+    assert state == openviking_module._LOCAL_SERVER_FAILED
     assert "Could not parse local OpenViking URL" in message
 
 
@@ -345,6 +463,49 @@ def test_local_openviking_port_is_open_detects_listener_and_closed_port():
     assert openviking_module._local_openviking_port_is_open("127.0.0.1", port) is False
 
 
+def test_describe_local_port_listener_reports_process(monkeypatch):
+    import psutil
+
+    connection = SimpleNamespace(
+        status=psutil.CONN_LISTEN,
+        laddr=SimpleNamespace(ip="0.0.0.0", port=1934),
+        pid=4242,
+    )
+    monkeypatch.setattr(psutil, "net_connections", lambda *, kind: [connection])
+    monkeypatch.setattr(
+        psutil,
+        "Process",
+        lambda pid: SimpleNamespace(name=lambda: "postgres"),
+    )
+
+    assert openviking_module._describe_local_port_listener("127.0.0.1", 1934) == (
+        "postgres (PID 4242)"
+    )
+
+
+def test_runtime_reports_occupied_port_and_does_not_wait_or_spawn(monkeypatch):
+    monkeypatch.setattr(
+        openviking_module,
+        "_start_local_openviking_server",
+        lambda endpoint: (
+            openviking_module._LOCAL_SERVER_OCCUPIED,
+            "Port 127.0.0.1:1934 is occupied by postgres (PID 99).",
+        ),
+    )
+    provider = OpenVikingMemoryProvider()
+    provider._endpoint = "http://127.0.0.1:1934"
+    provider._start_runtime_openviking_waiter = MagicMock()
+    warnings = []
+
+    provider._handle_runtime_openviking_unreachable(warning_callback=warnings.append)
+
+    provider._start_runtime_openviking_waiter.assert_not_called()
+    assert provider._client is None
+    assert len(warnings) == 1
+    assert "postgres (PID 99)" in warnings[0]
+    assert "temporarily unavailable" in warnings[0]
+
+
 def test_https_local_endpoint_is_not_runtime_autostart_eligible(monkeypatch):
     _clear_openviking_env(monkeypatch)
     monkeypatch.setenv("OPENVIKING_ENDPOINT", "https://localhost:1934")
@@ -370,7 +531,7 @@ def test_https_local_endpoint_is_not_runtime_autostart_eligible(monkeypatch):
     assert provider._client is None
     assert warnings == [
         "Remote OpenViking server at https://localhost:1934 is not reachable. "
-        "OpenViking memory disabled; will retry on a later access or when "
+        "OpenViking memory is temporarily unavailable; Hermes will retry on a later access or when "
         "the config changes. "
         "Check the configured endpoint and network connectivity."
     ]
@@ -403,8 +564,9 @@ def test_runtime_does_not_autostart_when_local_server_reports_unhealthy(monkeypa
 
     assert provider._client is None
     assert warnings == [
-        "OpenViking server at http://localhost:1934 responded but reported unhealthy status. "
-        "OpenViking memory disabled; will retry on a later access or when the config changes."
+        "Service at http://localhost:1934 responded but reported unhealthy OpenViking status. "
+        "OpenViking memory is temporarily unavailable; Hermes will retry on a later access "
+        "or when the config changes."
     ]
 
 
@@ -414,7 +576,10 @@ def test_handle_unreachable_endpoint_waits_long_enough_after_autostart(monkeypat
     monkeypatch.setattr(
         openviking_module,
         "_start_local_openviking_server",
-        lambda endpoint: (True, "Started openviking-server on 127.0.0.1:1934 in the background."),
+        lambda endpoint: (
+            openviking_module._LOCAL_SERVER_STARTED,
+            "Started openviking-server on 127.0.0.1:1934 in the background.",
+        ),
     )
     monkeypatch.setattr(
         openviking_module,
@@ -454,7 +619,8 @@ def test_initialize_autostarts_local_openviking_in_background_when_runtime_healt
     monkeypatch.setattr(
         openviking_module,
         "_start_local_openviking_server",
-        lambda endpoint: start_calls.append(endpoint) or (True, "started"),
+        lambda endpoint: start_calls.append(endpoint)
+        or (openviking_module._LOCAL_SERVER_STARTED, "started"),
     )
     monkeypatch.setattr(
         openviking_module,
@@ -1079,6 +1245,7 @@ def test_in_place_compression_rearms_commit_guard():
 
     # The session is still live, so new turns must be committable again.
     assert provider._has_committed_session("sid-123") is False
+    assert provider._turn_count == 0
     assert provider._session_needs_commit("sid-123", 2) is True
 
 
@@ -1111,7 +1278,7 @@ def test_undo_rewind_does_not_rearm_commit_guard():
     assert provider._has_committed_session("sid-123") is True
 
 
-def test_in_place_compression_lifecycle_allows_a_later_commit(monkeypatch):
+def test_in_place_compression_lifecycle_allows_a_later_commit():
     """End-to-end wiring, not a hand-set latch (#74695).
 
     Drives the real sequence a session goes through: commit at the compression
@@ -1121,9 +1288,7 @@ def test_in_place_compression_lifecycle_allows_a_later_commit(monkeypatch):
     """
     provider = _make_provider_with_session("sid-123", turn_count=3)
     provider._ensure_client = lambda: True
-    provider._drain_writers = lambda sid, timeout=None: True
-    # Keep the async write worker out of it; the counter bump is what matters.
-    monkeypatch.setattr(provider, "_queue_memory_write", lambda *a, **k: None, raising=False)
+    provider._new_client = lambda: provider._client
 
     def _commit_calls():
         return [
@@ -1139,9 +1304,18 @@ def test_in_place_compression_lifecycle_allows_a_later_commit(monkeypatch):
     # 2. In-place compression: same id back in, no rotation.
     provider.on_session_switch("sid-123", reason="compression")
 
+    # No new turns means no duplicate extraction at an immediate boundary.
+    provider.on_session_end([])
+    assert len(_commit_calls()) == 1
+
     # 3. A genuinely new turn lands on the still-live session.
     provider.sync_turn("after compression", "reply", session_id="sid-123")
+    assert provider._drain_writers("sid-123", timeout=5.0)
     assert provider._turn_count > 0
+    assert any(
+        call.args and str(call.args[0]).endswith("/messages/batch")
+        for call in provider._client.post.call_args_list
+    )
 
     # 4. That turn must still be committable.
     provider.on_session_end([{"role": "user", "content": "after"}])
@@ -1154,7 +1328,7 @@ def test_resolve_connection_settings_reads_config_yaml_non_secret_fields(monkeyp
     """#68209: non-secret fields saved to config.yaml feed the resolution chain."""
     _clear_openviking_env(monkeypatch)
     provider_config = {
-        "endpoint": "http://saved.local:1933",
+        "endpoint": "http://saved.test:1933",
         "account": "cfg-account",
         "user": "cfg-user",
         "agent": "cfg-agent",
@@ -1162,7 +1336,7 @@ def test_resolve_connection_settings_reads_config_yaml_non_secret_fields(monkeyp
 
     settings = openviking_module._resolve_connection_settings(provider_config)
 
-    assert settings["endpoint"] == "http://saved.local:1933"
+    assert settings["endpoint"] == "http://saved.test:1933"
     assert settings["account"] == "cfg-account"
     assert settings["user"] == "cfg-user"
     assert settings["agent"] == "cfg-agent"
@@ -1171,24 +1345,105 @@ def test_resolve_connection_settings_reads_config_yaml_non_secret_fields(monkeyp
 def test_env_overrides_config_yaml_non_secret_fields(monkeypatch):
     """env still wins over config.yaml (env -> ovcli -> config.yaml -> default)."""
     _clear_openviking_env(monkeypatch)
-    monkeypatch.setenv("OPENVIKING_ENDPOINT", "http://env.local")
+    monkeypatch.setenv("OPENVIKING_ENDPOINT", "http://env.test")
     monkeypatch.setenv("OPENVIKING_AGENT", "env-agent")
 
     settings = openviking_module._resolve_connection_settings(
-        {"endpoint": "http://saved.local", "agent": "cfg-agent"}
+        {"endpoint": "http://saved.test", "agent": "cfg-agent"}
     )
 
-    assert settings["endpoint"] == "http://env.local"
+    assert settings["endpoint"] == "http://env.test"
     assert settings["agent"] == "env-agent"
 
 
+def test_blocked_endpoint_does_not_fall_back_or_construct_client(monkeypatch, tmp_path):
+    _clear_openviking_env(monkeypatch)
+    monkeypatch.setenv(
+        "OPENVIKING_ENDPOINT",
+        "http://169.254.169.254/latest/meta-data/temporary-credential",
+    )
+    monkeypatch.setattr(
+        openviking_module,
+        "_VikingClient",
+        MagicMock(side_effect=AssertionError("blocked endpoint must not construct a client")),
+    )
+    warnings = []
+    provider = OpenVikingMemoryProvider()
+
+    provider.initialize(
+        "session-1",
+        hermes_home=str(tmp_path),
+        platform="cli",
+        warning_callback=warnings.append,
+    )
+
+    assert provider._client is None
+    assert provider._endpoint == ""
+    assert len(warnings) == 1
+    assert "blocked metadata address" in warnings[0]
+    assert "temporary-credential" not in warnings[0]
+    assert openviking_module._DEFAULT_ENDPOINT not in warnings[0]
+
+
+@pytest.mark.parametrize(
+    "health_payload",
+    [
+        {"status": "ok", "healthy": True},
+        ["not", "openviking"],
+    ],
+)
+def test_runtime_rejects_unrelated_json_health_response(
+    monkeypatch, tmp_path, health_payload
+):
+    _clear_openviking_env(monkeypatch)
+    monkeypatch.setenv("OPENVIKING_ENDPOINT", "http://localhost:1934")
+
+    class UnrelatedJsonService:
+        def __init__(self, *args, **kwargs):
+            pass
+
+        def health_payload(self):
+            return health_payload
+
+    monkeypatch.setattr(openviking_module, "_VikingClient", UnrelatedJsonService)
+    monkeypatch.setattr(
+        openviking_module,
+        "_local_openviking_port_is_open",
+        lambda host, port: True,
+    )
+    monkeypatch.setattr(
+        openviking_module,
+        "_describe_local_port_listener",
+        lambda host, port: "python-http-server (PID 4242)",
+    )
+    monkeypatch.setattr(
+        openviking_module,
+        "_start_local_openviking_server",
+        MagicMock(side_effect=AssertionError("responding non-OpenViking service must not auto-start")),
+    )
+    warnings = []
+    provider = OpenVikingMemoryProvider()
+
+    provider.initialize(
+        "session-1",
+        hermes_home=str(tmp_path),
+        platform="cli",
+        warning_callback=warnings.append,
+    )
+
+    assert provider._client is None
+    assert len(warnings) == 1
+    assert "/health response is not valid OpenViking" in warnings[0]
+    assert "python-http-server (PID 4242)" in warnings[0]
+
+
 def test_is_available_true_for_config_yaml_endpoint(monkeypatch):
     """#68209: a config.yaml endpoint (no env, no ovcli) counts as available."""
     _clear_openviking_env(monkeypatch)
     monkeypatch.setattr(
         openviking_module,
         "_load_hermes_openviking_config",
-        lambda: {"endpoint": "http://saved.local:1933"},
+        lambda: {"endpoint": "http://saved.test:1933"},
     )
     assert OpenVikingMemoryProvider().is_available() is True
 
diff --git a/tests/plugins/memory/test_retaindb_provider.py b/tests/plugins/memory/test_retaindb_provider.py
index 922e7b47a1..9fede04e8c 100644
--- a/tests/plugins/memory/test_retaindb_provider.py
+++ b/tests/plugins/memory/test_retaindb_provider.py
@@ -61,6 +61,53 @@ def _capture_initialized_client(monkeypatch, tmp_path):
     return retaindb_module, captured
 
 
+def test_retaindb_config_loader_uses_readonly_config(monkeypatch):
+    import hermes_cli.config as config_mod
+    import plugins.memory.retaindb as retaindb_module
+
+    backing_config = {
+        "memory": {
+            "retaindb": {
+                "base_url": "https://saved.example",
+                "project": "saved-project",
+            }
+        }
+    }
+    monkeypatch.setattr(config_mod, "load_config_readonly", lambda: backing_config)
+    monkeypatch.setattr(
+        config_mod,
+        "load_config",
+        MagicMock(side_effect=AssertionError("read-only provider path must not load a mutable copy")),
+    )
+
+    config = retaindb_module._load_retaindb_config()
+
+    assert config == backing_config["memory"]["retaindb"]
+    assert config is not backing_config["memory"]["retaindb"]
+
+
+def test_initialize_reads_real_dashboard_config_file(tmp_path, monkeypatch):
+    for var in ("RETAINDB_API_KEY", "RETAINDB_BASE_URL", "RETAINDB_PROJECT"):
+        monkeypatch.delenv(var, raising=False)
+    (tmp_path / "config.yaml").write_text(
+        """\
+memory:
+  provider: retaindb
+  retaindb:
+    base_url: https://retaindb.saved.example/
+    project: dashboard-project
+""",
+        encoding="utf-8",
+    )
+    monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+    _retaindb_module, captured = _capture_initialized_client(monkeypatch, tmp_path)
+
+    RetainDBMemoryProvider().initialize("sess-1")
+
+    assert captured["base_url"] == "https://retaindb.saved.example"
+    assert captured["project"] == "dashboard-project"
+
+
 def test_initialize_reads_base_url_and_project_from_config_yaml(tmp_path, monkeypatch):
     """#68209: non-secret base_url/project come from config.yaml when env is unset."""
     for var in ("RETAINDB_API_KEY", "RETAINDB_BASE_URL", "RETAINDB_PROJECT"):
diff --git a/web/src/lib/api.ts b/web/src/lib/api.ts
index 3cc2c5a0ad..54e5badbeb 100644
--- a/web/src/lib/api.ts
+++ b/web/src/lib/api.ts
@@ -1725,14 +1725,17 @@ export interface MemoryProviderFieldOption {
 export interface MemoryProviderField {
   key: string;
   label: string;
-  kind: "text" | "secret" | "select" | "boolean";
+  kind: "text" | "secret" | "select" | "boolean" | "integer" | "number";
   description: string;
   placeholder: string;
   required: boolean;
-  value: string | boolean;
+  value: string | boolean | number;
   is_set: boolean;
   options: MemoryProviderFieldOption[];
   url: string;
+  minimum?: number | null;
+  maximum?: number | null;
+  step?: number | null;
   when?: Record<string, string | boolean | number> | null;
 }
 
diff --git a/web/src/pages/PluginsPage.tsx b/web/src/pages/PluginsPage.tsx
index afd444c6fa..bde492f20c 100644
--- a/web/src/pages/PluginsPage.tsx
+++ b/web/src/pages/PluginsPage.tsx
@@ -32,7 +32,7 @@ import { usePageHeader } from "@/contexts/usePageHeader";
 /** Select value for built-in memory (`config` uses empty string). Never use `""` — UI Select maps empty value to an empty label. */
 const MEMORY_PROVIDER_BUILTIN = "__hermes_memory_builtin__";
 
-type MemoryFormValue = string | boolean;
+type MemoryFormValue = string | boolean | number;
 
 const MEMORY_STATUS_LABEL: Record<MemoryProviderInfo["status"], string> = {
   ready: "ready",
@@ -664,7 +664,18 @@ export default function PluginsPage() {
                               <div className="flex items-center gap-2">
                                 <Input
                                   id={`memory-${field.key}`}
-                                  type={field.kind === "secret" && !secretIsVisible ? "password" : "text"}
+                                  type={
+                                    field.kind === "secret" && !secretIsVisible
+                                      ? "password"
+                                      : field.kind === "integer" || field.kind === "number"
+                                        ? "number"
+                                        : "text"
+                                  }
+                                  min={field.minimum ?? undefined}
+                                  max={field.maximum ?? undefined}
+                                  step={
+                                    field.step ?? (field.kind === "integer" ? 1 : undefined)
+                                  }
                                   value={String(value ?? "")}
                                   placeholder={
                                     field.kind === "secret" && field.is_set