from __future__ import annotations

from datetime import date, timedelta
from decimal import Decimal
from typing import Any

from sqlalchemy import exists, func, or_, select
from sqlalchemy.orm import Session, selectinload

from app.modules.clients.association_models import ClientAssociation
from app.modules.clients.models import Client
from app.modules.alerts.workflow_escalations import list_workflow_escalations
from app.modules.core.iam.models import User
from app.modules.core.rbac.models import Role, UserRole
from app.modules.core.tenancy.models import Branch, Tenant
from app.modules.employees.models import Employee
from app.modules.services.execution import (
    CLOSED_TASK_STATUSES,
    aqmm_task_summary_for_subscription,
    closure_readiness_for_subscription,
)
from app.modules.services.models import (
    ClientServiceSubscription,
    ClientServiceTaskInstance,
    ServiceCatalogue,
    ServiceTaskComment,
)
from app.modules.employees.service import _engagement_sla, _engagement_team, _weighted_progress

try:
    from app.modules.billing.models import BillingInvoice
except Exception:  # pragma: no cover
    BillingInvoice = None

PARTNER_ROLES = {"Partner", "System Admin"}


def _count(db: Session, stmt) -> int:
    return int(db.execute(stmt).scalar() or 0)


def _money(value: Any) -> Decimal:
    return Decimal(str(value or "0"))


def get_user_role_names(db: Session, user_id: int) -> list[str]:
    return list(
        db.execute(
            select(Role.name)
            .join(UserRole, UserRole.role_id == Role.id)
            .where(UserRole.user_id == int(user_id), Role.is_active.is_(True))
            .order_by(Role.name.asc())
        ).scalars().all()
    )


def can_access_partner_dashboard(db: Session, current_user) -> bool:
    roles = set(get_user_role_names(db, current_user.id))
    return bool(roles.intersection(PARTNER_ROLES))


def _active_scope(db: Session, request, current_user, roles: set[str]) -> dict[str, Any]:
    tenant_id = request.session.get("active_tenant_id") or getattr(current_user, "tenant_id", None)
    branch_id = request.session.get("active_branch_id") or getattr(current_user, "branch_id", None)

    if "System Admin" not in roles:
        tenant_id = getattr(current_user, "tenant_id", None)

    # Partner operations are branch-centric. For normal partner users, do not allow
    # accidental all-branch view unless the branch context is genuinely missing.
    if "System Admin" not in roles:
        branch_id = getattr(current_user, "branch_id", None) or branch_id

    if branch_id in (None, "", 0, "0"):
        branch_id = None

    tenant = db.get(Tenant, int(tenant_id)) if tenant_id else None
    branch = db.get(Branch, int(branch_id)) if branch_id else None
    return {
        "tenant_id": int(tenant_id) if tenant_id else None,
        "branch_id": int(branch_id) if branch_id else None,
        "tenant": tenant,
        "branch": branch,
    }


def _financial_year(request) -> str | None:
    value = request.session.get("active_financial_year") or getattr(request.state, "year_code", None)
    value = (value or "").strip()
    return value or None


def _is_scoped_partner(roles: set[str]) -> bool:
    """Return True only for a normal Partner login.

    System Admin and Firm Admin retain their existing administrative scope.  A
    normal Partner is always ownership-scoped, even when a branch is selected.
    """
    return "Partner" in roles and "System Admin" not in roles and "Firm Admin" not in roles


def _partner_client_access_expression(*, partner_user_id: int, tenant_id: int | None):
    association_access = exists(
        select(ClientAssociation.id).where(
            ClientAssociation.client_id == Client.id,
            ClientAssociation.partner_user_id == int(partner_user_id),
            or_(
                ClientAssociation.firm_tenant_id == tenant_id,
                ClientAssociation.firm_tenant_id.is_(None),
            ) if tenant_id is not None else ClientAssociation.id.is_not(None),
        )
    )
    review_access = exists(
        select(ClientServiceSubscription.id).where(
            ClientServiceSubscription.client_id == Client.id,
            ClientServiceSubscription.review_partner_user_id == int(partner_user_id),
            ClientServiceSubscription.is_active.is_(True),
            ClientServiceSubscription.tenant_id == tenant_id if tenant_id is not None else ClientServiceSubscription.id.is_not(None),
        )
    )
    return or_(
        Client.partner_id == int(partner_user_id),
        association_access,
        review_access,
    )


