feat(audit): P1 cross-tenant/RBAC tests, P3 test fixes, P2/P3 frontend fixes

- P1-Tests: 12 test files with new cross-tenant isolation + RBAC tests
- P3-Tests: 8 fixes (duplicate fixtures, sys.path.insert, unused imports, KeyError)
- P3-Frontend: LucideIcons → ICON_MAP (2 files), inline styles → Tailwind (2 files)
- P3-Frontend: DOMPurify for iframe XSS, redundant regex removed, console.log → console.debug
- P2-Frontend: 2 notification API TODOs retained (requires larger refactor)
- conftest.py: create_no_perm_user helper added
- pyproject.toml: pythonpath for scripts/ added
- All checks green: ruff 0, F821 0, tsc 0, app 495 routes, cross-plugin 0

tests/conftest.py

@@ -434,6 +434,49 @@ async def seed_tenant_and_users(db: AsyncSession) -> dict[str, Any]:
     }
 
 
+async def create_no_perm_user(db: AsyncSession, seed: dict[str, Any]) -> User:
+    """Create a user in tenant A with no permissions (for RBAC tests).
+
+    Returns the created user. The user has a role with empty permissions,
+    so any require_permission check will return 403.
+    """
+    from app.core.auth import hash_password
+    from app.models.role import Role
+    from app.models.user import User, UserTenant
+
+    no_perm_role = Role(
+        tenant_id=seed["tenant_a"].id,
+        name="no_perm",
+        permissions={},
+        denied_permissions=[],
+        field_permissions={},
+    )
+    db.add(no_perm_role)
+    await db.flush()
+
+    no_perm_user = User(
+        email="noperm@tenanta.com",
+        name="No Perm",
+        password_hash=hash_password("TestPass123!"),
+        is_active=True,
+        preferences={},
+    )
+    db.add(no_perm_user)
+    await db.flush()
+
+    ut = UserTenant(
+        user_id=no_perm_user.id,
+        tenant_id=seed["tenant_a"].id,
+        is_default=True,
+        role="no_perm",
+        role_id=no_perm_role.id,
+    )
+    db.add(ut)
+    await db.flush()
+    await db.commit()
+    return no_perm_user
+
+
 async def login_client(
     client: AsyncClient, email: str, password: str = "TestPass123!"
 ) -> dict[str, str]:

tests/test_agent_subtasks.py

@@ -26,11 +26,6 @@ def db_setup():
     yield
 
 
-@pytest.fixture(autouse=True)
-def clean_tables(db_setup):
-    """No-op override of conftest clean_tables."""
-    yield
-
 @pytest.fixture(autouse=True)
 def clean_tables(db_setup):
     """No-op override of conftest clean_tables."""

tests/test_ai_deploy.py

@@ -4,14 +4,10 @@ from __future__ import annotations
 
 import json
 import os
-import sys
 from unittest.mock import patch
 
 import pytest
 
-# Add scripts dir to path
-sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "scripts"))
-
 
 def test_phase_result_to_dict():
     """PhaseResult.to_dict should serialize correctly."""

tests/test_ai_health_check.py

@@ -3,15 +3,10 @@
 from __future__ import annotations
 
 import json
-import os
-import sys
 from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
 
-# Add scripts dir to path
-sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "scripts"))
-
 
 def test_enumerate_routes_returns_api_routes():
     """enumerate_routes should return only /api/ routes with methods."""

tests/test_api_documentation.py

@@ -3,12 +3,9 @@
 from __future__ import annotations
 
 import os
-import sys
 
 import pytest
 
-sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
-
 
 def test_openapi_tags_configured():
     """FastAPI app should have openapi_tags configured with descriptions."""

tests/test_backup_restore.py

@@ -4,16 +4,12 @@ from __future__ import annotations
 
 import json
 import os
-import sys
 import tempfile
 from pathlib import Path
 from unittest.mock import patch
 
 import pytest
 
