feat(rbac): hide unauthorized menu sections and enforce route guards

backend/src/api/routes/__init__.py

@@ -6,7 +6,7 @@
 # @RELATION: DEPENDS_ON -> importlib
 # @INVARIANT: Only names listed in __all__ are importable via __getattr__.
 
-__all__ = ['plugins', 'tasks', 'settings', 'connections', 'environments', 'mappings', 'migration', 'git', 'storage', 'admin', 'reports', 'assistant', 'clean_release']
+__all__ = ['plugins', 'tasks', 'settings', 'connections', 'environments', 'mappings', 'migration', 'git', 'storage', 'admin', 'reports', 'assistant', 'clean_release', 'profile']
 
 
 # [DEF:__getattr__:Function]

backend/src/api/routes/__tests__/test_dashboards.py

@@ -11,9 +11,11 @@ from fastapi.testclient import TestClient
 from src.app import app
 from src.api.routes.dashboards import DashboardsResponse
 from src.dependencies import get_current_user, has_permission, get_config_manager, get_task_manager, get_resource_service, get_mapping_service
+from src.core.database import get_db
 
 # Global mock user for get_current_user dependency overrides
 mock_user = MagicMock()
+mock_user.id = "u-1"
 mock_user.username = "testuser"
 mock_user.roles = []
 admin_role = MagicMock()
@@ -27,11 +29,14 @@ def mock_deps():
     resource_service = MagicMock()
     mapping_service = MagicMock()
     
+    db = MagicMock()
+
     app.dependency_overrides[get_config_manager] = lambda: config_manager
     app.dependency_overrides[get_task_manager] = lambda: task_manager
     app.dependency_overrides[get_resource_service] = lambda: resource_service
     app.dependency_overrides[get_mapping_service] = lambda: mapping_service
     app.dependency_overrides[get_current_user] = lambda: mock_user
+    app.dependency_overrides[get_db] = lambda: db
     
     app.dependency_overrides[has_permission("plugin:migration", "READ")] = lambda: mock_user
     app.dependency_overrides[has_permission("plugin:migration", "EXECUTE")] = lambda: mock_user
@@ -42,7 +47,8 @@ def mock_deps():
         "config": config_manager,
         "task": task_manager,
         "resource": resource_service,
-        "mapping": mapping_service
+        "mapping": mapping_service,
+        "db": db,
     }
     app.dependency_overrides.clear()
 
@@ -495,4 +501,309 @@ def test_get_dashboard_thumbnail_success(mock_deps):
 # [/DEF:test_get_dashboard_thumbnail_success:Function]
 
 
+# [DEF:_build_profile_preference_stub:Function]
+# @PURPOSE: Creates profile preference payload stub for dashboards filter contract tests.
+# @PRE: username can be empty; enabled indicates profile-default toggle state.
+# @POST: Returns object compatible with ProfileService.get_my_preference contract.
+def _build_profile_preference_stub(username: str, enabled: bool):
+    preference = MagicMock()
+    preference.superset_username = username
+    preference.superset_username_normalized = str(username or "").strip().lower() or None
+    preference.show_only_my_dashboards = bool(enabled)
+
+    payload = MagicMock()
+    payload.preference = preference
+    return payload
+# [/DEF:_build_profile_preference_stub:Function]
+
+
+# [DEF:_matches_actor_case_insensitive:Function]
+# @PURPOSE: Applies trim + case-insensitive owners OR modified_by matching used by route contract tests.
+# @PRE: owners can be None or list-like values.
+# @POST: Returns True when bound username matches any owner or modified_by.
+def _matches_actor_case_insensitive(bound_username, owners, modified_by):
+    normalized_bound = str(bound_username or "").strip().lower()
+    if not normalized_bound:
+        return False
+
+    owner_tokens = []
+    for owner in owners or []:
+        token = str(owner or "").strip().lower()
+        if token:
+            owner_tokens.append(token)
+
+    modified_token = str(modified_by or "").strip().lower()
+    return normalized_bound in owner_tokens or bool(modified_token and modified_token == normalized_bound)
+# [/DEF:_matches_actor_case_insensitive:Function]
+
+
+# [DEF:test_get_dashboards_profile_filter_contract_owners_or_modified_by:Function]
+# @TEST: GET /api/dashboards applies profile-default filter with owners OR modified_by trim+case-insensitive semantics.
+# @PRE: Current user has enabled profile-default preference and bound username.
+# @POST: Response includes only matching dashboards and effective_profile_filter metadata.
+def test_get_dashboards_profile_filter_contract_owners_or_modified_by(mock_deps):
+    mock_env = MagicMock()
+    mock_env.id = "prod"
+    mock_deps["config"].get_environments.return_value = [mock_env]
+    mock_deps["task"].get_all_tasks.return_value = []
+    mock_deps["resource"].get_dashboards_with_status = AsyncMock(return_value=[
+        {
+            "id": 1,
+            "title": "Owner Match",
+            "slug": "owner-match",
+            "owners": ["  John_Doe  "],
+            "modified_by": "someone_else",
+        },
+        {
+            "id": 2,
+            "title": "Modifier Match",
+            "slug": "modifier-match",
+            "owners": ["analytics-team"],
+            "modified_by": "  JOHN_DOE ",
+        },
+        {
+            "id": 3,
+            "title": "No Match",
+            "slug": "no-match",
+            "owners": ["another-user"],
+            "modified_by": "nobody",
+        },
+    ])
+
+    with patch("src.api.routes.dashboards.ProfileService") as profile_service_cls:
+        profile_service = MagicMock()
+        profile_service.get_my_preference.return_value = _build_profile_preference_stub(
+            username="  JOHN_DOE  ",
+            enabled=True,
+        )
+        profile_service.matches_dashboard_actor.side_effect = _matches_actor_case_insensitive
+        profile_service_cls.return_value = profile_service
+
+        response = client.get(
+            "/api/dashboards?env_id=prod&page_context=dashboards_main&apply_profile_default=true"
+        )
+
+    assert response.status_code == 200
+    payload = response.json()
+
+    assert payload["total"] == 2
+    assert {item["id"] for item in payload["dashboards"]} == {1, 2}
+    assert payload["effective_profile_filter"]["applied"] is True
+    assert payload["effective_profile_filter"]["source_page"] == "dashboards_main"
+    assert payload["effective_profile_filter"]["override_show_all"] is False
+    assert payload["effective_profile_filter"]["username"] == "john_doe"
+    assert payload["effective_profile_filter"]["match_logic"] == "owners_or_modified_by"
+# [/DEF:test_get_dashboards_profile_filter_contract_owners_or_modified_by:Function]
+
+
+# [DEF:test_get_dashboards_override_show_all_contract:Function]
+# @TEST: GET /api/dashboards honors override_show_all and disables profile-default filter for current page.
+# @PRE: Profile-default preference exists but override_show_all=true query is provided.
+# @POST: Response remains unfiltered and effective_profile_filter.applied is false.
+def test_get_dashboards_override_show_all_contract(mock_deps):
+    mock_env = MagicMock()
+    mock_env.id = "prod"
+    mock_deps["config"].get_environments.return_value = [mock_env]
+    mock_deps["task"].get_all_tasks.return_value = []
+    mock_deps["resource"].get_dashboards_with_status = AsyncMock(return_value=[
+        {"id": 1, "title": "Dash A", "slug": "dash-a", "owners": ["john_doe"], "modified_by": "john_doe"},
+        {"id": 2, "title": "Dash B", "slug": "dash-b", "owners": ["other"], "modified_by": "other"},
+    ])
+
+    with patch("src.api.routes.dashboards.ProfileService") as profile_service_cls:
+        profile_service = MagicMock()
+        profile_service.get_my_preference.return_value = _build_profile_preference_stub(
+            username="john_doe",
+            enabled=True,
+        )
+        profile_service.matches_dashboard_actor.side_effect = _matches_actor_case_insensitive
+        profile_service_cls.return_value = profile_service
+
+        response = client.get(
+            "/api/dashboards?env_id=prod&page_context=dashboards_main&apply_profile_default=true&override_show_all=true"
+        )
+
+    assert response.status_code == 200
+    payload = response.json()
+
+    assert payload["total"] == 2
+    assert {item["id"] for item in payload["dashboards"]} == {1, 2}
+    assert payload["effective_profile_filter"]["applied"] is False
+    assert payload["effective_profile_filter"]["source_page"] == "dashboards_main"
+    assert payload["effective_profile_filter"]["override_show_all"] is True
+    assert payload["effective_profile_filter"]["username"] is None
+    assert payload["effective_profile_filter"]["match_logic"] is None
+    profile_service.matches_dashboard_actor.assert_not_called()
+# [/DEF:test_get_dashboards_override_show_all_contract:Function]
+
+
+# [DEF:test_get_dashboards_profile_filter_no_match_results_contract:Function]
+# @TEST: GET /api/dashboards returns empty result set when profile-default filter is active and no dashboard actors match.
+# @PRE: Profile-default preference is enabled with bound username and all dashboards are non-matching.
+# @POST: Response total is 0 with deterministic pagination and active effective_profile_filter metadata.
+def test_get_dashboards_profile_filter_no_match_results_contract(mock_deps):
+    mock_env = MagicMock()
+    mock_env.id = "prod"
+    mock_deps["config"].get_environments.return_value = [mock_env]
+    mock_deps["task"].get_all_tasks.return_value = []
+    mock_deps["resource"].get_dashboards_with_status = AsyncMock(return_value=[
+        {
+            "id": 101,
+            "title": "Team Dashboard",
+            "slug": "team-dashboard",
+            "owners": ["analytics-team"],
+            "modified_by": "someone_else",
+        },
+        {
+            "id": 102,
+            "title": "Ops Dashboard",
+            "slug": "ops-dashboard",
+            "owners": ["ops-user"],
+            "modified_by": "ops-user",
+        },
+    ])
+
+    with patch("src.api.routes.dashboards.ProfileService") as profile_service_cls:
+        profile_service = MagicMock()
+        profile_service.get_my_preference.return_value = _build_profile_preference_stub(
+            username="john_doe",
+            enabled=True,
+        )
+        profile_service.matches_dashboard_actor.side_effect = _matches_actor_case_insensitive
+        profile_service_cls.return_value = profile_service
+
+        response = client.get(
+            "/api/dashboards?env_id=prod&page_context=dashboards_main&apply_profile_default=true"
+        )
+
+    assert response.status_code == 200
+    payload = response.json()
+
+    assert payload["total"] == 0
+    assert payload["dashboards"] == []
+    assert payload["page"] == 1
+    assert payload["page_size"] == 10
+    assert payload["total_pages"] == 1
+    assert payload["effective_profile_filter"]["applied"] is True
+    assert payload["effective_profile_filter"]["source_page"] == "dashboards_main"
+    assert payload["effective_profile_filter"]["override_show_all"] is False
+    assert payload["effective_profile_filter"]["username"] == "john_doe"
+    assert payload["effective_profile_filter"]["match_logic"] == "owners_or_modified_by"
+# [/DEF:test_get_dashboards_profile_filter_no_match_results_contract:Function]
+
+
+# [DEF:test_get_dashboards_page_context_other_disables_profile_default:Function]
+# @TEST: GET /api/dashboards does not auto-apply profile-default filter outside dashboards_main page context.
+# @PRE: Profile-default preference exists but page_context=other query is provided.
+# @POST: Response remains unfiltered and metadata reflects source_page=other.
+def test_get_dashboards_page_context_other_disables_profile_default(mock_deps):
+    mock_env = MagicMock()
+    mock_env.id = "prod"
+    mock_deps["config"].get_environments.return_value = [mock_env]
+    mock_deps["task"].get_all_tasks.return_value = []
+    mock_deps["resource"].get_dashboards_with_status = AsyncMock(return_value=[
+        {"id": 1, "title": "Dash A", "slug": "dash-a", "owners": ["john_doe"], "modified_by": "john_doe"},
+        {"id": 2, "title": "Dash B", "slug": "dash-b", "owners": ["other"], "modified_by": "other"},
+    ])
+
+    with patch("src.api.routes.dashboards.ProfileService") as profile_service_cls:
+        profile_service = MagicMock()
+        profile_service.get_my_preference.return_value = _build_profile_preference_stub(
+            username="john_doe",
+            enabled=True,
+        )
+        profile_service.matches_dashboard_actor.side_effect = _matches_actor_case_insensitive
+        profile_service_cls.return_value = profile_service
+
+        response = client.get(
+            "/api/dashboards?env_id=prod&page_context=other&apply_profile_default=true"
+        )
+
+    assert response.status_code == 200
+    payload = response.json()
+
+    assert payload["total"] == 2
+    assert {item["id"] for item in payload["dashboards"]} == {1, 2}
+    assert payload["effective_profile_filter"]["applied"] is False
+    assert payload["effective_profile_filter"]["source_page"] == "other"
+    assert payload["effective_profile_filter"]["override_show_all"] is False
+    assert payload["effective_profile_filter"]["username"] is None
+    assert payload["effective_profile_filter"]["match_logic"] is None
+    profile_service.matches_dashboard_actor.assert_not_called()
+# [/DEF:test_get_dashboards_page_context_other_disables_profile_default:Function]
+
+
+# [DEF:test_get_dashboards_profile_filter_matches_display_alias_without_detail_fanout:Function]
+# @TEST: GET /api/dashboards resolves Superset display-name alias once and filters without per-dashboard detail calls.
+# @PRE: Profile-default filter is active, bound username is `admin`, dashboard actors contain display labels.
+# @POST: Route matches by alias (`Superset Admin`) and does not call `SupersetClient.get_dashboard` in list filter path.
+def test_get_dashboards_profile_filter_matches_display_alias_without_detail_fanout(mock_deps):
+    mock_env = MagicMock()
+    mock_env.id = "prod"
+    mock_deps["config"].get_environments.return_value = [mock_env]
+    mock_deps["task"].get_all_tasks.return_value = []
+    mock_deps["resource"].get_dashboards_with_status = AsyncMock(return_value=[
+        {
+            "id": 5,
+            "title": "Alias Match",
+            "slug": "alias-match",
+            "owners": [],
+            "created_by": None,
+            "modified_by": "Superset Admin",
+        },
+        {
+            "id": 6,
+            "title": "Alias No Match",
+            "slug": "alias-no-match",
+            "owners": [],
+            "created_by": None,
+            "modified_by": "Other User",
+        },
+    ])
+
+    with patch("src.api.routes.dashboards.ProfileService") as profile_service_cls, patch(
+        "src.api.routes.dashboards.SupersetClient"
+    ) as superset_client_cls, patch(
+        "src.api.routes.dashboards.SupersetAccountLookupAdapter"
+    ) as lookup_adapter_cls:
+        profile_service = MagicMock()
+        profile_service.get_my_preference.return_value = _build_profile_preference_stub(
+            username="admin",
+            enabled=True,
+        )
+        profile_service.matches_dashboard_actor.side_effect = _matches_actor_case_insensitive
+        profile_service_cls.return_value = profile_service
+
+        superset_client = MagicMock()
+        superset_client_cls.return_value = superset_client
+
+        lookup_adapter = MagicMock()
+        lookup_adapter.get_users_page.return_value = {
+            "items": [
+                {
+                    "environment_id": "prod",
+                    "username": "admin",
+                    "display_name": "Superset Admin",
+                    "email": "admin@example.com",
+                    "is_active": True,
+                }
+            ],
+            "total": 1,
+        }
+        lookup_adapter_cls.return_value = lookup_adapter
+
+        response = client.get(
+            "/api/dashboards?env_id=prod&page_context=dashboards_main&apply_profile_default=true"
+        )
+
+    assert response.status_code == 200
+    payload = response.json()
+    assert payload["total"] == 1
+    assert {item["id"] for item in payload["dashboards"]} == {5}
+    assert payload["effective_profile_filter"]["applied"] is True
+    lookup_adapter.get_users_page.assert_called_once()
+    superset_client.get_dashboard.assert_not_called()
+# [/DEF:test_get_dashboards_profile_filter_matches_display_alias_without_detail_fanout:Function]
+
+
 # [/DEF:backend.src.api.routes.__tests__.test_dashboards:Module]