def _task_scope(stmt, tenant_id: int | None, branch_id: int | None, current_user, roles: set[str], fy: str | None = None):
    if tenant_id:
        stmt = stmt.where(ClientServiceTaskInstance.tenant_id == tenant_id)
    stmt = stmt.where(ClientServiceTaskInstance.is_active.is_(True))
    if branch_id is not None:
        stmt = stmt.where(ClientServiceTaskInstance.branch_id == branch_id)
    if fy:
        stmt = stmt.where(ClientServiceTaskInstance.financial_year == fy)
    if _is_scoped_partner(roles):
        stmt = stmt.where(
            ClientServiceTaskInstance.subscription.has(
                or_(
                    ClientServiceSubscription.assigned_partner_user_id == current_user.id,
                    ClientServiceSubscription.performing_partner_user_id == current_user.id,
                    ClientServiceSubscription.review_partner_user_id == current_user.id,
                )
            )
        )
    return stmt


def _subscription_scope(stmt, tenant_id: int | None, branch_id: int | None, current_user, roles: set[str], fy: str | None = None):
    if tenant_id:
        stmt = stmt.where(ClientServiceSubscription.tenant_id == tenant_id)
    stmt = stmt.where(ClientServiceSubscription.is_active.is_(True))
    if branch_id is not None:
        stmt = stmt.where(ClientServiceSubscription.branch_id == branch_id)
    if fy:
        stmt = stmt.where(ClientServiceSubscription.financial_year == fy)
    if _is_scoped_partner(roles):
        stmt = stmt.where(
            or_(
                ClientServiceSubscription.assigned_partner_user_id == current_user.id,
                ClientServiceSubscription.performing_partner_user_id == current_user.id,
                ClientServiceSubscription.review_partner_user_id == current_user.id,
            )
        )
    return stmt


def _client_scope(stmt, tenant_id: int | None, branch_id: int | None, current_user, roles: set[str]):
    if tenant_id:
        stmt = stmt.where(Client.tenant_id == tenant_id)
    stmt = stmt.where(Client.is_active.is_(True))
    if branch_id is not None:
        stmt = stmt.where(Client.branch_id == branch_id)
    if _is_scoped_partner(roles):
        stmt = stmt.where(
            _partner_client_access_expression(
                partner_user_id=int(current_user.id),
                tenant_id=tenant_id,
            )
        )
    return stmt


def _is_open_status(status: str | None) -> bool:
    return (status or "pending").strip().lower() not in CLOSED_TASK_STATUSES


def _task_label(task: ClientServiceTaskInstance) -> dict[str, Any]:
    client = getattr(task, "client", None)
    catalogue = getattr(task, "catalogue", None)
    assignee = getattr(task, "assigned_to", None)
    status = (task.status or "pending").replace("_", " ").title()
    return {
        "id": task.id,
        "subscription_id": task.subscription_id,
        "client_name": getattr(client, "client_name", None) or "Unlinked Client",
        "client_code": getattr(client, "client_code", None) or "",
        "service_name": getattr(catalogue, "service_name", None) or "Service",
        "task_name": task.task_name,
        "period": task.financial_year or "-",
        "due_date": task.internal_target_date,
        "status": task.status or "pending",
        "status_label": status,
        "priority": task.priority or "normal",
        "assigned_to": getattr(assignee, "full_name", None) or getattr(assignee, "email", None) or "Unassigned",
        "is_overdue": bool(task.internal_target_date and task.internal_target_date < date.today() and _is_open_status(task.status)),
        "href": f"/work/engagements/{task.subscription_id}" if task.subscription_id else "/partner/reviews",
    }


