feat(vault): sign in with 1Password or Bitwarden logins, unlocked per session

The browser vault now draws from three login sources behind one handle
shape: the local encrypted vault (vault_…), 1Password Login items (op:…)
and Bitwarden Password Manager logins (bw:…). browser_vault_list aggregates
metadata across them; browser_vault_fill routes by prefix and resolves the
password at fill time only, through the manager CLI.

External managers are locked until the user unlocks them for the current
session. The new browser_vault_unlock tool (and the fill path, implicitly)
asks the surface to show a masked master-password prompt — CLI panel
(reuses the sudo panel state), TUI/Desktop via a vault.unlock.request
blocking card. The password goes to `op signin --raw` / `bw unlock --raw`
on stdin, never argv or env; only the session token is kept, in memory,
with a 30-minute idle TTL, cleared on session close or `vault.lock`.

Headless contexts (cron, webhook, api_server, -q) can never prompt: the
manager is reported as locked with unlock=unavailable_in_this_session and
fill refuses — the same posture approvals take where nobody can answer.

Config: vault.onepassword / vault.bitwarden {enabled, binary_path, …};
a 1Password service-account token skips the prompt for headless use.
RPC: vault.sources, vault.source.set, vault.unlock, vault.lock for Settings.

Tests (2, real subprocess against a fake bw; each proven red by sabotage):
headless never prompts or spawns; unlock feeds stdin only, token never
enters os.environ, fill routes by prefix and the password only reaches the
fill script.

agent/vault_backends/__init__.py

@@ -0,0 +1,18 @@
+"""Login backends for the browser credential vault.
+
+The local Fernet store (``agent/vault_store.py``) is one backend; 1Password
+(``op``) and Bitwarden Password Manager (``bw``) are the others. Every backend
+hands the agent the same shape — opaque handle + login metadata — and resolves
+the password server-side at fill time only. Handles are namespaced by backend
+(``vault_…`` local, ``op:…``, ``bw:…``) so the browser tools need no schema
+change to route to the right one.
+
+External managers are locked until the user unlocks them for the current
+session (``agent/vault_backends/unlock.py``); the master password is typed
+into a masked prompt owned by the surface (CLI panel, Desktop dialog) and is
+never a tool argument, never argv, never persisted.
+"""
+
+from agent.vault_backends.base import LoginBackend, UnlockRequired, backend_for_handle, enabled_backends
+
+__all__ = ["LoginBackend", "UnlockRequired", "backend_for_handle", "enabled_backends"]

agent/vault_backends/base.py

@@ -0,0 +1,86 @@
+"""Login-backend contract + registry for the browser credential vault.
+
+A ``LoginBackend`` lists login metadata (never secrets) and resolves ONE
+password at fill time. External managers (1Password, Bitwarden) additionally
+need a per-session unlock; ``resolve_password`` raises ``UnlockRequired``
+while locked so the tool can ask the surface to prompt. Handles are
+namespaced by ``prefix`` so ``backend_for_handle`` needs no lookup table.
+"""
+
+from __future__ import annotations
+
+import subprocess
+from abc import ABC, abstractmethod
+from typing import Dict, List, Optional, Sequence
+
+from agent.vault_store import VaultItemMeta
+
+
+class UnlockRequired(Exception):
+    """The backend is locked for this session; the surface must prompt for the master password."""
+
+    def __init__(self, backend: "LoginBackend"):
+        super().__init__(f"{backend.display_name} is locked")
+        self.backend = backend
+
+
+class LoginBackend(ABC):
+    name: str                # config key: local | onepassword | bitwarden
+    display_name: str        # user-facing
+    prefix: str              # handle prefix ("vault_", "op:", "bw:")
+    needs_unlock: bool = False
+
+    def owns(self, handle: str) -> bool:
+        return handle.startswith(self.prefix)
+
+    def is_unlocked(self) -> bool:
+        return True
+
+    @abstractmethod
+    def list_items(self) -> List[VaultItemMeta]:
+        """Metadata only. Locked external backends return [] (the agent sees a lock hint instead)."""
+
+    @abstractmethod
+    def get_meta(self, handle: str) -> Optional[VaultItemMeta]: ...
+
+    @abstractmethod
+    def resolve_password(self, handle: str) -> str:
+        """Server-side only; raises ``UnlockRequired`` when locked."""
+
+
+def run_with_stdin_secret(argv: Sequence[str], *, env: Dict[str, str], secret: str, timeout: float,
+                          label: str) -> subprocess.CompletedProcess:
+    """Run a manager CLI feeding *secret* on stdin (never argv, never env). Spawn/timeout → RuntimeError."""
+    try:
+        return subprocess.run(  # noqa: S603 — argv list, no shell
+            list(argv), env=env, input=secret + "\n", capture_output=True, text=True,
+            encoding="utf-8", errors="replace", timeout=timeout)
+    except subprocess.TimeoutExpired as exc:
+        raise RuntimeError(f"{label} unlock timed out after {timeout:.0f}s") from exc
+    except OSError as exc:
+        raise RuntimeError(f"failed to invoke {label}: {exc}") from exc
+
+
+def _cfg() -> Dict:
+    from hermes_cli.config import load_config_readonly
+    cfg = load_config_readonly().get("vault") or {}
+    return cfg if isinstance(cfg, dict) else {}
+
+
+def enabled_backends() -> List[LoginBackend]:
+    """Local first (always on), then each enabled external manager, in config order."""
+    from agent.vault_backends.bitwarden import BitwardenLoginBackend
+    from agent.vault_backends.local import LocalLoginBackend
+    from agent.vault_backends.onepassword import OnePasswordLoginBackend
+
+    cfg = _cfg()
+    out: List[LoginBackend] = [LocalLoginBackend()]
+    for cls in (OnePasswordLoginBackend, BitwardenLoginBackend):
+        section = cfg.get(cls.name) or {}
+        if isinstance(section, dict) and section.get("enabled"):
+            out.append(cls(section))
+    return out
+
+
+def backend_for_handle(handle: str) -> Optional[LoginBackend]:
+    return next((b for b in enabled_backends() if b.owns(handle)), None)

agent/vault_backends/bitwarden.py

