Merge branch agent/me-bootstrap

modules/database/services/bootstrap_service.py

@@ -0,0 +1,370 @@
+import os
+from datetime import datetime, timezone
+from typing import Any, Callable, Dict, List, Optional
+
+from modules.logger_tool import initialise_logger
+from modules.database.supabase.utils.client import SupabaseServiceRoleClient
+import modules.database.tools.neo4j_driver_tools as driver_tools
+
+logger = initialise_logger(__name__, os.getenv("LOG_LEVEL"), os.getenv("LOG_PATH"), "default", True)
+
+MembershipRow = Dict[str, Any]
+GraphProbe = Callable[..., Dict[str, Any]]
+
+
+ROLE_PERMISSIONS: Dict[str, Dict[str, bool]] = {
+    "school_admin": {
+        "can_manage_school": True,
+        "can_manage_calendar": True,
+        "can_manage_timetable": True,
+        "can_invite_staff": True,
+        "can_manage_classes": True,
+        "can_view_student_data": True,
+    },
+    "department_head": {
+        "can_manage_school": False,
+        "can_manage_calendar": False,
+        "can_manage_timetable": True,
+        "can_invite_staff": False,
+        "can_manage_classes": True,
+        "can_view_student_data": True,
+    },
+    "teacher": {
+        "can_manage_school": False,
+        "can_manage_calendar": False,
+        "can_manage_timetable": False,
+        "can_invite_staff": False,
+        "can_manage_classes": True,
+        "can_view_student_data": False,
+    },
+    "student": {
+        "can_manage_school": False,
+        "can_manage_calendar": False,
+        "can_manage_timetable": False,
+        "can_invite_staff": False,
+        "can_manage_classes": False,
+        "can_view_student_data": False,
+    },
+}
+
+
+BASE_PERMISSIONS: Dict[str, bool] = {
+    "platform_admin": False,
+    "platform_super_admin": False,
+    "can_create_school": True,
+    "can_manage_school": False,
+    "can_manage_calendar": False,
+    "can_manage_timetable": False,
+    "can_invite_staff": False,
+    "can_manage_classes": False,
+    "can_view_student_data": False,
+    "can_use_canvas": True,
+}
+
+
+def utc_now_iso() -> str:
+    return datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")
+
+
+def _safe_list(result: Any) -> List[Dict[str, Any]]:
+    data = getattr(result, "data", None)
+    if not data:
+        return []
+    if isinstance(data, list):
+        return data
+    return [data]
+
+
+def _safe_one(result: Any) -> Optional[Dict[str, Any]]:
+    data = getattr(result, "data", None)
+    return data if isinstance(data, dict) else None
+
+
+def default_graph_probe(user_id: str, user_email: str, active_institute: Optional[Dict[str, Any]], institute: Optional[Dict[str, Any]]) -> Dict[str, Any]:
+    """Best-effort derived graph status; never authoritative for bootstrap state."""
+    user_db = f"cc.users.teacher.{user_id.replace('-', '')}" if user_id else None
+    neo4j_uuid = (institute or {}).get("neo4j_uuid_string")
+    institute_db = f"cc.institutes.{neo4j_uuid}" if neo4j_uuid else None
+    status = {
+        "available": False,
+        "user_db": user_db,
+        "institute_db": institute_db,
+        "projection_state": "unknown",
+        "needs_rebuild": True,
+        "last_checked_at": utc_now_iso(),
+        "error_code": None,
+    }
+    try:
+        if not user_db and not institute_db:
+            status["projection_state"] = "missing"
+            return status
+        db_to_check = institute_db or user_db
+        with driver_tools.get_session(database=db_to_check) as session:
+            session.run("RETURN 1 AS ok").single()
+        status.update({"available": True, "projection_state": "ready", "needs_rebuild": False})
+        return status
+    except Exception:
+        status.update({"available": False, "projection_state": "error", "needs_rebuild": True, "error_code": "neo4j_unavailable"})
+        return status
+
+
+class BootstrapService:
+    def __init__(self, supabase: Any, graph_probe: Optional[GraphProbe] = None):
+        self.supabase = supabase
+        self.graph_probe = graph_probe or default_graph_probe
+
+    def build(self, credentials: Dict[str, Any]) -> Dict[str, Any]:
+        user_id = credentials.get("sub") or ""
+        user_email = credentials.get("email") or ""
+        if not user_id:
+            raise ValueError("Missing authenticated user id")
+
+        profile = self._get_profile(user_id, user_email)
+        admin_profile = self._get_admin_profile(user_id)
+        memberships = self._get_memberships(user_id)
+        active = self._select_active_institute(profile, memberships)
+        active_institute = self._membership_institute(active, memberships)
+
+        permissions = self._permissions(active, bool(admin_profile), bool((admin_profile or {}).get("is_super_admin")))
+        profile_user_type = "platform_admin" if admin_profile else (profile.get("user_type") or active.get("role") or "unknown")
+        school_status = self._school_status(active, memberships, admin_profile)
+
+        calendar_status = self._calendar_status(active.get("institute_id"))
+        timetable_status = self._timetable_status(user_id, active.get("institute_id"))
+        graph_status = self._graph_status(user_id, user_email, active, active_institute)
+        onboarding = self._onboarding(school_status, permissions, calendar_status, timetable_status, graph_status)
+
+        return {
+            "profile": {
+                "id": user_id,
+                "email": profile.get("email") or user_email,
+                "display_name": profile.get("display_name") or profile.get("full_name") or profile.get("username") or user_email,
+                "user_type": profile_user_type,
+                "school_id": profile.get("school_id"),
+            },
+            "memberships": [self._serialize_membership(row) for row in memberships],
+            "active_institute": {
+                "id": active.get("institute_id"),
+                "source": active.get("source", "none"),
+                "membership_role": active.get("role"),
+            },
+            "permissions": permissions,
+            "school_status": school_status,
+            "onboarding": onboarding,
+            "calendar_status": calendar_status,
+            "timetable_status": timetable_status,
+            "graph_status": graph_status,
+        }
+
+    def _get_profile(self, user_id: str, user_email: str) -> Dict[str, Any]:
+        try:
+            result = (
+                self.supabase.table("profiles")
+                .select("id,email,user_type,username,full_name,display_name,user_db_name,school_db_name")
+                .eq("id", user_id)
+                .single()
+                .execute()
+            )
+            profile = _safe_one(result) or {}
+        except Exception as exc:
+            logger.warning("Bootstrap profile lookup failed for authenticated user: %s", exc)
+            profile = {}
+        profile.setdefault("id", user_id)
+        profile.setdefault("email", user_email)
+        return profile
+
+    def _get_admin_profile(self, user_id: str) -> Optional[Dict[str, Any]]:
+        try:
+            result = (
+                self.supabase.table("admin_profiles")
+                .select("id,admin_role,is_super_admin")
+                .eq("id", user_id)
+                .single()
+                .execute()
+            )
+            return _safe_one(result)
+        except Exception:
+            return None
+
+    def _get_memberships(self, user_id: str) -> List[MembershipRow]:
+        try:
+            member_rows = _safe_list(
+                self.supabase.table("institute_memberships")
+                .select("profile_id,institute_id,role")
+                .eq("profile_id", user_id)
+                .execute()
+            )
+        except Exception as exc:
+            logger.warning("Bootstrap membership lookup failed: %s", exc)
+            return []
+
+        institute_ids = [str(row.get("institute_id")) for row in member_rows if row.get("institute_id")]
+        institutes_by_id: Dict[str, Dict[str, Any]] = {}
+        if institute_ids:
+            try:
+                inst_rows = _safe_list(
+                    self.supabase.table("institutes")
+                    .select("id,name,urn,website,address,metadata,status,neo4j_uuid_string")
+                    .in_("id", institute_ids)
+                    .execute()
+                )
+                institutes_by_id = {str(row.get("id")): row for row in inst_rows}
+            except Exception as exc:
+                logger.warning("Bootstrap institute lookup failed: %s", exc)
+
+        rows: List[MembershipRow] = []
+        for member in member_rows:
+            inst = institutes_by_id.get(str(member.get("institute_id")), {})
+            if inst and inst.get("status") not in (None, "active"):
+                continue
+            rows.append({**member, "institute": inst, "status": member.get("status") or "active", "is_active": True})
+        return rows
+
+    def _select_active_institute(self, profile: Dict[str, Any], memberships: List[MembershipRow]) -> MembershipRow:
+        active_memberships = [m for m in memberships if m.get("is_active")]
+        profile_school_id = profile.get("school_id")
+        if profile_school_id:
+            for member in active_memberships:
+                if str(member.get("institute_id")) == str(profile_school_id):
+                    return {**member, "source": "profile.school_id"}
+        if len(active_memberships) == 1:
+            return {**active_memberships[0], "source": "single_membership"}
+        return {"institute_id": None, "role": None, "source": "none"}
+
+    def _membership_institute(self, active: MembershipRow, memberships: List[MembershipRow]) -> Optional[Dict[str, Any]]:
+        active_id = active.get("institute_id")
+        if not active_id:
+            return None
+        for member in memberships:
+            if str(member.get("institute_id")) == str(active_id):
+                return member.get("institute") or None
+        return None
+
+    def _serialize_membership(self, row: MembershipRow) -> Dict[str, Any]:
+        inst = row.get("institute") or {}
+        return {
+            "institute_id": row.get("institute_id"),
+            "role": row.get("role") or "teacher",
+            "status": row.get("status") or "active",
+            "is_active": bool(row.get("is_active", True)),
+            "institute": {
+                "id": inst.get("id") or row.get("institute_id"),
+                "name": inst.get("name"),
+                "urn": inst.get("urn"),
+                "website": inst.get("website"),
+                "address": inst.get("address") or {},
+                "metadata": inst.get("metadata") or {},
+            },
+        }
+
+    def _permissions(self, active: MembershipRow, platform_admin: bool, super_admin: bool) -> Dict[str, bool]:
+        permissions = dict(BASE_PERMISSIONS)
+        permissions["platform_admin"] = platform_admin
+        permissions["platform_super_admin"] = super_admin
+        role_permissions = ROLE_PERMISSIONS.get(active.get("role") or "", {})
+        permissions.update(role_permissions)
+        if platform_admin:
+            # Platform authority is additive and must not be reduced by a user's
+            # school membership role (for example a platform admin who also has
+            # a teacher/student membership).
+            permissions.update({
+                "can_create_school": True,
+                "can_manage_school": True,
+                "can_manage_calendar": True,
+                "can_manage_timetable": True,
+                "can_invite_staff": True,
+                "can_manage_classes": True,
+                "can_view_student_data": True,
+            })
+        return permissions
+
+    def _school_status(self, active: MembershipRow, memberships: List[MembershipRow], admin_profile: Optional[Dict[str, Any]]) -> str:
+        if admin_profile:
+            return "platform_admin"
+        if active.get("institute_id"):
+            return "school_admin" if active.get("role") == "school_admin" else "member"
+        if len([m for m in memberships if m.get("is_active")]) > 1:
+            return "multi_school_needs_selection"
+        return "no_school"
+
+    def _calendar_status(self, institute_id: Optional[str]) -> Dict[str, Any]:
+        status = {"available": False, "academic_year_count": 0, "term_count": 0, "current_academic_year_id": None, "needs_setup": True}
+        if not institute_id:
+            return status
+        try:
+            years = _safe_list(self.supabase.table("academic_years").select("id,institute_id").eq("institute_id", institute_id).execute())
+            terms = _safe_list(self.supabase.table("academic_terms").select("id,institute_id").eq("institute_id", institute_id).execute())
+            status["academic_year_count"] = len(years)
+            status["term_count"] = len(terms)
+            status["current_academic_year_id"] = years[0].get("id") if years else None
+            status["available"] = bool(years and terms)
+            status["needs_setup"] = not status["available"]
+        except Exception as exc:
+            logger.warning("Bootstrap calendar status lookup failed: %s", exc)
+        return status
+
+    def _timetable_status(self, user_id: str, institute_id: Optional[str]) -> Dict[str, Any]:
+        status = {"available": False, "teacher_timetable_id": None, "slot_count": 0, "needs_setup": True}
+        if not institute_id:
+            return status
+        try:
+            timetables = _safe_list(
+                self.supabase.table("teacher_timetables")
+                .select("id,profile_id,institute_id")
+                .eq("institute_id", institute_id)
+                .eq("profile_id", user_id)
+                .limit(1)
+                .execute()
+            )
+            if not timetables:
+                # Test/future compatibility for teacher_profile_id naming.
+                timetables = _safe_list(
+                    self.supabase.table("teacher_timetables")
+                    .select("id,teacher_profile_id,institute_id")
+                    .eq("institute_id", institute_id)
+                    .eq("teacher_profile_id", user_id)
+                    .limit(1)
+                    .execute()
+                )
+            if timetables:
+                timetable_id = timetables[0].get("id")
+                slots = _safe_list(
+                    self.supabase.table("teacher_timetable_slots")
+                    .select("id,teacher_timetable_id")
+                    .eq("teacher_timetable_id", timetable_id)
+                    .execute()
+                )
+                status["teacher_timetable_id"] = timetable_id
+                status["slot_count"] = len(slots)
+                status["available"] = bool(slots)
+                status["needs_setup"] = not status["available"]
+        except Exception as exc:
+            logger.warning("Bootstrap timetable status lookup failed: %s", exc)
+        return status
+
+    def _graph_status(self, user_id: str, user_email: str, active: MembershipRow, institute: Optional[Dict[str, Any]]) -> Dict[str, Any]:
+        base = {"available": False, "user_db": None, "institute_db": None, "projection_state": "unknown", "needs_rebuild": True, "last_checked_at": utc_now_iso(), "error_code": None}
+        try:
+            result = self.graph_probe(user_id=user_id, user_email=user_email, active_institute=active, institute=institute)
+            return {**base, **(result or {}), "last_checked_at": (result or {}).get("last_checked_at") or base["last_checked_at"]}
+        except Exception:
+            return {**base, "projection_state": "error", "needs_rebuild": True, "error_code": "neo4j_unavailable"}
+
+    def _onboarding(self, school_status: str, permissions: Dict[str, bool], calendar: Dict[str, Any], timetable: Dict[str, Any], graph: Dict[str, Any]) -> Dict[str, Any]:
+        optional = [] if graph.get("available") else ["graph_rebuild"]
+        if school_status == "no_school":
+            return {"next_step": "create_or_join_school", "required": ["school_membership"], "optional": optional, "message": "Create or join a school to finish setup."}
+        if school_status == "multi_school_needs_selection":
+            return {"next_step": "select_school", "required": ["active_school_selection"], "optional": optional, "message": "Select which school to use for this session."}
+        if permissions.get("can_manage_calendar") and calendar.get("needs_setup"):
+            return {"next_step": "setup_calendar", "required": ["calendar"], "optional": optional, "message": "Set up the school calendar."}
+        if permissions.get("can_manage_timetable") and timetable.get("needs_setup"):
+            return {"next_step": "setup_timetable", "required": ["timetable"], "optional": optional, "message": "Set up the timetable."}
+        if school_status == "school_admin" and permissions.get("can_invite_staff"):
+            return {"next_step": "invite_staff", "required": [], "optional": optional, "message": "Invite staff or continue to your workspace."}
+        return {"next_step": "ready", "required": [], "optional": optional, "message": "Your workspace is ready."}
+
+
+def build_bootstrap_response(credentials: Dict[str, Any]) -> Dict[str, Any]:
+    sb = SupabaseServiceRoleClient()
+    return BootstrapService(sb.supabase).build(credentials)