def _load_tasks(db: Session, tenant_id: int | None, branch_id: int | None, current_user, roles: set[str], fy: str | None) -> list[ClientServiceTaskInstance]:
    stmt = (
        select(ClientServiceTaskInstance)
        .options(
            selectinload(ClientServiceTaskInstance.client),
            selectinload(ClientServiceTaskInstance.catalogue),
            selectinload(ClientServiceTaskInstance.assigned_to),
            selectinload(ClientServiceTaskInstance.subscription).selectinload(ClientServiceSubscription.catalogue),
        )
    )
    stmt = _task_scope(stmt, tenant_id, branch_id, current_user, roles, fy)
    stmt = stmt.order_by(
        ClientServiceTaskInstance.internal_target_date.is_(None),
        ClientServiceTaskInstance.internal_target_date.asc(),
        ClientServiceTaskInstance.priority.desc(),
        ClientServiceTaskInstance.id.desc(),
    ).limit(300)
    return list(db.execute(stmt).scalars().all())


def _load_clients(db: Session, tenant_id: int | None, branch_id: int | None, current_user, roles: set[str]) -> list[dict[str, Any]]:
    stmt = _client_scope(select(Client), tenant_id, branch_id, current_user, roles)
    clients = db.execute(stmt.order_by(Client.client_name.asc()).limit(5000)).scalars().all()
    out: list[dict[str, Any]] = []
    for client in clients:
        task_count = _count(db, _task_scope(select(func.count(ClientServiceTaskInstance.id)), tenant_id, branch_id, current_user, roles).where(ClientServiceTaskInstance.client_id == client.id))
        overdue_count = _count(
            db,
            _task_scope(select(func.count(ClientServiceTaskInstance.id)), tenant_id, branch_id, current_user, roles)
            .where(ClientServiceTaskInstance.client_id == client.id, ClientServiceTaskInstance.internal_target_date < date.today(), ClientServiceTaskInstance.status.notin_(list(CLOSED_TASK_STATUSES))),
        )
        service_count = _count(db, _subscription_scope(select(func.count(ClientServiceSubscription.id)), tenant_id, branch_id, current_user, roles).where(ClientServiceSubscription.client_id == client.id))
        out.append({
            "id": client.id,
            "client_name": client.client_name,
            "client_code": client.client_code,
            "pan": client.pan,
            "gstin": client.gstin,
            "email": client.email,
            "mobile": client.mobile,
            "status": client.status or ("active" if client.is_active else "inactive"),
            "service_count": service_count,
            "task_count": task_count,
            "overdue_count": overdue_count,
            "href": f"/clients/{client.id}",
        })
    return out


def _staff_rows(db: Session, tenant_id: int | None, branch_id: int | None, tasks: list[ClientServiceTaskInstance]) -> list[dict[str, Any]]:
    if not tenant_id:
        return []
    stmt = (
        select(User)
        .join(UserRole, UserRole.user_id == User.id)
        .join(Role, Role.id == UserRole.role_id)
        .where(
            User.tenant_id == tenant_id,
            User.is_active.is_(True),
            Role.is_active.is_(True),
            Role.name.in_(("Staff", "Branch Manager")),
        )
        .distinct()
    )
    if branch_id is not None:
        stmt = stmt.where(User.branch_id == branch_id)
    users = db.execute(
        stmt.order_by(User.full_name.asc(), User.email.asc()).limit(100)
    ).scalars().all()
    by_user: dict[int, dict[str, int]] = {}
    for task in tasks:
        uid = getattr(task, "assigned_to_user_id", None)
        if not uid:
            continue
        bucket = by_user.setdefault(int(uid), {"active": 0, "overdue": 0, "review": 0, "client_pending": 0})
        if _is_open_status(task.status):
            bucket["active"] += 1
        if task.internal_target_date and task.internal_target_date < date.today() and _is_open_status(task.status):
            bucket["overdue"] += 1
        if (task.status or "").lower() == "completed":
            bucket["review"] += 1
        if (task.status or "").lower() == "blocked":
            bucket["client_pending"] += 1
    rows: list[dict[str, Any]] = []
    for user in users:
        stats = by_user.get(int(user.id), {"active": 0, "overdue": 0, "review": 0, "client_pending": 0})
        total = stats["active"] + stats["review"]
        rows.append({
            "id": user.id,
            "name": user.full_name or user.email,
            "email": user.email,
            "designation": user.designation,
            "active": stats["active"],
            "overdue": stats["overdue"],
            "review": stats["review"],
            "client_pending": stats["client_pending"],
            "load_status": "Heavy" if total >= 40 else ("Balanced" if total >= 10 else "Light"),
        })
    rows.sort(key=lambda r: (r["active"] + r["review"], r["overdue"]), reverse=True)
    return rows[:25]