@@ -0,0 +1,112 @@
+"""Bitwarden Password Manager logins as a vault backend (``bw`` CLI).
+
+This is the personal/org *password* vault (``bw``), distinct from the
+Bitwarden Secrets Manager (``bws``) source that hydrates API keys at startup.
+Unlock: ``bw unlock --raw`` with the master password on stdin mints a
+``BW_SESSION`` token. List: ``bw list items`` filtered to type=1 (login) with
+a URI. Resolve: ``bw get password <id>``.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import os
+import shutil
+import subprocess
+from pathlib import Path
+from typing import Dict, List, Optional
+
+from agent.secret_sources.base import run_cli, scrub_ansi
+from agent.vault_backends import unlock as _unlock
+from agent.vault_backends.base import LoginBackend, UnlockRequired, run_with_stdin_secret
+from agent.vault_store import VaultItemMeta, normalize_origin
+
+logger = logging.getLogger(__name__)
+
+_TIMEOUT = 30.0
+_ENV_KEEP = ("PATH", "HOME", "USERPROFILE", "APPDATA", "LOCALAPPDATA", "SystemRoot",
+             "TMPDIR", "TMP", "TEMP", "XDG_CONFIG_HOME", "BITWARDENCLI_APPDATA_DIR")
+
+
+class BitwardenLoginBackend(LoginBackend):
+    name = "bitwarden"
+    display_name = "Bitwarden"
+    prefix = "bw:"
+    needs_unlock = True
+
+    def __init__(self, cfg: Optional[Dict] = None):
+        self.cfg = cfg or {}
+
+    def _bw(self) -> Path:
+        explicit = str(self.cfg.get("binary_path") or "")
+        found = explicit or shutil.which("bw")
+        if not found:
+            raise RuntimeError("Bitwarden CLI (bw) not found — install it or set vault.bitwarden.binary_path")
+        return Path(found)
+
+    def _env(self, session_token: Optional[str]) -> Dict[str, str]:
+        env = {k: os.environ[k] for k in _ENV_KEEP if k in os.environ}
+        env["NO_COLOR"] = "1"
+        if session_token:
+            env["BW_SESSION"] = session_token
+        return env
+
+    def is_unlocked(self) -> bool:
+        return _unlock.is_unlocked(self.name)
+
+    def unlock(self, master_password: str) -> None:
+        proc = run_with_stdin_secret([str(self._bw()), "unlock", "--raw", "--nointeraction"],
+                                     env=self._env(None), secret=master_password, timeout=_TIMEOUT, label="bw")
+        token = (proc.stdout or "").strip()
+        if proc.returncode != 0 or not token:
+            err = scrub_ansi(proc.stderr or "").strip()[:200]
+            if "not logged in" in err.lower():
+                err = "not logged in — run `bw login` once in a terminal first"
+            raise RuntimeError(f"Bitwarden unlock failed: {err or 'no session key'}")
+        _unlock.store_session_token(self.name, token)
+
+    def _run(self, *args: str) -> str:
+        token = _unlock.get_session_token(self.name)
+        if not token:
+            raise UnlockRequired(self)
+        proc = run_cli([str(self._bw()), *args, "--nointeraction"], env=self._env(token), timeout=_TIMEOUT,
+                       label="bw", timeout_message="bw timed out", stdin=subprocess.DEVNULL)
+        if proc.returncode != 0:
+            err = scrub_ansi(proc.stderr or "")
+            if "locked" in err.lower() or "session" in err.lower():
+                _unlock.lock(self.name)
+                raise UnlockRequired(self)
+            raise RuntimeError(f"bw failed: {err[:200]}")
+        return proc.stdout or ""
+
+    def list_items(self) -> List[VaultItemMeta]:
+        if not self.is_unlocked():
+            return []
+        raw = json.loads(self._run("list", "items") or "[]")
+        out: List[VaultItemMeta] = []
+        for item in raw if isinstance(raw, list) else []:
+            if item.get("type") != 1 or not isinstance(item.get("login"), dict):
+                continue
+            login = item["login"]
+            origin = None
+            for uri in login.get("uris") or []:
+                try:
+                    origin = normalize_origin(str(uri.get("uri") or ""))
+                    break
+                except Exception:
+                    continue
+            if not origin:
+                continue
+            username = str(login.get("username") or "").strip() or None
+            out.append(VaultItemMeta(
+                id=f"{self.prefix}{item.get('id')}", kind="login", label=str(item.get("name") or origin),
+                origin=origin, created_at=str(item.get("creationDate") or ""),
+                identifier_type="username" if username else None, identifier=username))
+        return out
+
+    def get_meta(self, handle: str) -> Optional[VaultItemMeta]:
+        return next((m for m in self.list_items() if m.id == handle), None)
+
+    def resolve_password(self, handle: str) -> str:
+        return self._run("get", "password", handle[len(self.prefix):]).rstrip("\r\n")

agent/vault_backends/local.py

@@ -0,0 +1,30 @@
+"""Local Fernet vault as a login backend (the always-on default)."""
+
+from __future__ import annotations
+
+from typing import List, Optional
+
+from agent.vault_backends.base import LoginBackend
+from agent.vault_store import VaultItemMeta
+
+
+def _store():
+    # Late import: tests and callers patch ``agent.vault_store.get_vault_store``; binding it here
+    # at call time keeps that facade name the single seam.
+    from agent.vault_store import get_vault_store
+    return get_vault_store()
+
+
+class LocalLoginBackend(LoginBackend):
+    name = "local"
+    display_name = "Hermes vault"
+    prefix = "vault_"
+
+    def list_items(self) -> List[VaultItemMeta]:
+        return _store().list_items()
+
+    def get_meta(self, handle: str) -> Optional[VaultItemMeta]:
+        return _store().get_meta(handle)
+
+    def resolve_password(self, handle: str) -> str:
+        return str(_store().resolve_secret(handle).get("password") or "")

agent/vault_backends/onepassword.py

@@ -0,0 +1,123 @@
+"""1Password Login items as a vault backend (``op`` CLI).
+
+Unlock: ``op signin --raw`` with the master password on stdin (desktop-app
+integration or account-level auth) mints an ``OP_SESSION_<account>`` token.
+A configured service-account token skips the prompt entirely (headless).
+List: ``op item list --categories Login --format json`` → title, urls,
+username. Resolve: ``op item get <id> --fields label=password --reveal``.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import os
+import subprocess
+from pathlib import Path
+from typing import Dict, List, Optional
+
+from agent.secret_sources.base import run_cli
+from agent.secret_sources.onepassword import _OP_ENV_ALLOWLIST, _scrub, find_op
+from agent.vault_backends.base import LoginBackend, UnlockRequired, run_with_stdin_secret
+from agent.vault_backends import unlock as _unlock
+from agent.vault_store import VaultItemMeta, normalize_origin
+
+logger = logging.getLogger(__name__)
+
+_TIMEOUT = 30.0
+
+
+class OnePasswordLoginBackend(LoginBackend):
+    name = "onepassword"
+    display_name = "1Password"
+    prefix = "op:"
+    needs_unlock = True
+
+    def __init__(self, cfg: Optional[Dict] = None):
+        self.cfg = cfg or {}
+        env_name = str(self.cfg.get("service_account_token_env") or "OP_SERVICE_ACCOUNT_TOKEN")
+        self._service_token = os.environ.get(env_name, "") or ""
+
+    # ── auth ────────────────────────────────────────────────────────────────
+
+    def _op(self) -> Path:
+        op = find_op(str(self.cfg.get("binary_path") or ""))
+        if op is None:
+            raise RuntimeError("1Password CLI (op) not found — install it or set vault.onepassword.binary_path")
+        return op
+
+    def _env(self, session_token: Optional[str]) -> Dict[str, str]:
+        env = {k: os.environ[k] for k in _OP_ENV_ALLOWLIST if k in os.environ}
+        env["NO_COLOR"] = "1"
+        account = str(self.cfg.get("account") or "")
+        if account:
+            env["OP_ACCOUNT"] = account
+        if self._service_token:
+            env["OP_SERVICE_ACCOUNT_TOKEN"] = self._service_token
+        elif session_token:
+            # op signin --raw prints the bare token; the env var name carries the account shorthand,
+            # which op also accepts as plain OP_SESSION for the default account.
+            env[f"OP_SESSION_{account}" if account else "OP_SESSION"] = session_token
+        return env
+
+    def is_unlocked(self) -> bool:
+        return bool(self._service_token) or _unlock.is_unlocked(self.name)
+
+    def unlock(self, master_password: str) -> None:
+        """Mint a session token from the master password (consumed on stdin, never argv)."""
+        cmd = [str(self._op()), "signin", "--raw"]
+        if account := str(self.cfg.get("account") or ""):
+            cmd += ["--account", account]
+        proc = run_with_stdin_secret(cmd, env=self._env(None), secret=master_password, timeout=_TIMEOUT, label="op")
+        token = (proc.stdout or "").strip()
+        if proc.returncode != 0 or not token:
+            raise RuntimeError(f"1Password unlock failed: {_scrub(proc.stderr or '')[:200] or 'no session token'}")
+        _unlock.store_session_token(self.name, token)
+
+    def _run(self, *args: str) -> str:
+        token = None if self._service_token else _unlock.get_session_token(self.name)
+        if not self._service_token and not token:
+            raise UnlockRequired(self)
+        proc = run_cli([str(self._op()), *args], env=self._env(token), timeout=_TIMEOUT, label="op",
+                       timeout_message="op timed out", stdin=subprocess.DEVNULL)
+        if proc.returncode != 0:
+            err = _scrub(proc.stderr or "")
+            if "session" in err.lower() or "sign in" in err.lower() or "not signed in" in err.lower():
+                _unlock.lock(self.name)
+                raise UnlockRequired(self)
+            raise RuntimeError(f"op failed: {err[:200]}")
+        return proc.stdout or ""
+
+    # ── backend contract ───────────────────────────────────────────────────
+    def list_items(self) -> List[VaultItemMeta]:
+        if not self.is_unlocked():
+            return []
+        raw = json.loads(self._run("item", "list", "--categories", "Login", "--format", "json") or "[]")
+        out: List[VaultItemMeta] = []
+        for item in raw if isinstance(raw, list) else []:
+            urls = [str(u["href"]) for u in item.get("urls") or [] if isinstance(u, dict) and u.get("href")]
+            origin = _first_origin(urls)
+            if not origin:
+                continue
+            username = str(item.get("additional_information") or "").strip() or None
+            out.append(VaultItemMeta(
+                id=f"{self.prefix}{item.get('id')}", kind="login", label=str(item.get("title") or origin),
+                origin=origin, created_at=str(item.get("created_at") or ""),
+                identifier_type="username" if username else None, identifier=username))
+        return out
+
+    def get_meta(self, handle: str) -> Optional[VaultItemMeta]:
+        return next((m for m in self.list_items() if m.id == handle), None)
+
+    def resolve_password(self, handle: str) -> str:
+        item_id = handle[len(self.prefix):]
+        return self._run("item", "get", item_id, "--fields", "label=password", "--reveal").rstrip("\r\n")
+
+
+def _first_origin(urls: List[str]) -> Optional[str]:
+    for u in urls:
+        try:
+            return normalize_origin(u)
+        except Exception:
+            continue
+    return None

