Add AQMM quality tags to existing service task checklist

app/modules/services/execution.py

@@ -6,6 +6,7 @@ from sqlalchemy import func, or_, select
 from sqlalchemy.orm import Session, selectinload
 
 from app.modules.clients.models import Client
+from app.modules.documents.models import EngagementDocument
 from app.modules.core.iam.models import User
 from app.modules.services.client_services import quality_required_for_engagement, QUALITY_APPROVED
 from app.modules.services.models import (
@@ -222,6 +223,16 @@ def generate_tasks_for_subscription(db: Session, *, subscription: ClientServiceS
                 internal_target_date=_default_internal_target_date(subscription, template),
                 status="pending",
                 priority="normal",
+                is_aqmm_task=getattr(template, "is_aqmm_task", False),
+                aqmm_mandatory=getattr(template, "aqmm_mandatory", False),
+                aqmm_evidence_required=getattr(template, "aqmm_evidence_required", False),
+                aqmm_manager_review_required=getattr(template, "aqmm_manager_review_required", False),
+                aqmm_partner_review_required=getattr(template, "aqmm_partner_review_required", False),
+                aqmm_review_partner_required=getattr(template, "aqmm_review_partner_required", False),
+                aqmm_blocks_final_release=getattr(template, "aqmm_blocks_final_release", False),
+                aqmm_reference=getattr(template, "aqmm_reference", None),
+                aqmm_status="pending" if getattr(template, "is_aqmm_task", False) else "not_required",
+                aqmm_review_status="pending_review" if (getattr(template, "aqmm_manager_review_required", False) or getattr(template, "aqmm_partner_review_required", False) or getattr(template, "aqmm_review_partner_required", False)) else "not_required",
                 is_active=True,
                 created_by_user_id=user_id,
                 updated_by_user_id=user_id,
@@ -231,6 +242,89 @@ def generate_tasks_for_subscription(db: Session, *, subscription: ClientServiceS
     return created
 
 
+
+def _task_has_evidence(db: Session, task_id: int) -> bool:
+    return bool(
+        db.execute(
+            select(EngagementDocument.id).where(
+                EngagementDocument.task_instance_id == int(task_id),
+                EngagementDocument.is_deleted.is_(False),
+            )
+        ).first()
+    )
+
+
+def _aqmm_task_issue(db: Session, task: ClientServiceTaskInstance) -> str | None:
+    """Return blocking reason for one AQMM-tagged task, or None if it passes.
+
+    This deliberately reuses the existing task status and task document upload
+    workflow. Review flags are captured for reporting and future review-specific
+    workflow; until dedicated manager/partner review statuses are added, the
+    task must at least be completed and evidence uploaded wherever marked.
+    """
+    if not getattr(task, "is_aqmm_task", False):
+        return None
+    if getattr(task, "aqmm_mandatory", False) and task.status != "completed":
+        return "Task not completed"
+    if getattr(task, "aqmm_evidence_required", False) and not _task_has_evidence(db, task.id):
+        return "Evidence not uploaded"
+    return None
+
+
+def list_aqmm_quality_tasks(db: Session, *, subscription_id: int) -> list[dict]:
+    tasks = db.execute(
+        select(ClientServiceTaskInstance)
+        .where(
+            ClientServiceTaskInstance.subscription_id == int(subscription_id),
+            ClientServiceTaskInstance.is_active.is_(True),
+            ClientServiceTaskInstance.is_aqmm_task.is_(True),
+        )
+        .order_by(ClientServiceTaskInstance.sequence_no.asc(), ClientServiceTaskInstance.id.asc())
+    ).scalars().all()
+    rows: list[dict] = []
+    for task in tasks:
+        has_evidence = _task_has_evidence(db, task.id)
+        issue = _aqmm_task_issue(db, task)
+        rows.append({"task": task, "has_evidence": has_evidence, "issue": issue, "passes": issue is None})
+    return rows
+
+
+def aqmm_task_summary_for_subscription(db: Session, *, subscription_id: int) -> dict:
+    rows = list_aqmm_quality_tasks(db, subscription_id=subscription_id)
+    total = len(rows)
+    completed = sum(1 for r in rows if r["task"].status == "completed")
+    mandatory = sum(1 for r in rows if getattr(r["task"], "aqmm_mandatory", False))
+    mandatory_done = sum(1 for r in rows if getattr(r["task"], "aqmm_mandatory", False) and r["passes"])
+    evidence_required = sum(1 for r in rows if getattr(r["task"], "aqmm_evidence_required", False))
+    evidence_missing = sum(1 for r in rows if getattr(r["task"], "aqmm_evidence_required", False) and not r["has_evidence"])
+    blockers = [r for r in rows if getattr(r["task"], "aqmm_blocks_final_release", False) and not r["passes"]]
+    status = "not_required"
+    if total:
+        status = "completed" if mandatory_done == mandatory and evidence_missing == 0 else "in_progress"
+    return {
+        "rows": rows,
+        "total": total,
+        "completed": completed,
+        "mandatory": mandatory,
+        "mandatory_done": mandatory_done,
+        "evidence_required": evidence_required,
+        "evidence_missing": evidence_missing,
+        "blockers": blockers,
+        "status": status,
+    }
+
+
+def assert_aqmm_quality_tasks_complete(db: Session, *, subscription_id: int) -> None:
+    summary = aqmm_task_summary_for_subscription(db, subscription_id=subscription_id)
+    if summary["blockers"]:
+        first = summary["blockers"][0]
+        task = first["task"]
+        raise ValueError(f"AQMM final release blocked: {task.task_name} - {first['issue']}")
+    if summary["mandatory"] and summary["mandatory_done"] < summary["mandatory"]:
+        raise ValueError("AQMM mandatory quality checklist tasks are pending.")
+    if summary["evidence_missing"]:
+        raise ValueError("AQMM evidence upload is pending for one or more quality checklist tasks.")
+
 def _decorate_task_for_tracker(task: ClientServiceTaskInstance, *, today: date) -> ClientServiceTaskInstance:
     target_date = getattr(task, "internal_target_date", None)
     subscription = getattr(task, "subscription", None)