def _billing_rows(
    db: Session,
    tenant_id: int | None,
    branch_id: int | None,
    fy: str | None,
    current_user,
    roles: set[str],
) -> dict[str, Any]:
    if BillingInvoice is None or not tenant_id:
        return {"available": False, "invoices": [], "invoice_count": 0, "draft_count": 0, "outstanding": Decimal("0")}
    stmt = select(BillingInvoice).where(BillingInvoice.tenant_id == tenant_id)
    if branch_id is not None:
        stmt = stmt.where(BillingInvoice.branch_id == branch_id)
    if fy:
        stmt = stmt.where(BillingInvoice.financial_year == fy)
    if _is_scoped_partner(roles):
        allowed_client_ids = select(Client.id).where(
            Client.tenant_id == tenant_id,
            _partner_client_access_expression(
                partner_user_id=int(current_user.id),
                tenant_id=tenant_id,
            ),
        )
        stmt = stmt.where(BillingInvoice.client_id.in_(allowed_client_ids))
    invoices = list(db.execute(stmt.order_by(BillingInvoice.invoice_date.desc(), BillingInvoice.id.desc()).limit(25)).scalars().all())
    outstanding = sum((_money(getattr(inv, "balance_amount", 0)) for inv in invoices), Decimal("0"))
    return {
        "available": True,
        "invoices": invoices,
        "invoice_count": len(invoices),
        "draft_count": len([i for i in invoices if (getattr(i, "status", "") or "").upper() == "DRAFT"]),
        "outstanding": outstanding,
    }


def build_partner_dashboard_payload(db: Session, request, current_user) -> dict[str, Any]:
    roles = set(get_user_role_names(db, current_user.id))
    scope = _active_scope(db, request, current_user, roles)
    tenant_id = scope["tenant_id"]
    branch_id = scope["branch_id"]
    fy = _financial_year(request)
    today = date.today()
    next_week = today + timedelta(days=7)

    tasks = _load_tasks(db, tenant_id, branch_id, current_user, roles, fy)
    task_rows = [_task_label(t) for t in tasks]
    open_rows = [r for r in task_rows if r["status"] not in CLOSED_TASK_STATUSES]
    overdue_rows = [r for r in open_rows if r["due_date"] and r["due_date"] < today]
    due_today_rows = [r for r in open_rows if r["due_date"] == today]
    due_week_rows = [r for r in open_rows if r["due_date"] and today <= r["due_date"] <= next_week]
    client_pending_rows = [r for r in open_rows if r["status"] == "blocked"]
    review_rows = [r for r in task_rows if r["status"] == "completed"]

    clients = _load_clients(db, tenant_id, branch_id, current_user, roles)
    staff = _staff_rows(db, tenant_id, branch_id, tasks)
    billing = _billing_rows(db, tenant_id, branch_id, fy, current_user, roles)
    unified_escalations = list_workflow_escalations(db, tenant_id=tenant_id, branch_id=branch_id, assigned_to_user_id=current_user.id)

    subscriptions_count = _count(db, _subscription_scope(select(func.count(ClientServiceSubscription.id)), tenant_id, branch_id, current_user, roles, fy)) if tenant_id else 0

    overview = {
        "tenant": scope["tenant"],
        "branch": scope["branch"],
        "tenant_id": tenant_id,
        "branch_id": branch_id,
        "financial_year": fy,
        "client_count": len(clients),
        "subscription_count": subscriptions_count,
        "open_task_count": len(open_rows),
        "overdue_count": len(overdue_rows),
        "due_today_count": len(due_today_rows),
        "due_week_count": len(due_week_rows),
        "client_pending_count": len(client_pending_rows),
        "review_pending_count": len(review_rows),
        "staff_count": len(staff),
        "invoice_count": billing["invoice_count"],
        "outstanding": billing["outstanding"],
        "today": today,
        "unified_escalation_count": len(unified_escalations),
        "escalations_over_3_days": len([r for r in unified_escalations if r["age_days"] >= 3]),
    }

    return {
        "roles": sorted(roles),
        "overview": overview,
        "branch_work": task_rows[:80],
        "due_today": due_today_rows[:25],
        "due_week": due_week_rows[:25],
        "overdue": overdue_rows[:25],
        "client_pending": client_pending_rows[:25],
        "review_queue": review_rows[:50],
        "partner_engagement_review_queue": _partner_review_queue_rows(db, request, current_user),
        "unified_escalations": unified_escalations,
        "clients": clients,
        "staff_rows": staff,
        "billing": billing,
        "reports": _report_cards(),
        "wizards": _wizard_cards(),
    }