-# Add scripts dir to path
-sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "scripts"))
-
 
 def test_get_db_connection_params_from_asyncpg_url():
     """get_db_connection_params should convert asyncpg URL to standard postgresql URL."""

tests/test_calendar.py

@@ -1078,3 +1078,38 @@ async def test_ac30_private_entry_visibility(calendar_app, db_session):
         # Direct access to private entry → 403
         resp = await viewer_c.get(f"/api/v1/calendar/entries/{entry_id}", headers=ORIGIN_HEADER)
         assert resp.status_code == 403
+
+
+# ─── Cross-tenant isolation test ───
+
+
+@pytest.mark.asyncio
+async def test_cross_tenant_calendar_isolation(calendar_app, db_session):
+    """Calendar created in tenant A is not accessible from tenant B."""
+    from httpx import ASGITransport, AsyncClient
+
+    from tests.conftest import seed_tenant_and_users
+
+    await seed_tenant_and_users(db_session)
+    transport = ASGITransport(app=calendar_app)
+
+    async with AsyncClient(transport=transport, base_url="http://test") as client_a:
+        await login_client(client_a, "admin@tenanta.com")
+        resp = await client_a.post(
+            "/api/v1/calendars",
+            json={"name": "Tenant A Calendar"},
+            headers=ORIGIN_HEADER,
+        )
+        assert resp.status_code == 201
+        cal_id = resp.json()["id"]
+
+        # Tenant B admin must not see tenant A's calendar
+        async with AsyncClient(transport=transport, base_url="http://test") as client_b:
+            await login_client(client_b, "admin@tenantb.com")
+            list_resp = await client_b.get("/api/v1/calendars", headers=ORIGIN_HEADER)
+            assert list_resp.status_code == 200
+            assert all(c["id"] != cal_id for c in list_resp.json())
+
+            # Direct access to tenant A's calendar → 404
+            get_resp = await client_b.get(f"/api/v1/calendars/{cal_id}", headers=ORIGIN_HEADER)
+            assert get_resp.status_code == 404

tests/test_companies.py

@@ -351,3 +351,45 @@ class TestCompanyAuditAndSoftDelete:
         names = [item["name"] for item in list_resp.json()["items"]]
         assert "SoftDelete Corp" not in names
         assert "Company Alpha" in names  # Seed company still present
+
+
+# ── Visibility filter test ──
+
+
+@pytest.mark.asyncio
+class TestCompanyVisibilityFilter:
+    """Row-level visibility filter hides non-owned, non-shared companies."""
+
+    async def test_visibility_filter_hides_owned_company_from_viewer(self, client: AsyncClient, db_session):
+        """A company owned by admin_a is not visible to viewer_a in the contacts list."""
+        await seed_tenant_and_users(db_session)
+        await login_client(client, "admin@tenanta.com")
+        # Create a company via /api/v1/contacts so owner_id is set to admin_a
+        create_resp = await client.post(
+            "/api/v1/contacts",
+            json={"type": "company", "name": "Owned Corp", "displayname": "Owned Corp"},
+            headers=ORIGIN_HEADER,
+        )
+        assert create_resp.status_code == 201
+        owned_id = create_resp.json()["id"]
+
+        # Admin (owner) sees it in the contacts list with type=company
+        list_resp = await client.get("/api/v1/contacts?type=company", headers=ORIGIN_HEADER)
+        assert list_resp.status_code == 200
+        assert any(c["id"] == owned_id for c in list_resp.json()["items"])
+
+        # Viewer (non-owner, not shared) must NOT see it
+        await login_client(client, "viewer@tenanta.com")
+        viewer_list = await client.get("/api/v1/contacts?type=company", headers=ORIGIN_HEADER)
+        assert viewer_list.status_code == 200
+        assert all(c["id"] != owned_id for c in viewer_list.json()["items"])
+
+    async def test_visibility_filter_shows_tenant_owned_company(self, client: AsyncClient, db_session):
+        """A tenant-owned company (owner_id NULL) is visible to all users with read permission."""
+        await seed_tenant_and_users(db_session)
+        await login_client(client, "viewer@tenanta.com")
+        # Seed company 'Company Alpha' has owner_id NULL → tenant-owned → visible to viewer
+        list_resp = await client.get("/api/v1/contacts?type=company", headers=ORIGIN_HEADER)
+        assert list_resp.status_code == 200
+        names = [c["name"] for c in list_resp.json()["items"]]
+        assert "Company Alpha" in names

