diff --git a/.hyperloop/checks/check-no-api-simulation.sh b/.hyperloop/checks/check-no-api-simulation.sh
index 57d99a9bc..7795d539c 100755
--- a/.hyperloop/checks/check-no-api-simulation.sh
+++ b/.hyperloop/checks/check-no-api-simulation.sh
@@ -47,6 +47,7 @@ check_dir() {
     --include="*.js" \
     --exclude="*.test.*" \
     --exclude="*.spec.*" \
+    --exclude-dir=.venv \
     -E "new Promise[^)]*setTimeout[[:space:]]*\([[:space:]]*resolve" \
     "$dir" 2>/dev/null || true)
 
diff --git a/src/api/tests/fakes/iam.py b/src/api/tests/fakes/iam.py
new file mode 100644
index 000000000..5d1fe2db8
--- /dev/null
+++ b/src/api/tests/fakes/iam.py
@@ -0,0 +1,96 @@
+"""In-memory fakes for IAM bounded context ports.
+
+Provides fast, self-contained test doubles for:
+- TenantServiceProbe
+
+These fakes implement the full port protocols using in-memory storage and
+record all method calls so tests can assert on behavior without resorting
+to MagicMock or AsyncMock.  They follow the "Fakes over Mocks" principle
+from specs/nfr/testing.spec.md.
+"""
+
+from __future__ import annotations
+
+
+class RecordingTenantServiceProbe:
+    """Concrete recording probe implementing TenantServiceProbe protocol.
+
+    Records every method call in per-method lists so tests can assert on
+    which events were raised and what arguments were passed.  This is a
+    concrete class — NOT a MagicMock(spec=...) — as required by the testing
+    NFR (specs/nfr/testing.spec.md).
+    """
+
+    def __init__(self) -> None:
+        self.tenant_created_calls: list[dict[str, str]] = []
+        self.tenant_retrieved_calls: list[dict[str, str]] = []
+        self.tenants_listed_calls: list[dict[str, object]] = []
+        self.tenant_deleted_calls: list[dict[str, str]] = []
+        self.tenant_not_found_calls: list[dict[str, str]] = []
+        self.duplicate_tenant_name_calls: list[dict[str, str]] = []
+        self.tenant_member_added_calls: list[dict[str, object]] = []
+        self.tenant_member_removed_calls: list[dict[str, str]] = []
+        self.tenant_members_listed_calls: list[dict[str, object]] = []
+        self.tenant_cascade_deletion_started_calls: list[dict[str, object]] = []
+
+    def tenant_created(self, tenant_id: str, name: str) -> None:
+        self.tenant_created_calls.append({"tenant_id": tenant_id, "name": name})
+
+    def tenant_retrieved(self, tenant_id: str) -> None:
+        self.tenant_retrieved_calls.append({"tenant_id": tenant_id})
+
+    def tenants_listed(self, count: int) -> None:
+        self.tenants_listed_calls.append({"count": count})
+
+    def tenant_deleted(self, tenant_id: str) -> None:
+        self.tenant_deleted_calls.append({"tenant_id": tenant_id})
+
+    def tenant_not_found(self, tenant_id: str) -> None:
+        self.tenant_not_found_calls.append({"tenant_id": tenant_id})
+
+    def duplicate_tenant_name(self, name: str) -> None:
+        self.duplicate_tenant_name_calls.append({"name": name})
+
+    def tenant_member_added(
+        self, tenant_id: str, user_id: str, role: str, added_by: str | None
+    ) -> None:
+        self.tenant_member_added_calls.append(
+            {
+                "tenant_id": tenant_id,
+                "user_id": user_id,
+                "role": role,
+                "added_by": added_by,
+            }
+        )
+
+    def tenant_member_removed(
+        self, tenant_id: str, user_id: str, removed_by: str
+    ) -> None:
+        self.tenant_member_removed_calls.append(
+            {"tenant_id": tenant_id, "user_id": user_id, "removed_by": removed_by}
+        )
+
+    def tenant_members_listed(self, tenant_id: str, member_count: int) -> None:
+        self.tenant_members_listed_calls.append(
+            {"tenant_id": tenant_id, "member_count": member_count}
+        )
+
+    def tenant_cascade_deletion_started(
+        self,
+        tenant_id: str,
+        workspaces_count: int,
+        groups_count: int,
+        api_keys_count: int,
+    ) -> None:
+        self.tenant_cascade_deletion_started_calls.append(
+            {
+                "tenant_id": tenant_id,
+                "workspaces_count": workspaces_count,
+                "groups_count": groups_count,
+                "api_keys_count": api_keys_count,
+            }
+        )
+
+    def with_context(self, context: object) -> "RecordingTenantServiceProbe":
+        """Return self — context is not used in tests."""
+        return self
diff --git a/src/api/tests/fakes/management.py b/src/api/tests/fakes/management.py
index ded5df71c..8cb6674d9 100644
--- a/src/api/tests/fakes/management.py
+++ b/src/api/tests/fakes/management.py
@@ -3,8 +3,10 @@
 Provides fast, self-contained test doubles for:
 - IKnowledgeGraphRepository
 - IDataSourceRepository
+- IDataSourceSyncRunRepository
 - ISecretStoreRepository
 - KnowledgeGraphServiceProbe
+- DataSourceServiceProbe
 
 These fakes implement the full port protocols using in-memory storage and
 record all mutation calls so tests can assert on behavior without resorting
@@ -15,6 +17,7 @@
 from __future__ import annotations
 
 from management.domain.aggregates import DataSource, KnowledgeGraph
+from management.domain.entities import DataSourceSyncRun
 from management.domain.value_objects import DataSourceId, KnowledgeGraphId
 
 
