from __future__ import annotations

from dataclasses import dataclass
from datetime import date, datetime, timezone, timedelta
from math import asin, cos, radians, sin, sqrt
from typing import Any
import ipaddress
from zoneinfo import ZoneInfo, ZoneInfoNotFoundError

from fastapi import HTTPException
from sqlalchemy import func, or_, select
from sqlalchemy.orm import Session, selectinload

from app.core.security.passwords import hash_password
from app.modules.core.iam.models import User
from app.modules.alerts.service import create_alert
from app.modules.alerts.workflow_escalations import create_workflow_escalation
from app.modules.core.iam.scope import build_scope, list_visible_branches, list_visible_tenants, validate_branch_matches_tenant
from app.modules.core.rbac.deps import get_user_roles
from app.modules.core.rbac.models import Role, UserRole
from app.modules.core.tenancy.models import Branch, Tenant
from app.modules.core.tenancy.settings_models import BranchSettings
from app.modules.employees.models import Employee, EmployeeAttendance, EmployeeRegistrationRequest, EmployeeLeaveType, EmployeeLeaveBalance, EmployeeLeaveRequest, EmployeeDocumentType, EmployeeDocument, EmployeeOnboardingChecklistItem, EmployeeOnboardingTask, EmployeeOffboardingRequest, EmployeeOffboardingTask, EmployeeSalaryStructure, EmployeePayrollRun, EmployeePayslip
from app.modules.clients.models import Client, ClientBusinessUnit, ClientBranch
from app.modules.documents.models import EngagementDocument
from app.modules.services.models import ClientServiceTaskInstance, ClientServiceSubscription, ServiceCatalogue, ServiceTaskComment
from app.modules.registrations.models import ClientRegistration, RegistrationType
from app.modules.services.execution import (
    CLOSED_TASK_STATUSES,
    TASK_PRIORITIES,
    TASK_STATUSES,
    apply_task_checklist_response,
    recalculate_task_aqmm_status,
    submit_task_for_review,
)

EMPLOYEE_STATUS = ["active", "inactive", "relieved"]
EMPLOYMENT_TYPES = ["full_time", "part_time", "article_assistant", "intern", "consultant", "contract"]
EMPLOYEE_ROLE_NAMES = ["Firm Admin", "Partner", "Branch Manager", "Staff"]

DOCUMENT_STATUS = ["uploaded", "verified", "rejected", "archived"]
DOCUMENT_VISIBILITY = ["employee_and_hr", "hr_only"]
ONBOARDING_TASK_STATUS = ["pending", "completed", "skipped"]
OFFBOARDING_REQUEST_STATUS = ["pending", "approved", "rejected", "completed", "cancelled"]
OFFBOARDING_TASK_STATUS = ["pending", "completed", "waived"]
PAYROLL_RUN_STATUS = ["draft", "generated", "approved", "paid", "cancelled"]
PAYSLIP_STATUS = ["generated", "approved", "paid", "cancelled"]

TASK_COMMUNICATION_TYPES = [
    ("internal_note", "Internal Note"),
    ("client_clarification", "Client Clarification"),
    ("consultant_communication", "Consultant Communication"),
    ("partner_review_note", "Partner Review Note"),
]

TASK_COMMUNICATION_VISIBILITIES = [
    ("internal", "Internal Team"),
    ("client_visible", "Client Visible Later"),
    ("consultant_visible", "Consultant Visible Later"),
    ("partner_review", "Partner / Review"),
]

TASK_COMMUNICATION_TYPE_CODES = {code for code, _ in TASK_COMMUNICATION_TYPES}

WORKFLOW_HOLD_REASONS = {
    "documents_from_client": "Documents awaited from client",
    "clarification_from_client": "Clarification awaited from client",
    "client_requested_hold": "Client requested hold",
    "manager_review": "Manager review pending",
    "partner_review": "Partner review pending",
    "review_partner_review": "Review Partner review pending",
    "payment_pending": "Payment pending",
    "portal_issue": "Portal or system issue",
    "internal_dependency": "Internal dependency",
    "other": "Other",
}

TASK_COMMUNICATION_VISIBILITY_CODES = {code for code, _ in TASK_COMMUNICATION_VISIBILITIES}

# Phase 4 derived controls deliberately reuse existing task, engagement, review and
# alert records. No duplicate workflow or capacity tables are introduced.
WORKFLOW_ESCALATION_LEVELS = {
    "manager": "Manager",
    "partner": "Engagement Partner",
    "review_partner": "Review Partner",
}



@dataclass
class EmployeeScope:
    tenant_id: int
    branch_id: int | None
    is_system_admin: bool
    is_firm_admin: bool
    is_partner: bool
    is_branch_manager: bool
    is_staff: bool
    allow_cross_tenant: bool
    allow_cross_branch: bool
    own_user_id: int


def _role_set(db: Session, user: User) -> set[str]:
    return set(get_user_roles(db, user.id))


def build_employee_scope(db: Session, user: User, *, tenant_id: int | None = None, branch_id: int | None = None) -> EmployeeScope:
    roles = _role_set(db, user)
    is_system_admin = "System Admin" in roles
    is_firm_admin = "Firm Admin" in roles
    is_partner = "Partner" in roles
    is_branch_manager = "Branch Manager" in roles
    is_staff = "Staff" in roles

    effective_tenant_id = int(tenant_id or user.tenant_id)
    effective_branch_id = branch_id

    if not is_system_admin:
        effective_tenant_id = int(user.tenant_id)

    # System Admin and Firm Admin may use all-branch context. Others stay locked to own branch.
    if not (is_system_admin or is_firm_admin):
        effective_branch_id = int(user.branch_id)
    elif effective_branch_id in (0, "0", "", None):
        effective_branch_id = None
    else:
        effective_branch_id = int(effective_branch_id)

    if effective_branch_id is not None:
        validate_branch_matches_tenant(db, effective_tenant_id, effective_branch_id)

    return EmployeeScope(
        tenant_id=effective_tenant_id,
        branch_id=effective_branch_id,
        is_system_admin=is_system_admin,
        is_firm_admin=is_firm_admin,
        is_partner=is_partner,
        is_branch_manager=is_branch_manager,
        is_staff=is_staff,
        allow_cross_tenant=is_system_admin,
        allow_cross_branch=is_system_admin or is_firm_admin,
        own_user_id=user.id,
    )


def list_employees(
    db: Session,
    scope: EmployeeScope,
    *,
    q: str = "",
    include_inactive: bool = False,
    link_status: str = "all",
) -> list[Employee]:
    stmt = select(Employee).where(Employee.tenant_id == scope.tenant_id)
    if scope.branch_id is not None:
        stmt = stmt.where(Employee.branch_id == scope.branch_id)
    if not include_inactive:
        stmt = stmt.where(Employee.is_active.is_(True))

    link_status = (link_status or "all").lower()
    if link_status == "linked":
        stmt = stmt.where(Employee.user_id.is_not(None))
    elif link_status == "unlinked":
        stmt = stmt.where(Employee.user_id.is_(None))

    if q:
        like = f"%{q.strip()}%"
        stmt = stmt.where(
            or_(
                Employee.employee_code.ilike(like),
                Employee.full_name.ilike(like),
                Employee.email.ilike(like),
                Employee.mobile.ilike(like),
                Employee.department.ilike(like),
                Employee.designation.ilike(like),
                Employee.pan.ilike(like),
            )
        )
    return db.execute(stmt.order_by(Employee.full_name, Employee.employee_code)).scalars().all()


def get_employee_or_404(db: Session, employee_id: int, scope: EmployeeScope) -> Employee:
    stmt = select(Employee).where(Employee.id == employee_id, Employee.tenant_id == scope.tenant_id)
    if scope.branch_id is not None:
        stmt = stmt.where(Employee.branch_id == scope.branch_id)
    emp = db.execute(stmt).scalar_one_or_none()
    if not emp:
        raise HTTPException(status_code=404, detail="Employee not found or not accessible.")
    return emp


def _blank_to_none(value: Any) -> Any:
    if value is None:
        return None
    if isinstance(value, str) and not value.strip():
        return None
    if isinstance(value, str):
        return value.strip()
    return value


def parse_date(value: Any) -> date | None:
    value = _blank_to_none(value)
    if not value:
        return None
    if isinstance(value, date):
        return value
    return date.fromisoformat(str(value))


def _clean_payload(data: dict[str, Any]) -> dict[str, Any]:
    cleaned = {k: _blank_to_none(v) for k, v in data.items()}
    for key in ("date_of_joining", "date_of_leaving"):
        cleaned[key] = parse_date(cleaned.get(key))
    status = (cleaned.get("status") or "active").lower()
    if status not in EMPLOYEE_STATUS:
        raise HTTPException(status_code=400, detail="Invalid employee status.")
    cleaned["status"] = status
    emp_type = (cleaned.get("employment_type") or "full_time").lower()
    if emp_type not in EMPLOYMENT_TYPES:
        raise HTTPException(status_code=400, detail="Invalid employment type.")
    cleaned["employment_type"] = emp_type
    cleaned["is_active"] = bool(cleaned.get("is_active", True)) and status == "active"
    return cleaned


def _ensure_unique(db: Session, *, tenant_id: int, employee_code: str, user_id: int | None, exclude_id: int | None = None) -> None:
    stmt = select(Employee).where(Employee.tenant_id == tenant_id, Employee.employee_code == employee_code)
    if exclude_id:
        stmt = stmt.where(Employee.id != exclude_id)
    if db.execute(stmt).scalar_one_or_none():
        raise HTTPException(status_code=409, detail="Employee code already exists in this tenant.")
    if user_id:
        stmt = select(Employee).where(Employee.tenant_id == tenant_id, Employee.user_id == user_id)
        if exclude_id:
            stmt = stmt.where(Employee.id != exclude_id)
        if db.execute(stmt).scalar_one_or_none():
            raise HTTPException(status_code=409, detail="Selected user is already linked to another employee.")


def _get_role(db: Session, role_name: str) -> Role | None:
    return db.execute(select(Role).where(Role.name == role_name, Role.is_active.is_(True))).scalar_one_or_none()


def _assign_role_if_needed(db: Session, user_id: int, role_name: str) -> None:
    role = _get_role(db, role_name)
    if not role:
        return
    exists = db.execute(select(UserRole).where(UserRole.user_id == user_id, UserRole.role_id == role.id)).scalar_one_or_none()
    if not exists:
        db.add(UserRole(user_id=user_id, role_id=role.id))


def create_login_user_for_employee(
    db: Session,
    *,
    tenant_id: int,
    branch_id: int,
    email: str,
    full_name: str,
    password: str,
    role_name: str = "Staff",
) -> User:
    if not email:
        raise HTTPException(status_code=400, detail="Login email is required to create an employee user.")
    if not password or len(password) < 8:
        raise HTTPException(status_code=400, detail="Temporary password must be at least 8 characters.")
    existing = db.execute(select(User).where(User.email == email)).scalar_one_or_none()
    if existing:
        raise HTTPException(status_code=409, detail="A user with this login email already exists. Select the existing user instead.")
    if role_name not in EMPLOYEE_ROLE_NAMES:
        role_name = "Staff"
    user = User(
        email=email.strip().lower(),
        full_name=full_name.strip(),
        password_hash=hash_password(password),
        tenant_id=tenant_id,
        branch_id=branch_id,
        is_active=True,
        allow_login=True,
        is_locked=False,
        deleted_at=None,
        must_change_password=True,
        password_changed_at_utc=None,
    )
    db.add(user)
    db.flush()
    _assign_role_if_needed(db, user.id, role_name)
    return user


def create_employee(db: Session, actor: User, scope: EmployeeScope, data: dict[str, Any]) -> Employee:
    cleaned = _clean_payload(data)
    partner_staff_mode = bool(scope.is_partner and not scope.is_system_admin and not scope.is_firm_admin)

    # Partners can create Staff or Branch Manager users only in their own
    # tenant and branch. Never trust tenant or branch values posted by the
    # browser for this workflow, and validate the requested role server-side.
    if partner_staff_mode:
        cleaned["tenant_id"] = scope.tenant_id
        cleaned["branch_id"] = scope.branch_id or actor.branch_id
        requested_role = (cleaned.get("employee_role") or "Staff").strip()
        if requested_role not in {"Staff", "Branch Manager"}:
            raise HTTPException(status_code=403, detail="Partners can create only Staff or Branch Manager users.")
        cleaned["employee_role"] = requested_role

    tenant_id = int(cleaned.get("tenant_id") or scope.tenant_id)
    branch_id = int(cleaned.get("branch_id") or actor.branch_id)

    if not scope.allow_cross_tenant and tenant_id != actor.tenant_id:
        raise HTTPException(status_code=403, detail="You cannot create employee in another tenant.")
    if not scope.allow_cross_branch and branch_id != actor.branch_id:
        raise HTTPException(status_code=403, detail="You cannot create employee in another branch.")
    validate_branch_matches_tenant(db, tenant_id, branch_id)

    employee_code = (cleaned.get("employee_code") or "").strip()
    full_name = (cleaned.get("full_name") or "").strip()
    if not employee_code or not full_name:
        raise HTTPException(status_code=400, detail="Employee code and full name are required.")

    user_id = cleaned.get("user_id")
    if user_id:
        linked_user = db.get(User, int(user_id))
        if not linked_user:
            raise HTTPException(status_code=404, detail="Selected user was not found.")
        if linked_user.tenant_id != tenant_id or linked_user.branch_id != branch_id:
            raise HTTPException(status_code=400, detail="Selected user must belong to the employee tenant and branch.")
        if partner_staff_mode:
            linked_roles = _role_set(db, linked_user)
            allowed_roles = {"Staff", "Branch Manager"}
            prohibited_roles = {"System Admin", "Firm Admin", "Partner"}
            if not linked_roles.intersection(allowed_roles) or linked_roles.intersection(prohibited_roles):
                raise HTTPException(status_code=403, detail="Partners can link only a Staff or Branch Manager login user.")
        user_id = linked_user.id
    elif cleaned.get("create_login_user"):
        login_user = create_login_user_for_employee(
            db,
            tenant_id=tenant_id,
            branch_id=branch_id,
            email=cleaned.get("login_email") or cleaned.get("email"),
            full_name=full_name,
            password=cleaned.get("temporary_password") or "",
            role_name=cleaned.get("employee_role") or "Staff",
        )
        user_id = login_user.id

    _ensure_unique(db, tenant_id=tenant_id, employee_code=employee_code, user_id=user_id)

    emp = Employee(
        tenant_id=tenant_id,
        branch_id=branch_id,
        user_id=user_id,
        employee_code=employee_code,
        full_name=full_name,
        email=cleaned.get("email") or cleaned.get("login_email"),
        mobile=cleaned.get("mobile"),
        alternate_mobile=cleaned.get("alternate_mobile"),
        date_of_joining=cleaned.get("date_of_joining"),
        date_of_leaving=cleaned.get("date_of_leaving"),
        employment_type=cleaned.get("employment_type") or "full_time",
        status=cleaned.get("status") or "active",
        is_active=cleaned.get("is_active", True),
        department=cleaned.get("department"),
        designation=cleaned.get("designation"),
        reporting_manager_user_id=cleaned.get("reporting_manager_user_id"),
        pan=cleaned.get("pan"),
        uan=cleaned.get("uan"),
        esi_no=cleaned.get("esi_no"),
        pf_no=cleaned.get("pf_no"),
        aadhaar_last4=cleaned.get("aadhaar_last4"),
        bank_name=cleaned.get("bank_name"),
        bank_account_no=cleaned.get("bank_account_no"),
        bank_ifsc=cleaned.get("bank_ifsc"),
        address=cleaned.get("address"),
        emergency_contact_name=cleaned.get("emergency_contact_name"),
        emergency_contact_mobile=cleaned.get("emergency_contact_mobile"),
        notes=cleaned.get("notes"),
        created_by_user_id=actor.id,
        updated_by_user_id=actor.id,
    )
    db.add(emp)
    db.commit()
    db.refresh(emp)
    return emp


def update_employee(db: Session, actor: User, emp: Employee, data: dict[str, Any]) -> Employee:
    cleaned = _clean_payload(data)
    employee_code = (cleaned.get("employee_code") or emp.employee_code).strip()
    full_name = (cleaned.get("full_name") or emp.full_name).strip()
    if not employee_code or not full_name:
        raise HTTPException(status_code=400, detail="Employee code and full name are required.")

    user_id = cleaned.get("user_id")
    if user_id:
        linked_user = db.get(User, int(user_id))
        if not linked_user:
            raise HTTPException(status_code=404, detail="Selected user was not found.")
        if linked_user.tenant_id != emp.tenant_id or linked_user.branch_id != emp.branch_id:
            raise HTTPException(status_code=400, detail="Selected user must belong to the employee tenant and branch.")
        user_id = linked_user.id
    else:
        user_id = None

    _ensure_unique(db, tenant_id=emp.tenant_id, employee_code=employee_code, user_id=user_id, exclude_id=emp.id)

    update_fields = [
        "employee_code", "full_name", "email", "mobile", "alternate_mobile", "date_of_joining", "date_of_leaving",
        "employment_type", "status", "is_active", "department", "designation", "reporting_manager_user_id",
        "pan", "uan", "esi_no", "pf_no", "aadhaar_last4", "bank_name", "bank_account_no", "bank_ifsc",
        "address", "emergency_contact_name", "emergency_contact_mobile", "notes",
    ]
    cleaned["employee_code"] = employee_code
    cleaned["full_name"] = full_name
    cleaned["user_id"] = user_id
    for field in update_fields + ["user_id"]:
        setattr(emp, field, cleaned.get(field))
    emp.updated_by_user_id = actor.id
    emp.updated_at_utc = datetime.now(timezone.utc)
    db.commit()
    db.refresh(emp)
    return emp


def change_employee_status(db: Session, actor: User, emp: Employee, status: str, date_of_leaving: date | None = None) -> Employee:
    status = (status or "").lower()
    if status not in EMPLOYEE_STATUS:
        raise HTTPException(status_code=400, detail="Invalid employee status.")
    emp.status = status
    emp.is_active = status == "active"
    if status == "relieved" and date_of_leaving:
        emp.date_of_leaving = date_of_leaving
    emp.updated_by_user_id = actor.id
    emp.updated_at_utc = datetime.now(timezone.utc)
    db.commit()
    db.refresh(emp)
    return emp


def visible_tenants(db: Session, user: User) -> list[Tenant]:
    return list_visible_tenants(db, build_scope(db, user))


def visible_branches(db: Session, user: User, tenant_id: int | None = None) -> list[Branch]:
    return list_visible_branches(db, build_scope(db, user), tenant_id or user.tenant_id)


def list_linkable_users(db: Session, scope: EmployeeScope, *, include_user_id: int | None = None) -> list[User]:
    """Return active login users that can be linked to an employee.

    Already-linked users are excluded to avoid accidental duplicate linkage.
    When editing an employee, include_user_id keeps that employee's current user
    visible in the dropdown.
    """
    linked_user_ids = set(
        db.execute(
            select(Employee.user_id).where(
                Employee.tenant_id == scope.tenant_id,
                Employee.user_id.is_not(None),
            )
        ).scalars().all()
    )
    if include_user_id:
        linked_user_ids.discard(int(include_user_id))

    stmt = select(User).where(
        User.tenant_id == scope.tenant_id,
        User.deleted_at.is_(None),
        User.is_active.is_(True),
    )
    if scope.branch_id is not None:
        stmt = stmt.where(User.branch_id == scope.branch_id)
    if linked_user_ids:
        stmt = stmt.where(User.id.not_in(linked_user_ids))
    return db.execute(stmt.order_by(User.full_name, User.email)).scalars().all()


def get_employee_user_link_summary(db: Session, scope: EmployeeScope) -> dict[str, int]:
    base = select(Employee).where(Employee.tenant_id == scope.tenant_id)
    if scope.branch_id is not None:
        base = base.where(Employee.branch_id == scope.branch_id)
    rows = db.execute(base).scalars().all()
    linked = sum(1 for emp in rows if emp.user_id)
    unlinked = len(rows) - linked
    return {"total": len(rows), "linked": linked, "unlinked": unlinked}


def link_employee_to_user(db: Session, actor: User, emp: Employee, user_id: int | None) -> Employee:
    """Link or unlink an employee master with an IAM login user.

    This is intentionally a narrow helper for Phase 7A.1 UX cleanup. It does
    not create users or change roles; it only updates Employee.user_id after
    validating tenant/branch and duplicate linkage.
    """
    resolved_user_id = None
    if user_id:
        linked_user = db.get(User, int(user_id))
        if not linked_user or linked_user.deleted_at is not None:
            raise HTTPException(status_code=404, detail="Selected user was not found or is inactive.")
        if int(linked_user.tenant_id) != int(emp.tenant_id) or int(linked_user.branch_id) != int(emp.branch_id):
            raise HTTPException(status_code=400, detail="Selected user must belong to the employee tenant and branch.")
        resolved_user_id = linked_user.id

    _ensure_unique(db, tenant_id=emp.tenant_id, employee_code=emp.employee_code, user_id=resolved_user_id, exclude_id=emp.id)
    emp.user_id = resolved_user_id
    emp.updated_by_user_id = actor.id
    emp.updated_at_utc = datetime.now(timezone.utc)
    db.commit()
    db.refresh(emp)
    return emp


def list_reporting_managers(db: Session, scope: EmployeeScope) -> list[User]:
    """Return only active Partner and Branch Manager users in scope.

    Client and other non-employee login roles must never appear in the
    Reporting Manager dropdown.
    """
    stmt = (
        select(User)
        .join(UserRole, UserRole.user_id == User.id)
        .join(Role, Role.id == UserRole.role_id)
        .where(
            User.tenant_id == scope.tenant_id,
            User.deleted_at.is_(None),
            User.is_active.is_(True),
            Role.is_active.is_(True),
            Role.name.in_(("Partner", "Branch Manager")),
        )
        .distinct()
    )
    if scope.branch_id is not None:
        stmt = stmt.where(User.branch_id == scope.branch_id)
    return db.execute(stmt.order_by(User.full_name, User.email)).scalars().all()


def _count_for_scope(db: Session, model, scope: EmployeeScope, *conditions) -> int:
    stmt = select(func.count(model.id)).where(model.tenant_id == scope.tenant_id)
    if scope.branch_id is not None and hasattr(model, "branch_id"):
        stmt = stmt.where(model.branch_id == scope.branch_id)
    for condition in conditions:
        stmt = stmt.where(condition)
    return int(db.execute(stmt).scalar() or 0)


