"""Tests for the credentials API endpoints."""

from collections.abc import Generator
from dataclasses import replace
from pathlib import Path
from unittest.mock import AsyncMock, MagicMock, patch

import pytest
from fastapi import Request
from fastapi.testclient import TestClient

from mindroom import constants
from mindroom.api import credentials_oauth_policy, credentials_target, main
from mindroom.api.main import app, initialize_api_app
from mindroom.config.main import Config
from mindroom.credential_policy import RUNTIME_BOOTSTRAPPED_CLIENT_CONFIG_KEY
from mindroom.credentials import get_runtime_credentials_manager
from mindroom.mcp.config import MCPServerConfig
from mindroom.mcp.oauth import mcp_oauth_provider
from mindroom.oauth.providers import OAuthProvider
from mindroom.tool_system.worker_routing import ToolExecutionIdentity, resolve_worker_key, resolve_worker_target
from tests.api.conftest import trusted_upstream_headers, use_trusted_upstream_runtime


def _config_with_worker_scope(
    worker_scope: str | None,
    *,
    authorization: dict[str, object] | None = None,
    worker_grantable_credentials: list[str] | None = None,
) -> Config:
    payload: dict[str, object] = {
        "models": {"default": {"provider": "openai", "id": "gpt-4o-mini"}},
        "agents": {
            "general": {
                "display_name": "General",
                "role": "test",
                "tools": ["calculator"],
                "instructions": ["hi"],
                "rooms": ["lobby"],
            },
        },
        "defaults": {
            "markdown": True,
            "worker_grantable_credentials": worker_grantable_credentials,
        },
    }
    if authorization is not None:
        payload["authorization"] = authorization
    config = Config.model_validate(payload)
    config.agents["general"].worker_scope = worker_scope
    return config


def _publish_committed_runtime_config(api_app: object, config: Config) -> None:
    """Publish one committed config snapshot for dashboard credential tests."""
    context = main._app_context(api_app)
    context.config_data = config.authored_model_dump()
    context.runtime_config = config
    context.config_load_result = main.ConfigLoadResult(success=True)


def _use_owner_runtime(api_app: object, matrix_user_id: str = "@alice:example.org") -> constants.RuntimePaths:
    runtime_paths = main._app_runtime_paths(api_app)
    owner_runtime_paths = constants.resolve_primary_runtime_paths(
        config_path=runtime_paths.config_path,
        storage_path=runtime_paths.storage_root,
        process_env={constants.OWNER_MATRIX_USER_ID_ENV: matrix_user_id},
    )
    initialize_api_app(api_app, owner_runtime_paths)
    return owner_runtime_paths


@pytest.fixture
def client(tmp_path: Path) -> TestClient:
    """Create a test client for the API."""
    initialize_api_app(
        app,
        constants.resolve_primary_runtime_paths(
            config_path=tmp_path / "config.yaml",
            storage_path=tmp_path / "mindroom_data",
            process_env={},
        ),
    )
    return TestClient(app)


@pytest.fixture
def mock_credentials_manager() -> Generator[MagicMock, None, None]:
    """Mock the credentials manager."""
    with patch("mindroom.api.credentials_target.get_runtime_credentials_manager") as mock:
        mock_manager = MagicMock()
        mock.return_value = mock_manager
        yield mock_manager