def _report_cards() -> list[dict[str, str]]:
    return [
        {"group": "Work Reports", "title": "Branch Work Status", "desc": "Due today, due this week, overdue and blocked work for the active branch.", "href": "/partner/dashboard?tab=branch-work"},
        {"group": "Work Reports", "title": "Overdue Task Report", "desc": "Tasks past internal target date requiring partner escalation.", "href": "/partner/dashboard?tab=branch-work"},
        {"group": "Review Reports", "title": "Partner Review Queue", "desc": "Completed work waiting for partner review or final sign-off.", "href": "/partner/dashboard?tab=review"},
        {"group": "Client Reports", "title": "Branch Client Health", "desc": "Clients, services, open tasks and overdue items in the branch.", "href": "/partner/dashboard?tab=clients"},
        {"group": "Staff Reports", "title": "Staff Workload", "desc": "Branch team workload, overdue count and client pending count.", "href": "/partner/dashboard?tab=staff"},
        {"group": "Billing Reports", "title": "Billing Control", "desc": "Invoices raised, drafts and outstanding amount for the active branch.", "href": "/partner/dashboard?tab=billing"},
    ]


def _wizard_cards() -> list[dict[str, str]]:
    return [
        {"title": "Add / Manage Clients", "desc": "Open client master for branch client creation and edits.", "href": "/clients"},
        {"title": "Client Service Subscriptions", "desc": "Assign firm services to branch clients and monitor engagements.", "href": "/services/engagements"},
        {"title": "Generate / Track Tasks", "desc": "Use work tracker for generated compliance and audit tasks.", "href": "/work"},
        {"title": "Partner Review Board", "desc": "Review completed work and send rework or clarifications.", "href": "/partner/reviews"},
        {"title": "Documents", "desc": "Open branch and engagement documents.", "href": "/documents"},
        {"title": "Billing", "desc": "Raise invoices and track collection follow-up.", "href": "/billing"},
    ]


# -----------------------------------------------------------------------------
# Phase 5B - Partner engagement review and release workspace
# -----------------------------------------------------------------------------

def _display_user(user) -> str:
    return getattr(user, "full_name", None) or getattr(user, "email", None) or "Not assigned"


def _partner_task_category(task: ClientServiceTaskInstance) -> str:
    return (getattr(task, "task_category", None) or "General Workflow").strip() or "General Workflow"


def _partner_review_level_for_task(task: ClientServiceTaskInstance, subscription: ClientServiceSubscription, current_user) -> str | None:
    """Return the review level the current partner is authorised to perform.

    Review Partner work is kept separate from Engagement Partner work. System and
    Firm Admin users may inspect the workspace, but review decisions are selected
    from the engagement assignments rather than from a new permission model.
    """
    uid = int(getattr(current_user, "id", 0) or 0)
    is_review_partner = uid and uid == int(getattr(subscription, "review_partner_user_id", 0) or 0)
    is_engagement_partner = uid and uid == int(getattr(subscription, "assigned_partner_user_id", 0) or 0)

    if is_review_partner and getattr(task, "aqmm_review_partner_required", False):
        return "review_partner"
    if is_engagement_partner and getattr(task, "aqmm_partner_review_required", False):
        return "partner"
    return None


