diff --git a/alembic/versions/20260720_document_vps_auto_cleanup.py b/alembic/versions/20260720_document_vps_auto_cleanup.py new file mode 100644 index 0000000..8dc6d13 --- /dev/null +++ b/alembic/versions/20260720_document_vps_auto_cleanup.py @@ -0,0 +1,42 @@ +"""Add VPS staging cleanup audit fields to document storage jobs. + +Revision ID: 20260720_document_vps_auto_cleanup +Revises: 20260720_phase5c_workflow_escalations +""" +from alembic import op +import sqlalchemy as sa + +revision = "20260720_document_vps_auto_cleanup" +down_revision = "20260720_phase5c_workflow_escalations" +branch_labels = None +depends_on = None + + +def _add_cleanup_columns(table_name: str) -> None: + op.add_column(table_name, sa.Column("vps_cleanup_status", sa.String(length=30), nullable=False, server_default="pending")) + op.add_column(table_name, sa.Column("vps_cleanup_attempts", sa.Integer(), nullable=False, server_default="0")) + op.add_column(table_name, sa.Column("vps_cleanup_error", sa.Text(), nullable=True)) + op.add_column(table_name, sa.Column("vps_deleted_at_utc", sa.DateTime(timezone=True), nullable=True)) + op.add_column(table_name, sa.Column("vps_deleted_bytes", sa.Integer(), nullable=True)) + op.create_index(f"ix_{table_name}_vps_cleanup_status", table_name, ["vps_cleanup_status"], unique=False) + op.create_index(f"ix_{table_name}_vps_deleted_at_utc", table_name, ["vps_deleted_at_utc"], unique=False) + + +def _drop_cleanup_columns(table_name: str) -> None: + op.drop_index(f"ix_{table_name}_vps_deleted_at_utc", table_name=table_name) + op.drop_index(f"ix_{table_name}_vps_cleanup_status", table_name=table_name) + op.drop_column(table_name, "vps_deleted_bytes") + op.drop_column(table_name, "vps_deleted_at_utc") + op.drop_column(table_name, "vps_cleanup_error") + op.drop_column(table_name, "vps_cleanup_attempts") + op.drop_column(table_name, "vps_cleanup_status") + + +def upgrade() -> None: + _add_cleanup_columns("document_storage_jobs") + _add_cleanup_columns("permanent_document_storage_jobs") + + +def downgrade() -> None: + _drop_cleanup_columns("permanent_document_storage_jobs") + _drop_cleanup_columns("document_storage_jobs") diff --git a/alembic/versions/20260720_phase5c_workflow_escalations.py b/alembic/versions/20260720_phase5c_workflow_escalations.py new file mode 100644 index 0000000..6308f70 --- /dev/null +++ b/alembic/versions/20260720_phase5c_workflow_escalations.py @@ -0,0 +1,15 @@ +"""Add durable workflow escalation register. + +Revision ID: 20260720_phase5c_workflow_escalations +Revises: 20260719_employee_workflow_phase3 +""" +from alembic import op +import sqlalchemy as sa +revision="20260720_phase5c_workflow_escalations" +down_revision="20260719_employee_workflow_phase3" +branch_labels=None +depends_on=None +def upgrade(): + op.create_table("workflow_escalations",sa.Column("id",sa.Integer(),primary_key=True),sa.Column("tenant_id",sa.Integer(),nullable=False),sa.Column("branch_id",sa.Integer(),nullable=True),sa.Column("subscription_id",sa.Integer(),nullable=False),sa.Column("task_id",sa.Integer(),nullable=True),sa.Column("raised_by_user_id",sa.Integer(),nullable=False),sa.Column("assigned_to_user_id",sa.Integer(),nullable=False),sa.Column("escalation_level",sa.String(30),nullable=False),sa.Column("category",sa.String(50),nullable=False,server_default="workflow_dependency"),sa.Column("priority",sa.String(20),nullable=False,server_default="high"),sa.Column("status",sa.String(20),nullable=False,server_default="open"),sa.Column("message",sa.Text(),nullable=False),sa.Column("follow_up_date",sa.Date(),nullable=True),sa.Column("acknowledged_at_utc",sa.DateTime(timezone=True),nullable=True),sa.Column("acknowledged_by_user_id",sa.Integer(),nullable=True),sa.Column("resolved_at_utc",sa.DateTime(timezone=True),nullable=True),sa.Column("resolved_by_user_id",sa.Integer(),nullable=True),sa.Column("resolution_note",sa.Text(),nullable=True),sa.Column("created_at_utc",sa.DateTime(timezone=True),nullable=False),sa.Column("updated_at_utc",sa.DateTime(timezone=True),nullable=False),sa.ForeignKeyConstraint(["tenant_id"],["tenants.id"],ondelete="CASCADE"),sa.ForeignKeyConstraint(["branch_id"],["branches.id"],ondelete="SET NULL"),sa.ForeignKeyConstraint(["subscription_id"],["client_service_subscriptions.id"],ondelete="CASCADE"),sa.ForeignKeyConstraint(["task_id"],["client_service_task_instances.id"],ondelete="SET NULL"),sa.ForeignKeyConstraint(["raised_by_user_id"],["users.id"],ondelete="RESTRICT"),sa.ForeignKeyConstraint(["assigned_to_user_id"],["users.id"],ondelete="RESTRICT"),sa.ForeignKeyConstraint(["acknowledged_by_user_id"],["users.id"],ondelete="SET NULL"),sa.ForeignKeyConstraint(["resolved_by_user_id"],["users.id"],ondelete="SET NULL")) + for c in ["tenant_id","branch_id","subscription_id","task_id","raised_by_user_id","assigned_to_user_id","escalation_level","category","priority","status","follow_up_date","created_at_utc"]: op.create_index(f"ix_workflow_escalations_{c}","workflow_escalations",[c]) +def downgrade(): op.drop_table("workflow_escalations") diff --git a/app/modules/alerts/models.py b/app/modules/alerts/models.py index 5948e77..a2f6928 100644 --- a/app/modules/alerts/models.py +++ b/app/modules/alerts/models.py @@ -2,7 +2,7 @@ from __future__ import annotations from datetime import datetime, timezone -from sqlalchemy import Boolean, DateTime, ForeignKey, Integer, String, Text +from sqlalchemy import Boolean, Date, DateTime, ForeignKey, Integer, String, Text from sqlalchemy.orm import Mapped, mapped_column, relationship from app.core.db.common import CommonBase @@ -39,3 +39,37 @@ class UserAlert(CommonBase): user = relationship("User", foreign_keys=[user_id]) created_by = relationship("User", foreign_keys=[created_by_user_id]) + + +class WorkflowEscalation(CommonBase): + """Durable engagement escalation register used by staff, managers and partners.""" + + __tablename__ = "workflow_escalations" + + id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) + tenant_id: Mapped[int] = mapped_column(ForeignKey("tenants.id", ondelete="CASCADE"), nullable=False, index=True) + branch_id: Mapped[int | None] = mapped_column(ForeignKey("branches.id", ondelete="SET NULL"), nullable=True, index=True) + subscription_id: Mapped[int] = mapped_column(ForeignKey("client_service_subscriptions.id", ondelete="CASCADE"), nullable=False, index=True) + task_id: Mapped[int | None] = mapped_column(ForeignKey("client_service_task_instances.id", ondelete="SET NULL"), nullable=True, index=True) + raised_by_user_id: Mapped[int] = mapped_column(ForeignKey("users.id", ondelete="RESTRICT"), nullable=False, index=True) + assigned_to_user_id: Mapped[int] = mapped_column(ForeignKey("users.id", ondelete="RESTRICT"), nullable=False, index=True) + escalation_level: Mapped[str] = mapped_column(String(30), nullable=False, index=True) + category: Mapped[str] = mapped_column(String(50), nullable=False, default="workflow_dependency", index=True) + priority: Mapped[str] = mapped_column(String(20), nullable=False, default="high", index=True) + status: Mapped[str] = mapped_column(String(20), nullable=False, default="open", index=True) + message: Mapped[str] = mapped_column(Text, nullable=False) + follow_up_date: Mapped[date | None] = mapped_column(Date, nullable=True, index=True) + acknowledged_at_utc: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) + acknowledged_by_user_id: Mapped[int | None] = mapped_column(ForeignKey("users.id", ondelete="SET NULL"), nullable=True) + resolved_at_utc: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) + resolved_by_user_id: Mapped[int | None] = mapped_column(ForeignKey("users.id", ondelete="SET NULL"), nullable=True) + resolution_note: Mapped[str | None] = mapped_column(Text, nullable=True) + created_at_utc: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=lambda: datetime.now(timezone.utc), nullable=False, index=True) + updated_at_utc: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=lambda: datetime.now(timezone.utc), onupdate=lambda: datetime.now(timezone.utc), nullable=False) + + subscription = relationship("ClientServiceSubscription") + task = relationship("ClientServiceTaskInstance") + raised_by = relationship("User", foreign_keys=[raised_by_user_id]) + assigned_to = relationship("User", foreign_keys=[assigned_to_user_id]) + acknowledged_by = relationship("User", foreign_keys=[acknowledged_by_user_id]) + resolved_by = relationship("User", foreign_keys=[resolved_by_user_id]) diff --git a/app/modules/alerts/workflow_escalations.py b/app/modules/alerts/workflow_escalations.py new file mode 100644 index 0000000..081d3f6 --- /dev/null +++ b/app/modules/alerts/workflow_escalations.py @@ -0,0 +1,43 @@ +from __future__ import annotations + +from datetime import date, datetime, timezone +from typing import Any + +from sqlalchemy import select +from sqlalchemy.orm import Session, selectinload + +from app.modules.alerts.models import WorkflowEscalation +from app.modules.services.models import ClientServiceSubscription + +OPEN_ESCALATION_STATUSES = {"open", "acknowledged"} + +def create_workflow_escalation(db: Session, *, subscription: ClientServiceSubscription, raised_by_user_id: int, assigned_to_user_id: int, escalation_level: str, message: str, task_id: int | None = None, category: str = "workflow_dependency", priority: str = "high", follow_up_date: date | None = None) -> WorkflowEscalation: + row = WorkflowEscalation(tenant_id=subscription.tenant_id, branch_id=subscription.branch_id, subscription_id=subscription.id, task_id=task_id, raised_by_user_id=raised_by_user_id, assigned_to_user_id=assigned_to_user_id, escalation_level=escalation_level, category=category, priority=priority, status="open", message=message.strip()[:4000], follow_up_date=follow_up_date) + db.add(row) + db.flush() + return row + +def list_workflow_escalations(db: Session, *, tenant_id: int | None, branch_id: int | None, assigned_to_user_id: int | None = None, include_resolved: bool = False, limit: int = 200) -> list[dict[str, Any]]: + stmt = select(WorkflowEscalation).options(selectinload(WorkflowEscalation.subscription).selectinload(ClientServiceSubscription.client), selectinload(WorkflowEscalation.subscription).selectinload(ClientServiceSubscription.catalogue), selectinload(WorkflowEscalation.raised_by), selectinload(WorkflowEscalation.assigned_to)) + if tenant_id: stmt = stmt.where(WorkflowEscalation.tenant_id == tenant_id) + if branch_id is not None: stmt = stmt.where(WorkflowEscalation.branch_id == branch_id) + if assigned_to_user_id: stmt = stmt.where(WorkflowEscalation.assigned_to_user_id == assigned_to_user_id) + if not include_resolved: stmt = stmt.where(WorkflowEscalation.status.in_(OPEN_ESCALATION_STATUSES)) + rows = db.execute(stmt.order_by(WorkflowEscalation.created_at_utc.asc()).limit(limit)).scalars().all() + now = datetime.now(timezone.utc) + out=[] + for row in rows: + created=row.created_at_utc + if created and created.tzinfo is None: created=created.replace(tzinfo=timezone.utc) + sub=row.subscription + out.append({"id":row.id,"subscription_id":row.subscription_id,"task_id":row.task_id,"client_name":getattr(getattr(sub,"client",None),"client_name",None) or "Unlinked Client","service_name":getattr(getattr(sub,"catalogue",None),"service_name",None) or "Service","financial_year":getattr(sub,"financial_year",None) or "-","level":row.escalation_level,"category":row.category,"priority":row.priority,"status":row.status,"message":row.message,"follow_up_date":row.follow_up_date,"raised_by":getattr(row.raised_by,"full_name",None) or getattr(row.raised_by,"email",None) or "User","assigned_to":getattr(row.assigned_to,"full_name",None) or getattr(row.assigned_to,"email",None) or "User","created_at":row.created_at_utc,"age_days":max(0,(now-created).days) if created else 0,"href":f"/manager/reviews/engagements/{row.subscription_id}" if row.escalation_level=="manager" else f"/partner/reviews/engagements/{row.subscription_id}"}) + return out + +def update_workflow_escalation(db: Session, *, escalation_id: int, actor_user_id: int, action: str, resolution_note: str = "") -> WorkflowEscalation: + row=db.get(WorkflowEscalation, escalation_id) + if not row or int(row.assigned_to_user_id)!=int(actor_user_id): raise ValueError("Escalation not found or not assigned to you.") + now=datetime.now(timezone.utc) + if action=="acknowledge": row.status="acknowledged"; row.acknowledged_at_utc=now; row.acknowledged_by_user_id=actor_user_id + elif action=="resolve": row.status="resolved"; row.resolved_at_utc=now; row.resolved_by_user_id=actor_user_id; row.resolution_note=(resolution_note or "").strip()[:4000] or None + else: raise ValueError("Invalid escalation action.") + db.flush(); return row diff --git a/app/modules/documents/models.py b/app/modules/documents/models.py index 1079521..e25fa4b 100644 --- a/app/modules/documents/models.py +++ b/app/modules/documents/models.py @@ -213,6 +213,15 @@ class DocumentStorageJob(CommonBase): completed_at_utc: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) acknowledged_hash_sha256: Mapped[str | None] = mapped_column(String(64), nullable=True) local_final_path: Mapped[str | None] = mapped_column(String(1000), nullable=True) + + # VPS staging cleanup audit. The file is deleted only after the branch agent + # acknowledges the exact expected SHA-256 hash. + vps_cleanup_status: Mapped[str] = mapped_column(String(30), nullable=False, default="pending", index=True) + vps_cleanup_attempts: Mapped[int] = mapped_column(Integer, nullable=False, default=0) + vps_cleanup_error: Mapped[str | None] = mapped_column(Text, nullable=True) + vps_deleted_at_utc: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True, index=True) + vps_deleted_bytes: Mapped[int | None] = mapped_column(Integer, nullable=True) + created_by_user_id: Mapped[int | None] = mapped_column(ForeignKey("users.id", ondelete="SET NULL"), nullable=True) node = relationship("BranchStorageNode") @@ -393,6 +402,15 @@ class PermanentDocumentStorageJob(CommonBase): completed_at_utc: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) acknowledged_hash_sha256: Mapped[str | None] = mapped_column(String(64), nullable=True) local_final_path: Mapped[str | None] = mapped_column(String(1000), nullable=True) + + # VPS staging cleanup audit. The file is deleted only after the branch agent + # acknowledges the exact expected SHA-256 hash. + vps_cleanup_status: Mapped[str] = mapped_column(String(30), nullable=False, default="pending", index=True) + vps_cleanup_attempts: Mapped[int] = mapped_column(Integer, nullable=False, default=0) + vps_cleanup_error: Mapped[str | None] = mapped_column(Text, nullable=True) + vps_deleted_at_utc: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True, index=True) + vps_deleted_bytes: Mapped[int | None] = mapped_column(Integer, nullable=True) + created_by_user_id: Mapped[int | None] = mapped_column(ForeignKey("users.id", ondelete="SET NULL"), nullable=True) node = relationship("BranchStorageNode") diff --git a/app/modules/documents/services.py b/app/modules/documents/services.py index c8472de..933023c 100644 --- a/app/modules/documents/services.py +++ b/app/modules/documents/services.py @@ -53,6 +53,11 @@ UDIN_REQUIRED_DOCUMENT_TYPES = {"AUDIT_REPORT", "SIGNED_OUTPUT", "ACKNOWLEDGEMEN MAX_UPLOAD_BYTES = int(os.getenv("DOCUMENT_MAX_UPLOAD_MB", "50")) * 1024 * 1024 +# A server copy is retained only as a temporary transfer buffer. It is deleted +# after the local storage agent confirms the exact expected SHA-256 hash. +# Emergency rollback: DOCUMENT_DELETE_VPS_AFTER_LOCAL_SYNC=false +DELETE_VPS_AFTER_LOCAL_SYNC = (os.getenv("DOCUMENT_DELETE_VPS_AFTER_LOCAL_SYNC", "true").strip().lower() not in {"0", "false", "no", "off"}) + def _resolve_document_storage_root(env_name: str, default_leaf: str) -> Path: """Return a short absolute storage root. @@ -815,6 +820,139 @@ def acknowledge_storage_job(db: Session, *, node: BranchStorageNode, job: Docume return False +def _prune_empty_stage_directories(file_path: Path, root: Path) -> None: + """Remove empty parent folders without crossing the configured root.""" + root = root.resolve() + parent = file_path.parent + while parent != root: + try: + parent.rmdir() + except OSError: + break + parent = parent.parent + + +def _verified_stage_path(version) -> Path | None: + """Resolve the staged path and reject traversal outside storage root.""" + if not version or not getattr(version, "local_relative_path", None): + return None + root = DEFAULT_STORAGE_ROOT.resolve() + candidate = (DEFAULT_STORAGE_ROOT / Path(version.local_relative_path)).resolve() + try: + candidate.relative_to(root) + except ValueError: + return None + return candidate + + +def _cleanup_verified_vps_stage(*, job, version) -> dict[str, object]: + """Delete one VPS stage only after verified branch-local storage.""" + result: dict[str, object] = {"enabled": DELETE_VPS_AFTER_LOCAL_SYNC, "deleted": False} + if not DELETE_VPS_AFTER_LOCAL_SYNC: + result["reason"] = "cleanup_disabled" + return result + if not job or job.status != "completed": + result["reason"] = "job_not_completed" + return result + if getattr(job, "vps_cleanup_status", "pending") == "deleted": + result.update({"deleted": True, "already_completed": True, "reason": "already_deleted"}) + return result + + expected = (job.expected_hash_sha256 or "").strip().lower() + acknowledged = (job.acknowledged_hash_sha256 or "").strip().lower() + if not expected or acknowledged != expected: + result["reason"] = "local_hash_not_verified" + return result + if not (job.local_final_path or "").strip(): + result["reason"] = "local_final_path_missing" + return result + + job.vps_cleanup_attempts = int(job.vps_cleanup_attempts or 0) + 1 + job.vps_cleanup_status = "deleting" + job.vps_cleanup_error = None + stage_path = _verified_stage_path(version) + if stage_path is None: + job.vps_cleanup_status = "failed" + job.vps_cleanup_error = "Unsafe or missing staged path." + result["reason"] = "unsafe_stage_path" + return result + + try: + deleted_bytes = int(stage_path.stat().st_size) if stage_path.exists() and stage_path.is_file() else 0 + if stage_path.exists(): + if not stage_path.is_file(): + raise OSError("Staged path is not a regular file.") + stage_path.unlink() + _prune_empty_stage_directories(stage_path, DEFAULT_STORAGE_ROOT) + job.vps_cleanup_status = "deleted" + job.vps_cleanup_error = None + job.vps_deleted_at_utc = _now_utc() + job.vps_deleted_bytes = deleted_bytes + result.update({ + "deleted": True, + "already_missing": deleted_bytes == 0, + "deleted_bytes": deleted_bytes, + "reason": "deleted_after_verified_local_sync", + }) + return result + except OSError as exc: + job.vps_cleanup_status = "retry" if int(job.vps_cleanup_attempts or 0) < 20 else "failed" + job.vps_cleanup_error = f"{type(exc).__name__}: {str(exc)[:900]}" + result.update({"reason": "delete_failed", "error": job.vps_cleanup_error}) + return result + + +def cleanup_completed_storage_job_vps_stage(db: Session, job: DocumentStorageJob) -> dict[str, object]: + version = db.get(EngagementDocumentVersion, job.version_id) if job else None + return _cleanup_verified_vps_stage(job=job, version=version) + + +def cleanup_completed_permanent_storage_job_vps_stage(db: Session, job: PermanentDocumentStorageJob) -> dict[str, object]: + version = db.get(PermanentClientDocumentVersion, job.version_id) if job else None + return _cleanup_verified_vps_stage(job=job, version=version) + + +def retry_verified_vps_cleanup_for_node(db: Session, node: BranchStorageNode, limit: int = 50) -> dict[str, int]: + """Retry interrupted/failed deletion during heartbeat or tunnel sync.""" + summary = {"checked": 0, "deleted": 0, "failed": 0} + if not DELETE_VPS_AFTER_LOCAL_SYNC or not node: + return summary + retry_states = ["pending", "retry", "failed", "deleting"] + engagement_jobs = db.execute( + select(DocumentStorageJob) + .where( + DocumentStorageJob.storage_node_id == node.id, + DocumentStorageJob.status == "completed", + DocumentStorageJob.vps_cleanup_status.in_(retry_states), + ) + .order_by(DocumentStorageJob.completed_at_utc.asc(), DocumentStorageJob.id.asc()) + .limit(limit) + ).scalars().all() + remaining = max(0, limit - len(engagement_jobs)) + permanent_jobs = [] + if remaining: + permanent_jobs = db.execute( + select(PermanentDocumentStorageJob) + .where( + PermanentDocumentStorageJob.storage_node_id == node.id, + PermanentDocumentStorageJob.status == "completed", + PermanentDocumentStorageJob.vps_cleanup_status.in_(retry_states), + ) + .order_by(PermanentDocumentStorageJob.completed_at_utc.asc(), PermanentDocumentStorageJob.id.asc()) + .limit(remaining) + ).scalars().all() + + for job in engagement_jobs: + summary["checked"] += 1 + result = cleanup_completed_storage_job_vps_stage(db, job) + summary["deleted" if result.get("deleted") else "failed"] += 1 + for job in permanent_jobs: + summary["checked"] += 1 + result = cleanup_completed_permanent_storage_job_vps_stage(db, job) + summary["deleted" if result.get("deleted") else "failed"] += 1 + return summary + + def list_storage_jobs(db: Session, tenant_id: int | None = None, branch_id: int | None = None, status: str | None = None, limit: int = 200): stmt = select(DocumentStorageJob).order_by(DocumentStorageJob.created_at_utc.desc()) if tenant_id: diff --git a/app/modules/documents/ui.py b/app/modules/documents/ui.py index da47f06..ad93814 100644 --- a/app/modules/documents/ui.py +++ b/app/modules/documents/ui.py @@ -27,6 +27,9 @@ from app.modules.documents.services import ( get_latest_version, get_version, acknowledge_storage_job, + cleanup_completed_storage_job_vps_stage, + cleanup_completed_permanent_storage_job_vps_stage, + retry_verified_vps_cleanup_for_node, authenticate_storage_node, create_branch_storage_node, generate_storage_secret, @@ -1568,8 +1571,9 @@ async def storage_agent_heartbeat(request: Request, x_node_code: str | None = He node, error = _agent_auth(db, request, x_node_code, x_node_secret) if error: return error + cleanup = retry_verified_vps_cleanup_for_node(db, node, limit=50) db.commit() - return {"ok": True, "node_code": node.node_code, "status": node.status} + return {"ok": True, "node_code": node.node_code, "status": node.status, "vps_cleanup": cleanup} finally: db.close() @@ -1644,6 +1648,8 @@ def _permanent_download_requests_payload(items): def _storage_agent_sync_payload(db, node): + # Retry cleanup in tunnel mode as well as normal heartbeat mode. + retry_verified_vps_cleanup_for_node(db, node, limit=25) jobs = _normal_storage_jobs_payload(list_pending_storage_jobs(db, node)) jobs += _permanent_storage_jobs_payload(list_pending_permanent_storage_jobs(db, node)) requests_ = _normal_download_requests_payload(list_pending_download_requests(db, node)) @@ -1719,8 +1725,17 @@ async def storage_agent_ack_job(request: Request, job_id: str, x_node_code: str if not job: return JSONResponse({"ok": False, "error": "job_not_found"}, status_code=404) ok = acknowledge_storage_job(db, node=node, job=job, acknowledged_hash=acknowledged_hash, local_final_path=local_final_path, success=bool(payload.get("success", True)), error=payload.get("error")) + # First commit the verified local acknowledgement. Only after that + # durable commit is the VPS staging file eligible for deletion. db.commit() - return {"ok": ok, "job_status": job.status} + cleanup = {"enabled": True, "deleted": False, "reason": "acknowledgement_failed"} + if ok: + if is_permanent: + cleanup = cleanup_completed_permanent_storage_job_vps_stage(db, job) + else: + cleanup = cleanup_completed_storage_job_vps_stage(db, job) + db.commit() + return {"ok": ok, "job_status": job.status, "vps_cleanup": cleanup} finally: db.close() diff --git a/app/modules/employees/service.py b/app/modules/employees/service.py index 34b4a39..d601136 100644 --- a/app/modules/employees/service.py +++ b/app/modules/employees/service.py @@ -14,6 +14,7 @@ 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 @@ -3255,6 +3256,11 @@ def escalate_employee_engagement_workflow( 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), diff --git a/app/modules/manager_dashboard/service.py b/app/modules/manager_dashboard/service.py index 0f50fb5..9c34993 100644 --- a/app/modules/manager_dashboard/service.py +++ b/app/modules/manager_dashboard/service.py @@ -7,6 +7,7 @@ from sqlalchemy import func, or_, select from sqlalchemy.orm import Session, selectinload from app.modules.clients.models import Client +from app.modules.alerts.workflow_escalations import list_workflow_escalations from app.modules.core.iam.models import User from app.modules.core.rbac.models import Permission, Role, RolePermission, UserRole from app.modules.employees.service import ( @@ -385,6 +386,7 @@ def build_manager_dashboard_payload(db: Session, request, current_user) -> dict[ staff_rows = _staff_rows(db, scope, task_rows) client_rows = _client_rows(db, scope, task_rows) advanced_engagements = _advanced_engagement_rows(tasks) + unified_escalations = list_workflow_escalations(db, tenant_id=getattr(scope, "tenant_id", None), branch_id=getattr(scope, "branch_id", None), assigned_to_user_id=current_user.id) capacity_rows = _capacity_rows(tasks) sla_breached = [row for row in advanced_engagements if row["sla"]["status"] == "breached"] sla_warning = [row for row in advanced_engagements if row["sla"]["status"] in {"critical", "warning"}] @@ -430,6 +432,8 @@ def build_manager_dashboard_payload(db: Session, request, current_user) -> dict[ "staff_count": len(staff_rows), "client_count": len(client_rows), "escalation_count": len(escalation_rows), + "unified_escalation_count": len(unified_escalations), + "escalations_over_3_days": len([r for r in unified_escalations if r["age_days"] >= 3]), "engagement_count": len(advanced_engagements), "sla_breached_count": len(sla_breached), "sla_warning_count": len(sla_warning), @@ -451,6 +455,7 @@ def build_manager_dashboard_payload(db: Session, request, current_user) -> dict[ "client_pending": client_pending_rows[:60], "documents_pending": client_pending_rows[:40], "escalations": escalation_rows[:80], + "unified_escalations": unified_escalations, "age_buckets": age_buckets, "status_summary": sorted([{"label": k, "count": v} for k, v in status_summary.items()], key=lambda x: x["count"], reverse=True)[:10], "service_summary": sorted([{"label": k, "count": v} for k, v in service_summary.items()], key=lambda x: x["count"], reverse=True)[:10], diff --git a/app/modules/manager_dashboard/templates/manager_dashboard/partials/escalations.html b/app/modules/manager_dashboard/templates/manager_dashboard/partials/escalations.html index 65d0678..67a3c3a 100644 --- a/app/modules/manager_dashboard/templates/manager_dashboard/partials/escalations.html +++ b/app/modules/manager_dashboard/templates/manager_dashboard/partials/escalations.html @@ -1,5 +1,5 @@
-
SLA Breached
{{ overview.sla_breached_count }}
SLA Warning
{{ overview.sla_warning_count }}
Task Escalations
{{ overview.escalation_count }}
Review Dependencies
{{ overview.engagement_review_count }}
-