tests/test_contacts.py

@@ -171,3 +171,45 @@ class TestContactDelete:
         al_entries = al_result.scalars().all()
         assert len(al_entries) >= 1
         assert any(e.action == "hard_delete" for e in al_entries)
+
+
+# ── Visibility filter test ──
+
+
+@pytest.mark.asyncio
+class TestContactVisibilityFilter:
+    """Row-level visibility filter hides non-owned, non-shared contacts."""
+
+    async def test_visibility_filter_hides_owned_contact_from_viewer(self, client: AsyncClient, db_session):
+        """A contact owned by admin_a is not visible to viewer_a in the list."""
+        await seed_tenant_and_users(db_session)
+        await login_client(client, "admin@tenanta.com")
+        # Create a contact as admin (owner_id = admin_a)
+        create_resp = await client.post(
+            "/api/v1/contacts",
+            json={"firstname": "Owned", "surname": "Contact"},
+            headers=ORIGIN_HEADER,
+        )
+        assert create_resp.status_code == 201
+        owned_id = create_resp.json()["id"]
+
+        # Admin (owner) sees it in the list
+        list_resp = await client.get("/api/v1/contacts", headers=ORIGIN_HEADER)
+        assert list_resp.status_code == 200
+        assert any(c["id"] == owned_id for c in list_resp.json()["items"])
+
+        # Viewer (non-owner, not shared) must NOT see it
+        await login_client(client, "viewer@tenanta.com")
+        viewer_list = await client.get("/api/v1/contacts", headers=ORIGIN_HEADER)
+        assert viewer_list.status_code == 200
+        assert all(c["id"] != owned_id for c in viewer_list.json()["items"])
+
+    async def test_visibility_filter_shows_tenant_owned_contact(self, client: AsyncClient, db_session):
+        """A tenant-owned contact (owner_id NULL) is visible to all users with read permission."""
+        await seed_tenant_and_users(db_session)
+        await login_client(client, "viewer@tenanta.com")
+        # Seed company 'Company Alpha' has owner_id NULL → tenant-owned → visible to viewer
+        list_resp = await client.get("/api/v1/contacts?type=company", headers=ORIGIN_HEADER)
+        assert list_resp.status_code == 200
+        names = [c["name"] for c in list_resp.json()["items"]]
+        assert "Company Alpha" in names

tests/test_cross_tenant_security.py

@@ -190,7 +190,6 @@ async def contact_b(db_session: AsyncSession, tenant_b: Tenant, user_b: User):
 
 # ── Cross-Tenant RLS Tests ────────────────────────────────────────────────────
 
-@pytest.mark.asyncio
 @pytest.mark.asyncio
 async def test_rls_blocks_cross_tenant_insert(
     db_session: AsyncSession,

tests/test_custom_fields.py

@@ -162,3 +162,50 @@ class TestCustomFieldDefinition:
         )
         assert len(manifest.custom_fields) == 1
         assert manifest.custom_fields[0].name == "score"