def get_employee_dashboard_stats(db: Session, scope: EmployeeScope) -> dict[str, Any]:
    """Return HR dashboard and report counters for the active tenant/branch context.

    Phase 6H intentionally adds reporting only. It does not change any existing
    employee, attendance, leave, document, onboarding, offboarding or payroll
    workflows.
    """
    today = date.today()
    month_start = today.replace(day=1)

    total_employees = _count_for_scope(db, Employee, scope)
    active_employees = _count_for_scope(db, Employee, scope, Employee.status == "active", Employee.is_active.is_(True))
    inactive_employees = _count_for_scope(db, Employee, scope, Employee.status == "inactive")
    relieved_employees = _count_for_scope(db, Employee, scope, Employee.status == "relieved")

    pending_registrations = _count_for_scope(db, EmployeeRegistrationRequest, scope, EmployeeRegistrationRequest.status == "pending")

    attendance_today_total = _count_for_scope(db, EmployeeAttendance, scope, EmployeeAttendance.attendance_date == today)
    attendance_today_present = _count_for_scope(db, EmployeeAttendance, scope, EmployeeAttendance.attendance_date == today, EmployeeAttendance.status == "present")
    attendance_today_pending = _count_for_scope(db, EmployeeAttendance, scope, EmployeeAttendance.attendance_date == today, EmployeeAttendance.approval_status == "pending")
    attendance_month_total = _count_for_scope(db, EmployeeAttendance, scope, EmployeeAttendance.attendance_date >= month_start)

    pending_leave_requests = _count_for_scope(db, EmployeeLeaveRequest, scope, EmployeeLeaveRequest.status == "pending")
    approved_leave_requests = _count_for_scope(db, EmployeeLeaveRequest, scope, EmployeeLeaveRequest.status == "approved")
    rejected_leave_requests = _count_for_scope(db, EmployeeLeaveRequest, scope, EmployeeLeaveRequest.status == "rejected")

    uploaded_documents = _count_for_scope(db, EmployeeDocument, scope, EmployeeDocument.status == "uploaded")
    verified_documents = _count_for_scope(db, EmployeeDocument, scope, EmployeeDocument.status == "verified")
    rejected_documents = _count_for_scope(db, EmployeeDocument, scope, EmployeeDocument.status == "rejected")

    onboarding_pending = _count_for_scope(db, EmployeeOnboardingTask, scope, EmployeeOnboardingTask.status == "pending")
    onboarding_completed = _count_for_scope(db, EmployeeOnboardingTask, scope, EmployeeOnboardingTask.status == "completed")

    offboarding_pending = _count_for_scope(db, EmployeeOffboardingRequest, scope, EmployeeOffboardingRequest.status == "pending")
    offboarding_approved = _count_for_scope(db, EmployeeOffboardingRequest, scope, EmployeeOffboardingRequest.status == "approved")

    salary_structures_active = _count_for_scope(db, EmployeeSalaryStructure, scope, EmployeeSalaryStructure.is_active.is_(True))
    payroll_runs_draft = _count_for_scope(db, EmployeePayrollRun, scope, EmployeePayrollRun.status == "draft")
    payroll_runs_generated = _count_for_scope(db, EmployeePayrollRun, scope, EmployeePayrollRun.status == "generated")
    payroll_runs_approved = _count_for_scope(db, EmployeePayrollRun, scope, EmployeePayrollRun.status == "approved")
    payroll_runs_paid = _count_for_scope(db, EmployeePayrollRun, scope, EmployeePayrollRun.status == "paid")
    payslips_generated = _count_for_scope(db, EmployeePayslip, scope, EmployeePayslip.status == "generated")
    payslips_paid = _count_for_scope(db, EmployeePayslip, scope, EmployeePayslip.status == "paid")

    recent_employees_stmt = select(Employee).where(Employee.tenant_id == scope.tenant_id)
    if scope.branch_id is not None:
        recent_employees_stmt = recent_employees_stmt.where(Employee.branch_id == scope.branch_id)
    recent_employees = db.execute(
        recent_employees_stmt.order_by(Employee.created_at_utc.desc()).limit(8)
    ).scalars().all()

    recent_leave_stmt = select(EmployeeLeaveRequest).options(selectinload(EmployeeLeaveRequest.employee), selectinload(EmployeeLeaveRequest.leave_type)).where(EmployeeLeaveRequest.tenant_id == scope.tenant_id)
    if scope.branch_id is not None:
        recent_leave_stmt = recent_leave_stmt.where(EmployeeLeaveRequest.branch_id == scope.branch_id)
    recent_leave_requests = db.execute(
        recent_leave_stmt.order_by(EmployeeLeaveRequest.created_at_utc.desc()).limit(8)
    ).scalars().all()

    recent_offboarding_stmt = select(EmployeeOffboardingRequest).options(selectinload(EmployeeOffboardingRequest.employee)).where(EmployeeOffboardingRequest.tenant_id == scope.tenant_id)
    if scope.branch_id is not None:
        recent_offboarding_stmt = recent_offboarding_stmt.where(EmployeeOffboardingRequest.branch_id == scope.branch_id)
    recent_offboarding_requests = db.execute(
        recent_offboarding_stmt.order_by(EmployeeOffboardingRequest.created_at_utc.desc()).limit(8)
    ).scalars().all()

    return {
        "as_on": today,
        "month_start": month_start,
        "employees": {
            "total": total_employees,
            "active": active_employees,
            "inactive": inactive_employees,
            "relieved": relieved_employees,
            "pending_registrations": pending_registrations,
        },
        "attendance": {
            "today_total": attendance_today_total,
            "today_present": attendance_today_present,
            "today_pending": attendance_today_pending,
            "month_total": attendance_month_total,
        },
        "leave": {
            "pending": pending_leave_requests,
            "approved": approved_leave_requests,
            "rejected": rejected_leave_requests,
        },
        "documents": {
            "uploaded": uploaded_documents,
            "verified": verified_documents,
            "rejected": rejected_documents,
        },
        "onboarding": {
            "pending": onboarding_pending,
            "completed": onboarding_completed,
        },
        "offboarding": {
            "pending": offboarding_pending,
            "approved": offboarding_approved,
        },
        "payroll": {
            "salary_structures_active": salary_structures_active,
            "runs_draft": payroll_runs_draft,
            "runs_generated": payroll_runs_generated,
            "runs_approved": payroll_runs_approved,
            "runs_paid": payroll_runs_paid,
            "payslips_generated": payslips_generated,
            "payslips_paid": payslips_paid,
        },
        "recent_employees": recent_employees,
        "recent_leave_requests": recent_leave_requests,
        "recent_offboarding_requests": recent_offboarding_requests,
    }


REGISTRATION_STATUS = ["pending", "approved", "rejected"]


def get_employee_for_user(db: Session, user: User) -> Employee | None:
    return db.execute(
        select(Employee).where(
            Employee.tenant_id == user.tenant_id,
            Employee.user_id == user.id,
        )
    ).scalar_one_or_none()


def update_own_employee_profile(db: Session, actor: User, emp: Employee, data: dict[str, Any]) -> Employee:
    """Allow employees to update only safe self-service fields."""
    if emp.user_id != actor.id:
        raise HTTPException(status_code=403, detail="You can update only your own employee profile.")
    allowed_fields = [
        "mobile",
        "alternate_mobile",
        "address",
        "emergency_contact_name",
        "emergency_contact_mobile",
        "bank_name",
        "bank_account_no",
        "bank_ifsc",
    ]
    cleaned = {k: _blank_to_none(data.get(k)) for k in allowed_fields}
    for field, value in cleaned.items():
        setattr(emp, field, value)
    emp.updated_by_user_id = actor.id
    emp.updated_at_utc = datetime.now(timezone.utc)
    db.commit()
    db.refresh(emp)
    return emp


def _next_employee_code(db: Session, tenant_id: int) -> str:
    prefix = "EMP"
    latest = db.execute(
        select(Employee.employee_code)
        .where(Employee.tenant_id == tenant_id, Employee.employee_code.ilike(f"{prefix}%"))
        .order_by(Employee.id.desc())
        .limit(1)
    ).scalar_one_or_none()
    if not latest:
        return "EMP0001"
    digits = "".join(ch for ch in str(latest) if ch.isdigit())
    next_no = (int(digits) + 1) if digits else 1
    return f"EMP{next_no:04d}"


def get_pending_registration_for_user(db: Session, user: User):
    from app.modules.employees.models import Employee, EmployeeAttendance, EmployeeRegistrationRequest, EmployeeLeaveType, EmployeeLeaveBalance, EmployeeRegistrationRequest

    return db.execute(
        select(EmployeeRegistrationRequest).where(
            EmployeeRegistrationRequest.tenant_id == user.tenant_id,
            EmployeeRegistrationRequest.user_id == user.id,
            EmployeeRegistrationRequest.status == "pending",
        )
    ).scalar_one_or_none()


def create_employee_registration_request(db: Session, actor: User, data: dict[str, Any]):
    from app.modules.employees.models import Employee, EmployeeAttendance, EmployeeRegistrationRequest, EmployeeLeaveType, EmployeeLeaveBalance, EmployeeRegistrationRequest

    if get_employee_for_user(db, actor):
        raise HTTPException(status_code=400, detail="Your user is already linked to an employee profile.")
    if get_pending_registration_for_user(db, actor):
        raise HTTPException(status_code=409, detail="A pending employee registration request already exists for your user.")
    full_name = (_blank_to_none(data.get("full_name")) or actor.full_name or actor.email).strip()
    req = EmployeeRegistrationRequest(
        tenant_id=actor.tenant_id,
        branch_id=actor.branch_id,
        user_id=actor.id,
        requested_employee_code=_blank_to_none(data.get("requested_employee_code")),
        full_name=full_name,
        email=_blank_to_none(data.get("email")) or actor.email,
        mobile=_blank_to_none(data.get("mobile")),
        department=_blank_to_none(data.get("department")),
        designation=_blank_to_none(data.get("designation")),
        date_of_joining=parse_date(data.get("date_of_joining")),
        remarks=_blank_to_none(data.get("remarks")),
        status="pending",
    )
    db.add(req)
    db.commit()
    db.refresh(req)
    return req


def list_employee_registration_requests(db: Session, scope: EmployeeScope, *, status: str | None = None):
    from app.modules.employees.models import Employee, EmployeeAttendance, EmployeeRegistrationRequest, EmployeeLeaveType, EmployeeLeaveBalance, EmployeeRegistrationRequest

    stmt = select(EmployeeRegistrationRequest).where(EmployeeRegistrationRequest.tenant_id == scope.tenant_id)
    if scope.branch_id is not None:
        stmt = stmt.where(EmployeeRegistrationRequest.branch_id == scope.branch_id)
    if status:
        stmt = stmt.where(EmployeeRegistrationRequest.status == status)
    return db.execute(stmt.order_by(EmployeeRegistrationRequest.created_at_utc.desc())).scalars().all()


def get_employee_registration_request_or_404(db: Session, request_id: int, scope: EmployeeScope):
    from app.modules.employees.models import Employee, EmployeeAttendance, EmployeeRegistrationRequest, EmployeeLeaveType, EmployeeLeaveBalance, EmployeeRegistrationRequest

    stmt = select(EmployeeRegistrationRequest).where(
        EmployeeRegistrationRequest.id == request_id,
        EmployeeRegistrationRequest.tenant_id == scope.tenant_id,
    )
    if scope.branch_id is not None:
        stmt = stmt.where(EmployeeRegistrationRequest.branch_id == scope.branch_id)
    row = db.execute(stmt).scalar_one_or_none()
    if not row:
        raise HTTPException(status_code=404, detail="Employee registration request not found or not accessible.")
    return row


def approve_employee_registration_request(db: Session, actor: User, req, *, employee_code: str | None = None, notes: str | None = None) -> Employee:
    if req.status != "pending":
        raise HTTPException(status_code=400, detail="Only pending registration requests can be approved.")
    existing_emp = db.execute(
        select(Employee).where(Employee.tenant_id == req.tenant_id, Employee.user_id == req.user_id)
    ).scalar_one_or_none()
    if existing_emp:
        req.status = "approved"
        req.review_notes = notes
        req.reviewed_by_user_id = actor.id
        req.reviewed_at_utc = datetime.now(timezone.utc)
        req.created_employee_id = existing_emp.id
        db.commit()
        db.refresh(existing_emp)
        return existing_emp
    code = (employee_code or req.requested_employee_code or _next_employee_code(db, req.tenant_id)).strip()
    _ensure_unique(db, tenant_id=req.tenant_id, employee_code=code, user_id=req.user_id)
    emp = Employee(
        tenant_id=req.tenant_id,
        branch_id=req.branch_id,
        user_id=req.user_id,
        employee_code=code,
        full_name=req.full_name,
        email=req.email,
        mobile=req.mobile,
        date_of_joining=req.date_of_joining,
        employment_type="full_time",
        status="active",
        is_active=True,
        department=req.department,
        designation=req.designation,
        notes=req.remarks,
        created_by_user_id=actor.id,
        updated_by_user_id=actor.id,
    )
    db.add(emp)
    db.flush()
    req.status = "approved"
    req.review_notes = notes
    req.reviewed_by_user_id = actor.id
    req.reviewed_at_utc = datetime.now(timezone.utc)
    req.created_employee_id = emp.id
    req.updated_at_utc = datetime.now(timezone.utc)
    db.commit()
    db.refresh(emp)
    return emp


def reject_employee_registration_request(db: Session, actor: User, req, *, notes: str | None = None):
    if req.status != "pending":
        raise HTTPException(status_code=400, detail="Only pending registration requests can be rejected.")
    req.status = "rejected"
    req.review_notes = _blank_to_none(notes)
    req.reviewed_by_user_id = actor.id
    req.reviewed_at_utc = datetime.now(timezone.utc)
    req.updated_at_utc = datetime.now(timezone.utc)
    db.commit()
    db.refresh(req)
    return req



ATTENDANCE_STATUS = ["present", "late", "absent", "half_day", "on_duty", "work_from_home"]
ATTENDANCE_APPROVAL_STATUS = ["pending", "approved", "rejected"]
ATTENDANCE_GEO_STATUS = ["not_configured", "inside_geofence", "outside_geofence", "location_missing", "invalid_location"]
ATTENDANCE_IP_STATUS = ["not_configured", "allowed_ip", "outside_allowed_ip", "ip_missing", "invalid_ip_rule"]
ATTENDANCE_RULE_STATUS = ["not_configured", "within_time", "within_grace", "late", "half_day", "weekly_off"]
WEEKDAY_CODES = ["MON", "TUE", "WED", "THU", "FRI", "SAT", "SUN"]


def _safe_timezone(tz_name: str | None) -> ZoneInfo:
    name = (tz_name or "Asia/Kolkata").strip() or "Asia/Kolkata"
    try:
        return ZoneInfo(name)
    except ZoneInfoNotFoundError:
        return ZoneInfo("Asia/Kolkata")


def _utc_now_naive() -> datetime:
    """Return UTC now as a naive datetime for SQLite-safe storage.

    The column names ending with `_utc` are still UTC values. Keeping them
    naive avoids SQLite/SQLAlchemy timezone stripping inconsistencies. Branch
    local display times are stored separately in *_local_at fields.
    """
    return datetime.now(timezone.utc).replace(tzinfo=None)


def _as_utc_aware(value: datetime | None) -> datetime | None:
    if value is None:
        return None
    if value.tzinfo is None:
        return value.replace(tzinfo=timezone.utc)
    return value.astimezone(timezone.utc)


def _branch_local_naive_from_utc(value: datetime | None, tz_name: str | None) -> datetime | None:
    utc_value = _as_utc_aware(value)
    if utc_value is None:
        return None
    return utc_value.astimezone(_safe_timezone(tz_name)).replace(tzinfo=None)


def _refresh_attendance_local_evidence(row: "EmployeeAttendance") -> "EmployeeAttendance":
    """Keep displayed punch time tied to the branch timezone snapshot.

    This also fixes old rows where local fields were missing or were saved
    incorrectly after timezone-related changes. It updates the in-memory row;
    callers may commit only when they intentionally edit attendance.
    """
    tz_name = row.branch_timezone or "Asia/Kolkata"
    if row.punch_in_utc:
        row.punch_in_local_at = _branch_local_naive_from_utc(row.punch_in_utc, tz_name)
    if row.punch_out_utc:
        row.punch_out_local_at = _branch_local_naive_from_utc(row.punch_out_utc, tz_name)
    return row


def _branch_settings(db: Session, branch_id: int) -> tuple[Branch | None, BranchSettings | None]:
    branch = db.execute(select(Branch).where(Branch.id == int(branch_id))).scalar_one_or_none()
    settings = db.execute(select(BranchSettings).where(BranchSettings.branch_id == int(branch_id))).scalar_one_or_none()
    return branch, settings


def _branch_now(db: Session, branch_id: int) -> tuple[datetime, datetime, date, str, Branch | None, BranchSettings | None]:
    branch, settings = _branch_settings(db, branch_id)
    tz_name = getattr(branch, "timezone", None) or "Asia/Kolkata"
    now_utc = _utc_now_naive()
    local_dt = _branch_local_naive_from_utc(now_utc, tz_name) or now_utc
    return now_utc, local_dt, local_dt.date(), tz_name, branch, settings


def _working_day_codes(settings: BranchSettings | None) -> set[str]:
    raw = (getattr(settings, "working_days_csv", None) or "MON,TUE,WED,THU,FRI,SAT").strip()
    return {x.strip().upper() for x in raw.split(',') if x.strip()}


