"""Tests for the centralized credentials manager."""

import base64
import stat
from pathlib import Path
from typing import Any

import pytest

import mindroom.constants as constants_mod
import mindroom.credentials as credentials_module
from mindroom.api.credentials_target import RequestCredentialsTarget
from mindroom.api.integrations import _save_spotify_credentials
from mindroom.credentials import (
    CredentialsManager,
    _merge_credential_layers,
    _reset_credentials_manager_cache,
    get_runtime_credentials_manager,
    load_scoped_credentials,
    save_scoped_credentials,
    sync_shared_credentials_to_worker,
)
from mindroom.runtime_env_policy import (
    CREDENTIALS_ENCRYPTION_KEY_ENV,
    SANDBOX_RUNTIME_ENV_BY_KEY,
    SHARED_CREDENTIALS_PATH_ENV,
)
from mindroom.tool_system.worker_routing import ResolvedWorkerTarget, ToolExecutionIdentity, resolve_worker_target


def _test_encryption_key() -> str:
    return base64.urlsafe_b64encode(b"0" * 32).decode("ascii")


@pytest.fixture
def temp_credentials_dir(tmp_path: Path) -> Path:
    """Create a temporary directory for testing credentials."""
    creds_dir = tmp_path / "test_credentials"
    creds_dir.mkdir(parents=True, exist_ok=True)
    return creds_dir


@pytest.fixture
def credentials_manager(temp_credentials_dir: Path) -> CredentialsManager:
    """Create a CredentialsManager instance with a temporary directory."""
    return CredentialsManager(base_path=temp_credentials_dir)


def _worker_target(
    worker_scope: str | None,
    routing_agent_name: str | None,
    execution_identity: ToolExecutionIdentity | None,
    *,
    tenant_id: str | None = None,
    account_id: str | None = None,
) -> ResolvedWorkerTarget:
    return resolve_worker_target(
        worker_scope,
        routing_agent_name,
        execution_identity,
        tenant_id=tenant_id,
        account_id=account_id,
    )


