from __future__ import annotations from datetime import date from sqlalchemy import or_, select from sqlalchemy.orm import Session, selectinload from app.modules.clients.models import Client from app.modules.core.iam.models import User from app.modules.core.rbac.models import Role, UserRole from app.modules.core.tenancy.models import Tenant from app.modules.services.models import ClientServicePlan, ClientServiceSubscription, FirmServiceSelection, ServiceCatalogue SUBSCRIPTION_STATUSES = [ ("draft", "Draft"), ("pending_acceptance", "Pending AQMM Acceptance"), ("active", "Active"), ("on_hold", "On Hold"), ("completed", "Completed"), ("cancelled", "Cancelled"), ("inactive", "Inactive"), ] ASSIGNMENT_ROLE_NAMES = ("Partner", "Branch Manager", "Staff") MONTHLY_PERIODS = ( (4, "Apr"), (5, "May"), (6, "Jun"), (7, "Jul"), (8, "Aug"), (9, "Sep"), (10, "Oct"), (11, "Nov"), (12, "Dec"), (1, "Jan"), (2, "Feb"), (3, "Mar"), ) QUARTERLY_PERIODS = ("Q1", "Q2", "Q3", "Q4") def normalized_recurrence_type(value: str | None) -> str: return (value or "").strip().lower().replace("-", "_").replace(" ", "_") def recurrence_requires_period(value: str | None) -> bool: return normalized_recurrence_type(value) in {"monthly", "quarterly"} def period_choices_for_service(financial_year: str | None, recurrence_type: str | None) -> list[tuple[str, str]]: recurrence = normalized_recurrence_type(recurrence_type) fy = normalize_financial_year(financial_year) start_year = int(fy.split("-", 1)[0]) end_year = start_year + 1 if recurrence == "monthly": result = [] for month, label in MONTHLY_PERIODS: year = start_year if month >= 4 else end_year result.append((f"{year:04d}-{month:02d}", f"{label} {year}")) return result if recurrence == "quarterly": return [(value, f"{value} {fy}") for value in QUARTERLY_PERIODS] return [("", "Not applicable")] def normalize_period_label(value: str | None, *, financial_year: str | None, recurrence_type: str | None) -> str: recurrence = normalized_recurrence_type(recurrence_type) raw = (value or "").strip() if not recurrence_requires_period(recurrence): return "" valid = {code for code, _ in period_choices_for_service(financial_year, recurrence)} if raw not in valid: raise ValueError("A valid month or quarter is required for this recurring service.") return raw def current_financial_year(today: date | None = None) -> str: today = today or date.today() if today.month >= 4: start = today.year else: start = today.year - 1 return f"{start}-{str(start + 1)[-2:]}" def assessment_year_from_financial_year(financial_year: str | None) -> str | None: if not financial_year or "-" not in financial_year: return None start = int(str(financial_year).split("-")[0]) return f"{start + 1}-{str(start + 2)[-2:]}" def normalize_financial_year(value: str | None) -> str: value = (value or "").strip() return value or current_financial_year() def parse_date(value: str | None) -> date | None: if not value: return None value = value.strip() if not value: return None return date.fromisoformat(value) def list_subscription_payload( db: Session, *, tenant_id: int, branch_id: int | None = None, financial_year: str | None = None, q: str = "", include_inactive: bool = True, ): fy = normalize_financial_year(financial_year) query = ( select(ClientServiceSubscription) .options( selectinload(ClientServiceSubscription.client), selectinload(ClientServiceSubscription.catalogue), selectinload(ClientServiceSubscription.assigned_partner), selectinload(ClientServiceSubscription.performing_partner), selectinload(ClientServiceSubscription.assigned_manager), selectinload(ClientServiceSubscription.assigned_staff), selectinload(ClientServiceSubscription.review_partner), selectinload(ClientServiceSubscription.due_date_rule), ) .where( ClientServiceSubscription.tenant_id == tenant_id, ClientServiceSubscription.financial_year == fy, ) ) if branch_id: query = query.where(ClientServiceSubscription.branch_id == branch_id) if not include_inactive: query = query.where(ClientServiceSubscription.is_active.is_(True)) if q.strip(): term = f"%{q.strip()}%" query = ( query.join(Client, Client.id == ClientServiceSubscription.client_id) .join(ServiceCatalogue, ServiceCatalogue.id == ClientServiceSubscription.service_catalogue_id) .where( or_( Client.client_name.ilike(term), Client.client_code.ilike(term), ServiceCatalogue.service_code.ilike(term), ServiceCatalogue.service_name.ilike(term), ) ) ) return db.execute( query.order_by( ClientServiceSubscription.is_locked.asc(), ClientServiceSubscription.is_active.desc(), ClientServiceSubscription.id.desc(), ) ).scalars().all() def get_subscription(db: Session, *, subscription_id: int, tenant_id: int) -> ClientServiceSubscription | None: return db.execute( select(ClientServiceSubscription) .options( selectinload(ClientServiceSubscription.client), selectinload(ClientServiceSubscription.catalogue), selectinload(ClientServiceSubscription.assigned_partner), selectinload(ClientServiceSubscription.performing_partner), selectinload(ClientServiceSubscription.assigned_manager), selectinload(ClientServiceSubscription.assigned_staff), selectinload(ClientServiceSubscription.review_partner), selectinload(ClientServiceSubscription.due_date_rule), ) .where( ClientServiceSubscription.id == subscription_id, ClientServiceSubscription.tenant_id == tenant_id, ) ).scalar_one_or_none() def get_existing_subscription( db: Session, *, tenant_id: int, client_id: int, service_catalogue_id: int, financial_year: str | None = None, period_label: str | None = None, ) -> ClientServiceSubscription | None: return db.execute( select(ClientServiceSubscription).where( ClientServiceSubscription.tenant_id == tenant_id, ClientServiceSubscription.client_id == client_id, ClientServiceSubscription.service_catalogue_id == service_catalogue_id, ClientServiceSubscription.financial_year == normalize_financial_year(financial_year), ClientServiceSubscription.period_label == (period_label or "").strip(), ) ).scalar_one_or_none() def list_clients_for_assignment(db: Session, *, tenant_id: int, branch_id: int | None = None, partner_id: int | None = None): query = select(Client).where(Client.tenant_id == tenant_id) if branch_id: query = query.where(Client.branch_id == branch_id) if partner_id: query = query.where(Client.partner_id == partner_id) return db.execute(query.order_by(Client.client_name.asc())).scalars().all() def list_enabled_services_for_assignment(db: Session, *, tenant_id: int): return db.execute( select(FirmServiceSelection) .join(ServiceCatalogue, ServiceCatalogue.id == FirmServiceSelection.service_catalogue_id) .options(selectinload(FirmServiceSelection.catalogue)) .where( FirmServiceSelection.tenant_id == tenant_id, FirmServiceSelection.is_enabled.is_(True), ServiceCatalogue.is_active.is_(True), ) .order_by(ServiceCatalogue.sort_order.asc(), ServiceCatalogue.service_name.asc()) ).scalars().all() def get_enabled_firm_service(db: Session, *, tenant_id: int, service_catalogue_id: int) -> FirmServiceSelection | None: return db.execute( select(FirmServiceSelection).where( FirmServiceSelection.tenant_id == tenant_id, FirmServiceSelection.service_catalogue_id == service_catalogue_id, FirmServiceSelection.is_enabled.is_(True), ) ).scalar_one_or_none() def list_assignable_users(db: Session, *, tenant_id: int, branch_id: int | None = None, role_names: tuple[str, ...] = ASSIGNMENT_ROLE_NAMES): query = ( select(User) .join(UserRole, UserRole.user_id == User.id) .join(Role, Role.id == UserRole.role_id) .where(User.tenant_id == tenant_id, User.is_active.is_(True), Role.name.in_(role_names)) ) if branch_id: query = query.where(or_(User.branch_id == branch_id, User.branch_id.is_(None))) return db.execute(query.order_by(User.full_name.asc(), User.email.asc()).distinct()).scalars().all() def tenant_requires_review_partner(db: Session, *, tenant_id: int) -> bool: tenant = db.get(Tenant, tenant_id) firm_type = (getattr(tenant, "firm_type", None) or "partnership").strip().lower() if tenant else "partnership" return firm_type == "partnership" def review_partner_required_for_engagement(db: Session, *, tenant_id: int, engagement_type: str | None) -> bool: return tenant_requires_review_partner(db, tenant_id=tenant_id) and (engagement_type or "").strip().lower() == "assurance" def list_review_partners(db: Session, *, tenant_id: int, branch_id: int | None = None): return list_assignable_users(db, tenant_id=tenant_id, branch_id=branch_id, role_names=("Partner",)) # ----------------------------------------------------------------------------- # AQMM engagement-level quality workflow helpers # ----------------------------------------------------------------------------- from datetime import datetime, timezone from app.modules.alerts.service import create_alert from app.modules.services.models import EngagementQualityDeclaration, EngagementKycVerification, EngagementLetter QUALITY_NOT_REQUIRED = "not_required" QUALITY_PENDING = "pending_acceptance" QUALITY_IN_PROGRESS = "in_progress" QUALITY_READY = "ready_for_approval" QUALITY_APPROVED = "approved" QUALITY_REJECTED = "rejected" DECLARATION_PENDING = "pending" DECLARATION_CLEAR = "declared_clear" DECLARATION_CONFLICT = "conflict_declared" DECLARATION_NOT_APPLICABLE = "not_applicable" def is_assurance_engagement(engagement_type: str | None) -> bool: return (engagement_type or "").strip().lower() == "assurance" def quality_required_for_engagement(engagement_type: str | None) -> bool: return is_assurance_engagement(engagement_type) def _engagement_team_user_ids(subscription: ClientServiceSubscription) -> list[int]: ids = [ subscription.assigned_partner_user_id, subscription.assigned_manager_user_id, subscription.assigned_staff_user_id, subscription.review_partner_user_id, ] seen: set[int] = set() result: list[int] = [] for uid in ids: if uid and int(uid) not in seen: seen.add(int(uid)) result.append(int(uid)) return result def _role_for_engagement_user(subscription: ClientServiceSubscription, user_id: int) -> str | None: if subscription.assigned_partner_user_id == user_id: return "Assigned Partner" if subscription.review_partner_user_id == user_id: return "Review Partner" if subscription.assigned_manager_user_id == user_id: return "Assigned Manager" if subscription.assigned_staff_user_id == user_id: return "Assigned Staff" return None def ensure_engagement_quality_workflow( db: Session, *, subscription: ClientServiceSubscription, actor_user_id: int | None = None, create_declarations: bool = False, ) -> ClientServiceSubscription: """Initialise or sync AQMM status for an engagement/subscription. Full quality workflow is mandatory only for assurance engagements. For non-assurance engagements, the quality fields are reset to not_required. """ required = quality_required_for_engagement(subscription.engagement_type) subscription.quality_workflow_required = required if not required: subscription.quality_workflow_status = QUALITY_NOT_REQUIRED subscription.quality_acceptance_status = QUALITY_NOT_REQUIRED subscription.quality_independence_status = QUALITY_NOT_REQUIRED subscription.quality_conflict_status = QUALITY_NOT_REQUIRED subscription.quality_kyc_status = QUALITY_NOT_REQUIRED subscription.quality_engagement_letter_status = QUALITY_NOT_REQUIRED subscription.quality_block_reason = None return subscription if subscription.quality_workflow_status in (None, "", QUALITY_NOT_REQUIRED): subscription.quality_workflow_status = QUALITY_PENDING if subscription.quality_acceptance_status in (None, "", QUALITY_NOT_REQUIRED): subscription.quality_acceptance_status = QUALITY_PENDING if subscription.quality_independence_status in (None, "", QUALITY_NOT_REQUIRED): subscription.quality_independence_status = "pending_declarations" if subscription.quality_conflict_status in (None, "", QUALITY_NOT_REQUIRED): subscription.quality_conflict_status = "pending_declarations" if subscription.quality_kyc_status in (None, "", QUALITY_NOT_REQUIRED): subscription.quality_kyc_status = "pending_verification" if subscription.quality_engagement_letter_status in (None, "", QUALITY_NOT_REQUIRED): subscription.quality_engagement_letter_status = "pending_client_acceptance" if create_declarations: request_engagement_quality_declarations(db, subscription=subscription, actor_user_id=actor_user_id) update_engagement_quality_summary(db, subscription=subscription) return subscription def request_engagement_quality_declarations( db: Session, *, subscription: ClientServiceSubscription, actor_user_id: int | None = None, ) -> list[EngagementQualityDeclaration]: ensure_engagement_quality_workflow(db, subscription=subscription, actor_user_id=actor_user_id, create_declarations=False) created_or_existing: list[EngagementQualityDeclaration] = [] target_url = f"/services/engagements/{subscription.id}" for user_id in _engagement_team_user_ids(subscription): role = _role_for_engagement_user(subscription, user_id) for declaration_type in ("independence", "conflict"): existing = db.execute( select(EngagementQualityDeclaration).where( EngagementQualityDeclaration.subscription_id == subscription.id, EngagementQualityDeclaration.declaration_type == declaration_type, EngagementQualityDeclaration.requested_user_id == user_id, ) ).scalar_one_or_none() if existing: created_or_existing.append(existing) continue row = EngagementQualityDeclaration( tenant_id=subscription.tenant_id, branch_id=subscription.branch_id, client_id=subscription.client_id, subscription_id=subscription.id, declaration_type=declaration_type, requested_user_id=user_id, requested_role=role, status=DECLARATION_PENDING, created_by_user_id=actor_user_id, ) db.add(row) db.flush() created_or_existing.append(row) try: create_alert( db, user_id=user_id, tenant_id=subscription.tenant_id, branch_id=subscription.branch_id, alert_type="general", priority="high" if declaration_type == "conflict" else "normal", title=f"AQMM {declaration_type.title()} Declaration Required", message=f"Please submit your {declaration_type} declaration for this assurance engagement.", target_url=target_url, created_by_user_id=actor_user_id, commit=False, ) except Exception: pass update_engagement_quality_summary(db, subscription=subscription) return created_or_existing def list_engagement_quality_declarations(db: Session, *, subscription_id: int) -> list[EngagementQualityDeclaration]: return db.execute( select(EngagementQualityDeclaration) .where(EngagementQualityDeclaration.subscription_id == subscription_id) .order_by(EngagementQualityDeclaration.declaration_type.asc(), EngagementQualityDeclaration.id.asc()) ).scalars().all() def respond_engagement_quality_declaration( db: Session, *, declaration_id: int, current_user_id: int, status: str, notes: str | None = None, request=None, ) -> EngagementQualityDeclaration: row = db.get(EngagementQualityDeclaration, declaration_id) if not row or int(row.requested_user_id) != int(current_user_id): raise ValueError("Declaration not found for current user.") if status not in {DECLARATION_CLEAR, DECLARATION_CONFLICT, DECLARATION_NOT_APPLICABLE}: raise ValueError("Invalid declaration status.") row.status = status row.response_notes = (notes or "").strip() or None row.responded_at_utc = datetime.now(timezone.utc) if request is not None: row.ip_address = getattr(getattr(request, "client", None), "host", None) row.user_agent = request.headers.get("user-agent") if hasattr(request, "headers") else None subscription = db.get(ClientServiceSubscription, row.subscription_id) if subscription: update_engagement_quality_summary(db, subscription=subscription) return row def get_latest_engagement_kyc_verification(db: Session, *, subscription_id: int) -> EngagementKycVerification | None: return db.execute( select(EngagementKycVerification) .where(EngagementKycVerification.subscription_id == subscription_id) .order_by(EngagementKycVerification.id.desc()) ).scalars().first() def verify_engagement_kyc_from_permanent_documents( db: Session, *, subscription: ClientServiceSubscription, actor_user_id: int, notes: str | None = None, ) -> EngagementKycVerification: ensure_engagement_quality_workflow(db, subscription=subscription, actor_user_id=actor_user_id, create_declarations=False) row = get_latest_engagement_kyc_verification(db, subscription_id=subscription.id) if not row: row = EngagementKycVerification( tenant_id=subscription.tenant_id, branch_id=subscription.branch_id, client_id=subscription.client_id, subscription_id=subscription.id, created_by_user_id=actor_user_id, ) db.add(row) row.status = "verified" row.source = "permanent_documents" row.verification_notes = (notes or "").strip() or "Verified from permanent document vault." row.verified_by_user_id = actor_user_id row.verified_at_utc = datetime.now(timezone.utc) subscription.quality_kyc_status = "verified" update_engagement_quality_summary(db, subscription=subscription) return row def get_current_engagement_letter(db: Session, *, subscription_id: int) -> EngagementLetter | None: return db.execute( select(EngagementLetter) .where(EngagementLetter.subscription_id == subscription_id) .order_by(EngagementLetter.version_no.desc(), EngagementLetter.id.desc()) ).scalars().first() def mark_engagement_letter_completed( db: Session, *, subscription: ClientServiceSubscription, actor_user_id: int, acceptance_mode: str, notes: str | None = None, request=None, ) -> EngagementLetter: ensure_engagement_quality_workflow(db, subscription=subscription, actor_user_id=actor_user_id, create_declarations=False) if acceptance_mode not in {"digital_otp", "manual_signed_upload"}: raise ValueError("Invalid engagement letter acceptance mode.") letter = get_current_engagement_letter(db, subscription_id=subscription.id) if not letter: letter = EngagementLetter( tenant_id=subscription.tenant_id, branch_id=subscription.branch_id, client_id=subscription.client_id, subscription_id=subscription.id, title="Engagement Letter", version_no=1, status="draft_pending", created_by_user_id=actor_user_id, ) db.add(letter) letter.status = "digitally_accepted" if acceptance_mode == "digital_otp" else "manual_signed_verified" letter.acceptance_mode = acceptance_mode letter.client_acceptance_declaration = (notes or "").strip() or None letter.client_accepted_by_user_id = actor_user_id if acceptance_mode == "digital_otp" else letter.client_accepted_by_user_id letter.client_accepted_at_utc = datetime.now(timezone.utc) if acceptance_mode == "digital_otp" else letter.client_accepted_at_utc letter.manual_verified_by_user_id = actor_user_id if acceptance_mode == "manual_signed_upload" else letter.manual_verified_by_user_id letter.manual_verified_at_utc = datetime.now(timezone.utc) if acceptance_mode == "manual_signed_upload" else letter.manual_verified_at_utc if request is not None: letter.ip_address = getattr(getattr(request, "client", None), "host", None) letter.user_agent = request.headers.get("user-agent") if hasattr(request, "headers") else None subscription.quality_engagement_letter_status = "completed" update_engagement_quality_summary(db, subscription=subscription) return letter def update_engagement_quality_summary(db: Session, *, subscription: ClientServiceSubscription) -> ClientServiceSubscription: if not quality_required_for_engagement(subscription.engagement_type): subscription.quality_workflow_required = False subscription.quality_workflow_status = QUALITY_NOT_REQUIRED subscription.quality_acceptance_status = QUALITY_NOT_REQUIRED subscription.quality_independence_status = QUALITY_NOT_REQUIRED subscription.quality_conflict_status = QUALITY_NOT_REQUIRED subscription.quality_kyc_status = QUALITY_NOT_REQUIRED subscription.quality_engagement_letter_status = QUALITY_NOT_REQUIRED subscription.quality_block_reason = None return subscription subscription.quality_workflow_required = True declarations = list_engagement_quality_declarations(db, subscription_id=subscription.id) if subscription.id else [] independence = [d for d in declarations if d.declaration_type == "independence"] conflicts = [d for d in declarations if d.declaration_type == "conflict"] if independence: if any(d.status == DECLARATION_PENDING for d in independence): subscription.quality_independence_status = "pending_declarations" elif any(d.status == DECLARATION_CONFLICT for d in independence): subscription.quality_independence_status = "issue_reported" else: subscription.quality_independence_status = "completed" else: subscription.quality_independence_status = "pending_declarations" if conflicts: if any(d.status == DECLARATION_PENDING for d in conflicts): subscription.quality_conflict_status = "pending_declarations" elif any(d.status == DECLARATION_CONFLICT for d in conflicts): subscription.quality_conflict_status = "conflict_reported" else: subscription.quality_conflict_status = "clear" else: subscription.quality_conflict_status = "pending_declarations" kyc = get_latest_engagement_kyc_verification(db, subscription_id=subscription.id) if subscription.id else None if kyc and kyc.status == "verified": subscription.quality_kyc_status = "verified" elif subscription.quality_kyc_status not in {"verified", "not_required"}: subscription.quality_kyc_status = "pending_verification" letter = get_current_engagement_letter(db, subscription_id=subscription.id) if subscription.id else None if letter and letter.status in {"digitally_accepted", "manual_signed_verified", "completed"}: subscription.quality_engagement_letter_status = "completed" elif subscription.quality_engagement_letter_status not in {"completed", "not_required"}: subscription.quality_engagement_letter_status = "pending_client_acceptance" ready = ( subscription.quality_independence_status == "completed" and subscription.quality_conflict_status == "clear" and subscription.quality_kyc_status == "verified" and subscription.quality_engagement_letter_status == "completed" ) if subscription.quality_acceptance_status == QUALITY_APPROVED: subscription.quality_workflow_status = QUALITY_APPROVED subscription.quality_block_reason = None elif ready: subscription.quality_workflow_status = QUALITY_READY subscription.quality_acceptance_status = QUALITY_READY subscription.quality_block_reason = None else: subscription.quality_workflow_status = QUALITY_IN_PROGRESS subscription.quality_acceptance_status = QUALITY_PENDING blockers = [] if subscription.quality_independence_status != "completed": blockers.append("independence declarations pending/issue") if subscription.quality_conflict_status != "clear": blockers.append("conflict declarations pending/conflict") if subscription.quality_kyc_status != "verified": blockers.append("KYC verification pending") if subscription.quality_engagement_letter_status != "completed": blockers.append("engagement letter acceptance pending") subscription.quality_block_reason = "; ".join(blockers) or None return subscription def approve_engagement_quality_workflow( db: Session, *, subscription: ClientServiceSubscription, actor_user_id: int, ) -> ClientServiceSubscription: ensure_engagement_quality_workflow(db, subscription=subscription, actor_user_id=actor_user_id, create_declarations=False) update_engagement_quality_summary(db, subscription=subscription) if subscription.quality_workflow_status != QUALITY_READY: raise ValueError(subscription.quality_block_reason or "AQMM workflow is not ready for approval.") subscription.quality_acceptance_status = QUALITY_APPROVED subscription.quality_workflow_status = QUALITY_APPROVED subscription.quality_approved_by_user_id = actor_user_id subscription.quality_approved_at_utc = datetime.now(timezone.utc) subscription.quality_block_reason = None subscription.status = "active" subscription.is_active = True return subscription def enforce_quality_gate_on_subscription(subscription: ClientServiceSubscription) -> None: """Prevent assurance engagement from becoming active before AQMM approval.""" if quality_required_for_engagement(subscription.engagement_type) and subscription.quality_acceptance_status != QUALITY_APPROVED: subscription.quality_workflow_required = True if subscription.quality_workflow_status in (None, "", QUALITY_NOT_REQUIRED): subscription.quality_workflow_status = QUALITY_PENDING if subscription.status == "active": subscription.status = "pending_acceptance" subscription.is_active = False if not subscription.quality_block_reason: subscription.quality_block_reason = "AQMM acceptance workflow pending for assurance engagement." def get_or_create_client_service_plan( db: Session, *, tenant_id: int, client: Client, catalogue: ServiceCatalogue, firm_selection: FirmServiceSelection | None, branch_id: int | None, partner_user_id: int | None, performing_partner_user_id: int | None, manager_user_id: int | None, staff_user_id: int | None, review_partner_user_id: int | None, actor_user_id: int | None, remarks: str | None = None, ) -> ClientServicePlan: plan = db.execute( select(ClientServicePlan).where( ClientServicePlan.tenant_id == tenant_id, ClientServicePlan.client_id == client.id, ClientServicePlan.service_catalogue_id == catalogue.id, ) ).scalar_one_or_none() if plan is None: plan = ClientServicePlan( tenant_id=tenant_id, client_id=client.id, service_catalogue_id=catalogue.id, created_by_user_id=actor_user_id, ) db.add(plan) plan.branch_id = branch_id or getattr(client, "branch_id", None) plan.firm_service_selection_id = getattr(firm_selection, "id", None) plan.default_partner_user_id = partner_user_id or getattr(client, "partner_id", None) plan.default_performing_partner_user_id = performing_partner_user_id or getattr(client, "default_performing_partner_user_id", None) or plan.default_partner_user_id plan.default_manager_user_id = manager_user_id plan.default_staff_user_id = staff_user_id plan.default_review_partner_user_id = review_partner_user_id or getattr(client, "default_review_partner_user_id", None) plan.recurrence_type = normalized_recurrence_type(getattr(catalogue, "recurrence_type", None)) or "one_time" plan.auto_generate_periods = recurrence_requires_period(plan.recurrence_type) plan.status = "active" plan.is_active = True if remarks and not plan.remarks: plan.remarks = remarks plan.updated_by_user_id = actor_user_id db.flush() return plan def attach_engagement_to_plan( db: Session, *, engagement: ClientServiceSubscription, client: Client, catalogue: ServiceCatalogue, firm_selection: FirmServiceSelection | None, actor_user_id: int | None, ) -> ClientServicePlan: plan = get_or_create_client_service_plan( db, tenant_id=engagement.tenant_id, client=client, catalogue=catalogue, firm_selection=firm_selection, branch_id=engagement.branch_id, partner_user_id=engagement.assigned_partner_user_id, performing_partner_user_id=getattr(engagement, "performing_partner_user_id", None), manager_user_id=engagement.assigned_manager_user_id, staff_user_id=engagement.assigned_staff_user_id, review_partner_user_id=engagement.review_partner_user_id, actor_user_id=actor_user_id, remarks=engagement.remarks, ) engagement.service_plan_id = plan.id return plan