# NousResearch/hermes-agent · 1Password and Bitwarden secret source integrations

## Open now

- Nothing open on GitHub.

These issues and pull requests concern adding support for 1Password (via CLI, op:// URIs, and SDK) as a secret source backend, along with a pluggable SecretSource interface that also includes a Bitwarden provider.

## TL;DR

- 5 member(s): 0 open issue(s), 1 closed issue(s), 0 open PR(s), 1 merged PR(s), 3 PR(s) closed without merging.
- A fix was merged to main in PR #59498. Verify it addresses your case.
- **Recommended:** read PR #59498 (and any competing PRs below) before filing. Confirm the merged change covers your symptom.

## Maintainer resolution verdict

- Internal-tracker verdict over this cluster's issue members: `unresolved`. This is distinct from the fix status above. Weigh it yourself.

## How this cluster was grouped

- Link evidence (`STRUCTURAL_SPANNING`): every member is connected by structural evidence: shared changed files, or a closing/superseded reference.
- Pairwise link density `0.40`: the share of member pairs that carry a link, counting a structural link in full and a relevance-only link at 0.36. `1.00` means every pair is linked.

## Closed / duplicate issues

- #36949: \[Feature\]: 1Password (op://) secret source backend

## Prior & competing PRs

- **#59498: feat(secrets): pluggable SecretSource interface + Bitwarden & 1Password providers**. merged to main · fix tier `verified`
  - maintainer direction: category `merged`; implication `superseded`
  - files: `.gitignore`, `agent/credential_pool.py`, `agent/secret_sources/__init__.py`, `agent/secret_sources/_cache.py`, `agent/secret_sources/base.py`, `agent/secret_sources/bitwarden.py`, `agent/secret_sources/onepassword.py`, `agent/secret_sources/registry.py`, `cli-config.yaml.example`, `hermes_cli/config.py`, `hermes_cli/env_loader.py`, `hermes_cli/main.py`, `hermes_cli/onepassword_secrets_cli.py`, `hermes_cli/plugins.py`, `scripts/release.py`, `tests/secret_sources/__init__.py`, `tests/secret_sources/conformance.py`, `tests/secret_sources/test_secret_source_registry.py`, `tests/test_bitwarden_secrets.py`, `tests/test_env_loader_op_bootstrap.py` (+4 more)
  - fix hunks:
  - `.gitignore`

    ```diff
    @@ -8,6 +8,7 @@ __pycache__/
     .venv
     .vscode/
     .env
    +.op.env
     .env.local
     .env.development.local
     .env.test.local
    ```
  - `agent/credential_pool.py`

    ```diff
    @@ -2105,8 +2105,20 @@ def _seed_from_env(provider: str, entries: List[PooledCredential]) -> Tuple[bool
         # changes to the .env file.
         def _get_env_prefer_dotenv(key: str) -> str:
             env_file = load_env()
    -        val = env_file.get(key) or _get_secret(key, "") or ""
    -        return val.strip()
    +        raw = env_file.get(key, "").strip()
    +        env_val = os.environ.get(key, "").strip()
    +        # If .env contains an unresolved op:// reference, prefer the
    +        # already-resolved value from os.environ (set by
    +        # load_hermes_dotenv() -> apply_onepassword_secrets()).  The raw
    +        # "op://Vault/Item/field" string would otherwise win and every
    +        # provider auth attempt would receive a URL instead of a key.  This
    +        # happens during a partial migration, or when the user wrote op://
    +        # references straight into .env rather than the secrets.onepassword
    +        # config block.  For every non-op:// value the original
    +        # .env-takes-precedence behaviour is preserved unchanged.
    +        if raw.startswith("op://") and env_val:
    +            return env_val
    +        return raw or _get_secret(key, "") or env_val
     
         # Honour user suppression — `hermes auth remove <provider> <N>` for an
         # env-seeded credential marks the env:<VAR> source as suppressed so it
    ```
  - `agent/secret_sources/__init__.py`

    ```diff
    @@ -1,13 +1,41 @@
     """External secret source integrations.
     
     A secret source is anything that can supply environment-variable-shaped
    -credentials at process startup, _after_ ~/.hermes/.env has loaded.  By
    -default sources are non-destructive: they only set values for env vars
    -that aren't already present, so .env and shell exports continue to win.
    +credentials at process startup, _after_ ~/.hermes/.env has loaded.
     
    -Currently shipped:
    +The contract every source implements is
    +:class:`agent.secret_sources.base.SecretSource`; the orchestrator that
    +runs the enabled sources (ordering, mapped-beats-bulk precedence,
    +first-claim-wins conflicts, ``override_existing`` semantics, provenance)
    +is :func:`agent.secret_sources.registry.apply_all`.  Multiple sources
    +can be enabled at once — see the registry module docstring for the
    +precedence ladder.  The atomic-write / 0600 / TTL disk-cache substrate
    +is shared across backends in ``agent.secret_sources._cache`` so the
    +security-sensitive bits live in exactly one place.
    +
    +Currently bundled:
     
       - ``bitwarden`` — Bitwarden Secrets Manager (`bws` CLI).  See
         ``agent.secret_sources.bitwarden`` for the integration and
         ``hermes_cli.secrets_cli`` for the user-facing setup wizard.
    +  - ``onepassword`` — 1Password ``op://`` secret references (`op` CLI).
    +    See ``agent.secret_sources.onepassword`` for the integration and
    +    ``hermes_cli.onepassword_secrets_cli`` for the user-facing commands.
    +
    +The bundled set is deliberately closed (policy mirrors memory
    +providers): new third-party secret managers ship as standalone plugin
    +repos that subclass ``SecretSource`` and register through
    +``PluginContext.register_secret_source()`` — they are NOT added to this
    +package.  A generic ``command`` source is a possible future exception;
    +OS keystores (Keychain/DPAPI/libsecret) are under discussion.
     """
    +
    +from agent.secret_sources.base import (  # noqa: F401
    +    SECRET_SOURCE_API_VERSION,
    +    ErrorKind,
    +    FetchResult,
    +    SecretSource,
    +    is_valid_env_name,
    +    run_secret_cli,
    +    scrub_ansi,
    +)
    ```
  - `agent/secret_sources/_cache.py`

    ```diff
    @@ -0,0 +1,213 @@
    +"""Shared substrate for external secret-source backends.
    +
    +Every backend (Bitwarden, 1Password, …) needs the same handful of
    +security-sensitive primitives:
    +
    +  * a uniform result object (:class:`FetchResult`),
    +  * environment-variable name validation (:func:`is_valid_env_name`),
    +  * a two-layer fetch cache whose disk half writes atomically with ``0600``
    +    permissions and honours a TTL (:class:`DiskCache`, :class:`CachedFetch`).
    +
    +These used to live inline inside ``bitwarden.py``.  Pulling them here means
    +the atomic-write / ``0600`` / TTL logic is audited and fixed in exactly one
    +place instead of drifting across copy-pasted per-backend modules — each
    +backend supplies only its own cache-key shape and a serializer for it.
    +
    +Nothing in this module ever raises out to the caller's hot path: the disk
    +layer is strictly best-effort (a miss just triggers a refetch), because a
    +cache problem must never block Hermes startup.
    +"""
    +
    +from __future__ import annotations
    +
    +import json
    +import os
    +import tempfile
    +import time
    +from dataclasses import dataclass
    +from pathlib import Path
    +from typing import Callable, Dict, Generic, Optional, TypeVar
    +
    +__all__ = [
    +    "FetchResult",
    +    "CachedFetch",
    +    "DiskCache",
    +    "is_valid_env_name",
    +    "resolve_cache_home",
    +]
    +
    +
    +# ---------------------------------------------------------------------------
    +# Result object + env-name validation — canonical definitions live in
    +# ``agent.secret_sources.base`` (the SecretSource contract module); re-exported
    +# here so backends that import from ``_cache`` keep working.
    +# ---------------------------------------------------------------------------
    +
    +from agent.secret_sources.base import (  # noqa: E402
    +    FetchResult,
    +    is_valid_env_name,
    +)
    +
    +
    +# ---------------------------------------------------------------------------
    +# Cache entry
    +# ---------------------------------------------------------------------------
    +
    +
    +@dataclass
    +class CachedFetch:
    +    """A set of fetched secret values plus when they were fetched."""
    +
    +    secrets: [redacted:secret-assignment] str]
    +    fetched_at: float
    +
    +    def is_fresh(self, ttl_seconds: float) -> bool:
    +        if ttl_seconds <= 0:
    +            return False
    +        return (time.time() - self.fetched_at) < ttl_seconds
    +
    +
    +
    +
    +# ---------------------------------------------------------------------------
    +# Disk cache
    +# ---------------------------------------------------------------------------
    +
    +
    +def resolve_cache_home(home_path: Optional[Path] = None) -> Path:
    +    """Resolve the Hermes home used for cache paths.
    +
    +    ``home_path`` is whatever ``load_hermes_dotenv()`` already resolved;
    +    falling back to ``$HERMES_HOME`` / ``~/.hermes`` keeps direct callers
    +    (and tests that don't thread a home through) working.
    +    """
    +    if home_path is None:
    +        home_path = Path(os.getenv("HERMES_HOME", Path.home() / ".hermes"))
    +    return home_path
    +
    +
    +K = TypeVar("K")
    +
    +
    +class DiskCache(Generic[K]):
    +    """Best-effort, profile-aware on-disk cache for fetched secret values.
    +
    +    One JSON object per backend lives at ``<hermes_home>/cache/<basename>``::
    +
    +        {"key": "<serialized cache key>", "secrets": {...}, "fetched_at": 1.0}
    +
    +    The file holds only secret *values* keyed by the serialized cache key —
    +    never raw auth material.  Backends are responsible for fingerprinting
    +    tokens/sessions *before* they reach ``key_serializer`` so the token can't
    +    land in the key.
    +
    +    Writes are atomic (``mkstemp`` → ``chmod 0600`` → ``os.replace``) and the
    +    containing ``cache/`` directory is forced to ``0700`` — ``mkdir``'s mode is
    +    umask-subject, so the chmod is the reliable form.  Both ``read`` and
    +    ``write`` short-circuit when ``ttl_seconds <= 0``, so setting the TTL to
    +    zero disables *both* cache layers symmetrically: a user opting out never
    +    gets secret values written to disk at all.
    +    """
    +
    +    def __init__(self, basename: str, *, key_serializer: Callable[[K], str]) -> None:
    +        self._basename = basename
    +        self._key_serializer = key_serializer
    +        # Temp-file prefix derived from the basename so concurrent writers for
    +        # different backends in the same dir don't collide on the staging name.
    +        stem = basename.split(".", 1)[0]
    +        self._tmp_prefix = f".{stem}_"
    +
    +    def path(self, home_path: Optional[Path] = None) -> Path:
    +        return resolve_cache_home(home_path) / "cache" / self._basename
    +
    +    def read(
    +        self,
    +        key: K,
    +        ttl_seconds: float,
    +        home_path: Optional[Path] = None,
    +    ) -> Optional[CachedFetch]:
    +        """Return a fresh cached entry for ``key``, or None.
    +
    +        Best-effort: any I/O or parse error, a key mismatch, or a stale entry
    +        all return None so the caller re-fetches.
    +        """
    +        if ttl_seconds <= 0:
    +            return None
    +        path = self.path(home_path)
    +        try:
    +            with open(path, "r", encoding="utf-8") as f:
    +                payload = json.load(f)
    +        except (OSError, json.JSONDecodeError):
    +            return None
    +        if not isinstance(payload, dict):
    +            return None
    +        if payload.get("key") != self._key_serializer(key):
    +            return None
    +        secrets = [redacted:secret-assignment]"secrets")
    +        fetched_at = payload.get("fetched_at")
    +        if not isinstance(secrets, dict) or not isinstance(fetched_at, (int, float)):
    +            return None
    +        # JSON permits non-string values; env vars need strings, so coerce by
    +        # dropping anything that isn't a str→str pair.
    +        typed: Dict[str, str] = {
    +            k: v for k, v in secrets.items() if isinstance(k, str) and isinstance(v, str)
    +        }
    +        entry = CachedFetch(secrets=[redacted:secret-assignment] fetched_at=float(fetched_at))
    +        if not entry.is_fresh(ttl_seconds):
    +            return None
    +        return entry
    +
    +    def write(
    +        self,
    +        key: K,
    +        entry: CachedFetch,
    +        ttl_seconds: float,
    +        home_path: Optional[Path] = None,
    +    ) -> None:
    +        """Persist ``entry`` for ``key`` atomically at mode ``0600``.
    +
    +        No-op when ``ttl_seconds <= 0`` (so caching is genuinely off) or on any
    +        I/O error — the next invocation just re-fetches.
    +        """
    +        if ttl_seconds <= 0:
    +            return
    +        path = self.path(home_path)
    +        try:
    +            cache_dir = path.parent
    +            cache_dir.mkdir(parents=True, exist_ok=True)
    +            # mkdir's mode is umask-subject; chmod the dir to 0700 so cache
    +            # metadata isn't exposed if HERMES_HOME is ever made traversable.
    +            try:
    +                os.chmod(cache_dir, 0o700)
    +            except OSError:
    +                pass
    +            payload = {
    +                "key": self._key_serializer(key),
    +                "secrets": entry.secrets,
    +                "fetched_at": entry.fetched_at,
    +            }
    +            # Write to a sibling temp file and atomic-rename.  tempfile honours
    +            # os.umask, so we explicitly chmod 0600 before the rename.
    +            fd, tmp = tempfile.mkstemp(
    +                prefix=self._tmp_prefix, suffix=".tmp", dir=str(cache_dir)
    +            )
    +            try:
    +                with os.fdopen(fd, "w", encoding="utf-8") as f:
    +                    json.dump(payload, f)
    +                os.chmod(tmp, 0o600)
    +                os.replace(tmp, path)
    +            except BaseException:
    +                try:
    +                    os.unlink(tmp)
    +                except OSError:
    +                    pass
    +                raise
    +        except OSError:
    +            pass  # best-effort — a disk-cache miss next invocation is fine
    +
    +    def clear(self, home_path: Optional[Path] = None) -> None:
    +        """Delete the on-disk cache file if present (idempotent)."""
    +        try:
    +            self.path(home_path).unlink()
    +        except (FileNotFoundError, OSError):
    +            pass
    ```
  - `agent/secret_sources/base.py`

    ```diff
    @@ -0,0 +1,274 @@
    +"""Secret-source contract: the ABC every secret backend implements.
    +
    +A *secret source* resolves credentials from an external secret manager
    +(Bitwarden Secrets Manager, 1Password, an OS keystore, a user script, ...)
    +into environment-variable-shaped values at process startup, AFTER
    +``~/.hermes/.env`` has loaded and BEFORE the rest of Hermes reads
    +``os.environ``.
    +
    +Scope of the contract (deliberate, please do not widen):
    +
    +* **Read-only.**  Sources resolve refs → values.  There is no write-back
    +  ("save this key to your vault"), no arbitrary secret objects, and no
    +  mid-session secret API.  If a future need for rotation/refresh appears
    +  it will arrive as a versioned optional hook — do not bolt it on.
    +* **Startup-time, synchronous.**  ``fetch()`` is called once per process
    +  (per HERMES_HOME) by the orchestrator in
    +  :mod:`agent.secret_sources.registry`, which enforces a wall-clock
    +  timeout around it.  Sources must not spawn background refreshers.
    +* **Never raises, never prompts.**  ``fetch()`` returns a
    +  :class:`FetchResult` — errors go in ``result.error`` with a
    +  machine-readable :class:`ErrorKind`.  Interactive auth belongs in the
    +  source's CLI ``setup`` flow, never on the startup path (non-TTY
    +  gateway/cron startup must never block on stdin).
    +* **Sources fetch; the orchestrator applies.**  A source returns the
    +  name→value mapping it *would* contribute.  Precedence (mapped-beats-bulk,
    +  first-wins, ``override_existing``, protected vars), conflict warnings,
    +  provenance tracking, and the actual ``os.environ`` writes are owned by
    +  the orchestrator so no backend can get them wrong.
    +
    +Versioning: ``SECRET_SOURCE_API_VERSION`` gates plugin compatibility.
    +New *optional* hooks with default implementations do not bump it;
    +required-signature changes do, and the registry skips (with a warning)
    +sources built against a different major version instead of crashing
    +startup.
    +"""
    +
    +from __future__ import annotations
    +
    +import os
    +import re
    +import subprocess
    +from abc import ABC, abstractmethod
    +from dataclasses import dataclass, field
    +from enum import Enum
    +from pathlib import Path
    +from typing import Dict, FrozenSet, List, Optional, Sequence
    +
    +# Bump ONLY for breaking changes to the required contract surface
    +# (abstract-method signatures, FetchResult required fields).  Additive
    +# optional hooks must ship with defaults and must NOT bump this.
    +SECRET_SOURCE_API_VERSION = 1
    +
    +# Timeout the orchestrator enforces around fetch() when the source's
    +# config section doesn't override it.  Generous because a first run may
    +# include a one-time CLI binary auto-install (e.g. bws download+verify).
    +DEFAULT_FETCH_TIMEOUT_SECONDS = 120.0
    +
    +# Default timeout for run_secret_cli() subprocess invocations.
    +DEFAULT_CLI_TIMEOUT_SECONDS = 30.0
    +
    +
    +class ErrorKind(str, Enum):
    +    """Machine-readable failure taxonomy for :class:`FetchResult.error`.
    +
    +    A fixed vocabulary keeps startup warnings and ``hermes secrets status``
    +    uniform across backends, and lets the orchestrator implement
    +    kind-dependent policy (e.g. a future stale-cache fallback on
    +    ``NETWORK``/``TIMEOUT`` but not on ``AUTH_FAILED``) exactly once.
    +    """
    +
    +    NOT_CONFIGURED = "not_configured"    # enabled but missing token/project/map
    +    BINARY_MISSING = "binary_missing"    # helper CLI not found / not installed
    +    AUTH_FAILED = "auth_failed"          # bad credentials
    +    AUTH_EXPIRED = "auth_expired"        # credentials were valid, aren't now
    +    REF_INVALID = "ref_invalid"          # a secret reference failed validation
    +    NETWORK = "network"                  # transport-level failure
    +    EMPTY_VALUE = "empty_value"          # backend returned nothing for a ref
    +    TIMEOUT = "timeout"                  # fetch exceeded its wall-clock budget
    +    INTERNAL = "internal"                # anything else (bug, unexpected shape)
    +
    +
    +@dataclass
    +class FetchResult:
    +    """Outcome of one source's fetch.
    +
    +    ``secrets`` holds what the source *would* contribute; whether each
    +    var is actually applied is the orchestrator's decision.  ``applied``
    +    and ``skipped`` exist for backward compatibility with the original
    +    Bitwarden fetch-and-apply entry point and are left empty by
    +    conforming ``fetch()`` implementations.
    +    """
    +
    +    secrets: [redacted:secret-assignment] str] = field(default_factory=dict)
    +    applied: List[str] = field(default_factory=list)
    +    skipped: List[str] = field(default_factory=list)
    +    warnings: List[str] = field(default_factory=list)
    +    error: Optional[str] = None
    +    error_kind: Optional[ErrorKind] = None
    +    # Path of the helper binary used, when the source is CLI-driven.
    +    # Surfaced by status commands; None for SDK/API-driven sources.
    +    binary_path: Optional[Path] = None
    +
    +    @property
    +    def ok(self) -> bool:
    +        return self.error is None
    +
    +
    +class SecretSource(ABC):
    +    """One external secret backend.
    +
    +    Subclasses set the class attributes and implement :meth:`fetch`.
    +    Everything else has a sensible default.
    +
    +    Attributes:
    +        name: Config-section key under ``secrets:`` in config.yaml.
    +            Lowercase ``[a-z0-9_]+``.  Also the provenance label stored
    +            for every var this source supplies.
    +        label: Human-readable name used in startup messages and
    +            ``hermes secrets status`` (e.g. ``"Bitwarden Secrets Manager"``).
    +        shape: ``"mapped"`` when the user explicitly binds env-var names
    +            to refs (1Password ``env:`` map, command source) or
    +            ``"bulk"`` when the backend injects whole projects/folders
    +            of secrets implicitly (Bitwarden BSM).  The orchestrator
    +            gives mapped sources precedence over bulk sources: an
    +            explicit binding is stronger intent than a project dump.
    +        scheme: Optional URI scheme this source owns for secret
    +            references (``"op"`` for ``op://...``).  Must be unique
    +            across registered sources — refs may eventually appear
    +            outside the ``secrets:`` block (e.g. credential-pool
    +            ``api_key`` fields), so scheme collisions are rejected at
    +            registration time to keep that future possible.
    +        api_version: Contract version this source was built against.
    +    """
    +
    +    api_version: int = SECRET_SOURCE_API_VERSION
    +    name: str = ""
    +    label: str = ""
    +    shape: str = "mapped"  # "mapped" | "bulk"
    +    scheme: Optional[str] = None
    +
    +    # -- required ----------------------------------------------------------
    +
    +    @abstractmethod
    +    def fetch(self, cfg: dict, home_path: Path) -> FetchResult:
    +        """Resolve this source's secrets. MUST NOT raise or prompt.
    +
    +        ``cfg`` is the source's raw config section (``secrets.<name>``)
    +        from config.yaml — treat every field defensively, the section
    +        may be malformed.  ``home_path`` is the resolved HERMES_HOME.
    +        """
    +
    +    # -- optional hooks (defaults are correct for most sources) ------------
    +
    +    def is_enabled(self, cfg: dict) -> bool:
    +        """Whether the user turned this source on."""
    +        return bool(isinstance(cfg, dict) and cfg.get("enabled"))
    +
    +    def override_existing(self, cfg: dict) -> bool:
    +        """May this source overwrite vars that .env / the shell already set?
    +
    +        This NEVER extends to vars claimed by another secret source in the
    +        same startup pass — cross-source overrides are a config error the
    +        orchestrator warns about, not a knob.
    +        """
    +        return bool(isinstance(cfg, dict) and cfg.get("override_existing", False))
    +
    +    def protected_env_vars(self, cfg: dict) -> FrozenSet[str]:
    +        """Env vars the orchestrator must never let ANY source overwrite.
    +
    +        Typically the source's own bootstrap-auth var (e.g.
    +        ``BWS_ACCESS_TOKEN``) so a vault that contains its own access
    +        token can't clobber the credential used to reach it.
    +        """
    +        return frozenset()
    +
    +    def fetch_timeout_seconds(self, cfg: dict) -> float:
    +        """Wall-clock budget the orchestrator enforces around fetch()."""
    +        try:
    +            val = float((cfg or {}).get("timeout_seconds", DEFAULT_FETCH_TIMEOUT_SECONDS))
    +        except (TypeError, ValueError):
    +            return DEFAULT_FETCH_TIMEOUT_SECONDS
    +        return val if val > 0 else DEFAULT_FETCH_TIMEOUT_SECONDS
    +
    +    def config_schema(self) -> dict:
    +        """Optional description of this source's config keys.
    +
    +        Shape: ``{key: {"description": str, "default": Any}}``.  Used by
    +        setup surfaces to render config without hardcoding per-source
    +        knowledge.  Purely informational.
    +        """
    +        return {}
    +
    +
    +# ---------------------------------------------------------------------------
    +# Shared helpers — use these instead of hand-rolling per backend
    +# ---------------------------------------------------------------------------
    +
    +
    +_ENV_NAME_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$")
    +
    +# ANSI CSI/OSC escape sequences — helper-CLI stderr often carries color
    +# codes that must not reach Hermes' own startup output.
    +_ANSI_RE = re.compile(r"\x1b(?:\[[0-9;?]*[ -/]*[@-~]|\][^\x07\x1b]*(?:\x07|\x1b\\)?)")
    +
    +
    +def is_valid_env_name(name: str) -> bool:
    +    """True when ``name`` is a legal environment-variable name."""
    +    return bool(name) and bool(_ENV_NAME_RE.match(name))
    +
    +
    +def scrub_ansi(text: str) -> str:
    +    """Strip ANSI escape sequences (whole CSI/OSC sequences, not just ESC)."""
    +    return _ANSI_RE.sub("", text or "")
    +
    +
    +def run_secret_cli(
    +    argv: Sequence[str],
    +    *,
    +    allow_env: Sequence[str] = (),
    +    extra_env: Optional[Dict[str, str]] = None,
    +    timeout: float = DEFAULT_CLI_TIMEOUT_SECONDS,
    +) -> subprocess.CompletedProcess:
    +    """Run a secret-manager helper CLI with a minimal, allowlisted env.
    +
    +    Security posture shared by every subprocess-driven backend:
    +
    +    * argv list only — never ``shell=True``.  Callers pass user-supplied
    +      reference strings AFTER a ``--`` option terminator in their argv.
    +    * The child gets ``PATH``/``HOME``/locale basics plus only the env
    +      vars named in ``allow_env`` (auth/session vars) and ``extra_env``
    +      — never a copy of the full post-dotenv ``os.environ``, which by
    +      this point holds every credential Hermes knows about.
    +    * ``NO_COLOR=1`` is set and stderr/stdout are ANSI-scrubbed so
    +      helper diagnostics can't smuggle escape sequences into Hermes
    +      output.
    +    * stdin is ``/dev/null`` so a helper that decides to prompt fails
    +      fast instead of hanging startup.
    +
    +    Raises ``RuntimeError`` on spawn failure or timeout (message safe to
    +    surface); returns the completed process otherwise — callers own
    +    returncode interpretation.
    +    """
    +    base_keep = ("PATH", "HOME", "USERPROFILE", "SYSTEMROOT", "TMPDIR", "TEMP",
    +                 "LANG", "LC_ALL", "XDG_CONFIG_HOME", "XDG_DATA_HOME")
    +    env: Dict[str, str] = {}
    +    for key in (*base_keep, *allow_env):
    +        val = os.environ.get(key)
    +        if val is not None:
    +            env[key] = val
    +    if extra_env:
    +        env.update(extra_env)
    +    env.setdefault("NO_COLOR", "1")
    +
    +    try:
    +        proc = subprocess.run(  # noqa: S603 — argv list, no shell
    +            list(argv),
    +            env=env,
    +            capture_output=True,
    +            text=True,
    +            timeout=timeout,
    +            stdin=subprocess.DEVNULL,
    +        )
    +    except subprocess.TimeoutExpired as exc:
    +        raise RuntimeError(
    +            f"{Path(str(argv[0])).name} timed out after {timeout:.0f}s"
    +        ) from exc
    +    except OSError as exc:
    +        raise RuntimeError(
    +            f"failed to invoke {Path(str(argv[0])).name}: {exc}"
    +        ) from exc
    +
    +    proc.stdout = proc.stdout or ""
    +    proc.stderr = scrub_ansi(proc.stderr or "")
    +    return proc
    ```
  - `agent/secret_sources/bitwarden.py`

    ```diff
    @@ -42,10 +42,17 @@
     import urllib.error
     import urllib.request
     import zipfile
    -from dataclasses import dataclass, field
     from pathlib import Path
     from typing import Dict, List, Optional, Tuple
     
    +from agent.secret_sources._cache import (
    +    CachedFetch as _CachedFetch,
    +    DiskCache,
    +    FetchResult,
    +    is_valid_env_name as _is_valid_env_name,
    +)
    +from agent.secret_sources.base import ErrorKind, SecretSource
    +
     logger = logging.getLogger(__name__)
     
     
    ```

    ```diff
    @@ -70,7 +77,7 @@
     # In-process cache so repeated load_hermes_dotenv() calls (CLI startup,
     # gateway hot-reload, test suites) don't re-fetch from BSM.
     _CacheKey = Tuple[str, str, str]  # (access_token_fingerprint, project_id, server_url)
    -_CACHE: Dict[_CacheKey, "_CachedFetch"] = {}
    +_CACHE: Dict[_CacheKey, _CachedFetch] = {}
     
     # Disk-persisted cache so back-to-back CLI invocations (e.g. `hermes chat -q ...`
     # called from scripts, cron, the gateway forking new agents) don't each pay the
    ```

    ```diff
    @@ -81,124 +88,29 @@
     # <hermes_home>/cache/bws_cache.json. The file holds only the secret VALUES,
     # never the access token. It's plaintext-equivalent to ~/.hermes/.env (which
     # we already accept) but kept out of the .env file so users editing it won't
    -# accidentally commit BSM-sourced secrets.
    +# accidentally commit BSM-sourced secrets. The atomic-write/0600/TTL mechanics
    +# live in agent.secret_sources._cache.DiskCache, shared with the other backends.
     _DISK_CACHE_BASENAME = "bws_cache.json"
     
     
    -def _disk_cache_path(home_path: Optional[Path] = None) -> Path:
    -    """Return the disk cache path under hermes_home/cache/.
    -
    -    `home_path` is what `load_hermes_dotenv()` already resolved; falling back
    -    to `$HERMES_HOME` / `~/.hermes` keeps direct callers working too.
    -    """
    -    if home_path is None:
    -        home_path = Path(os.getenv("HERMES_HOME", Path.home() / ".hermes"))
    -    return home_path / "cache" / _DISK_CACHE_BASENAME
    -
    -
     def _cache_key_str(cache_key: _CacheKey) -> str:
         """Serialize a cache key to a stable string for JSON storage."""
         token_fp, project_id, server_url = cache_key
         return f"{token_fp}|{project_id}|{server_url}"
     
     
    -def _read_disk_cache(cache_key: _CacheKey, ttl_seconds: float,
    -                     home_path: Optional[Path] = None) -> Optional["_CachedFetch"]:
    -    """Return a cached entry from disk if fresh, else None.
    -
    -    Best-effort: any I/O or parse error returns None and we re-fetch.
    -    """
    -    if ttl_seconds <= 0:
    -        return None
    -    path = _disk_cache_path(home_path)
    -    try:
    -        with open(path, "r", encoding="utf-8") as f:
    -            payload = json.load(f)
    -    except (OSError, json.JSONDecodeError):
    -        return None
    -    if not isinstance(payload, dict):
    -        return None
    -    if payload.get("key") != _cache_key_str(cache_key):
    -        return None
    -    secrets = [redacted:secret-assignment]"secrets")
    -    fetched_at = payload.get("fetched_at")
    -    if not isinstance(secrets, dict) or not isinstance(fetched_at, (int, float)):
    -        return None
    -    # Coerce all values to strings — JSON allows numbers but env vars need strings
    -    typed_secrets: [redacted:secret-assignment] str] = {
    -        k: v for k, v in secrets.items() if isinstance(k, str) and isinstance(v, str)
    -    }
    -    entry = _CachedFetch(secrets=[redacted:secret-assignment] fetched_at=float(fetched_at))
    -    if not entry.is_fresh(ttl_seconds):
    -        return None
    -    return entry
    -
    -
    -def _write_disk_cache(cache_key: _CacheKey, entry: "_CachedFetch",
    -                      home_path: Optional[Path] = None) -> None:
    -    """Persist a cache entry to disk atomically with mode 0600.
    -
    -    Best-effort: any I/O error is swallowed (the next invocation will just
    -    re-fetch). We never want disk cache failures to break startup.
    -    """
    -    path = _disk_cache_path(home_path)
    -    try:
    -        path.parent.mkdir(parents=True, exist_ok=True)
    -        payload = {
    -            "key": _cache_key_str(cache_key),
    -            "secrets": entry.secrets,
    -            "fetched_at": entry.fetched_at,
    -        }
    -        # Write to a temp file in the same directory and atomic-rename.
    -        # tempfile honors os.umask, so we explicitly chmod 0600 before rename.
    -        fd, tmp = tempfile.mkstemp(
    -            prefix=".bws_cache_", suffix=".tmp", dir=str(path.parent)
    -        )
    -        try:
    -            with os.fdopen(fd, "w", encoding="utf-8") as f:
    -                json.dump(payload, f)
    -            os.chmod(tmp, 0o600)
    -            os.replace(tmp, path)
    -        except BaseException:
    -            try:
    -                os.unlink(tmp)
    -            except OSError:
    -                pass
    -            raise
    -    except OSError:
    -        pass  # best-effort — disk cache miss on next invocation is fine
    -
    -
    -@dataclass
    -class _CachedFetch:
    -    secrets: [redacted:secret-assignment] str]
    -    fetched_at: float
    -
    -    def is_fresh(self, ttl_seconds: float) -> bool:
    -        if ttl_seconds <= 0:
    -            return False
    -        return (time.time() - self.fetched_at) < ttl_seconds
    -
    -
    -# ---------------------------------------------------------------------------
    -# Public dataclasses
    -# ---------------------------------------------------------------------------
    -
    +_DISK_CACHE: DiskCache = DiskCache(
    +    _DISK_CACHE_BASENAME, key_serializer=_cache_key_str
    +)
     
    -@dataclass
    -class FetchResult:
    -    """Outcome of a single BSM pull."""
     
    -    secrets: [redacted:secret-assignment] str] = field(default_factory=dict)
    -    applied: List[str] = field(default_factory=list)   # set into os.environ
    -    skipped: List[str] = field(default_factory=list)   # already set, not overridden
    -    warnings: List[str] = field(default_factory=list)  # non-fatal issues
    -    error: Optional[str] = None                        # fatal: nothing was fetched
    -    binary_path: Optional[Path] = None
    +def _disk_cache_path(home_path: Optional[Path] = None) -> Path:
    +    """Return the disk cache path under hermes_home/cache/.
     
    -    @property
    -    def ok(self) -> bool:
    -        return self.error is None
    +    Thin wrapper over the shared DiskCache, kept for tests and any direct
    +    callers; falls back to `$HERMES_HOME` / `~/.hermes` when home is None.
    +    """
    +    return _DISK_CACHE.path(home_path)
     
     
     # ---------------------------------------------------------------------------
    ```

    ```diff
    @@ -479,7 +391,7 @@ def fetch_bitwarden_secrets(
             if cached and cached.is_fresh(cache_ttl_seconds):
                 return cached.secrets, []
             # L2: disk cache. ~5ms on cache hit vs ~380ms for `bws secret list`.
    -        disk_cached = _read_disk_cache(cache_key, cache_ttl_seconds, home_path)
    +        disk_cached = _DISK_CACHE.read(cache_key, cache_ttl_seconds, home_path)
             if disk_cached is not None:
                 # Promote into in-process cache so subsequent fetches in the
                 # same process skip the disk read too.
    ```

    ```diff
    @@ -499,7 +411,7 @@ def fetch_bitwarden_secrets(
         entry = _CachedFetch(secrets=[redacted:secret-assignment] fetched_at=time.time())
         _CACHE[cache_key] = entry
         if use_cache:
    -        _write_disk_cache(cache_key, entry, home_path)
    +        _DISK_CACHE.write(cache_key, entry, cache_ttl_seconds, home_path)
         return secrets, warnings
     
     
    ```

    ```diff
    @@ -575,14 +487,6 @@ def _run_bws_list(
         return secrets, warnings
     
     
    -def _is_valid_env_name(name: str) -> bool:
    -    if not name:
    -        return False
    -    if not (name[0].isalpha() or name[0] == "_"):
    -        return False
    -    return all(c.isalnum() or c == "_" for c in name)
    -
    -
     # ---------------------------------------------------------------------------
     # Public entry point — called from hermes_cli.env_loader
     # ---------------------------------------------------------------------------
    ```

    ```diff
    @@ -673,6 +577,142 @@ def apply_bitwarden_secrets(
         return result
     
     
    +# ---------------------------------------------------------------------------
    +# SecretSource adapter — the registry-facing wrapper around this module.
    +# ---------------------------------------------------------------------------
    +
    +
    +class BitwardenSource(SecretSource):
    +    """Bitwarden Secrets Manager as a registered secret source.
    +
    +    Thin adapter over the module's fetch machinery.  ``fetch()`` only
    +    *fetches* — precedence, override semantics, conflict warnings, and
    +    the ``os.environ`` writes are the orchestrator's job
    +    (see ``agent.secret_sources.registry.apply_all``).
    +
    +    Bitwarden is a **bulk** source: it injects every secret in the
    +    configured BSM project, so explicit per-var bindings from mapped
    +    sources (e.g. the 1Password ``env:`` map) outrank it.
    +    """
    +
    +    name = "bitwarden"
    +    label = "Bitwarden Secrets Manager"
    +    shape = "bulk"
    +    scheme = "bws"
    +
    +    def override_existing(self, cfg: dict) -> bool:
    +        # Default True (matches DEFAULT_CONFIG): the point of BSM is
    +        # centralized rotation — if .env had the final say, rotating a
    +        # key in Bitwarden wouldn't take effect until the stale .env
    +        # line was also deleted.
    +        return bool(isinstance(cfg, dict) and cfg.get("override_existing", True))
    +
    +    def protected_env_vars(self, cfg: dict):
    +        token_env = "[redacted:secret-assignment]"
    +        if isinstance(cfg, dict):
    +            token_env = [redacted:secret-assignment]"access_token_env") or token_env)
    +        return frozenset({token_env})
    +
    +    def config_schema(self) -> dict:
    +        return {
    +            "enabled": {"description": "Master switch", "default": False},
    +            "access_token_env": {
    +                "description": "Env var holding the machine-account access token",
    +                "default": "BWS_ACCESS_TOKEN",
    +            },
    +            "project_id": {"description": "BSM project UUID", "default": ""},
    +            "cache_ttl_seconds": {
    +                "description": "Disk+memory cache TTL; 0 disables",
    +                "default": 300,
    +            },
    +            "override_existing": {
    +                "description": "BSM values overwrite .env/shell values",
    +                "default": True,
    +            },
    +            "auto_install": {
    +                "description": "Auto-download the pinned bws binary",
    +                "default": True,
    +            },
    +            "server_url": {
    +                "description": "Region / self-hosted endpoint (empty = US Cloud)",
    +                "default": "",
    +            },
    +        }
    +
    +    def fetch(self, cfg: dict, home_path: Path) -> FetchResult:
    +        cfg = cfg if isinstance(cfg, dict) else {}
    +        result = FetchResult()
    +
    +        access_token_env = [redacted:secret-assignment]"access_token_env") or "BWS_ACCESS_TOKEN")
    +        access_token = [redacted:secret-assignment] "").strip()
    +        if not access_token:
    +            result.error = (
    +                f"secrets.bitwarden.enabled is true but {access_token_env} is "
    +                "not set.  Run `hermes secrets bitwarden setup`."
    +            )
    +            result.error_kind = ErrorKind.NOT_CONFIGURED
    +            return result
    +
    +        project_id = str(cfg.get("project_id") or "")
    +        if not project_id:
    +            result.error = (
    +                "secrets.bitwarden.project_id is empty.  "
    +                "Run `hermes secrets bitwarden setup`."
    +            )
    +            result.error_kind = ErrorKind.NOT_CONFIGURED
    +            return result
    +
    +        auto_install = bool(cfg.get("auto_install", True))
    +        binary = find_bws(install_if_missing=auto_install)
    +        result.binary_path = binary
    +        if binary is None:
    +            result.error = (
    +                "bws binary not available and auto-install is disabled.  "
    +                "Run `hermes secrets bitwarden setup` to install."
    +            )
    +            result.error_kind = ErrorKind.BINARY_MISSING
    +            return result
    +
    +        try:
    +            ttl = float(cfg.get("cache_ttl_seconds", 300))
    +        except (TypeError, ValueError):
    +            ttl = 300.0
    +
    +        try:
    +            secrets, warnings = fetch_bitwarden_secrets(
    +                access_token=[redacted:secret-assignment]
    +                project_id=project_id,
    +                binary=binary,
    +                cache_ttl_seconds=ttl,
    +                server_url=str(cfg.get("server_url", "") or "").strip(),
    +                home_path=home_path,
    +            )
    +        except RuntimeError as exc:
    +            result.error = str(exc)
    +            result.error_kind = _classify_bws_error(str(exc))
    +            return result
    +
    +        result.secrets = secrets
    +        result.warnings.extend(warnings)
    +        return result
    +
    +
    +def _classify_bws_error(message: str) -> ErrorKind:
    +    """Best-effort mapping of bws failure text onto the shared taxonomy."""
    +    lowered = message.lower()
    +    if "timed out" in lowered:
    +        return ErrorKind.TIMEOUT
    +    if "binary not available" in lowered or "failed to invoke" in lowered:
    +        return ErrorKind.BINARY_MISSING
    +    if any(tok in lowered for tok in ("unauthorized", "invalid token",
    +                                      "access token", "401", "403")):
    +        return ErrorKind.AUTH_FAILED
    +    if any(tok in lowered for tok in ("network", "connection", "resolve",
    +                                      "download", "dns")):
    +        return ErrorKind.NETWORK
    +    return ErrorKind.INTERNAL
    +
    +
     # ---------------------------------------------------------------------------
     # Test hook — used by hermetic tests to flush the cache between cases.
     # ---------------------------------------------------------------------------
    ```

    ```diff
    @@ -686,7 +726,4 @@ def _reset_cache_for_tests(home_path: Optional[Path] = None) -> None:
         writer itself.
         """
         _CACHE.clear()
    -    try:
    -        _disk_cache_path(home_path).unlink()
    -    except (FileNotFoundError, OSError):
    -        pass
    +    _DISK_CACHE.clear(home_path)
    ```
  - `agent/secret_sources/onepassword.py`

    ```diff
    @@ -0,0 +1,643 @@
    +"""1Password (`op` CLI) secret source.
    +
    +Resolve provider credentials from 1Password ``op://vault/item/field``
    +references at process startup so they don't have to live in plaintext in
    +``~/.hermes/.env``.
    +
    +Design summary
    +--------------
    +
    +* Users map environment-variable names to official 1Password secret
    +  references in ``secrets.onepassword.env``::
    +
    +      secrets:
    +        onepassword:
    +          enabled: true
    +          env:
    +            OPENAI_API_KEY: "[redacted:secret-assignment] key"
    +            ANTHROPIC_API_KEY: "[redacted:secret-assignment]"
    +
    +* After ``.env`` loads, each reference is resolved with a single
    +  ``op read -- <reference>`` call and injected into ``os.environ`` (the
    +  same point in startup as the Bitwarden source).
    +* Authentication is whatever the user's ``op`` CLI already uses — a
    +  service-account token (``OP_SERVICE_ACCOUNT_TOKEN``) for headless boxes,
    +  or a desktop/interactive session (``OP_SESSION_*``).  Hermes never
    +  authenticates on the user's behalf; it shells out to an already-trusted,
    +  already-authenticated CLI.
    +* Failures NEVER block startup.  A missing ``op`` binary, expired auth, a
    +  bad reference, or a permission error each surface a one-line warning and
    +  Hermes continues with whatever credentials ``.env`` already had.
    +
    +The atomic-write / ``0600`` / TTL cache mechanics are shared with the other
    +backends via :mod:`agent.secret_sources._cache` — successful, complete pulls
    +are cached in-process and on disk under ``<hermes_home>/cache/op_cache.json``
    +so back-to-back short-lived ``hermes`` invocations don't re-shell ``op`` for
    +every reference.  The disk file holds only resolved secret *values*; auth
    +material is fingerprinted, never stored.
    +"""
    +
    +from __future__ import annotations
    +
    +import hashlib
    +import logging
    +import os
    +import re
    +import shutil
    +import subprocess
    +import time
    +from pathlib import Path
    +from typing import Dict, List, Optional, Tuple
    +
    +from agent.secret_sources._cache import (
    +    CachedFetch,
    +    DiskCache,
    +    FetchResult,
    +    is_valid_env_name,
    +)
    +from agent.secret_sources.base import ErrorKind, SecretSource
    +
    +logger = logging.getLogger(__name__)
    +
    +
    +# ---------------------------------------------------------------------------
    +# Configuration constants
    +# ---------------------------------------------------------------------------
    +
    +# How long to wait for a single `op read`, in seconds.
    +_OP_RUN_TIMEOUT = 30
    +
    +# Default env var the official `op` CLI reads for service-account auth.  Users
    +# can point `service_account_token_env` at a different name; we always export
    +# the value to the child as OP_SERVICE_ACCOUNT_TOKEN, which is what `op` itself
    +# looks for.
    +_DEFAULT_TOKEN_ENV = "[redacted:secret-assignment]"
    +
    +# Strip whole ANSI CSI sequences (colour, cursor moves, line erases) from any
    +# `op` diagnostic we surface — not just the lone ESC byte — so a control
    +# sequence can't reposition the cursor or hide text after a redaction marker.
    +_ANSI_CSI_RE = re.compile(r"\x1b\[[0-?]*[ -/]*[@-~]")
    +
    +# Env vars the `op` child actually needs.  We build a minimal allowlisted env
    +# rather than copying all of os.environ (which, post-dotenv, holds every
    +# provider credential) into the child — tighter blast radius if `op` or
    +# anything it execs ever misbehaves.  OP_SESSION_* and the token are added
    +# dynamically in _op_child_env().
    +_OP_ENV_ALLOWLIST = (
    +    "PATH",
    +    "HOME",
    +    "USERPROFILE",
    +    "APPDATA",
    +    "LOCALAPPDATA",
    +    "SystemRoot",
    +    "TMPDIR",
    +    "TMP",
    +    "TEMP",
    +    "XDG_CONFIG_HOME",
    +    "XDG_RUNTIME_DIR",
    +    "OP_ACCOUNT",
    +    "OP_CONNECT_HOST",
    +    "OP_CONNECT_TOKEN",
    +)
    +
    +
    +# ---------------------------------------------------------------------------
    +# Cache
    +# ---------------------------------------------------------------------------
    +
    +# In-process cache.  The key folds in str(home_path) so a HERMES_HOME switch
    +# inside one long-lived process (e.g. the gateway) can't return another
    +# profile's secrets from L1.  The disk layer omits home from its serialized
    +# key because the file already lives under the home dir (see _disk_key_str).
    +_CacheKey = Tuple[str, str, str, str]  # (auth_fp, account, home, refs_fp)
    +_CACHE: Dict[_CacheKey, CachedFetch] = {}
    +
    +_DISK_CACHE_BASENAME = "op_cache.json"
    +
    +
    +def _disk_key_str(cache_key: _CacheKey) -> str:
    +    """Serialize a cache key for on-disk storage, omitting home_path.
    +
    +    The disk file is already partitioned by home (it lives under
    +    ``<home>/cache/``), so the path provides the home dimension; folding it
    +    into the key string too would be redundant.
    +    """
    +    auth_fp, account, _home, refs_fp = cache_key
    +    return f"{auth_fp}|{account}|{refs_fp}"
    +
    +
    +_DISK_CACHE: DiskCache = DiskCache(
    +    _DISK_CACHE_BASENAME, key_serializer=_disk_key_str
    +)
    +
    +
    +def _disk_cache_path(home_path: Optional[Path] = None) -> Path:
    +    """Path to the on-disk cache (exposed for tests and direct callers)."""
    +    return _DISK_CACHE.path(home_path)
    +
    +
    +# ---------------------------------------------------------------------------
    +# Reference validation + fingerprinting
    +# ---------------------------------------------------------------------------
    +
    +
    +def _validate_references(
    +    references: Optional[Dict[str, str]],
    +) -> Tuple[Dict[str, str], List[str]]:
    +    """Return ``(valid_refs, warnings)`` from an ``env`` mapping.
    +
    +    A reference is kept only if its target env-var name is a valid POSIX
    +    name and the value is a stripped ``op://…`` reference string.  Everything
    +    else produces a warning and is dropped (never fatal).
    +    """
    +    valid: Dict[str, str] = {}
    +    warnings: List[str] = []
    +    for name, ref in (references or {}).items():
    +        if not is_valid_env_name(name):
    +            warnings.append(f"Skipping {name!r}: not a valid env-var name")
    +            continue
    +        if not isinstance(ref, str):
    +            warnings.append(f"Skipping {name!r}: reference is not a string")
    +            continue
    +        cleaned = ref.strip()
    +        if not cleaned.startswith("op://"):
    +            warnings.append(
    +                f"Skipping {name!r}: {ref!r} is not an op:// secret reference"
    +            )
    +            continue
    +        valid[name] = cleaned
    +    return valid, warnings
    +
    +
    +def _auth_fingerprint(token_env: str) -> str:
    +    """SHA-256 prefix over the auth material `op` would use.
    +
    +    Folds in the service-account token, ``OP_ACCOUNT``, and *all*
    +    ``OP_SESSION_*`` vars (the names `op` actually exports for interactive
    +    sessions — ``OP_SESSION_<account_shorthand>``).  Signing out and into a
    +    different identity therefore changes the cache key, so a value cached under
    +    a previous identity is never served under a new one.  Never logged or
    +    displayed; the raw token never leaves this hash.
    +    """
    +    parts: List[str] = [
    +        f"token=[redacted:secret-assignment] '')}",
    +        f"account={os.environ.get('OP_ACCOUNT', '')}",
    +    ]
    +    for key in sorted(os.environ):
    +        if key.startswith("OP_SESSION_"):
    +            parts.append(f"{key}={os.environ[key]}")
    +    material = "\n".join(parts)
    +    return hashlib.sha256(material.encode("utf-8")).hexdigest()[:16]
    +
    +
    +def _refs_fingerprint(references: Dict[str, str]) -> str:
    +    """SHA-256 prefix over the configured name→reference mapping."""
    +    material = "\n".join(f"{name}={references[name]}" for name in sorted(references))
    +    return hashlib.sha256(material.encode("utf-8")).hexdigest()[:16]
    +
    +
    +# ---------------------------------------------------------------------------
    +# Binary discovery
    +# ---------------------------------------------------------------------------
    +
    +
    +def find_op(binary_path: str = "") -> Optional[Path]:
    +    """Resolve a usable ``op`` binary, or None.
    +
    +    When ``binary_path`` is set it is used verbatim and PATH is NOT consulted
    +    — pinning an absolute path is a way to avoid trusting whatever ``op`` shows
    +    up first on ``PATH``.  A pinned-but-missing path returns None (the caller
    +    surfaces a clear error) rather than silently falling back.
    +    """
    +    if binary_path:
    +        pinned = Path(binary_path)
    +        if pinned.exists() and os.access(pinned, os.X_OK):
    +            return pinned
    +        return None
    +    found = shutil.which("op")
    +    return Path(found) if found else None
    +
    +
    +# ---------------------------------------------------------------------------
    +# `op read` invocation
    +# ---------------------------------------------------------------------------
    +
    +
    +def _scrub(text: str) -> str:
    +    """Remove ANSI control sequences and trim, for safe message surfacing."""
    +    return _ANSI_CSI_RE.sub("", text).replace("\x1b", "").strip()
    +
    +
    +def _op_child_env(token_value: str) -> Dict[str, str]:
    +    """Build a minimal allowlisted environment for the ``op`` child process."""
    +    env: Dict[str, str] = {}
    +    for key in _OP_ENV_ALLOWLIST:
    +        val = os.environ.get(key)
    +        if val is not None:
    +            env[key] = val
    +    # Desktop / interactive session credentials.
    +    for key, val in os.environ.items():
    +        if key.startswith("OP_SESSION_"):
    +            env[key] = val
    +    # `op` reads OP_SERVICE_ACCOUNT_TOKEN regardless of which env var the user
    +    # configured Hermes to source it from, so normalize to that name here.
    +    if token_value:
    +        env["OP_SERVICE_ACCOUNT_TOKEN"] = token_value
    +    env["NO_COLOR"] = "1"
    +    return env
    +
    +
    +def _run_op_read(
    +    op: Path,
    +    reference: str,
    +    *,
    +    account: str = "",
    +    token_value: str = "",
    +) -> str:
    +    """Resolve a single ``op://`` reference to its value.
    +
    +    Raises :class:`RuntimeError` on any failure — including a ``returncode 0``
    +    with empty output, which would otherwise silently clobber a good
    +    ``.env``/shell credential with ``""``.
    +    """
    +    cmd: List[str] = [str(op), "read"]
    +    if account:
    +        cmd += ["--account", account]
    +    # `--` terminates option parsing so a reference can never be mis-parsed as
    +    # an `op` flag even if validation is ever loosened.
    +    cmd += ["--", reference]
    +
    +    try:
    +        proc = subprocess.run(  # noqa: S603 — op path is user-trusted, argv list
    +            cmd,
    +            env=_op_child_env(token_value),
    +            capture_output=True,
    +            text=True,
    +            encoding="utf-8",
    +            errors="replace",
    +            timeout=_OP_RUN_TIMEOUT,
    +        )
    +    except subprocess.TimeoutExpired as exc:
    +        raise RuntimeError(
    +            f"op read timed out after {_OP_RUN_TIMEOUT}s for {reference!r}"
    +        ) from exc
    +    except OSError as exc:
    +        raise RuntimeError(f"failed to invoke op: {exc}") from exc
    +
    +    if proc.returncode != 0:
    +        err = _scrub(proc.stderr or "")[:200]
    +        if err:
    +            raise RuntimeError(f"op read failed for {reference!r}: {err}")
    +        raise RuntimeError(
    +            f"op read exited {proc.returncode} for {reference!r}"
    +        )
    +
    +    # `op` appends a trailing newline; strip only that so a value with
    +    # intentional internal/edge spaces survives.  But a value that is empty or
    +    # whitespace-only is treated as empty: applying it would silently clobber a
    +    # good .env/shell credential with effectively nothing.
    +    value = (proc.stdout or "").rstrip("\r\n")
    +    if not value.strip():
    +        raise RuntimeError(f"op read returned an empty value for {reference!r}")
    +    return value
    +
    +
    +# ---------------------------------------------------------------------------
    +# Fetch
    +# ---------------------------------------------------------------------------
    +
    +
    +def fetch_onepassword_secrets(
    +    *,
    +    references: Dict[str, str],
    +    account: str = "",
    +    token_env: str = _DEFAULT_TOKEN_ENV,
    +    binary: Optional[Path] = None,
    +    binary_path: str = "",
    +    use_cache: bool = True,
    +    cache_ttl_seconds: float = 300,
    +    home_path: Optional[Path] = None,
    +) -> Tuple[Dict[str, str], List[str]]:
    +    """Resolve ``references`` (name → ``op://…``) to ``(secrets, warnings)``.
    +
    +    Raises :class:`RuntimeError` only when no ``op`` binary is available — a
    +    fatal "can't fetch anything" condition.  Per-reference failures (expired
    +    auth, bad reference, empty value) are collected as warnings and the
    +    reference is dropped, so one bad entry never sinks the rest.
    +
    +    Only a complete, error-free pull is cached, so a transient auth failure
    +    isn't frozen in for the whole TTL window.
    +    """
    +    valid, warnings = _validate_references(references)
    +    if not valid:
    +        return {}, warnings
    +
    +    token_value = [redacted:secret-assignment] "").strip()
    +    cache_key: _CacheKey = (
    +        _auth_fingerprint(token_env),
    +        account or "",
    +        str(home_path) if home_path is not None else "",
    +        _refs_fingerprint(valid),
    +    )
    +
    +    if use_cache:
    +        cached = _CACHE.get(cache_key)
    +        if cached and cached.is_fresh(cache_ttl_seconds):
    +            return dict(cached.secrets), warnings
    +        disk_cached = _DISK_CACHE.read(cache_key, cache_ttl_seconds, home_path)
    +        if disk_cached is not None:
    +            # Promote into L1 so later fetches in this process skip the disk read.
    +            _CACHE[cache_key] = disk_cached
    +            return dict(disk_cached.secrets), warnings
    +
    +    op = binary or find_op(binary_path)
    +    if op is None:
    +        raise RuntimeError(
    +            "op CLI not found.  Install the 1Password CLI "
    +            "(https://developer.1password.com/docs/cli/get-started/) or set "
    +            "secrets.onepassword.binary_path to its absolute location."
    +        )
    +
    +    secrets: [redacted:secret-assignment] str] = {}
    +    read_errors = 0
    +    for name in sorted(valid):
    +        try:
    +            secrets[name] = _run_op_read(
    +                op, valid[name], account=account, token_value=[redacted:secret-assignment]
    +            )
    +        except RuntimeError as exc:
    +            warnings.append(str(exc))
    +            read_errors += 1
    +
    +    if use_cache and not read_errors and secrets:
    +        entry = CachedFetch(secrets=[redacted:secret-assignment] fetched_at=time.time())
    +        _CACHE[cache_key] = entry
    +        _DISK_CACHE.write(cache_key, entry, cache_ttl_seconds, home_path)
    +
    +    return secrets, warnings
    +
    +
    +# ---------------------------------------------------------------------------
    +# Public entry point — called from hermes_cli.env_loader
    +# ---------------------------------------------------------------------------
    +
    +
    +def apply_onepassword_secrets(
    +    *,
    +    enabled: bool,
    +    env: Optional[Dict[str, str]] = None,
    +    account: str = "",
    +    service_account_token_env: str = _DEFAULT_TOKEN_ENV,
    +    binary_path: str = "",
    +    override_existing: bool = True,
    +    cache_ttl_seconds: float = 300,
    +    home_path: Optional[Path] = None,
    +) -> FetchResult:
    +    """Resolve configured ``op://`` references and set them on ``os.environ``.
    +
    +    Called by ``load_hermes_dotenv()`` after the .env files have loaded.
    +    Intentionally defensive — any failure returns a :class:`FetchResult` with
    +    ``error`` set (or surfaces warnings); it never raises.
    +
    +    Parameters mirror the ``secrets.onepassword.*`` config keys so the caller
    +    can splat the dict in.  References that are already satisfied by the
    +    current environment (when ``override_existing`` is false) are skipped
    +    *before* fetching, so ``op`` is never invoked for a value that would be
    +    discarded.
    +    """
    +    result = FetchResult()
    +
    +    if not enabled:
    +        return result
    +
    +    valid, warnings = _validate_references(env)
    +    result.warnings.extend(warnings)
    +
    +    # Skip-before-fetch: never resolve a reference we'd only throw away.
    +    refs_to_fetch: Dict[str, str] = {}
    +    for name, ref in valid.items():
    +        if name == service_account_token_env:
    +            # Never let a resolved secret clobber the very token used to auth.
    +            result.skipped.append(name)
    +            continue
    +        if not override_existing and os.environ.get(name):
    +            result.skipped.append(name)
    +            continue
    +        refs_to_fetch[name] = ref
    +
    +    if not refs_to_fetch:
    +        return result
    +
    +    binary = find_op(binary_path)
    +    result.binary_path = binary
    +    if binary is None:
    +        if binary_path:
    +            result.error = (
    +                f"secrets.onepassword.binary_path ({binary_path!r}) is not an "
    +                "executable op binary."
    +            )
    +        else:
    +            result.error = (
    +                "secrets.onepassword.enabled is true but the op CLI was not "
    +                "found on PATH.  Install it "
    +                "(https://developer.1password.com/docs/cli/get-started/) or set "
    +                "secrets.onepassword.binary_path."
    +            )
    +        return result
    +
    +    try:
    +        secrets, fetch_warnings = fetch_onepassword_secrets(
    +            references=refs_to_fetch,
    +            account=account,
    +            token_env=[redacted:secret-assignment]
    +            binary=binary,
    +            cache_ttl_seconds=cache_ttl_seconds,
    +            home_path=home_path,
    +        )
    +    except RuntimeError as exc:
    +        result.error = str(exc)
    +        return result
    +
    +    result.secrets = secrets
    +    result.warnings.extend(fetch_warnings)
    +
    +    for name, value in secrets.items():
    +        # The token-var and override guards already filtered refs_to_fetch, but
    +        # re-check defensively in case the fetch layer ever returns extras.
    +        if name == service_account_token_env:
    +            if name not in result.skipped:
    +                result.skipped.append(name)
    +            continue
    +        if not override_existing and os.environ.get(name):
    +            if name not in result.skipped:
    +                result.skipped.append(name)
    +            continue
    +        os.environ[name] = value
    +        result.applied.append(name)
    +
    +    return result
    +
    +
    +# ---------------------------------------------------------------------------
    +# SecretSource adapter — the registry-facing wrapper around this module.
    +# ---------------------------------------------------------------------------
    +
    +
    +class OnePasswordSource(SecretSource):
    +    """1Password as a registered secret source.
    +
    +    Thin adapter over the module's fetch machinery.  ``fetch()`` only
    +    *fetches* — precedence, override semantics, conflict warnings, and
    +    the ``os.environ`` writes are the orchestrator's job
    +    (see ``agent.secret_sources.registry.apply_all``).
    +
    +    1Password is a **mapped** source: the user explicitly binds each env
    +    var to an ``op://`` reference under ``secrets.onepassword.env``, so
    +    its claims outrank bulk sources (e.g. a Bitwarden project dump) on
    +    contested vars.
    +    """
    +
    +    name = "onepassword"
    +    label = "1Password"
    +    shape = "mapped"
    +    scheme = "op"
    +
    +    def override_existing(self, cfg: dict) -> bool:
    +        # Default True: an explicit VAR→op:// binding is the strongest
    +        # user intent there is — leaving a stale .env line in place
    +        # should not silently defeat it (same rotation rationale as
    +        # Bitwarden).
    +        return bool(isinstance(cfg, dict) and cfg.get("override_existing", True))
    +
    +    def protected_env_vars(self, cfg: dict):
    +        token_env = [redacted:secret-assignment]
    +        if isinstance(cfg, dict):
    +            token_env = [redacted:secret-assignment]"service_account_token_env") or token_env)
    +        return frozenset({token_env})
    +
    +    def config_schema(self) -> dict:
    +        return {
    +            "enabled": {"description": "Master switch", "default": False},
    +            "env": {
    +                "description": "Map of ENV_VAR -> op://vault/item/field reference",
    +                "default": {},
    +            },
    +            "account": {
    +                "description": "op --account shorthand (empty = default account)",
    +                "default": "",
    +            },
    +            "service_account_token_env": {
    +                "description": "Env var holding the service-account token "
    +                               "(unset = desktop/interactive session)",
    +                "default": _DEFAULT_TOKEN_ENV,
    +            },
    +            "binary_path": {
    +                "description": "Pin the op binary (empty = resolve via PATH)",
    +                "default": "",
    +            },
    +            "cache_ttl_seconds": {
    +                "description": "Disk+memory cache TTL; 0 disables",
    +                "default": 300,
    +            },
    +            "override_existing": {
    +                "description": "Resolved values overwrite .env/shell values",
    +                "default": True,
    +            },
    +        }
    +
    +    def fetch(self, cfg: dict, home_path: Path) -> FetchResult:
    +        cfg = cfg if isinstance(cfg, dict) else {}
    +        result = FetchResult()
    +
    +        env_map = cfg.get("env")
    +        valid, warnings = _validate_references(
    +            env_map if isinstance(env_map, dict) else None
    +        )
    +        result.warnings.extend(warnings)
    +        if not valid:
    +            if not warnings:
    +                result.error = (
    +                    "secrets.onepassword.enabled is true but the env: map is "
    +                    "empty.  Add ENV_VAR: op://vault/item/field entries."
    +                )
    +                result.error_kind = ErrorKind.NOT_CONFIGURED
    +            return result
    +
    +        binary_path = str(cfg.get("binary_path") or "")
    +        binary = find_op(binary_path)
    +        result.binary_path = binary
    +        if binary is None:
    +            if binary_path:
    +                result.error = (
    +                    f"secrets.onepassword.binary_path ({binary_path!r}) is "
    +                    "not an executable op binary."
    +                )
    +            else:
    +                result.error = (
    +                    "secrets.onepassword.enabled is true but the op CLI was "
    +                    "not found on PATH.  Install it "
    +                    "(https://developer.1password.com/docs/cli/get-started/) "
    +                    "or set secrets.onepassword.binary_path."
    +                )
    +            result.error_kind = ErrorKind.BINARY_MISSING
    +            return result
    +
    +        try:
    +            ttl = float(cfg.get("cache_ttl_seconds", 300))
    +        except (TypeError, ValueError):
    +            ttl = 300.0
    +
    +        try:
    +            secrets, fetch_warnings = fetch_onepassword_secrets(
    +                references=valid,
    +                account=str(cfg.get("account") or ""),
    +                token_env=str(
    +                    cfg.get("service_account_token_env") or _DEFAULT_TOKEN_ENV
    +                ),
    +                binary=binary,
    +                cache_ttl_seconds=ttl,
    +                home_path=home_path,
    +            )
    +        except RuntimeError as exc:
    +            result.error = str(exc)
    +            result.error_kind = _classify_op_error(str(exc))
    +            return result
    +
    +        result.secrets = secrets
    +        result.warnings.extend(fetch_warnings)
    +        return result
    +
    +
    +def _classify_op_error(message: str) -> ErrorKind:
    +    """Best-effort mapping of op failure text onto the shared taxonomy."""
    +    lowered = message.lower()
    +    if "timed out" in lowered:
    +        return ErrorKind.TIMEOUT
    +    if "not found on path" in lowered or "not an executable" in lowered \
    +            or "failed to invoke" in lowered:
    +        return ErrorKind.BINARY_MISSING
    +    if any(tok in lowered for tok in ("unauthorized", "not signed in",
    +                                      "session expired", "authentication",
    +                                      "401", "403")):
    +        return ErrorKind.AUTH_FAILED
    +    if "empty value" in lowered:
    +        return ErrorKind.EMPTY_VALUE
    +    if any(tok in lowered for tok in ("network", "connection", "resolve host",
    +                                      "dns")):
    +        return ErrorKind.NETWORK
    +    return ErrorKind.INTERNAL
    +
    +
    +# ---------------------------------------------------------------------------
    +# Test hook — used by hermetic tests to flush the cache between cases.
    +# ---------------------------------------------------------------------------
    +
    +
    +def _reset_cache_for_tests(home_path: Optional[Path] = None) -> None:
    +    """Clear in-process AND disk caches.
    +
    +    Tests can pass ``home_path`` to scope the disk cleanup to a tmpdir.
    +    Without it we fall back to the same default resolution as the writer.
    +    """
    +    _CACHE.clear()
    +    _DISK_CACHE.clear(home_path)
    ```
  - `agent/secret_sources/registry.py`

    ```diff
    @@ -0,0 +1,370 @@
    +"""Secret-source registry + apply orchestrator.
    +
    +This module owns everything that must be uniform across secret backends
    +so no individual source can get it wrong:
    +
    +* registration (name/scheme uniqueness, API-version gating)
    +* per-source wall-clock timeout enforcement around ``fetch()``
    +* precedence: mapped sources beat bulk sources; within a shape,
    +  ``secrets.sources`` order (or registration order) decides; first
    +  claim wins — later sources never silently clobber an earlier one
    +* ``override_existing`` semantics (may beat .env/shell, never another
    +  secret source, never a protected var)
    +* cross-source conflict warnings (shadowed claims are always surfaced)
    +* provenance: which source supplied every applied var
    +
    +The single entry point for startup is :func:`apply_all`, called from
    +``hermes_cli.env_loader._apply_external_secret_sources()``.
    +
    +Plugins register additional sources via
    +``PluginContext.register_secret_source()`` which lands in
    +:func:`register_source`.  In-tree sources are registered lazily by
    +:func:`_ensure_builtin_sources` — the set of bundled sources is
    +deliberately closed (Bitwarden, and 1Password once it lands); new
    +third-party backends ship as standalone plugin repos implementing
    +:class:`agent.secret_sources.base.SecretSource`.
    +"""
    +
    +from __future__ import annotations
    +
    +import concurrent.futures
    +import logging
    +from dataclasses import dataclass, field
    +from pathlib import Path
    +from typing import Dict, List, Optional
    +
    +from agent.secret_sources.base import (
    +    SECRET_SOURCE_API_VERSION,
    +    ErrorKind,
    +    FetchResult,
    +    SecretSource,
    +    is_valid_env_name,
    +)
    +
    +logger = logging.getLogger(__name__)
    +
    +# Ordered registry: name → source instance.  Python dicts preserve
    +# insertion order, which doubles as the default apply order.
    +_SOURCES: Dict[str, SecretSource] = {}
    +_BUILTINS_LOADED = False
    +
    +
    +@dataclass
    +class AppliedVar:
    +    """Provenance record for one env var the orchestrator set."""
    +
    +    name: str
    +    source: str          # SecretSource.name
    +    shape: str           # "mapped" | "bulk"
    +    overrode_env: bool   # replaced a pre-existing .env/shell value
    +
    +
    +@dataclass
    +class SourceReport:
    +    """One source's outcome within an :class:`ApplyReport`."""
    +
    +    name: str
    +    label: str
    +    result: FetchResult
    +    applied: List[str] = field(default_factory=list)
    +    skipped_existing: List[str] = field(default_factory=list)   # .env/shell won
    +    skipped_claimed: List[str] = field(default_factory=list)    # earlier source won
    +    skipped_protected: List[str] = field(default_factory=list)  # bootstrap-auth guard
    +    skipped_invalid: List[str] = field(default_factory=list)    # bad env-var name
    +
    +
    +@dataclass
    +class ApplyReport:
    +    """Merged outcome of one orchestrated apply pass."""
    +
    +    sources: List[SourceReport] = field(default_factory=list)
    +    provenance: Dict[str, AppliedVar] = field(default_factory=dict)
    +    conflicts: List[str] = field(default_factory=list)  # human-readable warnings
    +
    +    @property
    +    def applied_any(self) -> bool:
    +        return bool(self.provenance)
    +
    +
    +# ---------------------------------------------------------------------------
    +# Registration
    +# ---------------------------------------------------------------------------
    +
    +
    +def register_source(source: SecretSource, *, replace: bool = False) -> bool:
    +    """Register a secret source.  Returns True on success.
    +
    +    Rejections are logged, never raised — a bad plugin must not take
    +    down startup.  ``replace`` allows tests / user plugins to override
    +    a bundled source of the same name (last-writer-wins like model
    +    providers), but scheme collisions across *different* names are
    +    always rejected.
    +    """
    +    if not isinstance(source, SecretSource):
    +        logger.warning(
    +            "Ignoring secret source %r: does not inherit from SecretSource",
    +            source,
    +        )
    +        return False
    +    name = getattr(source, "name", "") or ""
    +    if not name or not name.replace("_", "").isalnum() or name != name.lower():
    +        logger.warning("Ignoring secret source with invalid name %r", name)
    +        return False
    +    if getattr(source, "api_version", None) != SECRET_SOURCE_API_VERSION:
    +        logger.warning(
    +            "Ignoring secret source '%s': built against secret-source API v%s, "
    +            "this Hermes speaks v%s",
    +            name, getattr(source, "api_version", "?"), SECRET_SOURCE_API_VERSION,
    +        )
    +        return False
    +    if getattr(source, "shape", None) not in ("mapped", "bulk"):
    +        logger.warning(
    +            "Ignoring secret source '%s': shape must be 'mapped' or 'bulk', got %r",
    +            name, getattr(source, "shape", None),
    +        )
    +        return False
    +    if name in _SOURCES and not replace:
    +        logger.warning("Secret source '%s' already registered; ignoring duplicate", name)
    +        return False
    +    scheme = getattr(source, "scheme", None)
    +    if scheme:
    +        for other_name, other in _SOURCES.items():
    +            if other_name != name and getattr(other, "scheme", None) == scheme:
    +                logger.warning(
    +                    "Ignoring secret source '%s': scheme '%s://' is already "
    +                    "owned by source '%s'",
    +                    name, scheme, other_name,
    +                )
    +                return False
    +    _SOURCES[name] = source
    +    return True
    +
    +
    +def get_source(name: str) -> Optional[SecretSource]:
    +    _ensure_builtin_sources()
    +    return _SOURCES.get(name)
    +
    +
    +def list_sources() -> List[SecretSource]:
    +    _ensure_builtin_sources()
    +    return list(_SOURCES.values())
    +
    +
    +def _ensure_builtin_sources() -> None:
    +    """Idempotently register the bundled sources.
    +
    +    Lazy so importing this module stays cheap and so a broken bundled
    +    source can never break registration of the others.
    +    """
    +    global _BUILTINS_LOADED
    +    if _BUILTINS_LOADED:
    +        return
    +    _BUILTINS_LOADED = True
    +    try:
    +        from agent.secret_sources.bitwarden import BitwardenSource
    +
    +        register_source(BitwardenSource())
    +    except Exception:  # noqa: BLE001 — never block startup
    +        logger.warning("Failed to register bundled Bitwarden secret source",
    +                       exc_info=True)
    +    try:
    +        from agent.secret_sources.onepassword import OnePasswordSource
    +
    +        register_source(OnePasswordSource())
    +    except Exception:  # noqa: BLE001 — never block startup
    +        logger.warning("Failed to register bundled 1Password secret source",
    +                       exc_info=True)
    +
    +
    +def _reset_registry_for_tests() -> None:
    +    global _BUILTINS_LOADED
    +    _SOURCES.clear()
    +    _BUILTINS_LOADED = False
    +
    +
    +# ---------------------------------------------------------------------------
    +# Orchestrated apply
    +# ---------------------------------------------------------------------------
    +
    +
    +def _fetch_with_timeout(
    +    source: SecretSource, cfg: dict, home_path: Path
    +) -> FetchResult:
    +    """Run source.fetch() under a wall-clock budget; never raises.
    +
    +    The budget is enforced with a daemon worker thread: a source that
    +    blows its budget is reported as ``TIMEOUT`` and its (eventual)
    +    result is discarded.  The thread itself may linger until process
    +    exit — acceptable for a startup-only path, and strictly better than
    +    an unbounded hang on every ``hermes`` invocation.
    +    """
    +    timeout = source.fetch_timeout_seconds(cfg)
    +    executor = concurrent.futures.ThreadPoolExecutor(
    +        max_workers=1, thread_name_prefix=f"secret-src-{source.name}"
    +    )
    +    try:
    +        future = executor.submit(source.fetch, cfg, home_path)
    +        try:
    +            result = future.result(timeout=timeout)
    +        except concurrent.futures.TimeoutError:
    +            future.cancel()
    +            res = FetchResult()
    +            res.error = (
    +                f"fetch exceeded {timeout:.0f}s budget — startup continued "
    +                "without this source (raise secrets."
    +                f"{source.name}.timeout_seconds if the backend is just slow)"
    +            )
    +            res.error_kind = ErrorKind.TIMEOUT
    +            return res
    +        except Exception as exc:  # noqa: BLE001 — contract violation, contain it
    +            res = FetchResult()
    +            res.error = f"fetch raised {type(exc).__name__}: {exc}"
    +            res.error_kind = ErrorKind.INTERNAL
    +            return res
    +    finally:
    +        executor.shutdown(wait=False)
    +
    +    if not isinstance(result, FetchResult):
    +        res = FetchResult()
    +        res.error = (
    +            f"fetch returned {type(result).__name__} instead of FetchResult"
    +        )
    +        res.error_kind = ErrorKind.INTERNAL
    +        return res
    +    return result
    +
    +
    +def _ordered_enabled_sources(secrets_cfg: dict) -> List[SecretSource]:
    +    """Resolve which sources run, in which order.
    +
    +    Order: the optional ``secrets.sources`` list wins; sources not named
    +    there follow in registration order.  Enabled = the source's own
    +    ``is_enabled`` says so for its config section.  Mapped-vs-bulk
    +    precedence is applied on top of this order by :func:`apply_all`.
    +    """
    +    _ensure_builtin_sources()
    +
    +    explicit = secrets_cfg.get("sources")
    +    order: List[str] = []
    +    if isinstance(explicit, list):
    +        for entry in explicit:
    +            if isinstance(entry, str) and entry in _SOURCES and entry not in order:
    +                order.append(entry)
    +        unknown = [e for e in explicit
    +                   if isinstance(e, str) and e not in _SOURCES]
    +        if unknown:
    +            logger.warning(
    +                "secrets.sources names unknown source(s): %s (known: %s)",
    +                ", ".join(unknown), ", ".join(_SOURCES) or "none",
    +            )
    +    for name in _SOURCES:
    +        if name not in order:
    +            order.append(name)
    +
    +    enabled: List[SecretSource] = []
    +    for name in order:
    +        source = _SOURCES[name]
    +        cfg = secrets_cfg.get(name)
    +        cfg = cfg if isinstance(cfg, dict) else {}
    +        try:
    +            if source.is_enabled(cfg):
    +                enabled.append(source)
    +        except Exception:  # noqa: BLE001
    +            logger.warning("Secret source '%s' is_enabled() raised; skipping",
    +                           name, exc_info=True)
    +    return enabled
    +
    +
    +def apply_all(secrets_cfg: dict, home_path: Path,
    +              environ: Optional[Dict[str, str]] = None) -> ApplyReport:
    +    """Fetch from every enabled source and apply the merged result to env.
    +
    +    ``environ`` defaults to ``os.environ``; injectable for tests.
    +
    +    Precedence per env var (most-specific intent wins):
    +
    +    1. Pre-existing env (.env / shell) — unless the winning source has
    +       ``override_existing: true``.
    +    2. Mapped sources, in configured order.
    +    3. Bulk sources, in configured order.
    +
    +    First claim wins.  A later source that also carries the var gets a
    +    ``skipped_claimed`` entry and a conflict warning — never a silent
    +    clobber, and ``override_existing`` never applies across sources.
    +    """
    +    import os as _os
    +
    +    env = environ if environ is not None else _os.environ
    +    report = ApplyReport()
    +
    +    secrets_cfg = [redacted:secret-assignment] if isinstance(secrets_cfg, dict) else {}
    +    enabled = _ordered_enabled_sources(secrets_cfg)
    +    if not enabled:
    +        return report
    +
    +    # Mapped sources outrank bulk sources regardless of list order:
    +    # an explicit VAR→ref binding is stronger intent than a project dump.
    +    ordered = ([s for s in enabled if s.shape == "mapped"]
    +               + [s for s in enabled if s.shape == "bulk"])
    +
    +    # Fetch phase.
    +    fetches: List[tuple[SecretSource, dict, FetchResult]] = []
    +    protected: Dict[str, str] = {}  # var → source that protects it
    +    for source in ordered:
    +        cfg = secrets_cfg.get(source.name)
    +        cfg = cfg if isinstance(cfg, dict) else {}
    +        result = _fetch_with_timeout(source, cfg, home_path)
    +        fetches.append((source, cfg, result))
    +        try:
    +            for var in source.protected_env_vars(cfg):
    +                protected.setdefault(var, source.name)
    +        except Exception:  # noqa: BLE001
    +            pass
    +
    +    # Apply phase — sequential, first-wins, fully attributed.
    +    claimed: Dict[str, str] = {}  # var → source name that won it
    +    for source, cfg, result in fetches:
    +        sr = SourceReport(name=source.name,
    +                          label=source.label or source.name,
    +                          result=result)
    +        report.sources.append(sr)
    +        if not result.ok:
    +            continue
    +
    +        try:
    +            override = source.override_existing(cfg)
    +        except Exception:  # noqa: BLE001
    +            override = False
    +
    +        for var, value in result.secrets.items():
    +            if not isinstance(var, str) or not isinstance(value, str):
    +                continue
    +            if not is_valid_env_name(var):
    +                sr.skipped_invalid.append(var)
    +                continue
    +            if var in protected:
    +                sr.skipped_protected.append(var)
    +                continue
    +            if var in claimed:
    +                sr.skipped_claimed.append(var)
    +                report.conflicts.append(
    +                    f"{var}: kept value from {claimed[var]}; "
    +                    f"{source.name} also supplies it (first source wins — "
    +                    "remove one binding or reorder secrets.sources)"
    +                )
    +                continue
    +            existed = bool(env.get(var))
    +            if existed and not override:
    +                sr.skipped_existing.append(var)
    +                continue
    +            env[var] = value
    +            claimed[var] = source.name
    +            sr.applied.append(var)
    +            report.provenance[var] = AppliedVar(
    +                name=var,
    +                source=source.name,
    +                shape=source.shape,
    +                overrode_env=existed,
    +            )
    +
    +    return report
    ```
  - `cli-config.yaml.example`

    ```diff
    @@ -1380,3 +1380,50 @@ updates:
     #       # This is a CREDENTIAL: prefer setting HERMES_DASHBOARD_OIDC_CLIENT_SECRET
     #       # in ~/.hermes/.env over putting it here in config.yaml.
     #       #   client_secret: ""
    +
    +# =============================================================================
    +# External secret sources
    +# =============================================================================
    +# Pull provider credentials from external secret managers at process startup
    +# instead of storing them in ~/.hermes/.env. Only the manager's bootstrap
    +# credential (e.g. BWS_ACCESS_TOKEN / OP_SERVICE_ACCOUNT_TOKEN) lives in .env
    +# (or your shell / desktop session); everything else rotates centrally.
    +# Failures never block startup — Hermes warns once and continues with
    +# whatever .env already had.
    +#
    +# Multiple sources can be enabled at once:
    +#   - "mapped" sources (explicit VAR -> ref bindings, e.g. 1Password's env:
    +#     map) beat "bulk" sources (whole-project dumps like Bitwarden BSM)
    +#   - within a shape, the first source to claim a var wins; later claims
    +#     are skipped with a startup warning (never a silent clobber)
    +#   - a source's override_existing lets it beat .env/shell values, but
    +#     never another secret source's claim
    +# Docs: https://hermes-agent.nousresearch.com/docs/user-guide/secrets/
    +#
    +# secrets:
    +#   # Optional explicit ordering of enabled sources.
    +#   # sources: [onepassword, bitwarden]
    +#
    +#   # ---- Bitwarden Secrets Manager (bws CLI) --------------------------------
    +#   bitwarden:
    +#     enabled: false
    +#     access_token_env: [redacted:secret-assignment]   # bootstrap token, sourced from .env
    +#     project_id: ""                       # UUID of the BSM project to sync
    +#     server_url: ""                       # "" = US Cloud; EU/self-hosted URL otherwise
    +#     cache_ttl_seconds: 300               # 0 disables caching
    +#     override_existing: true              # BSM values win over existing env
    +#     auto_install: true                   # lazy-download bws into ~/.hermes/bin
    +#
    +#   # ---- 1Password (op CLI) -------------------------------------------------
    +#   onepassword:
    +#     enabled: false
    +#     # Map env-var names to op:// secret references. Each is resolved with a
    +#     # single `op read` at startup.
    +#     env:
    +#       OPENAI_API_KEY: "[redacted:secret-assignment] key"
    +#       ANTHROPIC_API_KEY: "[redacted:secret-assignment]"
    +#     account: ""                                   # op --account shorthand; "" = default
    +#     service_account_token_env: [redacted:secret-assignment]  # headless auth; unset = desktop session
    +#     binary_path: ""                               # "" = resolve op via PATH; else absolute path
    +#     cache_ttl_seconds: 300                        # 0 disables BOTH cache layers
    +#     override_existing: true                       # resolved values win over existing env
    ```
  - `hermes_cli/config.py`

    ```diff
    @@ -3073,6 +3073,15 @@ def _ensure_hermes_home_managed(home: Path):
         # Pull credentials from external secret managers at process startup
         # rather than storing them in ~/.hermes/.env.
         "secrets": {
    +        # Optional explicit ordering of enabled secret sources.  When
    +        # omitted, sources run in registration order (bundled first,
    +        # then plugin-registered).  Regardless of this list, "mapped"
    +        # sources (explicit VAR→ref bindings, e.g. a future 1Password
    +        # env: map) always take precedence over "bulk" sources
    +        # (project dumps like Bitwarden BSM), and the first source to
    +        # claim a var wins — later claims are skipped with a warning.
    +        # Example: sources: [onepassword, bitwarden]
    +        # "sources": [],
             "bitwarden": {
                 # Master switch.  When false, BSM is never contacted and the
                 # bws binary is never auto-installed — same as not having
    ```

    ```diff
    @@ -3104,6 +3113,34 @@ def _ensure_hermes_home_managed(home: Path):
                 # `hermes secrets bitwarden setup`.
                 "server_url": "",
             },
    +        "onepassword": {
    +            # Master switch.  When false, the op CLI is never invoked —
    +            # same as not having this section at all.
    +            "enabled": False,
    +            # Mapping of env-var name → 1Password secret reference
    +            # (op://vault/item/field).  Each entry is resolved with a
    +            # single `op read` at startup.
    +            "env": {},
    +            # Optional account shorthand / sign-in address passed as
    +            # `op read --account <account>`.  Empty = op's default account.
    +            "account": "",
    +            # Name of the env var holding a 1Password service-account token
    +            # for headless auth.  Sourced from ~/.hermes/.env (or the shell)
    +            # and exported to the op child as OP_SERVICE_ACCOUNT_TOKEN.
    +            # Leave the var unset to use an interactive/desktop op session.
    +            "service_account_token_env": "OP_SERVICE_ACCOUNT_TOKEN",
    +            # Optional absolute path to the op binary.  When set it is used
    +            # verbatim (PATH is not consulted) — pin this to avoid trusting
    +            # whatever `op` appears first on PATH.  Empty = resolve via PATH.
    +            "binary_path": "",
    +            # Seconds to cache resolved values in-process and on disk.  0
    +            # disables BOTH cache layers (no values are written to disk).
    +            "cache_ttl_seconds": 300,
    +            # When True (default), resolved values overwrite existing env
    +            # vars so rotating a secret in 1Password takes effect on next
    +            # start.  Flip to false to let .env / shell exports win locally.
    +            "override_existing": True,
    +        },
         },
     
         # Paste collapse thresholds (TUI + CLI).
    ```
  - `hermes_cli/env_loader.py`

    ```diff
    @@ -78,9 +78,17 @@ def format_secret_source_suffix(env_var: str) -> str:
             return ""
         if source == "bitwarden":
             return " (from Bitwarden)"
    -    # Generic fallback — future-proofing for additional secret sources
    -    # (e.g. 1Password, HashiCorp Vault) without having to update every
    -    # call site.
    +    # Ask the registry for the source's human label (e.g. "1Password").
    +    # Fall back to the raw source name for labels the registry doesn't
    +    # know (stale provenance from an uninstalled plugin, tests).
    +    try:
    +        from agent.secret_sources.registry import get_source
    +
    +        registered = get_source(source)
    +        if registered is not None and registered.label:
    +            return f" (from {registered.label})"
    +    except Exception:  # noqa: BLE001 — label lookup must never raise
    +        pass
         return f" (from {source})"
     
     
    ```

    ```diff
    @@ -238,6 +246,20 @@ def load_hermes_dotenv(
             _load_dotenv_with_fallback(user_env, override=True)
             loaded.append(user_env)
     
    +    # Load .op.env AFTER .env so that .env values win, but the bootstrap
    +    # token (OP_SERVICE_ACCOUNT_TOKEN) becomes available for
    +    # apply_onepassword_secrets() even in cron / subprocess environments
    +    # that inherit no shell state (no systemd EnvironmentFile, no op run).
    +    # .op.env is gitignored — the service-account token never enters the
    +    # committed .env file.
    +    # Users on systemd can alternatively use:
    +    #   EnvironmentFile=-/path/to/.hermes/.op.env
    +    # in their gateway unit, which takes precedence (override=False below
    +    # ensures .op.env never clobbers a token already in the environment).
    +    op_env = home_path / ".op.env"
    +    if op_env.exists() and not os.environ.get("OP_SERVICE_ACCOUNT_TOKEN"):
    +        _load_dotenv_with_fallback(op_env, override=False)
    +
         if project_env_path and project_env_path.exists():
             _load_dotenv_with_fallback(project_env_path, override=not loaded)
             loaded.append(project_env_path)
    ```

    ```diff
    @@ -281,21 +303,27 @@ def _apply_managed_env() -> None:
     
     
     def _apply_external_secret_sources(home_path: Path) -> None:
    -    """Pull secrets from external sources (currently Bitwarden) into env.
    +    """Pull secrets from every enabled external source into env.
     
    -    Runs AFTER dotenv loads so .env values are visible (we use them to
    -    locate the access token) but BEFORE the rest of Hermes reads
    +    Runs AFTER dotenv loads so .env values are visible (sources use them
    +    to locate bootstrap tokens) but BEFORE the rest of Hermes reads
         ``os.environ`` for credentials.  Any failure here is logged and
         swallowed — external secret sources must never block startup.
     
    +    The heavy lifting (source ordering, mapped-beats-bulk precedence,
    +    first-claim-wins conflict handling, override semantics, provenance)
    +    lives in ``agent.secret_sources.registry.apply_all``; this wrapper
    +    owns the once-per-HERMES_HOME guard, the post-apply ASCII
    +    sanitization sweep, the ``_SECRET_SOURCES`` provenance map that
    +    UI surfaces read, and the startup status lines.
    +
         Idempotent within a process: subsequent calls for the same
         ``home_path`` are no-ops.  ``load_hermes_dotenv()`` runs at import
         time from several hot modules (cli.py, hermes_cli/main.py,
         run_agent.py, trajectory_compressor.py, ...), so without this guard
    -    the Bitwarden status line would print 3-5x per CLI startup.  Use
    +    the status lines would print 3-5x per CLI startup.  Use
         ``reset_secret_source_cache()`` if you need to force a re-pull
    -    (tests, future ``hermes secrets bitwarden sync`` from a long-running
    -    process).
    +    (tests, long-running processes after a config change).
         """
         home_key = str(Path(home_path).resolve())
         if home_key in _APPLIED_HOMES:
    ```

    ```diff
    @@ -306,54 +334,45 @@ def _apply_external_secret_sources(home_path: Path) -> None:
             cfg = _load_secrets_config(home_path)
         except Exception:  # noqa: BLE001 — config errors must not block startup
             return
    -
    -    bw_cfg = (cfg or {}).get("bitwarden") or {}
    -    if not bw_cfg.get("enabled"):
    +    if not cfg:
             return
     
         try:
    -        from agent.secret_sources.bitwarden import apply_bitwarden_secrets
    +        from agent.secret_sources.registry import apply_all
         except ImportError:
             return
     
    -    result = apply_bitwarden_secrets(
    -        enabled=True,
    -        access_token_env=[redacted:secret-assignment]"access_token_env", "BWS_ACCESS_TOKEN"),
    -        project_id=bw_cfg.get("project_id", ""),
    -        override_existing=bool(bw_cfg.get("override_existing", False)),
    -        cache_ttl_seconds=float(bw_cfg.get("cache_ttl_seconds", 300)),
    -        auto_install=bool(bw_cfg.get("auto_install", True)),
    -        server_url=str(bw_cfg.get("server_url", "") or "").strip(),
    -        home_path=home_path,
    -    )
    -
    -    if result.applied:
    -        # Re-run the ASCII sanitization pass: BSM values are user-supplied
    -        # and might have the same copy-paste corruption as a manually
    -        # edited .env (see #6843).
    +    try:
    +        report = apply_all(cfg, home_path)
    +    except Exception:  # noqa: BLE001 — belt-and-braces; apply_all shouldn't raise
    +        return
    +
    +    if report.applied_any:
    +        # Re-run the ASCII sanitization pass: vault values are
    +        # user-supplied and might have the same copy-paste corruption as
    +        # a manually edited .env (see #6843).
             _sanitize_loaded_credentials()
    -        # Remember where these came from so the setup / `hermes model`
    -        # flows can label detected credentials with "(from Bitwarden)" —
    -        # otherwise users see "credentials ✓" with no hint that the value
    -        # came from BSM rather than .env.
    -        for name in result.applied:
    -            _SECRET_SOURCES[name] = "bitwarden"
    -        print(
    -            f"  Bitwarden Secrets Manager: applied {len(result.applied)} "
    -            f"secret{'s' if len(result.applied) != 1 else ''} "
    -            f"({', '.join(sorted(result.applied))})",
    -            file=sys.stderr,
    -        )
    -    if result.error:
    -        print(
    -            f"  Bitwarden Secrets Manager: {result.error}",
    -            file=sys.stderr,
    -        )
    -    for warn in result.warnings:
    -        print(
    -            f"  Bitwarden Secrets Manager: {warn}",
    -            file=sys.stderr,
    -        )
    +        # Remember where each var came from so setup / `hermes model`
    +        # flows can label detected credentials with "(from Bitwarden)" /
    +        # "(from 1Password)" — otherwise users see "credentials ✓" with
    +        # no hint the value came from a vault rather than .env.
    +        for name, applied in report.provenance.items():
    +            _SECRET_SOURCES[name] = applied.source
    +
    +    for src in report.sources:
    +        if src.applied:
    +            print(
    +                f"  {src.label}: applied {len(src.applied)} "
    +                f"secret{'s' if len(src.applied) != 1 else ''} "
    +                f"({', '.join(sorted(src.applied))})",
    +                file=sys.stderr,
    +            )
    +        if src.result.error:
    +            print(f"  {src.label}: {src.result.error}", file=sys.stderr)
    +        for warn in src.result.warnings:
    +            print(f"  {src.label}: {warn}", file=sys.stderr)
    +    for conflict in report.conflicts:
    +        print(f"  Secret sources: {conflict}", file=sys.stderr)
     
     
     def _load_secrets_config(home_path: Path) -> dict:
    ```
  - `hermes_cli/main.py`

    ```diff
    @@ -12787,16 +12787,16 @@ def main():
         fallback_parser.set_defaults(func=cmd_fallback)
     
         # =========================================================================
    -    # secrets command — external secret managers (currently: Bitwarden)
    +    # secrets command — external secret managers (Bitwarden, 1Password)
         # =========================================================================
         secrets_parser = [redacted:secret-assignment]
             "secrets",
    -        help="Manage external secret sources (Bitwarden Secrets Manager)",
    +        help="Manage external secret sources (Bitwarden, 1Password)",
             description=(
                 "Pull API keys from an external secret manager at process startup "
    -            "instead of storing them in ~/.hermes/.env.  Currently supports "
    -            "Bitwarden Secrets Manager.  See: "
    -            "https://hermes-agent.nousresearch.com/docs/user-guide/secrets/bitwarden"
    +            "instead of storing them in ~/.hermes/.env.  Supports Bitwarden "
    +            "Secrets Manager and 1Password.  See: "
    +            "https://hermes-agent.nousresearch.com/docs/user-guide/secrets/"
             ),
         )
         secrets_subparsers = [redacted:secret-assignment]"secrets_command")
    ```

    ```diff
    @@ -12807,16 +12807,27 @@ def main():
             help="Bitwarden Secrets Manager integration",
         )
     
    +    secrets_op = [redacted:secret-assignment]
    +        "onepassword",
    +        aliases=["op", "1password"],
    +        help="1Password (op:// references) integration",
    +    )
    +
         # Lazy import — only pays for itself when this subcommand is actually used.
         from hermes_cli import secrets_cli as _secrets_cli
    +    from hermes_cli import onepassword_secrets_cli as _op_secrets_cli
     
         _secrets_cli.register_cli(secrets_bw)
    +    _op_secrets_cli.register_cli(secrets_op)
     
         def _dispatch_secrets(args):  # noqa: ANN001
             sub = getattr(args, "secrets_command", None)
             bw_sub = getattr(args, "secrets_bw_command", None)
    +        op_sub = getattr(args, "secrets_op_command", None)
             if sub in ("bitwarden", "bw") and bw_sub is not None:
                 return args.func(args)
    +        if sub in ("onepassword", "op", "1password") and op_sub is not None:
    +            return args.func(args)
             secrets_parser.print_help()
             return 0
     
    ```
  - `hermes_cli/onepassword_secrets_cli.py`

    ```diff
    @@ -0,0 +1,432 @@
    +"""CLI handlers for ``hermes secrets onepassword ...``.
    +
    +Subcommands:
    +    setup    — verify the op CLI, set account / token env var, enable
    +    status   — show config + op binary + auth + configured references
    +    set      — map an env var to an ``op://…`` reference
    +    remove   — drop a mapping
    +    sync     — resolve references now and show what would be applied (dry-run)
    +    disable  — flip ``secrets.onepassword.enabled`` to False
    +
    +Unlike Bitwarden, the ``op`` binary is NOT auto-installed: 1Password publishes
    +the CLI through OS package managers and signed installers, so Hermes expects
    +an already-installed, already-authenticated ``op`` and never downloads one.
    +"""
    +
    +from __future__ import annotations
    +
    +import argparse
    +import os
    +import subprocess
    +from pathlib import Path
    +from typing import Optional
    +
    +from rich.console import Console
    +from rich.panel import Panel
    +from rich.table import Table
    +
    +from agent.secret_sources import onepassword as op_src
    +from hermes_cli.config import (
    +    get_env_path,
    +    load_config,
    +    save_config,
    +    save_env_value,
    +)
    +
    +_DEFAULT_TOKEN_ENV = "[redacted:secret-assignment]"
    +_DOCS_URL = "https://developer.1password.com/docs/cli/get-started/"
    +
    +
    +# ---------------------------------------------------------------------------
    +# Argparse wiring — called from hermes_cli.main
    +# ---------------------------------------------------------------------------
    +
    +
    +def register_cli(parent_parser: argparse.ArgumentParser) -> None:
    +    """Attach the ``onepassword`` subcommand tree to a parent parser."""
    +    sub = parent_parser.add_subparsers(dest="secrets_op_command")
    +
    +    setup = sub.add_parser(
    +        "setup",
    +        help="Verify the op CLI, set account / token env var, and enable",
    +    )
    +    setup.add_argument(
    +        "--account",
    +        help="1Password account shorthand or sign-in address (op --account)",
    +    )
    +    setup.add_argument(
    +        "--token-env",
    +        help=f"Env var holding a service-account token (default {_DEFAULT_TOKEN_ENV})",
    +    )
    +    setup.add_argument(
    +        "--token",
    +        help="Service-account token to store in .env non-interactively",
    +    )
    +    setup.add_argument(
    +        "--binary-path",
    +        help="Absolute path to the op binary (skips PATH lookup)",
    +    )
    +    setup.set_defaults(func=cmd_setup)
    +
    +    status = sub.add_parser("status", help="Show config + op binary + references")
    +    status.set_defaults(func=cmd_status)
    +
    +    set_p = sub.add_parser("set", help="Map an env var to an op:// reference")
    +    set_p.add_argument("env_var", help="Environment variable name, e.g. OPENAI_API_KEY")
    +    set_p.add_argument("reference", help="1Password reference, e.g. op://Private/OpenAI/api key")
    +    set_p.set_defaults(func=cmd_set)
    +
    +    remove = sub.add_parser("remove", help="Remove an env-var → reference mapping")
    +    remove.add_argument("env_var", help="Environment variable name to unmap")
    +    remove.set_defaults(func=cmd_remove)
    +
    +    sync = sub.add_parser("sync", help="Resolve references now and report what changed")
    +    sync.add_argument(
    +        "--apply",
    +        action="store_true",
    +        help="Actually export resolved values into the current shell (default: dry-run)",
    +    )
    +    sync.set_defaults(func=cmd_sync)
    +
    +    disable = sub.add_parser("disable", help="Turn off the 1Password integration")
    +    disable.set_defaults(func=cmd_disable)
    +
    +
    +# ---------------------------------------------------------------------------
    +# Handlers
    +# ---------------------------------------------------------------------------
    +
    +
    +def cmd_setup(args: argparse.Namespace) -> int:
    +    console = Console()
    +    console.print(
    +        Panel.fit(
    +            "[bold]1Password secret source setup[/bold]\n\n"
    +            "Hermes resolves [cyan]op://vault/item/field[/cyan] references through your\n"
    +            "already-installed, already-authenticated 1Password CLI (`op`).\n\n"
    +            f"Don't have it yet? Install + sign in: [cyan]{_DOCS_URL}[/cyan]",
    +            border_style="cyan",
    +        )
    +    )
    +
    +    cfg = load_config()
    +    op_cfg = cfg.setdefault("secrets", {}).setdefault("onepassword", {})
    +
    +    # ------------------------------------------------------------------ binary
    +    console.print()
    +    console.print("[bold]Step 1[/bold]  Locate the op CLI")
    +    binary_path = (args.binary_path or op_cfg.get("binary_path", "") or "").strip()
    +    binary = op_src.find_op(binary_path)
    +    if binary is None:
    +        if binary_path:
    +            console.print(f"  [red]✗ {binary_path} is not an executable op binary.[/red]")
    +        else:
    +            console.print("  [red]✗ op not found on PATH.[/red]")
    +        console.print(f"  Install the 1Password CLI: {_DOCS_URL}")
    +        return 1
    +    console.print(f"  [green]✓[/green] {binary}  ({_op_version(binary)})")
    +    if binary_path:
    +        op_cfg["binary_path"] = binary_path
    +
    +    # ----------------------------------------------------------------- account
    +    if args.account and args.account.strip():
    +        op_cfg["account"] = args.account.strip()
    +        console.print(f"  Account: [cyan]{op_cfg['account']}[/cyan]")
    +
    +    # ------------------------------------------------------------------- token
    +    console.print()
    +    console.print("[bold]Step 2[/bold]  Authentication")
    +    token_env = [redacted:secret-assignment] or op_cfg.get("service_account_token_env")
    +                 or _DEFAULT_TOKEN_ENV).strip()
    +    op_cfg["service_account_token_env"] = token_env
    +
    +    token = [redacted:secret-assignment] or "").strip()
    +    if token:
    +        save_env_value(token_env, token)
    +        os.environ[token_env] = token
    +        console.print(f"  [green]✓[/green] service-account token stored in "
    +                      f"{get_env_path()} as {token_env}")
    +    elif os.environ.get(token_env):
    +        console.print(f"  [green]✓[/green] using service-account token from {token_env}")
    +    else:
    +        who = _op_whoami(binary, op_cfg.get("account", ""))
    +        if who:
    +            console.print(f"  [green]✓[/green] using existing op session ({who})")
    +        else:
    +            console.print(
    +                "  [yellow]No service-account token and no active op session "
    +                "detected.[/yellow]\n"
    +                "  Either run [cyan]op signin[/cyan] (desktop/interactive) or set a "
    +                f"service-account token in {token_env}, then re-run status."
    +            )
    +
    +    # ----------------------------------------------------------------- enable
    +    op_cfg["enabled"] = True
    +    op_cfg.setdefault("env", {})
    +    op_cfg.setdefault("cache_ttl_seconds", 300)
    +    op_cfg.setdefault("override_existing", True)
    +    save_config(cfg)
    +
    +    console.print()
    +    console.print("[green]✓ 1Password secret source is enabled.[/green]")
    +    console.print(
    +        "  Map credentials:  [redacted:secret-assignment] secrets onepassword set OPENAI_API_KEY "
    +        "\"op://Private/OpenAI/api key\"[/cyan]\n"
    +        "  Preview:          [cyan]hermes secrets onepassword sync[/cyan]\n"
    +        "  Status:           [cyan]hermes secrets onepassword status[/cyan]"
    +    )
    +    return 0
    +
    +
    +def cmd_status(args: argparse.Namespace) -> int:
    +    console = Console()
    +    cfg = load_config()
    +    op_cfg = (cfg.get("secrets") or {}).get("onepassword") or {}
    +
    +    enabled = bool(op_cfg.get("enabled"))
    +    account = str(op_cfg.get("account", "") or "").strip()
    +    token_env = [redacted:secret-assignment]"service_account_token_env", _DEFAULT_TOKEN_ENV)
    +    binary_path = str(op_cfg.get("binary_path", "") or "").strip()
    +    references = op_cfg.get("env") if isinstance(op_cfg.get("env"), dict) else {}
    +    token_set = [redacted:secret-assignment]
    +
    +    binary = op_src.find_op(binary_path)
    +
    +    table = Table(show_header=False, box=None, padding=(0, 2))
    +    table.add_column("", style="bold")
    +    table.add_column("")
    +    table.add_row("Enabled", _yn(enabled))
    +    table.add_row("Account", account or "[dim]default[/dim]")
    +    table.add_row("Token env var", token_env)
    +    table.add_row("Token in env", _yn(token_set))
    +    table.add_row("Override existing", _yn(bool(op_cfg.get("override_existing", True))))
    +    table.add_row("Cache TTL (s)", str(op_cfg.get("cache_ttl_seconds", 300)))
    +    if binary:
    +        table.add_row("op binary", f"{binary} ({_op_version(binary)})")
    +    else:
    +        table.add_row("op binary", "[yellow]not found[/yellow]")
    +    table.add_row("References", str(len(references)))
    +
    +    console.print(Panel(table, title="1Password secret source", border_style="cyan"))
    +
    +    if references:
    +        ref_table = Table(show_header=True, header_style="bold")
    +        ref_table.add_column("Env var", style="cyan")
    +        ref_table.add_column("Reference")
    +        for name in sorted(references):
    +            ref_table.add_row(name, str(references[name]))
    +        console.print(ref_table)
    +
    +    if not enabled:
    +        console.print("\n  Run [cyan]hermes secrets onepassword setup[/cyan] to enable.")
    +        return 0
    +    if binary and not token_set:
    +        who = _op_whoami(binary, account)
    +        if who:
    +            console.print(f"\n  [green]Active op session:[/green] {who}")
    +        else:
    +            console.print(
    +                f"\n  [yellow]No active op session and {token_env} is unset — "
    +                "Hermes will warn and skip 1Password on next startup.[/yellow]"
    +            )
    +    if not references:
    +        console.print(
    +            "\n  [yellow]No references mapped yet.[/yellow]  Add one: "
    +            "[cyan]hermes secrets onepassword set ENV_VAR \"op://…\"[/cyan]"
    +        )
    +    return 0
    +
    +
    +def cmd_set(args: argparse.Namespace) -> int:
    +    console = Console()
    +    # Reuse the backend validator so the CLI and startup paths agree on what a
    +    # valid reference is — and store the *validated/stripped* value, not the
    +    # raw arg (so trailing whitespace never lands in config.yaml).
    +    valid, warnings = op_src._validate_references({args.env_var: args.reference})
    +    if args.env_var not in valid:
    +        for w in warnings:
    +            console.print(f"[red]{w}[/red]")
    +        return 1
    +
    +    cfg = load_config()
    +    op_cfg = cfg.setdefault("secrets", {}).setdefault("onepassword", {})
    +    env_map = op_cfg.get("env")
    +    if not isinstance(env_map, dict):
    +        env_map = {}
    +        op_cfg["env"] = env_map
    +    env_map[args.env_var] = valid[args.env_var]
    +    save_config(cfg)
    +    console.print(
    +        f"[green]✓[/green] mapped [cyan]{args.env_var}[/cyan] → "
    +        f"{valid[args.env_var]}"
    +    )
    +    if not op_cfg.get("enabled"):
    +        console.print(
    +            "  [yellow]Note: the integration is disabled — run "
    +            "[cyan]hermes secrets onepassword setup[/cyan] to turn it on.[/yellow]"
    +        )
    +    return 0
    +
    +
    +def cmd_remove(args: argparse.Namespace) -> int:
    +    console = Console()
    +    cfg = load_config()
    +    op_cfg = cfg.setdefault("secrets", {}).setdefault("onepassword", {})
    +    env_map = op_cfg.get("env")
    +    if not isinstance(env_map, dict) or args.env_var not in env_map:
    +        console.print(f"[yellow]{args.env_var} is not mapped.[/yellow]")
    +        return 1
    +    del env_map[args.env_var]
    +    save_config(cfg)
    +    console.print(f"[green]✓[/green] removed mapping for [cyan]{args.env_var}[/cyan]")
    +    return 0
    +
    +
    +def cmd_sync(args: argparse.Namespace) -> int:
    +    console = Console()
    +    cfg = load_config()
    +    op_cfg = (cfg.get("secrets") or {}).get("onepassword") or {}
    +    if not op_cfg.get("enabled"):
    +        console.print(
    +            "[yellow]1Password integration is disabled.  Run "
    +            "`hermes secrets onepassword setup` first.[/yellow]"
    +        )
    +        return 1
    +
    +    references = op_cfg.get("env") if isinstance(op_cfg.get("env"), dict) else {}
    +    if not references:
    +        console.print(
    +            "[yellow]No op:// references configured.  Add one with "
    +            "`hermes secrets onepassword set ENV_VAR \"op://…\"`.[/yellow]"
    +        )
    +        return 0
    +
    +    account = str(op_cfg.get("account", "") or "").strip()
    +    token_env = [redacted:secret-assignment]"service_account_token_env", _DEFAULT_TOKEN_ENV)
    +    binary_path = str(op_cfg.get("binary_path", "") or "").strip()
    +
    +    # --apply delegates to the same code path startup uses, so the skip /
    +    # override / token-guard policy lives in exactly one place.
    +    if args.apply:
    +        result = op_src.apply_onepassword_secrets(
    +            enabled=True,
    +            env=references,
    +            account=account,
    +            service_account_token_env=[redacted:secret-assignment]
    +            binary_path=binary_path,
    +            override_existing=bool(op_cfg.get("override_existing", True)),
    +            cache_ttl_seconds=0,  # an explicit sync always resolves fresh
    +        )
    +        if result.error:
    +            console.print(f"[red]{result.error}[/red]")
    +            return 1
    +        table = Table(show_header=True, header_style="bold")
    +        table.add_column("Env var", style="cyan")
    +        table.add_column("Action")
    +        for name in sorted(result.applied):
    +            table.add_row(name, "[green]exported[/green]")
    +        for name in sorted(result.skipped):
    +            table.add_row(name, "[dim]skipped (already set / token var)[/dim]")
    +        console.print(table)
    +        for w in result.warnings:
    +            console.print(f"[yellow]warning:[/yellow] {w}")
    +        console.print(
    +            f"\n  [green]Exported {len(result.applied)} secret(s) into current "
    +            "process.[/green]"
    +        )
    +        return 0
    +
    +    # Dry-run: resolve fresh (no cache) and preview, mutating nothing.
    +    try:
    +        secrets, warnings = op_src.fetch_onepassword_secrets(
    +            references=references,
    +            account=account,
    +            token_env=[redacted:secret-assignment]
    +            binary_path=binary_path,
    +            use_cache=False,
    +        )
    +    except RuntimeError as exc:
    +        console.print(f"[red]{exc}[/red]")
    +        return 1
    +
    +    override = bool(op_cfg.get("override_existing", True))
    +    table = Table(show_header=True, header_style="bold")
    +    table.add_column("Env var", style="cyan")
    +    table.add_column("Action")
    +    for name in sorted(references):
    +        if name == token_env:
    +            table.add_row(name, "[dim]skip (token var)[/dim]")
    +        elif name not in secrets:
    +            table.add_row(name, "[red]unresolved (see warnings)[/red]")
    +        elif os.environ.get(name) and not override:
    +            table.add_row(name, "[dim]skip (already set)[/dim]")
    +        else:
    +            already = bool(os.environ.get(name))
    +            table.add_row(
    +                name,
    +                "[green]would export[/green]" + (" (overrides)" if already else ""),
    +            )
    +    console.print(table)
    +    for w in warnings:
    +        console.print(f"[yellow]warning:[/yellow] {w}")
    +    console.print(
    +        "\n  This was a dry-run — references resolve automatically on the next "
    +        "[cyan]hermes[/cyan] invocation.  Re-run with [cyan]--apply[/cyan] to export "
    +        "into the current shell instead."
    +    )
    +    return 0
    +
    +
    +def cmd_disable(args: argparse.Namespace) -> int:
    +    console = Console()
    +    cfg = load_config()
    +    op_cfg = cfg.setdefault("secrets", {}).setdefault("onepassword", {})
    +    op_cfg["enabled"] = False
    +    save_config(cfg)
    +    console.print(
    +        "[green]Disabled.[/green]  1Password references will NOT be resolved on the "
    +        "next Hermes invocation.\n"
    +        "  Your reference mappings are left in config.yaml — remove them with "
    +        "[cyan]hermes secrets onepassword remove ENV_VAR[/cyan] if you no longer "
    +        "need them."
    +    )
    +    return 0
    +
    +
    +# ---------------------------------------------------------------------------
    +# Helpers
    +# ---------------------------------------------------------------------------
    +
    +
    +def _yn(b: bool) -> str:
    +    return "[green]yes[/green]" if b else "[dim]no[/dim]"
    +
    +
    +def _op_version(binary: Path) -> str:
    +    try:
    +        res = subprocess.run(
    +            [str(binary), "--version"],
    +            capture_output=True,
    +            text=True,
    +            timeout=5,
    +        )
    +        if res.returncode == 0:
    +            return (res.stdout or res.stderr).strip().splitlines()[0]
    +    except (OSError, subprocess.TimeoutExpired):
    +        pass
    +    return "version unknown"
    +
    +
    +def _op_whoami(binary: Path, account: str) -> Optional[str]:
    +    """Return a short identity string if op is authenticated, else None."""
    +    cmd = [str(binary), "whoami"]
    +    if account:
    +        cmd += ["--account", account]
    +    try:
    +        res = subprocess.run(cmd, capture_output=True, text=True, timeout=10)
    +    except (OSError, subprocess.TimeoutExpired):
    +        return None
    +    if res.returncode != 0:
    +        return None
    +    out = (res.stdout or "").strip()
    +    return out.replace("\n", " ")[:120] or "authenticated"
    ```
  - `hermes_cli/plugins.py`

    ```diff
    @@ -795,6 +795,53 @@ def register_browser_provider(self, provider) -> None:
                 self.manifest.name, provider.name,
             )
     
    +    # -- secret source registration -------------------------------------------
    +
    +    def register_secret_source(self, source) -> None:
    +        """Register an external secret-manager backend.
    +
    +        ``source`` must be an instance of
    +        :class:`agent.secret_sources.base.SecretSource`.  Registered
    +        sources run during ``load_hermes_dotenv()`` startup — after
    +        ``~/.hermes/.env`` loads, before Hermes reads credentials — when
    +        their ``secrets.<source.name>`` config section is enabled.  The
    +        orchestrator (``agent.secret_sources.registry.apply_all``) owns
    +        ordering, mapped-vs-bulk precedence, conflict warnings, and
    +        provenance; the source only fetches.
    +
    +        NOTE ON TIMING: plugin discovery happens later in startup than
    +        the first ``load_hermes_dotenv()`` call, so a plugin-registered
    +        source is not consulted by the initial env load of the process
    +        that discovers it.  It IS consulted by every subsequently
    +        spawned Hermes process (gateway children, cron sessions,
    +        subagents), and immediately after a
    +        ``reset_secret_source_cache()`` re-pull.  Plugin sources are
    +        therefore best for supplying credentials to the running fleet;
    +        the bundled sources cover first-process bootstrap.
    +
    +        Contract requirements (rejected with a warning otherwise):
    +        inherit from ``SecretSource``, ``api_version`` matching
    +        ``SECRET_SOURCE_API_VERSION``, lowercase unique ``name``,
    +        ``shape`` of ``"mapped"`` or ``"bulk"``, unique ``scheme`` (when
    +        set), and a ``fetch()`` that never raises and never prompts.
    +        See the base-module docstring for the full contract.
    +        """
    +        from agent.secret_sources.base import SecretSource
    +        from agent.secret_sources.registry import register_source
    +
    +        if not isinstance(source, SecretSource):
    +            logger.warning(
    +                "Plugin '%s' tried to register a secret source that does "
    +                "not inherit from SecretSource. Ignoring.",
    +                self.manifest.name,
    +            )
    +            return
    +        if register_source(source):
    +            logger.info(
    +                "Plugin '%s' registered secret source: %s",
    +                self.manifest.name, source.name,
    +            )
    +
         # -- TTS provider registration -------------------------------------------
     
         def register_tts_provider(self, provider) -> None:
    ```
  - `scripts/release.py`

    ```diff
    @@ -45,6 +45,7 @@
     
     # Auto-extracted from noreply emails + manual overrides
     AUTHOR_MAP = {
    +    "[redacted:email]": "hwrdprkns",  # PR #36896 salvage (secrets: [redacted:secret-assignment] op:// secret source + shared _cache substrate, adapted onto the SecretSource interface)
         "[redacted:email]": "isheng-eqi",  # PR #59428 salvage (cron: reject past one-shot timestamps in update_job fallback + resume_job; #59395). Also PR #59446 salvage (cron: advance one-shot next_run_at before dispatch so concurrent gateway+desktop schedulers can't double-execute; #59229).
         "[redacted:email]": "derek2000139",  # PR #57838 salvage (desktop/windows: pre-write update marker before quit dwell so the renderer's waitForUpdateToFinish gate parks instead of respawning a backend that re-locks venv .pyd files mid-update)
         "[redacted:email]": "AndreasHiltner",  # PR #56854 salvage (gateway: route multiplex profile responses through the profile's own adapter — 53-site _adapter_for_source sweep)
    ```
  - `tests/secret_sources/conformance.py`

    ```diff
    @@ -0,0 +1,123 @@
    +"""Conformance kit for :class:`agent.secret_sources.base.SecretSource`.
    +
    +Any secret-source backend — bundled or external plugin — can validate
    +itself against the contract by subclassing :class:`SecretSourceConformance`
    +and providing a ``source`` fixture (plus optional per-source config
    +fixtures).  Example::
    +
    +    from tests.secret_sources.conformance import SecretSourceConformance
    +
    +    class TestMySourceConformance(SecretSourceConformance):
    +        @pytest.fixture
    +        def source(self):
    +            return MySource()
    +
    +The checks encode the parts of the contract that break OTHER people
    +when violated: never raising, never prompting (stdin closed), respecting
    +disabled config, valid identity attributes, and orchestrator
    +compatibility.
    +"""
    +
    +from __future__ import annotations
    +
    +from pathlib import Path
    +
    +import pytest
    +
    +from agent.secret_sources.base import (
    +    SECRET_SOURCE_API_VERSION,
    +    FetchResult,
    +    SecretSource,
    +)
    +from agent.secret_sources.registry import (
    +    _reset_registry_for_tests,
    +    apply_all,
    +    register_source,
    +)
    +
    +
    +class SecretSourceConformance:
    +    """Base class of contract checks; subclass and provide ``source``."""
    +
    +    @pytest.fixture
    +    def source(self) -> SecretSource:  # pragma: no cover — must override
    +        raise NotImplementedError("conformance subclasses must provide a source fixture")
    +
    +    @pytest.fixture
    +    def minimal_cfg(self) -> dict:
    +        """An enabled-but-unconfigured section — the common misconfig case."""
    +        return {"enabled": True}
    +
    +    # -- identity ----------------------------------------------------------
    +
    +    def test_name_is_lowercase_identifier(self, source):
    +        assert source.name, "source.name must be non-empty"
    +        assert source.name == source.name.lower()
    +        assert source.name.replace("_", "").isalnum()
    +
    +    def test_label_present(self, source):
    +        assert source.label, "source.label must be a human-readable name"
    +
    +    def test_shape_valid(self, source):
    +        assert source.shape in ("mapped", "bulk")
    +
    +    def test_api_version_current(self, source):
    +        assert source.api_version == SECRET_SOURCE_API_VERSION
    +
    +    # -- contract behavior --------------------------------------------------
    +
    +    def test_fetch_never_raises_on_malformed_config(self, source, tmp_path):
    +        """Every degenerate config shape must produce a FetchResult, not a raise."""
    +        for cfg in ({}, {"enabled": True}, {"enabled": True, "env": "not-a-dict"},
    +                    {"enabled": True, "cache_ttl_seconds": "bogus"}, None):
    +            result = source.fetch(cfg if isinstance(cfg, dict) else {}, tmp_path)
    +            assert isinstance(result, FetchResult), (
    +                f"fetch() returned {type(result).__name__} for cfg={cfg!r}"
    +            )
    +
    +    def test_fetch_unconfigured_reports_error_not_secrets(self, source, tmp_path,
    +                                                          minimal_cfg, monkeypatch):
    +        """enabled=true with nothing else set must fail cleanly with a kind."""
    +        result = source.fetch(minimal_cfg, tmp_path)
    +        assert isinstance(result, FetchResult)
    +        if not result.ok:
    +            assert result.error_kind is not None, (
    +                "errors must carry a machine-readable ErrorKind"
    +            )
    +            assert not result.secrets
    +
    +    def test_disabled_by_default(self, source):
    +        assert source.is_enabled({}) is False
    +        assert source.is_enabled({"enabled": False}) is False
    +
    +    def test_timeout_is_positive(self, source, minimal_cfg):
    +        assert source.fetch_timeout_seconds(minimal_cfg) > 0
    +        # Garbage config must not break the timeout accessor either.
    +        assert source.fetch_timeout_seconds({"timeout_seconds": "junk"}) > 0
    +
    +    def test_protected_vars_are_valid_names(self, source, minimal_cfg):
    +        from agent.secret_sources.base import is_valid_env_name
    +
    +        for var in source.protected_env_vars(minimal_cfg):
    +            assert is_valid_env_name(var)
    +
    +    # -- orchestrator compatibility ------------------------------------------
    +
    +    def test_registers_and_applies_via_orchestrator(self, source, tmp_path,
    +                                                    monkeypatch):
    +        """The source must survive a full apply_all() pass without breaking it."""
    +        _reset_registry_for_tests()
    +        # Prevent the bundled sources from interfering.
    +        monkeypatch.setattr(
    +            "agent.secret_sources.registry._ensure_builtin_sources", lambda: None
    +        )
    +        try:
    +            assert register_source(source), "register_source() rejected the source"
    +            env: dict = {}
    +            report = apply_all(
    +                {source.name: {"enabled": True}}, tmp_path, environ=env
    +            )
    +            names = [sr.name for sr in report.sources]
    +            assert source.name in names
    +        finally:
    +            _reset_registry_for_tests()
    ```
  - `tests/secret_sources/test_secret_source_registry.py`

    ```diff
    @@ -0,0 +1,600 @@
    +"""Tests for the secret-source contract + orchestrator.
    +
    +Covers: registration gating (API version, name/scheme uniqueness, shape),
    +apply_all precedence (mapped beats bulk, first-wins, override_existing,
    +protected vars), conflict surfacing, timeout enforcement, provenance,
    +and Bitwarden's SecretSource adapter — plus the conformance kit run
    +against the bundled Bitwarden source.
    +"""
    +
    +from __future__ import annotations
    +
    +import sys
    +import time
    +from pathlib import Path
    +
    +import pytest
    +
    +ROOT = Path(__file__).resolve().parents[2]
    +if str(ROOT) not in sys.path:
    +    sys.path.insert(0, str(ROOT))
    +
    +from agent.secret_sources.base import (  # noqa: E402
    +    SECRET_SOURCE_API_VERSION,
    +    ErrorKind,
    +    FetchResult,
    +    SecretSource,
    +    is_valid_env_name,
    +    run_secret_cli,
    +    scrub_ansi,
    +)
    +from agent.secret_sources import registry as reg  # noqa: E402
    +from agent.secret_sources.bitwarden import BitwardenSource  # noqa: E402
    +from tests.secret_sources.conformance import SecretSourceConformance  # noqa: E402
    +
    +
    [redacted:email](autouse=True)
    +def _clean_registry(monkeypatch):
    +    """Each test starts with an empty registry and no builtin auto-load."""
    +    reg._reset_registry_for_tests()
    +    monkeypatch.setattr(reg, "_ensure_builtin_sources", lambda: None)
    +    yield
    +    reg._reset_registry_for_tests()
    +
    +
    +def _make_source(
    +    name="dummy",
    +    shape="mapped",
    +    secrets=None,
    +    error=None,
    +    error_kind=None,
    +    scheme=None,
    +    override=False,
    +    protected=(),
    +    api_version=SECRET_SOURCE_API_VERSION,
    +    fetch_fn=None,
    +):
    +    """Build a minimal conforming source for orchestrator tests."""
    +
    +    class _Src(SecretSource):
    +        def fetch(self, cfg, home_path):
    +            if fetch_fn is not None:
    +                return fetch_fn(cfg, home_path)
    +            res = FetchResult()
    +            if error:
    +                res.error = error
    +                res.error_kind = error_kind or ErrorKind.INTERNAL
    +            else:
    +                res.secrets = [redacted:secret-assignment] or {})
    +            return res
    +
    +        def override_existing(self, cfg):
    +            return override
    +
    +        def protected_env_vars(self, cfg):
    +            return frozenset(protected)
    +
    +    _Src.name = name
    +    _Src.label = name.title()
    +    _Src.shape = shape
    +    _Src.scheme = scheme
    +    _Src.api_version = api_version
    +    return _Src()
    +
    +
    +# ---------------------------------------------------------------------------
    +# Registration gating
    +# ---------------------------------------------------------------------------
    +
    +
    +class TestRegistration:
    +    def test_registers_conforming_source(self):
    +        assert reg.register_source(_make_source()) is True
    +        assert reg.get_source("dummy") is not None
    +
    +    def test_rejects_non_secretsource_instance(self):
    +        assert reg.register_source(object()) is False
    +
    +    def test_rejects_wrong_api_version(self):
    +        src = _make_source(api_version=SECRET_SOURCE_API_VERSION + 1)
    +        assert reg.register_source(src) is False
    +
    +    def test_rejects_invalid_name(self):
    +        assert reg.register_source(_make_source(name="Bad Name")) is False
    +        assert reg.register_source(_make_source(name="")) is False
    +        assert reg.register_source(_make_source(name="UPPER")) is False
    +
    +    def test_rejects_invalid_shape(self):
    +        assert reg.register_source(_make_source(shape="sideways")) is False
    +
    +    def test_rejects_duplicate_name_without_replace(self):
    +        assert reg.register_source(_make_source(name="dup")) is True
    +        assert reg.register_source(_make_source(name="dup")) is False
    +        assert reg.register_source(_make_source(name="dup"), replace=True) is True
    +
    +    def test_rejects_scheme_collision_across_names(self):
    +        assert reg.register_source(_make_source(name="one", scheme="op")) is True
    +        assert reg.register_source(_make_source(name="two", scheme="op")) is False
    +
    +    def test_same_name_replace_keeps_scheme(self):
    +        assert reg.register_source(_make_source(name="one", scheme="op")) is True
    +        assert reg.register_source(
    +            _make_source(name="one", scheme="op"), replace=True
    +        ) is True
    +
    +
    +# ---------------------------------------------------------------------------
    +# apply_all: precedence, conflicts, protection
    +# ---------------------------------------------------------------------------
    +
    +
    +class TestApplyAll:
    +    def test_disabled_sources_do_not_run(self, tmp_path):
    +        called = []
    +
    +        def _fetch(cfg, home):
    +            called.append(True)
    +            return FetchResult(secrets={"A": "1"})
    +
    +        reg.register_source(_make_source(fetch_fn=_fetch))
    +        env: dict = {}
    +        report = reg.apply_all({"dummy": {"enabled": False}}, tmp_path, environ=env)
    +        assert not called
    +        assert not report.sources
    +        assert env == {}
    +
    +    def test_applies_secrets_and_records_provenance(self, tmp_path):
    +        reg.register_source(_make_source(secrets={"API_KEY": "v1"}))
    +        env: dict = {}
    +        report = reg.apply_all({"dummy": {"enabled": True}}, tmp_path, environ=env)
    +        assert env["API_KEY"] == "v1"
    +        assert report.provenance["API_KEY"].source == "dummy"
    +        assert report.provenance["API_KEY"].shape == "mapped"
    +        assert report.provenance["API_KEY"].overrode_env is False
    +
    +    def test_existing_env_wins_without_override(self, tmp_path):
    +        reg.register_source(_make_source(secrets={"API_KEY": "vault"}))
    +        env = {"API_KEY": "dotenv"}
    +        report = reg.apply_all({"dummy": {"enabled": True}}, tmp_path, environ=env)
    +        assert env["API_KEY"] == "dotenv"
    +        assert "API_KEY" in report.sources[0].skipped_existing
    +
    +    def test_override_existing_beats_env_and_is_attributed(self, tmp_path):
    +        reg.register_source(_make_source(secrets={"API_KEY": "vault"}, override=True))
    +        env = {"API_KEY": "dotenv"}
    +        report = reg.apply_all({"dummy": {"enabled": True}}, tmp_path, environ=env)
    +        assert env["API_KEY"] == "vault"
    +        assert report.provenance["API_KEY"].overrode_env is True
    +
    +    def test_mapped_beats_bulk_regardless_of_order(self, tmp_path):
    +        reg.register_source(
    +            _make_source(name="bulky", shape="bulk", secrets={"K": "bulk"})
    +        )
    +        reg.register_source(
    +            _make_source(name="mappy", shape="mapped", secrets={"K": "mapped"})
    +        )
    +        env: dict = {}
    +        # bulk listed first in sources order — mapped must still win.
    +        report = reg.apply_all(
    +            {"sources": ["bulky", "mappy"],
    +             "bulky": {"enabled": True}, "mappy": {"enabled": True}},
    +            tmp_path, environ=env,
    +        )
    +        assert env["K"] == "mapped"
    +        assert report.provenance["K"].source == "mappy"
    +        assert report.conflicts, "shadowed bulk claim must surface a warning"
    +
    +    def test_first_source_wins_within_shape(self, tmp_path):
    +        reg.register_source(_make_source(name="alpha", secrets={"K": "a"}))
    +        reg.register_source(_make_source(name="beta", secrets={"K": "b"}))
    +        env: dict = {}
    +        report = reg.apply_all(
    +            {"sources": ["beta", "alpha"],
    +             "alpha": {"enabled": True}, "beta": {"enabled": True}},
    +            tmp_path, environ=env,
    +        )
    +        assert env["K"] == "b"  # beta listed first
    +        assert report.provenance["K"].source == "beta"
    +        beta_first = [s for s in report.sources if s.name == "alpha"][0]
    +        assert "K" in beta_first.skipped_claimed
    +
    +    def test_cross_source_override_never_clobbers_prior_claim(self, tmp_path):
    +        """override_existing beats .env, NEVER another source's claim."""
    +        reg.register_source(_make_source(name="alpha", secrets={"K": "a"}))
    +        reg.register_source(
    +            _make_source(name="beta", secrets={"K": "b"}, override=True)
    +        )
    +        env: dict = {}
    +        report = reg.apply_all(
    +            {"sources": ["alpha", "beta"],
    +             "alpha": {"enabled": True}, "beta": {"enabled": True}},
    +            tmp_path, environ=env,
    +        )
    +        assert env["K"] == "a"
    +        assert report.conflicts
    +
    +    def test_protected_vars_never_overwritten_by_any_source(self, tmp_path):
    +        reg.register_source(
    +            _make_source(name="alpha", secrets={"BOOT_TOKEN": "evil"},
    +                         override=True, protected=("BOOT_TOKEN",))
    +        )
    +        env = {"BOOT_TOKEN": "real"}
    +        report = reg.apply_all({"alpha": {"enabled": True}}, tmp_path, environ=env)
    +        assert env["BOOT_TOKEN"] == "real"
    +        assert "BOOT_TOKEN" in report.sources[0].skipped_protected
    +
    +    def test_invalid_env_names_skipped(self, tmp_path):
    +        reg.register_source(
    +            _make_source(secrets={"GOOD_NAME": "v", "bad-name": "v", "1BAD": "v"})
    +        )
    +        env: dict = {}
    +        report = reg.apply_all({"dummy": {"enabled": True}}, tmp_path, environ=env)
    +        assert "GOOD_NAME" in env and "bad-name" not in env and "1BAD" not in env
    +        assert set(report.sources[0].skipped_invalid) == {"bad-name", "1BAD"}
    +
    +    def test_failed_source_does_not_block_others(self, tmp_path):
    +        reg.register_source(
    +            _make_source(name="broken", error="boom", error_kind=ErrorKind.NETWORK)
    +        )
    +        reg.register_source(_make_source(name="works", secrets={"K": "v"}))
    +        env: dict = {}
    +        report = reg.apply_all(
    +            {"broken": {"enabled": True}, "works": {"enabled": True}},
    +            tmp_path, environ=env,
    +        )
    +        assert env["K"] == "v"
    +        broken = [s for s in report.sources if s.name == "broken"][0]
    +        assert broken.result.error_kind is ErrorKind.NETWORK
    +
    +    def test_raising_fetch_contained_as_internal_error(self, tmp_path):
    +        def _explode(cfg, home):
    +            raise ValueError("plugin bug")
    +
    +        reg.register_source(_make_source(name="buggy", fetch_fn=_explode))
    +        env: dict = {}
    +        report = reg.apply_all({"buggy": {"enabled": True}}, tmp_path, environ=env)
    +        assert report.sources[0].result.error_kind is ErrorKind.INTERNAL
    +        assert "plugin bug" in report.sources[0].result.error
    +
    +    def test_wrong_return_type_contained(self, tmp_path):
    +        reg.register_source(
    +            _make_source(name="liar", fetch_fn=lambda cfg, home: {"not": "a result"})
    +        )
    +        report = reg.apply_all({"liar": {"enabled": True}}, tmp_path, environ={})
    +        assert report.sources[0].result.error_kind is ErrorKind.INTERNAL
    +
    +    def test_timeout_enforced(self, tmp_path):
    +        def _slow(cfg, home):
    +            time.sleep(5)
    +            return FetchResult(secrets={"K": "late"})
    +
    +        src = _make_source(name="slow", fetch_fn=_slow)
    +        src.fetch_timeout_seconds = lambda cfg: 0.2
    +        reg.register_source(src)
    +        env: dict = {}
    +        start = time.monotonic()
    +        report = reg.apply_all({"slow": {"enabled": True}}, tmp_path, environ=env)
    +        assert time.monotonic() - start < 3
    +        assert report.sources[0].result.error_kind is ErrorKind.TIMEOUT
    +        assert "K" not in env
    +
    +    def test_malformed_secrets_cfg_shapes_are_safe(self, tmp_path):
    +        reg.register_source(_make_source(secrets={"K": "v"}))
    +        for cfg in (None, [], "junk", {"dummy": "not-a-dict"}, {"sources": "junk"}):
    +            report = reg.apply_all(cfg, tmp_path, environ={})
    +            assert isinstance(report, reg.ApplyReport)
    +
    +    def test_unknown_sources_entry_warns_but_continues(self, tmp_path, caplog):
    +        reg.register_source(_make_source(secrets={"K": "v"}))
    +        env: dict = {}
    +        reg.apply_all(
    +            {"sources": ["ghost", "dummy"], "dummy": {"enabled": True}},
    +            tmp_path, environ=env,
    +        )
    +        assert env["K"] == "v"
    +
    +
    +# ---------------------------------------------------------------------------
    +# Shared helpers
    +# ---------------------------------------------------------------------------
    +
    +
    +class TestHelpers:
    +    def test_is_valid_env_name(self):
    +        assert is_valid_env_name("GOOD_NAME")
    +        assert is_valid_env_name("_LEADING")
    +        assert not is_valid_env_name("")
    +        assert not is_valid_env_name("1BAD")
    +        assert not is_valid_env_name("bad-name")
    +        assert not is_valid_env_name("has space")
    +
    +    def test_scrub_ansi_removes_whole_sequences(self):
    +        assert scrub_ansi("\x1b[31mred\x1b[0m plain") == "red plain"
    +        assert scrub_ansi("\x1b]0;title\x07text") == "text"
    +        assert scrub_ansi("") == ""
    +
    +    def test_run_secret_cli_minimal_env(self):
    +        proc = run_secret_cli(
    +            [sys.executable, "-c",
    +             "import os, json; print(json.dumps(sorted(os.environ)))"],
    +        )
    +        import json
    +
    +        child_env = json.loads(proc.stdout)
    +        # No credential-bearing vars from the parent env leak through.
    +        assert not any(k.endswith(("_API_KEY", "_TOKEN", "_SECRET"))
    +                       for k in child_env)
    +        assert "NO_COLOR" in child_env
    +
    +    def test_run_secret_cli_allowlist_passes_named_vars(self, monkeypatch):
    +        monkeypatch.setenv("MY_AUTH_TOKEN", "tok")
    +        monkeypatch.setenv("OTHER_API_KEY", "leak")
    +        proc = run_secret_cli(
    +            [sys.executable, "-c",
    +             "import os; print(os.environ.get('MY_AUTH_TOKEN', '')); "
    +             "print(os.environ.get('OTHER_API_KEY', ''))"],
    +            allow_env=["MY_AUTH_TOKEN"],
    +        )
    +        lines = proc.stdout.splitlines()
    +        assert lines[0] == "tok"
    +        assert lines[1] == ""
    +
    +    def test_run_secret_cli_timeout_raises_runtime_error(self):
    +        with pytest.raises(RuntimeError, match="timed out"):
    +            run_secret_cli(
    +                [sys.executable, "-c", "import time; time.sleep(10)"],
    +                timeout=0.3,
    +            )
    +
    +    def test_run_secret_cli_stdin_devnull(self):
    +        # A helper that tries to prompt reads EOF immediately.
    +        proc = run_secret_cli(
    +            [sys.executable, "-c",
    +             "import sys; print(repr(sys.stdin.read()))"],
    +        )
    +        assert proc.stdout.strip() == "''"
    +
    +
    +# ---------------------------------------------------------------------------
    +# Bitwarden adapter
    +# ---------------------------------------------------------------------------
    +
    +
    +class TestBitwardenSource:
    +    def test_identity(self):
    +        src = BitwardenSource()
    +        assert src.name == "bitwarden"
    +        assert src.shape == "bulk"
    +        assert src.scheme == "bws"
    +
    +    def test_override_existing_defaults_true(self):
    +        src = BitwardenSource()
    +        assert src.override_existing({}) is True
    +        assert src.override_existing({"override_existing": False}) is False
    +
    +    def test_protected_vars_track_token_env(self):
    +        src = BitwardenSource()
    +        assert src.protected_env_vars({}) == frozenset({"BWS_ACCESS_TOKEN"})
    +        assert src.protected_env_vars(
    +            {"access_token_env": "CUSTOM_TOKEN"}
    +        ) == frozenset({"CUSTOM_TOKEN"})
    +
    +    def test_fetch_missing_token_not_configured(self, tmp_path, monkeypatch):
    +        monkeypatch.delenv("BWS_ACCESS_TOKEN", raising=False)
    +        result = BitwardenSource().fetch({"enabled": True}, tmp_path)
    +        assert result.error_kind is ErrorKind.NOT_CONFIGURED
    +        assert "BWS_ACCESS_TOKEN" in result.error
    +
    +    def test_fetch_missing_project_not_configured(self, tmp_path, monkeypatch):
    +        monkeypatch.setenv("BWS_ACCESS_TOKEN", "0.token")
    +        result = BitwardenSource().fetch({"enabled": True}, tmp_path)
    +        assert result.error_kind is ErrorKind.NOT_CONFIGURED
    +        assert "project_id" in result.error
    +
    +    def test_fetch_delegates_to_fetch_bitwarden_secrets(self, tmp_path, monkeypatch):
    +        monkeypatch.setenv("BWS_ACCESS_TOKEN", "0.token")
    +        import agent.secret_sources.bitwarden as bw
    +
    +        monkeypatch.setattr(bw, "find_bws", lambda **kw: Path("/fake/bws"))
    +        captured = {}
    +
    +        def _fake_fetch(**kwargs):
    +            captured.update(kwargs)
    +            return {"MY_KEY": "val"}, ["a warning"]
    +
    +        monkeypatch.setattr(bw, "fetch_bitwarden_secrets", _fake_fetch)
    +        result = BitwardenSource().fetch(
    +            {"enabled": True, "project_id": "proj",
    +             "server_url": " https://vault.bitwarden.eu "},
    +            tmp_path,
    +        )
    +        assert result.ok
    +        assert result.secrets == {"MY_KEY": "val"}
    +        assert result.warnings == ["a warning"]
    +        assert captured["project_id"] == "proj"
    +        assert captured["server_url"] == "https://vault.bitwarden.eu"
    +        assert captured["home_path"] == tmp_path
    +
    +    def test_fetch_runtime_error_classified(self, tmp_path, monkeypatch):
    +        monkeypatch.setenv("BWS_ACCESS_TOKEN", "0.token")
    +        import agent.secret_sources.bitwarden as bw
    +
    +        monkeypatch.setattr(bw, "find_bws", lambda **kw: Path("/fake/bws"))
    +
    +        def _fail(**kwargs):
    +            raise RuntimeError("bws exited 1: 401 unauthorized")
    +
    +        monkeypatch.setattr(bw, "fetch_bitwarden_secrets", _fail)
    +        result = BitwardenSource().fetch(
    +            {"enabled": True, "project_id": "proj"}, tmp_path
    +        )
    +        assert result.error_kind is ErrorKind.AUTH_FAILED
    +
    +    def test_e2e_through_orchestrator(self, tmp_path, monkeypatch):
    +        """Full path: registry → BitwardenSource → env, with fetch mocked."""
    +        monkeypatch.setenv("BWS_ACCESS_TOKEN", "0.token")
    +        import agent.secret_sources.bitwarden as bw
    +
    +        monkeypatch.setattr(bw, "find_bws", lambda **kw: Path("/fake/bws"))
    +        monkeypatch.setattr(
    +            bw, "fetch_bitwarden_secrets",
    +            lambda **kw: ({"ANTHROPIC_API_KEY": "sk-ant", "BWS_ACCESS_TOKEN": "steal"}, []),
    +        )
    +        reg.register_source(BitwardenSource())
    +        env = {"BWS_ACCESS_TOKEN": "0.token"}
    +        report = reg.apply_all(
    +            {"bitwarden": {"enabled": True, "project_id": "proj"}},
    +            tmp_path, environ=env,
    +        )
    +        assert env["ANTHROPIC_API_KEY"] == "sk-ant"
    +        # The bootstrap token is protected even though BSM carried it.
    +        assert env["BWS_ACCESS_TOKEN"] == "0.token"
    +        assert report.provenance["ANTHROPIC_API_KEY"].source == "bitwarden"
    +
    +
    +# ---------------------------------------------------------------------------
    +# Conformance kit applied to the bundled source
    +# ---------------------------------------------------------------------------
    +
    +
    +class TestBitwardenConformance(SecretSourceConformance):
    +    @pytest.fixture
    +    def source(self, monkeypatch):
    +        # Never hit the network / auto-install path in conformance runs.
    +        import agent.secret_sources.bitwarden as bw
    +
    +        monkeypatch.setattr(bw, "find_bws", lambda **kw: None)
    +        monkeypatch.delenv("BWS_ACCESS_TOKEN", raising=False)
    +        return BitwardenSource()
    +
    +
    +# ---------------------------------------------------------------------------
    +# 1Password adapter
    +# ---------------------------------------------------------------------------
    +
    +
    +class TestOnePasswordSource:
    +    def test_identity(self):
    +        from agent.secret_sources.onepassword import OnePasswordSource
    +
    +        src = OnePasswordSource()
    +        assert src.name == "onepassword"
    +        assert src.shape == "mapped"
    +        assert src.scheme == "op"
    +
    +    def test_override_existing_defaults_true(self):
    +        from agent.secret_sources.onepassword import OnePasswordSource
    +
    +        src = OnePasswordSource()
    +        assert src.override_existing({}) is True
    +        assert src.override_existing({"override_existing": False}) is False
    +
    +    def test_protected_vars_track_token_env(self):
    +        from agent.secret_sources.onepassword import OnePasswordSource
    +
    +        src = OnePasswordSource()
    +        assert src.protected_env_vars({}) == frozenset(
    +            {"OP_SERVICE_ACCOUNT_TOKEN"}
    +        )
    +        assert src.protected_env_vars(
    +            {"service_account_token_env": "MY_OP_TOKEN"}
    +        ) == frozenset({"MY_OP_TOKEN"})
    +
    +    def test_fetch_empty_map_not_configured(self, tmp_path):
    +        from agent.secret_sources.onepassword import OnePasswordSource
    +
    +        result = OnePasswordSource().fetch({"enabled": True}, tmp_path)
    +        assert result.error_kind is ErrorKind.NOT_CONFIGURED
    +
    +    def test_fetch_missing_binary(self, tmp_path, monkeypatch):
    +        import agent.secret_sources.onepassword as op
    +
    +        monkeypatch.setattr(op, "find_op", lambda *_a, **_kw: None)
    +        result = op.OnePasswordSource().fetch(
    +            {"enabled": True, "env": {"K": "op://V/I/F"}}, tmp_path
    +        )
    +        assert result.error_kind is ErrorKind.BINARY_MISSING
    +
    +    def test_fetch_delegates_and_passes_config(self, tmp_path, monkeypatch):
    +        import agent.secret_sources.onepassword as op
    +
    +        monkeypatch.setattr(op, "find_op", lambda *_a, **_kw: Path("/fake/op"))
    +        captured = {}
    +
    +        def _fake_fetch(**kwargs):
    +            captured.update(kwargs)
    +            return {"K": "v"}, ["warn"]
    +
    +        monkeypatch.setattr(op, "fetch_onepassword_secrets", _fake_fetch)
    +        result = op.OnePasswordSource().fetch(
    +            {"enabled": True, "env": {"K": "op://V/I/F"},
    +             "account": "team", "service_account_token_env": "MY_TOK"},
    +            tmp_path,
    +        )
    +        assert result.ok and result.secrets == {"K": "v"}
    +        assert captured["references"] == {"K": "op://V/I/F"}
    +        assert captured["account"] == "team"
    +        assert captured["token_env"] == "MY_TOK"
    +
    +    def test_invalid_refs_warned_not_fatal(self, tmp_path, monkeypatch):
    +        import agent.secret_sources.onepassword as op
    +
    +        monkeypatch.setattr(op, "find_op", lambda *_a, **_kw: Path("/fake/op"))
    +        monkeypatch.setattr(op, "fetch_onepassword_secrets",
    +                            lambda **kw: ({"GOOD": "v"}, []))
    +        result = op.OnePasswordSource().fetch(
    +            {"enabled": True,
    +             "env": {"GOOD": "op://V/I/F", "BAD": "not-a-ref",
    +                     "bad name": "op://V/I/F"}},
    +            tmp_path,
    +        )
    +        assert result.ok
    +        assert len(result.warnings) == 2
    +
    +    def test_mapped_op_beats_bulk_bitwarden_through_orchestrator(
    +        self, tmp_path, monkeypatch
    +    ):
    +        """The headline multi-source scenario: both vaults claim the same var."""
    +        import agent.secret_sources.bitwarden as bw
    +        import agent.secret_sources.onepassword as op
    +
    +        monkeypatch.setenv("BWS_ACCESS_TOKEN", "0.token")
    +        monkeypatch.setattr(bw, "find_bws", lambda **kw: Path("/fake/bws"))
    +        monkeypatch.setattr(
    +            bw, "fetch_bitwarden_secrets",
    +            lambda **kw: ({"SHARED_KEY": "from-bitwarden",
    +                           "BW_ONLY": "bw-val"}, []),
    +        )
    +        monkeypatch.setattr(op, "find_op", lambda *_a, **_kw: Path("/fake/op"))
    +        monkeypatch.setattr(
    +            op, "fetch_onepassword_secrets",
    +            lambda **kw: ({"SHARED_KEY": "from-1password"}, []),
    +        )
    +        reg.register_source(bw.BitwardenSource())
    +        reg.register_source(op.OnePasswordSource())
    +        env = {"BWS_ACCESS_TOKEN": "0.token"}
    +        report = reg.apply_all(
    +            {
    +                # bitwarden listed FIRST — mapped 1Password must still win.
    +                "sources": ["bitwarden", "onepassword"],
    +                "bitwarden": {"enabled": True, "project_id": "proj"},
    +                "onepassword": {"enabled": True,
    +                                "env": {"SHARED_KEY": "op://V/I/F"}},
    +            },
    +            tmp_path, environ=env,
    +        )
    +        assert env["SHARED_KEY"] == "from-1password"
    +        assert env["BW_ONLY"] == "bw-val"
    +        assert report.provenance["SHARED_KEY"].source == "onepassword"
    +        assert report.provenance["BW_ONLY"].source == "bitwarden"
    +        assert report.conflicts  # the shadowed bitwarden claim is surfaced
    +
    +
    +class TestOnePasswordConformance(SecretSourceConformance):
    +    @pytest.fixture
    +    def source(self, monkeypatch):
    +        import agent.secret_sources.onepassword as op
    +
    +        monkeypatch.setattr(op, "find_op", lambda *_a, **_kw: None)
    +        monkeypatch.delenv("OP_SERVICE_ACCOUNT_TOKEN", raising=False)
    +        return op.OnePasswordSource()
    ```
  - `tests/test_bitwarden_secrets.py`

    ```diff
    @@ -639,20 +639,23 @@ def test_env_loader_calls_bsm_when_enabled(tmp_path, monkeypatch):
         monkeypatch.delenv("MY_BSM_KEY", raising=False)
     
         called = {"n": 0}
    -    def fake_apply(**kwargs):
    +
    +    def fake_fetch(**kwargs):
             called["n"] += 1
    -        assert kwargs["enabled"] is True
             assert kwargs["project_id"] == "proj-1"
    -        os.environ["MY_BSM_KEY"] = "from-bsm"
    -        return bw.FetchResult(
    -            secrets={"MY_BSM_KEY": "from-bsm"},
    -            applied=["MY_BSM_KEY"],
    -        )
    +        return {"MY_BSM_KEY": "from-bsm"}, []
     
         monkeypatch.setattr(
    -        "agent.secret_sources.bitwarden.apply_bitwarden_secrets",
    -        fake_apply,
    +        "agent.secret_sources.bitwarden.find_bws",
    +        lambda **_kw: Path("/fake/bws"),
         )
    +    monkeypatch.setattr(
    +        "agent.secret_sources.bitwarden.fetch_bitwarden_secrets",
    +        fake_fetch,
    +    )
    +    from agent.secret_sources import registry as reg_module
    +
    +    reg_module._reset_registry_for_tests()
     
         from hermes_cli.env_loader import _apply_external_secret_sources
         _apply_external_secret_sources(home)
    ```
  - `tests/test_env_loader_op_bootstrap.py`

    ```diff
    @@ -0,0 +1,165 @@
    +"""Tests for the 1Password bootstrap-token reliability patches.
    +
    +Two behaviours are covered:
    +
    +1. ``load_hermes_dotenv()`` auto-loads ``~/.hermes/.op.env`` so the
    +   ``OP_SERVICE_ACCOUNT_TOKEN`` bootstrap token is available to
    +   ``apply_onepassword_secrets()`` in cron / subprocess / macOS / Docker
    +   contexts that inherit no shell state (no systemd EnvironmentFile, no
    +   ``op run``).  ``.op.env`` must never override a token already present
    +   in the environment (e.g. injected by a systemd ``EnvironmentFile``).
    +
    +2. ``credential_pool._seed_from_env`` (via the inner
    +   ``_get_env_prefer_dotenv``) must prefer an already-resolved value from
    +   ``os.environ`` over a raw ``op://`` reference still sitting in ``.env``,
    +   while leaving the normal ``.env``-takes-precedence behaviour untouched
    +   for every non-``op://`` value.
    +
    +These stay fully hermetic — the real ``op`` binary is never invoked and no
    +1Password integration is enabled.
    +"""
    +
    +from __future__ import annotations
    +
    +import os
    +import sys
    +from pathlib import Path
    +from unittest import mock
    +
    +import pytest
    +
    +# Make the worktree importable without depending on the installed wheel.
    +ROOT = Path(__file__).resolve().parents[1]
    +if str(ROOT) not in sys.path:
    +    sys.path.insert(0, str(ROOT))
    +
    +from hermes_cli import env_loader  # noqa: E402
    +import agent.credential_pool as credential_pool  # noqa: E402
    +
    +
    [redacted:email](autouse=True)
    +def _isolate_op_token(monkeypatch):
    +    """Each test starts with OP_SERVICE_ACCOUNT_TOKEN unset and a clean cache."""
    +    monkeypatch.delenv("OP_SERVICE_ACCOUNT_TOKEN", raising=False)
    +    env_loader.reset_secret_source_cache()
    +    yield
    +    env_loader.reset_secret_source_cache()
    +
    +
    +# ---------------------------------------------------------------------------
    +# Patch 1 — .op.env bootstrap-token auto-load
    +# ---------------------------------------------------------------------------
    +
    +
    +def test_op_env_autoloads_bootstrap_token_in_cron_context(tmp_path, monkeypatch):
    +    """A fresh interpreter (no inherited shell state) picks up the token."""
    +    home = tmp_path / ".hermes"
    +    home.mkdir()
    +    # .env carries user secrets / op:// references but NOT the bootstrap token.
    +    (home / ".env").write_text("FOO=bar\n", encoding="utf-8")
    +    # The gitignored .op.env holds only the service-account token.
    +    (home / ".op.env").write_text(
    +        "OP_SERVICE_ACCOUNT_TOKEN=[redacted:secret-assignment]", encoding="utf-8"
    +    )
    +
    +    assert os.environ.get("OP_SERVICE_ACCOUNT_TOKEN") is None
    +
    +    env_loader.load_hermes_dotenv(hermes_home=home)
    +
    +    assert os.environ["OP_SERVICE_ACCOUNT_TOKEN"] == "test-token"
    +
    +
    +def test_op_env_does_not_override_existing_token(tmp_path, monkeypatch):
    +    """A token already in the environment (e.g. systemd EnvironmentFile) wins."""
    +    home = tmp_path / ".hermes"
    +    home.mkdir()
    +    (home / ".env").write_text("FOO=bar\n", encoding="utf-8")
    +    (home / ".op.env").write_text(
    +        "OP_SERVICE_ACCOUNT_TOKEN=[redacted:secret-assignment]", encoding="utf-8"
    +    )
    +
    +    monkeypatch.setenv("OP_SERVICE_ACCOUNT_TOKEN", "live-token")
    +
    +    env_loader.load_hermes_dotenv(hermes_home=home)
    +
    +    # override=False AND the explicit guard both protect the live token.
    +    assert os.environ["OP_SERVICE_ACCOUNT_TOKEN"] == "live-token"
    +
    +
    +def test_missing_op_env_is_a_noop(tmp_path):
    +    """No .op.env present must not raise and must not invent a token."""
    +    home = tmp_path / ".hermes"
    +    home.mkdir()
    +    (home / ".env").write_text("FOO=bar\n", encoding="utf-8")
    +
    +    env_loader.load_hermes_dotenv(hermes_home=home)
    +
    +    assert os.environ.get("OP_SERVICE_ACCOUNT_TOKEN") is None
    +
    +
    +# ---------------------------------------------------------------------------
    +# Patch 2 — credential_pool prefers resolved value over raw op:// ref
    +# ---------------------------------------------------------------------------
    +
    +
    +def _seed_openrouter_token(monkeypatch, dotenv_value, environ_value):
    +    """Drive _seed_from_env('openrouter') and return the seeded access_token.
    +
    +    _get_env_prefer_dotenv is a closure inside _seed_from_env, so we exercise
    +    it through the openrouter seeding path, which calls
    +    _get_env_prefer_dotenv('OPENROUTER_API_KEY') and stores the result as the
    +    pooled credential's access_token.
    +    """
    +    monkeypatch.setattr(
    +        credential_pool,
    +        "load_env",
    +        lambda: {"OPENROUTER_API_KEY": dotenv_value},
    +    )
    +    if environ_value is None:
    +        monkeypatch.delenv("OPENROUTER_API_KEY", raising=False)
    +    else:
    +        monkeypatch.setenv("OPENROUTER_API_KEY", environ_value)
    +    # Never treat the synthetic source as suppressed.
    +    monkeypatch.setattr(
    +        "hermes_cli.auth.is_source_suppressed", lambda _p, _s: False
    +    )
    +
    +    entries: list = []
    +    changed, sources = credential_pool._seed_from_env("openrouter", entries)
    +    assert changed and entries, "expected a seeded openrouter credential"
    +    return entries[0].access_token
    +
    +
    +def test_credential_pool_prefers_resolved_env_over_raw_op_ref(monkeypatch):
    +    """A raw op:// reference in .env must lose to the resolved os.environ value."""
    +    token = [redacted:secret-assignment]
    +        monkeypatch,
    +        dotenv_value="op://Vault/Item/field",
    +        environ_value="resolved-value",
    +    )
    +    assert token == "resolved-value"
    +
    +
    +def test_credential_pool_still_prefers_dotenv_for_non_op_values(monkeypatch):
    +    """Regression guard: .env still beats os.environ for ordinary values."""
    +    token = [redacted:secret-assignment]
    +        monkeypatch,
    +        dotenv_value="dotenv-value",
    +        environ_value="shell-value",
    +    )
    +    assert token == "dotenv-value"
    +
    +
    +def test_credential_pool_falls_back_to_env_when_dotenv_is_only_op_ref(monkeypatch):
    +    """An unresolved op:// in .env with no resolved env value yields the raw ref.
    +
    +    This is the pre-resolution / misconfigured edge: there is nothing better
    +    to return, so behaviour is unchanged (the raw reference is surfaced rather
    +    than silently dropping the credential).
    +    """
    +    token = [redacted:secret-assignment]
    +        monkeypatch,
    +        dotenv_value="op://Vault/Item/field",
    +        environ_value=None,
    +    )
    +    assert token == "op://Vault/Item/field"
    ```
  - `tests/test_env_loader_secret_sources.py`

    ```diff
    @@ -63,11 +63,21 @@ def test_format_secret_source_suffix_generic_label_for_future_sources():
         )
     
     
    +def test_format_secret_source_suffix_onepassword_uses_proper_name():
    +    env_loader._SECRET_SOURCES["OPENAI_API_KEY"] = "onepassword"
    +    assert (
    +        env_loader.format_secret_source_suffix("OPENAI_API_KEY")
    +        == " (from 1Password)"
    +    )
    +
    +
     def test_apply_external_secret_sources_records_bitwarden_origin(tmp_path, monkeypatch):
    -    """End-to-end: when ``apply_bitwarden_secrets`` returns applied keys,
    -    they end up in ``_SECRET_SOURCES`` so the UI can label them."""
    +    """End-to-end: when the Bitwarden source fetches keys, applied vars
    +    end up in ``_SECRET_SOURCES`` so the UI can label them."""
     
         monkeypatch.setenv("HERMES_HOME", str(tmp_path))
    +    monkeypatch.setenv("BWS_ACCESS_TOKEN", "0.test-token")
    +    monkeypatch.delenv("ANTHROPIC_API_KEY", raising=False)
         config_path = tmp_path / "config.yaml"
         config_path.write_text(
             "secrets:\n"
    ```

    ```diff
    @@ -78,22 +88,19 @@ def test_apply_external_secret_sources_records_bitwarden_origin(tmp_path, monkey
             encoding="utf-8",
         )
     
    -    # Stub apply_bitwarden_secrets to return a synthetic FetchResult.
    -    from agent.secret_sources.bitwarden import FetchResult
    +    # Stub the fetch layer under the SecretSource adapter.
    +    import agent.secret_sources.bitwarden as bw_module
     
    -    fake_result = FetchResult(
    -        secrets={"ANTHROPIC_API_KEY": "sk-ant-test"},
    -        applied=["ANTHROPIC_API_KEY"],
    +    monkeypatch.setattr(bw_module, "find_bws", lambda **_kw: Path("/fake/bws"))
    +    monkeypatch.setattr(
    +        bw_module,
    +        "fetch_bitwarden_secrets",
    +        lambda **_kw: ({"ANTHROPIC_API_KEY": "sk-ant-test"}, []),
         )
     
    -    def _fake_apply(**_kwargs):
    -        return fake_result
    -
    -    # The import inside _apply_external_secret_sources is lazy, so we
    -    # patch the *module attribute* it will pull in.
    -    import agent.secret_sources.bitwarden as bw_module
    +    from agent.secret_sources import registry as reg_module
     
    -    monkeypatch.setattr(bw_module, "apply_bitwarden_secrets", _fake_apply)
    +    reg_module._reset_registry_for_tests()
     
         env_loader._apply_external_secret_sources(tmp_path)
     
    ```

    ```diff
    @@ -131,6 +138,8 @@ def test_apply_external_secret_sources_dedupes_within_process(tmp_path, monkeypa
         """
     
         monkeypatch.setenv("HERMES_HOME", str(tmp_path))
    +    monkeypatch.setenv("BWS_ACCESS_TOKEN", "0.test-token")
    +    monkeypatch.delenv("ANTHROPIC_API_KEY", raising=False)
         config_path = tmp_path / "config.yaml"
         config_path.write_text(
             "secrets:\n"
    ```

    ```diff
    @@ -141,19 +150,19 @@ def test_apply_external_secret_sources_dedupes_within_process(tmp_path, monkeypa
             encoding="utf-8",
         )
     
    -    from agent.secret_sources.bitwarden import FetchResult
    -
         call_count = {"n": 0}
     
    -    def _fake_apply(**_kwargs):
    +    def _fake_fetch(**_kwargs):
             call_count["n"] += 1
    -        return FetchResult(
    -            secrets={"ANTHROPIC_API_KEY": "sk-ant-test"},
    -            applied=["ANTHROPIC_API_KEY"],
    -        )
    +        return {"ANTHROPIC_API_KEY": "sk-ant-test"}, []
     
         import agent.secret_sources.bitwarden as bw_module
    -    monkeypatch.setattr(bw_module, "apply_bitwarden_secrets", _fake_apply)
    +    monkeypatch.setattr(bw_module, "find_bws", lambda **_kw: Path("/fake/bws"))
    +    monkeypatch.setattr(bw_module, "fetch_bitwarden_secrets", _fake_fetch)
    +
    +    from agent.secret_sources import registry as reg_module
    +
    +    reg_module._reset_registry_for_tests()
     
         # Five calls in a row, simulating module-import-time invocations from
         # cli.py, hermes_cli/main.py, run_agent.py, trajectory_compressor.py,
    ```

    ```diff
    @@ -173,3 +182,95 @@ def _fake_apply(**_kwargs):
         env_loader.reset_secret_source_cache()
         env_loader._apply_external_secret_sources(tmp_path)
         assert call_count["n"] == 2
    +
    +
    +def test_apply_external_secret_sources_records_onepassword_origin(tmp_path, monkeypatch):
    +    """When the 1Password source resolves refs, applied vars end up in
    +    ``_SECRET_SOURCES`` labeled ``onepassword``."""
    +
    +    monkeypatch.setenv("HERMES_HOME", str(tmp_path))
    +    monkeypatch.delenv("ANTHROPIC_API_KEY", raising=False)
    +    (tmp_path / "config.yaml").write_text(
    +        "secrets:\n"
    +        "  onepassword:\n"
    +        "    enabled: true\n"
    +        "    env:\n"
    +        "      ANTHROPIC_API_KEY: '[redacted:secret-assignment]'\n",
    +        encoding="utf-8",
    +    )
    +
    +    import agent.secret_sources.onepassword as op_module
    +
    +    monkeypatch.setattr(op_module, "find_op", lambda *_a, **_kw: Path("/fake/op"))
    +    monkeypatch.setattr(
    +        op_module,
    +        "fetch_onepassword_secrets",
    +        lambda **_kw: ({"ANTHROPIC_API_KEY": "sk-ant-test"}, []),
    +    )
    +
    +    from agent.secret_sources import registry as reg_module
    +
    +    reg_module._reset_registry_for_tests()
    +
    +    env_loader._apply_external_secret_sources(tmp_path)
    +
    +    assert env_loader.get_secret_source("ANTHROPIC_API_KEY") == "onepassword"
    +    assert (
    +        env_loader.format_secret_source_suffix("ANTHROPIC_API_KEY")
    +        == " (from 1Password)"
    +    )
    +
    +
    +def test_apply_external_secret_sources_survives_non_dict_section(tmp_path, monkeypatch):
    +    """A malformed `secrets:` section must not abort startup (fail-open).
    +
    +    Both `onepassword: true` (non-dict) and a bad bitwarden section must be
    +    coerced to empty config instead of raising AttributeError up through
    +    load_hermes_dotenv().
    +    """
    +
    +    monkeypatch.setenv("HERMES_HOME", str(tmp_path))
    +    (tmp_path / "config.yaml").write_text(
    +        "secrets:\n"
    +        "  bitwarden: true\n"
    +        "  onepassword: [redacted:secret-assignment]",
    +        encoding="utf-8",
    +    )
    +
    +    # Must not raise and must not record anything.
    +    env_loader._apply_external_secret_sources(tmp_path)
    +    assert env_loader.get_secret_source("ANYTHING") is None
    +
    +
    +def test_apply_external_secret_sources_bad_ttl_does_not_crash(tmp_path, monkeypatch):
    +    """A non-numeric cache_ttl_seconds must be coerced, not crash startup."""
    +
    +    monkeypatch.setenv("HERMES_HOME", str(tmp_path))
    +    (tmp_path / "config.yaml").write_text(
    +        "secrets:\n"
    +        "  onepassword:\n"
    +        "    enabled: true\n"
    +        "    cache_ttl_seconds: not-a-number\n"
    +        "    env:\n"
    +        "      K: 'op://V/I/F'\n",
    +        encoding="utf-8",
    +    )
    +
    +    captured = {}
    +
    +    def _fake_fetch(**kwargs):
    +        captured.update(kwargs)
    +        return {}, []
    +
    +    import agent.secret_sources.onepassword as op_module
    +    monkeypatch.setattr(op_module, "find_op", lambda *_a, **_kw: Path("/fake/op"))
    +    monkeypatch.setattr(op_module, "fetch_onepassword_secrets", _fake_fetch)
    +
    +    from agent.secret_sources import registry as reg_module
    +
    +    reg_module._reset_registry_for_tests()
    +
    +    env_loader._apply_external_secret_sources(tmp_path)
    +
    +    # Coerced to the 300s default rather than raising ValueError.
    +    assert captured["cache_ttl_seconds"] == 300
    ```
  - `tests/test_onepassword_secrets.py`

    ```diff
    @@ -0,0 +1,484 @@
    +"""Hermetic tests for the 1Password (`op` CLI) secret source.
    +
    +We never invoke the real ``op`` binary: ``subprocess.run`` is mocked so the
    +suite stays fast and offline-safe.  A live resolve is exercised manually via
    +``hermes secrets onepassword sync`` outside of pytest.
    +"""
    +
    +from __future__ import annotations
    +
    +import json
    +import os
    +import subprocess
    +import sys
    +import time
    +from pathlib import Path
    +from unittest import mock
    +
    +import pytest
    +
    +
    +# Make the worktree importable without depending on the installed wheel.
    +ROOT = Path(__file__).resolve().parents[1]
    +if str(ROOT) not in sys.path:
    +    sys.path.insert(0, str(ROOT))
    +
    +from agent.secret_sources import onepassword as op  # noqa: E402
    +
    +
    [redacted:email](autouse=True)
    +def _reset_caches():
    +    op._reset_cache_for_tests()
    +    yield
    +    op._reset_cache_for_tests()
    +
    +
    [redacted:email](autouse=True)
    +def _clean_op_env(monkeypatch):
    +    """Start every test from a known 1Password auth state."""
    +    for key in list(os.environ):
    +        if key.startswith("OP_SESSION_"):
    +            monkeypatch.delenv(key, raising=False)
    +    monkeypatch.delenv("OP_SERVICE_ACCOUNT_TOKEN", raising=False)
    +    monkeypatch.delenv("OP_ACCOUNT", raising=False)
    +    yield
    +
    +
    +def _ok(value: str):
    +    return mock.Mock(returncode=0, stdout=value, stderr="")
    +
    +
    +def _err(code: int, stderr: str):
    +    return mock.Mock(returncode=code, stdout="", stderr=stderr)
    +
    +
    +# ---------------------------------------------------------------------------
    +# Reference validation
    +# ---------------------------------------------------------------------------
    +
    +
    +def test_validate_references_filters_bad_names_and_refs():
    +    refs = {
    +        "OPENAI_API_KEY": "op://Private/OpenAI/api key",
    +        "1BAD_NAME": "op://Private/x/y",          # bad env name
    +        "HAS SPACE": "op://Private/x/y",          # bad env name
    +        "NOT_A_REF": "https://example.com",        # not op://
    +        "WHITESPACE": "  op://Private/z/field  ",  # stripped + kept
    +    }
    +    valid, warnings = op._validate_references(refs)
    +    assert valid == {
    +        "OPENAI_API_KEY": "op://Private/OpenAI/api key",
    +        "WHITESPACE": "op://Private/z/field",
    +    }
    +    assert len(warnings) == 3
    +
    +
    +# ---------------------------------------------------------------------------
    +# fetch_onepassword_secrets
    +# ---------------------------------------------------------------------------
    +
    +
    +def test_fetch_happy_path(monkeypatch, tmp_path):
    +    fake_op = tmp_path / "op"
    +    fake_op.write_text("")
    +    values = {
    +        "op://Private/OpenAI/api key": "sk-abc\n",
    +        "op://Private/Anthropic/credential": "sk-ant-xyz",
    +    }
    +
    +    def fake_run(cmd, **kwargs):
    +        # argv list, never shell=True; reference passed after `--`.
    +        assert "--" in cmd
    +        ref = cmd[cmd.index("--") + 1]
    +        return _ok(values[ref])
    +
    +    monkeypatch.setattr(op.subprocess, "run", fake_run)
    +
    +    secrets, warnings = op.fetch_onepassword_secrets(
    +        references={
    +            "OPENAI_API_KEY": "op://Private/OpenAI/api key",
    +            "ANTHROPIC_API_KEY": "op://Private/Anthropic/credential",
    +        },
    +        binary=fake_op,
    +        use_cache=False,
    +    )
    +    assert secrets == {"OPENAI_API_KEY": "sk-abc", "ANTHROPIC_API_KEY": "sk-ant-xyz"}
    +    assert warnings == []
    +
    +
    +def test_fetch_uses_option_terminator_and_account(monkeypatch, tmp_path):
    +    fake_op = tmp_path / "op"
    +    fake_op.write_text("")
    +    captured = {}
    +
    +    def fake_run(cmd, **kwargs):
    +        captured["cmd"] = cmd
    +        return _ok("value")
    +
    +    monkeypatch.setattr(op.subprocess, "run", fake_run)
    +
    +    op.fetch_onepassword_secrets(
    +        references={"K": "op://V/I/F"},
    +        account="my.1password.com",
    +        binary=fake_op,
    +        use_cache=False,
    +    )
    +    cmd = captured["cmd"]
    +    assert cmd[:2] == [str(fake_op), "read"]
    +    assert "--account" in cmd and "my.1password.com" in cmd
    +    # `--` must precede the positional reference.
    +    assert cmd[-2:] == ["--", "op://V/I/F"]
    +
    +
    +def test_fetch_empty_rc0_does_not_clobber(monkeypatch, tmp_path):
    +    """returncode 0 with empty stdout must surface as a warning, not a value."""
    +    fake_op = tmp_path / "op"
    +    fake_op.write_text("")
    +    monkeypatch.setattr(op.subprocess, "run", lambda *a, **k: _ok("   \n"))
    +
    +    secrets, warnings = op.fetch_onepassword_secrets(
    +        references={"K": "op://V/I/F"}, binary=fake_op, use_cache=False
    +    )
    +    assert secrets == {}
    +    assert any("empty value" in w for w in warnings)
    +
    +
    +def test_fetch_read_failure_becomes_warning(monkeypatch, tmp_path):
    +    fake_op = tmp_path / "op"
    +    fake_op.write_text("")
    +    monkeypatch.setattr(
    +        op.subprocess, "run", lambda *a, **k: _err(1, "\x1b[31m[ERROR] not signed in\x1b[0m")
    +    )
    +
    +    secrets, warnings = op.fetch_onepassword_secrets(
    +        references={"K": "op://V/I/F"}, binary=fake_op, use_cache=False
    +    )
    +    assert secrets == {}
    +    assert len(warnings) == 1
    +    # ANSI control sequences are fully scrubbed from the surfaced message.
    +    assert "\x1b" not in warnings[0]
    +    assert "[31m" not in warnings[0]
    +    assert "not signed in" in warnings[0]
    +
    +
    +def test_fetch_one_bad_one_good(monkeypatch, tmp_path):
    +    fake_op = tmp_path / "op"
    +    fake_op.write_text("")
    +
    +    def fake_run(cmd, **kwargs):
    +        ref = cmd[cmd.index("--") + 1]
    +        if ref == "op://V/good/f":
    +            return _ok("good-value")
    +        return _err(1, "no access")
    +
    +    monkeypatch.setattr(op.subprocess, "run", fake_run)
    +
    +    secrets, warnings = op.fetch_onepassword_secrets(
    +        references={"GOOD": "op://V/good/f", "BAD": "op://V/bad/f"},
    +        binary=fake_op,
    +        use_cache=False,
    +    )
    +    assert secrets == {"GOOD": "good-value"}
    +    assert len(warnings) == 1
    +
    +
    +def test_fetch_missing_binary_raises(monkeypatch):
    +    monkeypatch.setattr(op, "find_op", lambda binary_path="": None)
    +    with pytest.raises(RuntimeError, match="op CLI not found"):
    +        op.fetch_onepassword_secrets(
    +            references={"K": "op://V/I/F"}, use_cache=False
    +        )
    +
    +
    +def test_fetch_child_env_is_allowlisted(monkeypatch, tmp_path):
    +    """The op child must NOT inherit unrelated provider credentials."""
    +    fake_op = tmp_path / "op"
    +    fake_op.write_text("")
    +    monkeypatch.setenv("OPENAI_API_KEY", "leak-me")
    +    monkeypatch.setenv("OP_SERVICE_ACCOUNT_TOKEN", "ops_tok")
    +    monkeypatch.setenv("OP_SESSION_myacct", "sess123")
    +    captured = {}
    +
    +    def fake_run(cmd, **kwargs):
    +        captured["env"] = kwargs["env"]
    +        return _ok("v")
    +
    +    monkeypatch.setattr(op.subprocess, "run", fake_run)
    +    op.fetch_onepassword_secrets(
    +        references={"K": "op://V/I/F"}, binary=fake_op, use_cache=False
    +    )
    +    env = captured["env"]
    +    assert "OPENAI_API_KEY" not in env          # not inherited
    +    assert env["OP_SERVICE_ACCOUNT_TOKEN"] == "ops_tok"
    +    assert env["OP_SESSION_myacct"] == "sess123"
    +    assert env.get("NO_COLOR") == "1"
    +
    +
    +# ---------------------------------------------------------------------------
    +# Caching
    +# ---------------------------------------------------------------------------
    +
    +
    +def test_inprocess_cache_hit(monkeypatch, tmp_path):
    +    fake_op = tmp_path / "op"
    +    fake_op.write_text("")
    +    calls = {"n": 0}
    +
    +    def fake_run(*a, **k):
    +        calls["n"] += 1
    +        return _ok("v")
    +
    +    monkeypatch.setattr(op.subprocess, "run", fake_run)
    +    op._reset_cache_for_tests(tmp_path)
    +    for _ in range(2):
    +        op.fetch_onepassword_secrets(
    +            references={"K": "op://V/I/F"}, cache_ttl_seconds=60,
    +            binary=fake_op, home_path=tmp_path,
    +        )
    +    assert calls["n"] == 1  # second call served from L1 cache
    +
    +
    +def test_disk_cache_roundtrip_and_no_token_on_disk(monkeypatch, tmp_path):
    +    fake_op = tmp_path / "op"
    +    fake_op.write_text("")
    +    monkeypatch.setenv("OP_SERVICE_ACCOUNT_TOKEN", "ops_supersecret")
    +    calls = {"n": 0}
    +
    +    def fake_run(*a, **k):
    +        calls["n"] += 1
    +        return _ok("resolved")
    +
    +    monkeypatch.setattr(op.subprocess, "run", fake_run)
    +    op._reset_cache_for_tests(tmp_path)
    +
    +    op.fetch_onepassword_secrets(
    +        references={"K": "op://V/I/F"}, cache_ttl_seconds=300,
    +        binary=fake_op, home_path=tmp_path,
    +    )
    +    assert calls["n"] == 1
    +
    +    cache_path = op._disk_cache_path(tmp_path)
    +    assert cache_path.exists()
    +    assert (os.stat(cache_path).st_mode & 0o777) == 0o600
    +    text = cache_path.read_text()
    +    assert "ops_supersecret" not in text            # token never on disk
    +    payload = json.loads(text)
    +    assert payload["secrets"] == {"K": "resolved"}
    +
    +    # Simulate a fresh process: clear only the in-process cache.
    +    op._CACHE.clear()
    +    op.fetch_onepassword_secrets(
    +        references={"K": "op://V/I/F"}, cache_ttl_seconds=300,
    +        binary=fake_op, home_path=tmp_path,
    +    )
    +    assert calls["n"] == 1  # served from disk, op not re-invoked
    +
    +
    +def test_ttl_zero_disables_both_layers(monkeypatch, tmp_path):
    +    fake_op = tmp_path / "op"
    +    fake_op.write_text("")
    +    calls = {"n": 0}
    +
    +    def fake_run(*a, **k):
    +        calls["n"] += 1
    +        return _ok("v")
    +
    +    monkeypatch.setattr(op.subprocess, "run", fake_run)
    +    op._reset_cache_for_tests(tmp_path)
    +
    +    op.fetch_onepassword_secrets(
    +        references={"K": "op://V/I/F"}, cache_ttl_seconds=0,
    +        binary=fake_op, home_path=tmp_path,
    +    )
    +    # No disk file written when TTL is 0.
    +    assert not op._disk_cache_path(tmp_path).exists()
    +    op._CACHE.clear()
    +    op.fetch_onepassword_secrets(
    +        references={"K": "op://V/I/F"}, cache_ttl_seconds=0,
    +        binary=fake_op, home_path=tmp_path,
    +    )
    +    assert calls["n"] == 2  # never cached
    +
    +
    +def test_session_change_invalidates_cache(monkeypatch, tmp_path):
    +    """A different OP_SESSION_* identity must not reuse a cached value."""
    +    fake_op = tmp_path / "op"
    +    fake_op.write_text("")
    +    calls = {"n": 0}
    +
    +    def fake_run(*a, **k):
    +        calls["n"] += 1
    +        return _ok("v")
    +
    +    monkeypatch.setattr(op.subprocess, "run", fake_run)
    +    op._reset_cache_for_tests(tmp_path)
    +
    +    monkeypatch.setenv("OP_SESSION_acctA", "sessA")
    +    op.fetch_onepassword_secrets(
    +        references={"K": "op://V/I/F"}, cache_ttl_seconds=300,
    +        binary=fake_op, home_path=tmp_path,
    +    )
    +    # Switch identity.
    +    monkeypatch.delenv("OP_SESSION_acctA", raising=False)
    +    monkeypatch.setenv("OP_SESSION_acctB", "sessB")
    +    op._CACHE.clear()
    +    op.fetch_onepassword_secrets(
    +        references={"K": "op://V/I/F"}, cache_ttl_seconds=300,
    +        binary=fake_op, home_path=tmp_path,
    +    )
    +    assert calls["n"] == 2  # cache key changed → refetch
    +
    +
    +def test_partial_failure_not_cached(monkeypatch, tmp_path):
    +    fake_op = tmp_path / "op"
    +    fake_op.write_text("")
    +
    +    def fake_run(cmd, **kwargs):
    +        ref = cmd[cmd.index("--") + 1]
    +        return _ok("v") if ref == "op://V/good/f" else _err(1, "fail")
    +
    +    monkeypatch.setattr(op.subprocess, "run", fake_run)
    +    op._reset_cache_for_tests(tmp_path)
    +    op.fetch_onepassword_secrets(
    +        references={"G": "op://V/good/f", "B": "op://V/bad/f"},
    +        cache_ttl_seconds=300, binary=fake_op, home_path=tmp_path,
    +    )
    +    # A pull with any read error must not be persisted.
    +    assert not op._disk_cache_path(tmp_path).exists()
    +
    +
    +def test_reset_cache_clears_disk(tmp_path):
    +    cache_path = op._disk_cache_path(tmp_path)
    +    cache_path.parent.mkdir(parents=True, exist_ok=True)
    +    cache_path.write_text("{}")
    +    assert cache_path.exists()
    +    op._reset_cache_for_tests(tmp_path)
    +    assert not cache_path.exists()
    +    op._reset_cache_for_tests(tmp_path)  # idempotent
    +
    +
    +# ---------------------------------------------------------------------------
    +# find_op
    +# ---------------------------------------------------------------------------
    +
    +
    +def test_find_op_pinned_path_not_on_path(tmp_path, monkeypatch):
    +    pinned = tmp_path / "op"
    +    pinned.write_text("")
    +    pinned.chmod(0o755)
    +    # PATH lookup must NOT be consulted when a binary_path is pinned.
    +    monkeypatch.setattr(op.shutil, "which", lambda name: "/usr/bin/op")
    +    assert op.find_op(str(pinned)) == pinned
    +
    +
    +def test_find_op_pinned_missing_returns_none(tmp_path, monkeypatch):
    +    monkeypatch.setattr(op.shutil, "which", lambda name: "/usr/bin/op")
    +    assert op.find_op(str(tmp_path / "nope")) is None
    +
    +
    +# ---------------------------------------------------------------------------
    +# apply_onepassword_secrets
    +# ---------------------------------------------------------------------------
    +
    +
    +def test_apply_disabled_returns_empty():
    +    result = op.apply_onepassword_secrets(enabled=False, env={"K": "op://V/I/F"})
    +    assert result.ok
    +    assert not result.applied
    +
    +
    +def test_apply_missing_binary_sets_error(monkeypatch):
    +    monkeypatch.setattr(op, "find_op", lambda binary_path="": None)
    +    result = op.apply_onepassword_secrets(
    +        enabled=True, env={"K": "op://V/I/F"}
    +    )
    +    assert not result.ok
    +    assert "op CLI" in result.error
    +
    +
    +def test_apply_sets_env(monkeypatch, tmp_path):
    +    fake_op = tmp_path / "op"
    +    fake_op.write_text("")
    +    monkeypatch.setattr(op, "find_op", lambda binary_path="": fake_op)
    +    monkeypatch.setattr(op.subprocess, "run", lambda *a, **k: _ok("resolved-val"))
    +    monkeypatch.delenv("MY_OP_KEY", raising=False)
    +
    +    result = op.apply_onepassword_secrets(
    +        enabled=True, env={"MY_OP_KEY": "op://V/I/F"}, cache_ttl_seconds=0,
    +    )
    +    assert result.ok
    +    assert result.applied == ["MY_OP_KEY"]
    +    assert os.environ["MY_OP_KEY"] == "resolved-val"
    +
    +
    +def test_apply_skips_before_fetch_when_not_overriding(monkeypatch, tmp_path):
    +    fake_op = tmp_path / "op"
    +    fake_op.write_text("")
    +    monkeypatch.setattr(op, "find_op", lambda binary_path="": fake_op)
    +    monkeypatch.setenv("MY_OP_KEY", "from-env")
    +    calls = {"n": 0}
    +
    +    def fake_run(*a, **k):
    +        calls["n"] += 1
    +        return _ok("from-1password")
    +
    +    monkeypatch.setattr(op.subprocess, "run", fake_run)
    +
    +    result = op.apply_onepassword_secrets(
    +        enabled=True, env={"MY_OP_KEY": "op://V/I/F"},
    +        override_existing=False, cache_ttl_seconds=0,
    +    )
    +    assert "MY_OP_KEY" in result.skipped
    +    assert os.environ["MY_OP_KEY"] == "from-env"
    +    assert calls["n"] == 0  # never even called op for a value we'd discard
    +
    +
    +def test_apply_never_overrides_token_var(monkeypatch, tmp_path):
    +    fake_op = tmp_path / "op"
    +    fake_op.write_text("")
    +    monkeypatch.setattr(op, "find_op", lambda binary_path="": fake_op)
    +    monkeypatch.setenv("OP_SERVICE_ACCOUNT_TOKEN", "original")
    +    calls = {"n": 0}
    +
    +    def fake_run(*a, **k):
    +        calls["n"] += 1
    +        return _ok("malicious")
    +
    +    monkeypatch.setattr(op.subprocess, "run", fake_run)
    +
    +    result = op.apply_onepassword_secrets(
    +        enabled=True,
    +        env={"OP_SERVICE_ACCOUNT_TOKEN": "op://V/I/F"},
    +        override_existing=True, cache_ttl_seconds=0,
    +    )
    +    assert "OP_SERVICE_ACCOUNT_TOKEN" in result.skipped
    +    assert os.environ["OP_SERVICE_ACCOUNT_TOKEN"] == "original"
    +    assert calls["n"] == 0
    +
    +
    +def test_apply_never_raises_on_read_failure(monkeypatch, tmp_path):
    +    fake_op = tmp_path / "op"
    +    fake_op.write_text("")
    +    monkeypatch.setattr(op, "find_op", lambda binary_path="": fake_op)
    +    monkeypatch.setattr(op.subprocess, "run", lambda *a, **k: _err(1, "locked"))
    +    monkeypatch.delenv("MY_OP_KEY", raising=False)
    +
    +    result = op.apply_onepassword_secrets(
    +        enabled=True, env={"MY_OP_KEY": "op://V/I/F"}, cache_ttl_seconds=0,
    +    )
    +    # Fail-open: warnings, nothing applied, no fatal error, no exception.
    +    assert result.ok
    +    assert result.applied == []
    +    assert result.warnings
    +
    +
    +def test_apply_no_valid_refs_is_noop(monkeypatch):
    +    # find_op must never be reached when there's nothing to fetch.
    +    monkeypatch.setattr(
    +        op, "find_op",
    +        lambda binary_path="": (_ for _ in ()).throw(AssertionError("should not resolve op")),
    +    )
    +    result = op.apply_onepassword_secrets(enabled=True, env={"BAD NAME": "op://V/I/F"})
    +    assert result.ok
    +    assert result.applied == []
    +    assert result.warnings  # the bad mapping warned
    ```
  - `website/docs/user-guide/secrets/index.md`

    ```diff
    @@ -5,5 +5,30 @@ Hermes can pull API keys from external secret managers at process startup instea
     Supported:
     
     - [Bitwarden Secrets Manager](./bitwarden) — `bws` CLI, lazy-installed, free tier works.
    +- [1Password](./onepassword) — `op://` references via the official `op` CLI; service-account or desktop session auth.
     
    -More backends (Vault, AWS Secrets Manager, 1Password CLI) are easy to add behind the same interface — the lift is one module in `agent/secret_sources/` and one CLI handler. File a request if you have a specific one in mind.
    +## Multiple sources at once
    +
    +You can enable more than one secret source at the same time — for example a team Bitwarden project alongside a personal vault plugin. Sources compose per env var with a deterministic precedence ladder:
    +
    +1. **Your `.env` / shell wins by default.** A source only replaces a pre-existing value when its own `override_existing: true` is set (Bitwarden defaults to true so central rotation works).
    +2. **Mapped sources beat bulk sources.** A source where you explicitly bind env vars to references (an `env:` map) outranks a source that injects a whole project of secrets implicitly, regardless of ordering.
    +3. **First source wins.** Within the same shape, the order of the optional `secrets.sources` list (or registration order) decides. Later claims on an already-claimed var are skipped — with a startup warning, never silently.
    +
    +`override_existing` never lets one source overwrite a var another source already claimed, and no source can ever overwrite another source's bootstrap token (e.g. `BWS_ACCESS_TOKEN`).
    +
    +```yaml
    +secrets:
    +  sources: [bitwarden]     # optional explicit ordering
    +  bitwarden:
    +    enabled: true
    +    project_id: "..."
    +```
    +
    +Every credential injected by a source is labelled with its origin — setup flows and `hermes model` show `(from Bitwarden)` next to detected keys so you always know where a value came from.
    +
    +## Adding your own backend
    +
    +Third-party secret managers ship as standalone plugins, not core PRs. A backend subclasses `agent.secret_sources.base.SecretSource` (one required method: `fetch(cfg, home_path) -> FetchResult`) and registers via `ctx.register_secret_source(MySource())` in the plugin's `register(ctx)`. The orchestrator owns precedence, conflict handling, timeouts, and provenance — your source only fetches. Contract rules: `fetch()` never raises, never prompts, and returns within its timeout budget; validate your implementation against the conformance kit in `tests/secret_sources/conformance.py`.
    +
    +The bundled set is deliberately closed (same policy as memory providers): Bitwarden and 1Password ship in-tree. Everything else — Infisical, Proton Pass, HashiCorp Vault, AWS Secrets Manager, OS keystores — belongs in plugin repos; share them in the Nous Research Discord (`#plugins-skills-and-skins`).
    ```
  - `website/docs/user-guide/secrets/onepassword.md`

    ```diff
    @@ -0,0 +1,166 @@
    +# 1Password
    +
    +Resolve provider API keys from [1Password](https://1password.com/) at process startup instead of storing them in plaintext inside `~/.hermes/.env`. You keep your keys as 1Password items and reference them by `op://vault/item/field`; rotating a credential becomes a single change in 1Password.
    +
    +## How it works
    +
    +1. You install the official [1Password CLI](https://developer.1password.com/docs/cli/get-started/) (`op`) and authenticate it — either with a **service-account token** (headless servers) or an **interactive/desktop session** (your laptop).
    +2. You map environment-variable names to `op://` references in `~/.hermes/config.yaml`.
    +3. Every time `hermes` (or the gateway, or a cron job) starts, after `~/.hermes/.env` has loaded, Hermes runs `op read` for each reference and sets the resolved values into `os.environ`.
    +4. By default Hermes **overrides** values already in your environment, so 1Password is the source of truth — rotate a credential once and every Hermes process picks it up on next start. Flip `override_existing: false` if you want `.env` to win instead.
    +
    +Hermes never authenticates on your behalf and never downloads `op`: it shells out to your already-installed, already-trusted CLI. If `op` is missing, your session is locked, or a reference is wrong, Hermes prints a one-line warning and continues with whatever credentials `.env` already had — it never blocks startup.
    +
    +## Authentication
    +
    +`op` supports two non-interactive-friendly modes; Hermes works with either:
    +
    +- **Service accounts** (recommended for servers/CI): create a service account in 1Password, grant it read access to the relevant vault, and export its token as `OP_SERVICE_ACCOUNT_TOKEN` in `~/.hermes/.env`. The token is the credential — treat it like any other bearer token.
    +- **Desktop / interactive sessions** (laptops): run `op signin` (or enable CLI integration in the 1Password app). Hermes passes your `OP_SESSION_*` variables through to the `op` child process. The 1Password cache key includes those session variables, so signing into a different account never serves a value cached under the previous identity.
    +
    +## Bootstrap token
    +
    +When you authenticate with a **service-account token**, that token is itself the bootstrap credential Hermes needs *before* it can resolve any `op://` reference. It must be present in `os.environ` of every process that resolves secrets — including cron jobs (`kanban.dispatch_in_gateway: false`), subprocess invocations, CLI runs, macOS launchd agents, and Docker containers — not just the interactive gateway. There are three ways to make it available, in order of precedence:
    +
    +1. **In `~/.hermes/.env` (recommended).** `hermes secrets onepassword setup --token <token>` writes the token to `~/.hermes/.env`, exactly like Bitwarden's `BWS_ACCESS_TOKEN`. Because `load_hermes_dotenv()` always loads `.env`, the token is available everywhere with zero extra setup. This is the simplest reliable option.
    +
    +2. **In `~/.hermes/.op.env` (gitignored).** If you'd rather keep the service-account token out of `.env` — for example so `.env` can be checked into a private dotfiles repo while the token stays out of version control — place it in `~/.hermes/.op.env`:
    +
    +   ```bash
    +   echo 'OP_SERVICE_ACCOUNT_TOKEN=[redacted:secret-assignment]' > ~/.hermes/.op.env
    +   chmod 600 ~/.hermes/.op.env
    +   ```
    +
    +   Hermes auto-loads `.op.env` at startup, **after** `.env`, and **never** overrides a token already present in the environment. `.op.env` is gitignored so the token never enters a committed file.
    +
    +3. **Via systemd `EnvironmentFile` (Linux gateway).** If you run the gateway under systemd, you can inject the token directly into the service environment:
    +
    +   ```ini
    +   [Service]
    +   EnvironmentFile=-/home/youruser/.hermes/.op.env
    +   ```
    +
    +   A token injected this way takes precedence — Hermes detects that `OP_SERVICE_ACCOUNT_TOKEN` is already set and skips loading `.op.env` entirely.
    +
    +If the token is reachable only through an interactive shell (`op signin`, `OP_SESSION_*` exports in `.bashrc`, etc.), it will **not** be inherited by cron jobs or freshly spawned subprocesses, and those contexts will log a warning and fall back to whatever credentials `.env` already held. Use one of the three options above for any non-interactive workload.
    +
    +## Setup
    +
    +### 1. Install and sign in to `op`
    +
    +Follow the [1Password CLI getting-started guide](https://developer.1password.com/docs/cli/get-started/). Verify it works:
    +
    +```bash
    +op whoami
    +```
    +
    +### 2. Enable the integration
    +
    +```bash
    +hermes secrets onepassword setup
    +```
    +
    +This verifies `op` is on `PATH` (or use `--binary-path`), records your account/token settings, checks for an active session, and flips `secrets.onepassword.enabled: true`. Non-interactive flags:
    +
    +```bash
    +hermes secrets onepassword setup \
    +  --account my.1password.com \
    +  --token-env OP_SERVICE_ACCOUNT_TOKEN \
    +  --token "$OP_SERVICE_ACCOUNT_TOKEN"
    +```
    +
    +### 3. Map your credentials
    +
    +The reference format is `op://<vault>/<item>/<field>`:
    +
    +```bash
    +hermes secrets onepassword set OPENAI_API_KEY    "op://Private/OpenAI/api key"
    +hermes secrets onepassword set ANTHROPIC_API_KEY "op://Private/Anthropic/credential"
    +```
    +
    +### 4. Preview and confirm
    +
    +```bash
    +hermes secrets onepassword sync     # dry-run: resolve now, show what would apply
    +hermes secrets onepassword status   # config + binary + references + auth
    +```
    +
    +From now on, every `hermes` invocation resolves the references at startup. You'll see a one-line summary in stderr the first time secrets are applied in a process.
    +
    +## CLI
    +
    +| Command | What it does |
    +|---|---|
    +| `hermes secrets onepassword setup` | Verify `op`, set account / token env var, enable |
    +| `hermes secrets onepassword status` | Show config, binary, auth, and configured references |
    +| `hermes secrets onepassword set ENV_VAR "op://…"` | Map an env var to a reference (stored stripped + validated) |
    +| `hermes secrets onepassword remove ENV_VAR` | Drop a mapping |
    +| `hermes secrets onepassword sync` | Dry-run: resolve references now and show what would apply |
    +| `hermes secrets onepassword sync --apply` | Resolve and export into the current shell's environment |
    +| `hermes secrets onepassword disable` | Flip `enabled: false`; leaves mappings in place |
    +
    +`op` and `1password` are accepted as aliases for `onepassword`.
    +
    +## Configuration
    +
    +Defaults in `~/.hermes/config.yaml`:
    +
    +```yaml
    +secrets:
    +  onepassword:
    +    enabled: false
    +    env:
    +      OPENAI_API_KEY: "[redacted:secret-assignment] key"
    +      ANTHROPIC_API_KEY: "[redacted:secret-assignment]"
    +    account: ""
    +    service_account_token_env: [redacted:secret-assignment]
    +    binary_path: ""
    +    cache_ttl_seconds: 300
    +    override_existing: true
    +```
    +
    +| Key | Default | What it does |
    +|---|---|---|
    +| `enabled` | `false` | Master switch. When false, `op` is never invoked. |
    +| `env` | `{}` | Mapping of env-var name → `op://vault/item/field` reference. Entries whose name isn't a valid env-var name, or whose value isn't an `op://` reference, are skipped with a warning. |
    +| `account` | `""` | Account shorthand / sign-in address passed as `op read --account`. Empty uses `op`'s default account. |
    +| `service_account_token_env` | `OP_SERVICE_ACCOUNT_TOKEN` | Env var Hermes reads the service-account token from. Its value is exported to the `op` child as `OP_SERVICE_ACCOUNT_TOKEN` (the name `op` expects). Leave the var unset to use a desktop/interactive session. |
    +| `binary_path` | `""` | Absolute path to `op`. When set, it is used verbatim and `PATH` is **not** consulted — pin this to avoid trusting whatever `op` appears first on `PATH`. |
    +| `cache_ttl_seconds` | `300` | How long resolved values are reused (in-process and on disk). Set to `0` to disable **both** cache layers — no values are written to disk at all. |
    +| `override_existing` | `true` | When true, resolved values overwrite anything already in env (so rotation takes effect). Flip to `false` to let `.env` / shell exports win; those references are then skipped *before* `op` is invoked. |
    +
    +## Failure modes
    +
    +1Password never blocks Hermes startup. If anything goes wrong you'll see a one-line warning in stderr and Hermes continues:
    +
    +| Symptom | Cause | Fix |
    +|---|---|---|
    +| `the op CLI was not found on PATH` | `op` not installed / not on PATH | Install the CLI, or set `secrets.onepassword.binary_path` |
    +| `op read failed for 'op://…': …` | Locked session, expired token, or no vault access | `op signin`, refresh the token, or grant the service account access |
    +| `op read returned an empty value for 'op://…'` | The referenced field exists but is empty | Fix the item/field in 1Password (an empty value is never applied — your existing env var is left intact) |
    +| `… is not an op:// secret reference` | A mapping value isn't an `op://` reference | Re-set it with the correct `op://vault/item/field` form |
    +| `op read timed out` | Network blocked or 1Password slow | Check connectivity / the desktop app integration |
    +
    +## Caching
    +
    +Successful, complete pulls are cached in-process and on disk under `<hermes_home>/cache/op_cache.json` (written atomically, mode `0600`), so back-to-back short-lived `hermes` invocations don't re-shell `op` for every reference. The cache:
    +
    +- stores only resolved secret **values** — never the service-account token or any raw auth material (auth is fingerprinted into the cache key);
    +- is invalidated when the token, account, `OP_SESSION_*` variables, or the set of references change;
    +- is **not** written when a pull had any per-reference error, so a transient auth failure isn't frozen in for the TTL;
    +- is fully disabled — reads *and* writes — when `cache_ttl_seconds: 0`.
    +
    +## Security notes
    +
    +- A 1Password service-account token can read every secret the account has access to. Store it in `~/.hermes/.env` (not `config.yaml`), and revoke + regenerate from 1Password if it leaks.
    +- Hermes refuses to let a resolved value overwrite the token env var itself, even with `override_existing: true`.
    +- The `op` child process gets a minimal allowlisted environment (auth/session vars + `PATH`/`HOME`), not a copy of the full `os.environ`, so post-dotenv provider credentials aren't all inherited by the child.
    +- References are validated to start with `op://`, and the reference is passed after a `--` option terminator so a crafted value can't be parsed as an `op` flag.
    +
    +## When NOT to use this
    +
    +- **Single-machine personal setups** where `~/.hermes/.env` is fine.
    +- **Air-gapped environments** that can't reach 1Password.
    +- **CI/CD** where an existing secrets-injection mechanism is already wired up — pick one path, not two.
    +
    +The good case for this is multi-machine fleets, shared dev boxes, gateway VPSes, or anywhere you want centralized rotation and revocation across multiple Hermes installations.
    ```
- **#32254: feat(secrets): add 1Password secret source**. closed without merge
  - maintainer direction: category `implemented_elsewhere`; implication `superseded`; refs `#59498`, `#36896`
- **#36896: feat(secrets): add 1Password (op://) secret source**. closed without merge
  - maintainer direction: category `merged_via`; implication `superseded`; refs `#59498`
- **#39445: feat(secrets): add 1Password SDK-based secret source (onepassword-sdk)**. closed without merge
  - maintainer direction: category `implemented_elsewhere`; implication `superseded`; refs `#59498`, `#36896`, `#32254`

## Fix status

- Merged to main in PR #59498 (cited above). No release data is tracked. Verify downstream.
- Fix last verified (merged) `2026-07-06`.
- Subsystem: `agent`