def _partner_review_state(task: ClientServiceTaskInstance, review_level: str | None) -> str:
    if getattr(task, "rework_status", "none") == "open":
        return "rework"
    if review_level == "review_partner":
        status = getattr(task, "review_partner_review_status", "not_required")
    elif review_level == "partner":
        status = getattr(task, "partner_review_status", "not_required")
    else:
        return "read_only"
    if status == "reviewed":
        return "reviewed"
    if status == "rework_required":
        return "rework"
    # Partner review must follow any required Manager review.
    if getattr(task, "aqmm_manager_review_required", False) and getattr(task, "manager_review_status", "not_required") != "reviewed":
        return "not_ready"
    # Review Partner review must follow required Engagement Partner review.
    if review_level == "review_partner" and getattr(task, "aqmm_partner_review_required", False) and getattr(task, "partner_review_status", "not_required") != "reviewed":
        return "not_ready"
    if (getattr(task, "status", "pending") or "pending") not in CLOSED_TASK_STATUSES and getattr(task, "submitted_for_review_at_utc", None) is None:
        return "not_ready"
    return "pending_review"


def _partner_task_payload(task: ClientServiceTaskInstance, subscription: ClientServiceSubscription, current_user) -> dict[str, Any]:
    review_level = _partner_review_level_for_task(task, subscription, current_user)
    review_state = _partner_review_state(task, review_level)
    comments = [c for c in (getattr(task, "comments", None) or []) if not getattr(c, "is_deleted", False)]
    documents = [d for d in (getattr(task, "documents", None) or []) if not getattr(d, "is_deleted", False)]
    assignee = getattr(task, "assigned_to", None)
    return {
        "id": task.id,
        "task_name": task.task_name,
        "description": task.description,
        "sequence_no": task.sequence_no,
        "category": _partner_task_category(task),
        "status": task.status or "pending",
        "status_label": (task.status or "pending").replace("_", " ").title(),
        "assigned_to": _display_user(assignee),
        "response_type": getattr(task, "response_type", "NONE") or "NONE",
        "checklist_response": getattr(task, "checklist_response", None),
        "checklist_text_response": getattr(task, "checklist_text_response", None),
        "checklist_number_response": getattr(task, "checklist_number_response", None),
        "checklist_date_response": getattr(task, "checklist_date_response", None),
        "checklist_remarks": getattr(task, "checklist_remarks", None),
        "work_remarks": getattr(task, "remarks", None),
        "manager_review_status": getattr(task, "manager_review_status", "not_required"),
        "manager_review_note": getattr(task, "manager_review_note", None),
        "partner_review_status": getattr(task, "partner_review_status", "not_required"),
        "partner_review_note": getattr(task, "partner_review_note", None),
        "review_partner_review_status": getattr(task, "review_partner_review_status", "not_required"),
        "review_partner_review_note": getattr(task, "review_partner_review_note", None),
        "rework_status": getattr(task, "rework_status", "none"),
        "rework_reason": getattr(task, "rework_reason", None),
        "is_aqmm_task": bool(getattr(task, "is_aqmm_task", False)),
        "aqmm_reference": getattr(task, "aqmm_reference", None),
        "aqmm_status": getattr(task, "aqmm_status", "not_required"),
        "blocks_final_release": bool(getattr(task, "aqmm_blocks_final_release", False)),
        "partner_review_required": bool(getattr(task, "aqmm_partner_review_required", False)),
        "review_partner_required": bool(getattr(task, "aqmm_review_partner_required", False)),
        "review_level": review_level,
        "review_state": review_state,
        "can_review": review_level is not None and review_state in {"pending_review", "reviewed", "rework"},
        "is_exception": getattr(task, "checklist_response", None) == "NO" or getattr(task, "rework_status", "none") == "open",
        "evidence_count": len(documents),
        "comments": comments,
        "is_locked": bool(getattr(task, "is_locked", False) or getattr(subscription, "is_locked", False)),
    }