+
+
+# ── Cross-tenant isolation test ──
+@pytest.mark.asyncio
+class TestCustomFieldsCrossTenant:
+    """Custom fields must not leak across tenants."""
+
+    async def test_cross_tenant_isolation(self, client: AsyncClient, db_session):
+        """Tenant B cannot access custom fields of tenant A's contact."""
+        from httpx import ASGITransport
+        from httpx import AsyncClient as AC
+
+        import app.main
+
+        await seed_tenant_and_users(db_session)
+        await login_client(client, "admin@tenanta.com")
+        # Create a contact in tenant A
+        create_resp = await client.post(
+            "/api/v1/contacts",
+            json={"type": "company", "name": "Tenant A Custom Corp", "displayname": "Tenant A Custom Corp"},
+            headers=ORIGIN_HEADER,
+        )
+        assert create_resp.status_code == 201
+        contact_id = create_resp.json()["id"]
+
+        # Tenant B admin must not access tenant A's contact custom fields
+        app_instance = app.main.app
+        async with AC(transport=ASGITransport(app=app_instance), base_url="http://test") as client_b:
+            await login_client(client_b, "admin@tenantb.com")
+            resp = await client_b.get(
+                f"/api/v1/contacts/{contact_id}/custom-fields",
+                headers=ORIGIN_HEADER,
+            )
+            assert resp.status_code == 404
+
+    async def test_rbac_no_permission(self, client: AsyncClient, db_session):
+        """User without contacts:read permission gets 403 on GET custom fields."""
+        from tests.conftest import create_no_perm_user
+
+        seed = await seed_tenant_and_users(db_session)
+        await create_no_perm_user(db_session, seed)
+        await login_client(client, "noperm@tenanta.com")
+        resp = await client.get(
+            "/api/v1/contacts/00000000-0000-0000-0000-000000000000/custom-fields",
+            headers=ORIGIN_HEADER,
+        )
+        assert resp.status_code == 403

tests/test_dashboard.py

@@ -44,3 +44,50 @@ class TestDashboardWidgets:
             assert "id" in widget
             assert "component" in widget
             assert "label_key" in widget
+
+
+# ── Cross-tenant isolation test ──
+@pytest.mark.asyncio
+class TestDashboardCrossTenant:
+    """Dashboard counts must not leak across tenants."""
+
+    async def test_cross_tenant_isolation(self, client: AsyncClient, db_session):
+        """Tenant B admin does not see tenant A's contacts in dashboard counts."""
+        from httpx import ASGITransport
+        from httpx import AsyncClient as AC
+
+        import app.main
+
+        await seed_tenant_and_users(db_session)
+        await login_client(client, "admin@tenanta.com")
+        # Create a company in tenant A
+        create_resp = await client.post(
+            "/api/v1/companies",
+            json={"name": "Tenant A Dashboard Corp"},
+            headers=ORIGIN_HEADER,
+        )
+        assert create_resp.status_code == 201
+
+        # Tenant A admin sees it
+        counts_a = await client.get("/api/v1/dashboard/counts", headers=ORIGIN_HEADER)
+        assert counts_a.status_code == 200
+        assert counts_a.json()["companies"] >= 1
+
+        # Tenant B admin must not see it
+        app_instance = app.main.app
+        async with AC(transport=ASGITransport(app=app_instance), base_url="http://test") as client_b:
+            await login_client(client_b, "admin@tenantb.com")
+            counts_b = await client_b.get("/api/v1/dashboard/counts", headers=ORIGIN_HEADER)
+            assert counts_b.status_code == 200
+            # Tenant B only has its own seeded company (Company Beta)
+            assert counts_b.json()["companies"] == 1
+
+    async def test_rbac_no_permission(self, client: AsyncClient, db_session):
+        """User without dashboard:read permission gets 403."""
+        from tests.conftest import create_no_perm_user
+
+        seed = await seed_tenant_and_users(db_session)
+        await create_no_perm_user(db_session, seed)
+        await login_client(client, "noperm@tenanta.com")
+        resp = await client.get("/api/v1/dashboard/widgets", headers=ORIGIN_HEADER)
+        assert resp.status_code == 403

tests/test_dms.py

@@ -736,3 +736,38 @@ async def test_ac19_bulk_delete(authed_client):
     for fid in file_ids:
         resp = await client.get(f"/api/v1/dms/files/{fid}", headers=ORIGIN_HEADER)
         assert resp.status_code == 404