class TestCredentialsManager:
    """Test suite for CredentialsManager."""

    def test_initialization_explicit_runtime_path(self, tmp_path: Path) -> None:
        """Test that credentials managers use an explicitly resolved runtime root."""
        config_path = tmp_path / "config.yaml"
        config_path.write_text("agents: {}\nmodels: {}\nrouter:\n  model: default\n", encoding="utf-8")
        runtime_paths = constants_mod.resolve_runtime_paths(config_path=config_path, storage_path=tmp_path)
        manager = CredentialsManager(runtime_paths.storage_root / "credentials")
        assert manager.base_path == tmp_path / "credentials"
        assert manager.base_path.exists()

    def test_initialization_custom_path(self, temp_credentials_dir: Path) -> None:
        """Test initialization with custom path."""
        manager = CredentialsManager(base_path=temp_credentials_dir)
        assert manager.base_path == temp_credentials_dir
        assert manager.base_path.exists()

    def test_get_credentials_path(self, credentials_manager: CredentialsManager) -> None:
        """Test getting the path for a service's credentials."""
        google_path = credentials_manager.get_credentials_path("google")
        assert google_path == credentials_manager.base_path / "google_credentials.json"

        ha_path = credentials_manager.get_credentials_path("homeassistant")
        assert ha_path == credentials_manager.base_path / "homeassistant_credentials.json"

    @pytest.mark.parametrize(
        "service",
        ["", " ", "../etc", "bad/name", "bad name", "bad!name"],
    )
    def test_get_credentials_path_rejects_invalid_service_names(
        self,
        credentials_manager: CredentialsManager,
        service: str,
    ) -> None:
        """Test that invalid service names are rejected."""
        with pytest.raises(ValueError, match="Service name"):
            credentials_manager.get_credentials_path(service)

    def test_save_and_load_credentials(self, credentials_manager: CredentialsManager) -> None:
        """Test saving and loading credentials."""
        test_creds = {
            "token": "test_token_123",
            "refresh_token": "refresh_123",
            "client_id": "client_123",
            "client_secret": "secret_123",
            "scopes": ["scope1", "scope2"],
        }

        # Save credentials
        credentials_manager.save_credentials("test_service", test_creds)

        # Verify file was created
        creds_file = credentials_manager.get_credentials_path("test_service")
        assert creds_file.exists()

        # Load credentials
        loaded_creds = credentials_manager.load_credentials("test_service")
        assert loaded_creds == test_creds

    def test_plaintext_credentials_storage_is_private(self, tmp_path: Path) -> None:
        """Plaintext credential storage should only be accessible to the owning OS user."""
        credentials_dir = tmp_path / "credentials"
        credentials_dir.mkdir(mode=0o755)
        existing_path = credentials_dir / "existing_credentials.json"
        existing_path.write_text('{"token":"existing"}', encoding="utf-8")
        existing_path.chmod(0o644)

        manager = CredentialsManager(credentials_dir)
        manager.save_credentials("new", {"token": "new"})

        assert stat.S_IMODE(credentials_dir.stat().st_mode) == 0o700
        assert stat.S_IMODE(existing_path.stat().st_mode) == 0o600
        assert stat.S_IMODE(manager.get_credentials_path("new").stat().st_mode) == 0o600

    def test_credentials_hardening_permission_error_is_actionable(
        self,
        tmp_path: Path,
        monkeypatch: pytest.MonkeyPatch,
    ) -> None:
        """Permission failures should identify the path, mode, and ownership fix."""
        credentials_dir = tmp_path / "credentials"
        credentials_dir.mkdir()
        original_chmod = Path.chmod

        def deny_credentials_chmod(path: Path, mode: int) -> None:
            if path == credentials_dir:
                raise PermissionError
            original_chmod(path, mode)

        monkeypatch.setattr(Path, "chmod", deny_credentials_chmod)

        with pytest.raises(
            PermissionError,
            match=rf"{credentials_dir}.*0o700.*MindRoom OS user owns",
        ):
            CredentialsManager(credentials_dir)

    def test_primary_manager_hardens_existing_worker_credentials(self, tmp_path: Path) -> None:
        """Root initialization should secure dormant worker credential stores."""
        storage_root = tmp_path / "mindroom_data"
        worker_root = storage_root / "workers" / "worker-existing"
        worker_credentials = worker_root / "credentials"
        worker_shared_credentials = worker_root / ".shared_credentials"
        worker_credentials.mkdir(parents=True, mode=0o755)
        worker_shared_credentials.mkdir(mode=0o755)
        credentials_path = worker_credentials / "google_credentials.json"
        shared_credentials_path = worker_shared_credentials / "openai_credentials.json"
        credentials_path.write_text('{"token":"worker"}', encoding="utf-8")
        shared_credentials_path.write_text('{"api_key":"shared"}', encoding="utf-8")
        credentials_path.chmod(0o644)
        shared_credentials_path.chmod(0o644)

        CredentialsManager(storage_root / "credentials")

        assert stat.S_IMODE(worker_credentials.stat().st_mode) == 0o700
        assert stat.S_IMODE(worker_shared_credentials.stat().st_mode) == 0o700
        assert stat.S_IMODE(credentials_path.stat().st_mode) == 0o600
        assert stat.S_IMODE(shared_credentials_path.stat().st_mode) == 0o600

    def test_encrypted_save_and_load_credentials_round_trip(
        self,
        tmp_path: Path,
        monkeypatch: pytest.MonkeyPatch,
    ) -> None:
        """Encrypted credential storage should not leave JSON or token text on disk."""
        encryption_key = _test_encryption_key()
        monkeypatch.setenv("MINDROOM_CREDENTIALS_ENCRYPTION_KEY", encryption_key)
        manager = CredentialsManager(tmp_path / "credentials", encryption_key=encryption_key)
        test_creds = {
            "token": "test_token_123",
            "refresh_token": "refresh_123",
            "client_secret": "secret_123",
        }

        manager.save_credentials("oauth_service", test_creds)

        creds_file = manager.get_credentials_path("oauth_service")
        stored_bytes = creds_file.read_bytes()
        assert b"test_token_123" not in stored_bytes
        assert b"refresh_123" not in stored_bytes
        assert b'"token"' not in stored_bytes
        assert manager.load_credentials("oauth_service") == test_creds

    def test_encrypted_credentials_reject_plaintext_json(
        self,
        tmp_path: Path,
        monkeypatch: pytest.MonkeyPatch,
    ) -> None:
        """Encrypted mode should refuse to load plaintext credential JSON."""
        encryption_key = _test_encryption_key()
        monkeypatch.setenv("MINDROOM_CREDENTIALS_ENCRYPTION_KEY", encryption_key)
        manager = CredentialsManager(tmp_path / "credentials", encryption_key=encryption_key)
        creds_path = manager.get_credentials_path("oauth_service")
        creds_path.write_text('{"token":"plaintext-token"}', encoding="utf-8")

        assert manager.load_credentials("oauth_service") is None

    def test_encrypted_save_refuses_to_overwrite_unreadable_credentials(
        self,
        tmp_path: Path,
        monkeypatch: pytest.MonkeyPatch,
    ) -> None:
        """Encrypted save should fail closed when an existing credential file cannot be loaded."""
        encryption_key = _test_encryption_key()
        monkeypatch.setenv("MINDROOM_CREDENTIALS_ENCRYPTION_KEY", encryption_key)
        manager = CredentialsManager(tmp_path / "credentials", encryption_key=encryption_key)
        creds_path = manager.get_credentials_path("oauth_service")
        original_payload = b'{"api_key":"old","other":"preserve-me"}'
        creds_path.write_bytes(original_payload)

        with pytest.raises(ValueError, match="refusing to overwrite"):
            manager.save_credentials("oauth_service", {"api_key": "new"})

        assert creds_path.read_bytes() == original_payload

    def test_plaintext_save_refuses_to_overwrite_encrypted_credentials_without_key(
        self,
        tmp_path: Path,
        monkeypatch: pytest.MonkeyPatch,
    ) -> None:
        """A missing encryption key should not turn an existing encrypted file into plaintext JSON."""
        encryption_key = _test_encryption_key()
        monkeypatch.setenv("MINDROOM_CREDENTIALS_ENCRYPTION_KEY", encryption_key)
        encrypted_manager = CredentialsManager(tmp_path / "credentials", encryption_key=encryption_key)
        encrypted_manager.save_credentials("oauth_service", {"api_key": "old", "other": "preserve-me"})
        creds_path = encrypted_manager.get_credentials_path("oauth_service")
        original_payload = creds_path.read_bytes()

        monkeypatch.delenv("MINDROOM_CREDENTIALS_ENCRYPTION_KEY")
        plaintext_manager = CredentialsManager(tmp_path / "credentials")

        with pytest.raises(ValueError, match="refusing to overwrite without a key"):
            plaintext_manager.save_credentials("oauth_service", {"api_key": "new"})

        assert creds_path.read_bytes() == original_payload

    def test_encrypted_credentials_reject_plaintext_without_traceback_logging(
        self,
        tmp_path: Path,
        monkeypatch: pytest.MonkeyPatch,
    ) -> None:
        """Encrypted-mode load failures should not log traceback locals containing secrets."""

        class CapturingLogger:
            def __init__(self) -> None:
                self.warning_calls: list[tuple[str, dict[str, object]]] = []

            def warning(self, event: str, **kwargs: object) -> None:
                self.warning_calls.append((event, kwargs))

            def exception(self, event: str, **kwargs: object) -> None:
                _ = event, kwargs
                pytest.fail("encrypted credential load must not use traceback logging")

        encryption_key = _test_encryption_key()
        captured_logger = CapturingLogger()
        monkeypatch.setenv("MINDROOM_CREDENTIALS_ENCRYPTION_KEY", encryption_key)
        monkeypatch.setattr(credentials_module, "logger", captured_logger)
        manager = CredentialsManager(tmp_path / "credentials", encryption_key=encryption_key)
        creds_path = manager.get_credentials_path("oauth_service")
        creds_path.write_text('{"token":"plaintext-token"}', encoding="utf-8")

        assert manager.load_credentials("oauth_service") is None

        assert captured_logger.warning_calls == [
            (
                "Failed to load encrypted credentials",
                {
                    "service": "oauth_service",
                    "path": str(creds_path),
                    "error_type": "ValueError",
                },
            ),
        ]
        logged_payload = repr(captured_logger.warning_calls)
        assert "plaintext-token" not in logged_payload
        assert encryption_key not in logged_payload

    def test_plaintext_mode_rejects_encrypted_file_without_traceback_logging(
        self,
        tmp_path: Path,
        monkeypatch: pytest.MonkeyPatch,
    ) -> None:
        """Keyless load failures should not log traceback locals containing ciphertext."""

        class CapturingLogger:
            def __init__(self) -> None:
                self.warning_calls: list[tuple[str, dict[str, object]]] = []

            def warning(self, event: str, **kwargs: object) -> None:
                self.warning_calls.append((event, kwargs))

            def exception(self, event: str, **kwargs: object) -> None:
                _ = event, kwargs
                pytest.fail("keyless credential load must not use traceback logging")

        encryption_key = _test_encryption_key()
        monkeypatch.setenv("MINDROOM_CREDENTIALS_ENCRYPTION_KEY", encryption_key)
        encrypted_manager = CredentialsManager(tmp_path / "credentials", encryption_key=encryption_key)
        encrypted_manager.save_credentials("oauth_service", {"token": "secret-token"})
        creds_path = encrypted_manager.get_credentials_path("oauth_service")
        stored_payload = creds_path.read_text(encoding="utf-8")

        captured_logger = CapturingLogger()
        monkeypatch.delenv("MINDROOM_CREDENTIALS_ENCRYPTION_KEY")
        monkeypatch.setattr(credentials_module, "logger", captured_logger)
        plaintext_manager = CredentialsManager(tmp_path / "credentials")

        assert plaintext_manager.load_credentials("oauth_service") is None

        assert captured_logger.warning_calls == [
            (
                "Failed to load credentials",
                {
                    "service": "oauth_service",
                    "path": str(creds_path),
                    "error_type": "JSONDecodeError",
                },
            ),
        ]
        logged_payload = repr(captured_logger.warning_calls)
        assert stored_payload not in logged_payload
        assert encryption_key not in logged_payload

    def test_encrypted_credentials_file_mode_is_private(
        self,
        tmp_path: Path,
        monkeypatch: pytest.MonkeyPatch,
    ) -> None:
        """Encrypted credential files should be written with mode 0600."""
        encryption_key = _test_encryption_key()
        monkeypatch.setenv("MINDROOM_CREDENTIALS_ENCRYPTION_KEY", encryption_key)
        manager = CredentialsManager(tmp_path / "credentials", encryption_key=encryption_key)

        manager.save_credentials("oauth_service", {"token": "test-token"})

        mode = stat.S_IMODE(manager.get_credentials_path("oauth_service").stat().st_mode)
        assert mode == 0o600

    def test_encrypted_credentials_directory_mode_is_private(
        self,
        tmp_path: Path,
        monkeypatch: pytest.MonkeyPatch,
    ) -> None:
        """Encrypted credential directories should be created with mode 0700."""
        encryption_key = _test_encryption_key()
        monkeypatch.setenv("MINDROOM_CREDENTIALS_ENCRYPTION_KEY", encryption_key)

        manager = CredentialsManager(tmp_path / "credentials", encryption_key=encryption_key)

        mode = stat.S_IMODE(manager.base_path.stat().st_mode)
        assert mode == 0o700

    def test_encrypted_save_does_not_rechmod_existing_directory(
        self,
        tmp_path: Path,
        monkeypatch: pytest.MonkeyPatch,
    ) -> None:
        """Credential saves should not repeatedly chmod a directory hardened during construction."""
        encryption_key = _test_encryption_key()
        monkeypatch.setenv("MINDROOM_CREDENTIALS_ENCRYPTION_KEY", encryption_key)
        manager = CredentialsManager(tmp_path / "credentials", encryption_key=encryption_key)
        manager.base_path.chmod(0o755)

        manager.save_credentials("oauth_service", {"token": "test-token"})

        mode = stat.S_IMODE(manager.base_path.stat().st_mode)
        assert mode == 0o755

    def test_encrypted_scoped_credentials_directories_are_private(
        self,
        tmp_path: Path,
        monkeypatch: pytest.MonkeyPatch,
    ) -> None:
        """Encrypted scoped credential directories should be written with mode 0700."""
        encryption_key = _test_encryption_key()
        monkeypatch.setenv("MINDROOM_CREDENTIALS_ENCRYPTION_KEY", encryption_key)
        manager = CredentialsManager(tmp_path / "credentials", encryption_key=encryption_key)

        scoped_manager = manager.for_primary_runtime_scope("@user:example.test", "agent")

        scoped_root = tmp_path / "private_oauth"
        for directory_path in [scoped_root, scoped_manager.base_path.parent, scoped_manager.base_path]:
            mode = stat.S_IMODE(directory_path.stat().st_mode)
            assert mode == 0o700

    def test_encrypted_scoped_credentials_harden_existing_parent_directories(
        self,
        tmp_path: Path,
        monkeypatch: pytest.MonkeyPatch,
    ) -> None:
        """Encrypted scoped credentials should harden pre-existing credential-owned parents."""
        encryption_key = _test_encryption_key()
        monkeypatch.setenv("MINDROOM_CREDENTIALS_ENCRYPTION_KEY", encryption_key)
        requester_dir = credentials_module._scoped_credentials_dir_part("@user:example.test")
        scoped_root = tmp_path / "private_oauth"
        scoped_requester_path = scoped_root / requester_dir
        scoped_requester_path.mkdir(parents=True)
        scoped_root.chmod(0o755)
        scoped_requester_path.chmod(0o755)
        manager = CredentialsManager(tmp_path / "credentials", encryption_key=encryption_key)

        scoped_manager = manager.for_primary_runtime_scope("@user:example.test", "agent")

        for directory_path in [scoped_root, scoped_requester_path, scoped_manager.base_path]:
            mode = stat.S_IMODE(directory_path.stat().st_mode)
            assert mode == 0o700

    def test_encrypted_credentials_reject_corrupt_files(
        self,
        tmp_path: Path,
        monkeypatch: pytest.MonkeyPatch,
    ) -> None:
        """Encrypted mode should fail closed when ciphertext cannot be authenticated."""
        encryption_key = _test_encryption_key()
        monkeypatch.setenv("MINDROOM_CREDENTIALS_ENCRYPTION_KEY", encryption_key)
        manager = CredentialsManager(tmp_path / "credentials", encryption_key=encryption_key)
        creds_path = manager.get_credentials_path("oauth_service")
        creds_path.write_bytes(b"MINDROOM-CREDENTIALS-V1\nnot-valid-ciphertext")

        assert manager.load_credentials("oauth_service") is None

    def test_isolated_worker_runtime_loads_encrypted_shared_credentials(
        self,
        tmp_path: Path,
    ) -> None:
        """Isolated workers should keep the encryption key in RuntimePaths for credential reads."""
        encryption_key = _test_encryption_key()
        config_path = tmp_path / "config.yaml"
        config_path.write_text("agents: {}\nmodels: {}\nrouter:\n  model: default\n", encoding="utf-8")
        manager = CredentialsManager(tmp_path / "credentials", encryption_key=encryption_key)
        worker_manager = manager.for_worker("worker-a")
        worker_manager.shared_manager().save_credentials("openai", {"api_key": "shared-key", "_source": "env"})
        runtime_paths = constants_mod.resolve_runtime_paths(
            config_path=config_path,
            storage_path=worker_manager.storage_root,
            process_env={
                CREDENTIALS_ENCRYPTION_KEY_ENV: encryption_key,
                SHARED_CREDENTIALS_PATH_ENV: str(worker_manager.shared_base_path),
            },
        )

        isolated_runtime_paths = constants_mod.isolated_runtime_paths(runtime_paths)

        loaded_credentials = (
            get_runtime_credentials_manager(isolated_runtime_paths)
            .shared_manager()
            .load_credentials(
                "openai",
            )
        )
        assert loaded_credentials == {"api_key": "shared-key", "_source": "env"}

    def test_load_nonexistent_credentials(self, credentials_manager: CredentialsManager) -> None:
        """Test loading credentials that don't exist."""
        result = credentials_manager.load_credentials("nonexistent")
        assert result is None

    def test_load_corrupted_credentials(self, credentials_manager: CredentialsManager) -> None:
        """Test loading corrupted credentials file."""
        # Create a corrupted credentials file
        creds_path = credentials_manager.get_credentials_path("corrupted")
        creds_path.write_text("not valid json{")

        # Should return None on error
        result = credentials_manager.load_credentials("corrupted")
        assert result is None

    def test_delete_credentials(self, credentials_manager: CredentialsManager) -> None:
        """Test deleting credentials."""
        test_creds = {"key": "value"}

        # Save credentials
        credentials_manager.save_credentials("to_delete", test_creds)
        creds_file = credentials_manager.get_credentials_path("to_delete")
        assert creds_file.exists()

        # Delete credentials
        credentials_manager.delete_credentials("to_delete")
        assert not creds_file.exists()

        # Deleting non-existent credentials should not raise error
        credentials_manager.delete_credentials("nonexistent")

    def test_list_services(self, credentials_manager: CredentialsManager) -> None:
        """Test listing all services with stored credentials."""
        # Initially empty
        assert credentials_manager.list_services() == []

        # Add some credentials
        credentials_manager.save_credentials("google", {"token": "google_token"})
        credentials_manager.save_credentials("homeassistant", {"token": "ha_token"})
        credentials_manager.save_credentials("spotify", {"token": "spotify_token"})

        # List should be sorted
        services = credentials_manager.list_services()
        assert services == ["google", "homeassistant", "spotify"]

    def test_update_credentials(self, credentials_manager: CredentialsManager) -> None:
        """Test updating existing credentials."""
        original = {"token": "old_token", "refresh_token": "old_refresh"}
        updated = {"token": "new_token", "refresh_token": "new_refresh", "extra": "data"}

        # Save original
        credentials_manager.save_credentials("update_test", original)
        assert credentials_manager.load_credentials("update_test") == original

        # Update
        credentials_manager.save_credentials("update_test", updated)
        assert credentials_manager.load_credentials("update_test") == updated

    def test_credentials_isolation(self, credentials_manager: CredentialsManager) -> None:
        """Test that credentials for different services are isolated."""
        google_creds = {"service": "google", "token": "google_123"}
        ha_creds = {"service": "homeassistant", "token": "ha_456"}

        credentials_manager.save_credentials("google", google_creds)
        credentials_manager.save_credentials("homeassistant", ha_creds)

        # Each service should have its own credentials
        assert credentials_manager.load_credentials("google") == google_creds
        assert credentials_manager.load_credentials("homeassistant") == ha_creds

        # Deleting one shouldn't affect the other
        credentials_manager.delete_credentials("google")
        assert credentials_manager.load_credentials("google") is None
        assert credentials_manager.load_credentials("homeassistant") == ha_creds

    def test_worker_credentials_are_isolated_from_shared(self, temp_credentials_dir: Path) -> None:
        """Worker-scoped credentials should not overwrite or read from the shared credential directory."""
        manager = CredentialsManager(temp_credentials_dir)
        manager.save_credentials("openai", {"api_key": "shared-key", "_source": "ui"})

        worker_manager = manager.for_worker("worker-a")
        worker_manager.save_credentials("openai", {"api_key": "worker-key", "_source": "ui"})

        assert manager.load_credentials("openai") == {"api_key": "shared-key", "_source": "ui"}
        assert worker_manager.load_credentials("openai") == {"api_key": "worker-key", "_source": "ui"}
        assert worker_manager.get_credentials_path("openai").parent != manager.get_credentials_path("openai").parent

    def test_save_scoped_credentials_writes_to_worker_manager(self, temp_credentials_dir: Path) -> None:
        """Scoped saves should target the worker-owned credentials store."""
        manager = CredentialsManager(temp_credentials_dir)
        execution_identity = ToolExecutionIdentity(
            channel="matrix",
            agent_name="general",
            requester_id="@alice:example.org",
            room_id="!room:example.org",
            thread_id=None,
            resolved_thread_id=None,
            session_id=None,
            tenant_id="tenant-123",
            account_id="account-456",
        )

        save_scoped_credentials(
            "google",
            {"token": "worker-token", "_source": "ui"},
            credentials_manager=manager,
            worker_target=_worker_target("user", "general", execution_identity),
        )

        shared_credentials = manager.load_credentials("google")
        worker_credentials = manager.for_worker(
            "v1:tenant-123:user:@alice:example.org",
        ).load_credentials("google")

        assert shared_credentials is None
        assert worker_credentials == {"token": "worker-token", "_source": "ui"}

    @pytest.mark.parametrize("worker_scope", [None, "shared", "user", "user_agent"])
    def test_scoped_credentials_path_matches_save_and_delete_target(
        self,
        temp_credentials_dir: Path,
        worker_scope: str | None,
    ) -> None:
        """Scoped path resolution should be the write-path source of truth."""
        manager = CredentialsManager(temp_credentials_dir)
        execution_identity = ToolExecutionIdentity(
            channel="matrix",
            agent_name="general",
            requester_id="@alice:example.org",
            room_id="!room:example.org",
            thread_id=None,
            resolved_thread_id=None,
            session_id=None,
            tenant_id="tenant-123",
            account_id="account-456",
        )
        worker_target = _worker_target(worker_scope, "general", execution_identity)
        credentials_path = credentials_module.scoped_credentials_path(
            "mcp_demo_oauth",
            credentials_manager=manager,
            worker_target=worker_target,
        )

        save_scoped_credentials(
            "mcp_demo_oauth",
            {"token": "scoped-token", "_source": "oauth"},
            credentials_manager=manager,
            worker_target=worker_target,
        )

        assert credentials_path.exists()
        assert credentials_path.read_text(encoding="utf-8")

        credentials_module.delete_scoped_credentials(
            "mcp_demo_oauth",
            credentials_manager=manager,
            worker_target=worker_target,
        )

        assert not credentials_path.exists()

    def test_load_scoped_credentials_shared_scope_inherits_shared_ui_credentials(
        self,
        temp_credentials_dir: Path,
    ) -> None:
        """Shared worker scope should inherit allowlisted shared credentials regardless of source."""
        manager = CredentialsManager(temp_credentials_dir)
        execution_identity = ToolExecutionIdentity(
            channel="matrix",
            agent_name="general",
            requester_id="@alice:example.org",
            room_id="!room:example.org",
            thread_id=None,
            resolved_thread_id=None,
            session_id=None,
            tenant_id="tenant-123",
            account_id="account-456",
        )
        manager.save_credentials("google", {"api_key": "global-ui-key", "_source": "ui"})

        loaded_credentials = load_scoped_credentials(
            "google",
            credentials_manager=manager,
            worker_target=_worker_target("shared", "general", execution_identity),
            allowed_shared_services=frozenset({"google"}),
        )

        assert loaded_credentials == {"api_key": "global-ui-key", "_source": "ui"}

    def test_load_scoped_credentials_shared_scope_keeps_env_fallback(
        self,
        temp_credentials_dir: Path,
    ) -> None:
        """Shared worker scope should still inherit env-backed credentials."""
        manager = CredentialsManager(temp_credentials_dir)
        execution_identity = ToolExecutionIdentity(
            channel="matrix",
            agent_name="general",
            requester_id="@alice:example.org",
            room_id="!room:example.org",
            thread_id=None,
            resolved_thread_id=None,
            session_id=None,
            tenant_id="tenant-123",
            account_id="account-456",
        )
        manager.save_credentials("google", {"api_key": "env-key", "_source": "env"})

        loaded_credentials = load_scoped_credentials(
            "google",
            credentials_manager=manager,
            worker_target=_worker_target("shared", "general", execution_identity),
            allowed_shared_services=frozenset({"google"}),
        )

        assert loaded_credentials == {"api_key": "env-key", "_source": "env"}

    def test_load_scoped_credentials_shared_scope_blocks_non_grantable_shared_credentials(
        self,
        temp_credentials_dir: Path,
    ) -> None:
        """Shared worker scope should hide shared credentials that are not allowlisted for workers."""
        manager = CredentialsManager(temp_credentials_dir)
        execution_identity = ToolExecutionIdentity(
            channel="matrix",
            agent_name="general",
            requester_id="@alice:example.org",
            room_id="!room:example.org",
            thread_id=None,
            resolved_thread_id=None,
            session_id=None,
            tenant_id="tenant-123",
            account_id="account-456",
        )
        manager.save_credentials("google", {"api_key": "global-ui-key", "_source": "ui"})

        loaded_credentials = load_scoped_credentials(
            "google",
            credentials_manager=manager,
            worker_target=_worker_target("shared", "general", execution_identity),
            allowed_shared_services=frozenset(),
        )

        assert loaded_credentials is None

    def test_load_scoped_credentials_uses_worker_rooted_manager_without_nesting(
        self,
        temp_credentials_dir: Path,
    ) -> None:
        """Worker-rooted managers should merge their shared mirror with worker-local overrides."""
        base_manager = CredentialsManager(temp_credentials_dir)
        execution_identity = ToolExecutionIdentity(
            channel="matrix",
            agent_name="general",
            requester_id="@alice:example.org",
            room_id="!room:example.org",
            thread_id="$thread",
            resolved_thread_id="$thread",
            session_id="session-1",
            tenant_id="tenant-123",
            account_id="account-456",
        )
        worker_key = "v1:tenant-123:user:@alice:example.org"
        worker_manager = base_manager.for_worker(worker_key)
        base_manager.save_credentials("openweather", {"api_key": "shared-ui-key", "_source": "ui", "base": "yes"})
        sync_shared_credentials_to_worker(
            worker_key,
            allowed_services=frozenset({"openweather"}),
            credentials_manager=base_manager,
        )
        worker_manager.save_credentials("openweather", {"api_key": "worker-key", "_source": "ui"})

        loaded_credentials = load_scoped_credentials(
            "openweather",
            credentials_manager=worker_manager,
            worker_target=_worker_target("user", "general", execution_identity),
        )

        assert loaded_credentials == {"api_key": "worker-key", "_source": "ui", "base": "yes"}

    def test_load_scoped_credentials_shared_scope_synthesizes_worker_key_from_tenant_context(
        self,
        temp_credentials_dir: Path,
    ) -> None:
        """Shared worker scope should resolve worker credentials from explicit tenant context."""
        manager = CredentialsManager(temp_credentials_dir)
        worker_key = "v1:tenant-123:shared:general"
        manager.for_worker(worker_key).save_credentials("google", {"api_key": "worker-key", "_source": "ui"})

        loaded_credentials = load_scoped_credentials(
            "google",
            credentials_manager=manager,
            worker_target=_worker_target(
                "shared",
                "general",
                None,
                tenant_id="tenant-123",
                account_id="account-456",
            ),
        )

        assert loaded_credentials == {"api_key": "worker-key", "_source": "ui"}

    def test_shared_scope_oauth_tokens_stay_isolated_per_agent(
        self,
        temp_credentials_dir: Path,
    ) -> None:
        """OAuth tokens saved for one shared-scope agent should stay invisible to other agents."""
        manager = CredentialsManager(temp_credentials_dir)
        connecting_target = _worker_target("shared", "alpha", None, tenant_id="tenant-123", account_id="account-456")
        other_agent_target = _worker_target("shared", "beta", None, tenant_id="tenant-123", account_id="account-456")

        save_scoped_credentials(
            "google_drive_oauth",
            {"token": "alpha-token", "refresh_token": "alpha-refresh", "_source": "oauth"},
            credentials_manager=manager,
            worker_target=connecting_target,
        )

        connecting_agent_credentials = load_scoped_credentials(
            "google_drive_oauth",
            credentials_manager=manager,
            worker_target=connecting_target,
        )
        assert connecting_agent_credentials is not None
        assert connecting_agent_credentials["token"] == "alpha-token"  # noqa: S105
        assert (
            load_scoped_credentials(
                "google_drive_oauth",
                credentials_manager=manager,
                worker_target=other_agent_target,
            )
            is None
        )
        assert manager.load_credentials("google_drive_oauth") is None
        assert (
            load_scoped_credentials(
                "google_drive_oauth",
                credentials_manager=manager,
                worker_target=None,
            )
            is None
        )

    def test_shared_scope_oauth_tokens_ignore_global_store(
        self,
        temp_credentials_dir: Path,
    ) -> None:
        """Shared-scope agents should not inherit OAuth tokens from the global credentials store."""
        manager = CredentialsManager(temp_credentials_dir)
        manager.save_credentials(
            "google_drive_oauth",
            {"token": "global-token", "refresh_token": "global-refresh", "_source": "oauth"},
        )

        loaded_credentials = load_scoped_credentials(
            "google_drive_oauth",
            credentials_manager=manager,
            worker_target=_worker_target("shared", "alpha", None, tenant_id="tenant-123", account_id="account-456"),
        )

        assert loaded_credentials is None

    def test_shared_scope_oauth_save_requires_agent_name(
        self,
        temp_credentials_dir: Path,
    ) -> None:
        """Agent-scoped OAuth saves should fail loudly instead of widening to the global store."""
        manager = CredentialsManager(temp_credentials_dir)
        worker_target = ResolvedWorkerTarget(
            worker_scope="shared",
            routing_agent_name=None,
            execution_identity=None,
            tenant_id="tenant-123",
            account_id="account-456",
            worker_key=None,
        )

        with pytest.raises(ValueError, match="require an agent name"):
            save_scoped_credentials(
                "google_drive_oauth",
                {"token": "orphan-token", "_source": "oauth"},
                credentials_manager=manager,
                worker_target=worker_target,
            )

    def test_load_scoped_credentials_uses_shared_mirror_for_unscoped_worker_manager(
        self,
        temp_credentials_dir: Path,
    ) -> None:
        """Worker-rooted managers should load unscoped credentials from their mirrored shared layer."""
        base_manager = CredentialsManager(temp_credentials_dir)
        worker_key = "v1:tenant-123:unscoped:general"
        worker_manager = base_manager.for_worker(worker_key)
        base_manager.save_credentials("openai", {"api_key": "shared-ui-key", "_source": "ui"})
        sync_shared_credentials_to_worker(
            worker_key,
            allowed_services=frozenset({"openai"}),
            credentials_manager=base_manager,
        )

        loaded_credentials = load_scoped_credentials(
            "openai",
            credentials_manager=worker_manager,
            worker_target=None,
        )

        assert loaded_credentials == {"api_key": "shared-ui-key", "_source": "ui"}

    def test_save_scoped_credentials_unscoped_worker_manager_writes_local_override(
        self,
        temp_credentials_dir: Path,
    ) -> None:
        """Unscoped worker-rooted saves should create a worker-local override instead of mutating the shared mirror."""
        base_manager = CredentialsManager(temp_credentials_dir)
        worker_key = "v1:tenant-123:unscoped:general"
        worker_manager = base_manager.for_worker(worker_key)

        save_scoped_credentials(
            "google",
            {"refresh_token": "worker-refresh", "_source": "ui"},
            credentials_manager=worker_manager,
            worker_target=None,
        )

        assert worker_manager.load_credentials("google") == {
            "refresh_token": "worker-refresh",
            "_source": "ui",
        }
        assert worker_manager.shared_manager().load_credentials("google") is None

    def test_unscoped_worker_rooted_manager_keeps_local_refresh_across_resync(
        self,
        temp_credentials_dir: Path,
    ) -> None:
        """Resyncing the shared mirror should not clobber a worker-local unscoped refresh."""
        base_manager = CredentialsManager(temp_credentials_dir)
        worker_key = "v1:tenant-123:unscoped:general"
        worker_manager = base_manager.for_worker(worker_key)
        base_manager.save_credentials("google", {"client_id": "shared-client", "_source": "ui"})

        sync_shared_credentials_to_worker(
            worker_key,
            allowed_services=frozenset({"google"}),
            credentials_manager=base_manager,
        )
        save_scoped_credentials(
            "google",
            {"refresh_token": "worker-refresh", "_source": "ui"},
            credentials_manager=worker_manager,
            worker_target=None,
        )

        sync_shared_credentials_to_worker(
            worker_key,
            allowed_services=frozenset({"google"}),
            credentials_manager=base_manager,
        )

        loaded_credentials = load_scoped_credentials(
            "google",
            credentials_manager=worker_manager,
            worker_target=None,
        )

        assert loaded_credentials == {
            "client_id": "shared-client",
            "refresh_token": "worker-refresh",
            "_source": "ui",
        }

    def test_sync_shared_credentials_to_worker_copies_allowlisted_shared_credentials(
        self,
        temp_credentials_dir: Path,
    ) -> None:
        """Dedicated workers should mirror allowlisted shared credentials into their shared layer."""
        manager = CredentialsManager(temp_credentials_dir)
        manager.save_credentials("google", {"api_key": "env-key", "_source": "env"})

        sync_shared_credentials_to_worker(
            "worker-a",
            allowed_services=frozenset({"google"}),
            credentials_manager=manager,
        )

        worker_credentials = manager.for_worker("worker-a").shared_manager().load_credentials("google")
        assert worker_credentials == {"api_key": "env-key", "_source": "env"}

    def test_sync_shared_credentials_to_worker_empty_allowlist_mirrors_nothing(
        self,
        temp_credentials_dir: Path,
        monkeypatch: pytest.MonkeyPatch,
    ) -> None:
        """An empty allowlist should skip the shared store while cleaning stale mirrors."""
        manager = CredentialsManager(temp_credentials_dir)
        manager.save_credentials("google", {"api_key": "env-key", "_source": "env"})
        worker_shared_manager = manager.for_worker("worker-a").shared_manager()
        worker_shared_manager.save_credentials("google", {"api_key": "stale-key", "_source": "env"})

        def fail_shared_store_lookup(_manager: CredentialsManager) -> CredentialsManager:
            pytest.fail("empty allowlist must not open the shared credential store")

        monkeypatch.setattr(
            credentials_module,
            "_shared_credentials_manager",
            fail_shared_store_lookup,
        )

        sync_shared_credentials_to_worker(
            "worker-a",
            allowed_services=frozenset(),
            credentials_manager=manager,
        )

        assert worker_shared_manager.load_credentials("google") is None

    def test_sync_shared_credentials_to_worker_only_copies_allowed_services_and_deletes_stale_skips(
        self,
        temp_credentials_dir: Path,
    ) -> None:
        """Worker credential mirroring should enforce the allowlist and drop stale skipped services."""
        manager = CredentialsManager(temp_credentials_dir)
        manager.save_credentials("openai", {"api_key": "openai-key", "_source": "env"})
        manager.save_credentials("google", {"api_key": "google-key", "_source": "env"})
        manager.save_credentials("github_private", {"token": "github-pat", "_source": "env"})
        worker_shared_manager = manager.for_worker("worker-a").shared_manager()
        worker_shared_manager.save_credentials("github_private", {"token": "stale-pat", "_source": "env"})

        sync_shared_credentials_to_worker(
            "worker-a",
            allowed_services=frozenset({"openai", "google"}),
            credentials_manager=manager,
        )

        assert worker_shared_manager.load_credentials("openai") == {
            "api_key": "openai-key",
            "_source": "env",
        }
        assert worker_shared_manager.load_credentials("google") == {
            "api_key": "google-key",
            "_source": "env",
        }
        assert worker_shared_manager.load_credentials("github_private") is None

    def test_load_primary_runtime_scoped_credentials_do_not_inherit_shared_credentials(
        self,
        tmp_path: Path,
    ) -> None:
        """Private primary-runtime OAuth scopes must not fall back to shared deployment tokens."""
        shared_path = tmp_path / "shared"
        manager = CredentialsManager(
            base_path=tmp_path / "credentials",
            shared_base_path=shared_path,
        )
        manager.shared_manager().save_credentials(
            "google_drive_oauth",
            {
                "token": "shared-token",
                "_oauth_provider": "google_drive",
                "_source": "oauth",
            },
        )
        execution_identity = ToolExecutionIdentity(
            channel="matrix",
            agent_name="general",
            requester_id="@alice:example.org",
            room_id="!room:example.org",
            thread_id=None,
            resolved_thread_id=None,
            session_id=None,
        )

        loaded_credentials = load_scoped_credentials(
            "google_drive_oauth",
            credentials_manager=manager,
            worker_target=_worker_target("user_agent", "general", execution_identity),
        )

        assert loaded_credentials is None

    def test_sync_shared_credentials_to_worker_disallows_ui_credentials_for_non_grantable_services(
        self,
        temp_credentials_dir: Path,
    ) -> None:
        """UI-backed shared credentials should still respect the worker allowlist."""
        manager = CredentialsManager(temp_credentials_dir)
        manager.save_credentials("openai", {"api_key": "shared-ui-key", "_source": "ui"})
        manager.save_credentials("spotify", {"access_token": "spotify-ui-token", "_source": "ui"})
        worker_shared_manager = manager.for_worker("worker-a").shared_manager()

        sync_shared_credentials_to_worker(
            "worker-a",
            allowed_services=frozenset({"openai"}),
            credentials_manager=manager,
        )

        assert worker_shared_manager.load_credentials("openai") == {
            "api_key": "shared-ui-key",
            "_source": "ui",
        }
        assert worker_shared_manager.load_credentials("spotify") is None

    def test_sync_shared_credentials_to_worker_never_mirrors_oauth_client_config(
        self,
        temp_credentials_dir: Path,
    ) -> None:
        """OAuth app client secrets should stay in the primary runtime even if allowlisted."""
        manager = CredentialsManager(temp_credentials_dir)
        manager.save_credentials(
            "google_drive_oauth_client",
            {
                "client_id": "stored-client-id",
                "client_secret": "stored-client-secret",
                "_source": "ui",
            },
        )
        worker_shared_manager = manager.for_worker("worker-a").shared_manager()

        sync_shared_credentials_to_worker(
            "worker-a",
            allowed_services=frozenset({"google_drive_oauth_client"}),
            credentials_manager=manager,
        )

        assert worker_shared_manager.load_credentials("google_drive_oauth_client") is None

    def test_sync_shared_credentials_to_worker_default_denies_all_services(
        self,
        temp_credentials_dir: Path,
    ) -> None:
        """The built-in worker default should not mirror any shared credentials into workers."""
        manager = CredentialsManager(temp_credentials_dir)
        manager.save_credentials("openai", {"api_key": "openai-key", "_source": "env"})
        manager.save_credentials("github_private", {"token": "github-pat", "_source": "env"})

        sync_shared_credentials_to_worker(
            "worker-a",
            allowed_services=constants_mod.DEFAULT_WORKER_GRANTABLE_CREDENTIALS,
            credentials_manager=manager,
        )

        worker_shared_manager = manager.for_worker("worker-a").shared_manager()
        assert frozenset() == constants_mod.DEFAULT_WORKER_GRANTABLE_CREDENTIALS
        assert worker_shared_manager.load_credentials("openai") is None
        assert worker_shared_manager.load_credentials("github_private") is None

    def test_sync_shared_credentials_to_worker_default_denies_all_services_for_unscoped_workers(
        self,
        temp_credentials_dir: Path,
    ) -> None:
        """The built-in worker default should not mirror any shared credentials into unscoped workers."""
        manager = CredentialsManager(temp_credentials_dir)
        manager.save_credentials("openai", {"api_key": "shared-ui-key", "_source": "ui"})
        manager.save_credentials("github_private", {"token": "github-ui-pat", "_source": "ui"})

        sync_shared_credentials_to_worker(
            "v1:tenant-123:unscoped:general",
            allowed_services=constants_mod.DEFAULT_WORKER_GRANTABLE_CREDENTIALS,
            credentials_manager=manager,
        )

        worker_shared_manager = manager.for_worker("v1:tenant-123:unscoped:general").shared_manager()
        assert worker_shared_manager.load_credentials("openai") is None
        assert worker_shared_manager.load_credentials("github_private") is None

    def test_sync_shared_credentials_to_worker_preserves_worker_local_credentials(
        self,
        temp_credentials_dir: Path,
    ) -> None:
        """Dedicated worker seeding should not overwrite worker-local non-env credentials."""
        manager = CredentialsManager(temp_credentials_dir)
        manager.save_credentials("google", {"api_key": "env-key", "_source": "env"})
        worker_manager = manager.for_worker("worker-a")
        worker_manager.save_credentials("google", {"api_key": "worker-key", "_source": "ui"})

        sync_shared_credentials_to_worker(
            "worker-a",
            allowed_services=frozenset({"google"}),
            credentials_manager=manager,
        )

        assert worker_manager.load_credentials("google") == {"api_key": "worker-key", "_source": "ui"}
        assert worker_manager.shared_manager().load_credentials("google") == {
            "api_key": "env-key",
            "_source": "env",
        }

    def test_sync_shared_credentials_to_worker_can_copy_ui_credentials_for_unscoped_workers(
        self,
        temp_credentials_dir: Path,
    ) -> None:
        """Unscoped workers should mirror dashboard-saved shared UI credentials."""
        manager = CredentialsManager(temp_credentials_dir)
        manager.save_credentials("openai", {"api_key": "shared-ui-key", "_source": "ui"})

        sync_shared_credentials_to_worker(
            "v1:tenant-123:unscoped:general",
            allowed_services=frozenset({"openai"}),
            credentials_manager=manager,
        )

        shared_worker_credentials = (
            manager.for_worker(
                "v1:tenant-123:unscoped:general",
            )
            .shared_manager()
            .load_credentials("openai")
        )
        assert shared_worker_credentials == {"api_key": "shared-ui-key", "_source": "ui"}

    def test_sync_shared_credentials_to_worker_copies_legacy_shared_credentials_for_unscoped_workers(
        self,
        temp_credentials_dir: Path,
    ) -> None:
        """Unscoped workers should mirror legacy shared credentials that predate _source tagging."""
        manager = CredentialsManager(temp_credentials_dir)
        manager.save_credentials("spotify", {"access_token": "legacy-token"})

        sync_shared_credentials_to_worker(
            "v1:tenant-123:unscoped:general",
            allowed_services=frozenset({"spotify"}),
            credentials_manager=manager,
        )

        shared_worker_credentials = (
            manager.for_worker(
                "v1:tenant-123:unscoped:general",
            )
            .shared_manager()
            .load_credentials("spotify")
        )
        assert shared_worker_credentials == {"access_token": "legacy-token"}

    def test_merge_scoped_credentials_overlays_worker_credentials(
        self,
        temp_credentials_dir: Path,
    ) -> None:
        """Worker-scoped credentials should overlay shared credentials regardless of source."""
        manager = CredentialsManager(temp_credentials_dir)
        manager.save_credentials("google", {"api_key": "shared-ui-key", "_source": "ui", "shared_only": "yes"})
        worker_manager = manager.for_worker("worker-a")
        worker_manager.save_credentials("google", {"api_key": "worker-key", "_source": "ui"})

        merged = _merge_credential_layers(manager.load_credentials("google"), worker_manager.load_credentials("google"))

        assert merged == {"api_key": "worker-key", "_source": "ui", "shared_only": "yes"}

    def test_complex_credentials_structure(self, credentials_manager: CredentialsManager) -> None:
        """Test saving and loading complex nested credentials."""
        complex_creds: dict[str, Any] = {
            "token": "token_123",
            "nested": {
                "level1": {
                    "level2": ["item1", "item2", "item3"],
                    "data": {"key": "value"},
                },
            },
            "numbers": [1, 2, 3, 4.5],
            "boolean": True,
            "null_value": None,
        }

        credentials_manager.save_credentials("complex", complex_creds)
        loaded = credentials_manager.load_credentials("complex")
        assert loaded == complex_creds

    def test_get_api_key(self, temp_credentials_dir: Path) -> None:
        """Test getting API keys from credentials."""
        manager = CredentialsManager(temp_credentials_dir)

        # Test getting API key from simple structure
        manager.save_credentials("openai", {"api_key": "sk-test123"})
        assert manager.get_api_key("openai") == "sk-test123"

        # Test getting non-existent service
        assert manager.get_api_key("nonexistent") is None

        # Test getting custom key name
        manager.save_credentials("custom", {"token": "custom-token"})
        assert manager.get_api_key("custom", "token") == "custom-token"
        assert manager.get_api_key("custom", "api_key") is None