agent/vault_backends/unlock.py

@@ -0,0 +1,80 @@
+"""Per-process unlock state for external password managers.
+
+An unlock is a session token minted by the manager's CLI from the master
+password (``op signin --raw`` / ``bw unlock --raw``). The token lives in
+process memory only, keyed by backend, and expires after an idle TTL or an
+explicit lock. The master password itself is consumed by the CLI call and
+dropped; nothing is written to disk or env.
+
+The surface owns the prompt: ``set_unlock_prompt_callback`` is installed by
+the CLI panel / TUI gateway bridge for the current thread, exactly like the
+sudo-password callback. Headless contexts (cron, webhook, api_server,
+single-query) install none and the vault stays locked — the same posture
+approvals take where nobody can answer.
+"""
+
+from __future__ import annotations
+
+import threading
+import time
+from typing import Callable, Dict, Optional
+
+_IDLE_TTL_S = 30 * 60
+
+_lock = threading.Lock()
+_sessions: Dict[str, tuple[str, float]] = {}   # backend name → (token, last_used)
+_callback_tls = threading.local()
+
+UnlockPrompt = Callable[[str, str], str]  # (backend_name, display_name) -> master password ("" = cancelled)
+
+
+def set_unlock_prompt_callback(cb: Optional[UnlockPrompt]) -> None:
+    """Register the current surface's masked master-password prompt (per-thread slot)."""
+    _callback_tls.prompt = cb
+
+
+def get_unlock_prompt_callback() -> Optional[UnlockPrompt]:
+    return getattr(_callback_tls, "prompt", None)
+
+
+def get_session_token(backend: str) -> Optional[str]:
+    with _lock:
+        entry = _sessions.get(backend)
+        if entry is None:
+            return None
+        token, last = entry
+        if time.monotonic() - last > _IDLE_TTL_S:
+            del _sessions[backend]
+            return None
+        _sessions[backend] = (token, time.monotonic())
+        return token
+
+
+def store_session_token(backend: str, token: str) -> None:
+    with _lock:
+        _sessions[backend] = (token, time.monotonic())
+
+
+def lock(backend: Optional[str] = None) -> None:
+    """Forget one backend's session (or every one when *backend* is None)."""
+    with _lock:
+        if backend is None:
+            _sessions.clear()
+        else:
+            _sessions.pop(backend, None)
+
+
+def is_unlocked(backend: str) -> bool:
+    return get_session_token(backend) is not None
+
+
+def can_prompt_here() -> bool:
+    """False in contexts where no human can answer (cron, webhook, api_server, -q)."""
+    from tools.approval_context import (
+        _is_cron_approval_context,
+        _is_single_query_approval_context,
+        _is_unattended_platform_approval_context,
+    )
+    if _is_cron_approval_context() or _is_unattended_platform_approval_context() or _is_single_query_approval_context():
+        return False
+    return get_unlock_prompt_callback() is not None

