diff --git a/src/api/infrastructure/migrations/versions/d5e6f7a8b9c0_create_encrypted_credentials_table.py b/src/api/infrastructure/migrations/versions/d5e6f7a8b9c0_create_encrypted_credentials_table.py
new file mode 100644
index 00000000..5d0f8c6a
--- /dev/null
+++ b/src/api/infrastructure/migrations/versions/d5e6f7a8b9c0_create_encrypted_credentials_table.py
@@ -0,0 +1,39 @@
+"""create encrypted_credentials table
+
+Adds the encrypted_credentials table for Fernet-encrypted credential
+storage in the Management bounded context.
+
+Revision ID: d5e6f7a8b9c0
+Revises: c4d5e6f7a8b9
+Create Date: 2026-03-17
+"""
+
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+
+
+# revision identifiers, used by Alembic.
+revision: str = "d5e6f7a8b9c0"
+down_revision: Union[str, Sequence[str], None] = "c4d5e6f7a8b9"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    """Create encrypted_credentials table with composite PK (path, tenant_id)."""
+    op.create_table(
+        "encrypted_credentials",
+        sa.Column("path", sa.String(500), primary_key=True),
+        sa.Column("tenant_id", sa.String(26), primary_key=True),
+        sa.Column("encrypted_value", sa.LargeBinary, nullable=False),
+        sa.Column("key_version", sa.Integer, nullable=False, server_default="0"),
+        sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
+        sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False),
+    )
+
+
+def downgrade() -> None:
+    """Drop encrypted_credentials table."""
+    op.drop_table("encrypted_credentials")
diff --git a/src/api/infrastructure/settings.py b/src/api/infrastructure/settings.py
index 5d1d9c94..6f5a3b02 100644
--- a/src/api/infrastructure/settings.py
+++ b/src/api/infrastructure/settings.py
@@ -387,3 +387,31 @@ def get_oidc_settings() -> OIDCSettings:
     Uses lru_cache to ensure settings are only loaded once.
     """
     return OIDCSettings()
+
+
+class ManagementSettings(BaseSettings):
+    """Management bounded context settings.
+
+    Environment variables:
+        KARTOGRAPH_MGMT_ENCRYPTION_KEY: Comma-separated Fernet keys for MultiFernet
+    """
+
+    model_config = SettingsConfigDict(
+        env_prefix="KARTOGRAPH_MGMT_",
+        env_file=".env",
+        env_file_encoding="utf-8",
+        extra="ignore",
+    )
+
+    encryption_key: SecretStr = Field(
+        description="Comma-separated Fernet keys for MultiFernet",
+    )
+
+
+@lru_cache
+def get_management_settings() -> ManagementSettings:
+    """Get cached Management settings.
+
+    Uses lru_cache to ensure settings are only loaded once.
+    """
+    return ManagementSettings()  # type: ignore[call-arg]
diff --git a/src/api/management/infrastructure/models/__init__.py b/src/api/management/infrastructure/models/__init__.py
index 4de7ad9e..77c908b7 100644
--- a/src/api/management/infrastructure/models/__init__.py
+++ b/src/api/management/infrastructure/models/__init__.py
@@ -5,10 +5,14 @@
 
 from management.infrastructure.models.data_source import DataSourceModel
 from management.infrastructure.models.data_source_sync_run import DataSourceSyncRunModel
+from management.infrastructure.models.encrypted_credential import (
+    EncryptedCredentialModel,
+)
 from management.infrastructure.models.knowledge_graph import KnowledgeGraphModel
 
 __all__ = [
     "DataSourceModel",
     "DataSourceSyncRunModel",
+    "EncryptedCredentialModel",
     "KnowledgeGraphModel",
 ]
diff --git a/src/api/management/infrastructure/models/encrypted_credential.py b/src/api/management/infrastructure/models/encrypted_credential.py
new file mode 100644
index 00000000..f3fe7323
--- /dev/null
+++ b/src/api/management/infrastructure/models/encrypted_credential.py
@@ -0,0 +1,35 @@
+"""SQLAlchemy ORM model for the encrypted_credentials table.
+
+Stores Fernet-encrypted credentials in PostgreSQL, scoped by path
+and tenant_id. Part of the Management bounded context.
+"""
+
+from __future__ import annotations
+
+from sqlalchemy import Integer, LargeBinary, String
+from sqlalchemy.orm import Mapped, mapped_column
+
+from infrastructure.database.models import Base, TimestampMixin
+
+
+class EncryptedCredentialModel(Base, TimestampMixin):
+    """ORM model for encrypted_credentials table.
+
+    Stores encrypted credential blobs keyed by (path, tenant_id).
+    The key_version column tracks which Fernet key was used for
+    encryption to support key rotation.
+    """
+
+    __tablename__ = "encrypted_credentials"
+
+    path: Mapped[str] = mapped_column(String(500), primary_key=True)
+    tenant_id: Mapped[str] = mapped_column(String(26), primary_key=True)
+    encrypted_value: Mapped[bytes] = mapped_column(LargeBinary, nullable=False)
+    key_version: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
+
+    def __repr__(self) -> str:
+        """Return string representation."""
+        return (
+            f"<EncryptedCredentialModel(path={self.path}, "
+            f"tenant_id={self.tenant_id}, key_version={self.key_version})>"
+        )
diff --git a/src/api/management/infrastructure/observability/__init__.py b/src/api/management/infrastructure/observability/__init__.py
index 588d0869..8f45ff5b 100644
--- a/src/api/management/infrastructure/observability/__init__.py
+++ b/src/api/management/infrastructure/observability/__init__.py
@@ -12,12 +12,18 @@
     KnowledgeGraphRepositoryProbe,
     SyncRunRepositoryProbe,
 )
+from management.infrastructure.observability.secret_store_probe import (
+    DefaultSecretStoreProbe,
+    SecretStoreProbe,
+)
 
 __all__ = [
     "DataSourceRepositoryProbe",
     "DefaultDataSourceRepositoryProbe",
     "DefaultKnowledgeGraphRepositoryProbe",
+    "DefaultSecretStoreProbe",
     "DefaultSyncRunRepositoryProbe",
     "KnowledgeGraphRepositoryProbe",
+    "SecretStoreProbe",
     "SyncRunRepositoryProbe",
 ]
diff --git a/src/api/management/infrastructure/observability/secret_store_probe.py b/src/api/management/infrastructure/observability/secret_store_probe.py
new file mode 100644
index 00000000..8df1ef7c
--- /dev/null
+++ b/src/api/management/infrastructure/observability/secret_store_probe.py
@@ -0,0 +1,90 @@
+"""Domain probes for secret store operations.
+
+Following Domain-Oriented Observability patterns, these probes capture
+domain-significant events related to credential storage operations.
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Any, Protocol
+
+import structlog
+
+if TYPE_CHECKING:
+    from shared_kernel.observability_context import ObservationContext
+
+
+class SecretStoreProbe(Protocol):
+    """Domain probe for secret store operations."""
+
+    def credential_stored(self, path: str, tenant_id: str) -> None:
+        """Record that credentials were successfully stored."""
+        ...
+
+    def credential_retrieved(self, path: str, tenant_id: str) -> None:
+        """Record that credentials were successfully retrieved."""
+        ...
+
+    def credential_not_found(self, path: str, tenant_id: str) -> None:
+        """Record that credentials were not found."""
+        ...
+
+    def credential_deleted(self, path: str, tenant_id: str) -> None:
+        """Record that credentials were successfully deleted."""
+        ...
+
+    def with_context(self, context: ObservationContext) -> SecretStoreProbe:
+        """Create a new probe with observation context bound."""
+        ...
+
+
+class DefaultSecretStoreProbe:
+    """Default implementation of SecretStoreProbe using structlog."""
+
+    def __init__(
+        self,
+        logger: structlog.stdlib.BoundLogger | None = None,
+        context: ObservationContext | None = None,
+    ) -> None:
+        self._logger = logger or structlog.get_logger()
+        self._context = context
+
+    def _get_context_kwargs(self) -> dict[str, Any]:
+        if self._context is None:
+            return {}
+        return self._context.as_dict()
+
+    def with_context(self, context: ObservationContext) -> DefaultSecretStoreProbe:
+        return DefaultSecretStoreProbe(logger=self._logger, context=context)
+
+    def credential_stored(self, path: str, tenant_id: str) -> None:
+        self._logger.info(
+            "credential_stored",
+            path=path,
+            tenant_id=tenant_id,
+            **self._get_context_kwargs(),
+        )
+
+    def credential_retrieved(self, path: str, tenant_id: str) -> None:
+        self._logger.debug(
+            "credential_retrieved",
+            path=path,
+            tenant_id=tenant_id,
+            **self._get_context_kwargs(),
+        )
+
+    def credential_not_found(self, path: str, tenant_id: str) -> None:
+        self._logger.debug(
+            "credential_not_found",
+            path=path,
+            tenant_id=tenant_id,
+            **self._get_context_kwargs(),
+        )
+
+    def credential_deleted(self, path: str, tenant_id: str) -> None:
+        self._logger.info(
+            "credential_deleted",
+            path=path,
+            tenant_id=tenant_id,
+            **self._get_context_kwargs(),
+        )
diff --git a/src/api/management/infrastructure/repositories/__init__.py b/src/api/management/infrastructure/repositories/__init__.py
index c36d1a7e..646ffcf1 100644
--- a/src/api/management/infrastructure/repositories/__init__.py
+++ b/src/api/management/infrastructure/repositories/__init__.py
@@ -9,6 +9,9 @@
 from management.infrastructure.repositories.data_source_sync_run_repository import (
     DataSourceSyncRunRepository,
 )
