from __future__ import annotations from dataclasses import dataclass, field from datetime import datetime, timezone import hashlib import json import re from typing import Any from sqlalchemy import func, select from sqlalchemy.orm import Session from app.modules.services.models import ( ClientServiceTaskInstance, FirmServiceSelection, FirmServiceTaskTemplate, ServiceDefaultTaskTemplate, ServiceTaskCategory, ) # Fields that define the centrally maintained system-default task snapshot. # task_category_id is deliberately excluded because system and firm category # masters use different scoped rows; task_category text is the portable value. SYSTEM_TASK_FIELDS: tuple[str, ...] = ( "task_name", "description", "sequence_no", "default_role_name", "eligible_role_names", "is_mandatory", "requires_review", "normal_review_role", "task_category", "response_required", "response_type", "evidence_required", "remarks_required_if_no", "task_tool_code", "is_aqmm_task", "aqmm_mandatory", "aqmm_evidence_required", "aqmm_manager_review_required", "aqmm_partner_review_required", "aqmm_review_partner_required", "aqmm_blocks_final_release", "aqmm_reference", "is_active", ) _SPACE_RE = re.compile(r"\s+") @dataclass class FirmDefaultTaskSyncResult: created: int = 0 updated: int = 0 duplicates_disabled: int = 0 retired: int = 0 unchanged: int = 0 custom_updates_available: int = 0 linked_legacy: int = 0 engagement_created: int = 0 engagement_updated_pending: int = 0 engagement_deactivated_pending: int = 0 engagement_preserved_history: int = 0 @property def active_total_change(self) -> int: return self.created - self.duplicates_disabled @dataclass class SystemDefaultRolloutResult: firms_processed: int = 0 firms_changed: int = 0 created: int = 0 updated: int = 0 unchanged: int = 0 custom_updates_available: int = 0 duplicates_disabled: int = 0 engagement_created: int = 0 engagement_updated_pending: int = 0 engagement_deactivated_pending: int = 0 engagement_preserved_history: int = 0 tenant_results: dict[int, FirmDefaultTaskSyncResult] = field(default_factory=dict) def _normalise_name(value: str | None) -> str: return _SPACE_RE.sub(" ", (value or "").strip()).casefold() def _portable_value(value: Any) -> Any: if isinstance(value, (str, int, float, bool)) or value is None: return value return str(value) def _snapshot_dict(row: Any) -> dict[str, Any]: return {name: _portable_value(getattr(row, name, None)) for name in SYSTEM_TASK_FIELDS} def system_task_hash(source: ServiceDefaultTaskTemplate) -> str: payload = json.dumps(_snapshot_dict(source), sort_keys=True, separators=(",", ":"), ensure_ascii=False) return hashlib.sha256(payload.encode("utf-8")).hexdigest() def firm_task_hash(target: FirmServiceTaskTemplate) -> str: payload = json.dumps(_snapshot_dict(target), sort_keys=True, separators=(",", ":"), ensure_ascii=False) return hashlib.sha256(payload.encode("utf-8")).hexdigest() def _legacy_retirement_equivalent( source: ServiceDefaultTaskTemplate, target: FirmServiceTaskTemplate, ) -> bool: """Return True when a legacy firm row differs only because the system row was retired. Full-sync retirement moves omitted system rows into a high temporary sequence range before setting is_active=False. Legacy firm rows created before provenance tracking still carry their old active flag and old sequence. Those two differences must not make an otherwise inherited task look like a deliberate firm customization. Any substantive field difference still protects the firm row as customized. """ source_snapshot = _snapshot_dict(source) target_snapshot = _snapshot_dict(target) source_snapshot.pop("is_active", None) target_snapshot.pop("is_active", None) if int(getattr(source, "sequence_no", 0) or 0) >= 100000: source_snapshot.pop("sequence_no", None) target_snapshot.pop("sequence_no", None) return source_snapshot == target_snapshot def system_task_diff(source: ServiceDefaultTaskTemplate, target: FirmServiceTaskTemplate) -> list[dict[str, Any]]: labels = { "task_name": "Task name", "description": "Description", "sequence_no": "Sequence", "default_role_name": "Default role", "eligible_role_names": "Eligible roles", "is_mandatory": "Mandatory", "requires_review": "Review required", "normal_review_role": "Normal reviewer", "task_category": "Task category", "response_required": "Response required", "response_type": "Response type", "evidence_required": "Evidence required", "remarks_required_if_no": "Remarks required if No", "task_tool_code": "Task tool", "is_aqmm_task": "AQMM task", "aqmm_mandatory": "AQMM mandatory", "aqmm_evidence_required": "AQMM evidence", "aqmm_manager_review_required": "AQMM manager review", "aqmm_partner_review_required": "AQMM partner review", "aqmm_review_partner_required": "AQMM review partner", "aqmm_blocks_final_release": "Blocks final release", "aqmm_reference": "AQMM reference", "is_active": "Active", } changes: list[dict[str, Any]] = [] for name in SYSTEM_TASK_FIELDS: old = getattr(target, name, None) new = getattr(source, name, None) if old != new: changes.append({"field": name, "label": labels.get(name, name), "firm": old, "system": new}) return changes def mark_firm_task_customized(task: FirmServiceTaskTemplate) -> None: """Mark an explicitly edited firm task as protected from automatic overwrite.""" task.is_customized = True # Do not clear an already pending system update. If there is no pending update, # the next system revision/hash change will create one automatically. def _ensure_firm_category( db: Session, *, source: ServiceDefaultTaskTemplate, tenant_id: int, user_id: int | None, ) -> ServiceTaskCategory | None: name = (getattr(source, "task_category", None) or "").strip() if not name: return None system_category = None source_category_id = getattr(source, "task_category_id", None) if source_category_id: system_category = db.get(ServiceTaskCategory, int(source_category_id)) code = ((getattr(system_category, "code", None) or "").strip().upper() if system_category else "") if code: existing = db.execute( select(ServiceTaskCategory).where( ServiceTaskCategory.tenant_id == tenant_id, ServiceTaskCategory.service_catalogue_id == source.service_catalogue_id, ServiceTaskCategory.code == code, ) ).scalar_one_or_none() else: existing = db.execute( select(ServiceTaskCategory).where( ServiceTaskCategory.tenant_id == tenant_id, ServiceTaskCategory.service_catalogue_id == source.service_catalogue_id, func.lower(ServiceTaskCategory.name) == name.lower(), ) ).scalar_one_or_none() if existing: # Central category rename/status/order should remain aligned for inherited use. # Firm category rows are shared by firm templates, so update only metadata that # does not destroy task history. if system_category: existing.name = system_category.name existing.sort_order = system_category.sort_order existing.is_active = system_category.is_active if user_id is not None: existing.updated_by_user_id = user_id return existing if not code: # Match the application's existing category-code convention sufficiently for # automatic inherited categories without importing services.py (avoids cycles). code = re.sub(r"[^A-Z0-9]+", "_", name.upper()).strip("_")[:50] or "CATEGORY" base = code suffix = 2 while db.execute( select(ServiceTaskCategory.id).where( ServiceTaskCategory.tenant_id == tenant_id, ServiceTaskCategory.service_catalogue_id == source.service_catalogue_id, ServiceTaskCategory.code == code, ) ).first(): code = f"{base[:45]}_{suffix}" suffix += 1 row = ServiceTaskCategory( tenant_id=tenant_id, service_catalogue_id=source.service_catalogue_id, code=code, name=(system_category.name if system_category else name), sort_order=(system_category.sort_order if system_category else 100), is_active=(system_category.is_active if system_category else True), created_by_user_id=user_id, updated_by_user_id=user_id, ) db.add(row) db.flush() return row def _copy_default_columns( db: Session, source: ServiceDefaultTaskTemplate, target: FirmServiceTaskTemplate, *, tenant_id: int, user_id: int | None, allow_sequence_change: bool = True, ) -> bool: changed = False for name in SYSTEM_TASK_FIELDS: if name == "sequence_no" and not allow_sequence_change: continue new_value = getattr(source, name, None) if getattr(target, name, None) != new_value: setattr(target, name, new_value) changed = True category = _ensure_firm_category(db, source=source, tenant_id=tenant_id, user_id=user_id) category_id = category.id if category else None category_name = category.name if category else None if getattr(target, "task_category_id", None) != category_id: target.task_category_id = category_id changed = True if getattr(target, "task_category", None) != category_name: target.task_category = category_name changed = True return changed def _reference_counts(db: Session, *, firm_task_ids: list[int]) -> dict[int, int]: if not firm_task_ids: return {} rows = db.execute( select(ClientServiceTaskInstance.firm_task_template_id, func.count(ClientServiceTaskInstance.id)) .where(ClientServiceTaskInstance.firm_task_template_id.in_(firm_task_ids)) .group_by(ClientServiceTaskInstance.firm_task_template_id) ).all() return {int(template_id): int(count) for template_id, count in rows if template_id is not None} def _choose_canonical(candidates: list[FirmServiceTaskTemplate], *, reference_counts: dict[int, int]) -> FirmServiceTaskTemplate: return sorted(candidates, key=lambda row: (-reference_counts.get(int(row.id), 0), int(row.id)))[0] def _next_free_sequence(used: set[int], preferred: int) -> int: if preferred > 0 and preferred not in used: return preferred candidate = max(used or {0}) + 1 while candidate in used: candidate += 1 return candidate def sync_firm_tasks_from_system_defaults( db: Session, *, tenant_id: int, service_catalogue_id: int, updated_by_user_id: int | None = None, sync_open_engagements: bool = False, ) -> FirmDefaultTaskSyncResult: """Synchronise one firm's task templates from centrally maintained defaults. Inherited tasks are updated automatically. Explicitly customized linked tasks are never overwritten; a system_update_available flag is raised for Firm Admin review. Firm-only tasks remain untouched. Legacy rows are linked by exact normalized name. """ defaults = db.execute( select(ServiceDefaultTaskTemplate) .where(ServiceDefaultTaskTemplate.service_catalogue_id == service_catalogue_id) .order_by(ServiceDefaultTaskTemplate.sequence_no.asc(), ServiceDefaultTaskTemplate.id.asc()) ).scalars().all() firm_rows = db.execute( select(FirmServiceTaskTemplate) .where( FirmServiceTaskTemplate.tenant_id == tenant_id, FirmServiceTaskTemplate.service_catalogue_id == service_catalogue_id, ) .order_by(FirmServiceTaskTemplate.sequence_no.asc(), FirmServiceTaskTemplate.id.asc()) ).scalars().all() result = FirmDefaultTaskSyncResult() reference_counts = _reference_counts(db, firm_task_ids=[int(r.id) for r in firm_rows if r.id]) by_source: dict[int, list[FirmServiceTaskTemplate]] = {} legacy_by_name: dict[str, list[FirmServiceTaskTemplate]] = {} for row in firm_rows: if row.source_system_task_id: by_source.setdefault(int(row.source_system_task_id), []).append(row) else: key = _normalise_name(row.task_name) if key: legacy_by_name.setdefault(key, []).append(row) used_sequences = {int(r.sequence_no) for r in firm_rows if r.sequence_no is not None} for source in defaults: latest_hash = system_task_hash(source) candidates = by_source.get(int(source.id), []) linked_from_legacy = False if not candidates: legacy_candidates = legacy_by_name.get(_normalise_name(source.task_name), []) if not legacy_candidates: # One-time migration fallback for a firm that renamed a previously # copied default before provenance columns existed. Sequence is used # only when there is exactly one unlinked candidate at that position. legacy_candidates = [ row for row in firm_rows if row.source_system_task_id is None and int(row.sequence_no or 0) == int(source.sequence_no or 0) ] if len(legacy_candidates) != 1: legacy_candidates = [] if legacy_candidates: target = _choose_canonical(legacy_candidates, reference_counts=reference_counts) candidates = [target] target.source_system_task_id = source.id linked_from_legacy = True result.linked_legacy += 1 by_source.setdefault(int(source.id), []).append(target) # Remove from future legacy matching. key = _normalise_name(target.task_name) if key in legacy_by_name: legacy_by_name[key] = [r for r in legacy_by_name[key] if r.id != target.id] if candidates: target = _choose_canonical(candidates, reference_counts=reference_counts) for duplicate in candidates: if duplicate.id == target.id: continue if duplicate.is_active: duplicate.is_active = False if updated_by_user_id is not None: duplicate.updated_by_user_id = updated_by_user_id result.duplicates_disabled += 1 # For legacy rows with no inheritance history, infer whether they were # already firm-customized by comparing the full portable snapshot. if linked_from_legacy and target.last_synced_system_hash is None: if not source.is_active and _legacy_retirement_equivalent(source, target): # This was an inherited legacy row whose system source has just been # retired by Full Synchronization. Do not misclassify the active/old # sequence difference as a firm customization; allow normal retirement # propagation below. target.is_customized = False else: target.is_customized = firm_task_hash(target) != latest_hash if not target.is_customized: target.last_synced_system_hash = latest_hash target.last_reviewed_system_hash = latest_hash # Repair rows linked by an earlier rollout where retirement-only differences # were incorrectly interpreted as customization. This is deliberately limited # to rows that have never had a successful inherited snapshot recorded. if ( not source.is_active and target.is_customized and target.last_synced_system_hash is None and _legacy_retirement_equivalent(source, target) ): target.is_customized = False target.system_update_available = False target.system_update_detected_at_utc = None # Detect out-of-band edits to a linked inherited row as customization. if ( not target.is_customized and target.last_synced_system_hash and firm_task_hash(target) != target.last_synced_system_hash ): target.is_customized = True if target.is_customized: pending = target.last_reviewed_system_hash != latest_hash target.system_update_available = pending target.system_update_detected_at_utc = datetime.now(timezone.utc) if pending else None if pending: result.custom_updates_available += 1 else: result.unchanged += 1 continue # Inherited task: copy system values. Sequence changes are applied when # the desired number is free or currently owned by this same row. If a # firm-only/custom row owns it, preserve that row and place this inherited # task at the next free sequence instead of overwriting customization. desired_seq = int(source.sequence_no or 0) current_seq = int(target.sequence_no or 0) allow_seq = desired_seq == current_seq or desired_seq not in (used_sequences - {current_seq}) if allow_seq: used_sequences.discard(current_seq) used_sequences.add(desired_seq) changed = _copy_default_columns( db, source, target, tenant_id=tenant_id, user_id=updated_by_user_id, allow_sequence_change=allow_seq, ) target.source_system_task_id = source.id target.last_synced_system_hash = latest_hash target.last_reviewed_system_hash = latest_hash target.system_update_available = False target.system_update_detected_at_utc = None if updated_by_user_id is not None: target.updated_by_user_id = updated_by_user_id result.updated += 1 if changed else 0 result.unchanged += 0 if changed else 1 continue # Missing default: create a new inherited row. Avoid colliding with a # firm-only sequence; identity is source_system_task_id, not sequence number. preferred = int(source.sequence_no or 0) seq = _next_free_sequence(used_sequences, preferred) used_sequences.add(seq) row = FirmServiceTaskTemplate( tenant_id=tenant_id, service_catalogue_id=service_catalogue_id, task_name=source.task_name, sequence_no=seq, source_system_task_id=source.id, is_customized=False, last_synced_system_hash=latest_hash, last_reviewed_system_hash=latest_hash, system_update_available=False, created_by_user_id=updated_by_user_id, updated_by_user_id=updated_by_user_id, ) db.add(row) db.flush() _copy_default_columns( db, source, row, tenant_id=tenant_id, user_id=updated_by_user_id, allow_sequence_change=(seq == preferred), ) result.created += 1 firm_rows.append(row) by_source.setdefault(int(source.id), []).append(row) db.flush() if sync_open_engagements: from app.modules.services.execution import sync_open_engagement_tasks_for_service engagement = sync_open_engagement_tasks_for_service( db, tenant_id=tenant_id, catalogue_id=service_catalogue_id, user_id=updated_by_user_id or 0, include_started_open_tasks=False, safe_system_rollout=True, ) result.engagement_created = engagement.get("created", 0) result.engagement_updated_pending = engagement.get("updated_pending", 0) result.engagement_deactivated_pending = engagement.get("deactivated_pending", 0) result.engagement_preserved_history = engagement.get("preserved_history", 0) return result def fully_align_firm_tasks_to_system_defaults( db: Session, *, tenant_id: int, service_catalogue_id: int, updated_by_user_id: int, sync_open_engagements: bool = True, ) -> FirmDefaultTaskSyncResult: """Fully align one firm's system-derived checklist to the current system master. This is the bulk/explicit "Upgrade All to System" operation. It is intentionally stronger than the normal background sync: * every row already linked to a system task follows that system task, even when an earlier firm customization exists; * linked rows whose system source is retired are retired at firm level; * legacy unlinked rows are linked only by an exact normalized task-name match; * duplicate rows for the same system source are retired; * active system-derived rows receive the exact current system sequence; * genuine firm-only rows (no reliable system match) are preserved. If one occupies a sequence now required by the system master it is moved to the next free sequence; * engagement instances are synchronized once after the template reconciliation, using the existing safe rollout rules so historical/started/reviewed work is not rewritten. No task/template row is hard-deleted. """ defaults = db.execute( select(ServiceDefaultTaskTemplate) .where(ServiceDefaultTaskTemplate.service_catalogue_id == service_catalogue_id) .order_by(ServiceDefaultTaskTemplate.sequence_no.asc(), ServiceDefaultTaskTemplate.id.asc()) ).scalars().all() firm_rows = db.execute( select(FirmServiceTaskTemplate) .where( FirmServiceTaskTemplate.tenant_id == tenant_id, FirmServiceTaskTemplate.service_catalogue_id == service_catalogue_id, ) .order_by(FirmServiceTaskTemplate.sequence_no.asc(), FirmServiceTaskTemplate.id.asc()) ).scalars().all() result = FirmDefaultTaskSyncResult() reference_counts = _reference_counts(db, firm_task_ids=[int(r.id) for r in firm_rows if r.id]) defaults_by_id = {int(row.id): row for row in defaults} defaults_by_name: dict[str, list[ServiceDefaultTaskTemplate]] = {} for source in defaults: key = _normalise_name(source.task_name) if key: defaults_by_name.setdefault(key, []).append(source) # Capture original firm ordering before temporarily moving rows away from the # unique (tenant, service, sequence) namespace. original_sequence = {int(row.id): int(row.sequence_no or 0) for row in firm_rows if row.id is not None} # Safely recover provenance for legacy rows. Exact name is deliberately the only # automatic fallback here: anything ambiguous remains a genuine firm-only row. for row in firm_rows: if row.source_system_task_id is not None: continue matches = defaults_by_name.get(_normalise_name(row.task_name), []) if len(matches) == 1: row.source_system_task_id = matches[0].id result.linked_legacy += 1 db.flush() by_source: dict[int, list[FirmServiceTaskTemplate]] = {} firm_only: list[FirmServiceTaskTemplate] = [] for row in firm_rows: source_id = int(row.source_system_task_id) if row.source_system_task_id is not None else None if source_id is not None and source_id in defaults_by_id: by_source.setdefault(source_id, []).append(row) else: # A missing/deleted source becomes firm-only because the FK uses SET NULL; # never guess and retire it without a reliable system identity. if source_id is not None and source_id not in defaults_by_id: row.source_system_task_id = None firm_only.append(row) # Free every sequence first. This allows exact system sequencing even when an old # firm row currently occupies the target number. The unique constraint remains valid # because each temporary sequence is unique. highest_existing = max([int(r.sequence_no or 0) for r in firm_rows] + [0]) highest_default = max([int(r.sequence_no or 0) for r in defaults] + [0]) temporary_base = max(highest_existing, highest_default, 100000) + 1000000 for offset, row in enumerate(sorted(firm_rows, key=lambda r: int(r.id or 0)), start=1): row.sequence_no = temporary_base + offset db.flush() canonical_by_source: dict[int, FirmServiceTaskTemplate] = {} for source in defaults: source_id = int(source.id) latest_hash = system_task_hash(source) candidates = by_source.get(source_id, []) if candidates: # Prefer the currently active row, then the row referenced by the most # engagement history, then the oldest id. This avoids switching canonical # identity unnecessarily while still preserving historical references. target = sorted( candidates, key=lambda row: ( -int(bool(row.is_active)), -reference_counts.get(int(row.id), 0), int(row.id), ), )[0] canonical_by_source[source_id] = target for duplicate in candidates: if duplicate.id == target.id: continue if duplicate.is_active: result.retired += 1 duplicate.is_active = False duplicate.is_customized = False duplicate.last_synced_system_hash = latest_hash duplicate.last_reviewed_system_hash = latest_hash duplicate.system_update_available = False duplicate.system_update_detected_at_utc = None duplicate.updated_by_user_id = updated_by_user_id result.duplicates_disabled += 1 elif source.is_active: # Create only active system defaults. Retired defaults with no firm row do # not need a new historical firm record. target = FirmServiceTaskTemplate( tenant_id=tenant_id, service_catalogue_id=service_catalogue_id, task_name=source.task_name, sequence_no=temporary_base + len(firm_rows) + len(canonical_by_source) + 1, source_system_task_id=source.id, is_customized=False, last_synced_system_hash=latest_hash, last_reviewed_system_hash=latest_hash, system_update_available=False, created_by_user_id=updated_by_user_id, updated_by_user_id=updated_by_user_id, ) db.add(target) db.flush() firm_rows.append(target) by_source.setdefault(source_id, []).append(target) canonical_by_source[source_id] = target result.created += 1 else: continue target = canonical_by_source[source_id] was_active = bool(target.is_active) changed = _copy_default_columns( db, source, target, tenant_id=tenant_id, user_id=updated_by_user_id, allow_sequence_change=False, ) target.source_system_task_id = source.id target.is_customized = False target.last_synced_system_hash = latest_hash target.last_reviewed_system_hash = latest_hash target.system_update_available = False target.system_update_detected_at_utc = None target.updated_by_user_id = updated_by_user_id if was_active and not bool(source.is_active): result.retired += 1 if changed: result.updated += 1 else: result.unchanged += 1 db.flush() # Active system rows now own the exact master sequence numbers. used_sequences: set[int] = set() active_sources = [row for row in defaults if bool(row.is_active)] for source in sorted(active_sources, key=lambda r: (int(r.sequence_no or 0), int(r.id))): target = canonical_by_source.get(int(source.id)) if target is None: continue desired = int(source.sequence_no or 0) target.sequence_no = desired used_sequences.add(desired) db.flush() # Genuine firm-only active tasks survive the full upgrade. Keep their original # sequence when it is still free; otherwise move them after the system checklist. next_free = max(used_sequences or {0}) + 1 active_firm_only = sorted( [row for row in firm_only if bool(row.is_active)], key=lambda row: (original_sequence.get(int(row.id), 0), int(row.id)), ) for row in active_firm_only: preferred = original_sequence.get(int(row.id), 0) if preferred > 0 and preferred not in used_sequences: row.sequence_no = preferred used_sequences.add(preferred) else: while next_free in used_sequences: next_free += 1 row.sequence_no = next_free used_sequences.add(next_free) next_free += 1 # Retired/duplicate templates are retained for history but moved outside the live # sequence range so they can never block exact system sequencing. inactive_rows = [row for row in firm_rows if not bool(row.is_active)] retired_base = max(used_sequences or {0}) + 100000 for offset, row in enumerate(sorted(inactive_rows, key=lambda r: int(r.id or 0)), start=1): row.sequence_no = retired_base + offset db.flush() if sync_open_engagements: from app.modules.services.execution import sync_open_engagement_tasks_for_service engagement = sync_open_engagement_tasks_for_service( db, tenant_id=tenant_id, catalogue_id=service_catalogue_id, user_id=updated_by_user_id, include_started_open_tasks=False, safe_system_rollout=True, ) result.engagement_created = engagement.get("created", 0) result.engagement_updated_pending = engagement.get("updated_pending", 0) result.engagement_deactivated_pending = engagement.get("deactivated_pending", 0) result.engagement_preserved_history = engagement.get("preserved_history", 0) return result def sync_system_defaults_to_all_firms( db: Session, *, service_catalogue_id: int, updated_by_user_id: int | None = None, sync_open_engagements: bool = True, ) -> SystemDefaultRolloutResult: """Roll a system-default service checklist to every firm that enabled it.""" selections = db.execute( select(FirmServiceSelection).where( FirmServiceSelection.service_catalogue_id == service_catalogue_id, FirmServiceSelection.is_enabled.is_(True), ) ).scalars().all() aggregate = SystemDefaultRolloutResult() for selection in selections: result = sync_firm_tasks_from_system_defaults( db, tenant_id=int(selection.tenant_id), service_catalogue_id=service_catalogue_id, updated_by_user_id=updated_by_user_id, sync_open_engagements=sync_open_engagements, ) aggregate.firms_processed += 1 if result.created or result.updated or result.duplicates_disabled or result.custom_updates_available: aggregate.firms_changed += 1 aggregate.created += result.created aggregate.updated += result.updated aggregate.unchanged += result.unchanged aggregate.custom_updates_available += result.custom_updates_available aggregate.duplicates_disabled += result.duplicates_disabled aggregate.engagement_created += result.engagement_created aggregate.engagement_updated_pending += result.engagement_updated_pending aggregate.engagement_deactivated_pending += result.engagement_deactivated_pending aggregate.engagement_preserved_history += result.engagement_preserved_history aggregate.tenant_results[int(selection.tenant_id)] = result return aggregate def accept_system_update_for_firm_task( db: Session, *, task: FirmServiceTaskTemplate, user_id: int, sync_open_engagements: bool = True, ) -> FirmDefaultTaskSyncResult: if not task.source_system_task_id: raise ValueError("This firm task is not linked to a system default.") source = db.get(ServiceDefaultTaskTemplate, int(task.source_system_task_id)) if source is None: raise ValueError("The linked system default no longer exists.") latest_hash = system_task_hash(source) used = { int(v) for v in db.scalars( select(FirmServiceTaskTemplate.sequence_no).where( FirmServiceTaskTemplate.tenant_id == task.tenant_id, FirmServiceTaskTemplate.service_catalogue_id == task.service_catalogue_id, FirmServiceTaskTemplate.id != task.id, ) ).all() if v is not None } desired = int(source.sequence_no or 0) allow_seq = desired not in used changed = _copy_default_columns( db, source, task, tenant_id=int(task.tenant_id), user_id=user_id, allow_sequence_change=allow_seq, ) task.is_customized = False task.last_synced_system_hash = latest_hash task.last_reviewed_system_hash = latest_hash task.system_update_available = False task.system_update_detected_at_utc = None task.updated_by_user_id = user_id db.flush() result = FirmDefaultTaskSyncResult(updated=1 if changed else 0, unchanged=0 if changed else 1) if sync_open_engagements: from app.modules.services.execution import sync_open_engagement_tasks_for_service engagement = sync_open_engagement_tasks_for_service( db, tenant_id=int(task.tenant_id), catalogue_id=int(task.service_catalogue_id), user_id=user_id, include_started_open_tasks=False, safe_system_rollout=True, ) result.engagement_created = engagement.get("created", 0) result.engagement_updated_pending = engagement.get("updated_pending", 0) result.engagement_deactivated_pending = engagement.get("deactivated_pending", 0) result.engagement_preserved_history = engagement.get("preserved_history", 0) return result def keep_firm_customization_for_system_revision( db: Session, *, task: FirmServiceTaskTemplate, user_id: int, ) -> None: if not task.source_system_task_id: raise ValueError("This firm task is not linked to a system default.") source = db.get(ServiceDefaultTaskTemplate, int(task.source_system_task_id)) if source is None: raise ValueError("The linked system default no longer exists.") task.is_customized = True task.last_reviewed_system_hash = system_task_hash(source) task.system_update_available = False task.system_update_detected_at_utc = None task.updated_by_user_id = user_id db.flush()