+
+
+# ─── Cross-tenant isolation test ───
+
+
+@pytest.mark.asyncio
+async def test_cross_tenant_dms_isolation(dms_app, db_session):
+    """File uploaded in tenant A is not visible to tenant B."""
+    from httpx import ASGITransport, AsyncClient
+
+    from tests.conftest import seed_tenant_and_users
+
+    await seed_tenant_and_users(db_session)
+    transport = ASGITransport(app=dms_app)
+
+    async with AsyncClient(transport=transport, base_url="http://test") as client_a:
+        await login_client(client_a, "admin@tenanta.com")
+        upload_resp = await client_a.post(
+            "/api/v1/dms/files/upload",
+            files={"file": ("tenant_a.pdf", PDF_CONTENT, "application/pdf")},
+            headers=ORIGIN_HEADER,
+        )
+        assert upload_resp.status_code == 201
+        file_id = upload_resp.json()["id"]
+
+        # Tenant B admin must not see tenant A's file in the list
+        async with AsyncClient(transport=transport, base_url="http://test") as client_b:
+            await login_client(client_b, "admin@tenantb.com")
+            list_resp = await client_b.get("/api/v1/dms/files", headers=ORIGIN_HEADER)
+            assert list_resp.status_code == 200
+            assert all(f["id"] != file_id for f in list_resp.json())
+
+            # Direct access to tenant A's file → 404
+            get_resp = await client_b.get(f"/api/v1/dms/files/{file_id}", headers=ORIGIN_HEADER)
+            assert get_resp.status_code == 404

tests/test_dms_coverage.py

@@ -8,7 +8,6 @@ shared-with-me with data, bulk mixed IDs, tenant isolation.
 
 from __future__ import annotations
 
-import os
 import uuid
 from unittest.mock import patch
 

tests/test_entity_links.py

@@ -485,3 +485,45 @@ async def test_list_contact_files_empty(authed_client: AsyncClient):
     resp = await client.get(f"/api/v1/contacts/{contact_id}/files", headers=ORIGIN_HEADER)
     assert resp.status_code == 200
     assert resp.json() == []
+
+
+# ─── Cross-tenant isolation test ───
+
+
+@pytest.mark.asyncio
+async def test_cross_tenant_entity_link_isolation(plugin_app, db_session):
+    """Entity link created in tenant A is not visible to tenant B."""
+    from tests.conftest import seed_tenant_and_users
+
+    seed = await seed_tenant_and_users(db_session)
+    transport = ASGITransport(app=plugin_app)
+
+    async with AsyncClient(transport=transport, base_url="http://test") as client_a:
+        await login_client(client_a, "admin@tenanta.com")
+        # Upload a file in tenant A
+        upload_resp = await client_a.post(
+            "/api/v1/dms/files/upload",
+            files={"file": ("tenant_a_link.txt", b"tenant a link", "text/plain")},
+            headers=ORIGIN_HEADER,
+        )
+        assert upload_resp.status_code == 201
+        file_id = upload_resp.json()["id"]
+        company_id = str(seed["company_a"].id)
+
+        # Link file to company_a in tenant A
+        link_resp = await client_a.post(
+            f"/api/v1/entity-links/files/{file_id}/link",
+            json={"entity_type": "company", "entity_id": company_id},
+            headers=ORIGIN_HEADER,
+        )
+        assert link_resp.status_code == 200
+
+        # Tenant B admin must not see tenant A's link
+        async with AsyncClient(transport=transport, base_url="http://test") as client_b:
+            await login_client(client_b, "admin@tenantb.com")
+            links_resp = await client_b.get(
+                f"/api/v1/entity-links/files/{file_id}/links",
+                headers=ORIGIN_HEADER,
+            )
+            assert links_resp.status_code == 200
+            assert links_resp.json() == []

tests/test_notifications.py

@@ -116,3 +116,52 @@ class TestNotifications:
         assert "count" in data
         assert isinstance(data["count"], int)
         assert data["count"] >= 2