backend/src/api/routes/__tests__/test_profile_api.py

@@ -0,0 +1,243 @@
+# [DEF:backend.src.api.routes.__tests__.test_profile_api:Module]
+# @TIER: STANDARD
+# @SEMANTICS: tests, profile, api, preferences, lookup, contract
+# @PURPOSE: Verifies profile API route contracts for preference read/update and Superset account lookup.
+# @LAYER: API
+# @RELATION: TESTS -> backend.src.api.routes.profile
+
+# [SECTION: IMPORTS]
+from datetime import datetime, timezone
+from unittest.mock import MagicMock, patch
+
+from fastapi.testclient import TestClient
+
+from src.app import app
+from src.core.database import get_db
+from src.dependencies import get_config_manager, get_current_user
+from src.schemas.profile import (
+    ProfilePreference,
+    ProfilePreferenceResponse,
+    SupersetAccountCandidate,
+    SupersetAccountLookupResponse,
+)
+from src.services.profile_service import (
+    EnvironmentNotFoundError,
+    ProfileAuthorizationError,
+    ProfileValidationError,
+)
+# [/SECTION]
+
+
+client = TestClient(app)
+
+
+# [DEF:mock_profile_route_dependencies:Function]
+# @PURPOSE: Provides deterministic dependency overrides for profile route tests.
+# @PRE: App instance is initialized.
+# @POST: Dependencies are overridden for current test and restored afterward.
+def mock_profile_route_dependencies():
+    mock_user = MagicMock()
+    mock_user.id = "u-1"
+    mock_user.username = "test-user"
+
+    mock_db = MagicMock()
+    mock_config_manager = MagicMock()
+
+    app.dependency_overrides[get_current_user] = lambda: mock_user
+    app.dependency_overrides[get_db] = lambda: mock_db
+    app.dependency_overrides[get_config_manager] = lambda: mock_config_manager
+
+    return mock_user, mock_db, mock_config_manager
+# [/DEF:mock_profile_route_dependencies:Function]
+
+
+# [DEF:profile_route_deps_fixture:Function]
+# @PURPOSE: Pytest fixture wrapper for profile route dependency overrides.
+# @PRE: None.
+# @POST: Yields overridden dependencies and clears overrides after test.
+import pytest
+
+
+@pytest.fixture(autouse=True)
+def profile_route_deps_fixture():
+    yielded = mock_profile_route_dependencies()
+    yield yielded
+    app.dependency_overrides.clear()
+# [/DEF:profile_route_deps_fixture:Function]
+
+
+# [DEF:_build_preference_response:Function]
+# @PURPOSE: Builds stable profile preference response payload for route tests.
+# @PRE: user_id is provided.
+# @POST: Returns ProfilePreferenceResponse object with deterministic timestamps.
+def _build_preference_response(user_id: str = "u-1") -> ProfilePreferenceResponse:
+    now = datetime.now(timezone.utc)
+    return ProfilePreferenceResponse(
+        status="success",
+        message="Preference loaded",
+        preference=ProfilePreference(
+            user_id=user_id,
+            superset_username="John_Doe",
+            superset_username_normalized="john_doe",
+            show_only_my_dashboards=True,
+            created_at=now,
+            updated_at=now,
+        ),
+    )
+# [/DEF:_build_preference_response:Function]
+
+
+# [DEF:test_get_profile_preferences_returns_self_payload:Function]
+# @PURPOSE: Verifies GET /api/profile/preferences returns stable self-scoped payload.
+# @PRE: Authenticated user context is available.
+# @POST: Response status is 200 and payload contains current user preference.
+def test_get_profile_preferences_returns_self_payload(profile_route_deps_fixture):
+    mock_user, _, _ = profile_route_deps_fixture
+    service = MagicMock()
+    service.get_my_preference.return_value = _build_preference_response(user_id=mock_user.id)
+
+    with patch("src.api.routes.profile._get_profile_service", return_value=service):
+        response = client.get("/api/profile/preferences")
+
+    assert response.status_code == 200
+    payload = response.json()
+    assert payload["status"] == "success"
+    assert payload["preference"]["user_id"] == mock_user.id
+    assert payload["preference"]["superset_username_normalized"] == "john_doe"
+    service.get_my_preference.assert_called_once_with(mock_user)
+# [/DEF:test_get_profile_preferences_returns_self_payload:Function]
+
+
+# [DEF:test_patch_profile_preferences_success:Function]
+# @PURPOSE: Verifies PATCH /api/profile/preferences persists valid payload through route mapping.
+# @PRE: Valid request payload and authenticated user.
+# @POST: Response status is 200 with saved preference payload.
+def test_patch_profile_preferences_success(profile_route_deps_fixture):
+    mock_user, _, _ = profile_route_deps_fixture
+    service = MagicMock()
+    service.update_my_preference.return_value = _build_preference_response(user_id=mock_user.id)
+
+    with patch("src.api.routes.profile._get_profile_service", return_value=service):
+        response = client.patch(
+            "/api/profile/preferences",
+            json={
+                "superset_username": "John_Doe",
+                "show_only_my_dashboards": True,
+            },
+        )
+
+    assert response.status_code == 200
+    payload = response.json()
+    assert payload["status"] == "success"
+    assert payload["preference"]["superset_username"] == "John_Doe"
+    assert payload["preference"]["show_only_my_dashboards"] is True
+    service.update_my_preference.assert_called_once()
+# [/DEF:test_patch_profile_preferences_success:Function]
+
+
+# [DEF:test_patch_profile_preferences_validation_error:Function]
+# @PURPOSE: Verifies route maps domain validation failure to HTTP 422 with actionable details.
+# @PRE: Service raises ProfileValidationError.
+# @POST: Response status is 422 and includes validation messages.
+def test_patch_profile_preferences_validation_error(profile_route_deps_fixture):
+    service = MagicMock()
+    service.update_my_preference.side_effect = ProfileValidationError(
+        ["Superset username is required when default filter is enabled."]
+    )
+
+    with patch("src.api.routes.profile._get_profile_service", return_value=service):
+        response = client.patch(
+            "/api/profile/preferences",
+            json={
+                "superset_username": "",
+                "show_only_my_dashboards": True,
+            },
+        )
+
+    assert response.status_code == 422
+    payload = response.json()
+    assert "detail" in payload
+    assert "Superset username is required when default filter is enabled." in payload["detail"]
+# [/DEF:test_patch_profile_preferences_validation_error:Function]
+
+
+# [DEF:test_patch_profile_preferences_cross_user_denied:Function]
+# @PURPOSE: Verifies route maps domain authorization guard failure to HTTP 403.
+# @PRE: Service raises ProfileAuthorizationError.
+# @POST: Response status is 403 with denial message.
+def test_patch_profile_preferences_cross_user_denied(profile_route_deps_fixture):
+    service = MagicMock()
+    service.update_my_preference.side_effect = ProfileAuthorizationError(
+        "Cross-user preference mutation is forbidden"
+    )
+
+    with patch("src.api.routes.profile._get_profile_service", return_value=service):
+        response = client.patch(
+            "/api/profile/preferences",
+            json={
+                "superset_username": "john_doe",
+                "show_only_my_dashboards": True,
+            },
+        )
+
+    assert response.status_code == 403
+    payload = response.json()
+    assert payload["detail"] == "Cross-user preference mutation is forbidden"
+# [/DEF:test_patch_profile_preferences_cross_user_denied:Function]
+
+
+# [DEF:test_lookup_superset_accounts_success:Function]
+# @PURPOSE: Verifies lookup route returns success payload with normalized candidates.
+# @PRE: Valid environment_id and service success response.
+# @POST: Response status is 200 and items list is returned.
+def test_lookup_superset_accounts_success(profile_route_deps_fixture):
+    service = MagicMock()
+    service.lookup_superset_accounts.return_value = SupersetAccountLookupResponse(
+        status="success",
+        environment_id="dev",
+        page_index=0,
+        page_size=20,
+        total=1,
+        warning=None,
+        items=[
+            SupersetAccountCandidate(
+                environment_id="dev",
+                username="john_doe",
+                display_name="John Doe",
+                email="john@example.local",
+                is_active=True,
+            )
+        ],
+    )
+
+    with patch("src.api.routes.profile._get_profile_service", return_value=service):
+        response = client.get("/api/profile/superset-accounts?environment_id=dev")
+
+    assert response.status_code == 200
+    payload = response.json()
+    assert payload["status"] == "success"
+    assert payload["environment_id"] == "dev"
+    assert payload["total"] == 1
+    assert payload["items"][0]["username"] == "john_doe"
+# [/DEF:test_lookup_superset_accounts_success:Function]
+
+
+# [DEF:test_lookup_superset_accounts_env_not_found:Function]
+# @PURPOSE: Verifies lookup route maps missing environment to HTTP 404.
+# @PRE: Service raises EnvironmentNotFoundError.
+# @POST: Response status is 404 with explicit message.
+def test_lookup_superset_accounts_env_not_found(profile_route_deps_fixture):
+    service = MagicMock()
+    service.lookup_superset_accounts.side_effect = EnvironmentNotFoundError(
+        "Environment 'missing-env' not found"
+    )
+
+    with patch("src.api.routes.profile._get_profile_service", return_value=service):
+        response = client.get("/api/profile/superset-accounts?environment_id=missing-env")
+
+    assert response.status_code == 404
+    payload = response.json()
+    assert payload["detail"] == "Environment 'missing-env' not found"
+# [/DEF:test_lookup_superset_accounts_env_not_found:Function]
+
+# [/DEF:backend.src.api.routes.__tests__.test_profile_api:Module]