def _evaluate_attendance_timing(branch: Branch | None, settings: BranchSettings | None, local_dt: datetime) -> dict[str, Any]:
    rule_enabled = bool(getattr(settings, "attendance_rule_enabled", True)) if settings else True
    start_time = getattr(branch, "office_start_time", None) if branch else None
    end_time = getattr(branch, "office_end_time", None) if branch else None
    grace_minutes = int(getattr(settings, "attendance_grace_minutes", 10) or 0) if settings else 10
    half_day_after = getattr(settings, "attendance_half_day_after_time", None) if settings else None

    weekday_code = WEEKDAY_CODES[local_dt.weekday()]
    is_weekly_off = weekday_code not in _working_day_codes(settings)

    result = {
        "status": "present",
        "rule_status": "within_time",
        "late_by_minutes": None,
        "scheduled_start_local": start_time,
        "scheduled_end_local": end_time,
        "is_weekly_off": is_weekly_off,
    }

    if not rule_enabled:
        result["rule_status"] = "not_configured"
        return result
    if is_weekly_off:
        result["rule_status"] = "weekly_off"
        return result
    if not start_time:
        result["rule_status"] = "not_configured"
        return result

    local_time = local_dt.time().replace(tzinfo=None)
    start_dt = datetime.combine(local_dt.date(), start_time)
    local_naive = local_dt.replace(tzinfo=None)
    late_by = max(int((local_naive - start_dt).total_seconds() // 60), 0)
    result["late_by_minutes"] = late_by if late_by else None

    if half_day_after and local_time >= half_day_after:
        result["status"] = "half_day"
        result["rule_status"] = "half_day"
    elif late_by > grace_minutes:
        result["status"] = "late"
        result["rule_status"] = "late"
    elif late_by > 0:
        result["status"] = "present"
        result["rule_status"] = "within_grace"
    else:
        result["status"] = "present"
        result["rule_status"] = "within_time"
    return result


def _attendance_duration_minutes(row: EmployeeAttendance) -> int | None:
    if not row.punch_in_utc or not row.punch_out_utc:
        return None

    punch_in_utc = _as_utc_aware(row.punch_in_utc)
    punch_out_utc = _as_utc_aware(row.punch_out_utc)
    if not punch_in_utc or not punch_out_utc:
        return None

    delta = punch_out_utc - punch_in_utc
    minutes = int(delta.total_seconds() // 60)
    return max(minutes, 0)


def _to_float(value: float | str | None) -> float | None:
    if value is None or value == "":
        return None
    try:
        return float(value)
    except (TypeError, ValueError):
        return None


def _haversine_distance_meters(lat1: float, lon1: float, lat2: float, lon2: float) -> float:
    radius_m = 6371000.0
    lat1_r, lon1_r, lat2_r, lon2_r = map(radians, [lat1, lon1, lat2, lon2])
    dlat = lat2_r - lat1_r
    dlon = lon2_r - lon1_r
    a = sin(dlat / 2) ** 2 + cos(lat1_r) * cos(lat2_r) * sin(dlon / 2) ** 2
    c = 2 * asin(sqrt(a))
    return radius_m * c


def _ip_matches_allowed(ip_value: str | None, allowed_csv: str | None) -> tuple[bool, str]:
    if not allowed_csv or not allowed_csv.strip():
        return False, "not_configured"
    if not ip_value or not ip_value.strip():
        return False, "ip_missing"
    try:
        client_ip = ipaddress.ip_address(ip_value.strip())
    except ValueError:
        return False, "ip_missing"

    invalid_rule_found = False
    for raw_rule in allowed_csv.split(','):
        rule = raw_rule.strip()
        if not rule:
            continue
        try:
            if '/' in rule:
                if client_ip in ipaddress.ip_network(rule, strict=False):
                    return True, "allowed_ip"
            elif client_ip == ipaddress.ip_address(rule):
                return True, "allowed_ip"
        except ValueError:
            invalid_rule_found = True
    return False, "invalid_ip_rule" if invalid_rule_found else "outside_allowed_ip"


def _attendance_geo_is_enforced(settings: BranchSettings | None) -> bool:
    """Return True when branch GPS geofence is configured and must be captured by browser."""
    if not settings or not bool(getattr(settings, "attendance_geo_enabled", False)):
        return False
    return _to_float(getattr(settings, "latitude", None)) is not None and _to_float(getattr(settings, "longitude", None)) is not None


def _raise_if_required_geo_missing(settings: BranchSettings | None, evaluation: dict[str, Any]) -> None:
    """Block self-attendance when geofencing is enabled but browser location is missing/invalid."""
    if not _attendance_geo_is_enforced(settings):
        return
    if evaluation.get("geo_status") in {"location_missing", "invalid_location"}:
        raise HTTPException(
            status_code=400,
            detail=(
                "Location is required for attendance because branch geo-fencing is enabled. "
                "Please allow browser location/GPS and try again."
            ),
        )


def _evaluate_attendance_controls(
    db: Session,
    *,
    branch_id: int,
    latitude: float | str | None = None,
    longitude: float | str | None = None,
    client_ip: str | None = None,
) -> dict[str, Any]:
    settings = db.execute(select(BranchSettings).where(BranchSettings.branch_id == branch_id)).scalar_one_or_none()
    lat = _to_float(latitude)
    lon = _to_float(longitude)

    geo_enabled = bool(getattr(settings, 'attendance_geo_enabled', False)) if settings else False
    ip_enabled = bool(getattr(settings, 'attendance_ip_enabled', False)) if settings else False
    branch_lat = _to_float(getattr(settings, 'latitude', None)) if settings else None
    branch_lon = _to_float(getattr(settings, 'longitude', None)) if settings else None
    radius_m = int(getattr(settings, 'attendance_geo_radius_meters', 100) or 100) if settings else 100

    distance_m = None
    geo_ok = False
    if not geo_enabled:
        geo_status = "not_configured"
    elif branch_lat is None or branch_lon is None:
        geo_status = "not_configured"
    elif lat is None or lon is None:
        geo_status = "location_missing"
    else:
        distance_m = round(_haversine_distance_meters(branch_lat, branch_lon, lat, lon), 2)
        geo_ok = distance_m <= radius_m
        geo_status = "inside_geofence" if geo_ok else "outside_geofence"

    if ip_enabled:
        ip_ok, ip_status = _ip_matches_allowed(client_ip, getattr(settings, 'attendance_allowed_ip_csv', None) if settings else None)
    else:
        ip_ok, ip_status = False, "not_configured"

    if not geo_enabled and not ip_enabled:
        approval_status = "approved"
        status = "present"
        source = "self_punch"
    elif geo_ok or ip_ok:
        approval_status = "approved"
        status = "present"
        source = "geo_ip_punch" if geo_ok and ip_ok else ("geo_punch" if geo_ok else "ip_punch")
    else:
        approval_status = "pending"
        status = "on_duty"
        source = "od_request"

    return {
        "approval_status": approval_status,
        "status": status,
        "source": source,
        "geo_status": geo_status,
        "ip_status": ip_status,
        "distance_meters": distance_m,
    }


def get_today_attendance_for_user(db: Session, actor: User) -> EmployeeAttendance | None:
    emp = get_employee_for_user(db, actor)
    if not emp:
        return None
    _now_utc, _local_dt, local_date, _tz_name, _branch, _settings = _branch_now(db, emp.branch_id)
    row = db.execute(
        select(EmployeeAttendance).where(
            EmployeeAttendance.tenant_id == emp.tenant_id,
            EmployeeAttendance.employee_id == emp.id,
            EmployeeAttendance.attendance_date == local_date,
        )
    ).scalar_one_or_none()
    return _refresh_attendance_local_evidence(row) if row else None


def list_own_attendance(db: Session, actor: User, *, limit: int = 60) -> list[EmployeeAttendance]:
    emp = get_employee_for_user(db, actor)
    if not emp:
        return []
    rows = db.execute(
        select(EmployeeAttendance)
        .where(EmployeeAttendance.tenant_id == emp.tenant_id, EmployeeAttendance.employee_id == emp.id)
        .order_by(EmployeeAttendance.attendance_date.desc(), EmployeeAttendance.id.desc())
        .limit(limit)
    ).scalars().all()
    return [_refresh_attendance_local_evidence(row) for row in rows]


def punch_in_attendance(
    db: Session,
    actor: User,
    *,
    remarks: str | None = None,
    latitude: float | str | None = None,
    longitude: float | str | None = None,
    accuracy_meters: float | str | None = None,
    client_ip: str | None = None,
) -> EmployeeAttendance:
    emp = get_employee_for_user(db, actor)
    if not emp:
        raise HTTPException(status_code=400, detail="Your user is not linked to an employee profile. Please request employee registration first.")
    if not emp.is_active or emp.status != "active":
        raise HTTPException(status_code=400, detail="Attendance punch is allowed only for active employees.")
    now, local_dt, today, branch_tz, branch, branch_settings = _branch_now(db, emp.branch_id)
    existing = db.execute(
        select(EmployeeAttendance).where(
            EmployeeAttendance.tenant_id == emp.tenant_id,
            EmployeeAttendance.employee_id == emp.id,
            EmployeeAttendance.attendance_date == today,
        )
    ).scalar_one_or_none()
    if existing and existing.punch_in_utc:
        raise HTTPException(status_code=409, detail="You have already punched in today.")
    evaluation = _evaluate_attendance_controls(
        db,
        branch_id=emp.branch_id,
        latitude=latitude,
        longitude=longitude,
        client_ip=client_ip,
    )
    _raise_if_required_geo_missing(branch_settings, evaluation)
    timing = _evaluate_attendance_timing(branch, branch_settings, local_dt)
    final_status = evaluation["status"] if evaluation["approval_status"] == "pending" else timing["status"]
    punch_remarks = _blank_to_none(remarks)
    if evaluation["approval_status"] == "pending" and not punch_remarks:
        punch_remarks = "Outside branch geofence / office IP. Approval required for OD, client visit or remote duty."

    if not existing:
        existing = EmployeeAttendance(
            tenant_id=emp.tenant_id,
            branch_id=emp.branch_id,
            employee_id=emp.id,
            user_id=actor.id,
            attendance_date=today,
            punch_in_utc=now,
            punch_in_local_at=local_dt,
            branch_timezone=branch_tz,
            scheduled_start_local=timing["scheduled_start_local"],
            scheduled_end_local=timing["scheduled_end_local"],
            late_by_minutes=timing["late_by_minutes"],
            attendance_rule_status=timing["rule_status"],
            is_weekly_off=timing["is_weekly_off"],
            status=final_status,
            approval_status=evaluation["approval_status"],
            source=evaluation["source"],
            remarks=punch_remarks,
            punch_in_latitude=_to_float(latitude),
            punch_in_longitude=_to_float(longitude),
            punch_in_accuracy_meters=_to_float(accuracy_meters),
            punch_in_distance_meters=evaluation["distance_meters"],
            punch_in_ip=client_ip,
            punch_in_geo_status=evaluation["geo_status"],
            punch_in_ip_status=evaluation["ip_status"],
            created_by_user_id=actor.id,
            updated_by_user_id=actor.id,
        )
        db.add(existing)
    else:
        existing.punch_in_utc = now
        existing.punch_in_local_at = local_dt
        existing.branch_timezone = branch_tz
        existing.scheduled_start_local = timing["scheduled_start_local"]
        existing.scheduled_end_local = timing["scheduled_end_local"]
        existing.late_by_minutes = timing["late_by_minutes"]
        existing.attendance_rule_status = timing["rule_status"]
        existing.is_weekly_off = timing["is_weekly_off"]
        existing.status = final_status
        existing.approval_status = evaluation["approval_status"]
        existing.source = evaluation["source"]
        existing.remarks = punch_remarks
        existing.punch_in_latitude = _to_float(latitude)
        existing.punch_in_longitude = _to_float(longitude)
        existing.punch_in_accuracy_meters = _to_float(accuracy_meters)
        existing.punch_in_distance_meters = evaluation["distance_meters"]
        existing.punch_in_ip = client_ip
        existing.punch_in_geo_status = evaluation["geo_status"]
        existing.punch_in_ip_status = evaluation["ip_status"]
        existing.updated_by_user_id = actor.id
        existing.updated_at_utc = now
    db.commit()
    db.refresh(existing)
    return existing


def punch_out_attendance(
    db: Session,
    actor: User,
    *,
    remarks: str | None = None,
    latitude: float | str | None = None,
    longitude: float | str | None = None,
    accuracy_meters: float | str | None = None,
    client_ip: str | None = None,
) -> EmployeeAttendance:
    emp = get_employee_for_user(db, actor)
    if not emp:
        raise HTTPException(status_code=400, detail="Your user is not linked to an employee profile. Please request employee registration first.")
    now, local_dt, today, branch_tz, branch, branch_settings = _branch_now(db, emp.branch_id)
    row = db.execute(
        select(EmployeeAttendance).where(
            EmployeeAttendance.tenant_id == emp.tenant_id,
            EmployeeAttendance.employee_id == emp.id,
            EmployeeAttendance.attendance_date == today,
        )
    ).scalar_one_or_none()
    if not row or not row.punch_in_utc:
        raise HTTPException(status_code=400, detail="No punch-in found for today.")
    if row.punch_out_utc:
        raise HTTPException(status_code=409, detail="You have already punched out today.")
    evaluation = _evaluate_attendance_controls(
        db,
        branch_id=emp.branch_id,
        latitude=latitude,
        longitude=longitude,
        client_ip=client_ip,
    )
    _raise_if_required_geo_missing(branch_settings, evaluation)
    row.punch_out_utc = now
    row.punch_out_local_at = local_dt
    row.branch_timezone = row.branch_timezone or branch_tz
    row.work_duration_minutes = _attendance_duration_minutes(row)
    row.punch_out_latitude = _to_float(latitude)
    row.punch_out_longitude = _to_float(longitude)
    row.punch_out_accuracy_meters = _to_float(accuracy_meters)
    row.punch_out_distance_meters = evaluation["distance_meters"]
    row.punch_out_ip = client_ip
    row.punch_out_geo_status = evaluation["geo_status"]
    row.punch_out_ip_status = evaluation["ip_status"]
    if row.approval_status == "approved" and evaluation["approval_status"] == "pending":
        row.approval_status = "pending"
        row.status = "on_duty"
        row.source = "od_request"
    if _blank_to_none(remarks):
        row.remarks = _blank_to_none(remarks)
    elif row.approval_status == "pending" and not _blank_to_none(row.remarks):
        row.remarks = "Punch-out outside branch geofence / office IP. Approval required."
    row.updated_by_user_id = actor.id
    row.updated_at_utc = now
    db.commit()
    db.refresh(row)
    return row


def list_attendance_records(
    db: Session,
    scope: EmployeeScope,
    *,
    employee_id: int | None = None,
    from_date: date | None = None,
    to_date: date | None = None,
    status: str | None = None,
    approval_status: str | None = None,
) -> list[EmployeeAttendance]:
    stmt = select(EmployeeAttendance).options(selectinload(EmployeeAttendance.employee)).where(EmployeeAttendance.tenant_id == scope.tenant_id)
    if scope.branch_id is not None:
        stmt = stmt.where(EmployeeAttendance.branch_id == scope.branch_id)
    if employee_id:
        stmt = stmt.where(EmployeeAttendance.employee_id == int(employee_id))
    if from_date:
        stmt = stmt.where(EmployeeAttendance.attendance_date >= from_date)
    if to_date:
        stmt = stmt.where(EmployeeAttendance.attendance_date <= to_date)
    if status:
        stmt = stmt.where(EmployeeAttendance.status == status)
    if approval_status:
        stmt = stmt.where(EmployeeAttendance.approval_status == approval_status)
    rows = db.execute(stmt.order_by(EmployeeAttendance.attendance_date.desc(), EmployeeAttendance.id.desc())).scalars().all()
    return [_refresh_attendance_local_evidence(row) for row in rows]


def get_attendance_or_404(db: Session, attendance_id: int, scope: EmployeeScope) -> EmployeeAttendance:
    stmt = select(EmployeeAttendance).where(EmployeeAttendance.id == attendance_id, EmployeeAttendance.tenant_id == scope.tenant_id)
    if scope.branch_id is not None:
        stmt = stmt.where(EmployeeAttendance.branch_id == scope.branch_id)
    row = db.execute(stmt).scalar_one_or_none()
    if not row:
        raise HTTPException(status_code=404, detail="Attendance record not found or not accessible.")
    return row


def review_attendance_record(
    db: Session,
    actor: User,
    row: EmployeeAttendance,
    *,
    approval_status: str,
    review_notes: str | None = None,
) -> EmployeeAttendance:
    approval_status = (approval_status or "").lower()
    if approval_status not in ATTENDANCE_APPROVAL_STATUS:
        raise HTTPException(status_code=400, detail="Invalid attendance approval status.")
    now = _utc_now_naive()
    row.approval_status = approval_status
    row.review_notes = _blank_to_none(review_notes)
    row.reviewed_by_user_id = actor.id
    row.reviewed_at_utc = now
    row.updated_by_user_id = actor.id
    row.updated_at_utc = now
    db.commit()
    db.refresh(row)
    return row


def create_or_update_manual_attendance(
    db: Session,
    actor: User,
    scope: EmployeeScope,
    *,
    employee_id: int,
    attendance_date: date,
    status: str,
    remarks: str | None = None,
) -> EmployeeAttendance:
    emp = get_employee_or_404(db, int(employee_id), scope)
    status = (status or "present").lower()
    if status not in ATTENDANCE_STATUS:
        raise HTTPException(status_code=400, detail="Invalid attendance status.")
    today_now = _utc_now_naive()
    branch, branch_settings = _branch_settings(db, emp.branch_id)
    branch_tz = getattr(branch, "timezone", None) or "Asia/Kolkata"
    row = db.execute(
        select(EmployeeAttendance).where(
            EmployeeAttendance.tenant_id == emp.tenant_id,
            EmployeeAttendance.employee_id == emp.id,
            EmployeeAttendance.attendance_date == attendance_date,
        )
    ).scalar_one_or_none()
    if not row:
        row = EmployeeAttendance(
            tenant_id=emp.tenant_id,
            branch_id=emp.branch_id,
            employee_id=emp.id,
            user_id=emp.user_id,
            attendance_date=attendance_date,
            branch_timezone=branch_tz,
            scheduled_start_local=getattr(branch, "office_start_time", None) if branch else None,
            scheduled_end_local=getattr(branch, "office_end_time", None) if branch else None,
            attendance_rule_status="manual",
            status=status,
            approval_status="approved",
            source="manual",
            remarks=_blank_to_none(remarks),
            created_by_user_id=actor.id,
            updated_by_user_id=actor.id,
        )
        db.add(row)
    else:
        row.status = status
        row.approval_status = "approved"
        row.source = "manual"
        row.remarks = _blank_to_none(remarks)
        row.updated_by_user_id = actor.id
        row.updated_at_utc = today_now
    db.commit()
    db.refresh(row)
    return row



LEAVE_REQUEST_STATUS = ["pending", "approved", "rejected", "cancelled"]
LEAVE_TYPE_DEFAULTS = [
    ("CL", "Casual Leave", 12, True),
    ("SL", "Sick Leave", 12, True),
    ("EL", "Earned Leave", 0, True),
    ("LOP", "Loss of Pay", 0, False),
]


def _days_between(from_date: date, to_date: date) -> int:
    days = (to_date - from_date).days + 1
    if days <= 0:
        raise HTTPException(status_code=400, detail="Leave to-date must be on or after from-date.")
    return days


def list_leave_types(db: Session, scope: EmployeeScope, *, include_inactive: bool = False) -> list[EmployeeLeaveType]:
    stmt = select(EmployeeLeaveType).where(EmployeeLeaveType.tenant_id == scope.tenant_id)
    if scope.branch_id is not None:
        stmt = stmt.where(EmployeeLeaveType.branch_id == scope.branch_id)
    if not include_inactive:
        stmt = stmt.where(EmployeeLeaveType.is_active.is_(True))
    return db.execute(stmt.order_by(EmployeeLeaveType.code)).scalars().all()


def get_leave_type_or_404(db: Session, leave_type_id: int, scope: EmployeeScope) -> EmployeeLeaveType:
    stmt = select(EmployeeLeaveType).where(EmployeeLeaveType.id == leave_type_id, EmployeeLeaveType.tenant_id == scope.tenant_id)
    if scope.branch_id is not None:
        stmt = stmt.where(EmployeeLeaveType.branch_id == scope.branch_id)
    row = db.execute(stmt).scalar_one_or_none()
    if not row:
        raise HTTPException(status_code=404, detail="Leave type not found or not accessible.")
    return row


def ensure_default_leave_types(db: Session, actor: User, scope: EmployeeScope) -> int:
    if scope.branch_id is None:
        raise HTTPException(status_code=400, detail="Select a branch before creating default leave types.")
    created = 0
    for code, name, quota, paid in LEAVE_TYPE_DEFAULTS:
        exists = db.execute(
            select(EmployeeLeaveType).where(
                EmployeeLeaveType.tenant_id == scope.tenant_id,
                EmployeeLeaveType.branch_id == scope.branch_id,
                EmployeeLeaveType.code == code,
            )
        ).scalar_one_or_none()
        if exists:
            continue
        db.add(EmployeeLeaveType(
            tenant_id=scope.tenant_id,
            branch_id=scope.branch_id,
            code=code,
            name=name,
            annual_quota_days=quota,
            is_paid=paid,
            allow_negative_balance=(code == "LOP"),
            created_by_user_id=actor.id,
            updated_by_user_id=actor.id,
        ))
        created += 1
    db.commit()
    return created


def create_leave_type(db: Session, actor: User, scope: EmployeeScope, data: dict[str, Any]) -> EmployeeLeaveType:
    if scope.branch_id is None:
        raise HTTPException(status_code=400, detail="Select a branch before creating a leave type.")
    code = (_blank_to_none(data.get("code")) or "").upper()
    name = _blank_to_none(data.get("name")) or ""
    if not code or not name:
        raise HTTPException(status_code=400, detail="Leave code and name are required.")
    exists = db.execute(select(EmployeeLeaveType).where(
        EmployeeLeaveType.tenant_id == scope.tenant_id,
        EmployeeLeaveType.branch_id == scope.branch_id,
        EmployeeLeaveType.code == code,
    )).scalar_one_or_none()
    if exists:
        raise HTTPException(status_code=409, detail="Leave type code already exists for this branch.")
    row = EmployeeLeaveType(
        tenant_id=scope.tenant_id,
        branch_id=scope.branch_id,
        code=code,
        name=name,
        description=_blank_to_none(data.get("description")),
        annual_quota_days=int(data.get("annual_quota_days") or 0),
        carry_forward_allowed=bool(data.get("carry_forward_allowed")),
        allow_negative_balance=bool(data.get("allow_negative_balance")),
        requires_approval=bool(data.get("requires_approval", True)),
        is_paid=bool(data.get("is_paid", True)),
        is_active=bool(data.get("is_active", True)),
        created_by_user_id=actor.id,
        updated_by_user_id=actor.id,
    )
    db.add(row)
    db.commit()
    db.refresh(row)
    return row


def update_leave_type(db: Session, actor: User, row: EmployeeLeaveType, data: dict[str, Any]) -> EmployeeLeaveType:
    row.name = _blank_to_none(data.get("name")) or row.name
    row.description = _blank_to_none(data.get("description"))
    row.annual_quota_days = int(data.get("annual_quota_days") or 0)
    row.carry_forward_allowed = bool(data.get("carry_forward_allowed"))
    row.allow_negative_balance = bool(data.get("allow_negative_balance"))
    row.requires_approval = bool(data.get("requires_approval", True))
    row.is_paid = bool(data.get("is_paid", True))
    row.is_active = bool(data.get("is_active"))
    row.updated_by_user_id = actor.id
    row.updated_at_utc = datetime.now(timezone.utc)
    db.commit()
    db.refresh(row)
    return row


def _get_or_create_leave_balance(db: Session, emp: Employee, leave_type: EmployeeLeaveType, actor_id: int | None = None) -> EmployeeLeaveBalance:
    bal = db.execute(select(EmployeeLeaveBalance).where(
        EmployeeLeaveBalance.tenant_id == emp.tenant_id,
        EmployeeLeaveBalance.employee_id == emp.id,
        EmployeeLeaveBalance.leave_type_id == leave_type.id,
    )).scalar_one_or_none()
    if bal:
        return bal
    initial = int(leave_type.annual_quota_days or 0)
    bal = EmployeeLeaveBalance(
        tenant_id=emp.tenant_id,
        branch_id=emp.branch_id,
        employee_id=emp.id,
        leave_type_id=leave_type.id,
        opening_days=0,
        credited_days=initial,
        availed_days=0,
        adjusted_days=0,
        balance_days=initial,
        updated_by_user_id=actor_id,
    )
    db.add(bal)
    db.flush()
    return bal


def list_leave_balances(db: Session, scope: EmployeeScope, *, employee_id: int | None = None) -> list[EmployeeLeaveBalance]:
    stmt = select(EmployeeLeaveBalance).options(selectinload(EmployeeLeaveBalance.employee), selectinload(EmployeeLeaveBalance.leave_type)).where(EmployeeLeaveBalance.tenant_id == scope.tenant_id)
    if scope.branch_id is not None:
        stmt = stmt.where(EmployeeLeaveBalance.branch_id == scope.branch_id)
    if employee_id:
        stmt = stmt.where(EmployeeLeaveBalance.employee_id == employee_id)
    return db.execute(stmt.order_by(EmployeeLeaveBalance.employee_id, EmployeeLeaveBalance.leave_type_id)).scalars().all()


def list_own_leave_balances(db: Session, actor: User) -> list[EmployeeLeaveBalance]:
    emp = get_employee_for_user(db, actor)
    if not emp:
        return []
    return db.execute(select(EmployeeLeaveBalance).options(selectinload(EmployeeLeaveBalance.leave_type)).where(EmployeeLeaveBalance.employee_id == emp.id).order_by(EmployeeLeaveBalance.leave_type_id)).scalars().all()


def adjust_leave_balance(db: Session, actor: User, scope: EmployeeScope, *, employee_id: int, leave_type_id: int, adjusted_days: int, notes: str | None = None) -> EmployeeLeaveBalance:
    emp = get_employee_or_404(db, employee_id, scope)
    leave_type = get_leave_type_or_404(db, leave_type_id, scope)
    if leave_type.branch_id != emp.branch_id:
        raise HTTPException(status_code=400, detail="Leave type and employee branch do not match.")
    bal = _get_or_create_leave_balance(db, emp, leave_type, actor.id)
    bal.adjusted_days = int(adjusted_days or 0)
    bal.balance_days = int(bal.opening_days or 0) + int(bal.credited_days or 0) + int(bal.adjusted_days or 0) - int(bal.availed_days or 0)
    bal.updated_by_user_id = actor.id
    bal.updated_at_utc = datetime.now(timezone.utc)
    db.commit()
    db.refresh(bal)
    return bal


def list_leave_requests(db: Session, scope: EmployeeScope, *, employee_id: int | None = None, status: str | None = None) -> list[EmployeeLeaveRequest]:
    stmt = select(EmployeeLeaveRequest).options(selectinload(EmployeeLeaveRequest.employee), selectinload(EmployeeLeaveRequest.leave_type)).where(EmployeeLeaveRequest.tenant_id == scope.tenant_id)
    if scope.branch_id is not None:
        stmt = stmt.where(EmployeeLeaveRequest.branch_id == scope.branch_id)
    if employee_id:
        stmt = stmt.where(EmployeeLeaveRequest.employee_id == employee_id)
    if status:
        stmt = stmt.where(EmployeeLeaveRequest.status == status)
    return db.execute(stmt.order_by(EmployeeLeaveRequest.created_at_utc.desc())).scalars().all()


def list_own_leave_requests(db: Session, actor: User) -> list[EmployeeLeaveRequest]:
    emp = get_employee_for_user(db, actor)
    if not emp:
        return []
    return db.execute(select(EmployeeLeaveRequest).options(selectinload(EmployeeLeaveRequest.leave_type)).where(EmployeeLeaveRequest.employee_id == emp.id).order_by(EmployeeLeaveRequest.created_at_utc.desc())).scalars().all()


def get_leave_request_or_404(db: Session, request_id: int, scope: EmployeeScope) -> EmployeeLeaveRequest:
    stmt = select(EmployeeLeaveRequest).options(selectinload(EmployeeLeaveRequest.employee), selectinload(EmployeeLeaveRequest.leave_type)).where(EmployeeLeaveRequest.id == request_id, EmployeeLeaveRequest.tenant_id == scope.tenant_id)
    if scope.branch_id is not None:
        stmt = stmt.where(EmployeeLeaveRequest.branch_id == scope.branch_id)
    row = db.execute(stmt).scalar_one_or_none()
    if not row:
        raise HTTPException(status_code=404, detail="Leave request not found or not accessible.")
    return row


def apply_employee_leave(db: Session, actor: User, *, leave_type_id: int, from_date: date, to_date: date, reason: str | None = None) -> EmployeeLeaveRequest:
    emp = get_employee_for_user(db, actor)
    if not emp:
        raise HTTPException(status_code=400, detail="Your user is not linked to an employee profile.")
    days = _days_between(from_date, to_date)
    leave_type = db.get(EmployeeLeaveType, int(leave_type_id))
    if not leave_type or leave_type.tenant_id != emp.tenant_id or leave_type.branch_id != emp.branch_id or not leave_type.is_active:
        raise HTTPException(status_code=404, detail="Leave type not available for your branch.")
    row = EmployeeLeaveRequest(
        tenant_id=emp.tenant_id,
        branch_id=emp.branch_id,
        employee_id=emp.id,
        user_id=actor.id,
        leave_type_id=leave_type.id,
        from_date=from_date,
        to_date=to_date,
        days=days,
        reason=_blank_to_none(reason),
        status="pending" if leave_type.requires_approval else "approved",
        request_source="employee_portal",
        created_by_user_id=actor.id,
        updated_by_user_id=actor.id,
    )
    db.add(row)
    if not leave_type.requires_approval:
        bal = _get_or_create_leave_balance(db, emp, leave_type, actor.id)
        if not leave_type.allow_negative_balance and bal.balance_days < days:
            raise HTTPException(status_code=400, detail="Insufficient leave balance.")
        bal.availed_days += days
        bal.balance_days -= days
    db.commit()
    db.refresh(row)
    return row


def review_leave_request(db: Session, actor: User, row: EmployeeLeaveRequest, *, status: str, review_notes: str | None = None) -> EmployeeLeaveRequest:
    status = (status or "").lower()
    if status not in ("approved", "rejected"):
        raise HTTPException(status_code=400, detail="Invalid leave review status.")
    if row.status not in ("pending",):
        raise HTTPException(status_code=400, detail="Only pending leave requests can be reviewed.")
    leave_type = row.leave_type or db.get(EmployeeLeaveType, row.leave_type_id)
    emp = row.employee or db.get(Employee, row.employee_id)
    if status == "approved":
        bal = _get_or_create_leave_balance(db, emp, leave_type, actor.id)
        if not leave_type.allow_negative_balance and bal.balance_days < row.days:
            raise HTTPException(status_code=400, detail="Insufficient leave balance for approval.")
        bal.availed_days += row.days
        bal.balance_days -= row.days
        bal.updated_by_user_id = actor.id
        bal.updated_at_utc = datetime.now(timezone.utc)
    row.status = status
    row.review_notes = _blank_to_none(review_notes)
    row.reviewed_by_user_id = actor.id
    row.reviewed_at_utc = datetime.now(timezone.utc)
    row.updated_by_user_id = actor.id
    row.updated_at_utc = datetime.now(timezone.utc)
    db.commit()
    db.refresh(row)
    return row


def cancel_own_leave_request(db: Session, actor: User, request_id: int) -> EmployeeLeaveRequest:
    emp = get_employee_for_user(db, actor)
    if not emp:
        raise HTTPException(status_code=400, detail="Your user is not linked to an employee profile.")
    row = db.execute(select(EmployeeLeaveRequest).where(EmployeeLeaveRequest.id == request_id, EmployeeLeaveRequest.employee_id == emp.id)).scalar_one_or_none()
    if not row:
        raise HTTPException(status_code=404, detail="Leave request not found.")
    if row.status != "pending":
        raise HTTPException(status_code=400, detail="Only pending leave requests can be cancelled.")
    row.status = "cancelled"
    row.updated_by_user_id = actor.id
    row.updated_at_utc = datetime.now(timezone.utc)
    db.commit()
    db.refresh(row)
    return row



DOCUMENT_TYPE_DEFAULTS = [
    ("AADHAAR", "Aadhaar / ID Proof", True),
    ("PAN", "PAN Card", True),
    ("PHOTO", "Photo", False),
    ("ADDRESS", "Address Proof", False),
    ("EDU", "Education Certificate", False),
    ("EXP", "Experience Certificate", False),
    ("BANK", "Bank Proof / Cancelled Cheque", False),
    ("OTHER", "Other Document", False),
]


def list_document_types(db: Session, scope: EmployeeScope, *, include_inactive: bool = False) -> list[EmployeeDocumentType]:
    stmt = select(EmployeeDocumentType).where(EmployeeDocumentType.tenant_id == scope.tenant_id)
    if scope.branch_id is not None:
        stmt = stmt.where(EmployeeDocumentType.branch_id == scope.branch_id)
    if not include_inactive:
        stmt = stmt.where(EmployeeDocumentType.is_active.is_(True))
    return db.execute(stmt.order_by(EmployeeDocumentType.code)).scalars().all()


def get_document_type_or_404(db: Session, document_type_id: int, scope: EmployeeScope) -> EmployeeDocumentType:
    stmt = select(EmployeeDocumentType).where(EmployeeDocumentType.id == document_type_id, EmployeeDocumentType.tenant_id == scope.tenant_id)
    if scope.branch_id is not None:
        stmt = stmt.where(EmployeeDocumentType.branch_id == scope.branch_id)
    row = db.execute(stmt).scalar_one_or_none()
    if not row:
        raise HTTPException(status_code=404, detail="Document type not found or not accessible.")
    return row


def ensure_default_document_types(db: Session, actor: User, scope: EmployeeScope) -> int:
    if scope.branch_id is None:
        raise HTTPException(status_code=400, detail="Select a branch before creating default document types.")
    created = 0
    for code, name, mandatory in DOCUMENT_TYPE_DEFAULTS:
        exists = db.execute(
            select(EmployeeDocumentType).where(
                EmployeeDocumentType.tenant_id == scope.tenant_id,
                EmployeeDocumentType.branch_id == scope.branch_id,
                EmployeeDocumentType.code == code,
            )
        ).scalar_one_or_none()
        if exists:
            continue
        db.add(EmployeeDocumentType(
            tenant_id=scope.tenant_id,
            branch_id=scope.branch_id,
            code=code,
            name=name,
            is_mandatory=mandatory,
            allow_employee_upload=True,
            requires_verification=True,
            is_active=True,
            created_by_user_id=actor.id,
            updated_by_user_id=actor.id,
        ))
        created += 1
    db.commit()
    return created


def create_document_type(db: Session, actor: User, scope: EmployeeScope, data: dict[str, Any]) -> EmployeeDocumentType:
    if scope.branch_id is None:
        raise HTTPException(status_code=400, detail="Select a branch before creating a document type.")
    code = (_blank_to_none(data.get("code")) or "").upper()
    name = _blank_to_none(data.get("name")) or ""
    if not code or not name:
        raise HTTPException(status_code=400, detail="Document type code and name are required.")
    exists = db.execute(select(EmployeeDocumentType).where(
        EmployeeDocumentType.tenant_id == scope.tenant_id,
        EmployeeDocumentType.branch_id == scope.branch_id,
        EmployeeDocumentType.code == code,
    )).scalar_one_or_none()
    if exists:
        raise HTTPException(status_code=409, detail="Document type code already exists for this branch.")
    row = EmployeeDocumentType(
        tenant_id=scope.tenant_id,
        branch_id=scope.branch_id,
        code=code,
        name=name,
        description=_blank_to_none(data.get("description")),
        is_mandatory=bool(data.get("is_mandatory")),
        allow_employee_upload=bool(data.get("allow_employee_upload", True)),
        requires_verification=bool(data.get("requires_verification", True)),
        is_active=bool(data.get("is_active", True)),
        created_by_user_id=actor.id,
        updated_by_user_id=actor.id,
    )
    db.add(row)
    db.commit()
    db.refresh(row)
    return row


def update_document_type(db: Session, actor: User, row: EmployeeDocumentType, data: dict[str, Any]) -> EmployeeDocumentType:
    row.name = _blank_to_none(data.get("name")) or row.name
    row.description = _blank_to_none(data.get("description"))
    row.is_mandatory = bool(data.get("is_mandatory"))
    row.allow_employee_upload = bool(data.get("allow_employee_upload", True))
    row.requires_verification = bool(data.get("requires_verification", True))
    row.is_active = bool(data.get("is_active"))
    row.updated_by_user_id = actor.id
    row.updated_at_utc = datetime.now(timezone.utc)
    db.commit()
    db.refresh(row)
    return row


def list_employee_documents(db: Session, scope: EmployeeScope, *, employee_id: int | None = None, status: str | None = None) -> list[EmployeeDocument]:
    stmt = select(EmployeeDocument).options(selectinload(EmployeeDocument.employee), selectinload(EmployeeDocument.document_type)).where(EmployeeDocument.tenant_id == scope.tenant_id)
    if scope.branch_id is not None:
        stmt = stmt.where(EmployeeDocument.branch_id == scope.branch_id)
    if employee_id:
        stmt = stmt.where(EmployeeDocument.employee_id == employee_id)
    if status:
        stmt = stmt.where(EmployeeDocument.status == status)
    return db.execute(stmt.order_by(EmployeeDocument.created_at_utc.desc())).scalars().all()


def list_own_employee_documents(db: Session, actor: User) -> list[EmployeeDocument]:
    emp = get_employee_for_user(db, actor)
    if not emp:
        return []
    return db.execute(
        select(EmployeeDocument)
        .options(selectinload(EmployeeDocument.document_type))
        .where(
            EmployeeDocument.employee_id == emp.id,
            EmployeeDocument.visibility == "employee_and_hr",
            EmployeeDocument.status != "archived",
        )
        .order_by(EmployeeDocument.created_at_utc.desc())
    ).scalars().all()


def get_employee_document_or_404(db: Session, document_id: int, scope: EmployeeScope) -> EmployeeDocument:
    stmt = select(EmployeeDocument).options(selectinload(EmployeeDocument.employee), selectinload(EmployeeDocument.document_type)).where(EmployeeDocument.id == document_id, EmployeeDocument.tenant_id == scope.tenant_id)
    if scope.branch_id is not None:
        stmt = stmt.where(EmployeeDocument.branch_id == scope.branch_id)
    row = db.execute(stmt).scalar_one_or_none()
    if not row:
        raise HTTPException(status_code=404, detail="Employee document not found or not accessible.")
    return row


def get_own_employee_document_or_404(db: Session, actor: User, document_id: int) -> EmployeeDocument:
    emp = get_employee_for_user(db, actor)
    if not emp:
        raise HTTPException(status_code=400, detail="Your user is not linked to an employee profile.")
    row = db.execute(select(EmployeeDocument).where(EmployeeDocument.id == document_id, EmployeeDocument.employee_id == emp.id)).scalar_one_or_none()
    if not row or row.visibility != "employee_and_hr" or row.status == "archived":
        raise HTTPException(status_code=404, detail="Employee document not found.")
    return row


def create_employee_document_record(
    db: Session,
    actor: User,
    employee: Employee,
    *,
    document_type_id: int | None,
    title: str,
    document_no: str | None,
    issue_date: date | None,
    expiry_date: date | None,
    original_filename: str,
    stored_filename: str,
    storage_path: str,
    content_type: str | None,
    file_size_bytes: int | None,
    remarks: str | None = None,
    visibility: str = "employee_and_hr",
    uploaded_status: str = "uploaded",
) -> EmployeeDocument:
    title = (_blank_to_none(title) or original_filename or "Employee Document").strip()
    if visibility not in DOCUMENT_VISIBILITY:
        visibility = "employee_and_hr"
    document_type = None
    if document_type_id:
        document_type = db.get(EmployeeDocumentType, int(document_type_id))
        if not document_type or document_type.tenant_id != employee.tenant_id or document_type.branch_id != employee.branch_id:
            raise HTTPException(status_code=404, detail="Document type is not available for this employee branch.")
    row = EmployeeDocument(
        tenant_id=employee.tenant_id,
        branch_id=employee.branch_id,
        employee_id=employee.id,
        document_type_id=document_type.id if document_type else None,
        title=title,
        document_no=_blank_to_none(document_no),
        issue_date=issue_date,
        expiry_date=expiry_date,
        original_filename=original_filename,
        stored_filename=stored_filename,
        storage_path=storage_path,
        content_type=content_type,
        file_size_bytes=file_size_bytes,
        status=uploaded_status if uploaded_status in DOCUMENT_STATUS else "uploaded",
        visibility=visibility,
        remarks=_blank_to_none(remarks),
        uploaded_by_user_id=actor.id,
        updated_by_user_id=actor.id,
    )
    db.add(row)
    db.commit()
    db.refresh(row)
    return row


def review_employee_document(db: Session, actor: User, row: EmployeeDocument, *, status: str, verification_notes: str | None = None) -> EmployeeDocument:
    status = (status or "").lower()
    if status not in ("verified", "rejected"):
        raise HTTPException(status_code=400, detail="Invalid document review status.")
    if row.status == "archived":
        raise HTTPException(status_code=400, detail="Archived documents cannot be reviewed.")
    row.status = status
    row.verification_notes = _blank_to_none(verification_notes)
    row.verified_by_user_id = actor.id
    row.verified_at_utc = datetime.now(timezone.utc)
    row.updated_by_user_id = actor.id
    row.updated_at_utc = datetime.now(timezone.utc)
    db.commit()
    db.refresh(row)
    return row


def archive_employee_document(db: Session, actor: User, row: EmployeeDocument, *, notes: str | None = None) -> EmployeeDocument:
    row.status = "archived"
    row.verification_notes = _blank_to_none(notes) or row.verification_notes
    row.updated_by_user_id = actor.id
    row.updated_at_utc = datetime.now(timezone.utc)
    db.commit()
    db.refresh(row)
    return row



ONBOARDING_DEFAULTS = [
    ("WELCOME", "Welcome and joining confirmation", "joining", 0, 10),
    ("ID_DOCS", "Collect and verify identity documents", "joining", 1, 20),
    ("BANK", "Collect bank account details", "joining", 1, 30),
    ("SYSTEM_ACCESS", "Create system access and assign role", "first_week", 1, 40),
    ("POLICIES", "Share office policies and confidentiality instructions", "first_week", 2, 50),
]

OFFBOARDING_DEFAULT_TASKS = [
    ("Collect handover notes", "Collect pending work, client list and handover notes."),
    ("Recover office assets", "Recover laptop, tokens, books, keys and other assets."),
    ("Disable system access", "Disable/limit application, email and storage access after relieving."),
    ("Final settlement checklist", "Verify attendance, leave and final settlement points."),
    ("Relieving documentation", "Prepare relieving/experience documentation where applicable."),
]


def _task_due(base_date: date | None, days: int) -> date | None:
    if not base_date:
        return None
    return base_date + timedelta(days=int(days or 0))


def list_onboarding_checklist_items(db: Session, scope: EmployeeScope, *, include_inactive: bool = True) -> list[EmployeeOnboardingChecklistItem]:
    stmt = select(EmployeeOnboardingChecklistItem).where(EmployeeOnboardingChecklistItem.tenant_id == scope.tenant_id)
    if scope.branch_id is not None:
        stmt = stmt.where(EmployeeOnboardingChecklistItem.branch_id == scope.branch_id)
    if not include_inactive:
        stmt = stmt.where(EmployeeOnboardingChecklistItem.is_active.is_(True))
    return db.execute(stmt.order_by(EmployeeOnboardingChecklistItem.sort_order, EmployeeOnboardingChecklistItem.title)).scalars().all()


def ensure_default_onboarding_checklist(db: Session, user: User, scope: EmployeeScope) -> int:
    if scope.branch_id is None:
        raise HTTPException(status_code=400, detail="Please select a specific branch before creating onboarding defaults.")
    created = 0
    for code, title, stage, due_days, sort_order in ONBOARDING_DEFAULTS:
        exists = db.execute(select(EmployeeOnboardingChecklistItem).where(
            EmployeeOnboardingChecklistItem.tenant_id == scope.tenant_id,
            EmployeeOnboardingChecklistItem.branch_id == scope.branch_id,
            EmployeeOnboardingChecklistItem.code == code,
        )).scalar_one_or_none()
        if exists:
            continue
        db.add(EmployeeOnboardingChecklistItem(
            tenant_id=scope.tenant_id,
            branch_id=scope.branch_id,
            code=code,
            title=title,
            stage=stage,
            default_due_days=due_days,
            sort_order=sort_order,
            is_mandatory=True,
            is_active=True,
            created_by_user_id=user.id,
            updated_by_user_id=user.id,
        ))
        created += 1
    db.commit()
    return created


def create_onboarding_checklist_item(db: Session, user: User, scope: EmployeeScope, data: dict[str, Any]) -> EmployeeOnboardingChecklistItem:
    if scope.branch_id is None:
        raise HTTPException(status_code=400, detail="Please select a specific branch before creating checklist item.")
    code = str(_blank_to_none(data.get("code")) or "").upper()
    title = str(_blank_to_none(data.get("title")) or "")
    if not code or not title:
        raise HTTPException(status_code=400, detail="Code and title are required.")
    exists = db.execute(select(EmployeeOnboardingChecklistItem).where(
        EmployeeOnboardingChecklistItem.tenant_id == scope.tenant_id,
        EmployeeOnboardingChecklistItem.branch_id == scope.branch_id,
        EmployeeOnboardingChecklistItem.code == code,
    )).scalar_one_or_none()
    if exists:
        raise HTTPException(status_code=409, detail="Checklist code already exists for this branch.")
    item = EmployeeOnboardingChecklistItem(
        tenant_id=scope.tenant_id,
        branch_id=scope.branch_id,
        code=code,
        title=title,
        description=_blank_to_none(data.get("description")),
        stage=_blank_to_none(data.get("stage")) or "joining",
        default_due_days=int(data.get("default_due_days") or 0),
        sort_order=int(data.get("sort_order") or 0),
        is_mandatory=bool(data.get("is_mandatory", True)),
        is_active=bool(data.get("is_active", True)),
        created_by_user_id=user.id,
        updated_by_user_id=user.id,
    )
    db.add(item)
    db.commit()
    db.refresh(item)
    return item


def get_onboarding_checklist_item_or_404(db: Session, item_id: int, scope: EmployeeScope) -> EmployeeOnboardingChecklistItem:
    stmt = select(EmployeeOnboardingChecklistItem).where(EmployeeOnboardingChecklistItem.id == item_id, EmployeeOnboardingChecklistItem.tenant_id == scope.tenant_id)
    if scope.branch_id is not None:
        stmt = stmt.where(EmployeeOnboardingChecklistItem.branch_id == scope.branch_id)
    item = db.execute(stmt).scalar_one_or_none()
    if not item:
        raise HTTPException(status_code=404, detail="Onboarding checklist item not found or not accessible.")
    return item


def update_onboarding_checklist_item(db: Session, user: User, item: EmployeeOnboardingChecklistItem, data: dict[str, Any]) -> EmployeeOnboardingChecklistItem:
    item.title = str(_blank_to_none(data.get("title")) or item.title)
    item.description = _blank_to_none(data.get("description"))
    item.stage = _blank_to_none(data.get("stage")) or "joining"
    item.default_due_days = int(data.get("default_due_days") or 0)
    item.sort_order = int(data.get("sort_order") or 0)
    item.is_mandatory = bool(data.get("is_mandatory", True))
    item.is_active = bool(data.get("is_active", True))
    item.updated_by_user_id = user.id
    db.commit()
    db.refresh(item)
    return item


def generate_onboarding_tasks_for_employee(db: Session, user: User, employee: Employee, scope: EmployeeScope) -> int:
    items = db.execute(select(EmployeeOnboardingChecklistItem).where(
        EmployeeOnboardingChecklistItem.tenant_id == employee.tenant_id,
        EmployeeOnboardingChecklistItem.branch_id == employee.branch_id,
        EmployeeOnboardingChecklistItem.is_active.is_(True),
    ).order_by(EmployeeOnboardingChecklistItem.sort_order)).scalars().all()
    if not items:
        branch_scope = EmployeeScope(employee.tenant_id, employee.branch_id, scope.is_system_admin, scope.is_firm_admin, scope.is_partner, scope.is_branch_manager, scope.is_staff, scope.allow_cross_tenant, scope.allow_cross_branch, scope.own_user_id)
        ensure_default_onboarding_checklist(db, user, branch_scope)
        items = db.execute(select(EmployeeOnboardingChecklistItem).where(
            EmployeeOnboardingChecklistItem.tenant_id == employee.tenant_id,
            EmployeeOnboardingChecklistItem.branch_id == employee.branch_id,
            EmployeeOnboardingChecklistItem.is_active.is_(True),
        ).order_by(EmployeeOnboardingChecklistItem.sort_order)).scalars().all()
    created = 0
    base_date = employee.date_of_joining or date.today()
    for item in items:
        exists = db.execute(select(EmployeeOnboardingTask).where(
            EmployeeOnboardingTask.tenant_id == employee.tenant_id,
            EmployeeOnboardingTask.employee_id == employee.id,
            EmployeeOnboardingTask.checklist_item_id == item.id,
        )).scalar_one_or_none()
        if exists:
            continue
        db.add(EmployeeOnboardingTask(
            tenant_id=employee.tenant_id,
            branch_id=employee.branch_id,
            employee_id=employee.id,
            checklist_item_id=item.id,
            title=item.title,
            description=item.description,
            stage=item.stage,
            due_date=_task_due(base_date, item.default_due_days),
            status="pending",
            assigned_to_user_id=employee.reporting_manager_user_id,
            created_by_user_id=user.id,
            updated_by_user_id=user.id,
        ))
        created += 1
    db.commit()
    return created


def list_onboarding_tasks(db: Session, scope: EmployeeScope, *, employee_id: int | None = None) -> list[EmployeeOnboardingTask]:
    stmt = select(EmployeeOnboardingTask).options(selectinload(EmployeeOnboardingTask.employee), selectinload(EmployeeOnboardingTask.assigned_to)).where(EmployeeOnboardingTask.tenant_id == scope.tenant_id)
    if scope.branch_id is not None:
        stmt = stmt.where(EmployeeOnboardingTask.branch_id == scope.branch_id)
    if employee_id:
        stmt = stmt.where(EmployeeOnboardingTask.employee_id == employee_id)
    return db.execute(stmt.order_by(EmployeeOnboardingTask.status, EmployeeOnboardingTask.due_date)).scalars().all()


def get_onboarding_task_or_404(db: Session, task_id: int, scope: EmployeeScope) -> EmployeeOnboardingTask:
    stmt = select(EmployeeOnboardingTask).where(EmployeeOnboardingTask.id == task_id, EmployeeOnboardingTask.tenant_id == scope.tenant_id)
    if scope.branch_id is not None:
        stmt = stmt.where(EmployeeOnboardingTask.branch_id == scope.branch_id)
    task = db.execute(stmt).scalar_one_or_none()
    if not task:
        raise HTTPException(status_code=404, detail="Onboarding task not found or not accessible.")
    return task


def update_onboarding_task_status(db: Session, user: User, task: EmployeeOnboardingTask, status: str, notes: str | None = None) -> EmployeeOnboardingTask:
    status = (status or "pending").lower()
    if status not in ONBOARDING_TASK_STATUS:
        raise HTTPException(status_code=400, detail="Invalid onboarding task status.")
    task.status = status
    task.review_notes = _blank_to_none(notes)
    task.updated_by_user_id = user.id
    if status in ("completed", "skipped"):
        task.completed_by_user_id = user.id
        task.completed_at_utc = datetime.now(timezone.utc)
    else:
        task.completed_by_user_id = None
        task.completed_at_utc = None
    db.commit()
    db.refresh(task)
    return task


def list_offboarding_requests(db: Session, scope: EmployeeScope, *, status: str | None = None) -> list[EmployeeOffboardingRequest]:
    stmt = select(EmployeeOffboardingRequest).options(selectinload(EmployeeOffboardingRequest.employee)).where(EmployeeOffboardingRequest.tenant_id == scope.tenant_id)
    if scope.branch_id is not None:
        stmt = stmt.where(EmployeeOffboardingRequest.branch_id == scope.branch_id)
    if status:
        stmt = stmt.where(EmployeeOffboardingRequest.status == status)
    return db.execute(stmt.order_by(EmployeeOffboardingRequest.created_at_utc.desc())).scalars().all()


def list_own_offboarding_requests(db: Session, user: User) -> list[EmployeeOffboardingRequest]:
    employee = get_employee_for_user(db, user)
    if not employee:
        return []
    stmt = select(EmployeeOffboardingRequest).where(
        EmployeeOffboardingRequest.tenant_id == employee.tenant_id,
        EmployeeOffboardingRequest.employee_id == employee.id,
    ).order_by(EmployeeOffboardingRequest.created_at_utc.desc())
    return db.execute(stmt).scalars().all()


def create_offboarding_request(db: Session, user: User, employee: Employee, data: dict[str, Any], *, source: str = "admin") -> EmployeeOffboardingRequest:
    pending = db.execute(select(EmployeeOffboardingRequest).where(
        EmployeeOffboardingRequest.employee_id == employee.id,
        EmployeeOffboardingRequest.status.in_(["pending", "approved"]),
    )).scalar_one_or_none()
    if pending:
        raise HTTPException(status_code=409, detail="This employee already has an active offboarding request.")
    req = EmployeeOffboardingRequest(
        tenant_id=employee.tenant_id,
        branch_id=employee.branch_id,
        employee_id=employee.id,
        user_id=employee.user_id,
        request_type=_blank_to_none(data.get("request_type")) or "resignation",
        requested_relieving_date=parse_date(data.get("requested_relieving_date")),
        reason=_blank_to_none(data.get("reason")),
        handover_notes=_blank_to_none(data.get("handover_notes")),
        status="pending",
        requested_by_user_id=user.id,
    )
    db.add(req)
    db.commit()
    db.refresh(req)
    return req


def get_offboarding_request_or_404(db: Session, request_id: int, scope: EmployeeScope) -> EmployeeOffboardingRequest:
    stmt = select(EmployeeOffboardingRequest).options(selectinload(EmployeeOffboardingRequest.employee)).where(EmployeeOffboardingRequest.id == request_id, EmployeeOffboardingRequest.tenant_id == scope.tenant_id)
    if scope.branch_id is not None:
        stmt = stmt.where(EmployeeOffboardingRequest.branch_id == scope.branch_id)
    req = db.execute(stmt).scalar_one_or_none()
    if not req:
        raise HTTPException(status_code=404, detail="Offboarding request not found or not accessible.")
    return req


def _ensure_offboarding_tasks(db: Session, user: User, req: EmployeeOffboardingRequest) -> int:
    existing = db.execute(select(EmployeeOffboardingTask).where(EmployeeOffboardingTask.request_id == req.id)).scalars().all()
    if existing:
        return 0
    base = req.approved_relieving_date or req.requested_relieving_date or date.today()
    created = 0
    for title, desc in OFFBOARDING_DEFAULT_TASKS:
        db.add(EmployeeOffboardingTask(
            tenant_id=req.tenant_id,
            branch_id=req.branch_id,
            request_id=req.id,
            employee_id=req.employee_id,
            title=title,
            description=desc,
            due_date=base,
            status="pending",
            created_by_user_id=user.id,
            updated_by_user_id=user.id,
        ))
        created += 1
    db.commit()
    return created


def review_offboarding_request(db: Session, user: User, req: EmployeeOffboardingRequest, *, status: str, approved_relieving_date: date | None = None, review_notes: str | None = None) -> EmployeeOffboardingRequest:
    status = (status or "").lower()
    if status not in ("approved", "rejected"):
        raise HTTPException(status_code=400, detail="Offboarding review status must be approved or rejected.")
    if req.status not in ("pending", "approved"):
        raise HTTPException(status_code=400, detail="Only pending/approved requests can be reviewed.")
    req.status = status
    req.approved_relieving_date = approved_relieving_date or req.requested_relieving_date
    req.review_notes = _blank_to_none(review_notes)
    req.reviewed_by_user_id = user.id
    req.reviewed_at_utc = datetime.now(timezone.utc)
    db.commit()
    db.refresh(req)
    if status == "approved":
        _ensure_offboarding_tasks(db, user, req)
        db.refresh(req)
    return req


def list_offboarding_tasks(db: Session, req: EmployeeOffboardingRequest) -> list[EmployeeOffboardingTask]:
    return db.execute(select(EmployeeOffboardingTask).where(EmployeeOffboardingTask.request_id == req.id).order_by(EmployeeOffboardingTask.status, EmployeeOffboardingTask.due_date, EmployeeOffboardingTask.id)).scalars().all()


def get_offboarding_task_or_404(db: Session, task_id: int, scope: EmployeeScope) -> EmployeeOffboardingTask:
    stmt = select(EmployeeOffboardingTask).where(EmployeeOffboardingTask.id == task_id, EmployeeOffboardingTask.tenant_id == scope.tenant_id)
    if scope.branch_id is not None:
        stmt = stmt.where(EmployeeOffboardingTask.branch_id == scope.branch_id)
    task = db.execute(stmt).scalar_one_or_none()
    if not task:
        raise HTTPException(status_code=404, detail="Offboarding task not found or not accessible.")
    return task


def update_offboarding_task_status(db: Session, user: User, task: EmployeeOffboardingTask, status: str, notes: str | None = None) -> EmployeeOffboardingTask:
    status = (status or "pending").lower()
    if status not in OFFBOARDING_TASK_STATUS:
        raise HTTPException(status_code=400, detail="Invalid offboarding task status.")
    task.status = status
    task.review_notes = _blank_to_none(notes)
    task.updated_by_user_id = user.id
    if status in ("completed", "waived"):
        task.completed_by_user_id = user.id
        task.completed_at_utc = datetime.now(timezone.utc)
    else:
        task.completed_by_user_id = None
        task.completed_at_utc = None
    db.commit()
    db.refresh(task)
    return task


def complete_offboarding_request(db: Session, user: User, req: EmployeeOffboardingRequest) -> EmployeeOffboardingRequest:
    if req.status != "approved":
        raise HTTPException(status_code=400, detail="Only approved offboarding requests can be completed.")
    tasks = list_offboarding_tasks(db, req)
    pending = [t for t in tasks if t.status == "pending"]
    if pending:
        raise HTTPException(status_code=400, detail="Complete or waive all offboarding tasks before final completion.")
    employee = db.get(Employee, req.employee_id)
    if employee:
        employee.status = "relieved"
        employee.is_active = False
        employee.date_of_leaving = req.approved_relieving_date or req.requested_relieving_date or date.today()
        employee.updated_by_user_id = user.id
    req.status = "completed"
    req.completed_by_user_id = user.id
    req.completed_at_utc = datetime.now(timezone.utc)
    db.commit()
    db.refresh(req)
    return req



# -------------------------
# Phase 6G payroll foundation
# -------------------------

def _amount_int(value: Any) -> int:
    value = _blank_to_none(value)
    if value is None:
        return 0
    try:
        return int(round(float(value)))
    except Exception:
        raise HTTPException(status_code=400, detail=f"Invalid amount: {value}")


def list_salary_structures(db: Session, scope: EmployeeScope, *, employee_id: int | None = None, include_inactive: bool = True) -> list[EmployeeSalaryStructure]:
    stmt = select(EmployeeSalaryStructure).options(selectinload(EmployeeSalaryStructure.employee)).where(EmployeeSalaryStructure.tenant_id == scope.tenant_id)
    if scope.branch_id is not None:
        stmt = stmt.where(EmployeeSalaryStructure.branch_id == scope.branch_id)
    if employee_id:
        stmt = stmt.where(EmployeeSalaryStructure.employee_id == employee_id)
    if not include_inactive:
        stmt = stmt.where(EmployeeSalaryStructure.is_active.is_(True))
    return db.execute(stmt.order_by(EmployeeSalaryStructure.effective_from.desc(), EmployeeSalaryStructure.id.desc())).scalars().all()


def get_salary_structure_or_404(db: Session, structure_id: int, scope: EmployeeScope) -> EmployeeSalaryStructure:
    stmt = select(EmployeeSalaryStructure).options(selectinload(EmployeeSalaryStructure.employee)).where(EmployeeSalaryStructure.id == structure_id, EmployeeSalaryStructure.tenant_id == scope.tenant_id)
    if scope.branch_id is not None:
        stmt = stmt.where(EmployeeSalaryStructure.branch_id == scope.branch_id)
    row = db.execute(stmt).scalar_one_or_none()
    if not row:
        raise HTTPException(status_code=404, detail="Salary structure not found or not accessible.")
    return row


def _salary_payload(data: dict[str, Any]) -> dict[str, Any]:
    return {
        "effective_from": parse_date(data.get("effective_from")) or date.today(),
        "effective_to": parse_date(data.get("effective_to")),
        "pay_cycle": _blank_to_none(data.get("pay_cycle")) or "monthly",
        "monthly_ctc_amount": _amount_int(data.get("monthly_ctc_amount")),
        "basic_amount": _amount_int(data.get("basic_amount")),
        "hra_amount": _amount_int(data.get("hra_amount")),
        "allowance_amount": _amount_int(data.get("allowance_amount")),
        "employee_pf_amount": _amount_int(data.get("employee_pf_amount")),
        "employee_esi_amount": _amount_int(data.get("employee_esi_amount")),
        "professional_tax_amount": _amount_int(data.get("professional_tax_amount")),
        "tds_amount": _amount_int(data.get("tds_amount")),
        "other_deduction_amount": _amount_int(data.get("other_deduction_amount")),
        "is_active": bool(data.get("is_active", True)),
        "remarks": _blank_to_none(data.get("remarks")),
    }


def create_salary_structure(db: Session, actor: User, scope: EmployeeScope, data: dict[str, Any]) -> EmployeeSalaryStructure:
    employee_id = int(data.get("employee_id") or 0)
    emp = get_employee_or_404(db, employee_id, scope)
    payload = _salary_payload(data)
    if payload["effective_to"] and payload["effective_to"] < payload["effective_from"]:
        raise HTTPException(status_code=400, detail="Effective to date cannot be before effective from date.")
    exists = db.execute(select(EmployeeSalaryStructure).where(
        EmployeeSalaryStructure.tenant_id == emp.tenant_id,
        EmployeeSalaryStructure.employee_id == emp.id,
        EmployeeSalaryStructure.effective_from == payload["effective_from"],
    )).scalar_one_or_none()
    if exists:
        raise HTTPException(status_code=409, detail="Salary structure already exists for this employee and effective date.")
    row = EmployeeSalaryStructure(tenant_id=emp.tenant_id, branch_id=emp.branch_id, employee_id=emp.id, created_by_user_id=actor.id, updated_by_user_id=actor.id, **payload)
    db.add(row)
    db.commit()
    db.refresh(row)
    return row


def update_salary_structure(db: Session, actor: User, row: EmployeeSalaryStructure, data: dict[str, Any]) -> EmployeeSalaryStructure:
    payload = _salary_payload(data)
    if payload["effective_to"] and payload["effective_to"] < payload["effective_from"]:
        raise HTTPException(status_code=400, detail="Effective to date cannot be before effective from date.")
    for key, value in payload.items():
        setattr(row, key, value)
    row.updated_by_user_id = actor.id
    row.updated_at_utc = datetime.now(timezone.utc)
    db.commit()
    db.refresh(row)
    return row


def _active_salary_structure_for_employee(db: Session, emp: Employee, period_end: date) -> EmployeeSalaryStructure | None:
    stmt = select(EmployeeSalaryStructure).where(
        EmployeeSalaryStructure.employee_id == emp.id,
        EmployeeSalaryStructure.is_active.is_(True),
        EmployeeSalaryStructure.effective_from <= period_end,
        or_(EmployeeSalaryStructure.effective_to.is_(None), EmployeeSalaryStructure.effective_to >= period_end),
    ).order_by(EmployeeSalaryStructure.effective_from.desc(), EmployeeSalaryStructure.id.desc())
    return db.execute(stmt).scalar_one_or_none()


def list_payroll_runs(db: Session, scope: EmployeeScope) -> list[EmployeePayrollRun]:
    stmt = select(EmployeePayrollRun).where(EmployeePayrollRun.tenant_id == scope.tenant_id)
    if scope.branch_id is not None:
        stmt = stmt.where(EmployeePayrollRun.branch_id == scope.branch_id)
    return db.execute(stmt.order_by(EmployeePayrollRun.pay_year.desc(), EmployeePayrollRun.pay_month.desc(), EmployeePayrollRun.id.desc())).scalars().all()


def get_payroll_run_or_404(db: Session, run_id: int, scope: EmployeeScope) -> EmployeePayrollRun:
    stmt = select(EmployeePayrollRun).where(EmployeePayrollRun.id == run_id, EmployeePayrollRun.tenant_id == scope.tenant_id)
    if scope.branch_id is not None:
        stmt = stmt.where(EmployeePayrollRun.branch_id == scope.branch_id)
    row = db.execute(stmt).scalar_one_or_none()
    if not row:
        raise HTTPException(status_code=404, detail="Payroll run not found or not accessible.")
    return row


def create_payroll_run(db: Session, actor: User, scope: EmployeeScope, *, pay_year: int, pay_month: int, notes: str | None = None) -> EmployeePayrollRun:
    if scope.branch_id is None:
        raise HTTPException(status_code=400, detail="Select a branch before creating payroll run.")
    pay_year = int(pay_year)
    pay_month = int(pay_month)
    if pay_month < 1 or pay_month > 12:
        raise HTTPException(status_code=400, detail="Pay month must be between 1 and 12.")
    exists = db.execute(select(EmployeePayrollRun).where(EmployeePayrollRun.tenant_id == scope.tenant_id, EmployeePayrollRun.branch_id == scope.branch_id, EmployeePayrollRun.pay_year == pay_year, EmployeePayrollRun.pay_month == pay_month)).scalar_one_or_none()
    if exists:
        raise HTTPException(status_code=409, detail="Payroll run already exists for this branch and period.")
    row = EmployeePayrollRun(
        tenant_id=scope.tenant_id,
        branch_id=scope.branch_id,
        pay_year=pay_year,
        pay_month=pay_month,
        run_name=f"Payroll {pay_month:02d}/{pay_year}",
        status="draft",
        notes=_blank_to_none(notes),
        processed_by_user_id=actor.id,
    )
    db.add(row)
    db.commit()
    db.refresh(row)
    return row


def list_payslips(db: Session, scope: EmployeeScope, *, payroll_run_id: int | None = None, employee_id: int | None = None) -> list[EmployeePayslip]:
    stmt = select(EmployeePayslip).options(selectinload(EmployeePayslip.employee), selectinload(EmployeePayslip.payroll_run)).where(EmployeePayslip.tenant_id == scope.tenant_id)
    if scope.branch_id is not None:
        stmt = stmt.where(EmployeePayslip.branch_id == scope.branch_id)
    if payroll_run_id:
        stmt = stmt.where(EmployeePayslip.payroll_run_id == payroll_run_id)
    if employee_id:
        stmt = stmt.where(EmployeePayslip.employee_id == employee_id)
    return db.execute(stmt.order_by(EmployeePayslip.pay_year.desc(), EmployeePayslip.pay_month.desc(), EmployeePayslip.employee_id)).scalars().all()


def list_own_payslips(db: Session, actor: User) -> list[EmployeePayslip]:
    emp = get_employee_for_user(db, actor)
    if not emp:
        return []
    return db.execute(select(EmployeePayslip).options(selectinload(EmployeePayslip.payroll_run)).where(EmployeePayslip.employee_id == emp.id).order_by(EmployeePayslip.pay_year.desc(), EmployeePayslip.pay_month.desc())).scalars().all()


def generate_payslips_for_run(db: Session, actor: User, scope: EmployeeScope, run: EmployeePayrollRun) -> int:
    if run.status not in ("draft", "generated"):
        raise HTTPException(status_code=400, detail="Only draft/generated payroll runs can be regenerated.")
    period_end = date(run.pay_year, run.pay_month, 28)
    employees = db.execute(select(Employee).where(Employee.tenant_id == run.tenant_id, Employee.branch_id == run.branch_id, Employee.is_active.is_(True)).order_by(Employee.full_name)).scalars().all()
    created = 0
    gross_total = deduction_total = net_total = 0
    for emp in employees:
        structure = _active_salary_structure_for_employee(db, emp, period_end)
        if not structure:
            continue
        gross = int(structure.basic_amount or 0) + int(structure.hra_amount or 0) + int(structure.allowance_amount or 0)
        deductions = int(structure.employee_pf_amount or 0) + int(structure.employee_esi_amount or 0) + int(structure.professional_tax_amount or 0) + int(structure.tds_amount or 0) + int(structure.other_deduction_amount or 0)
        net = gross - deductions
        existing = db.execute(select(EmployeePayslip).where(EmployeePayslip.payroll_run_id == run.id, EmployeePayslip.employee_id == emp.id)).scalar_one_or_none()
        if not existing:
            existing = EmployeePayslip(tenant_id=run.tenant_id, branch_id=run.branch_id, payroll_run_id=run.id, employee_id=emp.id, generated_by_user_id=actor.id)
            db.add(existing)
            created += 1
        existing.salary_structure_id = structure.id
        existing.pay_year = run.pay_year
        existing.pay_month = run.pay_month
        existing.basic_amount = int(structure.basic_amount or 0)
        existing.hra_amount = int(structure.hra_amount or 0)
        existing.allowance_amount = int(structure.allowance_amount or 0)
        existing.gross_amount = gross
        existing.employee_pf_amount = int(structure.employee_pf_amount or 0)
        existing.employee_esi_amount = int(structure.employee_esi_amount or 0)
        existing.professional_tax_amount = int(structure.professional_tax_amount or 0)
        existing.tds_amount = int(structure.tds_amount or 0)
        existing.other_deduction_amount = int(structure.other_deduction_amount or 0)
        existing.deduction_amount = deductions
        existing.net_amount = net
        existing.status = "generated"
        gross_total += gross
        deduction_total += deductions
        net_total += net
    run.total_employees = len(list_payslips(db, scope, payroll_run_id=run.id))
    run.gross_amount = gross_total
    run.deduction_amount = deduction_total
    run.net_amount = net_total
    run.status = "generated"
    run.processed_by_user_id = actor.id
    run.updated_at_utc = datetime.now(timezone.utc)
    db.commit()
    db.refresh(run)
    return created


def approve_payroll_run(db: Session, actor: User, run: EmployeePayrollRun) -> EmployeePayrollRun:
    if run.status != "generated":
        raise HTTPException(status_code=400, detail="Only generated payroll runs can be approved.")
    run.status = "approved"
    run.approved_by_user_id = actor.id
    run.approved_at_utc = datetime.now(timezone.utc)
    for slip in db.execute(select(EmployeePayslip).where(EmployeePayslip.payroll_run_id == run.id)).scalars().all():
        slip.status = "approved"
    db.commit()
    db.refresh(run)
    return run


def mark_payroll_run_paid(db: Session, actor: User, run: EmployeePayrollRun) -> EmployeePayrollRun:
    if run.status != "approved":
        raise HTTPException(status_code=400, detail="Only approved payroll runs can be marked paid.")
    run.status = "paid"
    run.paid_by_user_id = actor.id
    run.paid_at_utc = datetime.now(timezone.utc)
    for slip in db.execute(select(EmployeePayslip).where(EmployeePayslip.payroll_run_id == run.id)).scalars().all():
        slip.status = "paid"
    db.commit()
    db.refresh(run)
    return run


# -------------------------
# Phase 7A Employee Work Dashboard
# -------------------------

PRIORITY_BUCKETS = [
    ("urgent", "Urgent"),
    ("high", "High"),
    ("normal", "Normal"),
    ("low", "Low"),
    ("none", "No Priority"),
]


def _task_priority_key(task: ClientServiceTaskInstance) -> str:
    value = (getattr(task, "priority", None) or "normal").strip().lower()
    return value if value in {code for code, _ in PRIORITY_BUCKETS} else "normal"


def _task_date_bucket(task: ClientServiceTaskInstance, *, today: date) -> str:
    target = getattr(task, "internal_target_date", None)
    status = (getattr(task, "status", "") or "").lower()
    if status in CLOSED_TASK_STATUSES:
        return "Completed / Closed"
    if target and target < today:
        return "Overdue"
    if target and target == today:
        return "Due Today"
    if target:
        return "Upcoming"
    return "No Target Date"


def _subscription_label(subscription: ClientServiceSubscription | None, task: ClientServiceTaskInstance) -> str:
    catalogue = getattr(task, "catalogue", None)
    service_name = getattr(catalogue, "service_name", None) or getattr(catalogue, "service_code", None)
    if not service_name and subscription:
        cat = getattr(subscription, "catalogue", None)
        service_name = getattr(cat, "service_name", None) or getattr(cat, "service_code", None)
    fy = getattr(subscription, "financial_year", None) or getattr(task, "financial_year", None) or "-"
    due = getattr(subscription, "current_due_date", None) if subscription else None
    label = f"{service_name or 'Service Engagement'} · FY {fy}"
    if due:
        label = f"{label} · Due {due}"
    return label


def _task_status_label(task: ClientServiceTaskInstance) -> str:
    return dict(TASK_STATUSES).get(getattr(task, "status", ""), getattr(task, "status", ""))


def _task_priority_label(task: ClientServiceTaskInstance) -> str:
    return dict(TASK_PRIORITIES).get(getattr(task, "priority", ""), getattr(task, "priority", ""))


def _employee_owned_task_condition(scope: EmployeeScope):
    """Return the employee self-work ownership condition.

    Existing explicit task assignments remain authoritative. For legacy/open
    engagements created before a staff assignee was propagated to task rows, an
    unassigned Staff/Employee task is also treated as belonging to the
    engagement's assigned staff member. Tasks explicitly assigned to another
    user are never taken over by this fallback.
    """
    role_name = func.lower(func.coalesce(ClientServiceTaskInstance.default_role_name, ""))
    staff_role = or_(
        role_name.contains("staff"),
        role_name.contains("employee"),
    )
    assigned_via_engagement = (
        ClientServiceTaskInstance.assigned_to_user_id.is_(None)
        & staff_role
        & ClientServiceTaskInstance.subscription.has(
            ClientServiceSubscription.assigned_staff_user_id == scope.own_user_id
        )
    )
    return or_(
        ClientServiceTaskInstance.assigned_to_user_id == scope.own_user_id,
        assigned_via_engagement,
    )


def list_employee_work_dashboard(
    db: Session,
    scope: EmployeeScope,
    *,
    q: str = "",
    status: str = "open",
    financial_year: str | None = None,
) -> dict[str, Any]:
    """Return assigned service/engagement tasks grouped for the employee workspace.

    Phase 7A is intentionally read-only. It does not modify existing engagement/task
    creation flows. It groups existing ClientServiceTaskInstance records assigned to
    the logged-in user by priority -> client -> engagement/service subscription.
    """
    today = date.today()
    status_filter = (status or "open").strip().lower()

    stmt = (
        select(ClientServiceTaskInstance)
        .options(
            selectinload(ClientServiceTaskInstance.client),
            selectinload(ClientServiceTaskInstance.catalogue),
            selectinload(ClientServiceTaskInstance.subscription).selectinload(ClientServiceSubscription.catalogue),
            selectinload(ClientServiceTaskInstance.subscription).selectinload(ClientServiceSubscription.assigned_partner),
            selectinload(ClientServiceTaskInstance.subscription).selectinload(ClientServiceSubscription.assigned_manager),
            selectinload(ClientServiceTaskInstance.subscription).selectinload(ClientServiceSubscription.assigned_staff),
            selectinload(ClientServiceTaskInstance.subscription).selectinload(ClientServiceSubscription.review_partner),
            selectinload(ClientServiceTaskInstance.assigned_to),
            selectinload(ClientServiceTaskInstance.template),
            selectinload(ClientServiceTaskInstance.comments).selectinload(ServiceTaskComment.created_by),
        )
        .where(
            ClientServiceTaskInstance.tenant_id == scope.tenant_id,
            _employee_owned_task_condition(scope),
            ClientServiceTaskInstance.is_active.is_(True),
        )
    )
    if scope.branch_id is not None:
        stmt = stmt.where(ClientServiceTaskInstance.branch_id == scope.branch_id)
    if financial_year:
        stmt = stmt.where(ClientServiceTaskInstance.financial_year == financial_year.strip())

    if status_filter == "open":
        stmt = stmt.where(ClientServiceTaskInstance.status.notin_(list(CLOSED_TASK_STATUSES)))
    elif status_filter == "closed":
        stmt = stmt.where(ClientServiceTaskInstance.status.in_(list(CLOSED_TASK_STATUSES)))
    elif status_filter in {code for code, _ in TASK_STATUSES}:
        stmt = stmt.where(ClientServiceTaskInstance.status == status_filter)

    if q.strip():
        like = f"%{q.strip()}%"
        stmt = stmt.where(
            or_(
                ClientServiceTaskInstance.task_name.ilike(like),
                ClientServiceTaskInstance.description.ilike(like),
                ClientServiceTaskInstance.client.has(or_(
                    Client.client_name.ilike(like),
                    Client.client_code.ilike(like),
                )),
                ClientServiceTaskInstance.catalogue.has(or_(
                    ServiceCatalogue.service_name.ilike(like),
                    ServiceCatalogue.service_code.ilike(like),
                )),
            )
        )

    tasks = db.execute(
        stmt.order_by(
            ClientServiceTaskInstance.priority.desc(),
            ClientServiceTaskInstance.internal_target_date.is_(None),
            ClientServiceTaskInstance.internal_target_date.asc(),
            ClientServiceTaskInstance.sequence_no.asc(),
            ClientServiceTaskInstance.id.desc(),
        )
    ).scalars().all()

    summary = {
        "total": len(tasks),
        "open": 0,
        "overdue": 0,
        "due_today": 0,
        "upcoming": 0,
        "completed": 0,
    }
    priority_map: dict[str, dict[str, Any]] = {
        code: {"code": code, "label": label, "task_count": 0, "clients": []}
        for code, label in PRIORITY_BUCKETS
    }
    client_lookup: dict[tuple[str, int], dict[str, Any]] = {}
    engagement_lookup: dict[tuple[str, int, int], dict[str, Any]] = {}

    for task in tasks:
        is_closed = (task.status or "") in CLOSED_TASK_STATUSES
        if is_closed:
            summary["completed"] += 1
        else:
            summary["open"] += 1
        target = task.internal_target_date
        if target and target < today and not is_closed:
            summary["overdue"] += 1
        elif target and target == today and not is_closed:
            summary["due_today"] += 1
        elif target and target > today and not is_closed:
            summary["upcoming"] += 1

        priority_key = _task_priority_key(task)
        priority_bucket = priority_map.setdefault(priority_key, {"code": priority_key, "label": priority_key.title(), "task_count": 0, "clients": []})
        priority_bucket["task_count"] += 1

        client = getattr(task, "client", None)
        client_id = getattr(client, "id", 0) or 0
        client_key = (priority_key, client_id)
        if client_key not in client_lookup:
            client_group = {
                "client": client,
                "client_name": getattr(client, "client_name", None) or "Unlinked Client",
                "client_code": getattr(client, "client_code", None) or "",
                "task_count": 0,
                "engagements": [],
            }
            client_lookup[client_key] = client_group
            priority_bucket["clients"].append(client_group)
        client_group = client_lookup[client_key]
        client_group["task_count"] += 1

        subscription = getattr(task, "subscription", None)
        engagement_id = getattr(subscription, "id", None) or getattr(task, "subscription_id", 0) or 0
        engagement_key = (priority_key, client_id, engagement_id)
        if engagement_key not in engagement_lookup:
            engagement_group = {
                "subscription": subscription,
                "label": _subscription_label(subscription, task),
                "status": getattr(subscription, "status", None) or "-",
                "due_date": getattr(subscription, "current_due_date", None) if subscription else None,
                "task_count": 0,
                "open_count": 0,
                "completed_count": 0,
                "tasks": [],
            }
            engagement_lookup[engagement_key] = engagement_group
            client_group["engagements"].append(engagement_group)
        engagement_group = engagement_lookup[engagement_key]
        engagement_group["task_count"] += 1
        if is_closed:
            engagement_group["completed_count"] += 1
        else:
            engagement_group["open_count"] += 1

        task.comment_count = len([c for c in getattr(task, "comments", []) if not getattr(c, "is_deleted", False)])
        task.latest_comment = next((c for c in getattr(task, "comments", []) if not getattr(c, "is_deleted", False)), None)
        task.status_label = _task_status_label(task)
        task.priority_label = _task_priority_label(task)
        task.date_bucket = _task_date_bucket(task, today=today)
        task.is_overdue = bool(target and target < today and not is_closed)
        task.is_due_today = bool(target and target == today and not is_closed)
        engagement_group["tasks"].append(task)

    groups = [priority_map[code] for code, _label in PRIORITY_BUCKETS if priority_map.get(code, {}).get("task_count")]
    return {"summary": summary, "groups": groups, "q": q, "status": status_filter, "today": today}



def _employee_work_task_query(db: Session, scope: EmployeeScope, *, assigned_only: bool = True, financial_year: str | None = None):
    """Base query for employee self-work views.

    Kept separate for Phase 7I so the staff kanban and engagement board reuse the
    same tenant/branch/assignee scoping and do not bypass existing security rules.
    """
    stmt = (
        select(ClientServiceTaskInstance)
        .options(
            selectinload(ClientServiceTaskInstance.client),
            selectinload(ClientServiceTaskInstance.catalogue),
            selectinload(ClientServiceTaskInstance.subscription).selectinload(ClientServiceSubscription.catalogue),
            selectinload(ClientServiceTaskInstance.subscription).selectinload(ClientServiceSubscription.assigned_partner),
            selectinload(ClientServiceTaskInstance.subscription).selectinload(ClientServiceSubscription.assigned_manager),
            selectinload(ClientServiceTaskInstance.subscription).selectinload(ClientServiceSubscription.assigned_staff),
            selectinload(ClientServiceTaskInstance.subscription).selectinload(ClientServiceSubscription.review_partner),
            selectinload(ClientServiceTaskInstance.assigned_to),
            selectinload(ClientServiceTaskInstance.comments).selectinload(ServiceTaskComment.created_by),
        )
        .where(
            ClientServiceTaskInstance.tenant_id == scope.tenant_id,
            ClientServiceTaskInstance.is_active.is_(True),
        )
    )
    if assigned_only:
        stmt = stmt.where(_employee_owned_task_condition(scope))
    if scope.branch_id is not None:
        stmt = stmt.where(ClientServiceTaskInstance.branch_id == scope.branch_id)
    if financial_year:
        stmt = stmt.where(ClientServiceTaskInstance.financial_year == financial_year.strip())
    return stmt


def _phase7i_task_card_enrich(task: ClientServiceTaskInstance, *, today: date) -> None:
    is_closed = (task.status or "") in CLOSED_TASK_STATUSES
    target = getattr(task, "internal_target_date", None)
    task.comment_count = len([c for c in getattr(task, "comments", []) if not getattr(c, "is_deleted", False)])
    task.latest_comment = next((c for c in getattr(task, "comments", []) if not getattr(c, "is_deleted", False)), None)
    task.status_label = _task_status_label(task)
    task.priority_label = _task_priority_label(task)
    task.date_bucket = _task_date_bucket(task, today=today)
    task.is_overdue = bool(target and target < today and not is_closed)
    task.is_due_today = bool(target and target == today and not is_closed)
    task.engagement_label = _subscription_label(getattr(task, "subscription", None), task)


def _workflow_manual_blocker(subscription: ClientServiceSubscription | None) -> dict[str, Any] | None:
    if not subscription or not getattr(subscription, "workflow_pause_reason", None):
        return None
    code = str(subscription.workflow_pause_reason).strip()
    return {
        "type": "manual",
        "code": code,
        "label": WORKFLOW_HOLD_REASONS.get(code, code.replace("_", " ").title()),
        "notes": (getattr(subscription, "workflow_pause_notes", None) or "").strip() or None,
        "follow_up_date": getattr(subscription, "workflow_follow_up_date", None),
    }


def _workflow_automatic_blocker(subscription: ClientServiceSubscription | None, tasks: list[ClientServiceTaskInstance]) -> dict[str, Any] | None:
    if subscription and getattr(subscription, "quality_workflow_required", False):
        qstatus = (getattr(subscription, "quality_acceptance_status", None) or "").strip().lower()
        if qstatus not in {"approved", "not_required"}:
            return {"type": "automatic", "code": "quality_acceptance", "label": "AQMM acceptance pending", "notes": getattr(subscription, "quality_block_reason", None), "follow_up_date": None}
    checks = [
        ("rework", "rework_status", {"open"}, "Rework response pending"),
        ("manager_review", "manager_review_status", {"pending", "rework_required"}, "Manager review pending"),
        ("partner_review", "partner_review_status", {"pending", "rework_required"}, "Partner review pending"),
        ("review_partner_review", "review_partner_review_status", {"pending", "rework_required"}, "Review Partner review pending"),
    ]
    for code, field, values, label in checks:
        for task in tasks:
            if (getattr(task, field, None) or "").strip().lower() in values:
                return {"type": "automatic", "code": code, "label": label, "notes": task.task_name, "follow_up_date": None}
    return None


def pause_employee_engagement_workflow(db: Session, scope: EmployeeScope, engagement_id: int, *, reason: str, notes: str, follow_up_date: str, actor_user_id: int, financial_year: str | None = None) -> ClientServiceSubscription:
    code = (reason or "").strip().lower()
    if code not in WORKFLOW_HOLD_REASONS:
        raise ValueError("Select a valid hold reason.")
    clean_notes = (notes or "").strip()
    if code == "other" and not clean_notes:
        raise ValueError("Remarks are required for Other hold reason.")
    tasks = db.execute(_employee_work_task_query(db, scope, assigned_only=True, financial_year=financial_year).where(ClientServiceTaskInstance.subscription_id == engagement_id)).scalars().all()
    if not tasks:
        raise HTTPException(status_code=404, detail="Engagement work not found or not assigned to you")
    subscription = tasks[0].subscription
    if subscription.is_locked:
        raise ValueError("Locked engagement cannot be paused.")
    parsed_follow_up = parse_date(follow_up_date) if (follow_up_date or "").strip() else None
    subscription.workflow_pause_reason = code
    subscription.workflow_pause_notes = clean_notes or None
    subscription.workflow_follow_up_date = parsed_follow_up
    subscription.workflow_paused_at_utc = datetime.now(timezone.utc)
    subscription.workflow_paused_by_user_id = actor_user_id
    subscription.workflow_resumed_at_utc = None
    subscription.workflow_resumed_by_user_id = None
    subscription.updated_by_user_id = actor_user_id
    manager_id = getattr(subscription, "assigned_manager_user_id", None)
    if manager_id and manager_id != actor_user_id:
        create_alert(db, user_id=manager_id, title=f"Engagement paused: {_subscription_label(subscription, tasks[0])}", message=f"{WORKFLOW_HOLD_REASONS[code]}" + (f". {clean_notes}" if clean_notes else "") + (f" Follow-up: {parsed_follow_up}." if parsed_follow_up else ""), tenant_id=subscription.tenant_id, branch_id=subscription.branch_id, role_context="manager", alert_type="clarification", priority="high", target_url=f"/employees/work/engagements/{engagement_id}", created_by_user_id=actor_user_id, commit=False)
    db.commit(); db.refresh(subscription); return subscription


def resume_employee_engagement_workflow(db: Session, scope: EmployeeScope, engagement_id: int, *, actor_user_id: int, financial_year: str | None = None) -> ClientServiceSubscription:
    tasks = db.execute(_employee_work_task_query(db, scope, assigned_only=True, financial_year=financial_year).where(ClientServiceTaskInstance.subscription_id == engagement_id)).scalars().all()
    if not tasks:
        raise HTTPException(status_code=404, detail="Engagement work not found or not assigned to you")
    subscription = tasks[0].subscription
    if subscription.is_locked:
        raise ValueError("Locked engagement cannot be resumed.")
    old_reason = WORKFLOW_HOLD_REASONS.get(getattr(subscription, "workflow_pause_reason", None), "Paused")
    subscription.workflow_pause_reason = None
    subscription.workflow_pause_notes = None
    subscription.workflow_follow_up_date = None
    subscription.workflow_resumed_at_utc = datetime.now(timezone.utc)
    subscription.workflow_resumed_by_user_id = actor_user_id
    subscription.updated_by_user_id = actor_user_id
    manager_id = getattr(subscription, "assigned_manager_user_id", None)
    if manager_id and manager_id != actor_user_id:
        create_alert(db, user_id=manager_id, title=f"Engagement resumed: {_subscription_label(subscription, tasks[0])}", message=f"The employee resumed work previously held for: {old_reason}.", tenant_id=subscription.tenant_id, branch_id=subscription.branch_id, role_context="manager", alert_type="general", priority="normal", target_url=f"/employees/work/engagements/{engagement_id}", created_by_user_id=actor_user_id, commit=False)
    db.commit(); db.refresh(subscription); return subscription



def _employee_work_scope_display_map(
    db: Session,
    *,
    tenant_id: int,
    subscriptions: list[ClientServiceSubscription],
) -> dict[int, dict[str, str]]:
    """Resolve the display identity for Employee My Work without changing ownership.

    Display priority:
    - registration scope: registration trade name -> business-unit trade/name ->
      branch name -> registration legal name -> client trade name -> client name
    - business-unit scope: business-unit trade/name -> client trade name -> client name
    - client-branch scope: branch name -> business-unit trade/name -> client trade/name
    - client scope: client trade name -> client name

    The legal client remains available as context for scoped engagements.
    """
    subscriptions = [row for row in subscriptions if row is not None]
    if not subscriptions:
        return {}

    business_ids = {
        int(row.business_unit_id)
        for row in subscriptions
        if getattr(row, "business_unit_id", None)
    }
    branch_ids = {
        int(row.client_branch_id)
        for row in subscriptions
        if getattr(row, "client_branch_id", None)
    }
    registration_ids = {
        int(row.registration_id)
        for row in subscriptions
        if getattr(row, "registration_id", None)
    }

    businesses: dict[int, ClientBusinessUnit] = {}
    if business_ids:
        businesses = {
            row.id: row
            for row in db.execute(
                select(ClientBusinessUnit).where(
                    ClientBusinessUnit.tenant_id == tenant_id,
                    ClientBusinessUnit.id.in_(business_ids),
                )
            ).scalars().all()
        }

    branches: dict[int, ClientBranch] = {}
    if branch_ids:
        branches = {
            row.id: row
            for row in db.execute(
                select(ClientBranch).where(
                    ClientBranch.tenant_id == tenant_id,
                    ClientBranch.id.in_(branch_ids),
                )
            ).scalars().all()
        }

    registrations: dict[int, ClientRegistration] = {}
    registration_types: dict[int, RegistrationType] = {}
    if registration_ids:
        for registration, registration_type in db.execute(
            select(ClientRegistration, RegistrationType)
            .join(
                RegistrationType,
                RegistrationType.id == ClientRegistration.registration_type_id,
            )
            .where(
                ClientRegistration.tenant_id == tenant_id,
                ClientRegistration.id.in_(registration_ids),
            )
        ).all():
            registrations[registration.id] = registration
            registration_types[registration.id] = registration_type

        # Registration scope can carry business/branch references even when the
        # engagement's denormalised scope columns are blank.
        extra_business_ids = {
            int(row.business_unit_id)
            for row in registrations.values()
            if row.business_unit_id and row.business_unit_id not in businesses
        }
        if extra_business_ids:
            businesses.update({
                row.id: row
                for row in db.execute(
                    select(ClientBusinessUnit).where(
                        ClientBusinessUnit.tenant_id == tenant_id,
                        ClientBusinessUnit.id.in_(extra_business_ids),
                    )
                ).scalars().all()
            })

        extra_branch_ids = {
            int(row.client_branch_id)
            for row in registrations.values()
            if row.client_branch_id and row.client_branch_id not in branches
        }
        if extra_branch_ids:
            branches.update({
                row.id: row
                for row in db.execute(
                    select(ClientBranch).where(
                        ClientBranch.tenant_id == tenant_id,
                        ClientBranch.id.in_(extra_branch_ids),
                    )
                ).scalars().all()
            })

    result: dict[int, dict[str, str]] = {}

    for subscription in subscriptions:
        client = getattr(subscription, "client", None)
        client_name = (getattr(client, "client_name", None) or "").strip()
        client_trade_name = (getattr(client, "trade_name", None) or "").strip()
        client_code = (getattr(client, "client_code", None) or "").strip()

        primary = client_trade_name or client_name or "Unlinked Client"
        secondary = ""
        context = ""
        scope_label = "Client"

        scope_type = (getattr(subscription, "scope_type", None) or "client").strip().lower()

        if scope_type == "registration":
            registration = registrations.get(getattr(subscription, "registration_id", None))
            registration_type = registration_types.get(getattr(subscription, "registration_id", None))

            if registration:
                business = businesses.get(
                    getattr(registration, "business_unit_id", None)
                    or getattr(subscription, "business_unit_id", None)
                )
                branch = branches.get(
                    getattr(registration, "client_branch_id", None)
                    or getattr(subscription, "client_branch_id", None)
                )

                business_trade_name = (
                    (getattr(business, "trade_name", None) or "").strip()
                    if business else ""
                )
                business_name = (
                    (getattr(business, "business_name", None) or "").strip()
                    if business else ""
                )
                branch_name = (
                    (getattr(branch, "branch_name", None) or "").strip()
                    if branch else ""
                )

                primary = (
                    (getattr(registration, "trade_name", None) or "").strip()
                    or business_trade_name
                    or business_name
                    or branch_name
                    or (getattr(registration, "legal_name", None) or "").strip()
                    or client_trade_name
                    or client_name
                    or "Unlinked Client"
                )

                registration_code = (
                    (getattr(registration_type, "code", None) or "").strip()
                    if registration_type else ""
                )
                registration_number = (
                    getattr(registration, "registration_number", None) or ""
                ).strip()

                if registration_code and registration_number:
                    secondary = f"{registration_code} · {registration_number}"
                else:
                    secondary = registration_number or registration_code

                if client_name and client_name != primary:
                    context = f"Client: {client_name}"

            scope_label = "Registration"

        elif scope_type == "business_unit":
            business = businesses.get(getattr(subscription, "business_unit_id", None))
            if business:
                primary = (
                    (getattr(business, "trade_name", None) or "").strip()
                    or (getattr(business, "business_name", None) or "").strip()
                    or primary
                )
                secondary = (getattr(business, "business_code", None) or "").strip()
                if client_name and client_name != primary:
                    context = f"Client: {client_name}"
            scope_label = "Business Unit"

        elif scope_type == "client_branch":
            branch = branches.get(getattr(subscription, "client_branch_id", None))
            business = businesses.get(
                getattr(branch, "business_unit_id", None)
                if branch else getattr(subscription, "business_unit_id", None)
            )
            if branch:
                primary = (getattr(branch, "branch_name", None) or "").strip() or primary
                secondary = (getattr(branch, "branch_code", None) or "").strip()

                business_display = ""
                if business:
                    business_display = (
                        (getattr(business, "trade_name", None) or "").strip()
                        or (getattr(business, "business_name", None) or "").strip()
                    )

                context_parts: list[str] = []
                if business_display and business_display != primary:
                    context_parts.append(business_display)
                if client_name and client_name not in {primary, business_display}:
                    context_parts.append(f"Client: {client_name}")
                context = " · ".join(context_parts)
            scope_label = "Client Branch"

        result[int(subscription.id)] = {
            "primary": primary,
            "secondary": secondary,
            "context": context,
            "scope_label": scope_label,
            "client_code": client_code,
            "client_name": client_name,
            "client_trade_name": client_trade_name,
        }

    return result

def list_employee_work_kanban(
    db: Session,
    scope: EmployeeScope,
    *,
    q: str = "",
    status: str = "open",
    financial_year: str | None = None,
    completed_page: int = 1,
    completed_page_size: int = 25,
) -> dict[str, Any]:
    """Return exactly one employee-board card per assigned engagement.

    The task engine remains unchanged. The board derives an engagement state from
    the staff member's assigned tasks using this precedence:

    completed -> all assigned tasks are closed
    blocked   -> at least one assigned task is blocked
    in_progress -> at least one assigned task is started/completed
    pending   -> no assigned task has started
    """
    today = date.today()
    status_filter = (status or "open").strip().lower()
    valid_filters = {"open", "closed", "pending", "in_progress", "blocked", "completed"}
    if status_filter not in valid_filters:
        status_filter = "open"

    stmt = _employee_work_task_query(
        db,
        scope,
        assigned_only=True,
        financial_year=financial_year,
    )
    if q.strip():
        like = f"%{q.strip()}%"
        stmt = stmt.where(
            or_(
                ClientServiceTaskInstance.task_name.ilike(like),
                ClientServiceTaskInstance.description.ilike(like),
                ClientServiceTaskInstance.client.has(
                    or_(Client.client_name.ilike(like), Client.trade_name.ilike(like), Client.client_code.ilike(like))
                ),
                ClientServiceTaskInstance.catalogue.has(
                    or_(
                        ServiceCatalogue.service_name.ilike(like),
                        ServiceCatalogue.service_code.ilike(like),
                    )
                ),
            )
        )

    tasks = db.execute(
        stmt.order_by(
            ClientServiceTaskInstance.subscription_id.asc(),
            ClientServiceTaskInstance.sequence_no.asc(),
            ClientServiceTaskInstance.internal_target_date.is_(None),
            ClientServiceTaskInstance.internal_target_date.asc(),
            ClientServiceTaskInstance.id.asc(),
        )
    ).scalars().all()

    work_subscriptions: dict[int, ClientServiceSubscription] = {}
    for task in tasks:
        subscription = getattr(task, "subscription", None)
        if subscription is not None and getattr(subscription, "id", None):
            work_subscriptions[int(subscription.id)] = subscription

    scope_display_map = _employee_work_scope_display_map(
        db,
        tenant_id=scope.tenant_id,
        subscriptions=list(work_subscriptions.values()),
    )

    columns = [
        {"code": "in_progress", "label": "In Progress", "cards": []},
        {"code": "pending", "label": "Pending", "cards": []},
        {"code": "blocked", "label": "Blocked", "cards": []},
        {"code": "completed", "label": "Completed", "cards": []},
    ]
    column_lookup = {column["code"]: column for column in columns}
    engagement_lookup: dict[int, dict[str, Any]] = {}

    for task in tasks:
        _phase7i_task_card_enrich(task, today=today)
        subscription = getattr(task, "subscription", None)
        engagement_id = (
            getattr(subscription, "id", None)
            or getattr(task, "subscription_id", 0)
            or 0
        )
        if engagement_id not in engagement_lookup:
            client = getattr(task, "client", None)
            scope_display = scope_display_map.get(
                int(engagement_id),
                {
                    "primary": (
                        (getattr(client, "trade_name", None) or "").strip()
                        or getattr(client, "client_name", None)
                        or "Unlinked Client"
                    ),
                    "secondary": "",
                    "context": "",
                    "scope_label": "Client",
                    "client_code": getattr(client, "client_code", None) or "",
                },
            )
            engagement_lookup[engagement_id] = {
                "engagement_id": engagement_id,
                "subscription": subscription,
                "label": _subscription_label(subscription, task),
                "client_name": scope_display["primary"],
                "client_code": scope_display.get("client_code", "") or getattr(client, "client_code", None) or "",
                "scope_secondary": scope_display.get("secondary", ""),
                "scope_context": scope_display.get("context", ""),
                "scope_label": scope_display.get("scope_label", "Client"),
                "service_name": (
                    getattr(getattr(subscription, "catalogue", None), "service_name", None)
                    or getattr(getattr(task, "catalogue", None), "service_name", None)
                    or "Service Engagement"
                ),
                "financial_year": (
                    getattr(subscription, "financial_year", None)
                    or getattr(task, "financial_year", None)
                    or "-"
                ),
                "due_date": (
                    getattr(subscription, "current_due_date", None)
                    if subscription
                    else getattr(task, "internal_target_date", None)
                ),
                "task_count": 0,
                "pending_count": 0,
                "in_progress_count": 0,
                "blocked_count": 0,
                "completed_count": 0,
                "overdue_count": 0,
                "due_today_count": 0,
                "latest_comment": None,
                "blocked_reason": None,
                "blocked_task_name": None,
                "next_task_name": None,
                "tasks": [],
            }

        card = engagement_lookup[engagement_id]
        task_status = (task.status or "pending").strip().lower()
        is_closed = task_status in CLOSED_TASK_STATUSES
        card["task_count"] += 1
        card["tasks"].append(task)
        if is_closed:
            card["completed_count"] += 1
        elif task_status == "blocked":
            card["blocked_count"] += 1
            if not card["blocked_reason"]:
                card["blocked_reason"] = (
                    (getattr(task, "remarks", None) or "").strip()
                    or (
                        getattr(task.latest_comment, "message", None)
                        if task.latest_comment
                        else None
                    )
                    or "Work is awaiting a dependency or clarification."
                )
                card["blocked_task_name"] = task.task_name
        elif task_status == "in_progress":
            card["in_progress_count"] += 1
        else:
            card["pending_count"] += 1

        if task.is_overdue:
            card["overdue_count"] += 1
        if task.is_due_today:
            card["due_today_count"] += 1
        if task.latest_comment and not card["latest_comment"]:
            card["latest_comment"] = task.latest_comment
        if not is_closed and not card["next_task_name"]:
            card["next_task_name"] = task.task_name

    summary = {
        "total": 0,
        "open": 0,
        "pending": 0,
        "in_progress": 0,
        "blocked": 0,
        "completed": 0,
        "overdue": 0,
        "due_today": 0,
    }

    for card in engagement_lookup.values():
        total = card["task_count"]
        completed = card["completed_count"]
        weighted = _weighted_progress(card.get("tasks", []))
        card.update(weighted)
        card["simple_progress_percent"] = int(round((completed * 100) / total)) if total else 0
        card["open_count"] = max(total - completed, 0)
        card["engagement_team"] = _engagement_team(card.get("subscription"))
        card["sla"] = _engagement_sla(card.get("subscription"), card.get("tasks", []))

        manual_blocker = _workflow_manual_blocker(card.get("subscription"))
        automatic_blocker = _workflow_automatic_blocker(card.get("subscription"), card.get("tasks", []))
        card["manual_blocker"] = manual_blocker
        card["automatic_blocker"] = automatic_blocker
        card["workflow_blocker"] = manual_blocker or automatic_blocker
        if card["workflow_blocker"]:
            card["blocked_reason"] = card["workflow_blocker"]["label"]
            card["blocked_notes"] = card["workflow_blocker"].get("notes")
            card["follow_up_date"] = card["workflow_blocker"].get("follow_up_date")

        if total and completed == total:
            card_status = "completed"
            card["action_label"] = "View"
        elif card["workflow_blocker"] or card["blocked_count"]:
            card_status = "blocked"
            card["action_label"] = "Open / Follow Up"
        elif card["in_progress_count"] or completed:
            card_status = "in_progress"
            card["action_label"] = "Continue"
        else:
            card_status = "pending"
            card["action_label"] = "Start"

        card["workflow_status"] = card_status
        card["is_overdue"] = bool(card["overdue_count"])
        card["is_due_today"] = bool(card["due_today_count"])

        include = (
            status_filter == card_status
            or (status_filter == "open" and card_status != "completed")
            or (status_filter == "closed" and card_status == "completed")
        )
        if not include:
            continue

        column_lookup[card_status]["cards"].append(card)
        summary["total"] += 1
        summary[card_status] += 1
        summary["completed" if card_status == "completed" else "open"] += 1
        if card["is_overdue"]:
            summary["overdue"] += 1
        if card["is_due_today"]:
            summary["due_today"] += 1

    for column in columns:
        column["cards"].sort(
            key=lambda card: (
                card["due_date"] is None,
                card["due_date"] or date.max,
                card["client_name"].lower(),
                card["service_name"].lower(),
            )
        )
        column["total_cards"] = len(column["cards"])

    try:
        completed_page = max(int(completed_page or 1), 1)
    except (TypeError, ValueError):
        completed_page = 1
    try:
        completed_page_size = max(1, min(int(completed_page_size or 25), 100))
    except (TypeError, ValueError):
        completed_page_size = 25

    completed_column = column_lookup["completed"]
    completed_total = int(completed_column["total_cards"])
    completed_pages = max(1, (completed_total + completed_page_size - 1) // completed_page_size)
    if completed_page > completed_pages:
        completed_page = completed_pages
    completed_start = (completed_page - 1) * completed_page_size
    completed_end = completed_start + completed_page_size
    completed_column["cards"] = completed_column["cards"][completed_start:completed_end]

    completed_pagination = {
        "page": completed_page,
        "page_size": completed_page_size,
        "total": completed_total,
        "pages": completed_pages,
        "has_previous": completed_page > 1,
        "has_next": completed_page < completed_pages,
        "previous_page": completed_page - 1 if completed_page > 1 else None,
        "next_page": completed_page + 1 if completed_page < completed_pages else None,
        "start": completed_start + 1 if completed_total else 0,
        "end": min(completed_end, completed_total),
    }

    return {
        "summary": summary,
        "columns": columns,
        "q": q,
        "status": status_filter,
        "today": today,
        "completed_pagination": completed_pagination,
    }


def start_employee_engagement_workflow(
    db: Session,
    scope: EmployeeScope,
    engagement_id: int,
    *,
    actor_user_id: int,
    financial_year: str | None = None,
) -> ClientServiceTaskInstance:
    """Start the first pending task assigned to the employee in an engagement.

    Existing task status, lock, tenant, branch and financial-year controls are
    reused. No engagement or task records are duplicated.
    """
    stmt = _employee_work_task_query(
        db,
        scope,
        assigned_only=True,
        financial_year=financial_year,
    ).where(ClientServiceTaskInstance.subscription_id == engagement_id)
    tasks = db.execute(
        stmt.order_by(
            ClientServiceTaskInstance.sequence_no.asc(),
            ClientServiceTaskInstance.internal_target_date.is_(None),
            ClientServiceTaskInstance.internal_target_date.asc(),
            ClientServiceTaskInstance.id.asc(),
        )
    ).scalars().all()
    if not tasks:
        raise HTTPException(
            status_code=404,
            detail="Engagement work not found or not assigned to you",
        )

    already_started = next(
        (
            task
            for task in tasks
            if (task.status or "").strip().lower() == "in_progress"
        ),
        None,
    )
    if already_started:
        return already_started

    first_pending = next(
        (
            task
            for task in tasks
            if (task.status or "pending").strip().lower() == "pending"
        ),
        None,
    )
    if not first_pending:
        return tasks[0]

    return update_own_service_task_status(
        db,
        scope,
        first_pending.id,
        status="in_progress",
        remarks=None,
        actor_user_id=actor_user_id,
        financial_year=financial_year,
    )

def list_employee_engagement_documents(db: Session, scope: EmployeeScope, engagement_id: int, *, financial_year: str | None = None) -> list[EngagementDocument]:
    stmt = (
        select(EngagementDocument)
        .options(selectinload(EngagementDocument.versions))
        .where(
            EngagementDocument.tenant_id == scope.tenant_id,
            EngagementDocument.engagement_id == engagement_id,
            EngagementDocument.is_deleted.is_(False),
        )
        .order_by(EngagementDocument.document_type.asc(), EngagementDocument.updated_at_utc.desc(), EngagementDocument.id.desc())
    )
    if scope.branch_id is not None:
        stmt = stmt.where(or_(EngagementDocument.branch_id == scope.branch_id, EngagementDocument.branch_id.is_(None)))
    if financial_year:
        stmt = stmt.where(EngagementDocument.financial_year == financial_year.strip())
    return db.execute(stmt).scalars().unique().all()


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



def _i1_required_review_levels(task: ClientServiceTaskInstance) -> list[str]:
    """Return configured review levels for I1 using existing task-template controls."""
    levels: list[str] = []
    if getattr(task, "aqmm_manager_review_required", False):
        levels.append("manager")
    if getattr(task, "aqmm_partner_review_required", False):
        levels.append("partner")
    if getattr(task, "aqmm_review_partner_required", False):
        levels.append("review_partner")

    # Preserve the pre-existing generic requires_review flag. Where no explicit
    # reviewer flag was configured, route to Manager when available, otherwise
    # Engagement Partner. No new review table/state is introduced.
    template = getattr(task, "template", None)
    if not levels and getattr(template, "requires_review", False):
        subscription = getattr(task, "subscription", None)
        if getattr(subscription, "assigned_manager_user_id", None):
            levels.append("manager")
        elif getattr(subscription, "assigned_partner_user_id", None):
            levels.append("partner")
    return levels


def _i1_review_level_pending(task: ClientServiceTaskInstance, level: str) -> bool:
    if level == "manager":
        return (getattr(task, "manager_review_status", None) or "not_required") != "reviewed"
    if level == "partner":
        return (getattr(task, "partner_review_status", None) or "not_required") != "reviewed"
    if level == "review_partner":
        return (getattr(task, "review_partner_review_status", None) or "not_required") != "reviewed"
    return False


def _i1_review_label(task: ClientServiceTaskInstance) -> str | None:
    levels = _i1_required_review_levels(task)
    if not levels:
        return None
    if getattr(task, "rework_status", "none") == "open":
        return "Returned for correction"
    pending = [level for level in levels if _i1_review_level_pending(task, level)]
    if not pending:
        return "Approved"
    labels = {
        "manager": "Manager",
        "partner": "Partner",
        "review_partner": "Review Partner",
    }
    return "Awaiting " + " + ".join(labels[level] for level in pending) + " approval"


def _i1_approval_blocker(
    tasks: list[ClientServiceTaskInstance],
    target_task: ClientServiceTaskInstance,
) -> dict[str, Any] | None:
    """Lock downstream staff work until required approval on an earlier task is complete."""
    ordered = sorted(tasks, key=lambda row: (int(getattr(row, "sequence_no", 0) or 0), int(row.id)))
    for row in ordered:
        if row.id == target_task.id:
            break
        levels = _i1_required_review_levels(row)
        if not levels:
            continue
        pending = [level for level in levels if _i1_review_level_pending(row, level)]
        if not pending:
            continue
        labels = {
            "manager": "Manager",
            "partner": "Partner",
            "review_partner": "Review Partner",
        }
        return {
            "task_id": int(row.id),
            "task_name": row.task_name,
            "label": _i1_review_label(row) or "Approval pending",
            "levels": [labels[level] for level in pending],
        }
    return None


def _i1_prepare_task_display(
    task: ClientServiceTaskInstance,
    *,
    tasks: list[ClientServiceTaskInstance],
    task_document_counts: dict[int, int],
) -> None:
    status = (getattr(task, "status", None) or "pending").strip().lower()
    task.i1_status = status
    task.i1_review_levels = _i1_required_review_levels(task)
    task.i1_review_label = _i1_review_label(task)
    task.i1_approval_blocker = _i1_approval_blocker(tasks, task)
    task.i1_has_evidence = task_document_counts.get(int(task.id), 0) > 0
    task.i1_documents = []
    task.i1_communications = [
        comment
        for comment in getattr(task, "comments", [])
        if not getattr(comment, "is_deleted", False)
    ]
    task.i1_needs_response = bool(
        getattr(task, "response_required", False)
        or (getattr(task, "response_type", "NONE") or "NONE").strip().upper() != "NONE"
    )
    task.i1_needs_evidence = bool(getattr(task, "evidence_required", False))
    task.i1_quick_complete = bool(
        not task.i1_approval_blocker
        and not task.i1_needs_response
        and (not task.i1_needs_evidence or task.i1_has_evidence)
        and not getattr(task, "is_locked", False)
        and not getattr(getattr(task, "subscription", None), "is_locked", False)
    )


def _employee_workflow_next_task(tasks: list[ClientServiceTaskInstance], current_task_id: int) -> ClientServiceTaskInstance | None:
    current_index = next((index for index, row in enumerate(tasks) if row.id == current_task_id), -1)
    ordered = tasks[current_index + 1 :] + tasks[: max(current_index, 0)]
    return next(
        (row for row in ordered if (row.status or "pending").strip().lower() not in CLOSED_TASK_STATUSES),
        None,
    )


def _workflow_task_weight(task: ClientServiceTaskInstance) -> int:
    """Stable derived progress weight without changing the task schema.

    Ordinary execution tasks count as 1. Review and AQMM/final-release tasks carry
    higher weight so completion cannot appear artificially high while significant
    review gates remain open.
    """
    weight = 1
    if bool(getattr(task, "is_aqmm_task", False)):
        weight += 2
    if bool(getattr(task, "aqmm_blocks_final_release", False)):
        weight += 2
    if any(bool(getattr(task, name, False)) for name in (
        "aqmm_manager_review_required",
        "aqmm_partner_review_required",
        "aqmm_review_partner_required",
    )):
        weight += 1
    if (getattr(task, "default_role_name", "") or "").strip().lower() in {"manager", "partner", "review partner"}:
        weight += 1
    return min(weight, 7)


def _weighted_progress(tasks: list[ClientServiceTaskInstance]) -> dict[str, int]:
    total_weight = sum(_workflow_task_weight(task) for task in tasks)
    completed_weight = sum(
        _workflow_task_weight(task)
        for task in tasks
        if (getattr(task, "status", None) or "pending").strip().lower() in CLOSED_TASK_STATUSES
    )
    percent = int(round((completed_weight * 100) / total_weight)) if total_weight else 0
    return {
        "total_weight": total_weight,
        "completed_weight": completed_weight,
        "progress_percent": percent,
    }


def _engagement_team(subscription: ClientServiceSubscription | None) -> list[dict[str, Any]]:
    if subscription is None:
        return []
    team = []
    for role, rel_name, id_name in (
        ("Engagement Partner", "assigned_partner", "assigned_partner_user_id"),
        ("Manager", "assigned_manager", "assigned_manager_user_id"),
        ("Primary Staff", "assigned_staff", "assigned_staff_user_id"),
        ("Review Partner", "review_partner", "review_partner_user_id"),
    ):
        user = getattr(subscription, rel_name, None)
        user_id = getattr(subscription, id_name, None)
        if user_id:
            team.append({
                "role": role,
                "user_id": user_id,
                "name": getattr(user, "full_name", None) or getattr(user, "email", None) or f"User #{user_id}",
                "email": getattr(user, "email", None) or "",
            })
    return team


def _engagement_sla(subscription: ClientServiceSubscription | None, tasks: list[ClientServiceTaskInstance]) -> dict[str, Any]:
    today = date.today()
    due_date = None
    if subscription is not None:
        due_date = getattr(subscription, "current_due_date", None) or getattr(subscription, "original_due_date", None) or getattr(subscription, "end_date", None)
    if due_date is None:
        dates = [getattr(task, "internal_target_date", None) for task in tasks if getattr(task, "internal_target_date", None)]
        due_date = max(dates) if dates else None
    started_dates = [
        _as_utc_aware(getattr(task, "started_at_utc", None))
        for task in tasks
        if getattr(task, "started_at_utc", None)
    ]
    created_at = (
        _as_utc_aware(getattr(subscription, "created_at_utc", None))
        if subscription is not None
        else None
    )
    anchor = min(started_dates) if started_dates else created_at
    age_days = max((datetime.now(timezone.utc) - anchor).days, 0) if anchor else 0
    days_remaining = (due_date - today).days if due_date else None
    if due_date and days_remaining < 0:
        status = "breached"
        label = f"Overdue by {abs(days_remaining)} day(s)"
    elif due_date and days_remaining <= 3:
        status = "critical"
        label = f"Due in {days_remaining} day(s)"
    elif due_date and days_remaining <= 7:
        status = "warning"
        label = f"Due in {days_remaining} day(s)"
    elif due_date:
        status = "on_track"
        label = f"{days_remaining} day(s) remaining"
    else:
        status = "not_set"
        label = "No SLA due date"
    return {
        "due_date": due_date,
        "days_remaining": days_remaining,
        "age_days": age_days,
        "status": status,
        "label": label,
    }


def escalate_employee_engagement_workflow(
    db: Session,
    scope: EmployeeScope,
    engagement_id: int,
    *,
    level: str,
    message: str,
    actor_user_id: int,
    financial_year: str | None = None,
) -> int:
    tasks = _employee_engagement_tasks(db, scope, engagement_id, financial_year=financial_year)
    if not tasks:
        raise ValueError("Engagement was not found or is not assigned to you.")
    subscription = getattr(tasks[0], "subscription", None)
    if subscription is None:
        raise ValueError("Engagement subscription was not found.")
    code = (level or "manager").strip().lower()
    target_id = {
        "manager": getattr(subscription, "assigned_manager_user_id", None),
        "partner": getattr(subscription, "assigned_partner_user_id", None),
        "review_partner": getattr(subscription, "review_partner_user_id", None),
    }.get(code)
    if not target_id:
        raise ValueError(f"{WORKFLOW_ESCALATION_LEVELS.get(code, 'Selected reviewer')} is not assigned to this engagement.")
    clean_message = (message or "").strip()[:1000]
    if not clean_message:
        raise ValueError("Escalation details are required.")
    create_workflow_escalation(
        db, subscription=subscription, raised_by_user_id=actor_user_id,
        assigned_to_user_id=int(target_id), escalation_level=code,
        message=clean_message, category="workflow_dependency", priority="high",
    )
    create_alert(
        db,
        user_id=int(target_id),
        title=f"Workflow escalation: {_subscription_label(subscription, tasks[0])}",
        message=clean_message,
        tenant_id=subscription.tenant_id,
        branch_id=subscription.branch_id,
        role_context=code,
        alert_type="escalation",
        priority="high",
        target_url=f"/employee/work/engagements/{engagement_id}",
        created_by_user_id=actor_user_id,
        commit=False,
    )
    db.commit()
    return int(target_id)


def get_employee_engagement_work_board(
    db: Session,
    scope: EmployeeScope,
    engagement_id: int,
    *,
    financial_year: str | None = None,
    active_task_id: int | None = None,
) -> dict[str, Any]:
    """Build the employee engagement workspace grouped by task_category.

    Only tasks assigned to the logged-in employee are returned. Existing task,
    document, checklist, AQMM and lock controls remain the source of truth.
    """
    today = date.today()
    stmt = _employee_work_task_query(
        db, scope, assigned_only=True, financial_year=financial_year
    ).where(ClientServiceTaskInstance.subscription_id == engagement_id)
    tasks = db.execute(
        stmt.order_by(
            ClientServiceTaskInstance.sequence_no.asc(),
            ClientServiceTaskInstance.internal_target_date.is_(None),
            ClientServiceTaskInstance.internal_target_date.asc(),
            ClientServiceTaskInstance.id.asc(),
        )
    ).scalars().all()
    if not tasks:
        raise HTTPException(
            status_code=404,
            detail="Engagement work not found or not assigned to you",
        )

    summary = {
        "total": len(tasks),
        "open": 0,
        "pending": 0,
        "in_progress": 0,
        "blocked": 0,
        "completed": 0,
        "overdue": 0,
        "due_today": 0,
    }
    category_lookup: dict[str, dict[str, Any]] = {}
    categories: list[dict[str, Any]] = []

    for task in tasks:
        _phase7i_task_card_enrich(task, today=today)
        status_code = (task.status or "pending").strip().lower()
        is_closed = status_code in CLOSED_TASK_STATUSES
        summary["completed" if is_closed else "open"] += 1
        if status_code in summary:
            summary[status_code] += 1
        if task.is_overdue:
            summary["overdue"] += 1
        if task.is_due_today:
            summary["due_today"] += 1

        category_name = _employee_task_category(task)
        category = category_lookup.get(category_name)
        if category is None:
            category = {
                "name": category_name,
                "tasks": [],
                "total": 0,
                "completed": 0,
                "blocked": 0,
                "in_progress": 0,
                "progress_percent": 0,
                "status": "pending",
            }
            category_lookup[category_name] = category
            categories.append(category)
        category["tasks"].append(task)
        category["total"] += 1
        if is_closed:
            category["completed"] += 1
        elif status_code == "blocked":
            category["blocked"] += 1
        elif status_code == "in_progress":
            category["in_progress"] += 1

    for category in categories:
        weighted_category = _weighted_progress(category["tasks"])
        category.update(weighted_category)
        category["simple_progress_percent"] = round(
            (category["completed"] / category["total"]) * 100
        ) if category["total"] else 0
        if category["completed"] == category["total"] and category["total"]:
            category["status"] = "completed"
        elif category["blocked"]:
            category["status"] = "blocked"
        elif category["in_progress"] or category["completed"]:
            category["status"] = "in_progress"

    selected_task = None
    if active_task_id is not None:
        selected_task = next((row for row in tasks if row.id == active_task_id), None)
    if selected_task is None:
        selected_task = next(
            (row for row in tasks if (row.status or "pending").strip().lower() == "in_progress"),
            None,
        )
    if selected_task is None:
        selected_task = next(
            (row for row in tasks if (row.status or "pending").strip().lower() not in CLOSED_TASK_STATUSES),
            tasks[0],
        )

    selected_category = _employee_task_category(selected_task)
    next_task = _employee_workflow_next_task(tasks, selected_task.id)
    weighted_overall = _weighted_progress(tasks)
    overall_progress = weighted_overall["progress_percent"]
    subscription = getattr(tasks[0], "subscription", None)
    manual_blocker = _workflow_manual_blocker(subscription)
    automatic_blocker = _workflow_automatic_blocker(subscription, tasks)

    engagement_documents = list_employee_engagement_documents(
        db, scope, engagement_id, financial_year=financial_year
    )
    task_document_counts: dict[int, int] = {}
    for document in engagement_documents:
        task_instance_id = getattr(document, "task_instance_id", None)
        if task_instance_id:
            task_document_counts[int(task_instance_id)] = task_document_counts.get(int(task_instance_id), 0) + 1

    task_documents: dict[int, list[EngagementDocument]] = {}
    for document in engagement_documents:
        task_instance_id = getattr(document, "task_instance_id", None)
        if task_instance_id:
            task_documents.setdefault(int(task_instance_id), []).append(document)

    for task in tasks:
        _i1_prepare_task_display(
            task,
            tasks=tasks,
            task_document_counts=task_document_counts,
        )
        task.i1_documents = task_documents.get(int(task.id), [])

    scope_display = _employee_work_scope_display_map(
        db,
        tenant_id=scope.tenant_id,
        subscriptions=[subscription] if subscription is not None else [],
    ).get(int(subscription.id), {}) if subscription is not None else {}

    return {
        "engagement_id": engagement_id,
        "subscription": subscription,
        "manual_blocker": manual_blocker,
        "automatic_blocker": automatic_blocker,
        "workflow_blocker": manual_blocker or automatic_blocker,
        "hold_reasons": WORKFLOW_HOLD_REASONS,
        "client": getattr(tasks[0], "client", None),
        "label": _subscription_label(getattr(tasks[0], "subscription", None), tasks[0]),
        "summary": summary,
        "progress_percent": overall_progress,
        "weighted_progress": weighted_overall,
        "engagement_team": _engagement_team(subscription),
        "sla": _engagement_sla(subscription, tasks),
        "escalation_levels": WORKFLOW_ESCALATION_LEVELS,
        "categories": categories,
        "active_task": selected_task,
        "active_category": selected_category,
        "next_task": next_task,
        "documents": engagement_documents,
        "scope_display": scope_display,
        "i1_enabled": True,
        "communication_types": TASK_COMMUNICATION_TYPES,
        "communication_visibilities": TASK_COMMUNICATION_VISIBILITIES,
        "today": today,
    }


def save_employee_workflow_task(
    db: Session,
    scope: EmployeeScope,
    task_id: int,
    *,
    status: str,
    remarks: str,
    checklist_response: str,
    checklist_text_response: str,
    checklist_number_response: str,
    checklist_date_response: str,
    checklist_remarks: str,
    actor_user_id: int,
    financial_year: str | None = None,
) -> tuple[ClientServiceTaskInstance, ClientServiceTaskInstance | None]:
    """Save one assigned workflow task and return its next open task.

    The operation is atomic: checklist validation, task status, AQMM status and
    timestamps are committed together. Locked tasks and locked engagements are
    never modified.
    """
    stmt = _employee_work_task_query(
        db, scope, assigned_only=True, financial_year=financial_year
    ).where(ClientServiceTaskInstance.id == task_id)
    task = db.execute(stmt).scalar_one_or_none()
    if not task:
        raise HTTPException(status_code=404, detail="Task not found or not assigned to you")
    if getattr(task, "is_locked", False) or getattr(getattr(task, "subscription", None), "is_locked", False):
        raise ValueError("Locked task cannot be changed.")

    normalised_status = (status or "pending").strip().lower()
    if normalised_status not in {"pending", "in_progress", "blocked", "completed", "not_applicable"}:
        raise ValueError("Invalid task status.")

    engagement_tasks_for_gate = db.execute(
        _employee_work_task_query(
            db, scope, assigned_only=True, financial_year=financial_year
        )
        .where(ClientServiceTaskInstance.subscription_id == task.subscription_id)
        .order_by(ClientServiceTaskInstance.sequence_no.asc(), ClientServiceTaskInstance.id.asc())
    ).scalars().all()
    approval_blocker = _i1_approval_blocker(engagement_tasks_for_gate, task)
    if approval_blocker and normalised_status in {"in_progress", "completed", "not_applicable"}:
        raise ValueError(approval_blocker["label"])

    clean_remarks = (remarks or "").strip()
    if normalised_status == "blocked" and not clean_remarks:
        raise ValueError("Select or enter a blocker reason.")
    if normalised_status == "not_applicable" and not clean_remarks:
        raise ValueError("A short reason is required for Not Applicable.")

    apply_task_checklist_response(
        db,
        task,
        checklist_response=checklist_response,
        checklist_text_response=checklist_text_response,
        checklist_number_response=checklist_number_response,
        checklist_date_response=checklist_date_response,
        checklist_remarks=checklist_remarks,
        requested_status=normalised_status,
    )

    task.status = normalised_status
    if normalised_status in CLOSED_TASK_STATUSES:
        task.completed_at_utc = datetime.now(timezone.utc)
    else:
        task.completed_at_utc = None
    if normalised_status == "in_progress" and not getattr(task, "started_at_utc", None):
        task.started_at_utc = datetime.now(timezone.utc)
    if clean_remarks:
        task.remarks = clean_remarks
    task.updated_by_user_id = actor_user_id
    recalculate_task_aqmm_status(db, task)

    # I1 approval gate: completing a task configured for Manager/Partner review
    # automatically sends the existing task to the existing review workflow.
    if normalised_status == "completed" and _i1_required_review_levels(task):
        submit_task_for_review(
            db,
            task=task,
            note=clean_remarks or "Completed by staff and submitted for approval.",
            user_id=actor_user_id,
        )

    db.add(task)
    db.commit()
    db.refresh(task)

    engagement_tasks = db.execute(
        _employee_work_task_query(
            db, scope, assigned_only=True, financial_year=financial_year
        )
        .where(ClientServiceTaskInstance.subscription_id == task.subscription_id)
        .order_by(ClientServiceTaskInstance.sequence_no.asc(), ClientServiceTaskInstance.id.asc())
    ).scalars().all()
    return task, _employee_workflow_next_task(engagement_tasks, task.id)



def quick_update_employee_workflow_task(
    db: Session,
    scope: EmployeeScope,
    task_id: int,
    *,
    action: str,
    reason: str,
    actor_user_id: int,
    financial_year: str | None = None,
) -> tuple[ClientServiceTaskInstance, ClientServiceTaskInstance | None]:
    """I1 one-click task action using the existing task/checklist validation path."""
    action_code = (action or "").strip().lower()
    status_map = {
        "done": "completed",
        "blocked": "blocked",
        "na": "not_applicable",
    }
    if action_code not in status_map:
        raise ValueError("Invalid checklist action.")

    stmt = _employee_work_task_query(
        db, scope, assigned_only=True, financial_year=financial_year
    ).where(ClientServiceTaskInstance.id == task_id)
    task = db.execute(stmt).scalar_one_or_none()
    if not task:
        raise HTTPException(status_code=404, detail="Task not found or not assigned to you")

    reason_text = (reason or "").strip()
    if action_code == "blocked" and not reason_text:
        raise ValueError("A blocker reason is required.")
    if action_code == "na" and not reason_text:
        raise ValueError("A short reason is required for Not Applicable.")

    return save_employee_workflow_task(
        db,
        scope,
        task_id,
        status=status_map[action_code],
        remarks=reason_text or (getattr(task, "remarks", None) or ""),
        checklist_response=getattr(task, "checklist_response", None) or "",
        checklist_text_response=getattr(task, "checklist_text_response", None) or "",
        checklist_number_response=(
            str(getattr(task, "checklist_number_response", ""))
            if getattr(task, "checklist_number_response", None) is not None
            else ""
        ),
        checklist_date_response=(
            str(getattr(task, "checklist_date_response", ""))
            if getattr(task, "checklist_date_response", None) is not None
            else ""
        ),
        checklist_remarks=getattr(task, "checklist_remarks", None) or "",
        actor_user_id=actor_user_id,
        financial_year=financial_year,
    )



def list_employee_work_assignable_users(db: Session, scope: EmployeeScope) -> list[User]:
    """Users that can be assigned engagement/service tasks in the active employee scope."""
    stmt = (
        select(User)
        .where(User.tenant_id == scope.tenant_id)
        .order_by(User.full_name.asc(), User.email.asc(), User.id.asc())
    )
    if scope.branch_id is not None:
        stmt = stmt.where(or_(User.branch_id == scope.branch_id, User.branch_id.is_(None)))
    return db.execute(stmt).scalars().all()


def list_visible_work_assignment_dashboard(
    db: Session,
    scope: EmployeeScope,
    *,
    q: str = "",
    status: str = "open",
    assigned_to_user_id: int | None = None,
    client_id: int | None = None,
    financial_year: str | None = None,
) -> dict[str, Any]:
    """Manager/admin work allocation view grouped by priority -> client -> engagement.

    This uses the existing client_service_task_instances table and therefore does
    not disturb the engagement/task generation flow. It is intentionally an
    assignment/review layer over existing service execution tasks.
    """
    today = date.today()
    status_filter = (status or "open").strip().lower()

    stmt = (
        select(ClientServiceTaskInstance)
        .options(
            selectinload(ClientServiceTaskInstance.client),
            selectinload(ClientServiceTaskInstance.catalogue),
            selectinload(ClientServiceTaskInstance.subscription).selectinload(ClientServiceSubscription.catalogue),
            selectinload(ClientServiceTaskInstance.subscription).selectinload(ClientServiceSubscription.assigned_partner),
            selectinload(ClientServiceTaskInstance.subscription).selectinload(ClientServiceSubscription.assigned_manager),
            selectinload(ClientServiceTaskInstance.subscription).selectinload(ClientServiceSubscription.assigned_staff),
            selectinload(ClientServiceTaskInstance.subscription).selectinload(ClientServiceSubscription.review_partner),
            selectinload(ClientServiceTaskInstance.assigned_to),
            selectinload(ClientServiceTaskInstance.comments).selectinload(ServiceTaskComment.created_by),
        )
        .where(
            ClientServiceTaskInstance.tenant_id == scope.tenant_id,
            ClientServiceTaskInstance.is_active.is_(True),
        )
    )
    if scope.branch_id is not None:
        stmt = stmt.where(ClientServiceTaskInstance.branch_id == scope.branch_id)
    if financial_year:
        stmt = stmt.where(ClientServiceTaskInstance.financial_year == financial_year.strip())
    if assigned_to_user_id:
        stmt = stmt.where(ClientServiceTaskInstance.assigned_to_user_id == int(assigned_to_user_id))
    if client_id:
        stmt = stmt.where(ClientServiceTaskInstance.client_id == int(client_id))

    if status_filter == "open":
        stmt = stmt.where(ClientServiceTaskInstance.status.notin_(list(CLOSED_TASK_STATUSES)))
    elif status_filter == "unassigned":
        stmt = stmt.where(ClientServiceTaskInstance.assigned_to_user_id.is_(None))
    elif status_filter == "closed":
        stmt = stmt.where(ClientServiceTaskInstance.status.in_(list(CLOSED_TASK_STATUSES)))
    elif status_filter in {code for code, _ in TASK_STATUSES}:
        stmt = stmt.where(ClientServiceTaskInstance.status == status_filter)

    if q.strip():
        like = f"%{q.strip()}%"
        stmt = stmt.where(
            or_(
                ClientServiceTaskInstance.task_name.ilike(like),
                ClientServiceTaskInstance.description.ilike(like),
                ClientServiceTaskInstance.client.has(or_(Client.client_name.ilike(like), Client.trade_name.ilike(like), Client.client_code.ilike(like))),
                ClientServiceTaskInstance.catalogue.has(or_(ServiceCatalogue.service_name.ilike(like), ServiceCatalogue.service_code.ilike(like))),
            )
        )

    tasks = db.execute(
        stmt.order_by(
            ClientServiceTaskInstance.priority.desc(),
            ClientServiceTaskInstance.internal_target_date.is_(None),
            ClientServiceTaskInstance.internal_target_date.asc(),
            ClientServiceTaskInstance.sequence_no.asc(),
            ClientServiceTaskInstance.id.desc(),
        )
    ).scalars().all()

    summary = {
        "total": len(tasks),
        "open": 0,
        "unassigned": 0,
        "overdue": 0,
        "due_today": 0,
        "completed": 0,
    }
    priority_map: dict[str, dict[str, Any]] = {code: {"code": code, "label": label, "task_count": 0, "clients": []} for code, label in PRIORITY_BUCKETS}
    client_lookup: dict[tuple[str, int], dict[str, Any]] = {}
    engagement_lookup: dict[tuple[str, int, int], dict[str, Any]] = {}

    for task in tasks:
        is_closed = (task.status or "") in CLOSED_TASK_STATUSES
        if is_closed:
            summary["completed"] += 1
        else:
            summary["open"] += 1
        if not getattr(task, "assigned_to_user_id", None):
            summary["unassigned"] += 1
        target = task.internal_target_date
        if target and target < today and not is_closed:
            summary["overdue"] += 1
        elif target and target == today and not is_closed:
            summary["due_today"] += 1

        priority_key = _task_priority_key(task)
        priority_bucket = priority_map.setdefault(priority_key, {"code": priority_key, "label": priority_key.title(), "task_count": 0, "clients": []})
        priority_bucket["task_count"] += 1

        client = getattr(task, "client", None)
        client_id_value = getattr(client, "id", 0) or getattr(task, "client_id", 0) or 0
        client_key = (priority_key, client_id_value)
        if client_key not in client_lookup:
            client_group = {
                "client": client,
                "client_name": getattr(client, "client_name", None) or "Unlinked Client",
                "client_code": getattr(client, "client_code", None) or "",
                "task_count": 0,
                "engagements": [],
            }
            client_lookup[client_key] = client_group
            priority_bucket["clients"].append(client_group)
        client_group = client_lookup[client_key]
        client_group["task_count"] += 1

        subscription = getattr(task, "subscription", None)
        engagement_id = getattr(subscription, "id", None) or getattr(task, "subscription_id", 0) or 0
        engagement_key = (priority_key, client_id_value, engagement_id)
        if engagement_key not in engagement_lookup:
            engagement_group = {
                "subscription": subscription,
                "label": _subscription_label(subscription, task),
                "status": getattr(subscription, "status", None) or "-",
                "due_date": getattr(subscription, "current_due_date", None) if subscription else None,
                "task_count": 0,
                "open_count": 0,
                "completed_count": 0,
                "tasks": [],
            }
            engagement_lookup[engagement_key] = engagement_group
            client_group["engagements"].append(engagement_group)
        engagement_group = engagement_lookup[engagement_key]
        engagement_group["task_count"] += 1
        if is_closed:
            engagement_group["completed_count"] += 1
        else:
            engagement_group["open_count"] += 1

        task.comment_count = len([c for c in getattr(task, "comments", []) if not getattr(c, "is_deleted", False)])
        task.latest_comment = next((c for c in getattr(task, "comments", []) if not getattr(c, "is_deleted", False)), None)
        task.status_label = _task_status_label(task)
        task.priority_label = _task_priority_label(task)
        task.date_bucket = _task_date_bucket(task, today=today)
        task.is_overdue = bool(target and target < today and not is_closed)
        task.is_due_today = bool(target and target == today and not is_closed)
        engagement_group["tasks"].append(task)

    groups = [priority_map[code] for code, _label in PRIORITY_BUCKETS if priority_map.get(code, {}).get("task_count")]
    return {"summary": summary, "groups": groups, "q": q, "status": status_filter, "today": today}



def update_service_task_assignment(
    db: Session,
    scope: EmployeeScope,
    task_id: int,
    *,
    assigned_to_user_id: int | None,
    status: str | None,
    priority: str | None,
    internal_target_date: date | None,
    remarks: str | None,
    actor_user_id: int,
    financial_year: str | None = None,
) -> ClientServiceTaskInstance:
    stmt = select(ClientServiceTaskInstance).where(
        ClientServiceTaskInstance.id == task_id,
        ClientServiceTaskInstance.tenant_id == scope.tenant_id,
    )
    if scope.branch_id is not None:
        stmt = stmt.where(ClientServiceTaskInstance.branch_id == scope.branch_id)
    if financial_year:
        stmt = stmt.where(ClientServiceTaskInstance.financial_year == financial_year.strip())
    task = db.execute(stmt).scalar_one_or_none()
    if not task:
        raise HTTPException(status_code=404, detail="Task not found")
    if getattr(task, "is_locked", False):
        raise HTTPException(status_code=400, detail="Locked task cannot be changed")
    if task.assigned_to_user_id is None:
        # Legacy engagement fallback: once the assigned staff member actually
        # works the task, persist the normal explicit assignment. This never
        # overwrites an existing/manual assignee.
        task.assigned_to_user_id = scope.own_user_id

    if assigned_to_user_id:
        assignee = db.get(User, int(assigned_to_user_id))
        if not assignee or assignee.tenant_id != scope.tenant_id:
            raise HTTPException(status_code=400, detail="Invalid assignee")
        if scope.branch_id is not None and getattr(assignee, "branch_id", None) not in (None, scope.branch_id):
            raise HTTPException(status_code=400, detail="Assignee is outside active branch")
        task.assigned_to_user_id = int(assigned_to_user_id)
    else:
        task.assigned_to_user_id = None

    allowed_statuses = {code for code, _ in TASK_STATUSES}
    if status and status in allowed_statuses:
        task.status = status
        if status == "completed" and not getattr(task, "completed_at_utc", None):
            task.completed_at_utc = datetime.now(timezone.utc)
        elif status != "completed":
            task.completed_at_utc = None
    allowed_priorities = {code for code, _ in TASK_PRIORITIES}
    if priority and priority in allowed_priorities:
        task.priority = priority
    task.internal_target_date = internal_target_date
    if remarks is not None:
        task.remarks = remarks.strip() or None
    task.updated_by_user_id = actor_user_id
    db.add(task)
    db.commit()
    db.refresh(task)
    return task


def update_own_service_task_status(
    db: Session,
    scope: EmployeeScope,
    task_id: int,
    *,
    status: str,
    remarks: str | None,
    actor_user_id: int,
    financial_year: str | None = None,
) -> ClientServiceTaskInstance:
    stmt = select(ClientServiceTaskInstance).where(
        ClientServiceTaskInstance.id == task_id,
        ClientServiceTaskInstance.tenant_id == scope.tenant_id,
        _employee_owned_task_condition(scope),
    )
    if scope.branch_id is not None:
        stmt = stmt.where(ClientServiceTaskInstance.branch_id == scope.branch_id)
    if financial_year:
        stmt = stmt.where(ClientServiceTaskInstance.financial_year == financial_year.strip())
    task = db.execute(stmt).scalar_one_or_none()
    if not task:
        raise HTTPException(status_code=404, detail="Task not found or not assigned to you")
    if getattr(task, "is_locked", False):
        raise HTTPException(status_code=400, detail="Locked task cannot be changed")
    allowed = {"pending", "in_progress", "blocked", "completed"}
    if status not in allowed:
        raise HTTPException(status_code=400, detail="Invalid status")
    task.status = status
    if status == "completed":
        task.completed_at_utc = datetime.now(timezone.utc)
    else:
        task.completed_at_utc = None
    if status == "in_progress" and not getattr(task, "started_at_utc", None):
        task.started_at_utc = datetime.now(timezone.utc)
    if remarks is not None and remarks.strip():
        task.remarks = remarks.strip()
    task.updated_by_user_id = actor_user_id
    db.add(task)
    db.commit()
    db.refresh(task)
    return task


# -------------------------
# Phase 7D Task Communication Timeline
# -------------------------

def _apply_task_scope(stmt, scope: EmployeeScope):
    stmt = stmt.where(ClientServiceTaskInstance.tenant_id == scope.tenant_id)
    if scope.branch_id is not None:
        stmt = stmt.where(ClientServiceTaskInstance.branch_id == scope.branch_id)
    return stmt


def get_work_task_with_communications(
    db: Session,
    scope: EmployeeScope,
    task_id: int,
    *,
    assigned_only: bool = False,
    financial_year: str | None = None,
) -> ClientServiceTaskInstance:
    stmt = (
        select(ClientServiceTaskInstance)
        .options(
            selectinload(ClientServiceTaskInstance.client),
            selectinload(ClientServiceTaskInstance.catalogue),
            selectinload(ClientServiceTaskInstance.subscription).selectinload(ClientServiceSubscription.catalogue),
            selectinload(ClientServiceTaskInstance.subscription).selectinload(ClientServiceSubscription.assigned_partner),
            selectinload(ClientServiceTaskInstance.subscription).selectinload(ClientServiceSubscription.assigned_manager),
            selectinload(ClientServiceTaskInstance.subscription).selectinload(ClientServiceSubscription.assigned_staff),
            selectinload(ClientServiceTaskInstance.subscription).selectinload(ClientServiceSubscription.review_partner),
            selectinload(ClientServiceTaskInstance.assigned_to),
            selectinload(ClientServiceTaskInstance.comments).selectinload(ServiceTaskComment.created_by),
        )
        .where(
            ClientServiceTaskInstance.id == task_id,
            ClientServiceTaskInstance.is_active.is_(True),
        )
    )
    stmt = _apply_task_scope(stmt, scope)
    if financial_year:
        stmt = stmt.where(ClientServiceTaskInstance.financial_year == financial_year.strip())
    if assigned_only:
        stmt = stmt.where(_employee_owned_task_condition(scope))

    task = db.execute(stmt).scalar_one_or_none()
    if not task:
        raise HTTPException(status_code=404, detail="Task not found")

    task.status_label = _task_status_label(task)
    task.priority_label = _task_priority_label(task)
    task.date_bucket = _task_date_bucket(task, today=date.today())
    task.engagement_label = _subscription_label(getattr(task, "subscription", None), task)
    task.communication_items = [c for c in getattr(task, "comments", []) if not getattr(c, "is_deleted", False)]
    return task


def add_work_task_communication(
    db: Session,
    scope: EmployeeScope,
    task_id: int,
    *,
    comment_type: str,
    visibility: str,
    message: str,
    actor_user_id: int,
    assigned_only: bool = False,
    financial_year: str | None = None,
) -> ServiceTaskComment:
    task = get_work_task_with_communications(db, scope, task_id, assigned_only=assigned_only, financial_year=financial_year)
    comment_type = (comment_type or "internal_note").strip()
    visibility = (visibility or "internal").strip()
    message = (message or "").strip()

    if comment_type not in TASK_COMMUNICATION_TYPE_CODES:
        raise HTTPException(status_code=400, detail="Invalid communication type")
    if visibility not in TASK_COMMUNICATION_VISIBILITY_CODES:
        raise HTTPException(status_code=400, detail="Invalid communication visibility")
    if not message:
        raise HTTPException(status_code=400, detail="Message is required")

    comment = ServiceTaskComment(
        tenant_id=task.tenant_id,
        branch_id=task.branch_id,
        subscription_id=task.subscription_id,
        task_instance_id=task.id,
        comment_type=comment_type,
        visibility=visibility,
        message=message,
        created_by_user_id=actor_user_id,
        is_deleted=False,
    )
    db.add(comment)
    task.updated_by_user_id = actor_user_id
    db.add(task)
    db.commit()
    db.refresh(comment)
    return comment


# -------------------------
# Phase 7C Engagement Progress Dashboard
# -------------------------

def _safe_progress(completed: int, total: int) -> int:
    if total <= 0:
        return 0
    return int(round((completed / total) * 100))


def _progress_status_bucket(open_count: int, overdue_count: int, completed_count: int, total_count: int) -> str:
    if total_count <= 0:
        return "no_tasks"
    if completed_count >= total_count:
        return "completed"
    if overdue_count > 0:
        return "overdue"
    if open_count > 0:
        return "in_progress"
    return "pending"


def list_engagement_progress_dashboard(
    db: Session,
    scope: EmployeeScope,
    *,
    q: str = "",
    status: str = "open",
    assigned_to_user_id: int | None = None,
    client_id: int | None = None,
    financial_year: str | None = None,
) -> dict[str, Any]:
    """Return engagement progress grouped by client -> engagement/service subscription.

    This is a read-only reporting layer over existing client_service_task_instances.
    It does not change task generation, assignment, or service execution logic.
    """
    today = date.today()
    status_filter = (status or "open").strip().lower()

    stmt = (
        select(ClientServiceTaskInstance)
        .options(
            selectinload(ClientServiceTaskInstance.client),
            selectinload(ClientServiceTaskInstance.catalogue),
            selectinload(ClientServiceTaskInstance.subscription).selectinload(ClientServiceSubscription.catalogue),
            selectinload(ClientServiceTaskInstance.subscription).selectinload(ClientServiceSubscription.assigned_partner),
            selectinload(ClientServiceTaskInstance.subscription).selectinload(ClientServiceSubscription.assigned_manager),
            selectinload(ClientServiceTaskInstance.subscription).selectinload(ClientServiceSubscription.assigned_staff),
            selectinload(ClientServiceTaskInstance.subscription).selectinload(ClientServiceSubscription.review_partner),
            selectinload(ClientServiceTaskInstance.assigned_to),
            selectinload(ClientServiceTaskInstance.comments).selectinload(ServiceTaskComment.created_by),
        )
        .where(
            ClientServiceTaskInstance.tenant_id == scope.tenant_id,
            ClientServiceTaskInstance.is_active.is_(True),
        )
    )
    if scope.branch_id is not None:
        stmt = stmt.where(ClientServiceTaskInstance.branch_id == scope.branch_id)
    if financial_year:
        stmt = stmt.where(ClientServiceTaskInstance.financial_year == financial_year.strip())
    if assigned_to_user_id:
        stmt = stmt.where(ClientServiceTaskInstance.assigned_to_user_id == int(assigned_to_user_id))
    if client_id:
        stmt = stmt.where(ClientServiceTaskInstance.client_id == int(client_id))

    if status_filter == "open":
        stmt = stmt.where(ClientServiceTaskInstance.status.notin_(list(CLOSED_TASK_STATUSES)))
    elif status_filter == "closed":
        stmt = stmt.where(ClientServiceTaskInstance.status.in_(list(CLOSED_TASK_STATUSES)))
    elif status_filter == "overdue":
        stmt = stmt.where(
            ClientServiceTaskInstance.internal_target_date.is_not(None),
            ClientServiceTaskInstance.internal_target_date < today,
            ClientServiceTaskInstance.status.notin_(list(CLOSED_TASK_STATUSES)),
        )
    elif status_filter == "due_today":
        stmt = stmt.where(
            ClientServiceTaskInstance.internal_target_date == today,
            ClientServiceTaskInstance.status.notin_(list(CLOSED_TASK_STATUSES)),
        )
    elif status_filter == "all":
        pass
    elif status_filter in {code for code, _ in TASK_STATUSES}:
        stmt = stmt.where(ClientServiceTaskInstance.status == status_filter)

    if q.strip():
        like = f"%{q.strip()}%"
        stmt = stmt.where(
            or_(
                ClientServiceTaskInstance.task_name.ilike(like),
                ClientServiceTaskInstance.description.ilike(like),
                ClientServiceTaskInstance.client.has(or_(Client.client_name.ilike(like), Client.trade_name.ilike(like), Client.client_code.ilike(like))),
                ClientServiceTaskInstance.catalogue.has(or_(ServiceCatalogue.service_name.ilike(like), ServiceCatalogue.service_code.ilike(like))),
            )
        )

    tasks = db.execute(
        stmt.order_by(
            ClientServiceTaskInstance.client_id.asc(),
            ClientServiceTaskInstance.subscription_id.asc(),
            ClientServiceTaskInstance.internal_target_date.is_(None),
            ClientServiceTaskInstance.internal_target_date.asc(),
            ClientServiceTaskInstance.sequence_no.asc(),
            ClientServiceTaskInstance.id.asc(),
        )
    ).scalars().all()

    summary = {
        "clients": 0,
        "engagements": 0,
        "tasks": len(tasks),
        "open": 0,
        "completed": 0,
        "overdue": 0,
        "due_today": 0,
        "unassigned": 0,
        "average_progress": 0,
    }

    clients: dict[int, dict[str, Any]] = {}
    engagement_lookup: dict[tuple[int, int], dict[str, Any]] = {}

    for task in tasks:
        is_closed = (task.status or "") in CLOSED_TASK_STATUSES
        target = task.internal_target_date
        if is_closed:
            summary["completed"] += 1
        else:
            summary["open"] += 1
        if target and target < today and not is_closed:
            summary["overdue"] += 1
        if target and target == today and not is_closed:
            summary["due_today"] += 1
        if not getattr(task, "assigned_to_user_id", None):
            summary["unassigned"] += 1

        client = getattr(task, "client", None)
        client_id_value = getattr(client, "id", None) or getattr(task, "client_id", None) or 0
        if client_id_value not in clients:
            clients[client_id_value] = {
                "client": client,
                "client_name": getattr(client, "client_name", None) or "Unlinked Client",
                "client_code": getattr(client, "client_code", None) or "",
                "task_count": 0,
                "open_count": 0,
                "completed_count": 0,
                "overdue_count": 0,
                "due_today_count": 0,
                "progress_percent": 0,
                "engagements": [],
            }
        client_group = clients[client_id_value]
        client_group["task_count"] += 1
        if is_closed:
            client_group["completed_count"] += 1
        else:
            client_group["open_count"] += 1
        if target and target < today and not is_closed:
            client_group["overdue_count"] += 1
        if target and target == today and not is_closed:
            client_group["due_today_count"] += 1

        subscription = getattr(task, "subscription", None)
        engagement_id = getattr(subscription, "id", None) or getattr(task, "subscription_id", None) or 0
        key = (client_id_value, engagement_id)
        if key not in engagement_lookup:
            engagement_group = {
                "subscription": subscription,
                "label": _subscription_label(subscription, task),
                "status": getattr(subscription, "status", None) or "-",
                "due_date": getattr(subscription, "current_due_date", None) if subscription else None,
                "task_count": 0,
                "open_count": 0,
                "completed_count": 0,
                "overdue_count": 0,
                "due_today_count": 0,
                "unassigned_count": 0,
                "progress_percent": 0,
                "progress_status": "pending",
                "tasks": [],
            }
            engagement_lookup[key] = engagement_group
            client_group["engagements"].append(engagement_group)
        engagement_group = engagement_lookup[key]
        engagement_group["task_count"] += 1
        if is_closed:
            engagement_group["completed_count"] += 1
        else:
            engagement_group["open_count"] += 1
        if target and target < today and not is_closed:
            engagement_group["overdue_count"] += 1
        if target and target == today and not is_closed:
            engagement_group["due_today_count"] += 1
        if not getattr(task, "assigned_to_user_id", None):
            engagement_group["unassigned_count"] += 1

        task.comment_count = len([c for c in getattr(task, "comments", []) if not getattr(c, "is_deleted", False)])
        task.latest_comment = next((c for c in getattr(task, "comments", []) if not getattr(c, "is_deleted", False)), None)
        task.status_label = _task_status_label(task)
        task.priority_label = _task_priority_label(task)
        task.date_bucket = _task_date_bucket(task, today=today)
        task.is_overdue = bool(target and target < today and not is_closed)
        task.is_due_today = bool(target and target == today and not is_closed)
        engagement_group["tasks"].append(task)

    total_progress = 0
    for client_group in clients.values():
        client_group["progress_percent"] = _safe_progress(client_group["completed_count"], client_group["task_count"])
        for engagement_group in client_group["engagements"]:
            engagement_group["progress_percent"] = _safe_progress(engagement_group["completed_count"], engagement_group["task_count"])
            engagement_group["progress_status"] = _progress_status_bucket(
                engagement_group["open_count"],
                engagement_group["overdue_count"],
                engagement_group["completed_count"],
                engagement_group["task_count"],
            )
            total_progress += engagement_group["progress_percent"]

    client_list = sorted(
        clients.values(),
        key=lambda c: (-c["overdue_count"], -c["due_today_count"], c["client_name"].lower()),
    )
    engagement_count = len(engagement_lookup)
    summary["clients"] = len(client_list)
    summary["engagements"] = engagement_count
    summary["average_progress"] = _safe_progress(total_progress, engagement_count * 100) if engagement_count else 0

    return {"summary": summary, "clients": client_list, "q": q, "status": status_filter, "today": today}