@@ -236,3 +239,104 @@ def permission_denied(
     def with_context(self, context: object) -> "RecordingKnowledgeGraphServiceProbe":
         """Return self — context is not used in tests."""
         return self
+
+
+class InMemoryDataSourceSyncRunRepository:
+    """In-memory fake implementing IDataSourceSyncRunRepository.
+
+    Stores DataSourceSyncRun entities in a dict keyed by run ID.
+    Records all save calls so tests can assert on call count and args.
+    """
+
+    def __init__(self) -> None:
+        self._store: dict[str, DataSourceSyncRun] = {}
+        self.saved: list[DataSourceSyncRun] = []
+
+    def seed(self, *sync_runs: DataSourceSyncRun) -> None:
+        """Pre-populate the store (used in test setup)."""
+        for run in sync_runs:
+            self._store[run.id] = run
+
+    async def save(self, sync_run: DataSourceSyncRun) -> None:
+        self.saved.append(sync_run)
+        self._store[sync_run.id] = sync_run
+
+    async def get_by_id(self, sync_run_id: str) -> DataSourceSyncRun | None:
+        return self._store.get(sync_run_id)
+
+    async def find_by_data_source(self, data_source_id: str) -> list[DataSourceSyncRun]:
+        return [r for r in self._store.values() if r.data_source_id == data_source_id]
+
+
+class RecordingDataSourceServiceProbe:
+    """Concrete recording probe implementing DataSourceServiceProbe protocol.
+
+    Records every method call in per-method lists so tests can assert on
+    which events were raised and what arguments were passed.  This is a
+    concrete class — NOT a MagicMock(spec=...) — as required by the testing
+    NFR (specs/nfr/testing.spec.md).
+    """
+
+    def __init__(self) -> None:
+        self.data_source_created_calls: list[dict[str, str]] = []
+        self.data_source_creation_failed_calls: list[dict[str, str]] = []
+        self.data_source_retrieved_calls: list[dict[str, str]] = []
+        self.data_source_updated_calls: list[dict[str, str]] = []
+        self.data_source_deleted_calls: list[dict[str, str]] = []
+        self.data_source_deletion_failed_calls: list[dict[str, str]] = []
+        self.data_sources_listed_calls: list[dict[str, object]] = []
+        self.sync_requested_calls: list[dict[str, str]] = []
+        self.permission_denied_calls: list[dict[str, str]] = []
+
+    def data_source_created(
+        self,
+        ds_id: str,
+        kg_id: str,
+        tenant_id: str,
+        name: str,
+    ) -> None:
+        self.data_source_created_calls.append(
+            {"ds_id": ds_id, "kg_id": kg_id, "tenant_id": tenant_id, "name": name}
+        )
+
+    def data_source_creation_failed(
+        self,
+        kg_id: str,
+        name: str,
+        error: str,
+    ) -> None:
+        self.data_source_creation_failed_calls.append(
+            {"kg_id": kg_id, "name": name, "error": error}
+        )
+
+    def data_source_retrieved(self, ds_id: str) -> None:
+        self.data_source_retrieved_calls.append({"ds_id": ds_id})
+
+    def data_source_updated(self, ds_id: str, name: str) -> None:
+        self.data_source_updated_calls.append({"ds_id": ds_id, "name": name})
+
+    def data_source_deleted(self, ds_id: str) -> None:
+        self.data_source_deleted_calls.append({"ds_id": ds_id})
+
+    def data_source_deletion_failed(self, ds_id: str, error: str) -> None:
+        self.data_source_deletion_failed_calls.append({"ds_id": ds_id, "error": error})
+
+    def data_sources_listed(self, kg_id: str, count: int) -> None:
+        self.data_sources_listed_calls.append({"kg_id": kg_id, "count": count})
+
+    def sync_requested(self, ds_id: str) -> None:
+        self.sync_requested_calls.append({"ds_id": ds_id})
+
+    def permission_denied(
+        self,
+        user_id: str,
+        resource_id: str,
+        permission: str,
+    ) -> None:
+        self.permission_denied_calls.append(
+            {"user_id": user_id, "resource_id": resource_id, "permission": permission}
+        )
+
+    def with_context(self, context: object) -> "RecordingDataSourceServiceProbe":
+        """Return self — context is not used in tests."""
+        return self
diff --git a/src/api/tests/integration/iam/test_group_service.py b/src/api/tests/integration/iam/test_group_service.py
new file mode 100644
index 000000000..88447aa42
--- /dev/null
+++ b/src/api/tests/integration/iam/test_group_service.py
@@ -0,0 +1,168 @@
+"""Integration tests for GroupService transactional atomicity.
+
+These tests verify that group deletion is atomic: if the transaction fails
+at any point, the group record is NOT deleted (no partial state).
+
+NOTE: Rollback semantics cannot be verified with mock sessions. These tests
+require a real PostgreSQL connection.
+"""
+
+from __future__ import annotations
+
+import pytest
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from iam.application.services.group_service import GroupService
+from iam.domain.aggregates import Group
+from iam.domain.value_objects import TenantId, UserId
+from iam.infrastructure.group_repository import GroupRepository
+from infrastructure.outbox.repository import OutboxRepository
+from shared_kernel.authorization.types import (
+    ResourceType,
+    format_resource,
+    format_subject,
+)
+from tests.fakes.authorization import InMemoryAuthorizationProvider
+
+pytestmark = pytest.mark.integration
+
+
+class TestGroupServiceDeleteRollback:
+    """Tests that group deletion rolls back fully on transaction failure.
+
+    GroupService.delete_group() wraps the delete inside
+    ``async with self._session.begin()``. If an exception escapes that block,
+    SQLAlchemy must roll back the entire unit of work. These tests confirm
+    that invariant holds at the real-database level.
+
+    This class verifies that the SERVICE-LEVEL transaction boundary provides
+    correct rollback semantics, not just the repository layer.
+    """
+
+    @pytest.mark.asyncio
+    async def test_group_service_delete_rollback_on_failure(
+        self,
+        async_session: AsyncSession,
+        test_tenant: TenantId,
+        clean_iam_data: None,
+    ) -> None:
+        """When group deletion fails mid-transaction, the group is not deleted.
+
+        Creates a group, starts a deletion via GroupService (which uses a
+        ``async with session.begin()`` block), injects a failure inside the
+        transaction, and asserts the group still exists after — verifying
+        full transactional rollback at the service level.
+        """
+        outbox = OutboxRepository(session=async_session)
+        authz = InMemoryAuthorizationProvider()
+        group_repo = GroupRepository(
+            session=async_session,
+            authz=authz,
+            outbox=outbox,
+        )
+
+        # Arrange: create a group for the test tenant
+        group = Group.create(name="Rollback Test Group", tenant_id=test_tenant)
+        async with async_session.begin():
+            await group_repo.save(group)
+
+        # Grant MANAGE permission so the authorization check (outside the
+        # session.begin() block) passes.
+        resource = format_resource(ResourceType.GROUP, group.id.value)
+        subject = format_subject(ResourceType.USER, "test-user")
+        await authz.write_relationship(
+            resource=resource, relation="admin", subject=subject
+        )
+
+        # Arrange: failing repo subclass that raises RuntimeError on delete()
+        class FailingOnDeleteGroupRepository(GroupRepository):
+            """Raises RuntimeError when delete() is called.
+
+            Simulates a failure that occurs inside the service's
+            ``async with session.begin()`` block, after all reads have
+            completed but before the transaction can commit.
+            """
+
+            async def delete(self, g: Group) -> bool:
+                raise RuntimeError(
+                    "Simulated group deletion failure to verify service rollback"
+                )
+
+        failing_repo = FailingOnDeleteGroupRepository(
+            session=async_session,
+            authz=authz,
+            outbox=outbox,
+        )
+
+        svc = GroupService(
+            session=async_session,
+            group_repository=failing_repo,
+            authz=authz,
+            scope_to_tenant=test_tenant,
+        )
+
+        # Act: delete must raise (the repo is wired to fail)
+        with pytest.raises(RuntimeError, match="Simulated group deletion failure"):
+            await svc.delete_group(
+                group_id=group.id,
+                user_id=UserId(value="test-user"),
+            )
+
+        # Assert: transaction rolled back — group still exists in the database
+        retrieved = await group_repo.get_by_id(group.id)
+        assert retrieved is not None, (
+            "Group must not be deleted when the service-level transaction rolls back; "
+            "the async with session.begin() block must roll back completely."
+        )
+
+    @pytest.mark.asyncio
+    async def test_group_service_delete_commits_fully_on_success(
+        self,
+        async_session: AsyncSession,
+        test_tenant: TenantId,
+        clean_iam_data: None,
+    ) -> None:
+        """When delete succeeds, the group is removed from the database.
+
+        Verifies the happy path of the service-level delete to complement the
+        rollback test — the transaction must commit and leave no orphaned rows.
+        """
+        outbox = OutboxRepository(session=async_session)
+        authz = InMemoryAuthorizationProvider()
+        group_repo = GroupRepository(
+            session=async_session,
+            authz=authz,
+            outbox=outbox,
+        )
+
+        # Arrange: create a group
+        group = Group.create(name="Happy Path Delete Group", tenant_id=test_tenant)
+        async with async_session.begin():
+            await group_repo.save(group)
+
+        # Grant MANAGE permission
+        resource = format_resource(ResourceType.GROUP, group.id.value)
+        subject = format_subject(ResourceType.USER, "test-user")
+        await authz.write_relationship(
+            resource=resource, relation="admin", subject=subject
+        )
+
+        svc = GroupService(
+            session=async_session,
+            group_repository=group_repo,
+            authz=authz,
+            scope_to_tenant=test_tenant,
+        )
+
+        # Act: successful delete
+        result = await svc.delete_group(
+            group_id=group.id,
+            user_id=UserId(value="test-user"),
+        )
+
+        # Assert: group removed from database
+        assert result is True, "service.delete_group() must return True on success"
+        retrieved = await group_repo.get_by_id(group.id)
+        assert retrieved is None, (
+            "Group must be deleted from the DB after successful delete"
+        )
diff --git a/src/api/tests/integration/iam/test_tenant_service.py b/src/api/tests/integration/iam/test_tenant_service.py
new file mode 100644
index 000000000..037aca54d
--- /dev/null
+++ b/src/api/tests/integration/iam/test_tenant_service.py
@@ -0,0 +1,188 @@
+"""Integration tests for TenantService transactional atomicity.
+
+These tests verify that tenant deletion is atomic: if the transaction fails
+at any point, the tenant record (and all cascade-deleted child records) are
+NOT deleted — no partial state is left in the database.
+
+The spec requires: 'if any step fails, the entire deletion rolls back with
+no partial state'. TenantService.delete_tenant() wraps the entire cascade
+(workspaces, groups, API keys, then the tenant itself) in a single
+``async with session.begin()`` block.
+
+NOTE: Rollback semantics cannot be verified with mock sessions. These tests
+require a real PostgreSQL connection.
+"""
+
+from __future__ import annotations
+
+import pytest
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from iam.application.services.tenant_service import TenantService
+from iam.domain.aggregates import Tenant
+from iam.domain.value_objects import UserId
+from iam.infrastructure.api_key_repository import APIKeyRepository
+from iam.infrastructure.group_repository import GroupRepository
+from iam.infrastructure.tenant_repository import TenantRepository
+from iam.infrastructure.workspace_repository import WorkspaceRepository
+from infrastructure.outbox.repository import OutboxRepository
+from shared_kernel.authorization.types import (
+    ResourceType,
+    format_resource,
+    format_subject,
+)
+from tests.fakes.authorization import InMemoryAuthorizationProvider
+
+pytestmark = pytest.mark.integration
+
+
+class TestTenantServiceDeleteRollback:
+    """Tests that tenant deletion rolls back fully on transaction failure.
+
+    TenantService.delete_tenant() wraps the full cascade deletion (workspaces,
+    groups, API keys, then the tenant itself) inside a single
+    ``async with self._session.begin()``. If an exception escapes that block,
+    SQLAlchemy must roll back the entire unit of work. These tests confirm
+    that invariant holds at the real-database level.
+
+    This class verifies SERVICE-LEVEL transactional atomicity, not just the
+    repository layer.
+    """
+
+    @pytest.mark.asyncio
+    async def test_tenant_service_delete_rollback_on_failure(
+        self,
+        async_session: AsyncSession,
+        clean_iam_data: None,
+    ) -> None:
+        """When tenant deletion fails mid-cascade, the tenant is not deleted.
+
+        Creates a tenant, starts a deletion via TenantService (which uses
+        ``async with session.begin()`` to wrap the full cascade), injects a
+        failure at the tenant's own delete() call (after the cascade steps
+        have run), and asserts the tenant still exists after — verifying
+        full transactional rollback at the service level.
+        """
+        outbox = OutboxRepository(session=async_session)
+        authz = InMemoryAuthorizationProvider()
+
+        tenant_repo = TenantRepository(session=async_session, outbox=outbox)
+        workspace_repo = WorkspaceRepository(
+            session=async_session, authz=authz, outbox=outbox
+        )
+        group_repo = GroupRepository(session=async_session, authz=authz, outbox=outbox)
+        api_key_repo = APIKeyRepository(session=async_session, outbox=outbox)
+
+        # Arrange: create a tenant for the deletion test
+        tenant = Tenant.create(name="Rollback Test Tenant")
+        async with async_session.begin():
+            await tenant_repo.save(tenant)
+
+        tenant_id = tenant.id
+
+        # Grant ADMINISTRATE permission so the authorization check (outside the
+        # session.begin() block) passes.
+        resource = format_resource(ResourceType.TENANT, tenant_id.value)
+        subject = format_subject(ResourceType.USER, "test-admin")
+        await authz.write_relationship(
+            resource=resource, relation="admin", subject=subject
+        )
+
+        # Arrange: failing tenant repo subclass that raises RuntimeError on delete()
+        class FailingOnDeleteTenantRepository(TenantRepository):
+            """Raises RuntimeError when delete() is called.
+
+            Simulates a failure that occurs AFTER all cascade deletions have run
+            (workspaces, groups, API keys) but BEFORE the tenant row is removed.
+            The service's ``async with session.begin()`` block must roll back
+            all writes made so far.
+            """
+
+            async def delete(self, t: Tenant) -> bool:
+                raise RuntimeError(
+                    "Simulated tenant deletion failure to verify service rollback"
+                )
+
+        failing_tenant_repo = FailingOnDeleteTenantRepository(
+            session=async_session, outbox=outbox
+        )
+
+        svc = TenantService(
+            tenant_repository=failing_tenant_repo,
+            workspace_repository=workspace_repo,
+            group_repository=group_repo,
+            api_key_repository=api_key_repo,
+            authz=authz,
+            session=async_session,
+        )
+
+        # Act: delete must raise (the tenant repo is wired to fail)
+        with pytest.raises(RuntimeError, match="Simulated tenant deletion failure"):
+            await svc.delete_tenant(
+                tenant_id=tenant_id,
+                requesting_user_id=UserId(value="test-admin"),
+            )
+
+        # Assert: transaction rolled back — tenant still exists in the database
+        retrieved = await tenant_repo.get_by_id(tenant_id)
+        assert retrieved is not None, (
+            "Tenant must not be deleted when the service-level transaction rolls back; "
+            "the async with session.begin() block must roll back completely."
+        )
+
+    @pytest.mark.asyncio
+    async def test_tenant_service_delete_commits_fully_on_success(
+        self,
+        async_session: AsyncSession,
+        clean_iam_data: None,
+    ) -> None:
+        """When tenant deletion succeeds, the tenant is removed from the database.
+
+        Verifies the happy path of the service-level cascade to complement the
+        rollback test — the transaction must commit and leave no orphaned rows.
+        """
+        outbox = OutboxRepository(session=async_session)
+        authz = InMemoryAuthorizationProvider()
+
+        tenant_repo = TenantRepository(session=async_session, outbox=outbox)
+        workspace_repo = WorkspaceRepository(
+            session=async_session, authz=authz, outbox=outbox
+        )
+        group_repo = GroupRepository(session=async_session, authz=authz, outbox=outbox)
+        api_key_repo = APIKeyRepository(session=async_session, outbox=outbox)
+
+        # Arrange: create a tenant to delete
+        tenant = Tenant.create(name="Happy Path Delete Tenant")
+        async with async_session.begin():
+            await tenant_repo.save(tenant)
+
+        tenant_id = tenant.id
+
+        # Grant ADMINISTRATE permission
+        resource = format_resource(ResourceType.TENANT, tenant_id.value)
+        subject = format_subject(ResourceType.USER, "test-admin")
+        await authz.write_relationship(
+            resource=resource, relation="admin", subject=subject
+        )
+
+        svc = TenantService(
+            tenant_repository=tenant_repo,
+            workspace_repository=workspace_repo,
+            group_repository=group_repo,
+            api_key_repository=api_key_repo,
+            authz=authz,
+            session=async_session,
+        )
+
+        # Act: successful delete
+        result = await svc.delete_tenant(
+            tenant_id=tenant_id,
+            requesting_user_id=UserId(value="test-admin"),
+        )
+
+        # Assert: tenant removed from database
+        assert result is True, "service.delete_tenant() must return True on success"
+        retrieved = await tenant_repo.get_by_id(tenant_id)
+        assert retrieved is None, (
+            "Tenant must be deleted from the DB after successful delete"
+        )
diff --git a/src/api/tests/integration/management/test_data_source_service.py b/src/api/tests/integration/management/test_data_source_service.py
new file mode 100644
index 000000000..238e0e4df
--- /dev/null
+++ b/src/api/tests/integration/management/test_data_source_service.py
@@ -0,0 +1,210 @@
+"""Integration tests for DataSourceService transactional atomicity.
+
+These tests verify that data source deletion is atomic: if the transaction
+fails at any point, the data source record is NOT deleted — no partial state
+is left in the database.
+
+DataSourceService.delete() wraps the deletion (credential cleanup + DS delete)
+inside a single ``async with session.begin()`` block.
+
+NOTE: Rollback semantics cannot be verified with mock sessions. These tests
+require a real PostgreSQL connection.
+"""
+
+from __future__ import annotations
+
+import pytest
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from management.application.observability import DefaultDataSourceServiceProbe
+from management.application.services.data_source_service import DataSourceService
+from management.domain.aggregates import DataSource, KnowledgeGraph
+from management.infrastructure.repositories.data_source_repository import (
+    DataSourceRepository,
+)
+from management.infrastructure.repositories.data_source_sync_run_repository import (
+    DataSourceSyncRunRepository,
+)
+from management.infrastructure.repositories.knowledge_graph_repository import (
+    KnowledgeGraphRepository,
+)
+from shared_kernel.authorization.types import (
+    ResourceType,
+    format_resource,
+    format_subject,
+)
+from shared_kernel.datasource_types import DataSourceAdapterType
+from tests.fakes.authorization import InMemoryAuthorizationProvider
+from tests.fakes.management import InMemorySecretStoreRepository
+
+pytestmark = pytest.mark.integration
+
+
+class TestDataSourceServiceDeleteRollback:
+    """Integration tests for DataSourceService.delete() transactional atomicity.
+
+    The spec requires: 'if any step fails, the entire deletion rolls back with
+    no partial state'. These tests exercise the FULL service path — including
+    the ``async with self._session.begin()`` boundary in
+    ``DataSourceService.delete()`` — with real SQLAlchemy sessions.
+    """
+
+    @pytest.mark.asyncio
+    async def test_data_source_service_delete_rollback_on_failure(
+        self,
+        knowledge_graph_repository: KnowledgeGraphRepository,
+        data_source_repository: DataSourceRepository,
+        data_source_sync_run_repository: DataSourceSyncRunRepository,
+        async_session: AsyncSession,
+        test_tenant: str,
+        test_workspace: str,
+        clean_management_data: None,
+    ) -> None:
+        """When DS deletion fails after credential cleanup, the DS is not deleted.
+
+        Simulates a hard failure in ``DataSourceRepository.delete()`` AFTER
+        credential cleanup has run within the same
+        ``async with session.begin()`` block in ``DataSourceService.delete()``.
+        The SQLAlchemy context manager must roll back all writes in the block.
+        """
+        authz = InMemoryAuthorizationProvider()
+        secret_store = InMemorySecretStoreRepository()
+
+        # --- Arrange: create KG and DS ---
+        kg = KnowledgeGraph.create(
+            tenant_id=test_tenant,
+            workspace_id=test_workspace,
+            name="Service DS Rollback KG",
+            description="Verifies service-level data source atomicity",
+        )
+        ds = DataSource.create(
+            knowledge_graph_id=kg.id.value,
+            tenant_id=test_tenant,
+            name="Service DS Rollback Test",
+            adapter_type=DataSourceAdapterType.GITHUB,
+            connection_config={"repo": "org/repo", "branch": "main"},
+        )
+
+        async with async_session.begin():
+            await knowledge_graph_repository.save(kg)
+
+        async with async_session.begin():
+            await data_source_repository.save(ds)
+
+        # --- Arrange: grant MANAGE permission on the data source ---
+        resource = format_resource(ResourceType.DATA_SOURCE, ds.id.value)
+        subject = format_subject(ResourceType.USER, "test-user")
+        # data_source has no special permission expansion — "manage" is a direct relation
+        await authz.write_relationship(
+            resource=resource, relation="manage", subject=subject
+        )
+
+        # --- Arrange: DS repo subclass that raises during delete() ---
+        class FailingOnDeleteDataSourceRepository(DataSourceRepository):
+            """Raises RuntimeError when delete() is called.
+
+            Simulates a failure that occurs AFTER all pre-delete steps have run
+            (credential cleanup, mark_for_deletion) but BEFORE the DS row is
+            removed. The service's ``async with session.begin()`` block must
+            roll back all writes made so far.
+            """
+
+            async def delete(self, data_source: DataSource) -> bool:
+                raise RuntimeError(
+                    "Simulated DS deletion failure to verify service rollback"
+                )
+
+        from infrastructure.outbox.repository import OutboxRepository
+
+        outbox = OutboxRepository(session=async_session)
+        failing_ds_repo = FailingOnDeleteDataSourceRepository(
+            session=async_session, outbox=outbox
+        )
+
+        svc = DataSourceService(
+            session=async_session,
+            data_source_repository=failing_ds_repo,
+            knowledge_graph_repository=knowledge_graph_repository,
+            secret_store=secret_store,
+            sync_run_repository=data_source_sync_run_repository,
+            authz=authz,
+            scope_to_tenant=test_tenant,
+            probe=DefaultDataSourceServiceProbe(),
+        )
+
+        # --- Act: delete must raise (the DS repo is wired to fail) ---
+        with pytest.raises(RuntimeError, match="Simulated DS deletion failure"):
+            await svc.delete(user_id="test-user", ds_id=ds.id.value)
+
+        # --- Assert: transaction rolled back — DS still exists ---
+        retrieved_ds = await data_source_repository.get_by_id(ds.id)
+        assert retrieved_ds is not None, (
+            "DataSource must survive when service-level delete fails; "
+            "the async with session.begin() block must roll back completely."
+        )
+
+    @pytest.mark.asyncio
+    async def test_data_source_service_delete_commits_fully_on_success(
+        self,
+        knowledge_graph_repository: KnowledgeGraphRepository,
+        data_source_repository: DataSourceRepository,
+        data_source_sync_run_repository: DataSourceSyncRunRepository,
+        async_session: AsyncSession,
+        test_tenant: str,
+        test_workspace: str,
+        clean_management_data: None,
+    ) -> None:
+        """When delete succeeds, the data source is removed from the database.
+
+        Verifies the happy path of the service-level delete to complement the
+        rollback test — the transaction must commit and leave no orphaned rows.
+        """
+        authz = InMemoryAuthorizationProvider()
+        secret_store = InMemorySecretStoreRepository()
+
+        kg = KnowledgeGraph.create(
+            tenant_id=test_tenant,
+            workspace_id=test_workspace,
+            name="Service DS Delete Commit KG",
+            description="Verifies successful service-level DS deletion",
+        )
+        ds = DataSource.create(
+            knowledge_graph_id=kg.id.value,
+            tenant_id=test_tenant,
+            name="Service DS Delete Commit",
+            adapter_type=DataSourceAdapterType.GITHUB,
+            connection_config={"repo": "org/repo", "branch": "main"},
+        )
+
+        async with async_session.begin():
+            await knowledge_graph_repository.save(kg)
+
+        async with async_session.begin():
+            await data_source_repository.save(ds)
+
+        # Grant MANAGE permission
+        resource = format_resource(ResourceType.DATA_SOURCE, ds.id.value)
+        subject = format_subject(ResourceType.USER, "test-user")
+        await authz.write_relationship(
+            resource=resource, relation="manage", subject=subject
+        )
+
+        svc = DataSourceService(
+            session=async_session,
+            data_source_repository=data_source_repository,
+            knowledge_graph_repository=knowledge_graph_repository,
+            secret_store=secret_store,
+            sync_run_repository=data_source_sync_run_repository,
+            authz=authz,
+            scope_to_tenant=test_tenant,
+            probe=DefaultDataSourceServiceProbe(),
+        )
+
+        result = await svc.delete(user_id="test-user", ds_id=ds.id.value)
+
+        assert result is True, "service.delete() must return True on success"
+
+        retrieved_ds = await data_source_repository.get_by_id(ds.id)
+        assert retrieved_ds is None, (
+            "DataSource must be deleted from the DB after successful delete"
+        )
diff --git a/src/api/tests/unit/iam/application/test_tenant_service.py b/src/api/tests/unit/iam/application/test_tenant_service.py
index a16c29e8a..63f904a75 100644
--- a/src/api/tests/unit/iam/application/test_tenant_service.py
+++ b/src/api/tests/unit/iam/application/test_tenant_service.py
@@ -6,6 +6,8 @@
 
 from unittest.mock import AsyncMock, Mock, patch
 
+from tests.fakes.iam import RecordingTenantServiceProbe
+
 import pytest
 from sqlalchemy.ext.asyncio import AsyncSession
 
@@ -1744,7 +1746,6 @@ async def test_cascade_deletion_probe_reports_api_key_count(
         API keys are being deleted for a given tenant deletion.
         """
         from datetime import UTC, datetime, timedelta
-        from unittest.mock import MagicMock
 
         tenant_id = TenantId.generate()
         admin_id = UserId.from_string("admin-456")
@@ -1767,10 +1768,8 @@ async def test_cascade_deletion_probe_reports_api_key_count(
             expires_at=datetime.now(UTC) + timedelta(days=30),
         )
 
-        # Use a mock probe to capture probe events
-        mock_probe = MagicMock()
-        mock_probe.tenant_cascade_deletion_started = MagicMock()
-        mock_probe.tenant_deleted = MagicMock()
+        # Use a recording probe (concrete class — no MagicMock) to capture events.
+        recording_probe = RecordingTenantServiceProbe()
 
         service_with_probe = TenantService(
             tenant_repository=mock_tenant_repo,
@@ -1779,7 +1778,7 @@ async def test_cascade_deletion_probe_reports_api_key_count(
             api_key_repository=mock_api_key_repo,
             authz=mock_authz,
             session=mock_session,
-            probe=mock_probe,
+            probe=recording_probe,
         )
 
         mock_authz.check_permission = AsyncMock(return_value=True)
@@ -1793,13 +1792,17 @@ async def test_cascade_deletion_probe_reports_api_key_count(
 
         await service_with_probe.delete_tenant(tenant_id, requesting_user_id=admin_id)
 
-        # Verify the probe was called with the correct API key count
-        mock_probe.tenant_cascade_deletion_started.assert_called_once_with(
-            tenant_id=tenant_id.value,
-            workspaces_count=0,
-            groups_count=0,
-            api_keys_count=2,
-        )
+        # Verify the recording probe received the correct cascade deletion event.
+        assert len(recording_probe.tenant_cascade_deletion_started_calls) == 1
+        call = recording_probe.tenant_cascade_deletion_started_calls[0]
+        assert call["tenant_id"] == tenant_id.value
+        assert call["workspaces_count"] == 0
+        assert call["groups_count"] == 0
+        assert call["api_keys_count"] == 2
+        # Also verify the tenant_deleted probe event fired (delete completed).
+        assert len(recording_probe.tenant_deleted_calls) == 1
+        deleted_call = recording_probe.tenant_deleted_calls[0]
+        assert deleted_call["tenant_id"] == tenant_id.value
 
     @pytest.mark.asyncio
     async def test_api_keys_deleted_before_tenant_on_cascade(
diff --git a/src/api/tests/unit/management/application/test_data_source_service.py b/src/api/tests/unit/management/application/test_data_source_service.py
index e1d044f33..e5ac0228e 100644
--- a/src/api/tests/unit/management/application/test_data_source_service.py
+++ b/src/api/tests/unit/management/application/test_data_source_service.py
@@ -2,13 +2,16 @@
 
 Tests verify authorization checks, repository interactions,
 credential storage, transaction management, and observability probe calls.
+
+Collaborators use in-memory fakes (no MagicMock/AsyncMock for domain or
+application-layer ports) per specs/nfr/testing.spec.md.
 """
 
 from __future__ import annotations
 
 from contextlib import asynccontextmanager
 from datetime import UTC, datetime
-from unittest.mock import AsyncMock, MagicMock
+from unittest.mock import MagicMock
 
 import pytest
 
@@ -21,13 +24,25 @@
     ScheduleType,
 )
 from management.ports.exceptions import UnauthorizedError
-from shared_kernel.authorization.types import Permission
 from shared_kernel.datasource_types import DataSourceAdapterType
+from tests.fakes.authorization import InMemoryAuthorizationProvider
+from tests.fakes.management import (
+    InMemoryDataSourceRepository,
+    InMemoryDataSourceSyncRunRepository,
+    InMemoryKnowledgeGraphRepository,
+    InMemorySecretStoreRepository,
+    RecordingDataSourceServiceProbe,
+)
 
 
 @pytest.fixture
 def mock_session():
-    """Create a mock AsyncSession with begin() context manager."""
+    """Create a mock AsyncSession with begin() context manager.
+
+    The session is mocked at the infrastructure boundary — unit tests of
+    the application service do not exercise real SQLAlchemy transaction
+    semantics.  Rollback behaviour is covered by integration tests.
+    """
     session = MagicMock()
 
     @asynccontextmanager
@@ -39,33 +54,39 @@ async def _begin():
 
 
 @pytest.fixture
-def mock_ds_repo():
-    return AsyncMock()
+def ds_repo():
+    """In-memory DataSource repository."""
+    return InMemoryDataSourceRepository()
 
 
 @pytest.fixture
-def mock_kg_repo():
-    return AsyncMock()
+def kg_repo():
+    """In-memory KnowledgeGraph repository."""
+    return InMemoryKnowledgeGraphRepository()
 
 
 @pytest.fixture
-def mock_secret_store():
-    return AsyncMock()
+def secret_store():
+    """In-memory secret store."""
+    return InMemorySecretStoreRepository()
 
 
 @pytest.fixture
-def mock_sync_run_repo():
-    return AsyncMock()
+def sync_run_repo():
+    """In-memory sync run repository."""
+    return InMemoryDataSourceSyncRunRepository()
 
 
 @pytest.fixture
-def mock_authz():
-    return AsyncMock()
+def authz():
+    """In-memory authorization provider (full fake, no mocks)."""
+    return InMemoryAuthorizationProvider()
 
 
 @pytest.fixture
-def mock_probe():
-    return MagicMock()
+def probe():
+    """Concrete recording probe (no MagicMock)."""
+    return RecordingDataSourceServiceProbe()
 
 
 @pytest.fixture
@@ -86,23 +107,24 @@ def kg_id():
 @pytest.fixture
 def service(
     mock_session,
-    mock_ds_repo,
-    mock_kg_repo,
-    mock_secret_store,
-    mock_sync_run_repo,
-    mock_authz,
-    mock_probe,
+    ds_repo,
+    kg_repo,
+    secret_store,
+    sync_run_repo,
+    authz,
+    probe,
     tenant_id,
 ):
+    """DataSourceService wired with in-memory fakes."""
     return DataSourceService(
         session=mock_session,
-        data_source_repository=mock_ds_repo,
-        knowledge_graph_repository=mock_kg_repo,
-        secret_store=mock_secret_store,
-        sync_run_repository=mock_sync_run_repo,
-        authz=mock_authz,
+        data_source_repository=ds_repo,
+        knowledge_graph_repository=kg_repo,
+        secret_store=secret_store,
+        sync_run_repository=sync_run_repo,
+        authz=authz,
         scope_to_tenant=tenant_id,
-        probe=mock_probe,
+        probe=probe,
     )
 
 
@@ -150,6 +172,50 @@ def _make_ds(
     return ds
 
 
+# ---------------------------------------------------------------------------
+# Authorization helpers
+# ---------------------------------------------------------------------------
+
+
+async def _grant_kg_edit(
+    authz: InMemoryAuthorizationProvider, kg_id: str, user_id: str
+) -> None:
+    """Grant EDIT permission on a knowledge graph to a user (editor relation)."""
+    await authz.write_relationship(
+        f"knowledge_graph:{kg_id}", "editor", f"user:{user_id}"
+    )
+
+
+async def _grant_kg_view(
+    authz: InMemoryAuthorizationProvider, kg_id: str, user_id: str
+) -> None:
+    """Grant VIEW-only permission on a knowledge graph to a user (viewer relation)."""
+    await authz.write_relationship(
+        f"knowledge_graph:{kg_id}", "viewer", f"user:{user_id}"
+    )
+
+
+async def _grant_ds_view(
+    authz: InMemoryAuthorizationProvider, ds_id: str, user_id: str
+) -> None:
+    """Grant VIEW permission on a data source to a user."""
+    await authz.write_relationship(f"data_source:{ds_id}", "view", f"user:{user_id}")
+
+
+async def _grant_ds_edit(
+    authz: InMemoryAuthorizationProvider, ds_id: str, user_id: str
+) -> None:
+    """Grant EDIT permission on a data source to a user."""
+    await authz.write_relationship(f"data_source:{ds_id}", "edit", f"user:{user_id}")
+
+
+async def _grant_ds_manage(
+    authz: InMemoryAuthorizationProvider, ds_id: str, user_id: str
+) -> None:
+    """Grant MANAGE permission on a data source to a user."""
+    await authz.write_relationship(f"data_source:{ds_id}", "manage", f"user:{user_id}")
+
+
 # ---- create ----
 
 
@@ -157,34 +223,40 @@ class TestDataSourceServiceCreate:
     """Tests for DataSourceService.create."""
 
     @pytest.mark.asyncio
-    async def test_create_checks_edit_permission_on_kg(
-        self, service, mock_authz, user_id, kg_id, mock_kg_repo, tenant_id
+    async def test_create_requires_edit_permission_on_kg(
+        self, service, authz, kg_repo, user_id, kg_id, tenant_id
     ):
-        """create() must check EDIT permission on the knowledge graph."""
-        mock_authz.check_permission.return_value = True
-        mock_kg_repo.get_by_id.return_value = _make_kg(kg_id=kg_id, tenant_id=tenant_id)
+        """create() requires EDIT permission on the knowledge graph — VIEW alone fails."""
+        kg_repo.seed(_make_kg(kg_id=kg_id, tenant_id=tenant_id))
 
-        await service.create(
+        # Viewer role only grants VIEW, not EDIT — create must be denied.
+        await _grant_kg_view(authz, kg_id, user_id)
+        with pytest.raises(UnauthorizedError):
+            await service.create(
+                user_id=user_id,
+                kg_id=kg_id,
+                name="DS-view-only",
+                adapter_type=DataSourceAdapterType.GITHUB,
+                connection_config={"url": "https://github.com"},
+            )
+
+        # Editor role grants EDIT — create must succeed.
+        await _grant_kg_edit(authz, kg_id, user_id)
+        result = await service.create(
             user_id=user_id,
             kg_id=kg_id,
-            name="My DS",
+            name="DS-with-edit",
             adapter_type=DataSourceAdapterType.GITHUB,
             connection_config={"url": "https://github.com"},
         )
-
-        mock_authz.check_permission.assert_called_once_with(
-            resource=f"knowledge_graph:{kg_id}",
-            permission=Permission.EDIT,
-            subject=f"user:{user_id}",
-        )
+        assert result is not None
 
     @pytest.mark.asyncio
     async def test_create_raises_unauthorized_when_permission_denied(
-        self, service, mock_authz, mock_probe, user_id, kg_id
+        self, service, probe, user_id, kg_id
     ):
         """create() raises UnauthorizedError when user lacks EDIT on KG."""
-        mock_authz.check_permission.return_value = False
-
+        # No authz relationship written → permission denied.
         with pytest.raises(UnauthorizedError):
             await service.create(
                 user_id=user_id,
@@ -194,15 +266,15 @@ async def test_create_raises_unauthorized_when_permission_denied(
                 connection_config={"url": "https://github.com"},
             )
 
-        mock_probe.permission_denied.assert_called_once()
+        assert len(probe.permission_denied_calls) == 1
 
     @pytest.mark.asyncio
     async def test_create_verifies_kg_exists_and_belongs_to_tenant(
-        self, service, mock_authz, mock_kg_repo, user_id, kg_id
+        self, service, authz, user_id, kg_id
     ):
         """create() raises ValueError when KG not found."""
-        mock_authz.check_permission.return_value = True
-        mock_kg_repo.get_by_id.return_value = None
+        await _grant_kg_edit(authz, kg_id, user_id)
+        # kg_repo is empty — KG not found.
 
         with pytest.raises(ValueError, match="not found"):
             await service.create(
@@ -215,13 +287,11 @@ async def test_create_verifies_kg_exists_and_belongs_to_tenant(
 
     @pytest.mark.asyncio
     async def test_create_rejects_kg_from_different_tenant(
-        self, service, mock_authz, mock_kg_repo, user_id, kg_id
+        self, service, authz, kg_repo, user_id, kg_id
     ):
         """create() raises ValueError when KG belongs to different tenant."""
-        mock_authz.check_permission.return_value = True
-        mock_kg_repo.get_by_id.return_value = _make_kg(
-            kg_id=kg_id, tenant_id="other-tenant"
-        )
+        await _grant_kg_edit(authz, kg_id, user_id)
+        kg_repo.seed(_make_kg(kg_id=kg_id, tenant_id="other-tenant"))
 
         with pytest.raises(ValueError, match="different tenant"):
             await service.create(
@@ -236,17 +306,16 @@ async def test_create_rejects_kg_from_different_tenant(
     async def test_create_stores_credentials_when_provided(
         self,
         service,
-        mock_authz,
-        mock_kg_repo,
-        mock_secret_store,
-        mock_ds_repo,
+        authz,
+        kg_repo,
+        secret_store,
         user_id,
         kg_id,
         tenant_id,
     ):
         """create() stores credentials via secret store when raw_credentials provided."""
-        mock_authz.check_permission.return_value = True
-        mock_kg_repo.get_by_id.return_value = _make_kg(kg_id=kg_id, tenant_id=tenant_id)
+        await _grant_kg_edit(authz, kg_id, user_id)
+        kg_repo.seed(_make_kg(kg_id=kg_id, tenant_id=tenant_id))
         creds = {"token": "abc123"}
 
         await service.create(
@@ -258,19 +327,19 @@ async def test_create_stores_credentials_when_provided(
             raw_credentials=creds,
         )
 
-        mock_secret_store.store.assert_called_once()
-        call_kwargs = mock_secret_store.store.call_args.kwargs
-        assert "datasource/" in call_kwargs.get("path", "")
-        assert call_kwargs.get("tenant_id") == tenant_id
-        assert call_kwargs.get("credentials") == creds
+        assert len(secret_store.store_calls) == 1
+        call = secret_store.store_calls[0]
+        assert "datasource/" in call.get("path", "")
+        assert call.get("tenant_id") == tenant_id
+        assert call.get("credentials") == creds
 
     @pytest.mark.asyncio
     async def test_create_probes_success(
-        self, service, mock_authz, mock_kg_repo, mock_probe, user_id, kg_id, tenant_id
+        self, service, authz, kg_repo, probe, user_id, kg_id, tenant_id
     ):
-        """create() calls probe on success."""
-        mock_authz.check_permission.return_value = True
-        mock_kg_repo.get_by_id.return_value = _make_kg(kg_id=kg_id, tenant_id=tenant_id)
+        """create() emits a probe event on success."""
+        await _grant_kg_edit(authz, kg_id, user_id)
+        kg_repo.seed(_make_kg(kg_id=kg_id, tenant_id=tenant_id))
 
         result = await service.create(
             user_id=user_id,
@@ -280,12 +349,12 @@ async def test_create_probes_success(
             connection_config={},
         )
 
-        mock_probe.data_source_created.assert_called_once_with(
-            ds_id=result.id.value,
-            kg_id=kg_id,
-            tenant_id=tenant_id,
-            name="My DS",
-        )
+        assert len(probe.data_source_created_calls) == 1
+        call = probe.data_source_created_calls[0]
+        assert call["ds_id"] == result.id.value
+        assert call["kg_id"] == kg_id
+        assert call["tenant_id"] == tenant_id
+        assert call["name"] == "My DS"
 
 
 # ---- get ----
@@ -295,40 +364,37 @@ class TestDataSourceServiceGet:
     """Tests for DataSourceService.get."""
 
     @pytest.mark.asyncio
-    async def test_get_returns_none_when_not_found(
-        self, service, mock_ds_repo, user_id
-    ):
-        """get() returns None when DS not found."""
-        mock_ds_repo.get_by_id.return_value = None
-
+    async def test_get_returns_none_when_not_found(self, service, user_id):
+        """get() returns None when DS not found in repository."""
+        # ds_repo is empty by default
         result = await service.get(user_id=user_id, ds_id="nonexistent")
 
         assert result is None
 
     @pytest.mark.asyncio
-    async def test_get_checks_view_permission(
-        self, service, mock_authz, mock_ds_repo, user_id
+    async def test_get_requires_view_permission(
+        self, service, authz, ds_repo, user_id, tenant_id
     ):
-        """get() checks VIEW permission on the data source."""
-        ds = _make_ds()
-        mock_ds_repo.get_by_id.return_value = ds
-        mock_authz.check_permission.return_value = True
+        """get() requires VIEW permission on the data source — no permission → None."""
+        ds = _make_ds(tenant_id=tenant_id)
+        ds_repo.seed(ds)
 
-        await service.get(user_id=user_id, ds_id=ds.id.value)
+        # No VIEW permission granted → should return None (no existence leakage).
+        result = await service.get(user_id=user_id, ds_id=ds.id.value)
+        assert result is None
 
-        mock_authz.check_permission.assert_called_once_with(
-            resource=f"data_source:{ds.id.value}",
-            permission=Permission.VIEW,
-            subject=f"user:{user_id}",
-        )
+        # VIEW permission granted → should return the aggregate.
+        await _grant_ds_view(authz, ds.id.value, user_id)
+        result = await service.get(user_id=user_id, ds_id=ds.id.value)
+        assert result is ds
 
     @pytest.mark.asyncio
     async def test_get_returns_none_for_different_tenant(
-        self, service, mock_ds_repo, user_id
+        self, service, ds_repo, user_id
     ):
         """get() returns None when DS belongs to a different tenant."""
         ds = _make_ds(tenant_id="other-tenant")
-        mock_ds_repo.get_by_id.return_value = ds
+        ds_repo.seed(ds)
 
         result = await service.get(user_id=user_id, ds_id=ds.id.value)
 
@@ -336,12 +402,12 @@ async def test_get_returns_none_for_different_tenant(
 
     @pytest.mark.asyncio
     async def test_get_returns_none_when_permission_denied(
-        self, service, mock_authz, mock_ds_repo, user_id
+        self, service, ds_repo, user_id, tenant_id
     ):
         """get() returns None when user lacks VIEW (no existence leakage)."""
-        ds = _make_ds()
-        mock_ds_repo.get_by_id.return_value = ds
-        mock_authz.check_permission.return_value = False
+        ds = _make_ds(tenant_id=tenant_id)
+        ds_repo.seed(ds)
+        # No permission written → denied.
 
         result = await service.get(user_id=user_id, ds_id=ds.id.value)
 
@@ -349,17 +415,18 @@ async def test_get_returns_none_when_permission_denied(
 
     @pytest.mark.asyncio
     async def test_get_returns_aggregate_on_success(
-        self, service, mock_authz, mock_ds_repo, mock_probe, user_id
+        self, service, authz, ds_repo, probe, user_id, tenant_id
     ):
-        """get() returns the aggregate when authorized."""
-        ds = _make_ds()
-        mock_ds_repo.get_by_id.return_value = ds
-        mock_authz.check_permission.return_value = True
+        """get() returns the aggregate and emits probe event when authorized."""
+        ds = _make_ds(tenant_id=tenant_id)
+        ds_repo.seed(ds)
+        await _grant_ds_view(authz, ds.id.value, user_id)
 
         result = await service.get(user_id=user_id, ds_id=ds.id.value)
 
         assert result is ds
-        mock_probe.data_source_retrieved.assert_called_once_with(ds_id=ds.id.value)
+        assert len(probe.data_source_retrieved_calls) == 1
+        assert probe.data_source_retrieved_calls[0]["ds_id"] == ds.id.value
 
 
 # ---- list_for_knowledge_graph ----
@@ -369,52 +436,46 @@ class TestDataSourceServiceListForKnowledgeGraph:
     """Tests for DataSourceService.list_for_knowledge_graph."""
 
     @pytest.mark.asyncio
-    async def test_list_checks_view_permission_on_kg(
-        self, service, mock_authz, mock_ds_repo, mock_kg_repo, user_id, kg_id, tenant_id
+    async def test_list_requires_view_permission_on_kg(
+        self, service, authz, kg_repo, ds_repo, user_id, kg_id, tenant_id
     ):
-        """list_for_knowledge_graph() checks VIEW on the KG."""
-        mock_authz.check_permission.return_value = True
-        mock_kg_repo.get_by_id.return_value = _make_kg(kg_id=kg_id, tenant_id=tenant_id)
-        mock_ds_repo.find_by_knowledge_graph.return_value = []
+        """list_for_knowledge_graph() requires VIEW on the KG."""
+        kg_repo.seed(_make_kg(kg_id=kg_id, tenant_id=tenant_id))
 
-        await service.list_for_knowledge_graph(user_id=user_id, kg_id=kg_id)
+        # No VIEW permission → raises UnauthorizedError.
+        with pytest.raises(UnauthorizedError):
+            await service.list_for_knowledge_graph(user_id=user_id, kg_id=kg_id)
 
-        mock_authz.check_permission.assert_called_once_with(
-            resource=f"knowledge_graph:{kg_id}",
-            permission=Permission.VIEW,
-            subject=f"user:{user_id}",
-        )
+        # Granting VIEW (via viewer relation) → succeeds.
+        await _grant_kg_view(authz, kg_id, user_id)
+        result = await service.list_for_knowledge_graph(user_id=user_id, kg_id=kg_id)
+        assert isinstance(result, list)
 
     @pytest.mark.asyncio
-    async def test_list_raises_unauthorized_when_denied(
-        self, service, mock_authz, user_id, kg_id
-    ):
+    async def test_list_raises_unauthorized_when_denied(self, service, user_id, kg_id):
         """list_for_knowledge_graph() raises UnauthorizedError when denied."""
-        mock_authz.check_permission.return_value = False
-
+        # No authz relationship → denied.
         with pytest.raises(UnauthorizedError):
             await service.list_for_knowledge_graph(user_id=user_id, kg_id=kg_id)
 
     @pytest.mark.asyncio
     async def test_list_raises_unauthorized_when_kg_not_found(
-        self, service, mock_authz, mock_kg_repo, user_id, kg_id
+        self, service, authz, user_id, kg_id
     ):
         """list_for_knowledge_graph() raises UnauthorizedError when KG not found."""
-        mock_authz.check_permission.return_value = True
-        mock_kg_repo.get_by_id.return_value = None
+        await _grant_kg_view(authz, kg_id, user_id)
+        # kg_repo is empty — KG not found.
 
         with pytest.raises(UnauthorizedError, match="not accessible"):
             await service.list_for_knowledge_graph(user_id=user_id, kg_id=kg_id)
 
     @pytest.mark.asyncio
     async def test_list_raises_unauthorized_for_different_tenant_kg(
-        self, service, mock_authz, mock_kg_repo, user_id, kg_id
+        self, service, authz, kg_repo, user_id, kg_id
     ):
         """list_for_knowledge_graph() rejects KG belonging to different tenant."""
-        mock_authz.check_permission.return_value = True
-        mock_kg_repo.get_by_id.return_value = _make_kg(
-            kg_id=kg_id, tenant_id="other-tenant"
-        )
+        await _grant_kg_view(authz, kg_id, user_id)
+        kg_repo.seed(_make_kg(kg_id=kg_id, tenant_id="other-tenant"))
 
         with pytest.raises(UnauthorizedError, match="not accessible"):
             await service.list_for_knowledge_graph(user_id=user_id, kg_id=kg_id)
@@ -423,28 +484,27 @@ async def test_list_raises_unauthorized_for_different_tenant_kg(
     async def test_list_returns_data_sources(
         self,
         service,
-        mock_authz,
-        mock_ds_repo,
-        mock_kg_repo,
-        mock_probe,
+        authz,
+        ds_repo,
+        kg_repo,
+        probe,
         user_id,
         kg_id,
         tenant_id,
     ):
         """list_for_knowledge_graph() returns data sources from repo."""
-        mock_authz.check_permission.return_value = True
-        mock_kg_repo.get_by_id.return_value = _make_kg(kg_id=kg_id, tenant_id=tenant_id)
-        ds1 = _make_ds(ds_id="ds-001")
-        ds2 = _make_ds(ds_id="ds-002")
-        mock_ds_repo.find_by_knowledge_graph.return_value = [ds1, ds2]
+        await _grant_kg_view(authz, kg_id, user_id)
+        kg_repo.seed(_make_kg(kg_id=kg_id, tenant_id=tenant_id))
+        ds1 = _make_ds(ds_id="ds-001", kg_id=kg_id, tenant_id=tenant_id)
+        ds2 = _make_ds(ds_id="ds-002", kg_id=kg_id, tenant_id=tenant_id)
+        ds_repo.seed(ds1, ds2)
 
         result = await service.list_for_knowledge_graph(user_id=user_id, kg_id=kg_id)
 
         assert len(result) == 2
-        mock_probe.data_sources_listed.assert_called_once_with(
-            kg_id=kg_id,
-            count=2,
-        )
+        assert len(probe.data_sources_listed_calls) == 1
+        assert probe.data_sources_listed_calls[0]["kg_id"] == kg_id
+        assert probe.data_sources_listed_calls[0]["count"] == 2
 
 
 # ---- update ----
@@ -454,34 +514,36 @@ class TestDataSourceServiceUpdate:
     """Tests for DataSourceService.update."""
 
     @pytest.mark.asyncio
-    async def test_update_checks_edit_permission_on_ds(
-        self, service, mock_authz, mock_ds_repo, user_id
+    async def test_update_requires_edit_permission_on_ds(
+        self, service, authz, ds_repo, user_id, tenant_id
     ):
-        """update() checks EDIT permission on the data source."""
-        ds = _make_ds()
-        mock_authz.check_permission.return_value = True
-        mock_ds_repo.get_by_id.return_value = ds
+        """update() requires EDIT permission on the data source."""
+        ds = _make_ds(tenant_id=tenant_id)
+        ds_repo.seed(ds)
 
-        await service.update(
+        # No permission → raises UnauthorizedError.
+        with pytest.raises(UnauthorizedError):
+            await service.update(
+                user_id=user_id,
+                ds_id=ds.id.value,
+                name="Updated",
+                connection_config={"url": "https://new.com"},
+            )
+
+        # EDIT permission → succeeds.
+        await _grant_ds_edit(authz, ds.id.value, user_id)
+        result = await service.update(
             user_id=user_id,
             ds_id=ds.id.value,
             name="Updated",
             connection_config={"url": "https://new.com"},
         )
-
-        mock_authz.check_permission.assert_called_once_with(
-            resource=f"data_source:{ds.id.value}",
-            permission=Permission.EDIT,
-            subject=f"user:{user_id}",
-        )
+        assert result is not None
 
     @pytest.mark.asyncio
-    async def test_update_raises_unauthorized_when_denied(
-        self, service, mock_authz, user_id
-    ):
+    async def test_update_raises_unauthorized_when_denied(self, service, user_id):
         """update() raises UnauthorizedError when denied."""
-        mock_authz.check_permission.return_value = False
-
+        # No permission written → denied.
         with pytest.raises(UnauthorizedError):
             await service.update(
                 user_id=user_id,
@@ -491,11 +553,11 @@ async def test_update_raises_unauthorized_when_denied(
 
     @pytest.mark.asyncio
     async def test_update_raises_value_error_when_not_found(
-        self, service, mock_authz, mock_ds_repo, user_id
+        self, service, authz, user_id
     ):
         """update() raises ValueError when DS not found."""
-        mock_authz.check_permission.return_value = True
-        mock_ds_repo.get_by_id.return_value = None
+        await _grant_ds_edit(authz, "nonexistent", user_id)
+        # ds_repo is empty — DS not found.
 
         with pytest.raises(ValueError):
             await service.update(
@@ -506,12 +568,12 @@ async def test_update_raises_value_error_when_not_found(
 
     @pytest.mark.asyncio
     async def test_update_rejects_different_tenant(
-        self, service, mock_authz, mock_ds_repo, user_id
+        self, service, authz, ds_repo, user_id
     ):
         """update() raises ValueError when DS belongs to a different tenant."""
         ds = _make_ds(tenant_id="other-tenant")
-        mock_authz.check_permission.return_value = True
-        mock_ds_repo.get_by_id.return_value = ds
+        ds_repo.seed(ds)
+        await _grant_ds_edit(authz, ds.id.value, user_id)
 
         with pytest.raises(ValueError):
             await service.update(
@@ -522,12 +584,12 @@ async def test_update_rejects_different_tenant(
 
     @pytest.mark.asyncio
     async def test_update_stores_credentials_when_provided(
-        self, service, mock_authz, mock_ds_repo, mock_secret_store, user_id, tenant_id
+        self, service, authz, ds_repo, secret_store, user_id, tenant_id
     ):
         """update() stores credentials via secret store when raw_credentials provided."""
-        ds = _make_ds()
-        mock_authz.check_permission.return_value = True
-        mock_ds_repo.get_by_id.return_value = ds
+        ds = _make_ds(tenant_id=tenant_id)
+        ds_repo.seed(ds)
+        await _grant_ds_edit(authz, ds.id.value, user_id)
         creds = {"token": "new-token"}
 
         await service.update(
@@ -536,20 +598,20 @@ async def test_update_stores_credentials_when_provided(
             raw_credentials=creds,
         )
 
-        mock_secret_store.store.assert_called_once()
-        call_kwargs = mock_secret_store.store.call_args.kwargs
-        assert "datasource/" in call_kwargs.get("path", "")
-        assert call_kwargs.get("tenant_id") == tenant_id
-        assert call_kwargs.get("credentials") == creds
+        assert len(secret_store.store_calls) == 1
+        call = secret_store.store_calls[0]
+        assert "datasource/" in call.get("path", "")
+        assert call.get("tenant_id") == tenant_id
+        assert call.get("credentials") == creds
 
     @pytest.mark.asyncio
     async def test_update_probes_success(
-        self, service, mock_authz, mock_ds_repo, mock_probe, user_id
+        self, service, authz, ds_repo, probe, user_id, tenant_id
     ):
-        """update() probes success when name is updated."""
-        ds = _make_ds()
-        mock_authz.check_permission.return_value = True
-        mock_ds_repo.get_by_id.return_value = ds
+        """update() emits probe event when name is updated."""
+        ds = _make_ds(tenant_id=tenant_id)
+        ds_repo.seed(ds)
+        await _grant_ds_edit(authz, ds.id.value, user_id)
 
         await service.update(
             user_id=user_id,
@@ -558,10 +620,9 @@ async def test_update_probes_success(
             connection_config={"url": "https://new.com"},
         )
 
-        mock_probe.data_source_updated.assert_called_once_with(
-            ds_id=ds.id.value,
-            name="Updated",
-        )
+        assert len(probe.data_source_updated_calls) == 1
+        assert probe.data_source_updated_calls[0]["ds_id"] == ds.id.value
+        assert probe.data_source_updated_calls[0]["name"] == "Updated"
 
 
 # ---- delete ----
@@ -571,40 +632,34 @@ class TestDataSourceServiceDelete:
     """Tests for DataSourceService.delete."""
 
     @pytest.mark.asyncio
-    async def test_delete_checks_manage_permission_on_ds(
-        self, service, mock_authz, mock_ds_repo, user_id
+    async def test_delete_requires_manage_permission_on_ds(
+        self, service, authz, ds_repo, user_id, tenant_id
     ):
-        """delete() checks MANAGE permission on the data source."""
-        ds = _make_ds()
-        mock_authz.check_permission.return_value = True
-        mock_ds_repo.get_by_id.return_value = ds
-        mock_ds_repo.delete.return_value = True
+        """delete() requires MANAGE permission on the data source."""
+        ds = _make_ds(tenant_id=tenant_id)
+        ds_repo.seed(ds)
 
-        await service.delete(user_id=user_id, ds_id=ds.id.value)
+        # No permission → raises UnauthorizedError.
+        with pytest.raises(UnauthorizedError):
+            await service.delete(user_id=user_id, ds_id=ds.id.value)
 
-        mock_authz.check_permission.assert_called_once_with(
-            resource=f"data_source:{ds.id.value}",
-            permission=Permission.MANAGE,
-            subject=f"user:{user_id}",
-        )
+        # MANAGE permission → succeeds.
+        await _grant_ds_manage(authz, ds.id.value, user_id)
+        result = await service.delete(user_id=user_id, ds_id=ds.id.value)
+        assert result is True
 
     @pytest.mark.asyncio
-    async def test_delete_raises_unauthorized_when_denied(
-        self, service, mock_authz, user_id
-    ):
+    async def test_delete_raises_unauthorized_when_denied(self, service, user_id):
         """delete() raises UnauthorizedError when denied."""
-        mock_authz.check_permission.return_value = False
-
+        # No permission written → denied.
         with pytest.raises(UnauthorizedError):
             await service.delete(user_id=user_id, ds_id="ds-001")
 
     @pytest.mark.asyncio
-    async def test_delete_returns_false_when_not_found(
-        self, service, mock_authz, mock_ds_repo, user_id
-    ):
+    async def test_delete_returns_false_when_not_found(self, service, authz, user_id):
         """delete() returns False when DS not found."""
-        mock_authz.check_permission.return_value = True
-        mock_ds_repo.get_by_id.return_value = None
+        await _grant_ds_manage(authz, "nonexistent", user_id)
+        # ds_repo is empty — DS not found.
 
         result = await service.delete(user_id=user_id, ds_id="nonexistent")
 
@@ -612,12 +667,12 @@ async def test_delete_returns_false_when_not_found(
 
     @pytest.mark.asyncio
     async def test_delete_returns_false_for_different_tenant(
-        self, service, mock_authz, mock_ds_repo, user_id
+        self, service, authz, ds_repo, user_id
     ):
         """delete() returns False when DS belongs to a different tenant."""
         ds = _make_ds(tenant_id="other-tenant")
-        mock_authz.check_permission.return_value = True
-        mock_ds_repo.get_by_id.return_value = ds
+        ds_repo.seed(ds)
+        await _grant_ds_manage(authz, ds.id.value, user_id)
 
         result = await service.delete(user_id=user_id, ds_id=ds.id.value)
 
@@ -625,34 +680,35 @@ async def test_delete_returns_false_for_different_tenant(
 
     @pytest.mark.asyncio
     async def test_delete_removes_credentials_if_path_exists(
-        self, service, mock_authz, mock_ds_repo, mock_secret_store, user_id, tenant_id
+        self, service, authz, ds_repo, secret_store, user_id, tenant_id
     ):
         """delete() deletes credentials from secret store if credentials_path is set."""
-        ds = _make_ds(credentials_path="datasource/ds-001/credentials")
-        mock_authz.check_permission.return_value = True
-        mock_ds_repo.get_by_id.return_value = ds
-        mock_ds_repo.delete.return_value = True
+        ds = _make_ds(
+            credentials_path="datasource/ds-001/credentials",
+            tenant_id=tenant_id,
+        )
+        ds_repo.seed(ds)
+        await _grant_ds_manage(authz, ds.id.value, user_id)
 
         await service.delete(user_id=user_id, ds_id=ds.id.value)
 
-        mock_secret_store.delete.assert_called_once_with(
-            path="datasource/ds-001/credentials",
-            tenant_id=tenant_id,
-        )
+        assert len(secret_store.delete_calls) == 1
+        assert secret_store.delete_calls[0]["path"] == "datasource/ds-001/credentials"
+        assert secret_store.delete_calls[0]["tenant_id"] == tenant_id
 
     @pytest.mark.asyncio
     async def test_delete_probes_success(
-        self, service, mock_authz, mock_ds_repo, mock_probe, user_id
+        self, service, authz, ds_repo, probe, user_id, tenant_id
     ):
-        """delete() calls probe on success."""
-        ds = _make_ds()
-        mock_authz.check_permission.return_value = True
-        mock_ds_repo.get_by_id.return_value = ds
-        mock_ds_repo.delete.return_value = True
+        """delete() emits probe event on success."""
+        ds = _make_ds(tenant_id=tenant_id)
+        ds_repo.seed(ds)
+        await _grant_ds_manage(authz, ds.id.value, user_id)
 
         await service.delete(user_id=user_id, ds_id=ds.id.value)
 
-        mock_probe.data_source_deleted.assert_called_once_with(ds_id=ds.id.value)
+        assert len(probe.data_source_deleted_calls) == 1
+        assert probe.data_source_deleted_calls[0]["ds_id"] == ds.id.value
 
 
 # ---- trigger_sync ----
@@ -662,68 +718,69 @@ class TestDataSourceServiceTriggerSync:
     """Tests for DataSourceService.trigger_sync."""
 
     @pytest.mark.asyncio
-    async def test_trigger_sync_checks_manage_permission(
-        self, service, mock_authz, mock_ds_repo, mock_sync_run_repo, user_id
+    async def test_trigger_sync_requires_manage_permission(
+        self, service, authz, ds_repo, user_id, tenant_id
     ):
-        """trigger_sync() checks MANAGE permission on the data source."""
-        ds = _make_ds()
-        mock_authz.check_permission.return_value = True
-        mock_ds_repo.get_by_id.return_value = ds
+        """trigger_sync() requires MANAGE permission on the data source."""
+        ds = _make_ds(tenant_id=tenant_id)
+        ds_repo.seed(ds)
 
-        await service.trigger_sync(user_id=user_id, ds_id=ds.id.value)
+        # No permission → raises UnauthorizedError.
+        with pytest.raises(UnauthorizedError):
+            await service.trigger_sync(user_id=user_id, ds_id=ds.id.value)
 
-        mock_authz.check_permission.assert_called_once_with(
-            resource=f"data_source:{ds.id.value}",
-            permission=Permission.MANAGE,
-            subject=f"user:{user_id}",
-        )
+        # MANAGE permission → succeeds.
+        await _grant_ds_manage(authz, ds.id.value, user_id)
+        result = await service.trigger_sync(user_id=user_id, ds_id=ds.id.value)
+        assert result is not None
 
     @pytest.mark.asyncio
-    async def test_trigger_sync_raises_unauthorized_when_denied(
-        self, service, mock_authz, user_id
-    ):
+    async def test_trigger_sync_raises_unauthorized_when_denied(self, service, user_id):
         """trigger_sync() raises UnauthorizedError when denied."""
-        mock_authz.check_permission.return_value = False
-
+        # No permission written → denied.
         with pytest.raises(UnauthorizedError):
             await service.trigger_sync(user_id=user_id, ds_id="ds-001")
 
     @pytest.mark.asyncio
     async def test_trigger_sync_raises_value_error_when_not_found(
-        self, service, mock_authz, mock_ds_repo, user_id
+        self, service, authz, user_id
     ):
         """trigger_sync() raises ValueError when DS not found."""
-        mock_authz.check_permission.return_value = True
-        mock_ds_repo.get_by_id.return_value = None
+        await _grant_ds_manage(authz, "nonexistent", user_id)
+        # ds_repo is empty — DS not found.
 
         with pytest.raises(ValueError):
             await service.trigger_sync(user_id=user_id, ds_id="nonexistent")
 
     @pytest.mark.asyncio
     async def test_trigger_sync_rejects_different_tenant(
-        self, service, mock_authz, mock_ds_repo, user_id
+        self, service, authz, ds_repo, user_id
     ):
         """trigger_sync() raises ValueError when DS belongs to a different tenant."""
         ds = _make_ds(tenant_id="other-tenant")
-        mock_authz.check_permission.return_value = True
-        mock_ds_repo.get_by_id.return_value = ds
+        ds_repo.seed(ds)
+        await _grant_ds_manage(authz, ds.id.value, user_id)
 
         with pytest.raises(ValueError):
             await service.trigger_sync(user_id=user_id, ds_id=ds.id.value)
 
     @pytest.mark.asyncio
     async def test_trigger_sync_creates_sync_run_and_saves_ds(
-        self, service, mock_authz, mock_ds_repo, mock_sync_run_repo, mock_probe, user_id
+        self, service, authz, ds_repo, sync_run_repo, probe, user_id, tenant_id
     ):
-        """trigger_sync() creates a sync run and saves the data source."""
-        ds = _make_ds()
-        mock_authz.check_permission.return_value = True
-        mock_ds_repo.get_by_id.return_value = ds
+        """trigger_sync() creates a sync run, saves it, and saves the data source."""
+        ds = _make_ds(tenant_id=tenant_id)
+        ds_repo.seed(ds)
+        await _grant_ds_manage(authz, ds.id.value, user_id)
 
         result = await service.trigger_sync(user_id=user_id, ds_id=ds.id.value)
 
         assert result.data_source_id == ds.id.value
         assert result.status == "pending"
-        mock_sync_run_repo.save.assert_called_once()
-        mock_ds_repo.save.assert_called_once()
-        mock_probe.sync_requested.assert_called_once_with(ds_id=ds.id.value)
+        # Sync run must be persisted.
+        assert len(sync_run_repo.saved) == 1
+        # Data source must be saved (with SyncStarted event).
+        assert len(ds_repo.saved) >= 1
+        # Probe emitted.
+        assert len(probe.sync_requested_calls) == 1
+        assert probe.sync_requested_calls[0]["ds_id"] == ds.id.value
diff --git a/src/dev-ui/app/tests/sync-logs.test.ts b/src/dev-ui/app/tests/sync-logs.test.ts
new file mode 100644
index 000000000..7c11c2917
--- /dev/null
+++ b/src/dev-ui/app/tests/sync-logs.test.ts
@@ -0,0 +1,356 @@
+import { describe, it, expect, vi } from 'vitest'
+
+// ── Sync Log Viewer (task-044) ────────────────────────────────────────────────
+//
+// Spec: ui/experience.spec.md — Requirement: Sync Monitoring
+//
+// Scenario: Sync logs
+//   GIVEN a sync run (in progress or completed)
+//   WHEN the user requests logs
+//   THEN detailed logs for that run are displayed
+//
+// These tests cover the viewLogs / fetchRunLogs / closeLogs state machine
+// in pages/data-sources/index.vue, focusing on aspects not covered by the
+// generic sync-monitoring tests: data-source ID capture, loading-state
+// lifecycle, empty-state handling, and log display format.
+
+// ── Types (mirrors data-sources/index.vue) ────────────────────────────────────
+
+interface SyncRun {
+  id: string
+  status: 'pending' | 'ingesting' | 'ai_extracting' | 'applying' | 'completed' | 'failed'
+  started_at: string
+  completed_at: string | null
+  error: string | null
+  created_at: string
+}
+
+interface DataSourceItem {
+  id: string
+  name: string
+  adapter_type: string
+  knowledge_graph_id: string
+  last_sync_at: string | null
+  created_at: string
+  sync_runs?: SyncRun[]
+}
+
+// ── State machine (mirrors the component) ─────────────────────────────────────
+
+function makeLogState() {
+  const logSheetOpen = { value: false }
+  const selectedLogRunId = { value: null as string | null }
+  const selectedLogDsId = { value: null as string | null }
+  const runLogs = { value: [] as string[] }
+  const logsLoading = { value: false }
+  const logsError = { value: null as string | null }
+
+  async function fetchRunLogs(
+    dsId: string,
+    runId: string,
+    apiFetch: (url: string) => Promise<{ logs: string[] }>,
+  ) {
+    logsLoading.value = true
+    logsError.value = null
+    try {
+      const result = await apiFetch(
+        `/management/data-sources/${dsId}/sync-runs/${runId}/logs`,
+      )
+      runLogs.value = result.logs ?? []
+    } catch (err) {
+      logsError.value = err instanceof Error ? err.message : 'Failed to load logs'
+      runLogs.value = []
+    } finally {
+      logsLoading.value = false
+    }
+  }
+
+  async function viewLogs(
+    ds: DataSourceItem,
+    run: SyncRun,
+    apiFetch: (url: string) => Promise<{ logs: string[] }>,
+  ) {
+    selectedLogDsId.value = ds.id
+    selectedLogRunId.value = run.id
+    runLogs.value = []
+    logsError.value = null
+    logSheetOpen.value = true
+    await fetchRunLogs(ds.id, run.id, apiFetch)
+  }
+
+  function closeLogs() {
+    logSheetOpen.value = false
+    selectedLogRunId.value = null
+    selectedLogDsId.value = null
+    runLogs.value = []
+    logsError.value = null
+  }
+
+  return {
+    logSheetOpen,
+    selectedLogRunId,
+    selectedLogDsId,
+    runLogs,
+    logsLoading,
+    logsError,
+    viewLogs,
+    fetchRunLogs,
+    closeLogs,
+  }
+}
+
+// ── Scenario: Sync logs — View Logs captures both dsId and runId ──────────────
+
+describe('Sync Logs - viewLogs captures both dsId and runId', () => {
+  it('sets selectedLogDsId when View Logs is clicked', async () => {
+    const state = makeLogState()
+    const ds: DataSourceItem = { id: 'ds-abc', name: 'my-repo', adapter_type: 'github', knowledge_graph_id: 'kg-1', last_sync_at: null, created_at: '2024-01-01T00:00:00Z' }
+    const run: SyncRun = { id: 'run-1', status: 'completed', started_at: '2024-01-01T10:00:00Z', completed_at: '2024-01-01T10:01:00Z', error: null, created_at: '2024-01-01T10:00:00Z' }
+    const apiFetch = vi.fn().mockResolvedValue({ logs: [] })
+
+    await state.viewLogs(ds, run, apiFetch)
+
+    expect(state.selectedLogDsId.value).toBe('ds-abc')
+    expect(state.selectedLogRunId.value).toBe('run-1')
+  })
+
+  it('opens the log sheet immediately (before fetch completes)', async () => {
+    const state = makeLogState()
+    const ds: DataSourceItem = { id: 'ds-1', name: 'repo', adapter_type: 'github', knowledge_graph_id: 'kg-1', last_sync_at: null, created_at: '2024-01-01T00:00:00Z' }
+    const run: SyncRun = { id: 'run-2', status: 'ingesting', started_at: '2024-01-01T10:00:00Z', completed_at: null, error: null, created_at: '2024-01-01T10:00:00Z' }
+    let sheetOpenDuringFetch = false
+
+    const apiFetch = vi.fn().mockImplementation(async () => {
+      // capture sheet state while "fetch is in progress"
+      sheetOpenDuringFetch = state.logSheetOpen.value
+      return { logs: ['log line'] }
+    })
+
+    await state.viewLogs(ds, run, apiFetch)
+
+    expect(sheetOpenDuringFetch).toBe(true)
+  })
+
+  it('clears previous logs when a new run is selected', async () => {
+    const state = makeLogState()
+    // Seed stale log data from a previous selection
+    state.runLogs.value = ['stale log from previous run']
+
+    const ds: DataSourceItem = { id: 'ds-1', name: 'repo', adapter_type: 'github', knowledge_graph_id: 'kg-1', last_sync_at: null, created_at: '2024-01-01T00:00:00Z' }
+    const run: SyncRun = { id: 'run-new', status: 'completed', started_at: '2024-01-02T10:00:00Z', completed_at: '2024-01-02T10:01:00Z', error: null, created_at: '2024-01-02T10:00:00Z' }
+    const apiFetch = vi.fn().mockResolvedValue({ logs: ['fresh log line'] })
+
+    await state.viewLogs(ds, run, apiFetch)
+
+    expect(state.runLogs.value).toEqual(['fresh log line'])
+    expect(state.runLogs.value).not.toContain('stale log from previous run')
+  })
+})
+
+// ── Scenario: Sync logs — Loading state lifecycle ─────────────────────────────
+
+describe('Sync Logs - loading state lifecycle', () => {
+  it('sets logsLoading to true while fetch is in progress', async () => {
+    const state = makeLogState()
+    const loadingValues: boolean[] = []
+
+    const apiFetch = vi.fn().mockImplementation(async () => {
+      loadingValues.push(state.logsLoading.value)
+      return { logs: ['line 1'] }
+    })
+
+    await state.fetchRunLogs('ds-1', 'run-1', apiFetch)
+
+    // logsLoading was true during the fetch
+    expect(loadingValues).toContain(true)
+  })
+
+  it('resets logsLoading to false after a successful fetch', async () => {
+    const state = makeLogState()
+    const apiFetch = vi.fn().mockResolvedValue({ logs: ['log line'] })
+
+    await state.fetchRunLogs('ds-1', 'run-1', apiFetch)
+
+    expect(state.logsLoading.value).toBe(false)
+  })
+
+  it('resets logsLoading to false after a failed fetch', async () => {
+    const state = makeLogState()
+    const apiFetch = vi.fn().mockRejectedValue(new Error('Network failure'))
+
+    await state.fetchRunLogs('ds-1', 'run-1', apiFetch)
+
+    expect(state.logsLoading.value).toBe(false)
+  })
+})
+
+// ── Scenario: Sync logs — Log fetch uses correct endpoint ─────────────────────
+
+describe('Sync Logs - API endpoint construction', () => {
+  it('includes both dsId and runId in the fetch URL', async () => {
+    const state = makeLogState()
+    const apiFetch = vi.fn().mockResolvedValue({ logs: [] })
+
+    await state.fetchRunLogs('ds-xyz-123', 'run-abc-456', apiFetch)
+
+    expect(apiFetch).toHaveBeenCalledWith(
+      '/management/data-sources/ds-xyz-123/sync-runs/run-abc-456/logs',
+    )
+  })
+
+  it('calls the logs endpoint for an in-progress sync run', async () => {
+    const state = makeLogState()
+    const apiFetch = vi.fn().mockResolvedValue({ logs: ['ingesting batch 1/10'] })
+
+    await state.fetchRunLogs('ds-1', 'run-inprogress', apiFetch)
+
+    expect(apiFetch).toHaveBeenCalledWith(
+      '/management/data-sources/ds-1/sync-runs/run-inprogress/logs',
+    )
+    expect(state.runLogs.value).toContain('ingesting batch 1/10')
+  })
+})
+
+// ── Scenario: Sync logs — Empty state ────────────────────────────────────────
+
+describe('Sync Logs - empty state handling', () => {
+  it('runLogs is empty when API returns an empty logs array', async () => {
+    const state = makeLogState()
+    const apiFetch = vi.fn().mockResolvedValue({ logs: [] })
+
+    await state.fetchRunLogs('ds-1', 'run-1', apiFetch)
+
+    expect(state.runLogs.value).toHaveLength(0)
+  })
+
+  it('runLogs defaults to empty when API omits the logs key', async () => {
+    const state = makeLogState()
+    // Defensive: API may return {} without a logs key
+    const apiFetch = vi.fn().mockResolvedValue({} as { logs: string[] })
+
+    await state.fetchRunLogs('ds-1', 'run-1', apiFetch)
+
+    expect(state.runLogs.value).toHaveLength(0)
+  })
+})
+
+// ── Scenario: Sync logs — Error state ────────────────────────────────────────
+
+describe('Sync Logs - error state', () => {
+  it('captures the error message when fetch fails', async () => {
+    const state = makeLogState()
+    const apiFetch = vi.fn().mockRejectedValue(new Error('Service Unavailable'))
+
+    await state.fetchRunLogs('ds-1', 'run-1', apiFetch)
+
+    expect(state.logsError.value).toBe('Service Unavailable')
+    expect(state.runLogs.value).toHaveLength(0)
+  })
+
+  it('clears a previous error when a new fetch begins', async () => {
+    const state = makeLogState()
+    // Set up a previous error
+    state.logsError.value = 'Previous error'
+
+    const apiFetch = vi.fn().mockResolvedValue({ logs: ['success'] })
+    await state.fetchRunLogs('ds-1', 'run-2', apiFetch)
+
+    expect(state.logsError.value).toBeNull()
+  })
+})
+
+// ── Scenario: Sync logs — Log display format ──────────────────────────────────
+
+describe('Sync Logs - log line display format', () => {
+  it('log lines are stored as strings (one line per entry)', async () => {
+    const state = makeLogState()
+    const logLines = [
+      '2024-01-01T10:00:01Z INFO Starting sync',
+      '2024-01-01T10:00:05Z INFO Fetched 100 items',
+      '2024-01-01T10:00:10Z INFO Extraction complete',
+    ]
+    const apiFetch = vi.fn().mockResolvedValue({ logs: logLines })
+
+    await state.fetchRunLogs('ds-1', 'run-1', apiFetch)
+
+    expect(state.runLogs.value).toHaveLength(3)
+    state.runLogs.value.forEach((line) => expect(typeof line).toBe('string'))
+  })
+
+  it('log lines joined by newlines produce valid monospace output', async () => {
+    const state = makeLogState()
+    const lines = ['INFO Starting', 'INFO Done']
+    const apiFetch = vi.fn().mockResolvedValue({ logs: lines })
+
+    await state.fetchRunLogs('ds-1', 'run-1', apiFetch)
+
+    const rendered = state.runLogs.value.join('\n')
+    expect(rendered).toBe('INFO Starting\nINFO Done')
+  })
+})
+
+// ── Scenario: Sync logs — closeLogs clears all log state ─────────────────────
+
+describe('Sync Logs - closeLogs resets full state', () => {
+  it('closeLogs sets logSheetOpen to false', () => {
+    const state = makeLogState()
+    state.logSheetOpen.value = true
+
+    state.closeLogs()
+
+    expect(state.logSheetOpen.value).toBe(false)
+  })
+
+  it('closeLogs clears selectedLogRunId', () => {
+    const state = makeLogState()
+    state.selectedLogRunId.value = 'run-123'
+
+    state.closeLogs()
+
+    expect(state.selectedLogRunId.value).toBeNull()
+  })
+
+  it('closeLogs clears selectedLogDsId', () => {
+    const state = makeLogState()
+    state.selectedLogDsId.value = 'ds-abc'
+
+    state.closeLogs()
+
+    expect(state.selectedLogDsId.value).toBeNull()
+  })
+
+  it('closeLogs clears runLogs', () => {
+    const state = makeLogState()
+    state.runLogs.value = ['log line 1', 'log line 2']
+
+    state.closeLogs()
+
+    expect(state.runLogs.value).toHaveLength(0)
+  })
+
+  it('closeLogs clears logsError', () => {
+    const state = makeLogState()
+    state.logsError.value = 'Some error'
+
+    state.closeLogs()
+
+    expect(state.logsError.value).toBeNull()
+  })
+
+  it('closeLogs resets all state to initial values in one call', () => {
+    const state = makeLogState()
+    state.logSheetOpen.value = true
+    state.selectedLogRunId.value = 'run-999'
+    state.selectedLogDsId.value = 'ds-999'
+    state.runLogs.value = ['line']
+    state.logsError.value = 'err'
+
+    state.closeLogs()
+
+    expect(state.logSheetOpen.value).toBe(false)
+    expect(state.selectedLogRunId.value).toBeNull()
+    expect(state.selectedLogDsId.value).toBeNull()
+    expect(state.runLogs.value).toHaveLength(0)
+    expect(state.logsError.value).toBeNull()
+  })
+})