backend/src/api/routes/dashboards.py

@@ -34,14 +34,26 @@
 # [SECTION: IMPORTS]
 from fastapi import APIRouter, Depends, HTTPException, Query, Response
 from fastapi.responses import JSONResponse
-from typing import List, Optional, Dict, Any
+from typing import List, Optional, Dict, Any, Literal
 import re
 from urllib.parse import urlparse
 from pydantic import BaseModel, Field
-from ...dependencies import get_config_manager, get_task_manager, get_resource_service, get_mapping_service, has_permission
+from sqlalchemy.orm import Session
+from ...dependencies import (
+    get_config_manager,
+    get_task_manager,
+    get_resource_service,
+    get_mapping_service,
+    get_current_user,
+    has_permission,
+)
+from ...core.database import get_db
 from ...core.logger import logger, belief_scope
 from ...core.superset_client import SupersetClient
+from ...core.superset_profile_lookup import SupersetAccountLookupAdapter
 from ...core.utils.network import DashboardNotFoundError
+from ...models.auth import User
+from ...services.profile_service import ProfileService
 from ...services.resource_service import ResourceService
 # [/SECTION]
 
@@ -79,6 +91,15 @@ class DashboardItem(BaseModel):
     last_task: Optional[LastTask] = None
 # [/DEF:DashboardItem:DataClass]
 
+# [DEF:EffectiveProfileFilter:DataClass]
+class EffectiveProfileFilter(BaseModel):
+    applied: bool
+    source_page: Literal["dashboards_main", "other"] = "dashboards_main"
+    override_show_all: bool = False
+    username: Optional[str] = None
+    match_logic: Optional[Literal["owners_or_modified_by"]] = None
+# [/DEF:EffectiveProfileFilter:DataClass]
+
 # [DEF:DashboardsResponse:DataClass]
 class DashboardsResponse(BaseModel):
     dashboards: List[DashboardItem]
@@ -86,6 +107,7 @@ class DashboardsResponse(BaseModel):
     page: int
     page_size: int
     total_pages: int
+    effective_profile_filter: Optional[EffectiveProfileFilter] = None
 # [/DEF:DashboardsResponse:DataClass]
 
 # [DEF:DashboardChartItem:DataClass]
@@ -242,6 +264,101 @@ def _dashboard_git_filter_value(dashboard: Dict[str, Any]) -> str:
     return "pending"
 # [/DEF:_dashboard_git_filter_value:Function]
 
+# [DEF:_normalize_actor_alias_token:Function]
+# @PURPOSE: Normalize actor alias token to comparable trim+lower text.
+# @PRE: value can be scalar/None.
+# @POST: Returns normalized token or None.
+def _normalize_actor_alias_token(value: Any) -> Optional[str]:
+    token = str(value or "").strip().lower()
+    return token or None
+# [/DEF:_normalize_actor_alias_token:Function]
+
+
+# [DEF:_resolve_profile_actor_aliases:Function]
+# @PURPOSE: Resolve stable actor aliases for profile filtering without per-dashboard detail fan-out.
+# @PRE: bound username is available and env is valid.
+# @POST: Returns at least normalized username; may include Superset display-name alias.
+# @SIDE_EFFECT: Performs at most one Superset users-lookup request.
+def _resolve_profile_actor_aliases(env: Any, bound_username: str) -> List[str]:
+    normalized_bound = _normalize_actor_alias_token(bound_username)
+    if not normalized_bound:
+        return []
+
+    aliases: List[str] = [normalized_bound]
+    try:
+        client = SupersetClient(env)
+        adapter = SupersetAccountLookupAdapter(
+            network_client=client.network,
+            environment_id=str(getattr(env, "id", "")),
+        )
+        lookup_payload = adapter.get_users_page(
+            search=normalized_bound,
+            page_index=0,
+            page_size=20,
+            sort_column="username",
+            sort_order="asc",
+        )
+        lookup_items = (
+            lookup_payload.get("items", [])
+            if isinstance(lookup_payload, dict)
+            else []
+        )
+
+        matched_item: Optional[Dict[str, Any]] = None
+        for item in lookup_items:
+            if not isinstance(item, dict):
+                continue
+            if _normalize_actor_alias_token(item.get("username")) == normalized_bound:
+                matched_item = item
+                break
+
+        if matched_item is None:
+            for item in lookup_items:
+                if isinstance(item, dict):
+                    matched_item = item
+                    break
+
+        display_alias = _normalize_actor_alias_token(
+            (matched_item or {}).get("display_name")
+        )
+        if display_alias and display_alias not in aliases:
+            aliases.append(display_alias)
+
+        logger.reflect(
+            "[REFLECT] Resolved profile actor aliases "
+            f"(env={getattr(env, 'id', None)}, bound_username={normalized_bound!r}, "
+            f"lookup_items={len(lookup_items)}, aliases={aliases!r})"
+        )
+    except Exception as alias_error:
+        logger.explore(
+            "[EXPLORE] Failed to resolve profile actor aliases via Superset users lookup "
+            f"(env={getattr(env, 'id', None)}, bound_username={normalized_bound!r}): {alias_error}"
+        )
+    return aliases
+# [/DEF:_resolve_profile_actor_aliases:Function]
+
+
+# [DEF:_matches_dashboard_actor_aliases:Function]
+# @PURPOSE: Apply profile actor matching against multiple aliases (username + optional display name).
+# @PRE: actor_aliases contains normalized non-empty tokens.
+# @POST: Returns True when any alias matches owners OR modified_by.
+def _matches_dashboard_actor_aliases(
+    profile_service: ProfileService,
+    actor_aliases: List[str],
+    owners: Optional[Any],
+    modified_by: Optional[str],
+) -> bool:
+    for actor_alias in actor_aliases:
+        if profile_service.matches_dashboard_actor(
+            bound_username=actor_alias,
+            owners=owners,
+            modified_by=modified_by,
+        ):
+            return True
+    return False
+# [/DEF:_matches_dashboard_actor_aliases:Function]
+
+
 # [DEF:get_dashboards:Function]
 # @PURPOSE: Fetch list of dashboards from a specific environment with Git status and last task status
 # @PRE: env_id must be a valid environment ID
@@ -249,6 +366,7 @@ def _dashboard_git_filter_value(dashboard: Dict[str, Any]) -> str:
 # @PRE: page_size must be between 1 and 100 if provided
 # @POST: Returns a list of dashboards with enhanced metadata and pagination info
 # @POST: Response includes pagination metadata (page, page_size, total, total_pages)
+# @POST: Response includes effective profile filter metadata for main dashboards page context
 # @PARAM: env_id (str) - The environment ID to fetch dashboards from
 # @PARAM: search (Optional[str]) - Filter by title/slug
 # @PARAM: page (Optional[int]) - Page number (default: 1)
@@ -261,6 +379,9 @@ async def get_dashboards(
     search: Optional[str] = None,
     page: int = 1,
     page_size: int = 10,
+    page_context: Literal["dashboards_main", "other"] = Query(default="dashboards_main"),
+    apply_profile_default: bool = Query(default=True),
+    override_show_all: bool = Query(default=False),
     filter_title: Optional[List[str]] = Query(default=None),
     filter_git_status: Optional[List[str]] = Query(default=None),
     filter_llm_status: Optional[List[str]] = Query(default=None),
@@ -269,26 +390,73 @@ async def get_dashboards(
     config_manager=Depends(get_config_manager),
     task_manager=Depends(get_task_manager),
     resource_service=Depends(get_resource_service),
+    current_user: User = Depends(get_current_user),
+    db: Session = Depends(get_db),
     _ = Depends(has_permission("plugin:migration", "READ"))
 ):
-    with belief_scope("get_dashboards", f"env_id={env_id}, search={search}, page={page}, page_size={page_size}"):
-        # Validate pagination parameters
+    with belief_scope(
+        "get_dashboards",
+        (
+            f"env_id={env_id}, search={search}, page={page}, page_size={page_size}, "
+            f"page_context={page_context}, apply_profile_default={apply_profile_default}, "
+            f"override_show_all={override_show_all}"
+        ),
+    ):
         if page < 1:
             logger.error(f"[get_dashboards][Coherence:Failed] Invalid page: {page}")
             raise HTTPException(status_code=400, detail="Page must be >= 1")
         if page_size < 1 or page_size > 100:
             logger.error(f"[get_dashboards][Coherence:Failed] Invalid page_size: {page_size}")
             raise HTTPException(status_code=400, detail="Page size must be between 1 and 100")
-        
-        # Validate environment exists
+
         environments = config_manager.get_environments()
         env = next((e for e in environments if e.id == env_id), None)
         if not env:
             logger.error(f"[get_dashboards][Coherence:Failed] Environment not found: {env_id}")
             raise HTTPException(status_code=404, detail="Environment not found")
-        
+
+        profile_service = ProfileService(db=db, config_manager=config_manager)
+        bound_username: Optional[str] = None
+        can_apply_profile_filter = False
+        effective_profile_filter = EffectiveProfileFilter(
+            applied=False,
+            source_page=page_context,
+            override_show_all=bool(override_show_all),
+            username=None,
+            match_logic=None,
+        )
+
+        try:
+            profile_preference = profile_service.get_my_preference(current_user).preference
+            normalized_username = str(
+                getattr(profile_preference, "superset_username_normalized", None) or ""
+            ).strip().lower()
+            raw_username = str(
+                getattr(profile_preference, "superset_username", None) or ""
+            ).strip().lower()
+            bound_username = normalized_username or raw_username or None
+
+            can_apply_profile_filter = (
+                page_context == "dashboards_main"
+                and bool(apply_profile_default)
+                and not bool(override_show_all)
+                and bool(getattr(profile_preference, "show_only_my_dashboards", False))
+                and bool(bound_username)
+            )
+
+            effective_profile_filter = EffectiveProfileFilter(
+                applied=bool(can_apply_profile_filter),
+                source_page=page_context,
+                override_show_all=bool(override_show_all),
+                username=bound_username if can_apply_profile_filter else None,
+                match_logic="owners_or_modified_by" if can_apply_profile_filter else None,
+            )
+        except Exception as profile_error:
+            logger.explore(
+                f"[EXPLORE] Profile preference unavailable; continuing without profile-default filter: {profile_error}"
+            )
+
         try:
-            # Get all tasks for status lookup
             all_tasks = task_manager.get_all_tasks()
             title_filters = _normalize_filter_values(filter_title)
             git_filters = _normalize_filter_values(filter_git_status)
@@ -304,9 +472,9 @@ async def get_dashboards(
                     actor_filters,
                 )
             )