+
+
+# ── Cross-tenant isolation + RBAC tests ──
+
+
+@pytest.mark.asyncio
+class TestNotificationCrossTenant:
+    """Notifications must not leak across tenants."""
+
+    async def test_cross_tenant_isolation(self, client: AsyncClient, db_session):
+        """Notification for tenant A user is not visible to tenant B admin."""
+        from httpx import ASGITransport
+        from httpx import AsyncClient as AC
+
+        import app.main
+
+        seed = await seed_tenant_and_users(db_session)
+        await create_notification(
+            db_session,
+            seed["tenant_a"].id,
+            seed["admin_a"].id,
+            "info",
+            "Tenant A Notif",
+            "Body",
+        )
+        await db_session.commit()
+
+        await login_client(client, "admin@tenanta.com")
+        resp = await client.get("/api/v1/notifications")
+        assert resp.status_code == 200
+        assert any(i["title"] == "Tenant A Notif" for i in resp.json()["items"])
+
+        # Tenant B admin must not see tenant A's notification
+        app_instance = app.main.app
+        async with AC(transport=ASGITransport(app=app_instance), base_url="http://test") as client_b:
+            await login_client(client_b, "admin@tenantb.com")
+            resp_b = await client_b.get("/api/v1/notifications")
+            assert resp_b.status_code == 200
+            assert all(i["title"] != "Tenant A Notif" for i in resp_b.json()["items"])
+
+    async def test_rbac_no_permission(self, client: AsyncClient, db_session):
+        """User without notifications:read permission gets 403 on list."""
+        from tests.conftest import create_no_perm_user
+
+        seed = await seed_tenant_and_users(db_session)
+        await create_no_perm_user(db_session, seed)
+        await login_client(client, "noperm@tenanta.com")
+        resp = await client.get("/api/v1/notifications")
+        assert resp.status_code == 403

tests/test_saved_filters.py

@@ -126,3 +126,55 @@ class TestSavedFilterDelete:
         await login_client(client, "admin@tenanta.com")
         resp = await client.delete("/api/v1/saved-filters/not-a-uuid", headers=ORIGIN_HEADER)
         assert resp.status_code == 400
+
+
+# ── Cross-tenant isolation test ──
+@pytest.mark.asyncio
+class TestSavedFilterCrossTenant:
+    """Saved filters must not leak across tenants."""
+
+    async def test_cross_tenant_isolation(self, client: AsyncClient, db_session):
+        """Filter created in tenant A is not visible to tenant B."""
+        from httpx import ASGITransport
+        from httpx import AsyncClient as AC
+
+        import app.main
+
+        await seed_tenant_and_users(db_session)
+        await login_client(client, "admin@tenanta.com")
+        # Create a filter in tenant A
+        create_resp = await client.post(
+            "/api/v1/saved-filters",
+            json={"name": "Tenant A Filter", "entity_type": "contacts", "filter_criteria": {}},
+            headers=ORIGIN_HEADER,
+        )
+        assert create_resp.status_code == 201
+        filter_id = create_resp.json()["id"]
+
+        # Login as tenant B admin
+        app_instance = app.main.app
+        async with AC(transport=ASGITransport(app=app_instance), base_url="http://test") as client_b:
+            await login_client(client_b, "admin@tenantb.com")
+            # List must not contain tenant A's filter
+            list_resp = await client_b.get("/api/v1/saved-filters", headers=ORIGIN_HEADER)
+            assert list_resp.status_code == 200
+            assert all(f["id"] != filter_id for f in list_resp.json())
+            # Delete must 404 (not found in tenant B)
+            del_resp = await client_b.delete(
+                f"/api/v1/saved-filters/{filter_id}", headers=ORIGIN_HEADER
+            )
+            assert del_resp.status_code == 404
+
+    async def test_rbac_no_permission(self, client: AsyncClient, db_session):
+        """User without contacts:read permission gets 403 on create."""
+        from tests.conftest import create_no_perm_user
+
+        seed = await seed_tenant_and_users(db_session)
+        await create_no_perm_user(db_session, seed)
+        await login_client(client, "noperm@tenanta.com")
+        resp = await client.post(
+            "/api/v1/saved-filters",
+            json={"name": "No Perm Filter", "entity_type": "contacts", "filter_criteria": {}},
+            headers=ORIGIN_HEADER,
+        )
+        assert resp.status_code == 403