SLA Breaches

Engagement-level statutory/internal due-date breaches.

{% for row in sla_breached %}{% else %}{% endfor %}
{{ row.client_name }} — {{ row.service_name }}
{{ row.financial_year }} · {{ row.progress_percent }}% complete
{{ row.sla.label }}
Age {{ row.sla.age_days }} day(s)
No SLA breaches.

SLA Warning Window

Due within seven days.

{% for row in sla_warning %}{% else %}{% endfor %}
{{ row.client_name }} — {{ row.service_name }}
{{ row.financial_year }} · {{ row.progress_percent }}% complete
{{ row.sla.label }}
No engagements in warning window.
-

Existing Task Escalation Register

Unassigned, overdue and client-pending tasks remain visible.

Open Alerts
{% set rows = escalations %}{% include 'modules/manager_dashboard/templates/manager_dashboard/partials/_task_table.html' %}
-
+
Open Escalations
{{ overview.unified_escalation_count }}
Age 3+ Days
{{ overview.escalations_over_3_days }}
SLA Breached
{{ overview.sla_breached_count }}
+

Unified Escalation Register

Durable staff-to-manager escalations with acknowledgement, resolution and ageing.

{% for row in unified_escalations %}{% else %}{% endfor %}
EngagementEscalationAgeAction
{{ row.client_name }} — {{ row.service_name }}
{{ row.financial_year }} · Raised by {{ row.raised_by }}
{{ row.category|replace('_',' ')|title }}
{{ row.message }}
Status: {{ row.status|title }}{% if row.follow_up_date %} · Follow-up {{ row.follow_up_date }}{% endif %}
{{ row.age_days }} day(s)
No open escalations assigned to you.
+