+            needs_full_scan = has_column_filters or bool(can_apply_profile_filter)
 
-            # Fast path: real ResourceService -> one Superset page call per API request.
-            if isinstance(resource_service, ResourceService) and not has_column_filters:
+            if isinstance(resource_service, ResourceService) and not needs_full_scan:
                 try:
                     page_payload = await resource_service.get_dashboards_page_with_status(
                         env,
@@ -333,9 +501,9 @@ async def get_dashboards(
                     if search:
                         search_lower = search.lower()
                         dashboards = [
-                            d for d in dashboards 
-                            if search_lower in d.get('title', '').lower() 
-                            or search_lower in d.get('slug', '').lower()
+                            d for d in dashboards
+                            if search_lower in d.get("title", "").lower()
+                            or search_lower in d.get("slug", "").lower()
                         ]
 
                     total = len(dashboards)
@@ -343,13 +511,52 @@ async def get_dashboards(
                     start_idx = (page - 1) * page_size
                     end_idx = start_idx + page_size
                     paginated_dashboards = dashboards[start_idx:end_idx]
-            elif isinstance(resource_service, ResourceService) and has_column_filters:
+            else:
                 dashboards = await resource_service.get_dashboards_with_status(
                     env,
                     all_tasks,
                     include_git_status=bool(git_filters),
                 )
 
+                if can_apply_profile_filter and bound_username:
+                    actor_aliases = _resolve_profile_actor_aliases(env, bound_username)
+                    if not actor_aliases:
+                        actor_aliases = [bound_username]
+                    logger.reason(
+                        "[REASON] Applying profile actor filter "
+                        f"(env={env_id}, bound_username={bound_username}, actor_aliases={actor_aliases!r}, "
+                        f"dashboards_before={len(dashboards)})"
+                    )
+                    filtered_dashboards: List[Dict[str, Any]] = []
+                    max_actor_samples = 15
+                    for index, dashboard in enumerate(dashboards):
+                        owners_value = dashboard.get("owners")
+                        created_by_value = dashboard.get("created_by")
+                        modified_by_value = dashboard.get("modified_by")
+                        matches_actor = _matches_dashboard_actor_aliases(
+                            profile_service=profile_service,
+                            actor_aliases=actor_aliases,
+                            owners=owners_value,
+                            modified_by=modified_by_value,
+                        )
+                        if index < max_actor_samples:
+                            logger.reflect(
+                                "[REFLECT] Profile actor filter sample "
+                                f"(env={env_id}, dashboard_id={dashboard.get('id')}, "
+                                f"bound_username={bound_username!r}, actor_aliases={actor_aliases!r}, "
+                                f"owners={owners_value!r}, created_by={created_by_value!r}, "
+                                f"modified_by={modified_by_value!r}, matches={matches_actor})"
+                            )
+                        if matches_actor:
+                            filtered_dashboards.append(dashboard)
+
+                    logger.reflect(
+                        "[REFLECT] Profile actor filter summary "
+                        f"(env={env_id}, bound_username={bound_username!r}, "
+                        f"dashboards_before={len(dashboards)}, dashboards_after={len(filtered_dashboards)})"
+                    )
+                    dashboards = filtered_dashboards
+
                 if search:
                     search_lower = search.lower()
                     dashboards = [
@@ -376,13 +583,21 @@ async def get_dashboards(
                         return False
 
                     changed_on_raw = str(dashboard.get("last_modified") or "").strip().lower()
-                    changed_on_prefix = changed_on_raw[:10] if len(changed_on_raw) >= 10 else changed_on_raw
-                    if changed_on_filters and changed_on_raw not in changed_on_filters and changed_on_prefix not in changed_on_filters:
+                    changed_on_prefix = (
+                        changed_on_raw[:10] if len(changed_on_raw) >= 10 else changed_on_raw
+                    )
+                    if (
+                        changed_on_filters
+                        and changed_on_raw not in changed_on_filters
+                        and changed_on_prefix not in changed_on_filters
+                    ):
                         return False
 
                     owners = dashboard.get("owners") or []
                     if isinstance(owners, list):
-                        actor_value = ", ".join(str(item).strip() for item in owners if str(item).strip()).lower()
+                        actor_value = ", ".join(
+                            str(item).strip() for item in owners if str(item).strip()
+                        ).lower()
                     else:
                         actor_value = str(owners).strip().lower()
                     if not actor_value:
@@ -391,44 +606,29 @@ async def get_dashboards(
                         return False
                     return True
 
-                dashboards = [d for d in dashboards if _matches_dashboard_filters(d)]
-                total = len(dashboards)
-                total_pages = (total + page_size - 1) // page_size if total > 0 else 1
-                start_idx = (page - 1) * page_size
-                end_idx = start_idx + page_size
-                paginated_dashboards = dashboards[start_idx:end_idx]
-            else:
-                # Compatibility path for mocked services in route tests.
-                dashboards = await resource_service.get_dashboards_with_status(
-                    env,
-                    all_tasks,
-                    include_git_status=False,
-                )
-
-                if search:
-                    search_lower = search.lower()
-                    dashboards = [
-                        d for d in dashboards 
-                        if search_lower in d.get('title', '').lower() 
-                        or search_lower in d.get('slug', '').lower()
-                    ]
+                if has_column_filters:
+                    dashboards = [d for d in dashboards if _matches_dashboard_filters(d)]
 
                 total = len(dashboards)
                 total_pages = (total + page_size - 1) // page_size if total > 0 else 1
                 start_idx = (page - 1) * page_size
                 end_idx = start_idx + page_size
                 paginated_dashboards = dashboards[start_idx:end_idx]
-            
-            logger.info(f"[get_dashboards][Coherence:OK] Returning {len(paginated_dashboards)} dashboards (page {page}/{total_pages}, total: {total})")
-            
+
+            logger.info(
+                f"[get_dashboards][Coherence:OK] Returning {len(paginated_dashboards)} dashboards "
+                f"(page {page}/{total_pages}, total: {total}, profile_filter_applied={effective_profile_filter.applied})"
+            )
+
             return DashboardsResponse(
                 dashboards=paginated_dashboards,
                 total=total,
                 page=page,
                 page_size=page_size,
-                total_pages=total_pages
+                total_pages=total_pages,
+                effective_profile_filter=effective_profile_filter,
             )
-            
+
         except Exception as e:
             logger.error(f"[get_dashboards][Coherence:Failed] Failed to fetch dashboards: {e}")
             raise HTTPException(status_code=503, detail=f"Failed to fetch dashboards: {str(e)}")

backend/src/api/routes/profile.py

@@ -0,0 +1,136 @@
+# [DEF:backend.src.api.routes.profile:Module]
+#
+# @TIER:      CRITICAL
+# @SEMANTICS: api, profile, preferences, self-service, account-lookup
+# @PURPOSE:   Exposes self-scoped profile preference endpoints and environment-based Superset account lookup.
+# @LAYER:     API
+# @RELATION:  DEPENDS_ON -> backend.src.services.profile_service
+# @RELATION:  DEPENDS_ON -> backend.src.dependencies.get_current_user
+# @RELATION:  DEPENDS_ON -> backend.src.core.database.get_db
+#
+# @INVARIANT: Endpoints are self-scoped and never mutate another user preference.
+# @UX_STATE: ProfileLoad -> Returns stable ProfilePreferenceResponse for authenticated user.
+# @UX_STATE: Saving -> Validation errors map to actionable 422 details.
+# @UX_STATE: LookupLoading -> Returns success/degraded Superset lookup payload.
+# @UX_FEEDBACK: Stable status/message/warning payloads support profile page feedback.
+# @UX_RECOVERY: Lookup degradation keeps manual username save path available.
+
+# [SECTION: IMPORTS]
+from typing import Optional
+
+from fastapi import APIRouter, Depends, HTTPException, Query
+from sqlalchemy.orm import Session
+
+from ...core.database import get_db
+from ...core.logger import logger, belief_scope
+from ...dependencies import get_config_manager, get_current_user
+from ...models.auth import User
+from ...schemas.profile import (
+    ProfilePreferenceResponse,
+    ProfilePreferenceUpdateRequest,
+    SupersetAccountLookupRequest,
+    SupersetAccountLookupResponse,
+)
+from ...services.profile_service import (
+    EnvironmentNotFoundError,
+    ProfileAuthorizationError,
+    ProfileService,
+    ProfileValidationError,
+)
+# [/SECTION]
+
+router = APIRouter(prefix="/api/profile", tags=["profile"])
+
+
+# [DEF:_get_profile_service:Function]
+# @PURPOSE: Build profile service for current request scope.
+# @PRE: db session and config manager are available.
+# @POST: Returns a ready ProfileService instance.
+def _get_profile_service(db: Session, config_manager) -> ProfileService:
+    return ProfileService(db=db, config_manager=config_manager)
+# [/DEF:_get_profile_service:Function]
+
+
+# [DEF:get_preferences:Function]
+# @PURPOSE: Get authenticated user's dashboard filter preference.
+# @PRE: Valid JWT and authenticated user context.
+# @POST: Returns preference payload for current user only.
+@router.get("/preferences", response_model=ProfilePreferenceResponse)
+async def get_preferences(
+    current_user: User = Depends(get_current_user),
+    db: Session = Depends(get_db),
+    config_manager=Depends(get_config_manager),
+):
+    with belief_scope("profile.get_preferences", f"user_id={current_user.id}"):
+        logger.reason("[REASON] Resolving current user preference")
+        service = _get_profile_service(db, config_manager)
+        return service.get_my_preference(current_user)
+# [/DEF:get_preferences:Function]
+
+
+# [DEF:update_preferences:Function]
+# @PURPOSE: Update authenticated user's dashboard filter preference.
+# @PRE: Valid JWT and valid request payload.
+# @POST: Persists normalized preference for current user or raises validation/authorization errors.
+@router.patch("/preferences", response_model=ProfilePreferenceResponse)
+async def update_preferences(
+    payload: ProfilePreferenceUpdateRequest,
+    current_user: User = Depends(get_current_user),
+    db: Session = Depends(get_db),
+    config_manager=Depends(get_config_manager),
+):
+    with belief_scope("profile.update_preferences", f"user_id={current_user.id}"):
+        service = _get_profile_service(db, config_manager)
+        try:
+            logger.reason("[REASON] Attempting preference save")
+            return service.update_my_preference(current_user=current_user, payload=payload)
+        except ProfileValidationError as exc:
+            logger.reflect("[REFLECT] Preference validation failed")
+            raise HTTPException(status_code=422, detail=exc.errors) from exc
+        except ProfileAuthorizationError as exc:
+            logger.explore("[EXPLORE] Cross-user mutation guard blocked request")
+            raise HTTPException(status_code=403, detail=str(exc)) from exc
+# [/DEF:update_preferences:Function]
+
+
+# [DEF:lookup_superset_accounts:Function]
+# @PURPOSE: Lookup Superset account candidates in selected environment.
+# @PRE: Valid JWT, authenticated context, and environment_id query parameter.
+# @POST: Returns success or degraded lookup payload with stable shape.
+@router.get("/superset-accounts", response_model=SupersetAccountLookupResponse)
+async def lookup_superset_accounts(
+    environment_id: str = Query(...),
+    search: Optional[str] = Query(default=None),
+    page_index: int = Query(default=0, ge=0),
+    page_size: int = Query(default=20, ge=1, le=100),
+    sort_column: str = Query(default="username"),
+    sort_order: str = Query(default="desc"),
+    current_user: User = Depends(get_current_user),
+    db: Session = Depends(get_db),
+    config_manager=Depends(get_config_manager),
+):
+    with belief_scope(
+        "profile.lookup_superset_accounts",
+        f"user_id={current_user.id}, environment_id={environment_id}",
+    ):
+        service = _get_profile_service(db, config_manager)
+        lookup_request = SupersetAccountLookupRequest(
+            environment_id=environment_id,
+            search=search,
+            page_index=page_index,
+            page_size=page_size,
+            sort_column=sort_column,
+            sort_order=sort_order,
+        )
+        try:
+            logger.reason("[REASON] Executing Superset account lookup")
+            return service.lookup_superset_accounts(
+                current_user=current_user,
+                request=lookup_request,
+            )
+        except EnvironmentNotFoundError as exc:
+            logger.explore("[EXPLORE] Lookup request references unknown environment")
+            raise HTTPException(status_code=404, detail=str(exc)) from exc
+# [/DEF:lookup_superset_accounts:Function]
+
+# [/DEF:backend.src.api.routes.profile:Module]

backend/src/app.py

@@ -21,7 +21,7 @@ import asyncio
 from .dependencies import get_task_manager, get_scheduler_service
 from .core.utils.network import NetworkError
 from .core.logger import logger, belief_scope
-from .api.routes import plugins, tasks, settings, environments, mappings, migration, connections, git, storage, admin, llm, dashboards, datasets, reports, assistant, clean_release
+from .api.routes import plugins, tasks, settings, environments, mappings, migration, connections, git, storage, admin, llm, dashboards, datasets, reports, assistant, clean_release, profile
 from .api import auth
 
 # [DEF:App:Global]
@@ -134,6 +134,7 @@ app.include_router(datasets.router)
 app.include_router(reports.router)
 app.include_router(assistant.router, prefix="/api/assistant", tags=["Assistant"])
 app.include_router(clean_release.router)
+app.include_router(profile.router)
 
 
 # [DEF:api.include_routers:Action]

backend/src/core/__tests__/test_superset_profile_lookup.py

@@ -0,0 +1,128 @@
+# [DEF:backend.src.core.__tests__.test_superset_profile_lookup:Module]
+# @TIER: STANDARD
+# @SEMANTICS: tests, superset, profile, lookup, fallback, sorting
+# @PURPOSE: Verifies Superset profile lookup adapter payload normalization and fallback error precedence.
+# @LAYER: Domain
+# @RELATION: TESTS -> backend.src.core.superset_profile_lookup
+
+# [SECTION: IMPORTS]
+import json
+import sys
+from pathlib import Path
+from typing import Any, Dict, List, Optional
+
+import pytest
+
+backend_dir = str(Path(__file__).parent.parent.parent.parent.resolve())
+if backend_dir not in sys.path:
+    sys.path.insert(0, backend_dir)
+
+from src.core.superset_profile_lookup import SupersetAccountLookupAdapter
+from src.core.utils.network import AuthenticationError, SupersetAPIError
+# [/SECTION]
+
+
+# [DEF:_RecordingNetworkClient:Class]
+# @PURPOSE: Records request payloads and returns scripted responses for deterministic adapter tests.
+class _RecordingNetworkClient:
+    # [DEF:__init__:Function]
+    # @PURPOSE: Initializes scripted network responses.
+    # @PRE: scripted_responses is ordered per expected request sequence.
+    # @POST: Instance stores response script and captures subsequent request calls.
+    def __init__(self, scripted_responses: List[Any]):
+        self._scripted_responses = scripted_responses
+        self.calls: List[Dict[str, Any]] = []
+    # [/DEF:__init__:Function]
+
+    # [DEF:request:Function]
+    # @PURPOSE: Mimics APIClient.request while capturing call arguments.
+    # @PRE: method and endpoint are provided.
+    # @POST: Returns scripted response or raises scripted exception.
+    def request(
+        self,
+        method: str,
+        endpoint: str,
+        params: Optional[Dict[str, Any]] = None,
+        **kwargs,
+    ) -> Dict[str, Any]:
+        self.calls.append(
+            {
+                "method": method,
+                "endpoint": endpoint,
+                "params": params or {},
+            }
+        )
+        index = len(self.calls) - 1
+        response = self._scripted_responses[index]
+        if isinstance(response, Exception):
+            raise response
+        return response
+    # [/DEF:request:Function]
+# [/DEF:_RecordingNetworkClient:Class]
+
+
+# [DEF:test_get_users_page_sends_lowercase_order_direction:Function]
+# @PURPOSE: Ensures adapter sends lowercase order_direction compatible with Superset rison schema.
+# @PRE: Adapter is initialized with recording network client.
+# @POST: First request query payload contains order_direction='asc' for asc sort.
+def test_get_users_page_sends_lowercase_order_direction():
+    client = _RecordingNetworkClient(
+        scripted_responses=[{"result": [{"username": "admin"}], "count": 1}]
+    )
+    adapter = SupersetAccountLookupAdapter(network_client=client, environment_id="ss-dev")
+
+    adapter.get_users_page(
+        search="admin",
+        page_index=0,
+        page_size=20,
+        sort_column="username",
+        sort_order="asc",
+    )
+
+    sent_query = json.loads(client.calls[0]["params"]["q"])
+    assert sent_query["order_direction"] == "asc"
+# [/DEF:test_get_users_page_sends_lowercase_order_direction:Function]
+
+
+# [DEF:test_get_users_page_preserves_primary_schema_error_over_fallback_auth_error:Function]
+# @PURPOSE: Ensures fallback auth error does not mask primary schema/query failure.
+# @PRE: Primary endpoint fails with SupersetAPIError and fallback fails with AuthenticationError.
+# @POST: Raised exception remains primary SupersetAPIError (non-auth) to preserve root cause.
+def test_get_users_page_preserves_primary_schema_error_over_fallback_auth_error():
+    client = _RecordingNetworkClient(
+        scripted_responses=[
+            SupersetAPIError("API Error 400: bad rison schema"),
+            AuthenticationError(),
+        ]
+    )
+    adapter = SupersetAccountLookupAdapter(network_client=client, environment_id="ss-dev")
+
+    with pytest.raises(SupersetAPIError) as exc_info:
+        adapter.get_users_page(sort_order="asc")
+
+    assert "API Error 400" in str(exc_info.value)
+    assert not isinstance(exc_info.value, AuthenticationError)
+# [/DEF:test_get_users_page_preserves_primary_schema_error_over_fallback_auth_error:Function]
+
+
+# [DEF:test_get_users_page_uses_fallback_endpoint_when_primary_fails:Function]
+# @PURPOSE: Verifies adapter retries second users endpoint and succeeds when fallback is healthy.
+# @PRE: Primary endpoint fails; fallback returns valid users payload.
+# @POST: Result status is success and both endpoints were attempted in order.
+def test_get_users_page_uses_fallback_endpoint_when_primary_fails():
+    client = _RecordingNetworkClient(
+        scripted_responses=[
+            SupersetAPIError("Primary endpoint failed"),
+            {"result": [{"username": "admin"}], "count": 1},
+        ]
+    )
+    adapter = SupersetAccountLookupAdapter(network_client=client, environment_id="ss-dev")
+
+    result = adapter.get_users_page()
+
+    assert result["status"] == "success"
+    assert [call["endpoint"] for call in client.calls] == ["/security/users/", "/security/users"]
+# [/DEF:test_get_users_page_uses_fallback_endpoint_when_primary_fails:Function]
+
+
+# [/DEF:backend.src.core.__tests__.test_superset_profile_lookup:Module]

backend/src/core/auth/repository.py

@@ -12,6 +12,7 @@
 from typing import Optional, List
 from sqlalchemy.orm import Session
 from ...models.auth import User, Role, Permission
+from ...models.profile import UserDashboardPreference
 from ..logger import belief_scope
 # [/SECTION]
 
@@ -109,6 +110,38 @@ class AuthRepository:
             ).first()
     # [/DEF:get_permission_by_resource_action:Function]
 
+    # [DEF:get_user_dashboard_preference:Function]
+    # @PURPOSE: Retrieves dashboard preference by owner user ID.
+    # @PRE:     user_id is a string.
+    # @POST:    Returns UserDashboardPreference if found, else None.
+    # @PARAM:   user_id (str) - Preference owner identifier.
+    # @RETURN:  Optional[UserDashboardPreference] - Found preference or None.
+    def get_user_dashboard_preference(self, user_id: str) -> Optional[UserDashboardPreference]:
+        with belief_scope("AuthRepository.get_user_dashboard_preference"):
+            return (
+                self.db.query(UserDashboardPreference)
+                .filter(UserDashboardPreference.user_id == user_id)
+                .first()
+            )
+    # [/DEF:get_user_dashboard_preference:Function]
+
+    # [DEF:save_user_dashboard_preference:Function]
+    # @PURPOSE: Persists dashboard preference entity and returns refreshed row.
+    # @PRE:     preference is a valid UserDashboardPreference entity.
+    # @POST:    Preference is committed and refreshed in database.
+    # @PARAM:   preference (UserDashboardPreference) - Preference entity to persist.
+    # @RETURN:  UserDashboardPreference - Persisted preference row.
+    def save_user_dashboard_preference(
+        self,
+        preference: UserDashboardPreference,
+    ) -> UserDashboardPreference:
+        with belief_scope("AuthRepository.save_user_dashboard_preference"):
+            self.db.add(preference)
+            self.db.commit()
+            self.db.refresh(preference)
+            return preference
+    # [/DEF:save_user_dashboard_preference:Function]
+
     # [DEF:list_permissions:Function]
     # @PURPOSE: Lists all available permissions.
     # @POST:    Returns a list of all Permission objects.

backend/src/core/database.py

@@ -20,6 +20,7 @@ from ..models import auth as _auth_models  # noqa: F401
 from ..models import config as _config_models  # noqa: F401
 from ..models import llm as _llm_models  # noqa: F401
 from ..models import assistant as _assistant_models  # noqa: F401
+from ..models import profile as _profile_models  # noqa: F401
 from .logger import belief_scope
 from .auth.config import auth_config
 import os

backend/src/core/superset_client.py

@@ -159,14 +159,37 @@ class SupersetClient:
 
             # Map fields to DashboardMetadata schema
             result = []
-            for dash in dashboards:
-                owners = self._extract_owner_labels(dash.get("owners"))
+            max_debug_samples = 12
+            for index, dash in enumerate(dashboards):
+                raw_owners = dash.get("owners")
+                raw_created_by = dash.get("created_by")
+                raw_changed_by = dash.get("changed_by")
+                raw_changed_by_name = dash.get("changed_by_name")
+
+                owners = self._extract_owner_labels(raw_owners)
                 # No per-dashboard detail requests here: keep list endpoint O(1).
                 if not owners:
                     owners = self._extract_owner_labels(
-                        [dash.get("created_by"), dash.get("changed_by")],
+                        [raw_created_by, raw_changed_by],
                     )
 
+                projected_created_by = self._extract_user_display(
+                    None,
+                    raw_created_by,
+                )
+                projected_modified_by = self._extract_user_display(
+                    raw_changed_by_name,
+                    raw_changed_by,
+                )
+
+                raw_owner_usernames: List[str] = []
+                if isinstance(raw_owners, list):
+                    for owner_payload in raw_owners:
+                        if isinstance(owner_payload, dict):
+                            owner_username = self._sanitize_user_text(owner_payload.get("username"))
+                            if owner_username:
+                                raw_owner_usernames.append(owner_username)
+
                 result.append({
                     "id": dash.get("id"),
                     "slug": dash.get("slug"),
@@ -174,16 +197,26 @@ class SupersetClient:
                     "url": dash.get("url"),
                     "last_modified": dash.get("changed_on_utc"),
                     "status": "published" if dash.get("published") else "draft",
-                    "created_by": self._extract_user_display(
-                        None,
-                        dash.get("created_by"),
-                    ),
-                    "modified_by": self._extract_user_display(
-                        dash.get("changed_by_name"),
-                        dash.get("changed_by"),
-                    ),
+                    "created_by": projected_created_by,
+                    "modified_by": projected_modified_by,
                     "owners": owners,
                 })
+
+                if index < max_debug_samples:
+                    app_logger.reflect(
+                        "[REFLECT] Dashboard actor projection sample "
+                        f"(env={getattr(self.env, 'id', None)}, dashboard_id={dash.get('id')}, "
+                        f"raw_owners={raw_owners!r}, raw_owner_usernames={raw_owner_usernames!r}, "
+                        f"raw_created_by={raw_created_by!r}, raw_changed_by={raw_changed_by!r}, "
+                        f"raw_changed_by_name={raw_changed_by_name!r}, projected_owners={owners!r}, "
+                        f"projected_created_by={projected_created_by!r}, projected_modified_by={projected_modified_by!r})"
+                    )
+
+            app_logger.reflect(
+                "[REFLECT] Dashboard actor projection summary "
+                f"(env={getattr(self.env, 'id', None)}, dashboards={len(result)}, "
+                f"sampled={min(len(result), max_debug_samples)})"
+            )
             return result
     # [/DEF:get_dashboards_summary:Function]
 

backend/src/core/superset_profile_lookup.py

@@ -0,0 +1,238 @@
+# [DEF:backend.src.core.superset_profile_lookup:Module]
+#
+# @TIER:      STANDARD
+# @SEMANTICS: superset, users, lookup, profile, pagination, normalization
+# @PURPOSE:   Provides environment-scoped Superset account lookup adapter with stable normalized output.
+# @LAYER:     Core
+# @RELATION:  DEPENDS_ON -> backend.src.core.utils.network.APIClient
+# @RELATION:  DEPENDS_ON -> backend.src.core.logger
+#
+# @INVARIANT: Adapter never leaks raw upstream payload shape to API consumers.
+
+# [SECTION: IMPORTS]
+import json
+from typing import Any, Dict, List, Optional
+
+from .logger import logger, belief_scope
+from .utils.network import APIClient, AuthenticationError, SupersetAPIError
+# [/SECTION]
+
+
+# [DEF:SupersetAccountLookupAdapter:Class]
+# @TIER: STANDARD
+# @PURPOSE: Lookup Superset users and normalize candidates for profile binding.
+class SupersetAccountLookupAdapter:
+    # [DEF:__init__:Function]
+    # @PURPOSE: Initializes lookup adapter with authenticated API client and environment context.
+    # @PRE: network_client supports request(method, endpoint, params=...).
+    # @POST: Adapter is ready to perform users lookup requests.
+    def __init__(self, network_client: APIClient, environment_id: str):
+        self.network_client = network_client
+        self.environment_id = str(environment_id or "")
+    # [/DEF:__init__:Function]
+
+    # [DEF:get_users_page:Function]
+    # @PURPOSE: Fetch one users page from Superset with passthrough search/sort parameters.
+    # @PRE: page_index >= 0 and page_size >= 1.
+    # @POST: Returns deterministic payload with normalized items and total count.
+    # @RETURN: Dict[str, Any]
+    def get_users_page(
+        self,
+        search: Optional[str] = None,
+        page_index: int = 0,
+        page_size: int = 20,
+        sort_column: str = "username",
+        sort_order: str = "desc",
+    ) -> Dict[str, Any]:
+        with belief_scope("SupersetAccountLookupAdapter.get_users_page"):
+            normalized_page_index = max(int(page_index), 0)
+            normalized_page_size = max(int(page_size), 1)
+
+            normalized_sort_column = str(sort_column or "username").strip().lower() or "username"
+            normalized_sort_order = str(sort_order or "desc").strip().lower()
+            if normalized_sort_order not in {"asc", "desc"}:
+                normalized_sort_order = "desc"
+
+            query: Dict[str, Any] = {
+                "page": normalized_page_index,
+                "page_size": normalized_page_size,
+                "order_column": normalized_sort_column,
+                "order_direction": normalized_sort_order,
+            }
+
+            normalized_search = str(search or "").strip()
+            if normalized_search:
+                query["filters"] = [{"col": "username", "opr": "ct", "value": normalized_search}]
+
+            logger.reason(
+                "[REASON] Lookup Superset users "
+                f"(env={self.environment_id}, page={normalized_page_index}, page_size={normalized_page_size})"
+            )
+            logger.reflect(
+                "[REFLECT] Prepared Superset users lookup query "
+                f"(env={self.environment_id}, order_column={normalized_sort_column}, "
+                f"normalized_sort_order={normalized_sort_order}, "
+                f"payload_order_direction={query.get('order_direction')})"
+            )
+
+            primary_error: Optional[Exception] = None
+            last_error: Optional[Exception] = None
+            for attempt_index, endpoint in enumerate(("/security/users/", "/security/users"), start=1):
+                try:
+                    logger.reason(
+                        "[REASON] Users lookup request attempt "
+                        f"(env={self.environment_id}, attempt={attempt_index}, endpoint={endpoint})"
+                    )
+                    response = self.network_client.request(
+                        method="GET",
+                        endpoint=endpoint,
+                        params={"q": json.dumps(query)},
+                    )
+                    logger.reflect(
+                        "[REFLECT] Users lookup endpoint succeeded "
+                        f"(env={self.environment_id}, attempt={attempt_index}, endpoint={endpoint})"
+                    )
+                    return self._normalize_lookup_payload(
+                        response=response,
+                        page_index=normalized_page_index,
+                        page_size=normalized_page_size,
+                    )
+                except Exception as exc:
+                    if primary_error is None:
+                        primary_error = exc
+                    last_error = exc
+                    cause = getattr(exc, "__cause__", None)
+                    cause_response = getattr(cause, "response", None)
+                    status_code = getattr(cause_response, "status_code", None)
+                    logger.explore(
+                        "[EXPLORE] Users lookup endpoint failed "
+                        f"(env={self.environment_id}, attempt={attempt_index}, endpoint={endpoint}, "
+                        f"error_type={type(exc).__name__}, status_code={status_code}, "
+                        f"payload_order_direction={query.get('order_direction')}): {exc}"
+                    )
+
+            if last_error is not None:
+                selected_error: Exception = last_error
+                if (
+                    primary_error is not None
+                    and primary_error is not last_error
+                    and isinstance(last_error, AuthenticationError)
+                    and not isinstance(primary_error, AuthenticationError)
+                ):
+                    selected_error = primary_error
+                    logger.reflect(
+                        "[REFLECT] Preserving primary lookup failure over fallback auth error "
+                        f"(env={self.environment_id}, primary_error_type={type(primary_error).__name__}, "
+                        f"fallback_error_type={type(last_error).__name__})"
+                    )
+
+                logger.explore(
+                    "[EXPLORE] All Superset users lookup endpoints failed "
+                    f"(env={self.environment_id}, payload_order_direction={query.get('order_direction')}, "
+                    f"selected_error_type={type(selected_error).__name__})"
+                )
+                raise selected_error
+            raise SupersetAPIError("Superset users lookup failed without explicit error")
+    # [/DEF:get_users_page:Function]
+
+    # [DEF:_normalize_lookup_payload:Function]
+    # @PURPOSE: Convert Superset users response variants into stable candidates payload.
+    # @PRE: response can be dict/list in any supported upstream shape.
+    # @POST: Output contains canonical keys: status, environment_id, page_index, page_size, total, items.
+    # @RETURN: Dict[str, Any]
+    def _normalize_lookup_payload(
+        self,
+        response: Any,
+        page_index: int,
+        page_size: int,
+    ) -> Dict[str, Any]:
+        with belief_scope("SupersetAccountLookupAdapter._normalize_lookup_payload"):
+            payload = response
+            if isinstance(payload, dict) and isinstance(payload.get("result"), dict):
+                payload = payload.get("result")
+
+            raw_items: List[Any] = []
+            total = 0
+
+            if isinstance(payload, dict):
+                if isinstance(payload.get("result"), list):
+                    raw_items = payload.get("result") or []
+                    total = int(payload.get("count", len(raw_items)) or 0)
+                elif isinstance(payload.get("users"), list):
+                    raw_items = payload.get("users") or []
+                    total = int(payload.get("total", len(raw_items)) or 0)
+                elif isinstance(payload.get("items"), list):
+                    raw_items = payload.get("items") or []
+                    total = int(payload.get("total", len(raw_items)) or 0)
+            elif isinstance(payload, list):
+                raw_items = payload
+                total = len(raw_items)
+
+            normalized_items: List[Dict[str, Any]] = []
+            seen_usernames = set()
+
+            for raw_user in raw_items:
+                candidate = self.normalize_user_payload(raw_user)
+                username_key = str(candidate.get("username") or "").strip().lower()
+                if not username_key:
+                    continue
+                if username_key in seen_usernames:
+                    continue
+                seen_usernames.add(username_key)
+                normalized_items.append(candidate)
+
+            logger.reflect(
+                "[REFLECT] Normalized lookup payload "
+                f"(env={self.environment_id}, items={len(normalized_items)}, total={max(total, len(normalized_items))})"
+            )
+
+            return {
+                "status": "success",
+                "environment_id": self.environment_id,
+                "page_index": max(int(page_index), 0),
+                "page_size": max(int(page_size), 1),
+                "total": max(int(total), len(normalized_items)),
+                "items": normalized_items,
+            }
+    # [/DEF:_normalize_lookup_payload:Function]
+
+    # [DEF:normalize_user_payload:Function]
+    # @PURPOSE: Project raw Superset user object to canonical candidate shape.
+    # @PRE: raw_user may have heterogenous key names between Superset versions.
+    # @POST: Returns normalized candidate keys (environment_id, username, display_name, email, is_active).
+    # @RETURN: Dict[str, Any]
+    def normalize_user_payload(self, raw_user: Any) -> Dict[str, Any]:
+        if not isinstance(raw_user, dict):
+            raw_user = {}
+
+        username = str(
+            raw_user.get("username")
+            or raw_user.get("userName")
+            or raw_user.get("name")
+            or ""
+        ).strip()
+
+        full_name = str(raw_user.get("full_name") or "").strip()
+        first_name = str(raw_user.get("first_name") or "").strip()
+        last_name = str(raw_user.get("last_name") or "").strip()
+        display_name = full_name or " ".join(
+            part for part in [first_name, last_name] if part
+        ).strip()
+        if not display_name:
+            display_name = username or None
+
+        email = str(raw_user.get("email") or "").strip() or None
+        is_active_raw = raw_user.get("is_active")
+        is_active = bool(is_active_raw) if is_active_raw is not None else None
+
+        return {
+            "environment_id": self.environment_id,
+            "username": username,
+            "display_name": display_name,
+            "email": email,
+            "is_active": is_active,
+        }
+    # [/DEF:normalize_user_payload:Function]
+# [/DEF:SupersetAccountLookupAdapter:Class]
+
+# [/DEF:backend.src.core.superset_profile_lookup:Module]

backend/src/models/profile.py

@@ -0,0 +1,46 @@
+# [DEF:backend.src.models.profile:Module]
+#
+# @TIER:      STANDARD
+# @SEMANTICS: profile, preferences, persistence, user, dashboard-filter, sqlalchemy
+# @PURPOSE:   Defines persistent per-user dashboard filter preferences.
+# @LAYER:     Domain
+# @RELATION:  DEPENDS_ON -> backend.src.models.auth
+# @RELATION:  INHERITS_FROM -> backend.src.models.mapping.Base
+#
+# @INVARIANT: Exactly one preference row exists per user_id.
+
+# [SECTION: IMPORTS]
+import uuid
+from datetime import datetime
+from sqlalchemy import Column, String, Boolean, DateTime, ForeignKey
+from sqlalchemy.orm import relationship
+from .mapping import Base
+# [/SECTION]
+
+
+# [DEF:UserDashboardPreference:Class]
+# @TIER: STANDARD
+# @PURPOSE: Stores Superset username binding and default "my dashboards" toggle for one authenticated user.
+class UserDashboardPreference(Base):
+    __tablename__ = "user_dashboard_preferences"
+
+    id = Column(String, primary_key=True, default=lambda: str(uuid.uuid4()))
+    user_id = Column(String, ForeignKey("users.id"), nullable=False, unique=True, index=True)
+
+    superset_username = Column(String, nullable=True)
+    superset_username_normalized = Column(String, nullable=True, index=True)
+
+    show_only_my_dashboards = Column(Boolean, nullable=False, default=False)
+
+    created_at = Column(DateTime, nullable=False, default=datetime.utcnow)
+    updated_at = Column(
+        DateTime,
+        nullable=False,
+        default=datetime.utcnow,
+        onupdate=datetime.utcnow,
+    )
+
+    user = relationship("User")
+# [/DEF:UserDashboardPreference:Class]
+
+# [/DEF:backend.src.models.profile:Module]

backend/src/schemas/profile.py

@@ -0,0 +1,98 @@
+# [DEF:backend.src.schemas.profile:Module]
+#
+# @TIER:      STANDARD
+# @SEMANTICS: profile, schemas, pydantic, preferences, superset, lookup
+# @PURPOSE:   Defines API schemas for profile preference persistence and Superset account lookup flows.
+# @LAYER:     API
+# @RELATION:  DEPENDS_ON -> pydantic
+#
+# @INVARIANT: Schema shapes stay stable for profile UI states and dashboards filter metadata.
+
+# [SECTION: IMPORTS]
+from datetime import datetime
+from typing import List, Literal, Optional
+from pydantic import BaseModel, Field
+# [/SECTION]
+
+
+# [DEF:ProfilePreference:Class]
+# @TIER: STANDARD
+# @PURPOSE: Represents persisted profile preference for a single authenticated user.
+class ProfilePreference(BaseModel):
+    user_id: str
+    superset_username: Optional[str] = None
+    superset_username_normalized: Optional[str] = None
+    show_only_my_dashboards: bool = False
+    created_at: datetime
+    updated_at: datetime
+
+    class Config:
+        from_attributes = True
+# [/DEF:ProfilePreference:Class]
+
+
+# [DEF:ProfilePreferenceUpdateRequest:Class]
+# @TIER: STANDARD
+# @PURPOSE: Request payload for updating current user's dashboard filter preference.
+class ProfilePreferenceUpdateRequest(BaseModel):
+    superset_username: Optional[str] = Field(
+        default=None,
+        description="Apache Superset username bound to current user profile.",
+    )
+    show_only_my_dashboards: bool = Field(
+        default=False,
+        description='When true, "/dashboards" can auto-apply profile filter in main context.',
+    )
+# [/DEF:ProfilePreferenceUpdateRequest:Class]
+
+
+# [DEF:ProfilePreferenceResponse:Class]
+# @TIER: STANDARD
+# @PURPOSE: Response envelope for profile preference read/update endpoints.
+class ProfilePreferenceResponse(BaseModel):
+    status: Literal["success", "error"] = "success"
+    message: Optional[str] = None
+    validation_errors: List[str] = Field(default_factory=list)
+    preference: ProfilePreference
+# [/DEF:ProfilePreferenceResponse:Class]
+
+
+# [DEF:SupersetAccountLookupRequest:Class]
+# @TIER: STANDARD
+# @PURPOSE: Query contract for Superset account lookup by selected environment.
+class SupersetAccountLookupRequest(BaseModel):
+    environment_id: str
+    search: Optional[str] = None
+    page_index: int = Field(default=0, ge=0)
+    page_size: int = Field(default=20, ge=1, le=100)
+    sort_column: str = Field(default="username")
+    sort_order: str = Field(default="desc")
+# [/DEF:SupersetAccountLookupRequest:Class]
+
+
+# [DEF:SupersetAccountCandidate:Class]
+# @TIER: STANDARD
+# @PURPOSE: Canonical account candidate projected from Superset users payload.
+class SupersetAccountCandidate(BaseModel):
+    environment_id: str
+    username: str
+    display_name: Optional[str] = None
+    email: Optional[str] = None
+    is_active: Optional[bool] = None
+# [/DEF:SupersetAccountCandidate:Class]
+
+
+# [DEF:SupersetAccountLookupResponse:Class]
+# @TIER: STANDARD
+# @PURPOSE: Response envelope for Superset account lookup (success or degraded mode).
+class SupersetAccountLookupResponse(BaseModel):
+    status: Literal["success", "degraded"]
+    environment_id: str
+    page_index: int = Field(ge=0)
+    page_size: int = Field(ge=1, le=100)
+    total: int = Field(ge=0)
+    warning: Optional[str] = None
+    items: List[SupersetAccountCandidate] = Field(default_factory=list)
+# [/DEF:SupersetAccountLookupResponse:Class]
+
+# [/DEF:backend.src.schemas.profile:Module]

backend/src/services/profile_service.py

@@ -0,0 +1,353 @@
+# [DEF:backend.src.services.profile_service:Module]
+#
+# @TIER:      CRITICAL
+# @SEMANTICS: profile, service, validation, ownership, filtering, superset, preferences
+# @PURPOSE:   Orchestrates profile preference persistence, Superset account lookup, and deterministic actor matching.
+# @LAYER:     Domain
+# @RELATION:  DEPENDS_ON -> backend.src.models.profile
+# @RELATION:  DEPENDS_ON -> backend.src.schemas.profile
+# @RELATION:  DEPENDS_ON -> backend.src.core.superset_client
+# @RELATION:  DEPENDS_ON -> backend.src.core.auth.repository
+# @RELATION:  DEPENDS_ON -> backend.src.models.auth
+# @RELATION:  DEPENDS_ON -> sqlalchemy.orm.Session
+#
+# @INVARIANT: Preference mutations are always scoped to authenticated user identity.
+# @INVARIANT: Username normalization is trim+lower and shared by save and matching paths.
+#
+# @TEST_CONTRACT: ProfilePreferenceUpdateRequest -> ProfilePreferenceResponse
+# @TEST_FIXTURE: valid_profile_update -> {"user_id":"u-1","superset_username":"John_Doe","show_only_my_dashboards":true}
+# @TEST_EDGE: enable_without_username -> toggle=true with empty username returns validation error
+# @TEST_EDGE: cross_user_mutation -> attempt to update another user preference returns forbidden
+# @TEST_EDGE: lookup_env_not_found -> unknown environment_id returns not found
+# @TEST_INVARIANT: normalization_consistency -> VERIFIED_BY: [valid_profile_update, enable_without_username]
+
+# [SECTION: IMPORTS]
+from datetime import datetime
+from typing import Any, Iterable, List, Optional, Sequence
+from sqlalchemy.orm import Session
+
+from ..core.auth.repository import AuthRepository
+from ..core.logger import logger, belief_scope
+from ..core.superset_client import SupersetClient
+from ..core.superset_profile_lookup import SupersetAccountLookupAdapter
+from ..models.auth import User
+from ..models.profile import UserDashboardPreference
+from ..schemas.profile import (
+    ProfilePreference,
+    ProfilePreferenceResponse,
+    ProfilePreferenceUpdateRequest,
+    SupersetAccountLookupRequest,
+    SupersetAccountLookupResponse,
+    SupersetAccountCandidate,
+)
+# [/SECTION]
+
+
+# [DEF:ProfileValidationError:Class]
+# @TIER: STANDARD
+# @PURPOSE: Domain validation error for profile preference update requests.
+class ProfileValidationError(Exception):
+    def __init__(self, errors: Sequence[str]):
+        self.errors = list(errors)
+        super().__init__("Profile preference validation failed")
+# [/DEF:ProfileValidationError:Class]
+
+
+# [DEF:EnvironmentNotFoundError:Class]
+# @TIER: STANDARD
+# @PURPOSE: Raised when environment_id from lookup request is unknown in app configuration.
+class EnvironmentNotFoundError(Exception):
+    pass
+# [/DEF:EnvironmentNotFoundError:Class]
+
+
+# [DEF:ProfileAuthorizationError:Class]
+# @TIER: STANDARD
+# @PURPOSE: Raised when caller attempts cross-user preference mutation.
+class ProfileAuthorizationError(Exception):
+    pass
+# [/DEF:ProfileAuthorizationError:Class]
+
+
+# [DEF:ProfileService:Class]
+# @TIER: CRITICAL
+# @PURPOSE: Implements profile preference read/update flow and Superset account lookup degradation strategy.
+class ProfileService:
+    # [DEF:__init__:Function]
+    # @PURPOSE: Initialize service with DB session and config manager.
+    # @PRE: db session is active and config_manager supports get_environments().
+    # @POST: Service is ready for preference persistence and lookup operations.
+    def __init__(self, db: Session, config_manager: Any):
+        self.db = db
+        self.config_manager = config_manager
+        self.auth_repository = AuthRepository(db)
+    # [/DEF:__init__:Function]
+
+    # [DEF:get_my_preference:Function]
+    # @PURPOSE: Return current user's persisted preference or default non-configured view.
+    # @PRE: current_user is authenticated.
+    # @POST: Returned payload belongs to current_user only.
+    def get_my_preference(self, current_user: User) -> ProfilePreferenceResponse:
+        with belief_scope("ProfileService.get_my_preference", f"user_id={current_user.id}"):
+            logger.reflect("[REFLECT] Loading current user's dashboard preference")
+            preference = self._get_preference_row(current_user.id)
+            if preference is None:
+                return ProfilePreferenceResponse(
+                    status="success",
+                    message="Preference not configured yet",
+                    preference=self._build_default_preference(current_user.id),
+                )
+            return ProfilePreferenceResponse(
+                status="success",
+                message="Preference loaded",
+                preference=ProfilePreference.model_validate(preference, from_attributes=True),
+            )
+    # [/DEF:get_my_preference:Function]
+
+    # [DEF:update_my_preference:Function]
+    # @PURPOSE: Validate and persist current user's profile preference in self-scoped mode.
+    # @PRE: current_user is authenticated and payload is provided.
+    # @POST: Preference row for current_user is created/updated when validation passes.
+    def update_my_preference(
+        self,
+        current_user: User,
+        payload: ProfilePreferenceUpdateRequest,
+        target_user_id: Optional[str] = None,
+    ) -> ProfilePreferenceResponse:
+        with belief_scope("ProfileService.update_my_preference", f"user_id={current_user.id}"):
+            logger.reason("[REASON] Evaluating self-scope guard before preference mutation")
+            requested_user_id = str(target_user_id or current_user.id)
+            if requested_user_id != str(current_user.id):
+                logger.explore("[EXPLORE] Cross-user mutation attempt blocked")
+                raise ProfileAuthorizationError("Cross-user preference mutation is forbidden")
+
+            validation_errors = self._validate_update_payload(payload)
+            if validation_errors:
+                logger.reflect("[REFLECT] Validation failed; mutation is denied")
+                raise ProfileValidationError(validation_errors)
+
+            normalized_username = self._normalize_username(payload.superset_username)
+            raw_username = self._sanitize_username(payload.superset_username)
+
+            preference = self._get_or_create_preference_row(current_user.id)
+            preference.superset_username = raw_username
+            preference.superset_username_normalized = normalized_username
+            preference.show_only_my_dashboards = bool(payload.show_only_my_dashboards)
+            preference.updated_at = datetime.utcnow()
+
+            self.auth_repository.save_user_dashboard_preference(preference)
+
+            logger.reason("[REASON] Preference persisted successfully")
+            return ProfilePreferenceResponse(
+                status="success",
+                message="Preference saved",
+                preference=ProfilePreference.model_validate(preference, from_attributes=True),
+            )
+    # [/DEF:update_my_preference:Function]
+
+    # [DEF:lookup_superset_accounts:Function]
+    # @PURPOSE: Query Superset users in selected environment and project canonical account candidates.
+    # @PRE: current_user is authenticated and environment_id exists.
+    # @POST: Returns success payload or degraded payload with warning while preserving manual fallback.
+    def lookup_superset_accounts(
+        self,
+        current_user: User,
+        request: SupersetAccountLookupRequest,
+    ) -> SupersetAccountLookupResponse:
+        with belief_scope(
+            "ProfileService.lookup_superset_accounts",
+            f"user_id={current_user.id}, environment_id={request.environment_id}",
+        ):
+            environment = self._resolve_environment(request.environment_id)
+            if environment is None:
+                logger.explore("[EXPLORE] Lookup aborted: environment not found")
+                raise EnvironmentNotFoundError(f"Environment '{request.environment_id}' not found")
+
+            sort_column = str(request.sort_column or "username").strip().lower()
+            sort_order = str(request.sort_order or "desc").strip().lower()
+            allowed_columns = {"username", "first_name", "last_name", "email"}
+            if sort_column not in allowed_columns:
+                sort_column = "username"
+            if sort_order not in {"asc", "desc"}:
+                sort_order = "desc"
+
+            logger.reflect(
+                "[REFLECT] Normalized lookup request "
+                f"(env={request.environment_id}, sort_column={sort_column}, sort_order={sort_order}, "
+                f"page_index={request.page_index}, page_size={request.page_size}, "
+                f"search={(request.search or '').strip()!r})"
+            )
+
+            try:
+                logger.reason("[REASON] Performing Superset account lookup")
+                superset_client = SupersetClient(environment)
+                adapter = SupersetAccountLookupAdapter(
+                    network_client=superset_client.network,
+                    environment_id=request.environment_id,
+                )
+                lookup_result = adapter.get_users_page(
+                    search=request.search,
+                    page_index=request.page_index,
+                    page_size=request.page_size,
+                    sort_column=sort_column,
+                    sort_order=sort_order,
+                )
+                items = [
+                    SupersetAccountCandidate.model_validate(item)
+                    for item in lookup_result.get("items", [])
+                ]
+                return SupersetAccountLookupResponse(
+                    status="success",
+                    environment_id=request.environment_id,
+                    page_index=request.page_index,
+                    page_size=request.page_size,
+                    total=max(int(lookup_result.get("total", len(items))), 0),
+                    warning=None,
+                    items=items,
+                )
+            except Exception as exc:
+                logger.explore(f"[EXPLORE] Lookup degraded due to upstream error: {exc}")
+                return SupersetAccountLookupResponse(
+                    status="degraded",
+                    environment_id=request.environment_id,
+                    page_index=request.page_index,
+                    page_size=request.page_size,
+                    total=0,
+                    warning=(
+                        "Cannot load Superset accounts for this environment right now. "
+                        "You can enter username manually."
+                    ),
+                    items=[],
+                )
+    # [/DEF:lookup_superset_accounts:Function]
+
+    # [DEF:matches_dashboard_actor:Function]
+    # @PURPOSE: Apply trim+case-insensitive actor match across owners OR modified_by.
+    # @PRE: bound_username can be empty; owners may contain mixed payload.
+    # @POST: Returns True when normalized username matches owners or modified_by.
+    def matches_dashboard_actor(
+        self,
+        bound_username: Optional[str],
+        owners: Optional[Iterable[Any]],
+        modified_by: Optional[str],
+    ) -> bool:
+        normalized_actor = self._normalize_username(bound_username)
+        if not normalized_actor:
+            return False
+
+        owner_tokens = self._normalize_owner_tokens(owners)
+        modified_token = self._normalize_username(modified_by)
+
+        if normalized_actor in owner_tokens:
+            return True
+        if modified_token and normalized_actor == modified_token:
+            return True
+        return False
+    # [/DEF:matches_dashboard_actor:Function]
+
+    # [DEF:_resolve_environment:Function]
+    # @PURPOSE: Resolve environment model from configured environments by id.
+    # @PRE: environment_id is provided.
+    # @POST: Returns environment object when found else None.
+    def _resolve_environment(self, environment_id: str):
+        environments = self.config_manager.get_environments()
+        for env in environments:
+            if str(getattr(env, "id", "")) == str(environment_id):
+                return env
+        return None
+    # [/DEF:_resolve_environment:Function]
+
+    # [DEF:_get_preference_row:Function]
+    # @PURPOSE: Return persisted preference row for user or None.
+    # @PRE: user_id is provided.
+    # @POST: Returns matching row or None.
+    def _get_preference_row(self, user_id: str) -> Optional[UserDashboardPreference]:
+        return self.auth_repository.get_user_dashboard_preference(str(user_id))
+    # [/DEF:_get_preference_row:Function]
+
+    # [DEF:_get_or_create_preference_row:Function]
+    # @PURPOSE: Return existing preference row or create new unsaved row.
+    # @PRE: user_id is provided.
+    # @POST: Returned row always contains user_id.
+    def _get_or_create_preference_row(self, user_id: str) -> UserDashboardPreference:
+        existing = self._get_preference_row(user_id)
+        if existing is not None:
+            return existing
+        return UserDashboardPreference(user_id=str(user_id))
+    # [/DEF:_get_or_create_preference_row:Function]
+
+    # [DEF:_build_default_preference:Function]
+    # @PURPOSE: Build non-persisted default preference DTO for unconfigured users.
+    # @PRE: user_id is provided.
+    # @POST: Returns ProfilePreference with disabled toggle and empty username.
+    def _build_default_preference(self, user_id: str) -> ProfilePreference:
+        now = datetime.utcnow()
+        return ProfilePreference(
+            user_id=str(user_id),
+            superset_username=None,
+            superset_username_normalized=None,
+            show_only_my_dashboards=False,
+            created_at=now,
+            updated_at=now,
+        )
+    # [/DEF:_build_default_preference:Function]
+
+    # [DEF:_validate_update_payload:Function]
+    # @PURPOSE: Validate username/toggle constraints for preference mutation.
+    # @PRE: payload is provided.
+    # @POST: Returns validation errors list; empty list means valid.
+    def _validate_update_payload(
+        self,
+        payload: ProfilePreferenceUpdateRequest,
+    ) -> List[str]:
+        errors: List[str] = []
+        sanitized_username = self._sanitize_username(payload.superset_username)
+
+        if sanitized_username and any(ch.isspace() for ch in sanitized_username):
+            errors.append(
+                "Username should not contain spaces. Please enter a valid Apache Superset username."
+            )
+        if payload.show_only_my_dashboards and not sanitized_username:
+            errors.append("Superset username is required when default filter is enabled.")
+        return errors
+    # [/DEF:_validate_update_payload:Function]
+
+    # [DEF:_sanitize_username:Function]
+    # @PURPOSE: Normalize raw username into trimmed form or None for empty input.
+    # @PRE: value can be empty or None.
+    # @POST: Returns trimmed username or None.
+    def _sanitize_username(self, value: Optional[str]) -> Optional[str]:
+        normalized = str(value or "").strip()
+        if not normalized:
+            return None
+        return normalized
+    # [/DEF:_sanitize_username:Function]
+
+    # [DEF:_normalize_username:Function]
+    # @PURPOSE: Apply deterministic trim+lower normalization for actor matching.
+    # @PRE: value can be empty or None.
+    # @POST: Returns lowercase normalized token or None.
+    def _normalize_username(self, value: Optional[str]) -> Optional[str]:
+        sanitized = self._sanitize_username(value)
+        if sanitized is None:
+            return None
+        return sanitized.lower()
+    # [/DEF:_normalize_username:Function]
+
+    # [DEF:_normalize_owner_tokens:Function]
+    # @PURPOSE: Normalize owners payload into deduplicated lower-cased tokens.
+    # @PRE: owners can be iterable of scalars or dict-like values.
+    # @POST: Returns list of unique normalized owner tokens.
+    def _normalize_owner_tokens(self, owners: Optional[Iterable[Any]]) -> List[str]:
+        if owners is None:
+            return []
+        normalized: List[str] = []
+        for owner in owners:
+            token = self._normalize_username(str(owner or ""))
+            if token and token not in normalized:
+                normalized.append(token)
+        return normalized
+    # [/DEF:_normalize_owner_tokens:Function]
+# [/DEF:ProfileService:Class]
+
+# [/DEF:backend.src.services.profile_service:Module]

backend/tests/fixtures/profile/fixtures_profile_filter.json

@@ -0,0 +1,26 @@
+{
+  "valid_profile_update": {
+    "user_id": "u-1",
+    "superset_username": "John_Doe",
+    "show_only_my_dashboards": true
+  },
+  "get_my_preference_ok": {
+    "auth_user": "u-1",
+    "expected_status": 200
+  },
+  "profile_filter_applied": {
+    "apply_profile_default": true,
+    "override_show_all": false
+  },
+  "enable_without_username": {
+    "superset_username": "",
+    "show_only_my_dashboards": true
+  },
+  "cross_user_mutation": {
+    "auth_user": "u-1",
+    "target_user": "u-2"
+  },
+  "lookup_env_not_found": {
+    "environment_id": "missing-env"
+  }
+}