tests/test_tags.py

@@ -513,3 +513,51 @@ async def test_list_tag_entities_invalid_id(authed_client: AsyncClient):
     """GET /api/v1/tags/{invalid}/entities → 400."""
     resp = await authed_client.get("/api/v1/tags/bad-uuid/entities", headers=ORIGIN_HEADER)
     assert resp.status_code == 400
+
+
+# ── Cross-tenant isolation + RBAC tests ──
+
+
+@pytest.mark.asyncio
+async def test_cross_tenant_tag_isolation(plugin_app, db_session):
+    """Tag created in tenant A is not visible to tenant B."""
+    from tests.conftest import seed_tenant_and_users
+
+    await seed_tenant_and_users(db_session)
+    transport = ASGITransport(app=plugin_app)
+
+    async with AsyncClient(transport=transport, base_url="http://test") as client_a:
+        await login_client(client_a, "admin@tenanta.com")
+        create_resp = await client_a.post(
+            "/api/v1/tags",
+            json={"name": "Tenant A Tag", "color": "#FF0000"},
+            headers=ORIGIN_HEADER,
+        )
+        assert create_resp.status_code == 201
+        tag_id = create_resp.json()["id"]
+
+        # Tenant B admin must not see tenant A's tag
+        async with AsyncClient(transport=transport, base_url="http://test") as client_b:
+            await login_client(client_b, "admin@tenantb.com")
+            list_resp = await client_b.get("/api/v1/tags", headers=ORIGIN_HEADER)
+            assert list_resp.status_code == 200
+            assert all(t["id"] != tag_id for t in list_resp.json())
+
+
+@pytest.mark.asyncio
+async def test_rbac_tag_no_permission(plugin_app, db_session):
+    """User without tags:write permission gets 403 on create."""
+    from tests.conftest import create_no_perm_user, seed_tenant_and_users
+
+    seed = await seed_tenant_and_users(db_session)
+    await create_no_perm_user(db_session, seed)
+    transport = ASGITransport(app=plugin_app)
+
+    async with AsyncClient(transport=transport, base_url="http://test") as client:
+        await login_client(client, "noperm@tenanta.com")
+        resp = await client.post(
+            "/api/v1/tags",
+            json={"name": "No Perm Tag", "color": "#00FF00"},
+            headers=ORIGIN_HEADER,
+        )
+        assert resp.status_code == 403

tests/test_tasks.py

@@ -180,3 +180,49 @@ class TestTaskDelete:
         # Verify it's gone from list
         list_resp = await tasks_client.get("/api/v1/tasks", headers=ORIGIN_HEADER)
         assert not any(t["id"] == task_id for t in list_resp.json()["items"])