class TestGlobalCredentialsManager:
    """Test the global credentials manager singleton."""

    @pytest.fixture(autouse=True)
    def reset_global_manager(self) -> None:
        """Reset the global credentials manager before each test."""
        _reset_credentials_manager_cache()

    def test_get_credentials_manager_returns_same_cached_instance(self, tmp_path: Path) -> None:
        """Same storage roots should reuse the same cached manager."""
        runtime_paths = constants_mod.resolve_runtime_paths(storage_path=tmp_path)
        manager1 = get_runtime_credentials_manager(runtime_paths)
        manager2 = get_runtime_credentials_manager(runtime_paths)
        assert manager1 is manager2

    def test_cached_manager_uses_explicit_storage_root(self, tmp_path: Path) -> None:
        """The cached manager should use the provided storage root."""
        runtime_paths = constants_mod.resolve_runtime_paths(storage_path=tmp_path)
        manager = get_runtime_credentials_manager(runtime_paths)
        assert manager.base_path == tmp_path / "credentials"

    def test_global_manager_uses_explicit_shared_credentials_path(
        self,
        tmp_path: Path,
    ) -> None:
        """Dedicated workers should be able to configure a distinct shared credential mirror path."""
        config_path = tmp_path / "config.yaml"
        config_path.write_text(
            "models:\n  default:\n    provider: openai\n    id: gpt-5.4\nagents: {}\nrouter:\n  model: default\n",
            encoding="utf-8",
        )
        storage_path = (tmp_path / "worker-root").resolve()
        shared_path = storage_path / ".shared_credentials"
        runtime_paths = constants_mod.resolve_runtime_paths(
            config_path=config_path,
            storage_path=storage_path,
            process_env={SHARED_CREDENTIALS_PATH_ENV: str(shared_path)},
        )

        manager = get_runtime_credentials_manager(runtime_paths)

        assert manager.base_path == storage_path / "credentials"
        assert manager.shared_base_path == shared_path

    def test_runtime_manager_does_not_fall_back_to_process_env_key(
        self,
        tmp_path: Path,
        monkeypatch: pytest.MonkeyPatch,
    ) -> None:
        """An explicit keyless runtime should stay keyless even if the host env has a key."""
        encryption_key = _test_encryption_key()
        config_path = tmp_path / "config.yaml"
        config_path.write_text("models: {}\nagents: {}\n", encoding="utf-8")
        storage_path = tmp_path / "storage"
        encrypted_runtime = constants_mod.resolve_runtime_paths(
            config_path=config_path,
            storage_path=storage_path,
            process_env={CREDENTIALS_ENCRYPTION_KEY_ENV: encryption_key},
        )
        encrypted_manager = get_runtime_credentials_manager(encrypted_runtime)
        encrypted_manager.save_credentials("oauth_service", {"token": "secret-token"})

        monkeypatch.setenv(CREDENTIALS_ENCRYPTION_KEY_ENV, encryption_key)
        keyless_runtime = constants_mod.resolve_runtime_paths(
            config_path=config_path,
            storage_path=storage_path,
            process_env={},
        )
        keyless_manager = get_runtime_credentials_manager(keyless_runtime)

        assert keyless_manager.load_credentials("oauth_service") is None

    def test_runtime_manager_rebuilds_when_shared_credentials_path_changes(
        self,
        tmp_path: Path,
    ) -> None:
        """Distinct runtime credential mirrors should not reuse the same cached manager."""
        config_path = tmp_path / "config.yaml"
        config_path.write_text(
            "models:\n  default:\n    provider: openai\n    id: gpt-5.4\nagents: {}\nrouter:\n  model: default\n",
            encoding="utf-8",
        )
        first_runtime_paths = constants_mod.resolve_runtime_paths(
            config_path=config_path,
            storage_path=tmp_path,
            process_env={SHARED_CREDENTIALS_PATH_ENV: str((tmp_path / "shared-a").resolve())},
        )
        second_runtime_paths = constants_mod.resolve_runtime_paths(
            config_path=config_path,
            storage_path=tmp_path,
            process_env={SHARED_CREDENTIALS_PATH_ENV: str((tmp_path / "shared-b").resolve())},
        )

        first_manager = get_runtime_credentials_manager(first_runtime_paths)
        second_manager = get_runtime_credentials_manager(second_runtime_paths)

        assert first_manager is not second_manager
        assert first_manager.shared_base_path == (tmp_path / "shared-a").resolve()
        assert second_manager.shared_base_path == (tmp_path / "shared-b").resolve()

    def test_global_manager_rebuilds_when_storage_root_changes(self, tmp_path: Path) -> None:
        """Changing the explicit storage root should invalidate the cached manager."""
        config_path = tmp_path / "config.yaml"
        config_path.write_text(
            "models:\n  default:\n    provider: openai\n    id: gpt-5.4\nagents: {}\nrouter:\n  model: default\n",
            encoding="utf-8",
        )
        first_root = tmp_path / "one"
        second_root = tmp_path / "two"

        first_runtime_paths = constants_mod.resolve_runtime_paths(config_path=config_path, storage_path=first_root)
        second_runtime_paths = constants_mod.resolve_runtime_paths(config_path=config_path, storage_path=second_root)

        manager_one = get_runtime_credentials_manager(first_runtime_paths)
        manager_two = get_runtime_credentials_manager(second_runtime_paths)

        assert manager_one.base_path == first_root / "credentials"
        assert manager_two.base_path == second_root / "credentials"
        assert manager_one is not manager_two

    def test_dedicated_worker_manager_reads_mirrored_shared_credentials(
        self,
        tmp_path: Path,
    ) -> None:
        """Dedicated worker processes should load mirrored shared credentials through the runtime cache."""
        root = (tmp_path / "shared-storage").resolve()
        base_manager = CredentialsManager(root / "credentials")
        config_path = tmp_path / "config.yaml"
        config_path.write_text(
            "models:\n  default:\n    provider: openai\n    id: gpt-5.4\nagents: {}\nrouter:\n  model: default\n",
            encoding="utf-8",
        )
        execution_identity = ToolExecutionIdentity(
            channel="matrix",
            agent_name="general",
            requester_id="@alice:example.org",
            room_id="!room:example.org",
            thread_id=None,
            resolved_thread_id=None,
            session_id=None,
            tenant_id="tenant-123",
            account_id="account-456",
        )
        worker_key = "v1:tenant-123:shared:general"
        base_manager.save_credentials("google", {"api_key": "env-key", "_source": "env"})
        sync_shared_credentials_to_worker(
            worker_key,
            allowed_services=frozenset({"google"}),
            credentials_manager=base_manager,
        )
        worker_root = base_manager.for_worker(worker_key).storage_root

        _reset_credentials_manager_cache()

        runtime_paths = constants_mod.resolve_runtime_paths(
            config_path=config_path,
            storage_path=worker_root,
            process_env={
                SHARED_CREDENTIALS_PATH_ENV: str(worker_root / ".shared_credentials"),
                SANDBOX_RUNTIME_ENV_BY_KEY["dedicated_worker_key"]: worker_key,
                SANDBOX_RUNTIME_ENV_BY_KEY["dedicated_worker_root"]: str(worker_root),
            },
        )
        manager = get_runtime_credentials_manager(runtime_paths)
        loaded_credentials = load_scoped_credentials(
            "google",
            credentials_manager=manager,
            worker_target=_worker_target("shared", "general", execution_identity),
        )

        assert loaded_credentials == {"api_key": "env-key", "_source": "env"}


