"""MeUniverseService — aggregates one user's IT footprint into a single response shape consumed by the unified ``/me`` page. The service is read-only and stateless. Every section's data comes from existing domain queries; this service is the assembly layer that orders, decorates, and decorates them with the user-vs-admin mode signal. It does NOT reach into any domain's mutating service. """ from __future__ import annotations import logging from collections import defaultdict from dataclasses import dataclass from datetime import date from decimal import Decimal from typing import Any, cast from uuid import UUID from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.compliance.models import ComplianceIncident from app.core.workflow_journal import WorkflowJournal from app.domains.apikey.models import ApiKeyAssignment, ApiKeyProvider from app.domains.asset.models import ( AssetItem, AssetSoftwareCatalog, SoftwareLicensePlan, SoftwareSeatAllocation, ) from app.domains.cmdb.models import ( CmdbCloudAccessGrant, CmdbCloudScope, CmdbDomain, CmdbServer, ) from app.domains.email_group.models import EmailGroup, EmailGroupOwner from app.domains.leaver_blocked.models import ( DUAL_SIGN_PENDING_SECONDARY, LeaverDualSignOverride, ) from app.domains.offboarding.models import ( OffboardingProcess, ) from app.domains.permission.models import SoftwarePermission, UserPermission from app.identity.person_models import IdentityPerson from app.notifications.models import NotificationOutbox logger = logging.getLogger(__name__) @dataclass class ItHomeResponse: """Final shape returned to the frontend.""" user: dict[str, Any] mode: str # 'self' | 'view_admin' | 'view_manager' | 'view_auditor' summary: dict[str, Any] assets: list[dict[str, Any]] seats: list[dict[str, Any]] permissions: list[dict[str, Any]] api_keys: list[dict[str, Any]] #: Third-party SAML / SSO systems this person holds an approved grant #: for (and recently-revoked ones whose IdP-side removal is still #: outstanding — the holder should see that the access is being taken #: away, not just that it vanished). Added 2026-09-05: this was the one #: access object with no employee-facing surface at all. sso_grants: list[dict[str, Any]] cloud_servers: list[dict[str, Any]] cloud_domains: list[dict[str, Any]] cloud_grants: list[dict[str, Any]] email_groups: list[dict[str, Any]] #: Internal systems this person is accountable for (系统台账). Read #: through the registry coordinator — this domain never imports #: another domain's models. registered_systems: list[dict[str, Any]] workflows: list[dict[str, Any]] todos: list[dict[str, Any]] notifications: list[dict[str, Any]] compliance_self_check: dict[str, Any] cost_estimate: dict[str, Any] leaver_status: dict[str, Any] | None = None class MeUniverseService: def __init__(self, db: AsyncSession) -> None: self.db = db self.journal = WorkflowJournal(db) # ── Authorisation helper ──────────────────────────────────── @staticmethod def resolve_view_mode(viewer: IdentityPerson, target_id: UUID) -> str: """Return the view mode label without performing the access check (the check is in :meth:`assert_can_view`). Used by the endpoint to tag the response so the frontend renders appropriately.""" if viewer.id == target_id: return "self" from app.core.rbac import ( AUDITOR_ROLE, GLOBAL_ADMIN_ROLE, HR_ROLE, MANAGEMENT_ROLE, has_role, ) if has_role( viewer.role_names, GLOBAL_ADMIN_ROLE, HR_ROLE, MANAGEMENT_ROLE, ): return "view_admin" if has_role(viewer.role_names, AUDITOR_ROLE): return "view_auditor" return "view_manager" # fallback — assert_can_view validates async def assert_can_view( self, viewer: IdentityPerson, target: IdentityPerson, ) -> None: """Raise ForbiddenError if the viewer cannot see this target. Self always allowed; global_admin / hr / auditor / management allowed for anyone; managers allowed for any direct or transitive report (depth 6 — caller-side explicit depth, matches the ``permission/router.py`` cross-user-grant walk; the prior depth=4 cut off a CEO viewing N=5 levels down in a 500-person org). Management is the read-only leadership view and mirrors auditor / global_admin's any-target scope. """ from app.core.exceptions import ForbiddenError from app.core.rbac import ( AUDITOR_ROLE, GLOBAL_ADMIN_ROLE, HR_ROLE, MANAGEMENT_ROLE, has_role, ) if viewer.id == target.id: return if has_role( viewer.role_names, GLOBAL_ADMIN_ROLE, HR_ROLE, AUDITOR_ROLE, MANAGEMENT_ROLE, ): return # Manager chain check — walk up to depth 6. Pre-launch # hardening 2026-05-04: skip soft-deleted managers so the # walk doesn't terminate prematurely at a leaver mid-chain # (and so we don't leak through a stale manager_person_id # pointer pointing at a deleted person). # Pre-launch hardening 2026-05-06: ALSO reject when the # walking ``cur`` is the viewer but viewer is in # ``employment_status='left'``. A left manager whose # IdentityPerson row hasn't yet been soft-deleted (the row # only soft-deletes via separate cleanup, not on # offboarding completion) would otherwise still gate access # via the chain match. The same guard is applied implicitly # at every step below — once we observe a "left" manager in # the chain we stop, since their access path is closed. if viewer.employment_status == "left": raise ForbiddenError( "Offboarded managers cannot view subordinates' /me data." ) # Cycle guard: data corruption (A.manager=B + B.manager=A) used # to silently burn the 6-iteration budget bouncing between two # rows; with this set, we detect and break out cleanly so a # wider corruption never produces an opaque ForbiddenError + # 6 wasted DB round-trips. (Round-17 audit.) visited: set = {target.id} cur = target for _ in range(6): if cur.manager_person_id is None: break if cur.manager_person_id == viewer.id: return if cur.manager_person_id in visited: logger.warning( "manager-chain cycle detected starting at %s — " "aborting walk", target.id, ) break visited.add(cur.manager_person_id) next_manager = await self.db.get( IdentityPerson, cur.manager_person_id, ) if next_manager is None or next_manager.deleted_at is not None: break cur = next_manager # Stop walking through a manager who has left — their # subordinates are pending re-point, and they should not # gate access for anyone above them via a stale chain. if cur.employment_status == "left": break raise ForbiddenError( f"You don't have permission to view {target.work_email}'s " f"IT home. Self / admin / auditor / manager-chain only." ) # ── Aggregation ───────────────────────────────────────────── async def get_it_home( self, viewer: IdentityPerson, target: IdentityPerson, ) -> ItHomeResponse: await self.assert_can_view(viewer, target) mode = self.resolve_view_mode(viewer, target.id) # Fetch every section in parallel-friendly fashion (one query # each — SQLAlchemy async session handles them sequentially # but each is a tight indexed query). assets = await self._assets_for(target.id) seats = await self._seats_for(target.id) permissions = await self._permissions_for(target.id) api_keys = await self._api_keys_for(target.id) sso_grants = await self._sso_grants_for(target.id) cloud_servers = await self._cloud_servers_for(target.id) cloud_domains = await self._cloud_domains_for(target.id) cloud_grants = await self._cloud_grants_for(target.id) email_groups = await self._email_groups_for(target.id) registered_systems = await self._registered_systems_for(target.id) workflows = await self._workflows_for(target.id) todos = await self._todos_for(target, mode) notifications = await self._notifications_for(target.id, mode) compliance = await self._compliance_self_check(target) cost = self._cost_estimate(seats) leaver_status = await self._leaver_status(target) summary = { "asset_count": len(assets), "seat_count": len(seats), "permission_count": len(permissions), "api_key_count": len(api_keys), "sso_grant_count": len([ g for g in sso_grants if g.get("status") == "approved" ]), "cloud_server_count": len(cloud_servers), "cloud_domain_count": len(cloud_domains), "cloud_grant_count": len(cloud_grants), "email_group_count": len(email_groups), "registered_system_count": len(registered_systems), # Counted separately from the total: being SOMEBODY'S backup # is not the same as being on the hook, and only primary # ownership blocks an offboarding. "owned_system_count": len([ s for s in registered_systems if s.get("is_primary_owner") ]), "active_workflow_count": sum( 1 for w in workflows if w["status"] == "running" ), "open_todo_count": len(todos), "unread_notification_count": sum( 1 for n in notifications if not n.get("read_at") ), "monthly_cost_usd": cost["monthly_total_usd"], } return ItHomeResponse( user={ "id": str(target.id), "work_email": target.work_email, "display_name": target.display_name or target.work_email, "employment_status": target.employment_status, "primary_department": target.primary_department, "manager_person_id": ( str(target.manager_person_id) if target.manager_person_id else None ), "termination_date": ( target.termination_date.isoformat() if target.termination_date else None ), "force_logout_at": ( target.force_logout_at.isoformat() if target.force_logout_at else None ), }, mode=mode, summary=summary, assets=assets, seats=seats, permissions=permissions, api_keys=api_keys, sso_grants=sso_grants, cloud_servers=cloud_servers, cloud_domains=cloud_domains, cloud_grants=cloud_grants, email_groups=email_groups, registered_systems=registered_systems, workflows=workflows, todos=todos, notifications=notifications, compliance_self_check=compliance, cost_estimate=cost, leaver_status=leaver_status, ) async def get_holdings( self, viewer: IdentityPerson, target: IdentityPerson, ) -> tuple[list[dict[str, Any]], list[str]]: """Return the complete lifecycle inventory after the normal view check. This is deliberately separate from :meth:`get_it_home`: the inventory scans every lifecycle resource category, whereas the operational graph must remain a fast, independently cached view. Both surfaces retain exactly the same target resolution and RBAC boundary. """ await self.assert_can_view(viewer, target) return await self._holdings_for(target.id) async def _holdings_for( self, target_id: UUID, ) -> tuple[list[dict[str, Any]], list[str]]: """Return the authoritative personal holding catalogue. This is intentionally *not* assembled from the older per-tab queries above: their predicates were designed for individual UI views and omit whole lifecycle categories. The coordinator calls the same protected read used by mover review, preserving both its coverage contract and its partial-failure visibility. """ from app.core.coordinators.person_holdings_coordinator import ( scan_person_holdings, ) scan = await scan_person_holdings(self.db, target_id) return ( [ { "resource_type": holding.resource_type, "resource_id": str(holding.resource_id), "label": holding.label, "disposition": holding.disposition, "detail": holding.detail, } for holding in scan.holdings ], scan.failed_types, ) # ── Per-section queries ───────────────────────────────────── async def _assets_for(self, target_id: UUID) -> list[dict[str, Any]]: rows = list((await self.db.execute( select(AssetItem).where( AssetItem.assigned_to == target_id, AssetItem.deleted_at.is_(None), # exclude soft-deleted / disposed assets ) )).scalars().all()) return [ { "id": str(a.id), "asset_tag": a.asset_tag, "asset_name": a.asset_name, "category": a.category, "brand": a.brand, "model": a.model, "serial_number": a.serial_number, "status": a.status, "assigned_at": a.assigned_at.isoformat() if a.assigned_at else None, "expected_return_date": ( a.expected_return_date.isoformat() if a.expected_return_date else None ), # Real model column is ``warranty_until`` — pre-2026-04-30 # this used the non-existent ``warranty_end_date`` and # ``getattr(default=None)`` silently dropped every value. "warranty_until": ( a.warranty_until.isoformat() if a.warranty_until else None ), } for a in rows ] async def _seats_for(self, target_id: UUID) -> list[dict[str, Any]]: rows = list((await self.db.execute( select(SoftwareSeatAllocation).where( SoftwareSeatAllocation.user_id == target_id, SoftwareSeatAllocation.status == "active", ) )).scalars().all()) # Pre-batch the joined Plan and Software rows so this stays O(1) # round-trips for the typical case. ``software_id`` lives on the # plan, NOT on ``software_seat_allocations`` itself — pre-2026-04-30 # the code referenced ``seat.software_id`` (silently None) and # ``seat.started_at`` (column is ``allocated_at``). plan_ids = {s.plan_id for s in rows if s.plan_id} plan_by_id: dict[UUID, SoftwareLicensePlan] = {} if plan_ids: plan_rows = (await self.db.execute( select(SoftwareLicensePlan).where( SoftwareLicensePlan.id.in_(plan_ids), ) )).scalars().all() plan_by_id = {p.id: p for p in plan_rows} sw_ids = {p.software_id for p in plan_by_id.values() if p.software_id} sw_name_by_id: dict[UUID, str] = {} if sw_ids: sw_rows = (await self.db.execute( select(AssetSoftwareCatalog).where( AssetSoftwareCatalog.id.in_(sw_ids), ) )).scalars().all() sw_name_by_id = {s.id: s.name for s in sw_rows} out: list[dict[str, Any]] = [] for seat in rows: plan = plan_by_id.get(seat.plan_id) if seat.plan_id else None software_id = plan.software_id if plan is not None else None out.append({ "id": str(seat.id), "software_id": str(software_id) if software_id else None, "software_name": ( sw_name_by_id.get(software_id) if software_id else None ), "plan_id": str(seat.plan_id) if seat.plan_id else None, "plan_name": plan.name if plan is not None else None, "plan_price": ( float(plan.price) if plan is not None and plan.price is not None else None ), "plan_cycle": plan.billing_cycle if plan is not None else None, "allocated_at": ( seat.allocated_at.isoformat() if seat.allocated_at else None ), }) return out async def _permissions_for(self, target_id: UUID) -> list[dict[str, Any]]: rows = list((await self.db.execute( select(UserPermission).where(UserPermission.user_id == target_id) )).scalars().all()) # ``permission_key`` and ``software_id`` live on SoftwarePermission # (the catalog row), not on the join row UserPermission. # Pre-2026-04-30 the code referenced ``p.permission_key`` / # ``p.software_id`` directly via ``getattr(default=None)`` — # every cell silently None. catalog_ids = {p.permission_id for p in rows if p.permission_id} catalog_by_id: dict[UUID, SoftwarePermission] = {} if catalog_ids: catalog_rows = (await self.db.execute( select(SoftwarePermission).where( SoftwarePermission.id.in_(catalog_ids), ) )).scalars().all() catalog_by_id = {c.id: c for c in catalog_rows} out: list[dict[str, Any]] = [] for p in rows: cat = catalog_by_id.get(p.permission_id) out.append({ "id": str(p.id), "permission_key": cat.permission_key if cat is not None else None, "permission_display_name": ( cat.display_name if cat is not None else None ), "software_id": ( str(cat.software_id) if cat is not None and cat.software_id else None ), "status": p.status, "granted_at": ( p.granted_at.isoformat() if p.granted_at else None ), }) return out async def _api_keys_for(self, target_id: UUID) -> list[dict[str, Any]]: rows = list((await self.db.execute( select(ApiKeyAssignment).where(ApiKeyAssignment.assigned_to == target_id) )).scalars().all()) # Resolve provider names in one batch — saves N+1 round-trips. provider_ids = {a.provider_id for a in rows if a.provider_id} provider_name_by_id: dict[UUID, str] = {} if provider_ids: prov_rows = (await self.db.execute( select(ApiKeyProvider).where( ApiKeyProvider.id.in_(provider_ids), ) )).scalars().all() provider_name_by_id = {p.id: p.name for p in prov_rows} today = date.today() out: list[dict[str, Any]] = [] for a in rows: # Compute expiry telemetry the same way the deleted MyAssetsPage # did, so the row can render the warning + extend button. valid_until = a.valid_until days_until_expiry: int | None = None expiry_alert: str | None = None if valid_until is not None: delta = (valid_until - today).days days_until_expiry = delta if delta < 0: expiry_alert = "expired" elif delta <= 14: expiry_alert = "expiring_soon" # has_key tells the UI whether a "reveal" button makes sense; # key_masked is a presentational hint (last 4 chars). The # actual plaintext is only returned by the dedicated # POST /api-keys/{id}/reveal endpoint, which audit-logs the # access. kv = a.key_value or "" has_key = bool(kv) key_masked = ( f"••••{kv[-4:]}" if has_key and len(kv) >= 4 else "••••" if has_key else None ) out.append({ "id": str(a.id), "request_number": a.request_number, "provider_id": str(a.provider_id) if a.provider_id else None, "provider_name": ( provider_name_by_id.get(a.provider_id) if a.provider_id else None ), "project_name": a.project_name, "monthly_budget_usd": ( float(a.monthly_budget_usd) if a.monthly_budget_usd is not None else None ), "valid_from": a.valid_from.isoformat() if a.valid_from else None, "valid_until": valid_until.isoformat() if valid_until else None, "days_until_expiry": days_until_expiry, "expiry_alert": expiry_alert, "has_key": has_key, "key_masked": key_masked, "extended_count": a.extended_count, "status": a.status, }) return out async def _sso_grants_for(self, target_id: UUID) -> list[dict[str, Any]]: """Approved SSO grants, plus revoked ones whose IdP-side removal is still open (``removal_status`` in the open set) — the holder should be able to see that an access is being withdrawn. Fully-settled revocations (confirmed / verified) and never-granted requests are not part of "what I hold".""" from app.domains.sso.models import ( # noqa: PLC0415 REMOVAL_OPEN_STATUSES, SSO_GRANT_STATUS_APPROVED, SSO_GRANT_STATUS_REVOKED, SsoGrant, ) rows = list((await self.db.execute( select(SsoGrant).where( SsoGrant.person_id == str(target_id), SsoGrant.status.in_((SSO_GRANT_STATUS_APPROVED, SSO_GRANT_STATUS_REVOKED)), ).order_by(SsoGrant.granted_at.desc()) )).scalars().all()) out: list[dict[str, Any]] = [] for g in rows: if ( g.status == SSO_GRANT_STATUS_REVOKED and g.removal_status not in REMOVAL_OPEN_STATUSES ): continue out.append({ "id": str(g.id), "system_name": g.system_name, "sso_provider": g.sso_provider, "system_owner_label": g.system_owner_label, "status": g.status, "granted_at": g.granted_at.isoformat() if g.granted_at else None, "revoked_at": g.revoked_at.isoformat() if g.revoked_at else None, "revoke_reason": g.revoke_reason, "removal_status": g.removal_status, }) return out async def _cloud_servers_for(self, target_id: UUID) -> list[dict[str, Any]]: rows = list((await self.db.execute( select(CmdbServer).where( CmdbServer.owner_person_id == str(target_id), CmdbServer.deleted_at.is_(None), ) )).scalars().all()) # ``provider`` lives on the integration row (one cloud account = # one provider); ``public_ip`` is the legacy singular shape, the # column is ``public_ips`` (JSON list). Pre-2026-04-30 these # were ``getattr(default=None)`` — every cell silently None. from app.integrations.models import Integration integ_ids = {s.integration_id for s in rows if s.integration_id} integ_provider_by_id: dict[str, str] = {} if integ_ids: integ_rows = (await self.db.execute( select(Integration).where( Integration.id.in_([UUID(str(i)) for i in integ_ids]), ) )).scalars().all() integ_provider_by_id = {str(i.id): i.provider for i in integ_rows} out: list[dict[str, Any]] = [] for s in rows: ips = s.public_ips if isinstance(s.public_ips, list) else [] out.append({ "id": str(s.id), "name": s.name, "provider": integ_provider_by_id.get(str(s.integration_id)), "region": s.region, "public_ip": ips[0] if ips else None, "status": s.status, }) return out async def _cloud_domains_for(self, target_id: UUID) -> list[dict[str, Any]]: # ``deleted_at IS NULL`` was missing here — sibling # ``_cloud_servers_for`` carries it but this method didn't, so # the /me payload silently surfaced soft-deleted domains. Same # bug class as the "Integration.enabled across CMDB lists" fix # this PR ships — both are "list-time hygiene filters that # were skipped on one branch of a copy-paste". rows = list((await self.db.execute( select(CmdbDomain).where( CmdbDomain.owner_person_id == str(target_id), CmdbDomain.deleted_at.is_(None), ) )).scalars().all()) return [ { "id": str(d.id), # Direct attribute access — these are real ORM columns # on ``CmdbDomain``. The previous ``getattr(d, "col", # None)`` form silently returned None on a future # column rename, degrading the /me payload to "no # domain data" without any AttributeError to surface # the bug. Round-16 sweep of the same bug class # (feedback_no_getattr_default_for_model_columns). "domain_name": d.domain_name, "expiry_date": ( d.expiry_date.isoformat() if d.expiry_date else None ), "icp_filing_no": d.icp_filing_no, } for d in rows ] async def _cloud_grants_for(self, target_id: UUID) -> list[dict[str, Any]]: rows = list((await self.db.execute( select(CmdbCloudAccessGrant).where( CmdbCloudAccessGrant.person_id == str(target_id), CmdbCloudAccessGrant.status == "active", ) )).scalars().all()) # One query for every referenced scope (#2193-1). ``/me`` is on # the hot path for every employee; a per-grant ``db.get`` is N # round-trips for one indexed IN. scope_ids = {str(g.scope_id) for g in rows if g.scope_id} scope_names: dict[str, str | None] = {} if scope_ids: scope_names = { str(sid): name for sid, name in (await self.db.execute( select(CmdbCloudScope.id, CmdbCloudScope.display_name) .where(CmdbCloudScope.id.in_(list(scope_ids))) )).all() } out: list[dict[str, Any]] = [] for g in rows: # Pre-launch hardening 2026-05-04: direct attribute access # so a model rename surfaces as AttributeError in tests # instead of silently returning None to /me payloads. scope_name = scope_names.get(str(g.scope_id)) if g.scope_id else None out.append({ "id": str(g.id), "scope_id": str(g.scope_id) if g.scope_id else None, "scope_name": scope_name, "role_key": g.role_key, "expires_at": ( g.expires_at.isoformat() if g.expires_at else None ), "last_login_at": ( g.last_login_at.isoformat() if g.last_login_at else None ), }) return out async def _registered_systems_for( self, target_id: UUID, ) -> list[dict[str, Any]]: """Internal systems this person owns or backs up. Why it belongs on this page: offboarding blocks on exactly these rows, and a blocker at departure for something the person could never see anywhere is how "I didn't know I owned that" happens. """ from app.core.coordinators import ( system_registry_coordinator as registry, ) return cast( list[dict[str, Any]], await registry.systems_owned_by(self.db, target_id), ) async def _email_groups_for(self, target_id: UUID) -> list[dict[str, Any]]: owner_rows = list((await self.db.execute( select(EmailGroupOwner).where( EmailGroupOwner.person_id == target_id, ) )).scalars().all()) # The FK column on EmailGroupOwner is ``email_group_id``, not # ``group_id``. Direct attribute access on the wrong name was # raising AttributeError for any user who owned at least one # email group — making /me 500 on prod. Bug fix 2026-04-30. # One query for every owned group (#2193-2) — same reasoning as # the cloud-grant prefetch above. group_ids = [ r.email_group_id for r in owner_rows if r.email_group_id ] groups_by_id = {} if group_ids: groups_by_id = { str(g.id): g for g in (await self.db.execute( select(EmailGroup).where(EmailGroup.id.in_(group_ids)) )).scalars().all() } out: list[dict[str, Any]] = [] for owner_row in owner_rows: grp = groups_by_id.get(str(owner_row.email_group_id)) if grp is None or grp.status != "active": continue out.append({ "id": str(grp.id), "email": grp.email, "name": grp.name, # ``member_count`` is not a stored column — Google's # directory is the source of truth. The unified /me # surface intentionally skips it; the dedicated # /company/mail-groups page does the live count. "status": grp.status, }) return out async def _workflows_for(self, target_id: UUID) -> list[dict[str, Any]]: runs = await self.journal.list_runs_for_person(target_id, limit=20) return [ { "id": str(r.id), "workflow_type": r.workflow_type, "status": r.status, "current_step": r.current_step, "step_count": r.step_count, "started_at": r.started_at.isoformat(), "last_step_at": ( r.last_step_at.isoformat() if r.last_step_at else None ), "completed_at": ( r.completed_at.isoformat() if r.completed_at else None ), "error_message": r.error_message, "step_log": r.step_log or [], "related_resource_type": r.related_resource_type, "related_resource_id": ( str(r.related_resource_id) if r.related_resource_id else None ), } for r in runs ] async def _todos_for( self, target: IdentityPerson, mode: str, ) -> list[dict[str, Any]]: """Self mode: incoming transfer requests + pending dual-sign secondary assignments. Other modes: empty (admin doesn't act on behalf).""" if mode != "self": return [] from app.domains.offboarding.models import ( XFER_PENDING, AssetTransferRequest, ) todos: list[dict[str, Any]] = [] # Incoming transfers waiting on this user's accept transfer_rows = list((await self.db.execute( select(AssetTransferRequest).where( AssetTransferRequest.recipient_id == target.id, AssetTransferRequest.status == XFER_PENDING, ) )).scalars().all()) for t in transfer_rows: # Title is rendered via i18n on the frontend using # `title_key` + `title_params`. The backend stays language- # agnostic so /me can render in zh-CN or en-US. todos.append({ "kind": "transfer_accept", "id": str(t.id), "title_key": "myItHome.todos.transferAccept", "title_params": { "assetType": t.asset_type, "assetLabel": t.asset_label, }, "created_at": ( t.created_at.isoformat() if t.created_at else None ), # No per-transfer detail route exists; ``/my/transfers`` # redirects to ``/my/todos`` which is the canonical hub # where every recipient sees their pending transfers. "actor_link": "/my/transfers", }) # Dual-sign override secondary assignments override_rows = list((await self.db.execute( select(LeaverDualSignOverride).where( LeaverDualSignOverride.secondary_signer_user_id == target.id, LeaverDualSignOverride.status == DUAL_SIGN_PENDING_SECONDARY, ) )).scalars().all()) for o in override_rows: todos.append({ "kind": "leaver_dual_sign_secondary", "id": str(o.id), "title_key": "myItHome.todos.leaverDualSignSecondary", "title_params": {}, "created_at": ( o.created_at.isoformat() if o.created_at else None ), "actor_link": f"/admin/leaver-blocked/override/{o.id}", }) # Sort newest first todos.sort(key=lambda x: x["created_at"] or "", reverse=True) return todos async def _notifications_for( self, target_id: UUID, mode: str, ) -> list[dict[str, Any]]: """Self mode: latest 20 notifications. Other modes: empty.""" if mode != "self": return [] rows = list((await self.db.execute( select(NotificationOutbox) .where(NotificationOutbox.recipient_person_id == target_id) .order_by(NotificationOutbox.created_at.desc()) .limit(20) )).scalars().all()) # ``channel`` (singular) doesn't exist on notification_outbox — # the column is ``channels`` (JSON list of dispatched channels # such as ``["email", "feishu"]``). Pre-2026-04-30 the row # showed an empty channel chip on /me because the wrong name # silently returned None. return [ { "id": str(n.id), "title": n.title, "channels": list(n.channels) if n.channels else [], "status": n.status, "created_at": ( n.created_at.isoformat() if n.created_at else None ), "read_at": ( n.read_at.isoformat() if n.read_at else None ), } for n in rows ] async def _compliance_self_check( self, target: IdentityPerson, ) -> dict[str, Any]: """Personal compliance posture. Surfaces things the user (or their admin viewing) can act on.""" # Outstanding compliance incidents naming this person incident_rows = list((await self.db.execute( select(ComplianceIncident).where( ComplianceIncident.person_id == target.id, ComplianceIncident.status == "open", ) .order_by(ComplianceIncident.severity, ComplianceIncident.created_at.desc()) .limit(20) )).scalars().all()) return { "open_incident_count": len(incident_rows), "open_incidents": [ { "id": str(i.id), "incident_type": i.incident_type, "severity": i.severity, "title": i.title, "created_at": ( i.created_at.isoformat() if i.created_at else None ), } for i in incident_rows[:5] ], "force_logout_at": ( target.force_logout_at.isoformat() if target.force_logout_at else None ), "termination_date": ( target.termination_date.isoformat() if target.termination_date else None ), "employment_status": target.employment_status, } def _cost_estimate( self, seats: list[dict[str, Any]], ) -> dict[str, Any]: """Sum monthly contributions of every active seat. Yearly is amortised /12; one-time billing is excluded.""" monthly_total = Decimal("0") per_software: dict[str, Decimal] = defaultdict(lambda: Decimal("0")) for seat in seats: price = seat.get("plan_price") cycle = (seat.get("plan_cycle") or "").lower() sw = seat.get("software_name") or "Unknown" if price is None: continue if cycle == "monthly": contrib = Decimal(str(price)) elif cycle == "yearly": contrib = Decimal(str(price)) / Decimal("12") else: # one-time / unknown — exclude from monthly recurring continue monthly_total += contrib per_software[sw] += contrib return { "monthly_total_usd": float(monthly_total.quantize(Decimal("0.01"))), "per_software": { k: float(v.quantize(Decimal("0.01"))) for k, v in per_software.items() }, } async def _leaver_status( self, target: IdentityPerson, ) -> dict[str, Any] | None: """Show the leaver process snapshot when target is in leaver_pending or has a recent process. None for never- offboarded active employees.""" proc = await self.db.scalar( select(OffboardingProcess).where( OffboardingProcess.person_id == target.id, ) ) if proc is None: return None return { "id": str(proc.id), "status": proc.status, "leaver_date": ( proc.leaver_date.isoformat() if proc.leaver_date else None ), "current_step": proc.current_step, "first_blocked_at": ( proc.first_blocked_at.isoformat() if proc.first_blocked_at else None ), "sla_breach_level": proc.sla_breach_level, "ready_at": ( proc.ready_at.isoformat() if proc.ready_at else None ), "completed_at": ( proc.completed_at.isoformat() if proc.completed_at else None ), "sso_suspended_at": ( proc.sso_suspended_at.isoformat() if proc.sso_suspended_at else None ), } __all__ = ["MeUniverseService", "ItHomeResponse"]