feat(secrets): adapt 1Password onto the SecretSource interface

Follow-up on the cherry-picked #36896 commits, wiring 1Password into
the new registry as the reference *mapped* source:

- OnePasswordSource adapter (shape=mapped, scheme=op): fetch-only —
  precedence, override semantics, conflict warnings, and env writes
  move to the orchestrator; apply_onepassword_secrets kept as legacy
  shim like Bitwarden's.
- Registered in _ensure_builtin_sources; mapped op:// bindings now
  outrank bulk Bitwarden project dumps on contested vars.
- _cache.py FetchResult/is_valid_env_name re-exported from base so
  there is exactly one canonical definition; bitwarden.py re-adapted
  onto the contributor's DiskCache substrate.
- ErrorKind classification for op failures (auth/binary/empty/network).
- Registry + conformance coverage for OnePasswordSource, incl. the
  headline multi-source test: both vaults claim the same var, mapped
  1Password wins, conflict surfaced, provenance correct.
- env_loader tests migrated off the legacy apply_* mocks onto the
  fetch layer; AUTHOR_MAP entry for @hwrdprkns.

tests/secret_sources/test_secret_source_registry.py

@@ -466,3 +466,135 @@ class TestBitwardenConformance(SecretSourceConformance):
         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_env_loader_secret_sources.py

@@ -185,10 +185,11 @@ def test_apply_external_secret_sources_dedupes_within_process(tmp_path, monkeypa
 
 
 def test_apply_external_secret_sources_records_onepassword_origin(tmp_path, monkeypatch):
-    """When ``apply_onepassword_secrets`` returns applied keys, they end up in
+    """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"
@@ -198,16 +199,18 @@ def test_apply_external_secret_sources_records_onepassword_origin(tmp_path, monk
         encoding="utf-8",
     )
 
-    from agent.secret_sources.onepassword import FetchResult
-
-    def _fake_apply(**_kwargs):
-        return FetchResult(
-            secrets={"ANTHROPIC_API_KEY": "sk-ant-test"},
-            applied=["ANTHROPIC_API_KEY"],
-        )
-
     import agent.secret_sources.onepassword as op_module
-    monkeypatch.setattr(op_module, "apply_onepassword_secrets", _fake_apply)
+
+    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)
 
@@ -255,14 +258,17 @@ def test_apply_external_secret_sources_bad_ttl_does_not_crash(tmp_path, monkeypa
 
     captured = {}
 
-    from agent.secret_sources.onepassword import FetchResult
-
-    def _fake_apply(**kwargs):
+    def _fake_fetch(**kwargs):
         captured.update(kwargs)
-        return FetchResult()
+        return {}, []
 
     import agent.secret_sources.onepassword as op_module
-    monkeypatch.setattr(op_module, "apply_onepassword_secrets", _fake_apply)
+    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)