class TestSharedIntegrationCredentialTagging:
    """Regression tests for shared-only integration credential saves."""

    def test_spotify_credentials_saved_from_dashboard_are_tagged_as_ui_source(
        self,
        temp_credentials_dir: Path,
        monkeypatch: pytest.MonkeyPatch,
    ) -> None:
        """Spotify OAuth saves should mark credentials as UI-managed so unscoped workers mirror them."""
        manager = CredentialsManager(temp_credentials_dir)
        target = RequestCredentialsTarget(
            runtime_paths=constants_mod.resolve_primary_runtime_paths(
                config_path=temp_credentials_dir / "config.yaml",
                storage_path=temp_credentials_dir,
            ),
            base_manager=manager,
            target_manager=manager,
            worker_scope=None,
            agent_name=None,
            execution_identity=None,
        )

        def _resolve_target(*_args: object, **_kwargs: object) -> RequestCredentialsTarget:
            return target

        monkeypatch.setattr(
            "mindroom.api.integrations.resolve_request_credentials_target",
            _resolve_target,
        )

        _save_spotify_credentials({"access_token": "spotify-token"}, object())

        assert manager.load_credentials("spotify") == {
            "access_token": "spotify-token",
            "_source": "ui",
        }

    def test_dedicated_worker_manager_uses_current_worker_root_for_shared_scope(
        self,
        tmp_path: Path,
    ) -> None:
        """Dedicated workers mounted at arbitrary paths should read and write shared-scope credentials in the mounted root."""
        worker_root = (tmp_path / "app-worker").resolve()
        worker_manager = CredentialsManager(
            base_path=worker_root / "credentials",
            shared_base_path=worker_root / ".shared_credentials",
            current_worker_key="v1:tenant-123:shared:general",
            current_worker_root=worker_root,
        )
        execution_identity = ToolExecutionIdentity(
            channel="matrix",
            agent_name="general",
            requester_id="@alice:example.org",
            room_id="!room:example.org",
            thread_id=None,
            resolved_thread_id=None,
            session_id=None,
            tenant_id="tenant-123",
            account_id="account-456",
        )
        worker_manager.save_credentials("google", {"token": "ui-token", "_source": "ui"})

        loaded_credentials = load_scoped_credentials(
            "google",
            credentials_manager=worker_manager,
            worker_target=_worker_target("shared", "general", execution_identity),
        )

        assert loaded_credentials == {"token": "ui-token", "_source": "ui"}

        save_scoped_credentials(
            "google",
            {"token": "refreshed-token", "_source": "ui"},
            credentials_manager=worker_manager,
            worker_target=_worker_target("shared", "general", execution_identity),
        )

        assert worker_manager.load_credentials("google") == {"token": "refreshed-token", "_source": "ui"}
        assert not (worker_root / "workers").exists()

    def test_dedicated_worker_manager_uses_current_worker_root_for_isolating_scope(
        self,
        tmp_path: Path,
    ) -> None:
        """Dedicated workers mounted at arbitrary paths should not nest isolating-scope credentials under another workers/ tree."""
        worker_root = (tmp_path / "app-worker").resolve()
        worker_manager = CredentialsManager(
            base_path=worker_root / "credentials",
            shared_base_path=worker_root / ".shared_credentials",
            current_worker_key="v1:tenant-123:user:@alice:example.org",
            current_worker_root=worker_root,
        )
        execution_identity = ToolExecutionIdentity(
            channel="matrix",
            agent_name="persistent_worker_lab",
            requester_id="@alice:example.org",
            room_id="!room:example.org",
            thread_id=None,
            resolved_thread_id=None,
            session_id=None,
            tenant_id="tenant-123",
            account_id="account-456",
        )
        worker_manager.shared_manager().save_credentials("google", {"api_key": "env-key", "_source": "env"})
        worker_manager.save_credentials("google", {"token": "ui-token", "_source": "ui"})

        loaded_credentials = load_scoped_credentials(
            "google",
            credentials_manager=worker_manager,
            worker_target=_worker_target("user", "persistent_worker_lab", execution_identity),
        )

        assert loaded_credentials == {"api_key": "env-key", "token": "ui-token", "_source": "ui"}

        save_scoped_credentials(
            "google",
            {"token": "refreshed-token", "_source": "ui"},
            credentials_manager=worker_manager,
            worker_target=_worker_target("user", "persistent_worker_lab", execution_identity),
        )

        assert worker_manager.load_credentials("google") == {"token": "refreshed-token", "_source": "ui"}
        assert not (worker_root / "workers").exists()
