"""Close task model service — templates, instances, roll-forward, aggregation. Owns the per-period close *tasks*: seeding the 129-row template library from the captured checklist, resolving owners (via the close coordinator) and entity codes, instantiating a period's tasks (initial population or roll- forward), task CRUD + status, deadline (re)computation from the run's anchors, and the per-owner / per-process / overall progress rollups the console shows. """ from __future__ import annotations import uuid from datetime import UTC, datetime from sqlalchemy import func, select from sqlalchemy import text as sa_text from sqlalchemy.exc import IntegrityError from sqlalchemy.ext.asyncio import AsyncSession from app.core.coordinators import finance_close_coordinator as coord from app.core.exceptions import NotFoundError, ValidationError from app.domains.finance import close_seed_loader as loader from app.domains.finance import close_service from app.domains.finance.close_models import ( AI_WORK_VALUES, FREQ_ONGOING, FREQUENCIES, RUN_FROZEN_STATUSES, TASK_COMPLETED, TASK_IN_PROGRESS, TASK_NOT_STARTED, TASK_STATUSES, FinanceCloseEntity, FinanceCloseTask, FinanceCloseTaskTemplate, ) _SHARED_FIELDS = ( "process", "sub_process", "description", "owner_person_id", "owner_label", "frequency", "calendar_day", "estimated_hours", "submit_to_fpa", "ai_work", "link", "entity_codes", "sort_order", "signal_key", ) def _now() -> datetime: return datetime.now(UTC) # ── template + entity seeding ── async def ensure_seeded(db: AsyncSession) -> dict: """Idempotently seed the entity catalogue + 129 task templates. Re-resolves owners (email hints) and entity→legal-entity mappings on each call so a later HRIS / legal-entity sync fills previously-NULL links. Safe to call repeatedly. """ # Serialize every explicit seed/instantiate/roll-forward transaction on # the close-config singleton before reading either seed table. Without a # stable lock, two first writers both observe empty tables and the loser # hits the entity/template UNIQUE keys before the task-level idempotency # savepoints are reached (#2442). await close_service.lock_config_for_mutation(db) # 1. entity catalogue codes = loader.entity_codes() code_resolution = await coord.resolve_entity_codes(db, codes) existing_entities = { e.code: e for e in ( await db.execute(select(FinanceCloseEntity).with_for_update()) ).scalars().all() } for idx, code in enumerate(codes): res = code_resolution.get(code, {}) ent = existing_entities.get(code) if ent is None: db.add(FinanceCloseEntity( code=code, label=code, sort_order=idx, legal_entity_id=res.get("legal_entity_id"), legal_entity_name=res.get("legal_entity_name"), )) elif ent.legal_entity_id is None and res.get("legal_entity_id"): ent.legal_entity_id = res["legal_entity_id"] ent.legal_entity_name = res["legal_entity_name"] await db.flush() # 2. owner resolution (email-hint only) # Resolve owner labels to real people: email hints + finance-role name # match. Pass the full label set so name-matching can reach labels with # no email hint (Grace / Eric / Kevin Sun / …). all_labels = {r["owner_label"] for r in loader.template_rows() if r["owner_label"]} owner_ids = await coord.resolve_owner_person_ids( db, labels=all_labels, email_hints=loader.owner_email_hints(), ) # 3. templates (upsert by source_key) existing_tpl = { t.source_key: t for t in ( await db.execute(select(FinanceCloseTaskTemplate).with_for_update()) ).scalars().all() } created = 0 for row in loader.template_rows(): owner_pid = owner_ids.get(row["owner_label"]) tpl = existing_tpl.get(row["source_key"]) if tpl is None: db.add(FinanceCloseTaskTemplate( source_key=row["source_key"], owner_person_id=owner_pid, **{k: row[k] for k in _SHARED_FIELDS if k != "owner_person_id"}, )) created += 1 else: # Refresh the pure-definition fields from the seed, but PRESERVE # admin-owned fields on an existing template: owner (person id + # label) and signal_key are editable in the console, so a re-seed # (every instantiate / roll_forward calls ensure_seeded) must not # revert them. Owner id is only filled when newly resolvable and # not already set. _admin_owned = {"owner_person_id", "owner_label", "signal_key"} for k in _SHARED_FIELDS: if k in _admin_owned: continue setattr(tpl, k, row[k]) if owner_pid and not tpl.owner_person_id: tpl.owner_person_id = owner_pid await db.flush() # Backfill existing instances whose owner is still unresolved from their # now-resolved template (owner labels → real people land on prior periods # too, not just future roll-forwards). NULL-only, idempotent. New periods # already copy the resolved owner via _SHARED_FIELDS at instantiation. await db.execute( sa_text( "UPDATE finance_close_tasks t " "JOIN finance_close_task_templates tpl " " ON tpl.source_key = t.template_source_key " "SET t.owner_person_id = tpl.owner_person_id " "WHERE t.owner_person_id IS NULL AND tpl.owner_person_id IS NOT NULL" ) ) await db.flush() return { "templates": len(loader.template_rows()), "templates_created": created, "entities": len(codes), "owners_resolved": len(owner_ids), } async def list_templates( db: AsyncSession, *, for_update: bool = False, ) -> list[FinanceCloseTaskTemplate]: stmt = ( select(FinanceCloseTaskTemplate) .where(FinanceCloseTaskTemplate.is_active.is_(True)) .order_by(FinanceCloseTaskTemplate.sort_order, FinanceCloseTaskTemplate.id) ) if for_update: # Mutation callers may have waited behind the singleton seed lock # after establishing an older REPEATABLE READ snapshot. A locking # current-read is required to observe the winner's committed seed. stmt = stmt.with_for_update() return list((await db.execute(stmt)).scalars().all()) async def list_entities(db: AsyncSession) -> list[FinanceCloseEntity]: return list((await db.execute( select(FinanceCloseEntity) .where(FinanceCloseEntity.active.is_(True)) .order_by(FinanceCloseEntity.sort_order) )).scalars().all()) # ── deadline computation ── async def _window_and_deadline_fn(db: AsyncSession, period: str): """Return a ``calendar_day -> date|None`` resolver for ``period``.""" run = await close_service.ensure_run(db, period) window = await close_service.compute_window(db, run) return run, (lambda token: window.deadline_for(token)) async def recompute_deadlines(db: AsyncSession, period: str) -> int: """Recompute every task's deadline from the run's current anchors.""" _run, deadline_for = await _window_and_deadline_fn(db, period) tasks = await list_tasks(db, period) for t in tasks: t.deadline = deadline_for(t.calendar_day) await db.flush() return len(tasks) # ── instantiation + roll-forward ── def _initial_status(frequency: str) -> str: # "On going" tasks have no discrete start, so they begin in progress # (mirrors the source checklist's roll-forward behaviour). return TASK_IN_PROGRESS if frequency == FREQ_ONGOING else TASK_NOT_STARTED async def instantiate_from_templates( db: AsyncSession, period: str, *, actor_id: str | None = None ) -> dict: """Populate ``period`` with one task per active template (idempotent).""" close_service.validate_period(period) await ensure_seeded(db) _run, deadline_for = await _window_and_deadline_fn(db, period) existing_keys = await _existing_instance_keys(db, period) created = 0 for tpl in await list_templates(db, for_update=True): if tpl.source_key in existing_keys: continue if await _insert_task_idempotent(db, FinanceCloseTask( period=period, instance_key=tpl.source_key, template_source_key=tpl.source_key, status=_initial_status(tpl.frequency), deadline=deadline_for(tpl.calendar_day), created_by_id=actor_id, **{k: getattr(tpl, k) for k in _SHARED_FIELDS}, )): created += 1 return {"period": period, "created": created} async def roll_forward( db: AsyncSession, *, from_period: str, to_period: str, actor_id: str | None = None ) -> dict: """Clone ``from_period`` tasks into ``to_period`` (statuses reset). Carries owner reassignments and ad-hoc tasks forward (clones the live period, not the templates). Status resets to not-started (ongoing → in-progress); comments/completion are cleared; deadlines recomputed for the new period. Idempotent per (period, instance_key). Falls back to template instantiation if ``from_period`` has no tasks. """ close_service.validate_period(from_period) close_service.validate_period(to_period) src = await list_tasks(db, from_period) if not src: return await instantiate_from_templates(db, to_period, actor_id=actor_id) await close_service.ensure_run(db, to_period, actor_id=actor_id) _run, deadline_for = await _window_and_deadline_fn(db, to_period) existing_keys = await _existing_instance_keys(db, to_period) created = 0 for t in src: if t.instance_key in existing_keys: continue if await _insert_task_idempotent(db, FinanceCloseTask( period=to_period, instance_key=t.instance_key, template_source_key=t.template_source_key, status=_initial_status(t.frequency), deadline=deadline_for(t.calendar_day), created_by_id=actor_id, **{k: getattr(t, k) for k in _SHARED_FIELDS}, )): created += 1 return {"from_period": from_period, "to_period": to_period, "created": created} async def _insert_task_idempotent(db: AsyncSession, task: FinanceCloseTask) -> bool: """INSERT one task instance, tolerating a concurrent duplicate. The pre-check against ``_existing_instance_keys`` closes the common case, but two concurrent instantiate/roll_forward calls on the same period both see an empty key set and both INSERT — the loser trips ``uq_finance_close_task_period_key`` and, without a SAVEPOINT, poisons the async session so the router's later commit 500s instead of the documented idempotent no-op. A per-row ``begin_nested`` confines the IntegrityError to that write; a duplicate simply means another caller already created the row (skip, ``False``). Returns ``True`` when this call created the row. """ try: async with db.begin_nested(): db.add(task) await db.flush() except IntegrityError: return False return True async def _existing_instance_keys(db: AsyncSession, period: str) -> set[str]: rows = (await db.execute( select(FinanceCloseTask.instance_key) .where(FinanceCloseTask.period == period) .with_for_update() )).scalars().all() return set(rows) # ── task queries + aggregation ── async def list_tasks( db: AsyncSession, period: str, *, process: str | None = None, owner_person_id: str | None = None, status: str | None = None, calendar_day: str | None = None, q: str | None = None, ) -> list[FinanceCloseTask]: stmt = select(FinanceCloseTask).where(FinanceCloseTask.period == period) if process: stmt = stmt.where(FinanceCloseTask.process == process) if owner_person_id: stmt = stmt.where(FinanceCloseTask.owner_person_id == owner_person_id) if status: stmt = stmt.where(FinanceCloseTask.status == status) if calendar_day: stmt = stmt.where(FinanceCloseTask.calendar_day == calendar_day) if q: # Escape LIKE metacharacters so a literal '%' / '_' / '\' in the search # term matches literally instead of acting as a wildcard / scanning. esc = q.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_") stmt = stmt.where(FinanceCloseTask.description.like(f"%{esc}%", escape="\\")) stmt = stmt.order_by(FinanceCloseTask.sort_order, FinanceCloseTask.id) return list((await db.execute(stmt)).scalars().all()) async def get_task(db: AsyncSession, task_id: int) -> FinanceCloseTask: t = await db.get(FinanceCloseTask, task_id) if t is None: raise NotFoundError("close task not found") return t async def _assert_period_not_frozen(db: AsyncSession, period: str) -> None: """Reject task-list mutations once the owning run is soft/hard-closed. The freeze invariant used to protect only the run's close *date* / anchors (``close_service.update_anchors`` / ``set_override``), leaving the checklist itself editable after close: task status, comments, owner reassignment, ad-hoc create, and delete all still landed on a ``soft_closed`` / ``closed`` period, so ``aggregate().completed_pct`` could silently move post-close and weaken the close audit trail. Gate every task mutation here — both the admin (:mod:`close_task_router`) and self-service (:mod:`close_self_service`) paths funnel through the same service functions, so one guard covers both. A run with no row yet (never instantiated) is treated as open. """ run = await close_service.get_run(db, period) if run is not None and run.status in RUN_FROZEN_STATUSES: raise ValidationError( f"period {period} is {run.status}; reopen the close run before " "editing its checklist" ) async def aggregate(db: AsyncSession, period: str) -> dict: """Overall + per-owner + per-process status counts for the period.""" rows = (await db.execute( select( FinanceCloseTask.status, FinanceCloseTask.process, FinanceCloseTask.owner_person_id, FinanceCloseTask.owner_label, func.count().label("n"), ).where(FinanceCloseTask.period == period) .group_by( FinanceCloseTask.status, FinanceCloseTask.process, FinanceCloseTask.owner_person_id, FinanceCloseTask.owner_label, ) )).all() def _blank() -> dict: return {s: 0 for s in TASK_STATUSES} | {"total": 0} overall = _blank() by_process: dict[str, dict] = {} by_owner: dict[str, dict] = {} for status, process, owner_id, owner_label, n in rows: for bucket in ( overall, by_process.setdefault(process, _blank() | {"process": process}), by_owner.setdefault( owner_id or f"label:{owner_label}", _blank() | {"owner_person_id": owner_id, "owner_label": owner_label}, ), ): bucket[status] += n bucket["total"] += n def _pct(b: dict) -> int: return round(100 * b[TASK_COMPLETED] / b["total"]) if b["total"] else 0 overall["completed_pct"] = _pct(overall) for b in by_process.values(): b["completed_pct"] = _pct(b) for b in by_owner.values(): b["completed_pct"] = _pct(b) return { "overall": overall, "by_process": sorted(by_process.values(), key=lambda b: b["process"]), "by_owner": sorted(by_owner.values(), key=lambda b: (b["owner_label"] or "")), } # ── task mutations ── def _validate_enums(*, frequency=None, ai_work=None, status=None, calendar_day=None): if frequency is not None and frequency not in FREQUENCIES: raise ValidationError(f"frequency must be one of {FREQUENCIES}") if ai_work is not None and ai_work not in AI_WORK_VALUES: raise ValidationError(f"ai_work must be one of {AI_WORK_VALUES}") if status is not None and status not in TASK_STATUSES: raise ValidationError(f"status must be one of {TASK_STATUSES}") if calendar_day is not None: from app.domains.finance.close_calendar import _DAY_TOKEN_RE, ONGOING token = calendar_day.strip() m = _DAY_TOKEN_RE.match(token) # Reject Day+0 / Day-0: the regex admits them but resolve_anchor returns # None for N<1, which would silently leave the task with no deadline. if token != ONGOING and (m is None or int(m.group(2)) < 1): raise ValidationError("calendar_day must be 'On going' or 'Day±N' (N ≥ 1)") async def create_task(db: AsyncSession, period: str, *, fields: dict, actor_id: str | None) -> FinanceCloseTask: close_service.validate_period(period) await _assert_period_not_frozen(db, period) _validate_enums( frequency=fields.get("frequency"), ai_work=fields.get("ai_work"), calendar_day=fields.get("calendar_day"), ) _run, deadline_for = await _window_and_deadline_fn(db, period) instance_key = f"adhoc-{uuid.uuid4().hex[:16]}" task = FinanceCloseTask( period=period, instance_key=instance_key, template_source_key=None, status=fields.get("status", TASK_NOT_STARTED), deadline=deadline_for(fields.get("calendar_day", "On going")), created_by_id=actor_id, process=fields["process"], sub_process=fields.get("sub_process", ""), description=fields["description"], owner_person_id=fields.get("owner_person_id"), owner_label=fields.get("owner_label"), frequency=fields.get("frequency", "monthly"), calendar_day=fields.get("calendar_day", "On going"), estimated_hours=fields.get("estimated_hours"), submit_to_fpa=bool(fields.get("submit_to_fpa", False)), ai_work=fields.get("ai_work", "none"), link=fields.get("link"), entity_codes=fields.get("entity_codes"), sort_order=fields.get("sort_order", 9999), ) db.add(task) await db.flush() return task async def update_task(db: AsyncSession, task_id: int, *, fields: dict) -> FinanceCloseTask: task = await get_task(db, task_id) await _assert_period_not_frozen(db, task.period) _validate_enums( frequency=fields.get("frequency"), ai_work=fields.get("ai_work"), calendar_day=fields.get("calendar_day"), ) for k in ( "process", "sub_process", "description", "owner_person_id", "owner_label", "frequency", "calendar_day", "estimated_hours", "submit_to_fpa", "ai_work", "link", "entity_codes", ): if k in fields: setattr(task, k, fields[k]) if "calendar_day" in fields: _run, deadline_for = await _window_and_deadline_fn(db, task.period) task.deadline = deadline_for(task.calendar_day) await db.flush() return task async def set_status( db: AsyncSession, task_id: int, *, status: str, actor_id: str | None ) -> FinanceCloseTask: _validate_enums(status=status) task = await get_task(db, task_id) await _assert_period_not_frozen(db, task.period) task.status = status if status == TASK_COMPLETED: task.completed_at = _now() task.completed_by_id = actor_id else: task.completed_at = None task.completed_by_id = None # First real activity bumps the run open → in_progress. if status not in (TASK_NOT_STARTED,): run = await close_service.get_run(db, task.period) if run is not None: await close_service.mark_in_progress(db, run) await db.flush() return task async def set_comment( db: AsyncSession, task_id: int, *, comment: str | None, commenter_person_id: str | None, commenter_label: str | None, ) -> FinanceCloseTask: task = await get_task(db, task_id) await _assert_period_not_frozen(db, task.period) task.comment = comment task.commenter_person_id = commenter_person_id task.commenter_label = commenter_label await db.flush() return task async def delete_task(db: AsyncSession, task_id: int) -> None: task = await get_task(db, task_id) await _assert_period_not_frozen(db, task.period) await db.delete(task) await db.flush()