+
+
+# ── Cross-tenant isolation test ──
+@pytest.mark.asyncio
+class TestTaskCrossTenant:
+    """Tasks must not leak across tenants."""
+
+    async def test_cross_tenant_isolation(self, tasks_app, db_session):
+        """Task created in tenant A is not visible to tenant B."""
+        from httpx import ASGITransport, AsyncClient
+
+        await seed_tenant_and_users(db_session)
+        transport = ASGITransport(app=tasks_app)
+        async with AsyncClient(transport=transport, base_url="http://test") as client_a:
+            await login_client(client_a, "admin@tenanta.com")
+            create_resp = await client_a.post(
+                "/api/v1/tasks",
+                json={"title": "Tenant A Task"},
+                headers=ORIGIN_HEADER,
+            )
+            assert create_resp.status_code == 201
+            task_id = create_resp.json()["id"]
+
+            async with AsyncClient(transport=transport, base_url="http://test") as client_b:
+                await login_client(client_b, "admin@tenantb.com")
+                # List must not contain tenant A's task
+                list_resp = await client_b.get("/api/v1/tasks", headers=ORIGIN_HEADER)
+                assert list_resp.status_code == 200
+                assert all(t["id"] != task_id for t in list_resp.json()["items"])
+                # Get must 404
+                get_resp = await client_b.get(f"/api/v1/tasks/{task_id}", headers=ORIGIN_HEADER)
+                assert get_resp.status_code == 404
+
+    async def test_rbac_no_permission(self, tasks_client: AsyncClient, db_session):
+        """User without tasks:write permission gets 403 on create."""
+        from tests.conftest import create_no_perm_user
+
+        seed = await seed_tenant_and_users(db_session)
+        await create_no_perm_user(db_session, seed)
+        await login_client(tasks_client, "noperm@tenanta.com")
+        resp = await tasks_client.post(
+            "/api/v1/tasks",
+            json={"title": "No Perm Task"},
+            headers=ORIGIN_HEADER,
+        )
+        assert resp.status_code == 403

tests/test_tenant.py

@@ -186,7 +186,8 @@ class TestFieldPermissions:
         )
         db_session.add(sales_user)
         await db_session.flush()
-        ut = UserTenant(user_id=sales_user.id, tenant_id=seed["tenant_a"].id, is_default=True, role="sales_rep", role_id=seed["custom_role"].id)
+        custom_role = seed.get("custom_role")
+        ut = UserTenant(user_id=sales_user.id, tenant_id=seed["tenant_a"].id, is_default=True, role="sales_rep", role_id=custom_role.id if custom_role else None)
         db_session.add(ut)
         await db_session.commit()
 

tests/test_workflows.py

@@ -1777,3 +1777,52 @@ async def test_service_get_instance_not_found(db_session):
 
     result = await get_instance(db_session, tenant_id, str(uuid.uuid4()))
     assert result is None
+
+
+# ── Cross-tenant isolation + RBAC tests ──
+
+
+@pytest.mark.asyncio
+async def test_cross_tenant_workflow_isolation(client: AsyncClient, db_session):
+    """Workflow created in tenant A is not visible to tenant B."""
+    from httpx import ASGITransport
+    from httpx import AsyncClient as AC
+
+    import app.main
+
+    await seed_tenant_and_users(db_session)
+    await login_client(client, "admin@tenanta.com")
+    create_resp = await client.post(
+        "/api/v1/workflows",
+        json={"name": "Tenant A WF", "steps": VALID_STEPS},
+        headers=ORIGIN_HEADER,
+    )
+    assert create_resp.status_code == 201
+    wf_id = create_resp.json()["id"]
+
+    # Tenant B admin must not see tenant A's workflow
+    app_instance = app.main.app
+    async with AC(transport=ASGITransport(app=app_instance), base_url="http://test") as client_b:
+        await login_client(client_b, "admin@tenantb.com")
+        list_resp = await client_b.get("/api/v1/workflows")
+        assert list_resp.status_code == 200
+        assert all(w["id"] != wf_id for w in list_resp.json()["items"])
+        # Direct access to tenant A's workflow → 404
+        get_resp = await client_b.get(f"/api/v1/workflows/{wf_id}")
+        assert get_resp.status_code == 404
+
+
+@pytest.mark.asyncio
+async def test_rbac_workflow_no_permission(client: AsyncClient, db_session):
+    """User without workflows:write permission gets 403 on create."""
+    from tests.conftest import create_no_perm_user
+
+    seed = await seed_tenant_and_users(db_session)
+    await create_no_perm_user(db_session, seed)
+    await login_client(client, "noperm@tenanta.com")
+    resp = await client.post(
+        "/api/v1/workflows",
+        json={"name": "No Perm WF", "steps": VALID_STEPS},
+        headers=ORIGIN_HEADER,
+    )
+    assert resp.status_code == 403