cli.py

@@ -3000,6 +3000,8 @@ class HermesCLI(CLIProcessNotificationsMixin, CLIAgentSetupMixin, CLICommandsMix
         set_sudo_password_callback(self._sudo_password_callback)
         set_approval_callback(self._approval_callback)
         set_secret_capture_callback(self._secret_capture_callback)
+        from agent.vault_backends.unlock import set_unlock_prompt_callback
+        set_unlock_prompt_callback(self._vault_unlock_callback)
         try:
             from tools.computer_use_tool import set_approval_callback as _set_cu_cb
 
@@ -3940,8 +3942,11 @@ class HermesCLI(CLIProcessNotificationsMixin, CLIAgentSetupMixin, CLICommandsMix
         with suppress(Exception):
             from tools.voice_mode import cleanup_temp_recordings
             cleanup_temp_recordings()
-        for _unset in (set_sudo_password_callback, set_approval_callback, set_secret_capture_callback):
+        from agent.vault_backends.unlock import lock as _vault_lock, set_unlock_prompt_callback
+        for _unset in (set_sudo_password_callback, set_approval_callback, set_secret_capture_callback,
+                       set_unlock_prompt_callback):
             _unset(None)
+        _vault_lock()  # session tokens for external password managers die with the session
         # On SIGHUP/SIGTERM the agent thread may be reaped before its own persistence runs.
         self._persist_active_session_before_close()
 

hermes_cli/cli_chat_turn_mixin.py

@@ -278,10 +278,12 @@ class CLIChatTurnMixin:
             _prepend_note_to_message, set_approval_callback, set_secret_capture_callback,
             set_sudo_password_callback,
         )
+        from agent.vault_backends.unlock import set_unlock_prompt_callback
         # terminal_tool callbacks are thread-local: run()'s registration is invisible here.
         set_sudo_password_callback(self._sudo_password_callback)
         set_approval_callback(self._approval_callback)
         set_secret_capture_callback(self._secret_capture_callback)
+        set_unlock_prompt_callback(self._vault_unlock_callback)
         # Bind the approval session key so ``is_current_session_yolo_enabled()`` resolves
         # against the same key ``/yolo`` toggles under (``enable_session_yolo(self.session_id)``).
         try:
@@ -341,6 +343,7 @@ class CLIChatTurnMixin:
                 set_sudo_password_callback(None)
                 set_approval_callback(None)
                 set_secret_capture_callback(None)
+                set_unlock_prompt_callback(None)
             except Exception:
                 pass
             # Unbind the per-turn key; ``_session_yolo`` state itself persists across turns.

hermes_cli/cli_commands_mixin.py

@@ -1921,8 +1921,10 @@ class CLICommandsMixin:
         runtime = turn_route["runtime"]
 
         def produce():
+            from agent.vault_backends.unlock import set_unlock_prompt_callback
             set_sudo_password_callback(self._sudo_password_callback)
             set_approval_callback(self._approval_callback)
+            set_unlock_prompt_callback(self._vault_unlock_callback)
             with suppress(Exception):
                 set_secret_capture_callback(self._secret_capture_callback)
             try:
@@ -1960,6 +1962,7 @@ class CLICommandsMixin:
                     set_sudo_password_callback(None)
                     set_approval_callback(None)
                     set_secret_capture_callback(None)
+                    set_unlock_prompt_callback(None)
 
         def done():
             self._background_tasks.pop(task_id, None)

hermes_cli/cli_modal_mixin.py

@@ -852,6 +852,29 @@ class CLIModalMixin:
         self._approval_state = None
         self._invalidate()
 
+    def _vault_unlock_callback(self, backend_name: str, display_name: str) -> str:
+        """Masked master-password prompt for an external password manager (agent thread).
+        Reuses the sudo panel state so rendering, Enter/ESC handling and interrupt cleanup are shared."""
+        from cli import _DIM, _RST, _cprint
+
+        response_queue = queue.Queue()
+        self._capture_modal_input_snapshot()
+        self._sudo_state = {"response_queue": response_queue, "vault_backend": display_name}
+        self._sudo_deadline = _time.monotonic() + 120
+        self._ring_bell(prompt=True, context=f"unlock {display_name}")
+        self._paint_now()
+
+        result = self._poll_modal_queue(response_queue, "_sudo_deadline", refresh=0)
+        self._sudo_state = None
+        self._sudo_deadline = 0
+        self._restore_modal_input_snapshot()
+        self._paint_now()
+        if result is _TIMED_OUT or not result:
+            _cprint(f"\n{_DIM}  ⏭ {display_name} stays locked{_RST}")
+            return ""
+        _cprint(f"\n{_DIM}  ✓ Unlocking {display_name} for this session{_RST}")
+        return result
+
     def _secret_capture_callback(self, var_name: str, prompt: str, metadata=None) -> dict:
         self._capture_modal_input_snapshot()
         try:

hermes_cli/cli_tui_mixin.py

@@ -688,6 +688,12 @@ class CLITuiMixin:
     def _get_sudo_display_fragments(self):
         if not self._sudo_state:
             return []
+        if backend := self._sudo_state.get("vault_backend"):
+            return self._render_sudo_style_panel(
+                f'🔐 Unlock {backend}',
+                [f'The agent wants to sign into a site with a login saved in {backend}.',
+                 'Enter your master password below (hidden) to unlock it for this session, or press Enter to keep it locked.',
+                 'The password never reaches the model; only a session token is kept in memory.'])
         return self._render_sudo_style_panel(
             '🔐 Sudo Password Required', ['Enter password below (hidden), or press Enter to skip'])
 

hermes_cli/config_defaults.py

@@ -2158,6 +2158,23 @@ DEFAULT_CONFIG = {
     },
     # External secret sources — pull credentials from secret managers at startup instead of storing
     # them in ~/.hermes/.env.
+    # Browser credential vault: which login sources browser_vault_list/fill may draw from. The local
+    # encrypted vault (`hermes vault add`, Desktop → Settings → Credential Vault) is always on.
+    # External password managers are unlocked per session with a masked master-password prompt;
+    # headless sessions (cron, webhook, API) never prompt and see them as locked.
+    "vault": {
+        "onepassword": {
+            "enabled": False,       # `op` CLI: Login items with a website URL become fillable handles.
+            "account": "",          # account shorthand for `op --account`; empty = default account.
+            "binary_path": "",      # absolute path to op; empty = PATH.
+            # Env var holding a service-account token (headless auth, no unlock prompt). Unset = prompt.
+            "service_account_token_env": "OP_SERVICE_ACCOUNT_TOKEN",
+        },
+        "bitwarden": {
+            "enabled": False,       # `bw` CLI (Password Manager, not Secrets Manager); run `bw login` once first.
+            "binary_path": "",      # absolute path to bw; empty = PATH.
+        },
+    },
     "secrets": {
         # Optional ordering of enabled sources (e.g. [onepassword, bitwarden]); default registration
         # order. Mapped sources (explicit VAR→ref) always beat bulk sources (BSM project dumps);

tests/agent/test_vault_backends.py

@@ -0,0 +1,133 @@
+"""Invariants for external password-manager vault backends (1Password / Bitwarden).
+
+Two contracts that must never regress:
+1. A locked manager never prompts where nobody can answer (cron/headless) and never leaks a
+   value: browser_vault_list reports it under ``locked``, browser_vault_fill refuses.
+2. The unlock path hands the master password to the manager CLI on stdin only (never argv),
+   keeps just the session token in memory, and a fill then routes by handle prefix through
+   the real subprocess path. Locking forgets the token.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import stat
+from unittest.mock import patch
+
+import pytest
+
+from agent.vault_backends import unlock as unlock_mod
+from agent.vault_backends.bitwarden import BitwardenLoginBackend
+
+# A stand-in `bw` that mimics the three commands the backend uses. It records argv + stdin so the
+# test can prove the master password travelled on stdin only, and a fake session key gates `list`.
+# (No env passthrough: the backend's allowlisted child env is part of what is under test.)
+_FAKE_BW = r'''#!/usr/bin/env python3
+import json, os, sys
+log = open(os.path.join(os.path.dirname(os.path.abspath(__file__)), "bw.log"), "a")
+argv = sys.argv[1:]
+stdin = sys.stdin.read() if not sys.stdin.isatty() else ""
+log.write(json.dumps({"argv": argv, "stdin": stdin, "BW_SESSION": os.environ.get("BW_SESSION")}) + "\n")
+if argv[:2] == ["unlock", "--raw"]:
+    if stdin.strip() != "correct horse":
+        sys.stderr.write("Invalid master password.\n"); sys.exit(1)
+    print("SESSION-TOKEN-123"); sys.exit(0)
+if os.environ.get("BW_SESSION") != "SESSION-TOKEN-123":
+    sys.stderr.write("Vault is locked.\n"); sys.exit(1)
+if argv[:2] == ["list", "items"]:
+    print(json.dumps([{"id": "abc", "type": 1, "name": "Example", "creationDate": "2026-01-01T00:00:00Z",
+                       "login": {"username": "jane@example.com", "uris": [{"uri": "https://example.com/login"}]}},
+                      {"id": "note", "type": 2, "name": "Secure note"}])); sys.exit(0)
+if argv[:2] == ["get", "password"]:
+    print("plain sentence nobody would flag 7"); sys.exit(0)
+sys.exit(2)
+'''
+
+
+pytestmark = pytest.mark.skipif(os.name == "nt", reason="fake bw is a shebang script; the backend under test is host-agnostic")
+
+
+@pytest.fixture
+def fake_bw(tmp_path, monkeypatch):
+    exe = tmp_path / "bw"
+    exe.write_text(_FAKE_BW, encoding="utf-8")
+    exe.chmod(exe.stat().st_mode | stat.S_IXUSR)
+    log = tmp_path / "bw.log"  # the backend runs bw with an allowlisted env, so the fake logs beside itself
+    monkeypatch.setenv("HERMES_HOME", str(tmp_path / ".hermes"))
+    unlock_mod.lock()
+    yield exe, log
+    unlock_mod.lock()
+
+
+def _enabled(exe):
+    """The tool late-imports ``enabled_backends`` from the package facade; ``backend_for_handle`` reads
+    the sibling's own binding — patch both so the fake is the only backend anywhere."""
+    backend = BitwardenLoginBackend({"enabled": True, "binary_path": str(exe)})
+    return patch("agent.vault_backends.base.enabled_backends", return_value=[backend]), backend
+
+
+def test_locked_manager_is_reported_not_prompted_when_headless(fake_bw, monkeypatch):
+    exe, log = fake_bw
+    from tools.browser_vault_tool import browser_vault_fill, browser_vault_list
+
+    patcher, backend = _enabled(exe)
+    monkeypatch.setenv("HERMES_CRON_SESSION", "1")  # headless: nobody can answer a prompt
+    unlock_mod.set_unlock_prompt_callback(lambda *_: "correct horse")  # even a wired prompt must not fire
+    try:
+        with patcher, patch("agent.vault_backends.enabled_backends", return_value=[backend]):
+            listed = json.loads(browser_vault_list())
+            assert listed["items"] == []
+            assert listed["locked"] == [{"backend": "bitwarden", "display_name": "Bitwarden",
+                                         "unlock": "unavailable_in_this_session"}]
+            filled = json.loads(browser_vault_fill("bw:abc", task_id="t"))
+            assert filled["success"] is False and filled["error_type"] == "unlock_unavailable"
+    finally:
+        unlock_mod.set_unlock_prompt_callback(None)
+    assert not log.exists(), "bw must not be invoked at all while locked in a headless session"
+    assert not unlock_mod.is_unlocked("bitwarden")
+
+
+def test_unlock_feeds_stdin_only_then_fill_routes_by_prefix(fake_bw):
+    exe, log = fake_bw
+    from tools.browser_vault_tool import browser_vault_fill, browser_vault_list
+
+    patcher, backend = _enabled(exe)
+    prompts = []
+
+    def prompt(name, display):
+        prompts.append((name, display))
+        return "correct horse"
+
+    unlock_mod.set_unlock_prompt_callback(prompt)
+    try:
+        with patcher, patch("agent.vault_backends.enabled_backends", return_value=[backend]), \
+             patch("tools.browser_vault_tool._current_page_origin", return_value="https://example.com"), \
+             patch("tools.browser_vault_tool._eval_js", return_value={"success": True, "result": json.dumps([
+                 {"tag": "input", "type": "password", "name": "password", "id": "pw", "autocomplete": "current-password",
+                  "visible": True}])}), \
+             patch("tools.browser_vault_tool._eval_js_secret", return_value={"success": True, "result": json.dumps(
+                 {"filled": 1})}) as secret_eval:
+            out = json.loads(browser_vault_fill("bw:abc", task_id="t"))
+            assert out == {"success": True, "filled_fields": 1, "backend": "bitwarden", "kind": "login",
+                           "origin": "https://example.com"}
+            assert prompts == [("bitwarden", "Bitwarden")]
+            # Now unlocked: listing exposes metadata only, never the password.
+            listed = json.loads(browser_vault_list())
+            assert listed["items"][0]["handle"] == "bw:abc"
+            assert listed["items"][0]["identifier"] == "jane@example.com"
+            assert "plain sentence nobody would flag 7" not in json.dumps(listed)
+            # The password reached the fill script, and only there.
+            assert "plain sentence nobody would flag 7" in secret_eval.call_args.args[1]
+    finally:
+        unlock_mod.set_unlock_prompt_callback(None)
+
+    calls = [json.loads(line) for line in log.read_text(encoding="utf-8").splitlines()]
+    unlock_calls = [c for c in calls if c["argv"][:2] == ["unlock", "--raw"]]
+    assert len(unlock_calls) == 1 and unlock_calls[0]["stdin"].strip() == "correct horse"
+    assert all("correct horse" not in " ".join(c["argv"]) for c in calls), "master password must never be argv"
+    assert all(c["BW_SESSION"] == "SESSION-TOKEN-123" for c in calls if c["argv"][0] != "unlock")
+
+    unlock_mod.lock("bitwarden")
+    assert not backend.is_unlocked()
+    assert os.environ.get("BW_SESSION") is None, "session token must never touch the process env"

tests/test_browser_vault.py

@@ -314,6 +314,7 @@ class TestBrowserVaultTools:
         assert out == {
             "success": True,
             "filled_fields": 1,
+            "backend": "local",
             "kind": "login",
             "origin": "https://example.com",
         }

tools/browser_vault_tool.py

@@ -37,11 +37,11 @@ logger = logging.getLogger(__name__)
 # ---------------------------------------------------------------------------
 
 def _check_vault_available() -> bool:
-    """Tools are only in the schema when the local vault has ≥1 item."""
+    """Schema-gate: the tools appear only when the local vault has items or an external manager is enabled."""
     try:
+        from agent.vault_backends import enabled_backends
         from agent.vault_store import get_vault_store
-
-        return get_vault_store().has_items()
+        return get_vault_store().has_items() or any(b.needs_unlock for b in enabled_backends())
     except Exception:
         return False
 
@@ -157,28 +157,67 @@ def _current_page_origin(task_id: str) -> Optional[str]:
 # ---------------------------------------------------------------------------
 
 def browser_vault_list() -> str:
-    """List vault items as handles + metadata. Secret values never included.
+    """List login handles + metadata across every enabled backend. Passwords are never included.
 
-    Login identifiers (email/username/phone) ARE included — they are
-    metadata, not secrets, so the agent can type the identifier itself.
+    A locked external manager contributes no items; instead it is reported under ``locked`` so the
+    agent knows to call browser_vault_fill (which prompts the user to unlock) or tell the user.
     """
-    from agent.vault_store import get_vault_store
+    from agent.vault_backends import enabled_backends
+    from agent.vault_backends.unlock import can_prompt_here
 
-    items = []
-    for meta in get_vault_store().list_items():
-        entry = {
-            "handle": meta.id,
-            "label": meta.label,
-            "kind": meta.kind,
-            "origin": meta.origin,
-            # Phase 1: only login items are fillable.
-            "available": meta.kind == "login",
-        }
-        if meta.identifier:
-            entry["identifier"] = meta.identifier
-            entry["identifier_type"] = meta.identifier_type
-        items.append(entry)
-    return json.dumps({"success": True, "items": items}, ensure_ascii=False)
+    items, locked, errors = [], [], []
+    for backend in enabled_backends():
+        if backend.needs_unlock and not backend.is_unlocked():
+            locked.append({"backend": backend.name, "display_name": backend.display_name,
+                           "unlock": "browser_vault_unlock" if can_prompt_here() else "unavailable_in_this_session"})
+            continue
+        try:
+            metas = backend.list_items()
+        except Exception as exc:
+            errors.append({"backend": backend.name, "error": str(exc)[:200]})
+            continue
+        for meta in metas:
+            entry = {"handle": meta.id, "backend": backend.name, "label": meta.label, "kind": meta.kind,
+                     "origin": meta.origin, "available": meta.kind == "login"}
+            if meta.identifier:
+                entry["identifier"] = meta.identifier
+                entry["identifier_type"] = meta.identifier_type
+            items.append(entry)
+    out: Dict[str, Any] = {"success": True, "items": items}
+    if locked:
+        out["locked"] = locked
+    if errors:
+        out["errors"] = errors
+    return json.dumps(out, ensure_ascii=False)
+
+
+def browser_vault_unlock(backend_name: str) -> str:
+    """Ask the user (via the surface's masked prompt) to unlock an external manager for this session."""
+    from agent.vault_backends import enabled_backends
+    from agent.vault_backends.unlock import can_prompt_here, get_unlock_prompt_callback
+
+    backend = next((b for b in enabled_backends() if b.name == backend_name and b.needs_unlock), None)
+    if backend is None:
+        return json.dumps({"success": False, "error": f"No unlockable vault backend named {backend_name!r}."})
+    if backend.is_unlocked():
+        return json.dumps({"success": True, "backend": backend.name, "already_unlocked": True})
+    if not can_prompt_here():
+        return json.dumps({"success": False, "error_type": "unlock_unavailable",
+                           "error": (f"{backend.display_name} is locked and this session cannot prompt for the "
+                                     "master password (headless/cron/API). Unlock it from an interactive Hermes "
+                                     "session or the Desktop app first.")})
+    prompt = get_unlock_prompt_callback()
+    master = prompt(backend.name, backend.display_name) if prompt else ""
+    if not master:
+        return json.dumps({"success": False, "error_type": "unlock_cancelled",
+                           "error": f"The user declined to unlock {backend.display_name}."})
+    try:
+        backend.unlock(master)  # type: ignore[attr-defined]
+    except Exception as exc:
+        return json.dumps({"success": False, "error_type": "unlock_failed", "error": str(exc)[:300]})
+    finally:
+        del master
+    return json.dumps({"success": True, "backend": backend.name})
 
 
 def browser_vault_fill(handle: str, task_id: Optional[str] = None) -> str:
@@ -198,12 +237,21 @@ def browser_vault_fill(handle: str, task_id: Optional[str] = None) -> str:
         classify_login_control,
         select_password_fill,
     )
-    from agent.vault_store import VaultError, get_vault_store, scrub_secret_from_text
+    from agent.vault_backends import UnlockRequired, backend_for_handle
+    from agent.vault_store import scrub_secret_from_text
 
     effective_task_id = task_id or "default"
-    store = get_vault_store()
+    backend = backend_for_handle(handle)
+    if backend is not None and backend.needs_unlock and not backend.is_unlocked():
+        unlocked = json.loads(browser_vault_unlock(backend.name))
+        if not unlocked.get("success"):
+            return json.dumps(unlocked)
 
-    meta = store.get_meta(handle)
+    try:
+        meta = backend.get_meta(handle) if backend is not None else None
+    except UnlockRequired:
+        return json.dumps({"success": False, "error_type": "unlock_required",
+                           "error": f"{backend.display_name} locked again; call browser_vault_unlock."})
     if meta is None:
         return json.dumps(
             {
@@ -263,8 +311,12 @@ def browser_vault_fill(handle: str, task_id: Optional[str] = None) -> str:
         return json.dumps({"success": False, "error": "No login form fields were found on the current page."})
 
     # ── Resolve secret and fill (secret never enters any logged string) ─────
-    secret = store.resolve_secret(handle)
-    password = str(secret.get("password") or "")
+    try:
+        password = backend.resolve_password(handle)
+    except UnlockRequired:
+        return json.dumps({"success": False, "error_type": "unlock_required",
+                           "error": f"{backend.display_name} locked again; call browser_vault_unlock."})
+    secret = {"password": password}
     fills = select_password_fill(classified, password)
     if not fills:
         return json.dumps(
@@ -314,6 +366,7 @@ def browser_vault_fill(handle: str, task_id: Optional[str] = None) -> str:
         {
             "success": bool(filled),
             "filled_fields": int(filled),
+            "backend": backend.name,
             "kind": meta.kind,
             "origin": meta.origin,
         }
@@ -327,33 +380,48 @@ def browser_vault_fill(handle: str, task_id: Optional[str] = None) -> str:
 BROWSER_VAULT_LIST_SCHEMA = {
     "name": "browser_vault_list",
     "description": (
-        "List credentials stored in the local encrypted vault as handles "
-        "with metadata (label, kind, bound origin, and for logins the "
-        "identifier + identifier_type — identifiers are visible so you can "
-        "type them yourself with fill_input). Passwords are NEVER returned. "
-        "Workflow: type the identifier with fill_input, then call "
-        "browser_vault_fill with the handle to fill the password."
+        "List saved website logins as handles with metadata (label, backend, bound origin, and the "
+        "identifier + identifier_type so you can type the username yourself with fill_input). "
+        "Passwords are NEVER returned. Sources: the local Hermes vault plus any enabled password "
+        "manager (1Password, Bitwarden). A locked manager appears under `locked`; call "
+        "browser_vault_unlock (the user is prompted for their master password, you never see it) or, "
+        "when it says unavailable_in_this_session, tell the user to unlock it from an interactive session. "
+        "Workflow: fill_input the identifier, then browser_vault_fill with the handle."
     ),
     "input_schema": {"type": "object", "properties": {}, "required": []},
 }
 
+BROWSER_VAULT_UNLOCK_SCHEMA = {
+    "name": "browser_vault_unlock",
+    "description": (
+        "Ask the user to unlock a password manager (1Password or Bitwarden) for this session. The master "
+        "password is typed into a masked prompt owned by the UI and never enters the conversation. "
+        "Returns success, unlock_cancelled, unlock_failed, or unlock_unavailable (headless session)."
+    ),
+    "input_schema": {
+        "type": "object",
+        "properties": {"backend": {"type": "string", "enum": ["onepassword", "bitwarden"],
+                                   "description": "Backend name from browser_vault_list `locked`."}},
+        "required": ["backend"],
+    },
+}
+
 BROWSER_VAULT_FILL_SCHEMA = {
     "name": "browser_vault_fill",
     "description": (
-        "Fill ONLY the password field of the CURRENT browser page's login "
-        "form from a vault handle (see browser_vault_list). Type the "
-        "identifier/username yourself first with fill_input (it is visible "
-        "in the vault metadata), then call this to fill the password. The "
-        "password is resolved and injected server-side; it never appears in "
-        "the conversation. Refused unless the page origin exactly matches "
-        "the credential's bound origin (re-checked atomically at fill time)."
+        "Fill ONLY the password field of the CURRENT browser page's login form from a vault handle "
+        "(see browser_vault_list). Type the identifier/username yourself first with fill_input, then "
+        "call this. The password is resolved from the local vault or the password manager and injected "
+        "server-side; it never appears in the conversation. Refused unless the page origin exactly "
+        "matches the credential's bound origin (re-checked atomically at fill time). If the manager is "
+        "locked the user is prompted to unlock first."
     ),
     "input_schema": {
         "type": "object",
         "properties": {
             "handle": {
                 "type": "string",
-                "description": "Vault item handle from browser_vault_list (e.g. vault_ab12cd34ef56)",
+                "description": "Handle from browser_vault_list (vault_… local, op:… 1Password, bw:… Bitwarden)",
             }
         },
         "required": ["handle"],
@@ -365,6 +433,10 @@ def _handle_vault_list(args: Dict[str, Any], **kwargs) -> str:
     return browser_vault_list()
 
 
+def _handle_vault_unlock(args: Dict[str, Any], **kwargs) -> str:
+    return browser_vault_unlock(str(args.get("backend") or ""))
+
+
 def _handle_vault_fill(args: Dict[str, Any], **kwargs) -> str:
     return browser_vault_fill(
         handle=str(args.get("handle") or ""), task_id=kwargs.get("task_id")
@@ -382,6 +454,15 @@ registry.register(
     emoji="🔐",
 )
 
+registry.register(
+    name="browser_vault_unlock",
+    toolset="browser",
+    schema=BROWSER_VAULT_UNLOCK_SCHEMA,
+    handler=_handle_vault_unlock,
+    check_fn=_check_vault_available,
+    emoji="🔐",
+)
+
 registry.register(
     name="browser_vault_fill",
     toolset="browser",

toolsets.py

@@ -18,7 +18,7 @@ _HERMES_CORE_TOOLS = [
     "browser_type", "browser_scroll", "browser_back",
     "browser_press", "browser_get_images",
     "browser_vision", "browser_console", "browser_cdp", "browser_dialog",
-    "browser_vault_list", "browser_vault_fill",  # gated on a non-empty vault via check_fn
+    "browser_vault_list", "browser_vault_unlock", "browser_vault_fill",  # check_fn: vault items or a manager enabled
     "browser_exec",  # replaces the other browser tools when browser.backend is "browser-use"
     "text_to_speech",
     "todo_list", "memory",

tui_gateway/agent_callbacks.py

@@ -169,6 +169,11 @@ def _wire_callbacks(sid: str):
     set_sudo_password_callback(lambda: _block("sudo.request", sid, {}, timeout=120))
     set_project_workspace_callback(_apply_project_workspace)
     set_secret_capture_callback(secret_cb)
+    # External password-manager unlock: the renderer shows a masked master-password card; the
+    # answer is consumed by the manager CLI on stdin and only a session token stays in memory.
+    from agent.vault_backends.unlock import set_unlock_prompt_callback
+    set_unlock_prompt_callback(lambda backend, display_name: _block(
+        "vault.unlock.request", sid, {"backend": backend, "display_name": display_name}, timeout=120))
 
 
 def _available_personalities(cfg: dict | None = None) -> dict:

tui_gateway/methods_prompt.py

@@ -1097,7 +1097,7 @@ def _(rid, params: dict) -> dict:
 _LATE_RESPOND_KEYS = {
     "terminal.read.respond": "text", "preview.read.respond": "text", "preview.act.respond": "text",
     "window.read.respond": "text", "tour.respond": "text", "mcp.setup.respond": "result",
-    "sudo.respond": "password", "secret.respond": "value"}
+    "sudo.respond": "password", "secret.respond": "value", "vault.unlock.respond": "password"}
 for _name, _key in _LATE_RESPOND_KEYS.items():
     method(_name)(lambda rid, params, _k=_key: _respond(rid, params, _k, allow_expired=True))
 del _name, _key

tui_gateway/methods_vault.py

@@ -12,6 +12,9 @@ JSON-RPC channel every other Settings surface uses. Contracts:
   encrypted store, and is never logged. Error strings are defensively
   scrubbed with ``scrub_secret_from_text`` before they leave the handler.
 - ``vault.remove`` → {removed: bool}.
+- ``vault.sources`` / ``vault.source.set`` → external password-manager status and enable toggle.
+- ``vault.unlock`` / ``vault.lock`` → per-session unlock of a manager from Settings; the master
+  password is consumed by the manager CLI on stdin and never stored or logged.
 
 Handlers are rebound onto server.py's globals at install time (see
 method_ctx.py) and may reference server module globals (``_ok``, ``_err``).
@@ -29,16 +32,95 @@ method = _registry.method
 
 @method("vault.list")
 def _(rid, params: dict) -> dict:
-    """Metadata-only listing of vault items. Secret values are never included."""
+    """Metadata-only listing across every enabled backend (local + unlocked password managers).
+    Each item carries ``backend``; locked managers contribute nothing (see vault.sources)."""
     try:
-        from agent.vault_store import get_vault_store
+        from agent.vault_backends import enabled_backends
 
-        items = [meta.to_dict() for meta in get_vault_store().list_items()]
+        items = []
+        for backend in enabled_backends():
+            if backend.needs_unlock and not backend.is_unlocked():
+                continue
+            items.extend({**meta.to_dict(), "backend": backend.name} for meta in backend.list_items())
         return _ok(rid, {"items": items})
     except Exception as e:
         return _err(rid, 5095, str(e))
 
 
+_EXTERNAL_SOURCES = ("onepassword", "bitwarden")
+
+
+@method("vault.sources")
+def _(rid, params: dict) -> dict:
+    """Status of every login source: {name, display_name, enabled, needs_unlock, unlocked, installed}."""
+    import shutil
+
+    from agent.vault_backends import enabled_backends
+    from agent.vault_backends.bitwarden import BitwardenLoginBackend
+    from agent.vault_backends.onepassword import OnePasswordLoginBackend
+    from agent.secret_sources.onepassword import find_op
+
+    enabled = {b.name: b for b in enabled_backends()}
+    rows = [{"name": "local", "display_name": "Hermes vault", "enabled": True, "needs_unlock": False,
+             "unlocked": True, "installed": True}]
+    for cls, installed in ((OnePasswordLoginBackend, find_op() is not None),
+                           (BitwardenLoginBackend, shutil.which("bw") is not None)):
+        live = enabled.get(cls.name)
+        rows.append({"name": cls.name, "display_name": cls.display_name, "enabled": live is not None,
+                     "needs_unlock": True, "unlocked": bool(live and live.is_unlocked()), "installed": installed})
+    return _ok(rid, {"sources": rows})
+
+
+@method("vault.source.set")
+def _(rid, params: dict) -> dict:
+    """Enable/disable an external manager: writes ``vault.<name>.enabled`` and locks it when disabling."""
+    from agent.vault_backends.unlock import lock
+    from hermes_cli.config import load_config, save_config
+
+    name = str(params.get("name") or "")
+    if name not in _EXTERNAL_SOURCES:
+        return _err(rid, 5095, f"unknown vault source: {name}")
+    enabled = bool(params.get("enabled"))
+    cfg = load_config()
+    section = cfg.setdefault("vault", {}).setdefault(name, {})
+    section["enabled"] = enabled
+    if not enabled:
+        lock(name)
+    save_config(cfg)
+    return _ok(rid, {"name": name, "enabled": enabled})
+
+
+@method("vault.unlock")
+def _(rid, params: dict) -> dict:
+    """Unlock a manager with the master password typed in the Settings dialog (consumed by the CLI on stdin)."""
+    from agent.vault_backends import enabled_backends
+
+    name = str(params.get("name") or "")
+    password = str(params.get("password") or "")
+    backend = next((b for b in enabled_backends() if b.name == name and b.needs_unlock), None)
+    if backend is None:
+        return _err(rid, 5095, f"{name} is not an enabled password manager")
+    if not password:
+        return _err(rid, 5095, "master password is required")
+    try:
+        backend.unlock(password)  # type: ignore[attr-defined]
+    except Exception as e:
+        return _err(rid, 5095, str(e).replace(password, "[REDACTED]"))
+    finally:
+        del password
+    return _ok(rid, {"name": name, "unlocked": True})
+
+
+@method("vault.lock")
+def _(rid, params: dict) -> dict:
+    """Forget a manager's session token (or every one when ``name`` is omitted)."""
+    from agent.vault_backends.unlock import lock
+
+    name = params.get("name")
+    lock(str(name) if name else None)
+    return _ok(rid, {"locked": True})
+
+
 @method("vault.add")
 def _(rid, params: dict) -> dict:
     """Add a vault item. ``secret`` values go straight into the encrypted store.

tui_gateway/server.py

@@ -1247,7 +1247,7 @@ def _enable_gateway_prompts() -> None:
 # Blocking bridges whose `*.respond` tolerates a late reply (allow_expired=True): on timeout the tool
 # returns empty, but a slow renderer could still answer and hit a raw 4009 — `.expire` tears the card down.
 _EXPIRING_REQUESTS = frozenset({
-    "secret.request", "sudo.request", "clarify.request", "terminal.read.request",
+    "secret.request", "sudo.request", "vault.unlock.request", "clarify.request", "terminal.read.request",
     "preview.read.request", "preview.act.request", "window.read.request", "mcp.setup.request",
     "tour.request",
 })