def _partner_review_queue_rows(db: Session, request, current_user, *, limit: int = 80) -> list[dict[str, Any]]:
    roles = set(get_user_role_names(db, current_user.id))
    scope = _active_scope(db, request, current_user, roles)
    fy = _financial_year(request)
    stmt = (
        select(ClientServiceSubscription)
        .options(
            selectinload(ClientServiceSubscription.client),
            selectinload(ClientServiceSubscription.catalogue),
            selectinload(ClientServiceSubscription.assigned_partner),
            selectinload(ClientServiceSubscription.assigned_manager),
            selectinload(ClientServiceSubscription.assigned_staff),
            selectinload(ClientServiceSubscription.review_partner),
        )
    )
    stmt = _subscription_scope(stmt, scope["tenant_id"], scope["branch_id"], current_user, roles, fy)
    if "System Admin" not in roles and "Firm Admin" not in roles:
        stmt = stmt.where(or_(
            ClientServiceSubscription.assigned_partner_user_id == current_user.id,
            ClientServiceSubscription.review_partner_user_id == current_user.id,
        ))
    subscriptions = list(db.execute(stmt.order_by(ClientServiceSubscription.current_due_date.is_(None), ClientServiceSubscription.current_due_date.asc(), ClientServiceSubscription.id.desc()).limit(limit)).scalars().all())
    rows: list[dict[str, Any]] = []
    for sub in subscriptions:
        task_stmt = select(ClientServiceTaskInstance).where(
            ClientServiceTaskInstance.subscription_id == sub.id,
            ClientServiceTaskInstance.is_active.is_(True),
        )
        tasks = list(db.execute(task_stmt.order_by(ClientServiceTaskInstance.sequence_no.asc(), ClientServiceTaskInstance.id.asc())).scalars().all())
        pending_partner = sum(1 for t in tasks if getattr(t, "aqmm_partner_review_required", False) and getattr(t, "partner_review_status", "not_required") != "reviewed")
        pending_review_partner = sum(1 for t in tasks if getattr(t, "aqmm_review_partner_required", False) and getattr(t, "review_partner_review_status", "not_required") != "reviewed")
        rework = sum(1 for t in tasks if getattr(t, "rework_status", "none") == "open")
        blockers = sum(1 for t in tasks if getattr(t, "aqmm_blocks_final_release", False) and getattr(t, "aqmm_status", "not_required") != "completed")
        exceptions = sum(1 for t in tasks if getattr(t, "checklist_response", None) == "NO")
        if not any((pending_partner, pending_review_partner, rework, blockers, exceptions)):
            continue
        weighted = _weighted_progress(tasks)
        sla = _engagement_sla(sub, tasks)
        rows.append({
            "subscription_id": sub.id,
            "client_name": getattr(getattr(sub, "client", None), "client_name", None) or "Unlinked Client",
            "service_name": getattr(getattr(sub, "catalogue", None), "service_name", None) or "Service",
            "financial_year": sub.financial_year or "-",
            "pending_partner": pending_partner,
            "pending_review_partner": pending_review_partner,
            "rework_count": rework,
            "release_blockers": blockers,
            "exception_count": exceptions,
            "weighted_progress": weighted,
            "sla": sla,
            "href": f"/partner/reviews/engagements/{sub.id}",
        })
    rows.sort(key=lambda r: (r["release_blockers"], r["rework_count"], r["pending_partner"] + r["pending_review_partner"]), reverse=True)
    return rows