class TestCredentialsAPI:
    """Test the credentials API endpoints."""

    def test_set_credentials_endpoint(
        self,
        client: TestClient,
        mock_credentials_manager: MagicMock,
    ) -> None:
        """Test setting multiple credentials for a service."""
        response = client.post(
            "/api/credentials/email",
            json={
                "credentials": {
                    "SMTP_HOST": "smtp.example.com",
                    "SMTP_PORT": "587",
                    "SMTP_USERNAME": "user@example.com",
                    "SMTP_PASSWORD": "secret",
                },
            },
        )

        assert response.status_code == 200
        assert response.json() == {
            "status": "success",
            "message": "Credentials saved for email",
        }

        # Verify the manager was called correctly (includes _source: ui)
        mock_credentials_manager.save_credentials.assert_called_once_with(
            "email",
            {
                "SMTP_HOST": "smtp.example.com",
                "SMTP_PORT": "587",
                "SMTP_USERNAME": "user@example.com",
                "SMTP_PASSWORD": "secret",
                "_source": "ui",
            },
        )

    def test_set_api_key_endpoint(
        self,
        client: TestClient,
        mock_credentials_manager: MagicMock,
    ) -> None:
        """Test setting a single API key."""
        mock_credentials_manager.load_credentials.return_value = None

        response = client.post(
            "/api/credentials/openai/api-key",
            json={
                "service": "openai",
                "api_key": "sk-test123",
                "key_name": "api_key",
            },
        )

        assert response.status_code == 200
        assert response.json() == {
            "status": "success",
            "message": "API key set for openai",
        }

        mock_credentials_manager.save_credentials.assert_called_once_with(
            "openai",
            {"api_key": "sk-test123", "_source": "ui"},
        )

    def test_set_credentials_returns_400_when_store_refuses_write(
        self,
        client: TestClient,
        mock_credentials_manager: MagicMock,
    ) -> None:
        """Credential store fail-closed write errors should be actionable dashboard errors."""
        mock_credentials_manager.load_credentials.return_value = None
        mock_credentials_manager.save_credentials.side_effect = ValueError(
            "Stored credentials for openai could not be loaded; refusing to overwrite",
        )

        response = client.post(
            "/api/credentials/openai",
            json={"credentials": {"api_key": "new-key"}},
        )

        assert response.status_code == 400
        assert response.json()["detail"] == "Stored credentials for openai could not be loaded; refusing to overwrite"

    def test_set_api_key_returns_400_when_store_refuses_write(
        self,
        client: TestClient,
        mock_credentials_manager: MagicMock,
    ) -> None:
        """API key helper should translate credential store write refusal to HTTP 400."""
        mock_credentials_manager.load_credentials.return_value = None
        mock_credentials_manager.save_credentials.side_effect = ValueError(
            "Stored credentials for openai could not be loaded; refusing to overwrite",
        )

        response = client.post(
            "/api/credentials/openai/api-key",
            json={
                "service": "openai",
                "api_key": "new-key",
                "key_name": "api_key",
            },
        )

        assert response.status_code == 400
        assert response.json()["detail"] == "Stored credentials for openai could not be loaded; refusing to overwrite"

    def test_copy_credentials_returns_400_when_store_refuses_write(
        self,
        client: TestClient,
        mock_credentials_manager: MagicMock,
    ) -> None:
        """Credential copy should use the same dashboard write error boundary."""
        mock_credentials_manager.load_credentials.side_effect = [
            {"api_key": "source-key", "_source": "ui"},
            None,
        ]
        mock_credentials_manager.save_credentials.side_effect = ValueError(
            "Stored credentials for destination could not be loaded; refusing to overwrite",
        )

        response = client.post("/api/credentials/destination/copy-from/source")

        assert response.status_code == 400
        assert (
            response.json()["detail"] == "Stored credentials for destination could not be loaded; refusing to overwrite"
        )

    def test_rejects_raw_worker_key_query_param(
        self,
        client: TestClient,
    ) -> None:
        """Dashboard credential API should not accept raw worker_key targeting."""
        response = client.post(
            "/api/credentials/openai/api-key?worker_key=worker-a",
            json={
                "service": "openai",
                "api_key": "sk-test123",
                "key_name": "api_key",
            },
        )

        assert response.status_code == 400
        assert "worker_key" in response.json()["detail"]

    def test_agent_name_rejects_user_scope(
        self,
        client: TestClient,
    ) -> None:
        """Dashboard credential management should reject user-scoped agents."""
        config = _config_with_worker_scope("user")
        _publish_committed_runtime_config(client.app, config)
        response = client.get("/api/credentials/openai/api-key?agent_name=general")

        assert response.status_code == 400
        assert "worker_scope=user" in response.json()["detail"]

    def test_agent_name_rejects_user_agent_scope(
        self,
        client: TestClient,
    ) -> None:
        """Dashboard credential management should reject user-agent scoped agents."""
        config = _config_with_worker_scope("user_agent")
        _publish_committed_runtime_config(client.app, config)
        response = client.get("/api/credentials/openai/api-key?agent_name=general")

        assert response.status_code == 400
        assert "worker_scope=user_agent" in response.json()["detail"]

    def test_shared_agent_name_uses_customer_id_for_worker_tenant(
        self,
        client: TestClient,
        mock_credentials_manager: MagicMock,
        monkeypatch: pytest.MonkeyPatch,
    ) -> None:
        """Dashboard worker targeting must use the same tenant identity as runtime routing."""
        config = _config_with_worker_scope("shared")
        worker_manager = MagicMock()
        worker_manager.load_credentials.return_value = {
            "api_key": "sk-worker-scope",
            "_source": "ui",
        }
        mock_credentials_manager.for_worker.return_value = worker_manager
        monkeypatch.setenv("CUSTOMER_ID", "tenant-123")
        monkeypatch.setenv("ACCOUNT_ID", "account-456")
        runtime_paths = main._app_runtime_paths(client.app)
        main.initialize_api_app(
            client.app,
            constants.resolve_primary_runtime_paths(
                config_path=runtime_paths.config_path,
                storage_path=runtime_paths.storage_root,
                process_env={
                    **dict(runtime_paths.process_env),
                    "CUSTOMER_ID": "tenant-123",
                    "ACCOUNT_ID": "account-456",
                },
            ),
        )
        main._app_context(client.app).auth_state = None

        expected_worker_key = resolve_worker_key(
            "shared",
            ToolExecutionIdentity(
                channel="matrix",
                agent_name="general",
                requester_id=None,
                room_id=None,
                thread_id=None,
                resolved_thread_id=None,
                session_id=None,
                tenant_id="tenant-123",
                account_id="account-456",
            ),
            agent_name="general",
        )
        assert expected_worker_key is not None
        _publish_committed_runtime_config(client.app, config)
        response = client.get("/api/credentials/openai/api-key?agent_name=general")

        assert response.status_code == 200
        mock_credentials_manager.for_worker.assert_called_once_with(expected_worker_key)

    def test_credentials_target_worker_resolver_matches_worker_routing(
        self,
        client: TestClient,
    ) -> None:
        """Credentials and OAuth paths should share one target-to-worker conversion."""
        runtime_paths = main._app_runtime_paths(client.app)
        manager = get_runtime_credentials_manager(runtime_paths)
        identity = ToolExecutionIdentity(
            channel="matrix",
            agent_name="general",
            requester_id="@alice:example.org",
            room_id=None,
            thread_id=None,
            resolved_thread_id=None,
            session_id=None,
        )
        target = credentials_target.RequestCredentialsTarget(
            runtime_paths=runtime_paths,
            base_manager=manager,
            target_manager=manager,
            worker_scope="user_agent",
            agent_name="general",
            execution_identity=identity,
        )

        worker_target = credentials_target.worker_target_for_credentials_target(target)

        assert worker_target == resolve_worker_target(
            "user_agent",
            "general",
            execution_identity=identity,
        )

    def test_credentials_target_worker_resolver_returns_none_for_unscoped_target(
        self,
        client: TestClient,
    ) -> None:
        """Unscoped credential operations must keep using the primary credentials store."""
        runtime_paths = main._app_runtime_paths(client.app)
        manager = get_runtime_credentials_manager(runtime_paths)
        target = credentials_target.RequestCredentialsTarget(
            runtime_paths=runtime_paths,
            base_manager=manager,
            target_manager=manager,
            worker_scope=None,
            agent_name="general",
            execution_identity=None,
        )

        assert credentials_target.worker_target_for_credentials_target(target) is None

    def test_rejects_shared_only_integration_services_for_isolating_scope(
        self,
        client: TestClient,
    ) -> None:
        """Dashboard credential management should fail early for unsupported worker scopes."""
        config = _config_with_worker_scope("user")
        _publish_committed_runtime_config(client.app, config)
        response = client.get("/api/credentials/google?agent_name=general")

        assert response.status_code == 400
        assert "worker_scope=user" in response.json()["detail"]

    def test_list_services_allows_empty_isolating_scope(
        self,
        client: TestClient,
    ) -> None:
        """Dashboard service listing should support private scoped OAuth services."""
        _use_owner_runtime(client.app)
        config = _config_with_worker_scope("user")
        _publish_committed_runtime_config(client.app, config)
        response = client.get("/api/credentials/list?agent_name=general")

        assert response.status_code == 200
        assert response.json() == []

    def test_execution_scope_override_rejects_draft_isolating_scope(
        self,
        client: TestClient,
    ) -> None:
        """Credential management must reject draft-only execution-scope overrides."""
        config = _config_with_worker_scope(None)
        _publish_committed_runtime_config(client.app, config)
        response = client.get("/api/credentials/google?agent_name=general&execution_scope=user")

        assert response.status_code == 409
        assert "Save the configuration before managing credentials" in response.json()["detail"]
        assert "execution_scope=user" in response.json()["detail"]

    def test_execution_scope_override_rejects_draft_unscoped_scope(
        self,
        client: TestClient,
    ) -> None:
        """Credential management must reject draft unscoped overrides too."""
        config = _config_with_worker_scope("shared")
        _publish_committed_runtime_config(client.app, config)
        response = client.get(
            "/api/credentials/openai/api-key?agent_name=general&execution_scope=unscoped",
        )

        assert response.status_code == 409
        assert "execution_scope=unscoped" in response.json()["detail"]
        assert "Persisted scope is worker_scope=shared" in response.json()["detail"]

    def test_oauth_tool_settings_do_not_touch_global_token_service(
        self,
        client: TestClient,
    ) -> None:
        """Saving OAuth-backed tool options should not write or overwrite OAuth tokens."""
        runtime_paths = main._app_runtime_paths(client.app)
        manager = get_runtime_credentials_manager(runtime_paths)
        expected_access_value = "drive-access-value"
        expected_refresh_value = "drive-refresh-value"
        manager.save_credentials(
            "google_drive_oauth",
            {
                "token": expected_access_value,
                "refresh_token": expected_refresh_value,
                "_oauth_provider": "google_drive",
                "_source": "oauth",
            },
        )

        response = client.post(
            "/api/credentials/google_drive",
            json={
                "credentials": {
                    "token": "posted-token",
                    "list_files": False,
                    "max_read_size": 42,
                },
            },
        )

        assert response.status_code == 200
        saved_tokens = manager.load_credentials("google_drive_oauth")
        saved_settings = manager.load_credentials("google_drive")
        assert saved_tokens["token"] == expected_access_value
        assert saved_tokens["refresh_token"] == expected_refresh_value
        assert saved_tokens["_oauth_provider"] == "google_drive"
        assert saved_tokens["_source"] == "oauth"
        assert saved_settings == {
            "list_files": False,
            "max_read_size": 42,
            "_source": "ui",
        }

    def test_get_oauth_credentials_filters_token_fields(
        self,
        client: TestClient,
    ) -> None:
        """OAuth-backed config reads should expose editable tool settings only."""
        runtime_paths = main._app_runtime_paths(client.app)
        manager = get_runtime_credentials_manager(runtime_paths)
        manager.save_credentials(
            "google_drive_oauth",
            {
                "token": "drive-access-value",
                "refresh_token": "drive-refresh-value",
                "client_id": "client-id",
                "token_uri": "https://oauth2.googleapis.com/token",
                "scopes": ["https://www.googleapis.com/auth/drive.metadata.readonly"],
                "expires_at": 1234.0,
                "_oauth_provider": "google_drive",
                "_source": "oauth",
            },
        )
        manager.save_credentials(
            "google_drive",
            {
                "list_files": False,
                "max_read_size": 42,
                "_source": "ui",
            },
        )

        response = client.get("/api/credentials/google_drive")
        status_response = client.get("/api/credentials/google_drive/status")
        token_response = client.get("/api/credentials/google_drive_oauth")
        token_status_response = client.get("/api/credentials/google_drive_oauth/status")

        assert response.status_code == 200
        assert response.json() == {
            "service": "google_drive",
            "credentials": {
                "list_files": False,
                "max_read_size": 42,
            },
        }
        assert status_response.status_code == 200
        assert status_response.json()["has_credentials"] is True
        assert set(status_response.json()["key_names"]) == {"list_files", "max_read_size"}
        assert token_response.status_code == 400
        assert "OAuth token credentials" in token_response.json()["detail"]
        assert token_status_response.status_code == 400
        assert "OAuth token credentials" in token_status_response.json()["detail"]

    def test_oauth_client_config_service_redacts_secret_fields(
        self,
        client: TestClient,
    ) -> None:
        """OAuth app client config services should be editable without echoing secrets."""
        runtime_paths = main._app_runtime_paths(client.app)
        manager = get_runtime_credentials_manager(runtime_paths)

        response = client.post(
            "/api/credentials/google_drive_oauth_client",
            json={
                "credentials": {
                    "client_id": "stored-client-id",
                    "client_secret": "stored-client-secret",
                    "redirect_uri": "https://stored.example.test/callback",
                },
            },
        )
        get_response = client.get("/api/credentials/google_drive_oauth_client")
        status_response = client.get("/api/credentials/google_drive_oauth_client/status")

        assert response.status_code == 200
        assert manager.load_credentials("google_drive_oauth_client") == {
            "client_id": "stored-client-id",
            "client_secret": "stored-client-secret",
            "redirect_uri": "https://stored.example.test/callback",
            "_source": "ui",
        }
        assert get_response.status_code == 200
        assert get_response.json() == {
            "service": "google_drive_oauth_client",
            "credentials": {
                "client_id": "stored-client-id",
                "redirect_uri": "https://stored.example.test/callback",
            },
        }
        assert status_response.status_code == 200
        assert set(status_response.json()["key_names"]) == {"client_id", "redirect_uri"}

    def test_oauth_client_config_save_preserves_existing_secret_when_omitted(
        self,
        client: TestClient,
    ) -> None:
        """Editing redacted OAuth client config should not delete the existing secret."""
        runtime_paths = main._app_runtime_paths(client.app)
        manager = get_runtime_credentials_manager(runtime_paths)
        manager.save_credentials(
            "google_drive_oauth_client",
            {
                "client_id": "stored-client-id",
                "client_secret": "stored-client-secret",
                "redirect_uri": "https://old.example.test/callback",
                "_source": "ui",
            },
        )

        response = client.post(
            "/api/credentials/google_drive_oauth_client",
            json={
                "credentials": {
                    "client_id": "stored-client-id",
                    "redirect_uri": "https://new.example.test/callback",
                },
            },
        )

        assert response.status_code == 200
        assert manager.load_credentials("google_drive_oauth_client") == {
            "client_id": "stored-client-id",
            "client_secret": "stored-client-secret",
            "redirect_uri": "https://new.example.test/callback",
            "_source": "ui",
        }

        blank_response = client.post(
            "/api/credentials/google_drive_oauth_client",
            json={
                "credentials": {
                    "client_id": "stored-client-id",
                    "client_secret": "",
                    "redirect_uri": "https://blank.example.test/callback",
                },
            },
        )

        assert blank_response.status_code == 200
        assert manager.load_credentials("google_drive_oauth_client") == {
            "client_id": "stored-client-id",
            "client_secret": "stored-client-secret",
            "redirect_uri": "https://blank.example.test/callback",
            "_source": "ui",
        }

    def test_oauth_client_config_save_rejects_blank_secret_when_client_id_changes(
        self,
        client: TestClient,
    ) -> None:
        """Changing OAuth client IDs should require the matching new client secret."""
        runtime_paths = main._app_runtime_paths(client.app)
        manager = get_runtime_credentials_manager(runtime_paths)
        existing_credentials = {
            "client_id": "stored-client-id",
            "client_secret": "stored-client-secret",
            "redirect_uri": "https://old.example.test/callback",
            "_source": "ui",
        }
        manager.save_credentials("google_drive_oauth_client", existing_credentials)

        response = client.post(
            "/api/credentials/google_drive_oauth_client",
            json={
                "credentials": {
                    "client_id": "new-client-id",
                    "client_secret": "",
                    "redirect_uri": "https://new.example.test/callback",
                },
            },
        )

        assert response.status_code == 400
        assert "client_secret is required when client_id changes" in response.json()["detail"]
        assert manager.load_credentials("google_drive_oauth_client") == existing_credentials

    def test_provisioned_oauth_client_config_rejects_redacted_resave(
        self,
        client: TestClient,
    ) -> None:
        """Re-saving a provisioned client must not silently pin it as a custom client."""
        runtime_paths = main._app_runtime_paths(client.app)
        manager = get_runtime_credentials_manager(runtime_paths)
        existing_credentials = {
            "client_id": "provisioned-client-id",
            "client_secret": "provisioned-client-secret",
            RUNTIME_BOOTSTRAPPED_CLIENT_CONFIG_KEY: True,
        }
        manager.save_credentials("google_oauth_client", existing_credentials)

        response = client.post(
            "/api/credentials/google_oauth_client",
            json={"credentials": {"client_id": "provisioned-client-id"}},
        )

        assert response.status_code == 400
        assert "Provisioned OAuth client configuration does not need to be saved" in response.json()["detail"]
        assert manager.load_credentials("google_oauth_client") == existing_credentials

    def test_oauth_client_config_save_rejects_missing_first_time_secret(
        self,
        client: TestClient,
    ) -> None:
        """First-time OAuth client config saves should require usable client secret material."""
        runtime_paths = main._app_runtime_paths(client.app)
        manager = get_runtime_credentials_manager(runtime_paths)

        response = client.post(
            "/api/credentials/google_drive_oauth_client",
            json={
                "credentials": {
                    "client_id": "stored-client-id",
                    "client_secret": "",
                },
            },
        )

        assert response.status_code == 400
        assert "client_secret is required" in response.json()["detail"]
        assert manager.load_credentials("google_drive_oauth_client") is None

    def test_public_mcp_oauth_client_config_save_allows_missing_first_time_secret(
        self,
        client: TestClient,
    ) -> None:
        """Generated public MCP OAuth client config should allow client_id-only saves."""
        runtime_paths = main._app_runtime_paths(client.app)
        manager = get_runtime_credentials_manager(runtime_paths)
        provider = mcp_oauth_provider(
            "demo",
            MCPServerConfig(
                transport="streamable-http",
                url="https://mcp.example.test/mcp",
                auth={
                    "type": "oauth",
                    "discovery": "manual",
                    "authorization_url": "https://auth.example.test/authorize",
                    "token_url": "https://auth.example.test/token",
                },
            ),
        )

        with patch.object(
            credentials_oauth_policy,
            "load_oauth_providers_for_snapshot",
            return_value={provider.id: provider},
        ):
            response = client.post(
                "/api/credentials/mcp_demo_oauth_client",
                json={
                    "credentials": {
                        "client_id": "stored-client-id",
                        "client_secret": "",
                    },
                },
            )

        assert response.status_code == 200
        assert manager.load_credentials("mcp_demo_oauth_client") == {
            "client_id": "stored-client-id",
            "_source": "ui",
        }

    def test_oauth_client_config_save_rejects_missing_first_time_client_id(
        self,
        client: TestClient,
    ) -> None:
        """First-time OAuth client config saves should require usable client ID material."""
        runtime_paths = main._app_runtime_paths(client.app)
        manager = get_runtime_credentials_manager(runtime_paths)

        response = client.post(
            "/api/credentials/google_drive_oauth_client",
            json={
                "credentials": {
                    "client_secret": "stored-client-secret",
                },
            },
        )

        assert response.status_code == 400
        assert "client_id is required" in response.json()["detail"]
        assert manager.load_credentials("google_drive_oauth_client") is None

    def test_oauth_client_config_rejects_non_client_config_fields(
        self,
        client: TestClient,
    ) -> None:
        """OAuth client config services should not store token material."""
        runtime_paths = main._app_runtime_paths(client.app)
        manager = get_runtime_credentials_manager(runtime_paths)

        response = client.post(
            "/api/credentials/google_drive_oauth_client",
            json={
                "credentials": {
                    "client_id": "stored-client-id",
                    "client_secret": "stored-client-secret",
                    "refresh_token": "smuggled-refresh-token",
                },
            },
        )

        assert response.status_code == 400
        assert "refresh_token" in response.json()["detail"]
        assert manager.load_credentials("google_drive_oauth_client") is None

    def test_oauth_client_config_rejects_non_string_redirect_uri(
        self,
        client: TestClient,
    ) -> None:
        """OAuth client redirect URI should not store values the generic API-key route cannot mask."""
        runtime_paths = main._app_runtime_paths(client.app)
        manager = get_runtime_credentials_manager(runtime_paths)

        response = client.post(
            "/api/credentials/google_drive_oauth_client",
            json={
                "credentials": {
                    "client_id": "stored-client-id",
                    "client_secret": "stored-client-secret",
                    "redirect_uri": 123,
                },
            },
        )

        assert response.status_code == 400
        assert "redirect_uri must be a string" in response.json()["detail"]
        assert manager.load_credentials("google_drive_oauth_client") is None

    def test_oauth_client_config_api_key_read_rejects_legacy_non_string_field(
        self,
        client: TestClient,
    ) -> None:
        """Malformed legacy client config fields should fail as 400 instead of API-key masking errors."""
        runtime_paths = main._app_runtime_paths(client.app)
        manager = get_runtime_credentials_manager(runtime_paths)
        manager.save_credentials(
            "google_drive_oauth_client",
            {
                "client_id": "stored-client-id",
                "client_secret": "stored-client-secret",
                "redirect_uri": 123,
                "_source": "ui",
            },
        )

        response = client.get(
            "/api/credentials/google_drive_oauth_client/api-key?key_name=redirect_uri&include_value=true",
        )

        assert response.status_code == 400
        assert "not a readable string" in response.json()["detail"]

    def test_stored_oauth_client_config_token_fields_are_not_readable(
        self,
        client: TestClient,
    ) -> None:
        """Legacy malformed client config documents should not expose token material."""
        runtime_paths = main._app_runtime_paths(client.app)
        manager = get_runtime_credentials_manager(runtime_paths)
        manager.save_credentials(
            "google_drive_oauth_client",
            {
                "client_id": "stored-client-id",
                "client_secret": "stored-client-secret",
                "refresh_token": "smuggled-refresh-token",
                "_source": "ui",
            },
        )

        get_response = client.get("/api/credentials/google_drive_oauth_client")
        api_key_response = client.get(
            "/api/credentials/google_drive_oauth_client/api-key?key_name=refresh_token&include_value=true",
        )

        assert get_response.status_code == 200
        assert get_response.json() == {
            "service": "google_drive_oauth_client",
            "credentials": {
                "client_id": "stored-client-id",
            },
        }
        assert api_key_response.status_code == 400
        assert "OAuth client config field" in api_key_response.json()["detail"]

    def test_oauth_client_config_rejects_api_key_partial_writes(
        self,
        client: TestClient,
    ) -> None:
        """OAuth client config should not be created through single-field API key writes."""
        runtime_paths = main._app_runtime_paths(client.app)
        manager = get_runtime_credentials_manager(runtime_paths)

        response = client.post(
            "/api/credentials/google_drive_oauth_client/api-key",
            json={
                "service": "google_drive_oauth_client",
                "api_key": "stored-client-id",
                "key_name": "client_id",
            },
        )

        assert response.status_code == 400
        assert "OAuth client config credentials" in response.json()["detail"]
        assert manager.load_credentials("google_drive_oauth_client") is None

    def test_oauth_client_config_service_rejects_copy_to_regular_service(
        self,
        client: TestClient,
    ) -> None:
        """OAuth client secrets should not be copied into non-redacted services."""
        runtime_paths = main._app_runtime_paths(client.app)
        manager = get_runtime_credentials_manager(runtime_paths)
        manager.save_credentials(
            "google_drive_oauth_client",
            {
                "client_id": "stored-client-id",
                "client_secret": "stored-client-secret",
                "_source": "ui",
            },
        )

        response = client.post("/api/credentials/copied_service/copy-from/google_drive_oauth_client")

        assert response.status_code == 400
        assert "OAuth client config credentials cannot be copied" in response.json()["detail"]
        assert manager.load_credentials("copied_service") is None

    def test_orphaned_oauth_client_config_service_redacts_and_rejects_copy(
        self,
        client: TestClient,
    ) -> None:
        """OAuth client config suffix should protect secrets without a loaded provider."""
        runtime_paths = main._app_runtime_paths(client.app)
        manager = get_runtime_credentials_manager(runtime_paths)
        manager.save_credentials(
            "acme_oauth_client",
            {
                "client_id": "stored-client-id",
                "client_secret": "stored-client-secret",
                "_source": "ui",
            },
        )

        get_response = client.get("/api/credentials/acme_oauth_client")
        api_key_response = client.get(
            "/api/credentials/acme_oauth_client/api-key?key_name=client_secret&include_value=true",
        )
        copy_response = client.post("/api/credentials/copied_service/copy-from/acme_oauth_client")

        assert get_response.status_code == 200
        assert get_response.json() == {
            "service": "acme_oauth_client",
            "credentials": {
                "client_id": "stored-client-id",
            },
        }
        assert api_key_response.status_code == 400
        assert "OAuth client secret" in api_key_response.json()["detail"]
        assert copy_response.status_code == 400
        assert "OAuth client config credentials cannot be copied" in copy_response.json()["detail"]
        assert manager.load_credentials("copied_service") is None

    def test_oauth_client_config_agent_save_uses_primary_runtime_store(
        self,
        client: TestClient,
    ) -> None:
        """OAuth client config is deployment config even when edited from an agent-scoped view."""
        runtime_paths = _use_owner_runtime(client.app)
        config = _config_with_worker_scope("shared")
        _publish_committed_runtime_config(client.app, config)
        manager = get_runtime_credentials_manager(runtime_paths)
        worker_key = resolve_worker_key(
            "shared",
            ToolExecutionIdentity(
                channel="matrix",
                agent_name="general",
                requester_id="@alice:example.org",
                room_id=None,
                thread_id=None,
                resolved_thread_id=None,
                session_id=None,
            ),
            agent_name="general",
        )
        assert worker_key is not None

        response = client.post(
            "/api/credentials/google_drive_oauth_client?agent_name=general",
            json={
                "credentials": {
                    "client_id": "stored-client-id",
                    "client_secret": "stored-client-secret",
                },
            },
        )

        assert response.status_code == 200
        assert manager.load_credentials("google_drive_oauth_client") == {
            "client_id": "stored-client-id",
            "client_secret": "stored-client-secret",
            "_source": "ui",
        }
        assert manager.for_worker(worker_key).load_credentials("google_drive_oauth_client") is None

        list_response = client.get("/api/credentials/list?agent_name=general")

        assert list_response.status_code == 200
        assert "google_drive_oauth_client" in list_response.json()

    def test_oauth_client_config_private_agent_save_uses_primary_runtime_store(
        self,
        client: TestClient,
    ) -> None:
        """OAuth client config may be edited from private agent scopes but stays global."""
        runtime_paths = _use_owner_runtime(client.app)
        config = _config_with_worker_scope("user_agent")
        _publish_committed_runtime_config(client.app, config)
        manager = get_runtime_credentials_manager(runtime_paths)
        worker_key = resolve_worker_key(
            "user_agent",
            ToolExecutionIdentity(
                channel="matrix",
                agent_name="general",
                requester_id="@alice:example.org",
                room_id=None,
                thread_id=None,
                resolved_thread_id=None,
                session_id=None,
            ),
            agent_name="general",
        )
        assert worker_key is not None

        response = client.post(
            "/api/credentials/google_drive_oauth_client?agent_name=general",
            json={
                "credentials": {
                    "client_id": "stored-client-id",
                    "client_secret": "stored-client-secret",
                },
            },
        )

        assert response.status_code == 200
        assert manager.load_credentials("google_drive_oauth_client") == {
            "client_id": "stored-client-id",
            "client_secret": "stored-client-secret",
            "_source": "ui",
        }
        assert manager.for_worker(worker_key).load_credentials("google_drive_oauth_client") is None

    def test_plugin_oauth_client_config_agent_save_uses_primary_runtime_store(
        self,
        client: TestClient,
    ) -> None:
        """Plugin OAuth client config should not use worker credential storage."""
        runtime_paths = _use_owner_runtime(client.app)
        config = _config_with_worker_scope("shared")
        _publish_committed_runtime_config(client.app, config)
        manager = get_runtime_credentials_manager(runtime_paths)
        provider = OAuthProvider(
            id="acme",
            display_name="Acme",
            authorization_url="https://auth.example.test/authorize",
            token_url="https://auth.example.test/token",  # noqa: S106
            scopes=("acme.read",),
            credential_service="acme_oauth",
            client_config_services=("acme_oauth_client",),
        )
        worker_key = resolve_worker_key(
            "shared",
            ToolExecutionIdentity(
                channel="matrix",
                agent_name="general",
                requester_id="@alice:example.org",
                room_id=None,
                thread_id=None,
                resolved_thread_id=None,
                session_id=None,
            ),
            agent_name="general",
        )
        assert worker_key is not None

        with patch.object(
            credentials_oauth_policy,
            "load_oauth_providers_for_snapshot",
            return_value={"acme": provider},
        ):
            response = client.post(
                "/api/credentials/acme_oauth_client?agent_name=general",
                json={
                    "credentials": {
                        "client_id": "stored-client-id",
                        "client_secret": "stored-client-secret",
                    },
                },
            )

        assert response.status_code == 200
        assert manager.load_credentials("acme_oauth_client") == {
            "client_id": "stored-client-id",
            "client_secret": "stored-client-secret",
            "_source": "ui",
        }
        assert manager.for_worker(worker_key).load_credentials("acme_oauth_client") is None

    def test_orphaned_oauth_credentials_reject_generic_routes(
        self,
        client: TestClient,
    ) -> None:
        """OAuth-shaped credentials should stay opaque even if their provider is no longer registered."""
        runtime_paths = main._app_runtime_paths(client.app)
        manager = get_runtime_credentials_manager(runtime_paths)
        manager.save_credentials(
            "removed_oauth_provider",
            {
                "access_token": "orphaned-access-token-value",
                "token": "orphaned-access-value",
                "refresh_token": "orphaned-refresh-value",
                "client_id": "client-id",
                "client_secret": "client-secret",
                "id_token": "id-token",
                "provider_session_secret": "provider-secret",
                "token_uri": "https://oauth.example.test/token",
                "scopes": ["drive.read"],
                "expires_at": 1234.0,
                "_oauth_provider": "removed_provider",
                "_source": "oauth",
            },
        )
        manager.save_credentials("normal_source", {"api_key": "normal-key", "_source": "ui"})
        manager.save_credentials(
            "removed_oauth_destination",
            {
                "token": "orphaned-destination-token",
                "_oauth_provider": "removed_provider",
                "_source": "oauth",
            },
        )

        response = client.get("/api/credentials/removed_oauth_provider")
        status_response = client.get("/api/credentials/removed_oauth_provider/status")
        set_response = client.post(
            "/api/credentials/removed_oauth_provider",
            json={"credentials": {"api_key": "replacement"}},
        )
        api_key_response = client.get(
            "/api/credentials/removed_oauth_provider/api-key?key_name=access_token&include_value=true",
        )
        delete_response = client.delete("/api/credentials/removed_oauth_provider")
        copy_response = client.post("/api/credentials/copied_service/copy-from/removed_oauth_provider")
        copy_destination_response = client.post(
            "/api/credentials/removed_oauth_destination/copy-from/normal_source",
        )
        test_response = client.post("/api/credentials/removed_oauth_provider/test")

        responses = [
            response,
            status_response,
            set_response,
            api_key_response,
            delete_response,
            copy_response,
            copy_destination_response,
            test_response,
        ]
        for route_response in responses:
            assert route_response.status_code == 400
            assert "OAuth token credentials" in route_response.json()["detail"]
        assert manager.load_credentials("removed_oauth_provider")["token"] == "orphaned-access-value"  # noqa: S105
        assert manager.load_credentials("removed_oauth_destination")["token"] == "orphaned-destination-token"  # noqa: S105

    def test_oauth_token_service_rejects_generic_credential_writes(
        self,
        client: TestClient,
    ) -> None:
        """OAuth token services should only be written by the OAuth callback path."""
        response = client.post(
            "/api/credentials/google_drive_oauth",
            json={"credentials": {"token": "posted-token"}},
        )

        assert response.status_code == 400
        assert "OAuth token credentials" in response.json()["detail"]

    def test_oauth_token_service_rejects_legacy_credential_routes(
        self,
        client: TestClient,
    ) -> None:
        """OAuth token services should not be manageable through generic credential routes."""
        runtime_paths = main._app_runtime_paths(client.app)
        manager = get_runtime_credentials_manager(runtime_paths)
        manager.save_credentials("google_drive", {"list_files": True, "_source": "ui"})
        manager.save_credentials(
            "google_drive_oauth",
            {
                "token": "drive-access-value",
                "refresh_token": "drive-refresh-value",
                "_oauth_provider": "google_drive",
                "_source": "oauth",
            },
        )

        responses = [
            client.post(
                "/api/credentials/google_drive_oauth/api-key",
                json={"service": "google_drive_oauth", "api_key": "posted-token", "key_name": "token"},
            ),
            client.get("/api/credentials/google_drive_oauth/api-key?key_name=token&include_value=true"),
            client.delete("/api/credentials/google_drive_oauth"),
            client.post("/api/credentials/copied_service/copy-from/google_drive_oauth"),
            client.post("/api/credentials/google_drive_oauth/copy-from/google_drive"),
            client.post("/api/credentials/google_drive_oauth/test"),
        ]

        for response in responses:
            assert response.status_code == 400
            assert "OAuth token credentials" in response.json()["detail"]

    def test_oauth_tool_settings_reject_oauth_reserved_api_key_fields(
        self,
        client: TestClient,
    ) -> None:
        """OAuth tool settings should not accept token-shaped fields through the api-key helper."""
        post_response = client.post(
            "/api/credentials/google_drive/api-key",
            json={"service": "google_drive", "api_key": "drive-token", "key_name": "token"},
        )
        get_response = client.get("/api/credentials/google_drive/api-key?key_name=token&include_value=true")

        assert post_response.status_code == 400
        assert "OAuth field 'token'" in post_response.json()["detail"]
        assert get_response.status_code == 400
        assert "OAuth field 'token'" in get_response.json()["detail"]

    def test_non_oauth_services_can_use_access_token_api_key_field(
        self,
        client: TestClient,
    ) -> None:
        """Reserved OAuth field names remain valid for unrelated non-OAuth integrations."""
        post_response = client.post(
            "/api/credentials/homeassistant/api-key",
            json={"service": "homeassistant", "api_key": "ha-token", "key_name": "access_token"},
        )
        get_response = client.get(
            "/api/credentials/homeassistant/api-key?key_name=access_token&include_value=true",
        )
        status_response = client.get("/api/credentials/homeassistant/status")

        assert post_response.status_code == 200
        assert get_response.status_code == 200
        assert get_response.json()["api_key"] == "ha-token"
        assert status_response.status_code == 200
        assert status_response.json()["key_names"] == ["access_token"]

    def test_oauth_tool_settings_do_not_touch_private_token_service(
        self,
        client: TestClient,
    ) -> None:
        """OAuth-backed tool options may save in private scopes without replacing tokens."""
        _use_owner_runtime(client.app)
        config = _config_with_worker_scope("user_agent")
        _publish_committed_runtime_config(client.app, config)
        runtime_paths = main._app_runtime_paths(client.app)
        manager = get_runtime_credentials_manager(runtime_paths)
        identity = ToolExecutionIdentity(
            channel="matrix",
            agent_name="general",
            requester_id="@alice:example.org",
            room_id=None,
            thread_id=None,
            resolved_thread_id=None,
            session_id=None,
        )
        worker_key = resolve_worker_key("user_agent", identity, agent_name="general")
        assert worker_key is not None
        worker_manager = manager.for_worker(worker_key)
        scoped_manager = manager.for_primary_runtime_scope("@alice:example.org", "general")
        expected_access_value = "scoped-drive-access-value"
        expected_refresh_value = "scoped-drive-refresh-value"
        scoped_manager.save_credentials(
            "google_drive_oauth",
            {
                "token": expected_access_value,
                "refresh_token": expected_refresh_value,
                "_oauth_provider": "google_drive",
                "_source": "oauth",
            },
        )

        response = client.post(
            "/api/credentials/google_drive?agent_name=general",
            json={"credentials": {"list_files": False, "max_read_size": 42}},
        )

        assert response.status_code == 200
        saved_tokens = scoped_manager.load_credentials("google_drive_oauth")
        saved_settings = scoped_manager.load_credentials("google_drive")
        assert saved_tokens["token"] == expected_access_value
        assert saved_tokens["refresh_token"] == expected_refresh_value
        assert saved_tokens["_oauth_provider"] == "google_drive"
        assert saved_tokens["_source"] == "oauth"
        assert saved_settings == {
            "list_files": False,
            "max_read_size": 42,
            "_source": "ui",
        }
        assert worker_manager.load_credentials("google_drive_oauth") is None
        assert worker_manager.load_credentials("google_drive") is None

    def test_get_private_oauth_credentials_filters_token_fields(
        self,
        client: TestClient,
    ) -> None:
        """Private-scope OAuth config reads should not return stored OAuth tokens."""
        _use_owner_runtime(client.app)
        config = _config_with_worker_scope("user_agent")
        _publish_committed_runtime_config(client.app, config)
        runtime_paths = main._app_runtime_paths(client.app)
        manager = get_runtime_credentials_manager(runtime_paths)
        identity = ToolExecutionIdentity(
            channel="matrix",
            agent_name="general",
            requester_id="@alice:example.org",
            room_id=None,
            thread_id=None,
            resolved_thread_id=None,
            session_id=None,
        )
        worker_key = resolve_worker_key("user_agent", identity, agent_name="general")
        assert worker_key is not None
        worker_manager = manager.for_worker(worker_key)
        scoped_manager = manager.for_primary_runtime_scope("@alice:example.org", "general")
        scoped_manager.save_credentials(
            "google_drive_oauth",
            {
                "token": "scoped-drive-access-value",
                "refresh_token": "scoped-drive-refresh-value",
                "client_id": "client-id",
                "token_uri": "https://oauth2.googleapis.com/token",
                "scopes": ["https://www.googleapis.com/auth/drive.metadata.readonly"],
                "expires_at": 1234.0,
                "_oauth_provider": "google_drive",
                "_source": "oauth",
            },
        )
        scoped_manager.save_credentials(
            "google_drive",
            {
                "list_files": False,
                "max_read_size": 42,
                "_source": "ui",
            },
        )

        response = client.get("/api/credentials/google_drive?agent_name=general")
        status_response = client.get("/api/credentials/google_drive/status?agent_name=general")
        token_response = client.get("/api/credentials/google_drive_oauth?agent_name=general")

        assert response.status_code == 200
        assert response.json()["credentials"] == {"list_files": False, "max_read_size": 42}
        assert status_response.status_code == 200
        assert status_response.json()["has_credentials"] is True
        assert set(status_response.json()["key_names"]) == {"list_files", "max_read_size"}
        assert token_response.status_code == 400
        assert "OAuth token credentials" in token_response.json()["detail"]
        assert worker_manager.load_credentials("google_drive_oauth") is None
        assert worker_manager.load_credentials("google_drive") is None

    def test_list_services_includes_private_oauth_tool_settings(
        self,
        client: TestClient,
    ) -> None:
        """Private-scope OAuth tool settings should appear in service listings."""
        _use_owner_runtime(client.app)
        config = _config_with_worker_scope("user_agent")
        _publish_committed_runtime_config(client.app, config)
        runtime_paths = main._app_runtime_paths(client.app)
        manager = get_runtime_credentials_manager(runtime_paths)
        scoped_manager = manager.for_primary_runtime_scope("@alice:example.org", "general")
        scoped_manager.save_credentials(
            "google_drive_oauth",
            {
                "token": "scoped-drive-access-value",
                "refresh_token": "scoped-drive-refresh-value",
                "_oauth_provider": "google_drive",
                "_source": "oauth",
            },
        )
        scoped_manager.save_credentials(
            "google_drive",
            {
                "list_files": False,
                "max_read_size": 42,
                "_source": "ui",
            },
        )

        response = client.get("/api/credentials/list?agent_name=general")
        delete_response = client.delete("/api/credentials/google_drive?agent_name=general")
        deleted_list_response = client.get("/api/credentials/list?agent_name=general")

        assert response.status_code == 200
        assert response.json() == ["google_drive"]
        assert delete_response.status_code == 200
        assert scoped_manager.load_credentials("google_drive") is None
        assert scoped_manager.load_credentials("google_drive_oauth") is not None
        assert deleted_list_response.status_code == 200
        assert deleted_list_response.json() == []

    def test_list_services_discovers_shared_agent_oauth_store(
        self,
        client: TestClient,
    ) -> None:
        """Shared-scope listings should discover agent-store tokens and ignore stale global ones."""
        _use_owner_runtime(client.app)
        config = _config_with_worker_scope("shared")
        _publish_committed_runtime_config(client.app, config)
        runtime_paths = main._app_runtime_paths(client.app)
        manager = get_runtime_credentials_manager(runtime_paths)
        agent_store = manager.for_primary_runtime_agent_scope("general")
        agent_store.save_credentials(
            "google_drive_oauth",
            {
                "token": "agent-drive-access-value",
                "_oauth_provider": "google_drive",
                "_source": "oauth",
            },
        )
        agent_store.save_credentials(
            "acme_oauth",
            {
                "token": "agent-acme-access-value",
                "_oauth_provider": "acme",
                "_source": "oauth",
            },
        )
        manager.save_credentials(
            "legacy_oauth",
            {
                "token": "stale-global-access-value",
                "_source": "oauth",
            },
        )

        response = client.get("/api/credentials/list?agent_name=general")

        assert response.status_code == 200
        # The orphaned agent-store token service is discoverable, the registered
        # provider token stays hidden, and the stale global token never surfaces.
        assert response.json() == ["acme_oauth"]

    def test_primary_runtime_scoped_services_for_shared_agent_target(
        self,
        tmp_path: Path,
    ) -> None:
        """Shared-scope targets should list only agent-store OAuth token services."""
        runtime_paths = constants.resolve_primary_runtime_paths(
            config_path=tmp_path / "config.yaml",
            storage_path=tmp_path / "mindroom_data",
            process_env={},
        )
        manager = get_runtime_credentials_manager(runtime_paths)
        agent_store = manager.for_primary_runtime_agent_scope("general")
        agent_store.save_credentials("acme_oauth", {"token": "agent-token", "_source": "oauth"})
        agent_store.save_credentials("weather", {"api_key": "not-a-token", "_source": "ui"})
        target = credentials_target.RequestCredentialsTarget(
            runtime_paths=runtime_paths,
            base_manager=manager,
            target_manager=manager,
            worker_scope="shared",
            agent_name="general",
            execution_identity=None,
        )

        assert credentials_target.primary_runtime_scoped_services_for_target(target) == {"acme_oauth"}
        assert credentials_target.primary_runtime_scoped_services_for_target(replace(target, agent_name=None)) == set()

    def test_non_oauth_tool_settings_still_reject_private_scopes(
        self,
        client: TestClient,
    ) -> None:
        """Private-scope writes stay limited to registered OAuth credential services."""
        config = _config_with_worker_scope("user_agent")
        _publish_committed_runtime_config(client.app, config)

        response = client.post(
            "/api/credentials/weather?agent_name=general",
            json={"credentials": {"api_key": "weather-key"}},
        )

        assert response.status_code == 400
        assert "worker_scope=user_agent" in response.json()["detail"]

    def test_resolve_request_credentials_target_keeps_one_runtime_for_identity(
        self,
        tmp_path: Path,
    ) -> None:
        """Credential targeting should derive worker identity from the same bound runtime it validated."""
        runtime_a = constants.resolve_primary_runtime_paths(
            config_path=tmp_path / "first.yaml",
            storage_path=tmp_path / "first-store",
            process_env={"CUSTOMER_ID": "tenant-a", "ACCOUNT_ID": "account-a"},
        )
        runtime_b = constants.resolve_primary_runtime_paths(
            config_path=tmp_path / "second.yaml",
            storage_path=tmp_path / "second-store",
            process_env={"CUSTOMER_ID": "tenant-b", "ACCOUNT_ID": "account-b"},
        )
        initialize_api_app(app, runtime_a)
        request = Request(
            {
                "type": "http",
                "app": app,
                "headers": [],
                "query_string": b"",
                "auth_user": {"user_id": "dashboard-user"},
            },
        )
        config = _config_with_worker_scope("shared")
        base_manager = MagicMock()
        base_manager.for_worker.return_value = MagicMock()
        _publish_committed_runtime_config(app, config)

        def _swap_runtime_on_manager_lookup(runtime_paths: object) -> MagicMock:
            assert runtime_paths == runtime_a
            initialize_api_app(app, runtime_b)
            _publish_committed_runtime_config(app, config)
            return base_manager

        with patch(
            "mindroom.api.credentials_target.get_runtime_credentials_manager",
            side_effect=_swap_runtime_on_manager_lookup,
        ):
            target = credentials_target.resolve_request_credentials_target(request, agent_name="general")

        assert target.runtime_paths == runtime_a
        assert target.execution_identity is not None
        assert target.execution_identity.tenant_id == "tenant-a"
        assert target.execution_identity.account_id == "account-a"

    def test_credentials_routes_use_committed_snapshot_until_reload(
        self,
        client: TestClient,
    ) -> None:
        """Credential routes should ignore newer on-disk edits until a snapshot reload is published."""
        config = _config_with_worker_scope("shared")
        _publish_committed_runtime_config(client.app, config)
        runtime_paths = main._app_runtime_paths(client.app)
        runtime_paths.config_path.write_text(
            ("models:\n  default:\n    provider: openai\n    id: gpt-4o-mini\nrouter:\n  model: default\nagents: {}\n"),
            encoding="utf-8",
        )

        response = client.get("/api/credentials/openai/api-key?agent_name=general")

        assert response.status_code == 200

    def test_unknown_agent_rejected_for_dashboard_credentials(self, client: TestClient) -> None:
        """Dashboard credentials must reject unknown agents instead of falling back to shared state."""
        config = _config_with_worker_scope("shared")
        _publish_committed_runtime_config(client.app, config)
        response = client.get("/api/credentials/openai/api-key?agent_name=missing")

        assert response.status_code == 404
        assert response.json()["detail"] == "Unknown agent: missing"

    def test_agent_credentials_require_agent_reply_permission(self, client: TestClient) -> None:
        """Agent-scoped credential routes should reject requesters outside the agent allowlist."""
        use_trusted_upstream_runtime(client.app)
        config = _config_with_worker_scope(
            "shared",
            authorization={"agent_reply_permissions": {"general": ["@alice:example.org"]}},
        )
        _publish_committed_runtime_config(client.app, config)
        bob_headers = trusted_upstream_headers(
            user_id="bob",
            email="bob@example.org",
            matrix_user_id="@bob:example.org",
        )

        agent_response = client.get(
            "/api/credentials/openai/api-key?agent_name=general",
            headers=bob_headers,
        )
        global_response = client.get(
            "/api/credentials/openai/api-key",
            headers=bob_headers,
        )

        assert agent_response.status_code == 403
        assert global_response.status_code == 200
        assert global_response.json()["has_key"] is False

    def test_agent_oauth_token_credentials_authorize_before_generic_rejection(self, client: TestClient) -> None:
        """Unauthorized agent-scoped OAuth token routes should return 403 before route-specific 400s."""
        use_trusted_upstream_runtime(client.app)
        config = _config_with_worker_scope(
            "shared",
            authorization={"agent_reply_permissions": {"general": ["@alice:example.org"]}},
        )
        _publish_committed_runtime_config(client.app, config)
        bob_headers = trusted_upstream_headers(
            user_id="bob",
            email="bob@example.org",
            matrix_user_id="@bob:example.org",
        )

        token_response = client.get(
            "/api/credentials/google_drive_oauth?agent_name=general",
            headers=bob_headers,
        )
        copy_response = client.post(
            "/api/credentials/copied_service/copy-from/google_drive_oauth?agent_name=general",
            headers=bob_headers,
        )

        assert token_response.status_code == 403
        assert copy_response.status_code == 403

    def test_homeassistant_token_connect_authorizes_before_probe(self, client: TestClient) -> None:
        """Unauthorized agent-scoped Home Assistant connects should not contact the provider."""
        use_trusted_upstream_runtime(client.app)
        config = _config_with_worker_scope(
            "shared",
            authorization={"agent_reply_permissions": {"general": ["@alice:example.org"]}},
        )
        _publish_committed_runtime_config(client.app, config)
        bob_headers = trusted_upstream_headers(
            user_id="bob",
            email="bob@example.org",
            matrix_user_id="@bob:example.org",
        )

        with patch(
            "mindroom.api.homeassistant_integration._test_connection",
            new_callable=AsyncMock,
        ) as test_connection:
            response = client.post(
                "/api/homeassistant/connect/token?agent_name=general",
                headers=bob_headers,
                json={
                    "instance_url": "homeassistant.local:8123",
                    "long_lived_token": "ha-token",
                },
            )

        assert response.status_code == 403
        test_connection.assert_not_awaited()

    def test_shared_agent_name_hides_non_allowlisted_shared_credentials(
        self,
        client: TestClient,
    ) -> None:
        """Shared worker scope should not expose shared credentials outside the worker allowlist."""
        config = _config_with_worker_scope("shared")
        runtime_paths = main._app_runtime_paths(client.app)
        manager = get_runtime_credentials_manager(runtime_paths)
        manager.save_credentials("openai", {"api_key": "sk-global-ui", "_source": "ui"})

        _publish_committed_runtime_config(client.app, config)
        response = client.get("/api/credentials/openai/api-key?agent_name=general")

        assert response.status_code == 200
        assert response.json()["has_key"] is False

    def test_shared_agent_name_merges_allowlisted_shared_credentials(
        self,
        client: TestClient,
    ) -> None:
        """Shared worker scope should inherit allowlisted shared credentials regardless of source."""
        config = _config_with_worker_scope("shared", worker_grantable_credentials=["openai"])
        runtime_paths = main._app_runtime_paths(client.app)
        manager = get_runtime_credentials_manager(runtime_paths)
        manager.save_credentials("openai", {"api_key": "sk-global-ui", "_source": "ui"})

        _publish_committed_runtime_config(client.app, config)
        response = client.get("/api/credentials/openai/api-key?agent_name=general")

        assert response.status_code == 200
        assert response.json()["has_key"] is True
        assert response.json()["source"] == "ui"

    def test_shared_agent_name_local_shared_credentials_bypass_worker_allowlist(
        self,
        client: TestClient,
    ) -> None:
        """Shared-scope local integrations should stay visible without worker mirroring allowlists."""
        config = _config_with_worker_scope("shared")
        runtime_paths = main._app_runtime_paths(client.app)
        manager = get_runtime_credentials_manager(runtime_paths)
        manager.save_credentials(
            "homeassistant",
            {
                "instance_url": "http://homeassistant.local:8123",
                "access_token": "ha-token",
                "_source": "ui",
            },
        )

        _publish_committed_runtime_config(client.app, config)

        list_response = client.get("/api/credentials/list?agent_name=general")
        ha_status_response = client.get("/api/credentials/homeassistant/status?agent_name=general")

        assert list_response.status_code == 200
        assert "homeassistant" in list_response.json()

        assert ha_status_response.status_code == 200
        assert ha_status_response.json()["has_credentials"] is True
        assert set(ha_status_response.json()["key_names"]) == {"instance_url", "access_token"}

    def test_rejects_raw_source_worker_key_query_param(
        self,
        client: TestClient,
    ) -> None:
        """Credential copy API should not accept raw source_worker_key targeting."""
        response = client.post("/api/credentials/model:new/copy-from/model:old?source_worker_key=worker-a")

        assert response.status_code == 400
        assert "source_worker_key" in response.json()["detail"]

    def test_get_credentials(
        self,
        client: TestClient,
        mock_credentials_manager: MagicMock,
    ) -> None:
        """Test getting credentials for a service (for editing)."""
        mock_credentials_manager.load_credentials.return_value = {
            "TELEGRAM_TOKEN": "test-token-123",
        }

        response = client.get("/api/credentials/telegram")

        assert response.status_code == 200
        assert response.json() == {
            "service": "telegram",
            "credentials": {
                "TELEGRAM_TOKEN": "test-token-123",
            },
        }

    def test_get_credentials_empty(
        self,
        client: TestClient,
        mock_credentials_manager: MagicMock,
    ) -> None:
        """Test getting credentials when none exist."""
        mock_credentials_manager.load_credentials.return_value = None

        response = client.get("/api/credentials/telegram")

        assert response.status_code == 200
        assert response.json() == {
            "service": "telegram",
            "credentials": {},
        }

    def test_get_credential_status(
        self,
        client: TestClient,
        mock_credentials_manager: MagicMock,
    ) -> None:
        """Test getting credential status."""
        mock_credentials_manager.load_credentials.return_value = {
            "SMTP_HOST": "smtp.example.com",
            "SMTP_PORT": "587",
        }

        response = client.get("/api/credentials/email/status")

        assert response.status_code == 200
        data = response.json()
        assert data["service"] == "email"
        assert data["has_credentials"] is True
        assert set(data["key_names"]) == {"SMTP_HOST", "SMTP_PORT"}

    def test_delete_credentials(
        self,
        client: TestClient,
        mock_credentials_manager: MagicMock,
    ) -> None:
        """Test deleting credentials."""
        response = client.delete("/api/credentials/email")

        assert response.status_code == 200
        assert response.json() == {
            "status": "success",
            "message": "Credentials deleted for email",
        }

        mock_credentials_manager.delete_credentials.assert_called_once_with("email")

    def test_list_services(
        self,
        client: TestClient,
        mock_credentials_manager: MagicMock,
    ) -> None:
        """Test listing services with credentials."""
        mock_credentials_manager.list_services.return_value = ["email", "openai", "github"]

        response = client.get("/api/credentials/list")

        assert response.status_code == 200
        assert response.json() == ["email", "openai", "github"]

    def test_get_api_key_returns_source_env(
        self,
        client: TestClient,
        mock_credentials_manager: MagicMock,
    ) -> None:
        """Test that get_api_key returns source field for env-sourced keys."""
        mock_credentials_manager.load_credentials.return_value = {
            "api_key": "sk-test-long-key-value",
            "_source": "env",
        }

        response = client.get("/api/credentials/openai/api-key")

        assert response.status_code == 200
        data = response.json()
        assert data["has_key"] is True
        assert data["source"] == "env"
        assert data["masked_key"] == "sk-t...alue"

    def test_get_api_key_returns_source_ui(
        self,
        client: TestClient,
        mock_credentials_manager: MagicMock,
    ) -> None:
        """Test that get_api_key returns source field for UI-sourced keys."""
        mock_credentials_manager.load_credentials.return_value = {
            "api_key": "sk-test-long-key-value",
            "_source": "ui",
        }

        response = client.get("/api/credentials/openai/api-key")

        assert response.status_code == 200
        data = response.json()
        assert data["has_key"] is True
        assert data["source"] == "ui"

    def test_get_api_key_include_value(
        self,
        client: TestClient,
        mock_credentials_manager: MagicMock,
    ) -> None:
        """Test that get_api_key can return the full value when explicitly requested."""
        mock_credentials_manager.load_credentials.return_value = {
            "api_key": "sk-real-value",
            "_source": "ui",
        }

        response = client.get("/api/credentials/openai/api-key?include_value=true")

        assert response.status_code == 200
        data = response.json()
        assert data["has_key"] is True
        assert data["api_key"] == "sk-real-value"
        assert data["source"] == "ui"

    def test_get_api_key_returns_source_none_for_legacy(
        self,
        client: TestClient,
        mock_credentials_manager: MagicMock,
    ) -> None:
        """Test that get_api_key returns null source for legacy credentials."""
        mock_credentials_manager.load_credentials.return_value = {
            "api_key": "sk-test-long-key-value",
        }

        response = client.get("/api/credentials/openai/api-key")

        assert response.status_code == 200
        data = response.json()
        assert data["has_key"] is True
        assert data["source"] is None

    def test_get_credentials_filters_internal_keys(
        self,
        client: TestClient,
        mock_credentials_manager: MagicMock,
    ) -> None:
        """Test that get_credentials filters out _source and other internal keys."""
        mock_credentials_manager.load_credentials.return_value = {
            "api_key": "sk-test123",
            "_source": "ui",
        }

        response = client.get("/api/credentials/openai")

        assert response.status_code == 200
        data = response.json()
        assert data["credentials"] == {"api_key": "sk-test123"}
        assert "_source" not in data["credentials"]

    def test_get_credential_status_filters_internal_keys(
        self,
        client: TestClient,
        mock_credentials_manager: MagicMock,
    ) -> None:
        """Test that credential status key_names excludes internal keys."""
        mock_credentials_manager.load_credentials.return_value = {
            "api_key": "sk-test123",
            "_source": "env",
        }

        response = client.get("/api/credentials/openai/status")

        assert response.status_code == 200
        data = response.json()
        assert data["has_credentials"] is True
        assert data["key_names"] == ["api_key"]

    def test_set_api_key_merges_with_existing(
        self,
        client: TestClient,
        mock_credentials_manager: MagicMock,
    ) -> None:
        """Test that set_api_key merges into existing credentials and flips source to ui."""
        mock_credentials_manager.load_credentials.return_value = {
            "api_key": "old-key",
            "_source": "env",
        }

        response = client.post(
            "/api/credentials/openai/api-key",
            json={
                "service": "openai",
                "api_key": "new-key-from-ui",
                "key_name": "api_key",
            },
        )

        assert response.status_code == 200
        mock_credentials_manager.save_credentials.assert_called_once_with(
            "openai",
            {"api_key": "new-key-from-ui", "_source": "ui"},
        )

    def test_rejects_invalid_service_name(
        self,
        client: TestClient,
        mock_credentials_manager: MagicMock,
    ) -> None:
        """Test that invalid service names are rejected server-side."""
        response = client.get("/api/credentials/bad!service/status")

        assert response.status_code == 400
        assert "Service name can only include" in response.json()["detail"]
        mock_credentials_manager.load_credentials.assert_not_called()
