"""HTTP surface for the mover review packet. **The access decision here is the feature.** Every other role-change endpoint is ``require_admin`` — and that is exactly why the old review never worked as intended: an IT admin looking at 「Jason 有 9 条云授权」 has no way to know whether Jason still needs them. The person who knows is the manager who just picked him up. So a packet is readable and answerable by: * ``global_admin`` — the platform-wide operator, as before; * ``auditor`` / ``management`` — read only, same governance surface the rest of the role-change endpoints already expose; * **the assigned reviewer** — the new manager, who is an ordinary employee with no admin role at all. The reviewer gate is per-packet (``reviewer_person_id``), never a role. A manager can answer the packet they were handed and cannot enumerate anybody else's — the ordinary-employee path is a lookup, not a listing. Nothing here executes anything. ``revoke`` and ``transfer`` are answers recorded against a holding; carrying them out happens on that domain's own admin surface, which has its approval, audit and idempotency behaviour. Issue #623's no-auto-revoke-on-a-move policy is unchanged. """ from __future__ import annotations from typing import Any, Literal from uuid import UUID from fastapi import APIRouter, Depends, Query, Request from pydantic import BaseModel, Field from sqlalchemy import func, select from sqlalchemy.ext.asyncio import AsyncSession from app.audit.service import AuditService from app.auth.models import AuthUser from app.core.database import get_db from app.core.dependencies import get_current_user, require_active_employee from app.core.exceptions import AppException, NotFoundError from app.core.rbac import ( AUDITOR_ROLE, GLOBAL_ADMIN_ROLE, MANAGEMENT_ROLE, canonicalize_role_name, ) from app.core.response import success_response from app.domains.hr.models import ( ITEM_DECISIONS, ROLE_REVIEW_OPEN, ROLE_REVIEW_STATUSES, HrRoleChangeReview, HrRoleChangeReviewItem, ) from app.hr_lifecycle.mover_delta import COUNTED_DELTA_KINDS from app.hr_lifecycle.mover_holdings import REVIEW from app.hr_lifecycle.mover_review_ai import build_facts, deterministic_priority from app.hr_lifecycle.mover_review_service import record_decision, refresh_counts from app.identity.person_models import IdentityPerson router = APIRouter( prefix="/api/v1/role-changes/reviews", tags=["role-changes"], ) _READ_ROLES = (GLOBAL_ADMIN_ROLE, AUDITOR_ROLE, MANAGEMENT_ROLE) _WRITE_ROLES = (GLOBAL_ADMIN_ROLE,) def _roles(user: AuthUser) -> set[str]: """Canonical role names for this caller. ``role_names`` is the same property ``require_roles`` reads, so the per-packet gate and the role gates cannot disagree about what a caller holds — including the legacy aliases (``it_admin`` → ``global_admin``) that ``canonicalize_role_name`` folds. """ return {canonicalize_role_name(r) for r in (user.role_names or [])} async def _load_for_read( db: AsyncSession, review_id: UUID, user: AuthUser, ) -> HrRoleChangeReview: review = await db.get(HrRoleChangeReview, review_id) if review is None: raise NotFoundError("HrRoleChangeReview", str(review_id)) if _roles(user) & set(_READ_ROLES): return review if review.reviewer_person_id and review.reviewer_person_id == user.id: return review # Deliberately 404, not 403: an ordinary employee learning that a # packet EXISTS for a colleague is itself a disclosure about that # colleague's org move. raise NotFoundError("HrRoleChangeReview", str(review_id)) async def _load_for_write( db: AsyncSession, review_id: UUID, user: AuthUser, ) -> HrRoleChangeReview: review = await db.get(HrRoleChangeReview, review_id) if review is None: raise NotFoundError("HrRoleChangeReview", str(review_id)) if _roles(user) & set(_WRITE_ROLES): return review if review.reviewer_person_id and review.reviewer_person_id == user.id: return review raise NotFoundError("HrRoleChangeReview", str(review_id)) def _serialize_review( review: HrRoleChangeReview, *, person: IdentityPerson | None = None, names: dict[UUID, str] | None = None, ) -> dict[str, Any]: """One packet as JSON. ``names`` resolves the three person references the reviewer surface shows by name (old manager, new manager, reviewer). Passed in rather than looked up here so a list of 50 packets is three queries, not 150 — and so the caller decides whether it needs them at all. """ lookup = names or {} def _person_ref(pid: UUID | None) -> dict[str, str] | None: if pid is None: return None return {"id": str(pid), "display_name": lookup.get(pid) or ""} return { "id": str(review.id), "person_id": str(review.person_id), "person_name": person.display_name if person else None, "person_email": person.work_email if person else None, "effective_date": review.effective_date.isoformat(), "move_kind": review.move_kind, "status": review.status, "before_dept_code": review.before_dept_code, "after_dept_code": review.after_dept_code, "before_dept_path": review.before_dept_path, "after_dept_path": review.after_dept_path, "before_job_title": review.before_job_title, "after_job_title": review.after_job_title, # The manager axis was stored from day one and never exposed. # It is the whole explanation for eight of the nine cards sent # on 2026-09-05: same department on both sides, only the # reporting line moved. A reviewer who cannot see that cannot # tell why they were asked. "before_manager": _person_ref(review.before_manager_person_id), "after_manager": _person_ref(review.after_manager_person_id), "reviewer_person_id": ( str(review.reviewer_person_id) if review.reviewer_person_id else None ), "reviewer": _person_ref(review.reviewer_person_id), "notified_at": ( review.notified_at.isoformat() if review.notified_at else None ), "notify_suppressed_at": ( review.notify_suppressed_at.isoformat() if review.notify_suppressed_at else None ), "notify_suppressed_reason": review.notify_suppressed_reason, "total_items": review.total_items, "reviewable_items": review.reviewable_items, "decided_items": review.decided_items, # ``reviewable_items`` is how much the person HOLDS; # ``delta_items`` is how much this MOVE calls into question. # Conflating the two is the 2026-09-05 incident in one sentence, # so both are returned and every surface says which it is # showing. "baseline_source": review.baseline_source, "delta_items": review.delta_items, # The engine's working. NULL means "not computed yet" — a # distinct state from "computed and found nothing", and the page # says which. "delta_summary": review.delta_summary, "delta_computed_at": ( review.delta_computed_at.isoformat() if review.delta_computed_at else None ), # Surfaced rather than swallowed: a packet that silently dropped # a category reads as "not much to decide", which is the wrong # conclusion to hand somebody. "failed_scan_types": review.failed_scan_types or [], "ai_summary_zh": review.ai_summary_zh, "ai_summary_en": review.ai_summary_en, "ai_model": review.ai_model, "ai_failure_code": review.ai_failure_code, "ai_generated_at": ( review.ai_generated_at.isoformat() if review.ai_generated_at else None ), "created_at": review.created_at.isoformat() if review.created_at else None, "resolved_at": ( review.resolved_at.isoformat() if review.resolved_at else None ), } def _serialize_item(item: HrRoleChangeReviewItem) -> dict[str, Any]: return { "id": str(item.id), "resource_type": item.resource_type, "resource_id": str(item.resource_id), "label": item.label, "disposition": item.disposition, "detail": item.detail or {}, "decision": item.decision, "decision_note": item.decision_note, "decided_at": item.decided_at.isoformat() if item.decided_at else None, "decided_by_person_id": ( str(item.decided_by_person_id) if item.decided_by_person_id else None ), "ai_recommendation": item.ai_recommendation, "ai_reason": item.ai_reason, # What the move said about this holding, and why. NULL means no # verdict has been computed for it yet. "delta_kind": item.delta_kind, "delta_evidence": item.delta_evidence or {}, } async def _resolve_names( db: AsyncSession, reviews: list[HrRoleChangeReview], ) -> dict[UUID, str]: """Display names for every person a batch of packets references. One query for the whole batch. Missing / soft-deleted people are simply absent from the map — the serializer renders an empty name rather than inventing one, and the caller's own "this person has left" copy is what explains it. """ ids: set[UUID] = set() for review in reviews: for pid in ( review.before_manager_person_id, review.after_manager_person_id, review.reviewer_person_id, ): if pid is not None: ids.add(pid) if not ids: return {} rows = ( await db.execute( select(IdentityPerson.id, IdentityPerson.display_name) .where(IdentityPerson.id.in_(ids)) ) ).all() return {row[0]: row[1] or "" for row in rows} @router.get("", summary="List mover review packets") async def list_reviews( request: Request, status: str | None = Query(None, max_length=16), mine: bool = Query( False, description=( "Only packets assigned to me as reviewer. The default view " "for a manager, who holds no admin role." ), ), person_id: UUID | None = Query( None, description=( "Packets about ONE person — how the admin role-change page " "finds the packet for the row it already has open. Narrowing " "only; it cannot widen what the caller may already see." ), ), page: int = Query(1, ge=1), page_size: int = Query(20, ge=1, le=100), current_user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """Packets this caller may see. An admin/auditor/management caller sees everything; anybody else sees only what they were asked to review. ``mine=true`` narrows an admin to their own queue — a global admin who is also somebody's manager has both, and conflating them buries the two packets they personally owe an answer on inside two hundred they do not. """ stmt = select(HrRoleChangeReview, IdentityPerson).join( IdentityPerson, IdentityPerson.id == HrRoleChangeReview.person_id, ) privileged = bool(_roles(current_user) & set(_READ_ROLES)) if mine or not privileged: stmt = stmt.where( HrRoleChangeReview.reviewer_person_id == current_user.id, ) if person_id is not None: stmt = stmt.where(HrRoleChangeReview.person_id == person_id) if status: if status not in ROLE_REVIEW_STATUSES: raise AppException( code="INVALID_STATUS", message=f"status must be one of {ROLE_REVIEW_STATUSES}", status_code=400, ) stmt = stmt.where(HrRoleChangeReview.status == status) total = await db.scalar( select(func.count()).select_from(stmt.subquery()), ) or 0 rows = ( await db.execute( stmt.order_by(HrRoleChangeReview.created_at.desc()) .offset((page - 1) * page_size) .limit(page_size) ) ).all() names = await _resolve_names(db, [r for r, _ in rows]) return success_response( data=[_serialize_review(r, person=p, names=names) for r, p in rows], meta={"page": page, "page_size": page_size, "total": int(total)}, request=request, ) @router.get("/{review_id}", summary="One packet, grouped by category") async def get_review( review_id: UUID, request: Request, resource_type: str | None = Query( None, max_length=32, description="Drill into one category's items.", ), undecided_only: bool = Query(False), delta_only: bool = Query( False, description=( "Only holdings this MOVE calls into question, as opposed to " "everything the person happens to hold. Empty whenever no " "baseline could speak — which is not the same as 'nothing to " "worry about', and the page says so." ), ), page: int = Query(1, ge=1), page_size: int = Query(50, ge=1, le=200), current_user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """The packet, its per-category tallies, and one page of items. Categories come back whole and items come back paged, because the shape of the real thing is "ten categories, 238 items": a reviewer decides at the category level and drills in, and a surface that forces them to page through 238 undifferentiated rows is the review storm this feature exists to avoid. """ review = await _load_for_read(db, review_id, current_user) person = await db.get(IdentityPerson, review.person_id) facts = await build_facts(db, review) order = deterministic_priority(facts) by_type = {c.resource_type: c for c in facts.categories} # Per-category delta counts in one grouped query, not one per # category: a real packet is ten-plus categories and this endpoint # is on the reviewer's critical path. # How many of the delta items still have no answer. ``delta_items`` # is a fact about the MOVE and does not shrink as decisions land, so # a headline built on it alone still reads 「让 4 项需要重新判断」 to # somebody who has just answered all four. Computed at read time # because it depends on decisions, which the engine cannot know when # it classifies. delta_undecided = int( await db.scalar( select(func.count()) .select_from(HrRoleChangeReviewItem) .where( HrRoleChangeReviewItem.review_id == review.id, HrRoleChangeReviewItem.delta_kind.in_(COUNTED_DELTA_KINDS), HrRoleChangeReviewItem.decision.is_(None), ) ) or 0 ) delta_by_type = { rt: int(n) for rt, n in ( await db.execute( select( HrRoleChangeReviewItem.resource_type, func.count(), ) .where( HrRoleChangeReviewItem.review_id == review.id, HrRoleChangeReviewItem.delta_kind.in_(COUNTED_DELTA_KINDS), ) .group_by(HrRoleChangeReviewItem.resource_type) ) ).all() } categories = [ { "resource_type": rt, "total": by_type[rt].total, "undecided": by_type[rt].undecided, "examples": by_type[rt].examples, # How many in this category the MOVE calls into question, as # opposed to how many the person holds. Zero here with a # non-zero total is a meaningful, and common, answer. "delta": delta_by_type.get(rt, 0), } for rt in order if rt in by_type ] item_stmt = select(HrRoleChangeReviewItem).where( HrRoleChangeReviewItem.review_id == review.id, ) if resource_type: item_stmt = item_stmt.where( HrRoleChangeReviewItem.resource_type == resource_type, ) if undecided_only: item_stmt = item_stmt.where( HrRoleChangeReviewItem.decision.is_(None), HrRoleChangeReviewItem.disposition == REVIEW, ) if delta_only: item_stmt = item_stmt.where( HrRoleChangeReviewItem.delta_kind.in_(COUNTED_DELTA_KINDS), ) item_total = await db.scalar( select(func.count()).select_from(item_stmt.subquery()), ) or 0 items = ( await db.execute( item_stmt.order_by( HrRoleChangeReviewItem.resource_type.asc(), HrRoleChangeReviewItem.label.asc(), ) .offset((page - 1) * page_size) .limit(page_size) ) ).scalars().all() return success_response( data={ "review": { **_serialize_review( review, person=person, names=await _resolve_names(db, [review]), ), "delta_undecided": delta_undecided, }, "categories": categories, "items": [_serialize_item(i) for i in items], }, meta={ "page": page, "page_size": page_size, "total": int(item_total), }, request=request, ) class _DecisionBody(BaseModel): decision: Literal["keep", "revoke", "transfer"] note: str | None = Field(default=None, max_length=500) @router.put( "/{review_id}/items/{item_id}", summary="Record one decision on one holding", ) async def decide_item( review_id: UUID, item_id: UUID, body: _DecisionBody, request: Request, current_user: AuthUser = Depends(require_active_employee), db: AsyncSession = Depends(get_db), ): review = await _load_for_write(db, review_id, current_user) item = await db.get(HrRoleChangeReviewItem, item_id) if item is None or item.review_id != review.id: raise NotFoundError("HrRoleChangeReviewItem", str(item_id)) try: await record_decision( db, item, decision=body.decision, decided_by=current_user.id, note=body.note, ) except ValueError as exc: raise AppException( code="INVALID_DECISION", message=str(exc), status_code=400, ) from exc await refresh_counts(db, review) await AuditService(db).log( trace_id=str(request.state.trace_id), actor_user_id=current_user.id, actor_type="user", action="hr.role_change_review.decide", resource_type="hr_role_change_review_item", resource_id=str(item.id), status="success", detail={ "review_id": str(review.id), "resource_type": item.resource_type, "decision": body.decision, # Recorded so a later audit can ask whether the suggestions # were any good — and whether people simply took them. "ai_recommendation": item.ai_recommendation, "followed_ai": ( item.ai_recommendation == body.decision if item.ai_recommendation else None ), }, ) await db.commit() return success_response(data=_serialize_item(item), request=request) class _BulkBody(BaseModel): resource_type: str = Field(max_length=32) decision: Literal["keep", "revoke", "transfer"] note: str | None = Field(default=None, max_length=500) @router.put( "/{review_id}/categories", summary="Record the same decision for every undecided holding in one category", ) async def decide_category( review_id: UUID, body: _BulkBody, request: Request, current_user: AuthUser = Depends(require_active_employee), db: AsyncSession = Depends(get_db), ): """The bulk answer, and the reason the packet is usable at all. 111 DNS records is one decision about DNS records, not 111 decisions. Forcing a row-by-row answer on a category a reviewer has already judged as a whole does not produce 111 considered judgements — it produces 111 clicks and a reviewer who stops reading. Only UNDECIDED items are touched: a bulk action must not overwrite an answer somebody gave deliberately on one row. """ review = await _load_for_write(db, review_id, current_user) if body.decision not in ITEM_DECISIONS: raise AppException( code="INVALID_DECISION", message=f"decision must be one of {ITEM_DECISIONS}", status_code=400, ) items = ( await db.execute( select(HrRoleChangeReviewItem).where( HrRoleChangeReviewItem.review_id == review.id, HrRoleChangeReviewItem.resource_type == body.resource_type, HrRoleChangeReviewItem.disposition == REVIEW, HrRoleChangeReviewItem.decision.is_(None), ) ) ).scalars().all() for item in items: await record_decision( db, item, decision=body.decision, decided_by=current_user.id, note=body.note, ) await refresh_counts(db, review) await AuditService(db).log( trace_id=str(request.state.trace_id), actor_user_id=current_user.id, actor_type="user", action="hr.role_change_review.decide_category", resource_type="hr_role_change_review", resource_id=str(review.id), status="success", detail={ "resource_type": body.resource_type, "decision": body.decision, "affected": len(items), }, ) await db.commit() return success_response( data={ "affected": len(items), "review": _serialize_review(review), }, request=request, ) @router.post( "/{review_id}/rescan", summary="Re-scan the person's holdings into this packet", ) async def rescan( review_id: UUID, request: Request, current_user: AuthUser = Depends(require_active_employee), db: AsyncSession = Depends(get_db), ): """Top up a packet whose first scan ran while a domain was failing. Idempotent by construction: items are keyed on ``(review_id, resource_type, resource_id)``, so a re-scan adds what is new and cannot touch a decision that has already been made. """ review = await _load_for_write(db, review_id, current_user) from app.hr_lifecycle.mover_review_service import build_review_packet await build_review_packet( db, review.person_id, before={ "dept_code": review.before_dept_code, "manager_person_id": ( str(review.before_manager_person_id) if review.before_manager_person_id else None ), "job_title": review.before_job_title, }, after={ "dept_code": review.after_dept_code, "manager_person_id": ( str(review.after_manager_person_id) if review.after_manager_person_id else None ), "job_title": review.after_job_title, }, effective_date=review.effective_date, lifecycle_event_id=review.lifecycle_event_id, ) await db.commit() await db.refresh(review) return success_response( data=_serialize_review(review), request=request, ) __all__ = ["router", "ROLE_REVIEW_OPEN"]