modules/database/supabase/utils/client.py

@@ -27,9 +27,9 @@ def _create_base_client(url: str, key: str, access_token: Optional[str] = None,
     client_options = SyncClientOptions(
         schema="public",
         storage=SyncMemoryStorage(),
-        headers={{
+        headers={
             "Authorization": auth_header
-        }}
+        }
     )
     return create_client(url, key, options=client_options)
 

routers/me/bootstrap_router.py

@@ -0,0 +1,18 @@
+from typing import Any, Dict
+
+from fastapi import APIRouter, Depends, HTTPException
+
+from modules.auth.supabase_bearer import SupabaseBearer
+from modules.database.services.bootstrap_service import build_bootstrap_response
+
+router = APIRouter()
+auth_scheme = SupabaseBearer()
+
+
+@router.get("/bootstrap")
+async def get_me_bootstrap(credentials: Dict[str, Any] = Depends(auth_scheme)) -> Dict[str, Any]:
+    """Authenticated Supabase-first session/onboarding bootstrap contract."""
+    try:
+        return build_bootstrap_response(credentials)
+    except ValueError as exc:
+        raise HTTPException(status_code=403, detail=str(exc)) from exc

run/routers.py

@@ -38,6 +38,7 @@ from routers import provisioning as provisioning_router
 from routers.transcribe.sessions import router as sessions_router
 from routers.transcribe.canvas_events import router as canvas_events_router
 from routers.transcribe.keywords import router as keywords_router
+from routers.me.bootstrap_router import router as me_bootstrap_router
 
 def register_routes(app: FastAPI):
     logger.info("Starting to register routes...")
@@ -63,6 +64,9 @@ def register_routes(app: FastAPI):
     app.include_router(calendar_structure_router.router, prefix="/database/calendar-structure", tags=["Calendar"])
     app.include_router(worker_structure_router.router, prefix="/database/worker-structure", tags=["Worker"])
 
+    # Session/bootstrap routes
+    app.include_router(me_bootstrap_router, prefix="/me", tags=["Bootstrap"])
+
     # Graph navigation
     app.include_router(graph_tree_router, prefix="/graph", tags=["Graph Navigation"])
     app.include_router(user_init_router, prefix="/user", tags=["User"])

tests/test_me_bootstrap.py

@@ -0,0 +1,278 @@
+import os
+from fastapi import FastAPI
+from fastapi.testclient import TestClient
+
+from routers.me.bootstrap_router import router, auth_scheme
+from modules.database.services.bootstrap_service import BootstrapService
+
+
+USER_ID = "00000000-0000-0000-0000-000000000001"
+INST_A = "10000000-0000-0000-0000-000000000001"
+INST_B = "10000000-0000-0000-0000-000000000002"
+
+
+class FakeResult:
+    def __init__(self, data):
+        self.data = data
+
+
+class FakeQuery:
+    def __init__(self, rows):
+        self.rows = list(rows)
+        self._single = False
+        self._limit = None
+
+    def select(self, *_args, **_kwargs):
+        return self
+
+    def eq(self, key, value):
+        self.rows = [row for row in self.rows if row.get(key) == value]
+        return self
+
+    def in_(self, key, values):
+        values = set(values)
+        self.rows = [row for row in self.rows if row.get(key) in values]
+        return self
+
+    def order(self, *_args, **_kwargs):
+        return self
+
+    def limit(self, count):
+        self._limit = count
+        return self
+
+    def single(self):
+        self._single = True
+        return self
+
+    def execute(self):
+        rows = self.rows[: self._limit] if self._limit is not None else self.rows
+        if self._single:
+            return FakeResult(rows[0] if rows else None)
+        return FakeResult(rows)
+
+
+class FakeSupabase:
+    def __init__(self, tables):
+        self.tables = tables
+
+    def table(self, name):
+        return FakeQuery(self.tables.get(name, []))
+
+
+def profile(user_type="teacher", school_id=None):
+    return {
+        "id": USER_ID,
+        "email": "teacher@example.test",
+        "full_name": "Example Teacher",
+        "display_name": None,
+        "user_type": user_type,
+        "school_id": school_id,
+    }
+
+
+def institute(id_=INST_A, name="Example School", status="active"):
+    return {
+        "id": id_,
+        "name": name,
+        "urn": "123456",
+        "website": "https://school.example",
+        "address": {"town": "Testville"},
+        "metadata": {"internal_note": "should remain institute metadata only"},
+        "status": status,
+        "neo4j_uuid_string": "neo-a" if id_ == INST_A else "neo-b",
+    }
+
+
+def membership(institute_id=INST_A, role="teacher"):
+    return {"profile_id": USER_ID, "institute_id": institute_id, "role": role}
+
+
+def service(tables, graph_probe=None):
+    return BootstrapService(FakeSupabase(tables), graph_probe=graph_probe or (lambda **_: {"available": True, "projection_state": "ready"}))
+
+
+def build(credentials=None, tables=None, graph_probe=None):
+    credentials = credentials or {"sub": USER_ID, "email": "teacher@example.test"}
+    tables = tables or {"profiles": [profile()]}
+    return service(tables, graph_probe).build(credentials)
+
+
+def test_supabase_client_for_user_uses_access_token_authorization(monkeypatch):
+    from modules.database.supabase.utils import client as client_module
+
+    captured = {}
+
+    class FakeOptions:
+        def __init__(self, **kwargs):
+            captured["options_kwargs"] = kwargs
+
+    def fake_create_client(url, key, options=None):
+        captured["url"] = url
+        captured["key"] = key
+        captured["options"] = options
+        return {"ok": True}
+
+    monkeypatch.setenv("SUPABASE_URL", "http://supabase.test")
+    monkeypatch.setenv("ANON_KEY", "anon-key")
+    monkeypatch.setattr(client_module, "SyncClientOptions", FakeOptions)
+    monkeypatch.setattr(client_module, "create_client", fake_create_client)
+
+    anon = client_module.SupabaseAnonClient.for_user("user-token")
+
+    assert anon.access_token == "user-token"
+    assert captured["url"] == "http://supabase.test"
+    assert captured["key"] == "anon-key"
+    assert captured["options_kwargs"]["headers"] == {"Authorization": "Bearer user-token"}
+
+
+def test_no_school_bootstrap_requires_school_membership_but_allows_canvas():
+    payload = build(tables={"profiles": [profile()]})
+
+    assert payload["school_status"] == "no_school"
+    assert payload["active_institute"]["source"] == "none"
+    assert payload["memberships"] == []
+    assert payload["permissions"]["can_create_school"] is True
+    assert payload["permissions"]["can_use_canvas"] is True
+    assert payload["onboarding"]["next_step"] == "create_or_join_school"
+    assert "school_membership" in payload["onboarding"]["required"]
+
+
+def test_single_membership_becomes_active_and_uses_supabase_calendar_timetable_status():
+    payload = build(tables={
+        "profiles": [profile()],
+        "institute_memberships": [membership()],
+        "institutes": [institute()],
+        "academic_years": [{"id": "ay-1", "institute_id": INST_A, "is_current": True}],
+        "academic_terms": [{"id": "term-1", "institute_id": INST_A}],
+        "school_timetables": [{"id": "school-tt-1", "institute_id": INST_A}],
+        "teacher_timetables": [{"id": "teacher-tt-1", "institute_id": INST_A, "teacher_profile_id": USER_ID}],
+        "teacher_timetable_slots": [
+            {"id": "slot-1", "teacher_timetable_id": "teacher-tt-1"},
+            {"id": "slot-2", "teacher_timetable_id": "teacher-tt-1"},
+        ],
+    })
+
+    assert payload["school_status"] == "member"
+    assert payload["active_institute"] == {"id": INST_A, "source": "single_membership", "membership_role": "teacher"}
+    assert payload["calendar_status"] == {
+        "available": True,
+        "academic_year_count": 1,
+        "term_count": 1,
+        "current_academic_year_id": "ay-1",
+        "needs_setup": False,
+    }
+    assert payload["timetable_status"] == {
+        "available": True,
+        "teacher_timetable_id": "teacher-tt-1",
+        "slot_count": 2,
+        "needs_setup": False,
+    }
+    assert payload["onboarding"]["next_step"] == "ready"
+
+
+def test_multiple_memberships_require_selection_when_no_profile_school_id_matches():
+    payload = build(tables={
+        "profiles": [profile()],
+        "institute_memberships": [membership(INST_A, "teacher"), membership(INST_B, "department_head")],
+        "institutes": [institute(INST_A, "Alpha"), institute(INST_B, "Beta")],
+    })
+
+    assert payload["school_status"] == "multi_school_needs_selection"
+    assert payload["active_institute"] == {"id": None, "source": "none", "membership_role": None}
+    assert payload["onboarding"]["next_step"] == "select_school"
+    assert "active_school_selection" in payload["onboarding"]["required"]
+
+
+def test_school_admin_permissions_and_onboarding_invite_staff_after_calendar_timetable_ready():
+    payload = build(tables={
+        "profiles": [profile(user_type="school_admin", school_id=INST_A)],
+        "institute_memberships": [membership(INST_A, "school_admin")],
+        "institutes": [institute()],
+        "academic_years": [{"id": "ay-1", "institute_id": INST_A, "is_current": True}],
+        "academic_terms": [{"id": "term-1", "institute_id": INST_A}],
+        "teacher_timetables": [{"id": "teacher-tt-1", "institute_id": INST_A, "teacher_profile_id": USER_ID}],
+        "teacher_timetable_slots": [{"id": "slot-1", "teacher_timetable_id": "teacher-tt-1"}],
+    })
+
+    assert payload["school_status"] == "school_admin"
+    perms = payload["permissions"]
+    assert perms["can_manage_school"] is True
+    assert perms["can_manage_calendar"] is True
+    assert perms["can_manage_timetable"] is True
+    assert perms["can_invite_staff"] is True
+    assert perms["can_view_student_data"] is True
+    assert payload["onboarding"]["next_step"] == "invite_staff"
+
+
+def test_platform_admin_uses_admin_profiles_without_relying_on_profile_super_admin_type():
+    payload = build(credentials={"sub": USER_ID, "email": "admin@example.test"}, tables={
+        "profiles": [profile(user_type="teacher")],
+        "admin_profiles": [{"id": USER_ID, "admin_role": "owner", "is_super_admin": True}],
+    })
+
+    assert payload["profile"]["user_type"] == "platform_admin"
+    assert payload["school_status"] == "platform_admin"
+    assert payload["permissions"]["platform_admin"] is True
+    assert payload["permissions"]["platform_super_admin"] is True
+    assert payload["permissions"]["can_create_school"] is True
+
+
+def test_platform_admin_membership_role_does_not_reduce_platform_permissions():
+    payload = build(credentials={"sub": USER_ID, "email": "admin@example.test"}, tables={
+        "profiles": [profile(user_type="student")],
+        "admin_profiles": [{"id": USER_ID, "admin_role": "owner", "is_super_admin": True}],
+        "institute_memberships": [membership(INST_A, "student")],
+        "institutes": [institute()],
+    })
+
+    assert payload["school_status"] == "platform_admin"
+    assert payload["permissions"]["platform_admin"] is True
+    assert payload["permissions"]["can_manage_school"] is True
+    assert payload["permissions"]["can_manage_calendar"] is True
+    assert payload["permissions"]["can_view_student_data"] is True
+
+
+def test_neo4j_probe_failure_does_not_block_supabase_bootstrap_state():
+    def failing_probe(**_kwargs):
+        raise RuntimeError("connection refused with host details that must not leak")
+
+    payload = build(tables={
+        "profiles": [profile()],
+        "institute_memberships": [membership()],
+        "institutes": [institute()],
+    }, graph_probe=failing_probe)
+
+    assert payload["school_status"] == "member"
+    assert payload["active_institute"]["id"] == INST_A
+    assert payload["graph_status"]["available"] is False
+    assert payload["graph_status"]["projection_state"] == "error"
+    assert payload["graph_status"]["needs_rebuild"] is True
+    assert payload["graph_status"]["error_code"] == "neo4j_unavailable"
+    assert "connection refused" not in str(payload)
+
+
+def test_route_is_authenticated_and_does_not_return_raw_auth_metadata(monkeypatch):
+    app = FastAPI()
+    app.include_router(router, prefix="/me")
+    app.dependency_overrides[auth_scheme] = lambda: {"sub": USER_ID, "email": "teacher@example.test", "app_metadata": {"secret": "nope"}}
+
+    monkeypatch.setattr(
+        "routers.me.bootstrap_router.build_bootstrap_response",
+        lambda credentials: {
+            "profile": {"id": USER_ID, "email": credentials["email"], "display_name": "Example", "user_type": "teacher", "school_id": None},
+            "memberships": [],
+            "active_institute": {"id": None, "source": "none", "membership_role": None},
+            "permissions": {"platform_admin": False},
+            "school_status": "no_school",
+            "onboarding": {"next_step": "create_or_join_school", "required": [], "optional": [], "message": "Create or join a school."},
+            "calendar_status": {"available": False, "academic_year_count": 0, "term_count": 0, "current_academic_year_id": None, "needs_setup": True},
+            "timetable_status": {"available": False, "teacher_timetable_id": None, "slot_count": 0, "needs_setup": True},
+            "graph_status": {"available": False, "user_db": None, "institute_db": None, "projection_state": "unknown", "needs_rebuild": True, "last_checked_at": "2026-05-28T00:00:00Z", "error_code": None},
+        },
+    )
+
+    response = TestClient(app).get("/me/bootstrap")
+    assert response.status_code == 200
+    assert "app_metadata" not in response.text
+    assert "secret" not in response.text