+from management.infrastructure.repositories.fernet_secret_store import (
+    FernetSecretStore,
+)
 from management.infrastructure.repositories.knowledge_graph_repository import (
     KnowledgeGraphRepository,
 )
@@ -16,5 +19,6 @@
 __all__ = [
     "DataSourceRepository",
     "DataSourceSyncRunRepository",
+    "FernetSecretStore",
     "KnowledgeGraphRepository",
 ]
diff --git a/src/api/management/infrastructure/repositories/fernet_secret_store.py b/src/api/management/infrastructure/repositories/fernet_secret_store.py
new file mode 100644
index 00000000..68a2dfbe
--- /dev/null
+++ b/src/api/management/infrastructure/repositories/fernet_secret_store.py
@@ -0,0 +1,112 @@
+"""Fernet-encrypted secret store implementation.
+
+Uses cryptography.fernet.MultiFernet to encrypt credentials at rest
+in PostgreSQL, supporting key rotation via multiple Fernet keys.
+"""
+
+from __future__ import annotations
+
+import json
+
+from cryptography.fernet import Fernet, MultiFernet
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from management.infrastructure.models.encrypted_credential import (
+    EncryptedCredentialModel,
+)
+from management.infrastructure.observability.secret_store_probe import (
+    DefaultSecretStoreProbe,
+    SecretStoreProbe,
+)
+
+
+def _validate_inputs(path: str, tenant_id: str) -> None:
+    """Validate that path and tenant_id are non-empty."""
+    if not path or not path.strip():
+        raise ValueError("path must not be empty or whitespace-only")
+    if not tenant_id or not tenant_id.strip():
+        raise ValueError("tenant_id must not be empty or whitespace-only")
+
+
+class FernetSecretStore:
+    """Fernet-encrypted credential storage backed by PostgreSQL.
+
+    Implements both ISecretStoreRepository (management port) and
+    ICredentialReader (shared kernel) for a single implementation
+    that satisfies both read-write and read-only consumers.
+
+    Uses MultiFernet to support key rotation: the first key in the
+    list is used for encryption, and all keys are tried for decryption.
+    """
+
+    def __init__(
+        self,
+        session: AsyncSession,
+        encryption_keys: list[str],
+        probe: SecretStoreProbe | None = None,
+    ) -> None:
+        self._session = session
+        self._multi_fernet = MultiFernet([Fernet(key) for key in encryption_keys])
+        self._probe: SecretStoreProbe = probe or DefaultSecretStoreProbe()
+
+    async def store(
+        self, path: str, tenant_id: str, credentials: dict[str, str]
+    ) -> None:
+        """Encrypt and persist credentials via upsert."""
+        _validate_inputs(path, tenant_id)
+
+        plaintext = json.dumps(credentials).encode("utf-8")
+        encrypted = self._multi_fernet.encrypt(plaintext)
+
+        model = EncryptedCredentialModel(
+            path=path,
+            tenant_id=tenant_id,
+            encrypted_value=encrypted,
+            key_version=0,
+        )
+        await self._session.merge(model)
+        await self._session.flush()
+        self._probe.credential_stored(path, tenant_id)
+
+    async def retrieve(self, path: str, tenant_id: str) -> dict[str, str]:
+        """Decrypt and return stored credentials.
+
+        Raises:
+            KeyError: If no credentials exist at the given path for the tenant.
+        """
+        _validate_inputs(path, tenant_id)
+
+        stmt = select(EncryptedCredentialModel).where(
+            EncryptedCredentialModel.path == path,
+            EncryptedCredentialModel.tenant_id == tenant_id,
+        )
+        result = await self._session.execute(stmt)
+        model = result.scalar_one_or_none()
+
+        if model is None:
+            self._probe.credential_not_found(path, tenant_id)
+            raise KeyError("Credentials not found")
+
+        decrypted = self._multi_fernet.decrypt(model.encrypted_value)
+        self._probe.credential_retrieved(path, tenant_id)
+        return json.loads(decrypted.decode("utf-8"))
+
+    async def delete(self, path: str, tenant_id: str) -> bool:
+        """Remove stored credentials. Returns True if deleted."""
+        _validate_inputs(path, tenant_id)
+
+        stmt = select(EncryptedCredentialModel).where(
+            EncryptedCredentialModel.path == path,
+            EncryptedCredentialModel.tenant_id == tenant_id,
+        )
+        result = await self._session.execute(stmt)
+        model = result.scalar_one_or_none()
+
+        if model is None:
+            return False
+
+        await self._session.delete(model)
+        await self._session.flush()
+        self._probe.credential_deleted(path, tenant_id)
+        return True
diff --git a/src/api/management/ports/secret_store.py b/src/api/management/ports/secret_store.py
new file mode 100644
index 00000000..562c3984
--- /dev/null
+++ b/src/api/management/ports/secret_store.py
@@ -0,0 +1,61 @@
+"""Secret store port for encrypted credential management.
+
+Defines the ISecretStoreRepository protocol for storing, retrieving,
+and deleting encrypted credentials within the Management bounded context.
+"""
+
+from __future__ import annotations
+
+from typing import Protocol, runtime_checkable
+
+
+@runtime_checkable
+class ISecretStoreRepository(Protocol):
+    """Port for encrypted credential storage.
+
+    Implementations encrypt credentials at rest and scope them by
+    path and tenant_id for defense-in-depth isolation.
+
+    The retrieve method signature matches ICredentialReader.retrieve
+    from shared_kernel/credential_reader.py so that a single
+    implementation can satisfy both protocols.
+    """
+
+    async def store(
+        self, path: str, tenant_id: str, credentials: dict[str, str]
+    ) -> None:
+        """Encrypt and persist credentials.
+
+        Args:
+            path: The credential path (e.g., "datasource/{id}/credentials").
+            tenant_id: The tenant ID for scoping.
+            credentials: Key-value pairs to encrypt and store.
+        """
+        ...
+
+    async def retrieve(self, path: str, tenant_id: str) -> dict[str, str]:
+        """Decrypt and return stored credentials.
+
+        Args:
+            path: The credential path.
+            tenant_id: The tenant ID for scoping.
+
+        Returns:
+            A dictionary of credential key-value pairs.
+
+        Raises:
+            KeyError: If no credentials exist at the given path for the tenant.
+        """
+        ...
+
+    async def delete(self, path: str, tenant_id: str) -> bool:
+        """Remove stored credentials.
+
+        Args:
+            path: The credential path.
+            tenant_id: The tenant ID for scoping.
+
+        Returns:
+            True if credentials were deleted, False if not found.
+        """
+        ...
diff --git a/src/api/tests/integration/management/conftest.py b/src/api/tests/integration/management/conftest.py
index 8d1982ab..8167f93b 100644
--- a/src/api/tests/integration/management/conftest.py
+++ b/src/api/tests/integration/management/conftest.py
@@ -95,7 +95,9 @@ async def clean_management_data(
     async def cleanup() -> None:
         """Perform cleanup with proper FK constraint ordering."""
         try:
-            # Clean management-related outbox entries first
+            # Clean encrypted credentials first
+            await async_session.execute(text("DELETE FROM encrypted_credentials"))
+            # Clean management-related outbox entries
             await async_session.execute(
                 text(
                     "DELETE FROM outbox WHERE aggregate_type "
diff --git a/src/api/tests/integration/management/test_fernet_secret_store.py b/src/api/tests/integration/management/test_fernet_secret_store.py
new file mode 100644
index 00000000..0e1b9223
--- /dev/null
+++ b/src/api/tests/integration/management/test_fernet_secret_store.py
@@ -0,0 +1,154 @@
+"""Integration tests for FernetSecretStore.
+
+These tests require a running PostgreSQL instance with the
+encrypted_credentials table created via Alembic migration.
+"""
+
+from __future__ import annotations
+
+from collections.abc import AsyncGenerator
+
+import pytest
+import pytest_asyncio
+from cryptography.fernet import Fernet
+from sqlalchemy import text
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from management.infrastructure.repositories.fernet_secret_store import FernetSecretStore
+
+pytestmark = pytest.mark.integration
+
+
+@pytest.fixture(scope="module")
+def encryption_key() -> str:
+    """Generate a Fernet key for integration tests."""
+    return Fernet.generate_key().decode()
+
+
+@pytest_asyncio.fixture
+async def clean_encrypted_credentials(
+    async_session: AsyncSession,
+) -> AsyncGenerator[None, None]:
+    """Clean encrypted_credentials table before each test."""
+    try:
+        await async_session.execute(text("DELETE FROM encrypted_credentials"))
+        await async_session.commit()
+    except Exception:
+        await async_session.rollback()
+
+    yield
+
+    try:
+        await async_session.execute(text("DELETE FROM encrypted_credentials"))
+        await async_session.commit()
+    except Exception:
+        await async_session.rollback()
+
+
+@pytest.fixture
+def secret_store(async_session: AsyncSession, encryption_key: str) -> FernetSecretStore:
+    """Provide a FernetSecretStore for integration tests."""
+    return FernetSecretStore(session=async_session, encryption_keys=[encryption_key])
+
+
+class TestRoundTrip:
+    """Test store-then-retrieve returns original credentials."""
+
+    @pytest.mark.asyncio
+    async def test_store_and_retrieve(
+        self,
+        secret_store: FernetSecretStore,
+        async_session: AsyncSession,
+        test_tenant: str,
+        clean_encrypted_credentials: None,
+    ):
+        credentials = {"token": "test-value-abc-123"}
+        path = "datasource/integ-1/creds"
+
+        async with async_session.begin():
+            await secret_store.store(path, test_tenant, credentials)
+
+        result = await secret_store.retrieve(path, test_tenant)
+        assert result == credentials
+
+
+class TestOverwrite:
+    """Test that storing twice with same path overwrites."""
+
+    @pytest.mark.asyncio
+    async def test_overwrite_credentials(
+        self,
+        secret_store: FernetSecretStore,
+        async_session: AsyncSession,
+        test_tenant: str,
+        clean_encrypted_credentials: None,
+    ):
+        path = "datasource/integ-2/creds"
+
+        async with async_session.begin():
+            await secret_store.store(path, test_tenant, {"token": "old_value"})
+
+        async with async_session.begin():
+            await secret_store.store(path, test_tenant, {"token": "new_value"})
+
+        result = await secret_store.retrieve(path, test_tenant)
+        assert result == {"token": "new_value"}
+
+
+class TestDelete:
+    """Test store-then-delete-then-retrieve raises KeyError."""
+
+    @pytest.mark.asyncio
+    async def test_delete_then_retrieve_raises_key_error(
+        self,
+        secret_store: FernetSecretStore,
+        async_session: AsyncSession,
+        test_tenant: str,
+        clean_encrypted_credentials: None,
+    ):
+        path = "datasource/integ-3/creds"
+
+        async with async_session.begin():
+            await secret_store.store(path, test_tenant, {"token": "to_delete"})
+
+        async with async_session.begin():
+            deleted = await secret_store.delete(path, test_tenant)
+
+        assert deleted is True
+
+        with pytest.raises(KeyError):
+            await secret_store.retrieve(path, test_tenant)
+
+
+class TestTenantIsolation:
+    """Test that credentials are scoped by tenant_id."""
+
+    @pytest.mark.asyncio
+    async def test_tenant_a_cannot_read_tenant_b(
+        self,
+        secret_store: FernetSecretStore,
+        async_session: AsyncSession,
+        test_tenant: str,
+        clean_encrypted_credentials: None,
+    ):
+        path = "datasource/integ-4/creds"
+
+        async with async_session.begin():
+            await secret_store.store(path, test_tenant, {"token": "tenant_a_only"})
+
+        with pytest.raises(KeyError):
+            await secret_store.retrieve(path, "different-tenant-id")
+
+
+class TestNotFound:
+    """Test that retrieve raises KeyError for nonexistent path."""
+
+    @pytest.mark.asyncio
+    async def test_nonexistent_path_raises_key_error(
+        self,
+        secret_store: FernetSecretStore,
+        test_tenant: str,
+        clean_encrypted_credentials: None,
+    ):
+        with pytest.raises(KeyError):
+            await secret_store.retrieve("nonexistent/path", test_tenant)
diff --git a/src/api/tests/unit/management/infrastructure/test_fernet_secret_store.py b/src/api/tests/unit/management/infrastructure/test_fernet_secret_store.py
new file mode 100644
index 00000000..8fd0c01b
--- /dev/null
+++ b/src/api/tests/unit/management/infrastructure/test_fernet_secret_store.py
@@ -0,0 +1,290 @@
+"""Unit tests for FernetSecretStore.
+
+Tests the Fernet encryption/decryption logic without a database.
+Database operations are mocked with unittest.mock.AsyncMock.
+"""
+
+from __future__ import annotations
+
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+from cryptography.fernet import Fernet, InvalidToken
+
+from management.ports.secret_store import ISecretStoreRepository
+from shared_kernel.credential_reader import ICredentialReader
+
+
+@pytest.fixture
+def fernet_key() -> str:
+    """Generate a fresh Fernet key."""
+    return Fernet.generate_key().decode()
+
+
+@pytest.fixture
+def fernet_key_2() -> str:
+    """Generate a second Fernet key for rotation tests."""
+    return Fernet.generate_key().decode()
+
+
+@pytest.fixture
+def mock_session() -> AsyncMock:
+    """Provide a mocked AsyncSession."""
+    session = AsyncMock()
+    session.merge = AsyncMock()
+    session.execute = AsyncMock()
+    session.delete = AsyncMock()
+    return session
+
+
+def _make_store(session: AsyncMock, keys: list[str]):
+    """Create a FernetSecretStore with mocked session."""
+    from management.infrastructure.repositories.fernet_secret_store import (
+        FernetSecretStore,
+    )
+
+    return FernetSecretStore(session=session, encryption_keys=keys)
+
+
+class TestProtocolConformance:
+    """Verify FernetSecretStore satisfies both port protocols."""
+
+    def test_implements_secret_store_repository(
+        self, mock_session: AsyncMock, fernet_key: str
+    ):
+        store = _make_store(mock_session, [fernet_key])
+        assert isinstance(store, ISecretStoreRepository)
+
+    def test_implements_credential_reader(
+        self, mock_session: AsyncMock, fernet_key: str
+    ):
+        store = _make_store(mock_session, [fernet_key])
+        assert isinstance(store, ICredentialReader)
+
+
+class TestFernetRoundTrip:
+    """Test encrypt-then-decrypt produces original credentials."""
+
+    @pytest.mark.asyncio
+    async def test_round_trip_single_key(
+        self, mock_session: AsyncMock, fernet_key: str
+    ):
+        store = _make_store(mock_session, [fernet_key])
+        credentials = {"token": "ghp_abc123"}
+
+        # Store: capture the encrypted value passed to merge
+        await store.store("datasource/1/creds", "tenant-1", credentials)
+        merge_call = mock_session.merge.call_args
+        model = merge_call[0][0]
+
+        # Retrieve: mock the DB to return the encrypted blob
+        mock_result = MagicMock()
+        mock_result.scalar_one_or_none.return_value = model
+        mock_session.execute.return_value = mock_result
+
+        result = await store.retrieve("datasource/1/creds", "tenant-1")
+        assert result == credentials
+
+    @pytest.mark.asyncio
+    async def test_round_trip_multiple_credentials(
+        self, mock_session: AsyncMock, fernet_key: str
+    ):
+        store = _make_store(mock_session, [fernet_key])
+        credentials = {"username": "admin", "password": "secret", "host": "db.local"}
+
+        await store.store("datasource/2/creds", "tenant-1", credentials)
+        model = mock_session.merge.call_args[0][0]
+
+        mock_result = MagicMock()
+        mock_result.scalar_one_or_none.return_value = model
+        mock_session.execute.return_value = mock_result
+
+        result = await store.retrieve("datasource/2/creds", "tenant-1")
+        assert result == credentials
+
+    @pytest.mark.asyncio
+    async def test_round_trip_empty_values(
+        self, mock_session: AsyncMock, fernet_key: str
+    ):
+        store = _make_store(mock_session, [fernet_key])
+        credentials = {"token": ""}
+
+        await store.store("datasource/3/creds", "tenant-1", credentials)
+        model = mock_session.merge.call_args[0][0]
+
+        mock_result = MagicMock()
+        mock_result.scalar_one_or_none.return_value = model
+        mock_session.execute.return_value = mock_result
+
+        result = await store.retrieve("datasource/3/creds", "tenant-1")
+        assert result == credentials
+
+
+class TestMultiFernetRotation:
+    """Test that key rotation works via MultiFernet."""
+
+    @pytest.mark.asyncio
+    async def test_decrypt_with_rotated_keys(
+        self, mock_session: AsyncMock, fernet_key: str, fernet_key_2: str
+    ):
+        # Encrypt with key1
+        store_v1 = _make_store(mock_session, [fernet_key])
+        credentials = {"token": "ghp_rotate_me"}
+
+        await store_v1.store("datasource/4/creds", "tenant-1", credentials)
+        model = mock_session.merge.call_args[0][0]
+
+        # Decrypt with [key2, key1] — key2 is primary, key1 is legacy
+        store_v2 = _make_store(mock_session, [fernet_key_2, fernet_key])
+
+        mock_result = MagicMock()
+        mock_result.scalar_one_or_none.return_value = model
+        mock_session.execute.return_value = mock_result
+
+        result = await store_v2.retrieve("datasource/4/creds", "tenant-1")
+        assert result == credentials
+
+
+class TestRetrieveNotFound:
+    """Test that retrieve raises KeyError when not found."""
+
+    @pytest.mark.asyncio
+    async def test_raises_key_error(self, mock_session: AsyncMock, fernet_key: str):
+        store = _make_store(mock_session, [fernet_key])
+
+        mock_result = MagicMock()
+        mock_result.scalar_one_or_none.return_value = None
+        mock_session.execute.return_value = mock_result
+
+        with pytest.raises(KeyError):
+            await store.retrieve("nonexistent/path", "tenant-1")
+
+
+class TestInvalidKey:
+    """Test that invalid Fernet keys raise an error."""
+
+    def test_invalid_key_raises_error(self, mock_session: AsyncMock):
+        with pytest.raises(Exception):
+            _make_store(mock_session, ["not-a-valid-fernet-key"])
+
+
+class TestDelete:
+    """Test delete behavior with mocked session."""
+
+    @pytest.mark.asyncio
+    async def test_delete_returns_true_when_credentials_exist(
+        self, mock_session: AsyncMock, fernet_key: str
+    ):
+        store = _make_store(mock_session, [fernet_key])
+
+        from management.infrastructure.models.encrypted_credential import (
+            EncryptedCredentialModel,
+        )
+
+        model = EncryptedCredentialModel(
+            path="datasource/5/creds",
+            tenant_id="tenant-1",
+            encrypted_value=b"encrypted",
+            key_version=0,
+        )
+        mock_result = MagicMock()
+        mock_result.scalar_one_or_none.return_value = model
+        mock_session.execute.return_value = mock_result
+
+        result = await store.delete("datasource/5/creds", "tenant-1")
+        assert result is True
+        mock_session.delete.assert_awaited_once_with(model)
+        mock_session.flush.assert_awaited()
+
+    @pytest.mark.asyncio
+    async def test_delete_returns_false_when_credentials_not_found(
+        self, mock_session: AsyncMock, fernet_key: str
+    ):
+        store = _make_store(mock_session, [fernet_key])
+
+        mock_result = MagicMock()
+        mock_result.scalar_one_or_none.return_value = None
+        mock_session.execute.return_value = mock_result
+
+        result = await store.delete("datasource/6/creds", "tenant-1")
+        assert result is False
+        mock_session.delete.assert_not_awaited()
+
+
+class TestInputValidation:
+    """Test input validation for path and tenant_id."""
+
+    @pytest.mark.asyncio
+    async def test_store_with_empty_path_raises_value_error(
+        self, mock_session: AsyncMock, fernet_key: str
+    ):
+        store = _make_store(mock_session, [fernet_key])
+        with pytest.raises(ValueError, match="path must not be empty"):
+            await store.store("", "tenant-1", {"token": "abc"})
+
+    @pytest.mark.asyncio
+    async def test_retrieve_with_empty_tenant_id_raises_value_error(
+        self, mock_session: AsyncMock, fernet_key: str
+    ):
+        store = _make_store(mock_session, [fernet_key])
+        with pytest.raises(ValueError, match="tenant_id must not be empty"):
+            await store.retrieve("datasource/1/creds", "")
+
+    @pytest.mark.asyncio
+    async def test_store_with_whitespace_path_raises_value_error(
+        self, mock_session: AsyncMock, fernet_key: str
+    ):
+        store = _make_store(mock_session, [fernet_key])
+        with pytest.raises(ValueError, match="path must not be empty"):
+            await store.store("   ", "tenant-1", {"token": "abc"})
+
+    @pytest.mark.asyncio
+    async def test_retrieve_with_whitespace_tenant_id_raises_value_error(
+        self, mock_session: AsyncMock, fernet_key: str
+    ):
+        store = _make_store(mock_session, [fernet_key])
+        with pytest.raises(ValueError, match="tenant_id must not be empty"):
+            await store.retrieve("datasource/1/creds", "   ")
+
+    @pytest.mark.asyncio
+    async def test_delete_with_empty_path_raises_value_error(
+        self, mock_session: AsyncMock, fernet_key: str
+    ):
+        store = _make_store(mock_session, [fernet_key])
+        with pytest.raises(ValueError, match="path must not be empty"):
+            await store.delete("", "tenant-1")
+
+    @pytest.mark.asyncio
+    async def test_delete_with_empty_tenant_id_raises_value_error(
+        self, mock_session: AsyncMock, fernet_key: str
+    ):
+        store = _make_store(mock_session, [fernet_key])
+        with pytest.raises(ValueError, match="tenant_id must not be empty"):
+            await store.delete("datasource/1/creds", "")
+
+
+class TestCorruptedCiphertext:
+    """Test that corrupted ciphertext raises InvalidToken."""
+
+    @pytest.mark.asyncio
+    async def test_corrupted_encrypted_value_raises_invalid_token(
+        self, mock_session: AsyncMock, fernet_key: str
+    ):
+        store = _make_store(mock_session, [fernet_key])
+
+        from management.infrastructure.models.encrypted_credential import (
+            EncryptedCredentialModel,
+        )
+
+        model = EncryptedCredentialModel(
+            path="datasource/corrupt/creds",
+            tenant_id="tenant-1",
+            encrypted_value=b"this-is-not-valid-fernet-ciphertext",
+            key_version=0,
+        )
+        mock_result = MagicMock()
+        mock_result.scalar_one_or_none.return_value = model
+        mock_session.execute.return_value = mock_result
+
+        with pytest.raises(InvalidToken):
+            await store.retrieve("datasource/corrupt/creds", "tenant-1")
diff --git a/src/api/uv.lock b/src/api/uv.lock
index 8c225ae0..d9a98e23 100644
--- a/src/api/uv.lock
+++ b/src/api/uv.lock
@@ -1,5 +1,5 @@
 version = 1
-revision = 3
+revision = 2
 requires-python = ">=3.12"
 resolution-markers = [
     "python_full_version == '3.13.*'",
@@ -1171,7 +1171,7 @@ wheels = [
 
 [[package]]
 name = "kartograph-api"
-version = "3.25.1"
+version = "3.29.0"
 source = { virtual = "." }
 dependencies = [
     { name = "alembic" },