Review Ageing — SLA Breaches

{% for row in sla_breached %}
{{ row.client_name }} — {{ row.service_name }}
{{ row.sla.label }} · Age {{ row.sla.age_days }} day(s)
{% else %}
No breached engagements.
{% endfor %}

Review Ageing — Warning

{% for row in sla_warning %}
{{ row.client_name }} — {{ row.service_name }}
{{ row.sla.label }}
{% else %}
No warning-window engagements.
{% endfor %}
+ \ No newline at end of file diff --git a/app/modules/manager_dashboard/ui.py b/app/modules/manager_dashboard/ui.py index ca18cd1..81310f2 100644 --- a/app/modules/manager_dashboard/ui.py +++ b/app/modules/manager_dashboard/ui.py @@ -16,6 +16,7 @@ from app.modules.manager_dashboard.service import ( get_next_manager_review_task_id, ) from app.modules.services.execution import apply_task_review +from app.modules.alerts.workflow_escalations import update_workflow_escalation router = APIRouter(prefix="/manager", tags=["manager-dashboard-v2-ui"]) @@ -180,3 +181,17 @@ def manager_review_task_submit( return RedirectResponse(url=f"/manager/reviews/engagements/{subscription_id}?task_id={task_id}&reviewed=1", status_code=303) finally: db.close() + + +@router.post("/escalations/{escalation_id}/action") +def manager_escalation_action(request: Request, escalation_id: int, action: str = Form(...), resolution_note: str = Form(""), csrf_token: str = Form(...)): + validate_csrf(request, csrf_token) + db=CommonSessionLocal() + try: + current_user=get_current_user(request, db=db) + if not current_user: return RedirectResponse(url="/login",status_code=303) + if not can_access_manager_dashboard(db,current_user): return ui_access_denied() + try: update_workflow_escalation(db, escalation_id=escalation_id, actor_user_id=current_user.id, action=action, resolution_note=resolution_note); db.commit() + except ValueError: db.rollback() + return RedirectResponse(url="/manager/dashboard?tab=escalations",status_code=303) + finally: db.close() diff --git a/app/modules/partner_dashboard/service.py b/app/modules/partner_dashboard/service.py index 382656c..a3c91db 100644 --- a/app/modules/partner_dashboard/service.py +++ b/app/modules/partner_dashboard/service.py @@ -8,6 +8,7 @@ from sqlalchemy import func, or_, select from sqlalchemy.orm import Session, selectinload from app.modules.clients.models import Client +from app.modules.alerts.workflow_escalations import list_workflow_escalations from app.modules.core.iam.models import User from app.modules.core.rbac.models import Role, UserRole from app.modules.core.tenancy.models import Branch, Tenant @@ -297,6 +298,7 @@ def build_partner_dashboard_payload(db: Session, request, current_user) -> dict[ clients = _load_clients(db, tenant_id, branch_id, current_user, roles) staff = _staff_rows(db, tenant_id, branch_id, tasks) billing = _billing_rows(db, tenant_id, branch_id, fy) + unified_escalations = list_workflow_escalations(db, tenant_id=tenant_id, branch_id=branch_id, assigned_to_user_id=current_user.id) subscriptions_count = _count(db, _subscription_scope(select(func.count(ClientServiceSubscription.id)), tenant_id, branch_id, current_user, roles, fy)) if tenant_id else 0 @@ -318,6 +320,8 @@ def build_partner_dashboard_payload(db: Session, request, current_user) -> dict[ "invoice_count": billing["invoice_count"], "outstanding": billing["outstanding"], "today": today, + "unified_escalation_count": len(unified_escalations), + "escalations_over_3_days": len([r for r in unified_escalations if r["age_days"] >= 3]), } return { @@ -330,6 +334,7 @@ def build_partner_dashboard_payload(db: Session, request, current_user) -> dict[ "client_pending": client_pending_rows[:25], "review_queue": review_rows[:50], "partner_engagement_review_queue": _partner_review_queue_rows(db, request, current_user), + "unified_escalations": unified_escalations, "clients": clients, "staff_rows": staff, "billing": billing, diff --git a/app/modules/partner_dashboard/templates/partner_dashboard/partials/escalations.html b/app/modules/partner_dashboard/templates/partner_dashboard/partials/escalations.html new file mode 100644 index 0000000..b7f7300 --- /dev/null +++ b/app/modules/partner_dashboard/templates/partner_dashboard/partials/escalations.html @@ -0,0 +1 @@ +
Open Escalations
{{ overview.unified_escalation_count }}
Age 3+ Days
{{ overview.escalations_over_3_days }}
Partner Reviews
{{ overview.review_pending_count }}

Unified Partner Escalation & Review Ageing

Escalations assigned to the Engagement Partner or Review Partner.

{% for row in unified_escalations %}{% else %}{% endfor %}
{{ row.client_name }} — {{ row.service_name }}
{{ row.financial_year }} · {{ row.level|replace('_',' ')|title }} · {{ row.age_days }} day(s)
{{ row.message }}
No open escalations assigned to you.

Partner Review Ageing

{% for row in partner_engagement_review_queue %}
{{ row.client_name }} — {{ row.service_name }}
{{ row.financial_year }} · Partner {{ row.pending_partner }} · Review Partner {{ row.pending_review_partner }} · {{ row.sla.label }}
{% else %}
No pending partner reviews.
{% endfor %}
\ No newline at end of file diff --git a/app/modules/partner_dashboard/templates/partner_dashboard/partials/review.html b/app/modules/partner_dashboard/templates/partner_dashboard/partials/review.html index 2dcaed4..eebfa53 100644 --- a/app/modules/partner_dashboard/templates/partner_dashboard/partials/review.html +++ b/app/modules/partner_dashboard/templates/partner_dashboard/partials/review.html @@ -1,4 +1,5 @@
+
Escalations & Review Ageing
diff --git a/app/modules/partner_dashboard/ui.py b/app/modules/partner_dashboard/ui.py index b99495d..2653043 100644 --- a/app/modules/partner_dashboard/ui.py +++ b/app/modules/partner_dashboard/ui.py @@ -16,6 +16,7 @@ from app.modules.partner_dashboard.service import ( get_partner_review_workspace, ) from app.modules.services.execution import apply_task_review +from app.modules.alerts.workflow_escalations import update_workflow_escalation router = APIRouter(prefix="/partner", tags=["partner-dashboard-v2-ui"]) @@ -25,6 +26,7 @@ VALID_TABS = { "clients": "modules/partner_dashboard/templates/partner_dashboard/partials/clients.html", "staff": "modules/partner_dashboard/templates/partner_dashboard/partials/staff.html", "review": "modules/partner_dashboard/templates/partner_dashboard/partials/review.html", + "escalations": "modules/partner_dashboard/templates/partner_dashboard/partials/escalations.html", "billing": "modules/partner_dashboard/templates/partner_dashboard/partials/billing.html", "reports": "modules/partner_dashboard/templates/partner_dashboard/partials/reports.html", "wizards": "modules/partner_dashboard/templates/partner_dashboard/partials/wizards.html", @@ -168,3 +170,17 @@ def partner_review_task_submit( return RedirectResponse(url=f"/partner/reviews/engagements/{subscription_id}?task_id={task_id}&reviewed=1", status_code=303) finally: db.close() + + +@router.post("/escalations/{escalation_id}/action") +def partner_escalation_action(request: Request, escalation_id: int, action: str = Form(...), resolution_note: str = Form(""), csrf_token: str = Form(...)): + validate_csrf(request, csrf_token) + db=CommonSessionLocal() + try: + current_user=get_current_user(request, db=db) + if not current_user: return RedirectResponse(url="/login",status_code=303) + if not can_access_partner_dashboard(db,current_user): return ui_access_denied() + try: update_workflow_escalation(db, escalation_id=escalation_id, actor_user_id=current_user.id, action=action, resolution_note=resolution_note); db.commit() + except ValueError: db.rollback() + return RedirectResponse(url="/partner/dashboard?tab=escalations",status_code=303) + finally: db.close()