def get_partner_review_workspace(db: Session, request, current_user, *, subscription_id: int, active_task_id: int | None = None) -> dict[str, Any] | None:
    roles = set(get_user_role_names(db, current_user.id))
    scope = _active_scope(db, request, current_user, roles)
    fy = _financial_year(request)
    stmt = (
        select(ClientServiceSubscription)
        .options(
            selectinload(ClientServiceSubscription.client),
            selectinload(ClientServiceSubscription.catalogue),
            selectinload(ClientServiceSubscription.assigned_partner),
            selectinload(ClientServiceSubscription.assigned_manager),
            selectinload(ClientServiceSubscription.assigned_staff),
            selectinload(ClientServiceSubscription.review_partner),
        )
        .where(ClientServiceSubscription.id == int(subscription_id))
    )
    stmt = _subscription_scope(stmt, scope["tenant_id"], scope["branch_id"], current_user, roles, fy)
    subscription = db.execute(stmt).scalar_one_or_none()
    if not subscription:
        return None
    if not roles.intersection({"System Admin", "Firm Admin"}) and int(current_user.id) not in {
        int(getattr(subscription, "assigned_partner_user_id", 0) or 0),
        int(getattr(subscription, "review_partner_user_id", 0) or 0),
    }:
        return None

    task_stmt = (
        select(ClientServiceTaskInstance)
        .options(
            selectinload(ClientServiceTaskInstance.assigned_to),
            selectinload(ClientServiceTaskInstance.comments).selectinload(ServiceTaskComment.created_by),
            selectinload(ClientServiceTaskInstance.documents),
        )
        .where(
            ClientServiceTaskInstance.subscription_id == subscription.id,
            ClientServiceTaskInstance.is_active.is_(True),
        )
        .order_by(ClientServiceTaskInstance.sequence_no.asc(), ClientServiceTaskInstance.id.asc())
    )
    tasks = list(db.execute(task_stmt).scalars().all())
    task_rows = [_partner_task_payload(task, subscription, current_user) for task in tasks]

    categories_by_name: dict[str, dict[str, Any]] = {}
    for row in task_rows:
        cat = categories_by_name.setdefault(row["category"], {
            "name": row["category"], "tasks": [], "total": 0, "pending": 0,
            "reviewed": 0, "rework": 0, "exceptions": 0, "blockers": 0,
        })
        cat["tasks"].append(row)
        cat["total"] += 1
        if row["review_state"] == "pending_review": cat["pending"] += 1
        if row["review_state"] == "reviewed": cat["reviewed"] += 1
        if row["review_state"] == "rework": cat["rework"] += 1
        if row["is_exception"]: cat["exceptions"] += 1
        if row["blocks_final_release"] and row["aqmm_status"] != "completed": cat["blockers"] += 1
    categories = list(categories_by_name.values())
    for cat in categories:
        denominator = cat["pending"] + cat["reviewed"] + cat["rework"]
        cat["progress_percent"] = int(round((cat["reviewed"] / denominator) * 100)) if denominator else 100

    actionable = [r for r in task_rows if r["can_review"]]
    active = next((r for r in task_rows if active_task_id and r["id"] == int(active_task_id)), None)
    if active is None:
        active = next((r for r in actionable if r["review_state"] == "pending_review"), None)
        if active is None:
            active = next((r for r in actionable if r["review_state"] in {"rework", "reviewed"}), None)
        if active is None and task_rows:
            active = task_rows[0]

    weighted = _weighted_progress(tasks)
    sla = _engagement_sla(subscription, tasks)
    team = _engagement_team(subscription)
    aqmm = aqmm_task_summary_for_subscription(db, subscription_id=subscription.id)
    closure = closure_readiness_for_subscription(db, subscription=subscription)
    release_ready = bool(
        closure["normal_tasks_completed"]
        and closure["aqmm_acceptance_completed"]
        and closure["aqmm_tasks_completed"]
        and closure["evidence_review_completed"]
        and closure["udin_completed"]
    )
    return {
        "subscription": subscription,
        "client_name": getattr(getattr(subscription, "client", None), "client_name", None) or "Unlinked Client",
        "service_name": getattr(getattr(subscription, "catalogue", None), "service_name", None) or "Service",
        "financial_year": subscription.financial_year or "-",
        "categories": categories,
        "tasks": task_rows,
        "active_task": active,
        "weighted_progress": weighted,
        "sla": sla,
        "team": team,
        "aqmm": aqmm,
        "closure": closure,
        "release_ready": release_ready,
        "is_locked": bool(getattr(subscription, "is_locked", False)),
        "pending_partner_count": sum(1 for r in task_rows if r["review_level"] == "partner" and r["review_state"] == "pending_review"),
        "pending_review_partner_count": sum(1 for r in task_rows if r["review_level"] == "review_partner" and r["review_state"] == "pending_review"),
        "rework_count": sum(1 for r in task_rows if r["review_state"] == "rework"),
        "exception_count": sum(1 for r in task_rows if r["is_exception"]),
        "release_blocker_count": len(aqmm.get("blockers") or []),
        "documents_href": f"/documents/engagements/{subscription.id}",
    }


def get_next_partner_review_task_id(workspace: dict[str, Any], current_task_id: int) -> int | None:
    rows = [r for r in workspace.get("tasks", []) if r.get("can_review") and r.get("review_state") == "pending_review"]
    if not rows:
        return None
    for index, row in enumerate(rows):
        if int(row["id"]) == int(current_task_id):
            return int(rows[index + 1]["id"]) if index + 1 < len(rows) else None
    return int(rows[0]["id"])
