"""Tips engine. Architecture: - `Tip` is a plain dataclass returned to the router. - `RULE_REGISTRY` is a list of async functions; each takes a `TipsContext` and returns a `Tip | None`. None means the rule doesn't apply right now. - `TipsService.get_tips()` runs all rules and returns the non-None results. All 12 rules from ARCHITECTURE.md §6.1 are registered; the stubs return None until implemented. This lets the junior wire rules one at a time by editing the corresponding function without touching the plumbing. To add a new rule: 1. Write an async function `_rule_(ctx: TipsContext) -> Tip | None`. 2. Append it to RULE_REGISTRY. """ from __future__ import annotations import asyncio from collections import defaultdict from dataclasses import dataclass from datetime import date, timedelta from typing import Callable, Coroutine, TypeAlias from uuid import UUID, uuid4 from dateutil.relativedelta import relativedelta as _relativedelta from app.config import settings from app.entities.child import Child from app.entities.leave_wish import LeaveWish from app.entities.parental_leave import ParentalLeaveFetch from app.entities.personal_data import PersonalData from app.entities.planned_leave import PlannedLeaveDay from app.entities.tip_dismissal import TipDismissal from app.entities.vacation_days import VacationDays from app.repositories.child_repository import ChildRepository from app.repositories.leave_wish_repository import LeaveWishRepository from app.repositories.parental_leave_repository import ParentalLeaveRepository from app.repositories.personal_data_repository import PersonalDataRepository from app.repositories.planned_leave_repository import PlannedLeaveRepository from app.repositories.tip_dismissal_repository import TipDismissalRepository from app.repositories.vacation_days_repository import VacationDaysRepository from app.utils.errors import NotFoundError from app.utils.tip_predicates import ( age4_pooling_loss, days_left_at_age_4_projection, has_vacation_near_expiry, hash_tip_data, is_child_over_age, preschool_buffer_needed, red_day_clusters_in_range, resolve_age4_savable_cap, sgi_at_risk, weekly_intensity, ) @dataclass class Tip: id: str severity: str # 'info' | 'warning' | 'action' message_key: str # i18n key for frontend data: dict # context for frontend rendering (e.g. {days: 5, period: "Jul–Aug"}) # Where this tip is anchored on a calendar: a single day (e.g. "use this # vacation day"), a span (e.g. a break period), or neither. Lifted out of # `data` — where the value already lived — rather than invented; see the # individual rules for which `data` key each maps to. Rules with no # meaningful date (balance_between_parents, lagsta_niva_coverage, # stretch_time_at_home, sgi_minimum_after_age_1, # reserve_for_preschool_planning_days) leave both null. anchor_date: date | None = None anchor_range: tuple[date, date] | None = None @dataclass class TipsContext: workspace_id: UUID # Either empty (the workspace-level context — see `_WORKSPACE_SCOPED_RULES` # in `TipsService`) or a single child (a per-child context, built once per # child in `_build_context`). Never more than one: entitlement and # planned days are tracked per child, so a rule scoped to "the" child # would silently pick an arbitrary one in a multi-child workspace. children: list[Child] parents: list[PersonalData] leave_by_parent: dict[UUID, ParentalLeaveFetch | None] # personal_data_uuid → latest # Future planned days for `children[0]` only (empty for the workspace # context) — see `_build_context`. planned_days: list[PlannedLeaveDay] vacation_by_parent: dict[UUID, list[VacationDays]] leave_wishes: list[LeaveWish] today: date # injected for deterministic testing # Type alias for a rule function. RuleFn: TypeAlias = Callable[[TipsContext], Coroutine[None, None, Tip | None]] # --------------------------------------------------------------------------- # Rule implementations (stub → return None until fully wired) # --------------------------------------------------------------------------- async def _rule_use_vacation_before_expiry(ctx: TipsContext) -> Tip | None: for parent in ctx.parents: for vd in ctx.vacation_by_parent.get(parent.uuid, []): if has_vacation_near_expiry(vd.total_days, vd.saved_days_in): # `vd.year` is the only date-shaped fact this rule has; anchor # to that vacation year's last day — the natural forcing point # for "use it before it's gone" — rather than leaving the rail # with nothing to place this tip on. year_end = date(vd.year, 12, 31) return Tip( id="use_vacation_before_expiry", severity="warning", message_key="tips.use_vacation_before_expiry", data={ "parent_name": parent.name, "saved_days": vd.saved_days_in, "year": vd.year, }, anchor_date=year_end, ) return None async def _rule_balance_between_parents(ctx: TipsContext) -> Tip | None: if len(ctx.parents) < 2: return None usages = [ (p, ctx.leave_by_parent[p.uuid].days_used) for p in ctx.parents if ctx.leave_by_parent.get(p.uuid) is not None ] if len(usages) < 2: return None leading, leading_used = max(usages, key=lambda x: x[1]) lagging, lagging_used = min(usages, key=lambda x: x[1]) gap = leading_used - lagging_used if gap > 30: return Tip( id="balance_between_parents", severity="info", message_key="tips.balance_between_parents", data={ "leading_parent": leading.name, "lagging_parent": lagging.name, "gap_days": gap, }, ) return None async def _rule_preserve_for_age_4(ctx: TipsContext) -> Tip | None: if not ctx.children: return None child = ctx.children[0] fourth_bday = child.birth_date + _relativedelta(years=4) savable_cap = resolve_age4_savable_cap(len(ctx.parents)) for parent in ctx.parents: leave = ctx.leave_by_parent.get(parent.uuid) if leave is None: continue planned_fp_before_b4 = sum( 1 for d in ctx.planned_days if d.parent_personal_data_uuid == parent.uuid and d.day_type in ("sjukpenning", "lagsta") and d.date < fourth_bday ) _, expiring_at_b4 = days_left_at_age_4_projection( leave.days_used, planned_fp_before_b4, leave.total_days, savable_cap ) if expiring_at_b4 > 0: return Tip( id="preserve_for_age_4", severity="warning", message_key="tips.preserve_for_age_4", data={ "parent_name": parent.name, "expiring_days": expiring_at_b4, "fourth_birthday": fourth_bday.isoformat(), }, anchor_date=fourth_bday, ) return None async def _rule_age4_pooling_loss(ctx: TipsContext) -> Tip | None: """Warn a family when one parent is over the per-parent age-4 cap while another has unused headroom — days that could have been pooled but won't be, because `resolve_age4_savable_cap` splits the ceiling per parent as soon as the child has 2+ parents (see `age4_pooling_loss`). This fires for any 2+-parent family — the per-parent cut (and the pooling loss it can cause) is a legal fact about the child having two parents, purely a function of parent count. Advice order is the substance of this tip: taking more leave before the birthday converts entitlement into actual time at home (strictly better), so it is offered first. Transferring days at Försäkringskassan only preserves the entitlement — it is offered second, and the transfer figure is bounded by what `ParentalLeaveFetch.transferable_days_left` (not the raw days-left total, which includes non-transferable reserved days) says is actually movable. When the partner has no FK snapshot the exact transfer figure can't be known, so the tip degrades to a hedged "if you both have days left" phrasing instead of guessing a number. """ if len(ctx.parents) < 2 or not ctx.children: return None child = ctx.children[0] if not is_child_over_age(child.birth_date, settings.rules.age4_pooling_tip_min_child_age_years): return None # first couple of years: too early to usefully act on this fourth_bday = child.birth_date + _relativedelta(years=4) per_parent_cap = resolve_age4_savable_cap(len(ctx.parents)) per_child_cap = settings.rules.max_savable_days_at_age_4 projected_left_by_parent: dict[UUID, int] = {} excess_by_parent: dict[UUID, int] = {} for parent in ctx.parents: leave = ctx.leave_by_parent.get(parent.uuid) if leave is None: continue planned_fp_before_b4 = sum( 1 for d in ctx.planned_days if d.parent_personal_data_uuid == parent.uuid and d.day_type in ("sjukpenning", "lagsta") and d.date < fourth_bday ) projected_left, expiring = days_left_at_age_4_projection( leave.days_used, planned_fp_before_b4, leave.total_days, per_parent_cap ) projected_left_by_parent[parent.uuid] = projected_left excess_by_parent[parent.uuid] = expiring if not projected_left_by_parent: return None family_loss = age4_pooling_loss( list(projected_left_by_parent.values()), per_parent_cap, per_child_cap ) if family_loss <= 0: return None surplus_uuid = max(excess_by_parent, key=lambda u: excess_by_parent[u]) surplus_parent = next(p for p in ctx.parents if p.uuid == surplus_uuid) surplus_leave = ctx.leave_by_parent[surplus_uuid] excess_days = excess_by_parent[surplus_uuid] other_parents = [p for p in ctx.parents if p.uuid != surplus_uuid] partner_has_snapshot = False lagging_parent: PersonalData | None = None lagging_headroom = 0 for other in other_parents: other_leave = ctx.leave_by_parent.get(other.uuid) if other_leave is None: continue partner_has_snapshot = True headroom = max(per_parent_cap - projected_left_by_parent[other.uuid], 0) if headroom > lagging_headroom: lagging_headroom = headroom lagging_parent = other if lagging_parent is not None and lagging_headroom > 0: transfer_days = min(excess_days, lagging_headroom, surplus_leave.transferable_days_left) transfer_days = max(transfer_days, 0) return Tip( id="age4_pooling_loss", severity="warning", message_key="tips.age4_pooling_loss_exact", data={ "surplus_parent": surplus_parent.name, "lagging_parent": lagging_parent.name, "excess_days": excess_days, "transfer_days": transfer_days, "family_loss_days": family_loss, "parent_cap": per_parent_cap, "fourth_birthday": fourth_bday.isoformat(), }, ) if not partner_has_snapshot and other_parents: return Tip( id="age4_pooling_loss", severity="warning", message_key="tips.age4_pooling_loss_hedged", data={ "surplus_parent": surplus_parent.name, "excess_days": excess_days, "family_loss_days": family_loss, "parent_cap": per_parent_cap, "fourth_birthday": fourth_bday.isoformat(), }, ) return None async def _rule_use_dubbeldagar_strategically(ctx: TipsContext) -> Tip | None: if len(ctx.parents) < 2 or not ctx.children: return None child = ctx.children[0] window_end = child.birth_date + _relativedelta(months=settings.rules.dubbeldagar_months_limit) # Map date → set of parent UUIDs with FP planned that day (within dubbeldagar window) fp_by_date: dict[date, set[UUID]] = defaultdict(set) for d in ctx.planned_days: if d.day_type in ("sjukpenning", "lagsta") and d.date <= window_end: fp_by_date[d.date].add(d.parent_personal_data_uuid) for day_date, parent_uuids in fp_by_date.items(): if len(parent_uuids) < 2: continue already_marked = any( d.is_dubbeldag for d in ctx.planned_days if d.date == day_date and d.parent_personal_data_uuid in parent_uuids ) if not already_marked: return Tip( id="use_dubbeldagar_strategically", severity="action", message_key="tips.use_dubbeldagar_strategically", data={"date": day_date.isoformat()}, anchor_date=day_date, ) return None async def _rule_lagsta_niva_coverage(ctx: TipsContext) -> Tip | None: for parent in ctx.parents: leave = ctx.leave_by_parent.get(parent.uuid) if leave is None: continue tier = leave.tier_data if tier.sjukpenning_left < 30 and tier.lagsta_niva_left > 0: return Tip( id="lagsta_niva_coverage", severity="info", message_key="tips.lagsta_niva_coverage", data={ "parent_name": parent.name, "sjukpenning_left": tier.sjukpenning_left, "lagsta_left": tier.lagsta_niva_left, }, ) return None async def _rule_stretch_time_at_home(ctx: TipsContext) -> Tip | None: if not ctx.children: return None child = ctx.children[0] fourth_bday = child.birth_date + _relativedelta(years=4) today = ctx.today window_end = today + timedelta(weeks=4) savable_cap = resolve_age4_savable_cap(len(ctx.parents)) for parent in ctx.parents: parent_window_days = [ d for d in ctx.planned_days if d.parent_personal_data_uuid == parent.uuid and today <= d.date <= window_end ] intensity = weekly_intensity(parent_window_days, 4.0) if intensity < 7: continue leave = ctx.leave_by_parent.get(parent.uuid) if leave is None: continue planned_fp_before_b4 = sum( 1 for d in ctx.planned_days if d.parent_personal_data_uuid == parent.uuid and d.day_type in ("sjukpenning", "lagsta") and d.date < fourth_bday ) projected_left, _ = days_left_at_age_4_projection( leave.days_used, planned_fp_before_b4, leave.total_days, savable_cap ) if projected_left < 60: return Tip( id="stretch_time_at_home", severity="info", message_key="tips.stretch_time_at_home", data={ "parent_name": parent.name, "weekly_intensity": round(intensity, 1), "projected_left": projected_left, }, ) return None async def _rule_sgi_minimum_after_age_1(ctx: TipsContext) -> Tip | None: today = ctx.today for child in ctx.children: if not is_child_over_age(child.birth_date, 1): continue period_end = today + timedelta(weeks=4) for parent in ctx.parents: if sgi_at_risk(parent, ctx.planned_days, child.birth_date, today, period_end): return Tip( id="sgi_minimum_after_age_1", severity="warning", message_key="tips.sgi_minimum_after_age_1", data={ "parent_name": parent.name, "min_days_per_week": settings.rules.sgi_min_days_per_week_after_age_1, }, ) return None async def _rule_wrap_weekend_with_vacation(ctx: TipsContext) -> Tip | None: today = ctx.today look_ahead = today + timedelta(weeks=8) for parent in ctx.parents: if parent.salary_type != "monthly": continue fp_dates = { d.date for d in ctx.planned_days if d.parent_personal_data_uuid == parent.uuid and d.day_type in ("sjukpenning", "lagsta") and today <= d.date <= look_ahead } for fp_date in sorted(fp_dates): if fp_date.weekday() == 4: # Friday monday = fp_date + timedelta(days=3) if monday in fp_dates: return Tip( id="wrap_weekend_with_vacation", severity="info", message_key="tips.wrap_weekend_with_vacation", data={ "parent_name": parent.name, "friday": fp_date.isoformat(), "monday": monday.isoformat(), }, anchor_range=(fp_date, monday), ) return None async def _rule_wrap_red_days_with_vacation(ctx: TipsContext) -> Tip | None: today = ctx.today look_ahead = today + timedelta(weeks=12) clusters = red_day_clusters_in_range(today, look_ahead) if not clusters: return None for parent in ctx.parents: if parent.salary_type != "monthly": continue fp_dates = { d.date for d in ctx.planned_days if d.parent_personal_data_uuid == parent.uuid and d.day_type in ("sjukpenning", "lagsta") and today <= d.date <= look_ahead } for cluster_start, cluster_end in clusters: day_before = cluster_start - timedelta(days=1) day_after = cluster_end + timedelta(days=1) if day_before in fp_dates or day_after in fp_dates: return Tip( id="wrap_red_days_with_vacation", severity="info", message_key="tips.wrap_red_days_with_vacation", data={ "parent_name": parent.name, "cluster_start": cluster_start.isoformat(), "cluster_end": cluster_end.isoformat(), }, anchor_range=(cluster_start, cluster_end), ) return None def _concrete_windows_near_today(wish: LeaveWish, today: date) -> list[tuple[date, date]]: """This wish's window, if it is anchored near `today` — else nothing. One wish row is one year (see `app.entities.leave_wish`), so there is nothing to concretize any more; the tips engine just ignores occurrences far from today. The neighbourhood is last year / this year / next year — "last year" is what catches a window that started last December and wraps into January. Reasoning over the whole horizon is the allocator's job (PLANNING_ENGINE_SPEC.md §3.2), not the tips engine's. """ if wish.first_year not in (today.year - 1, today.year, today.year + 1): return [] return [wish.window()] async def _rule_long_break_parent_swap(ctx: TipsContext) -> Tip | None: if len(ctx.parents) < 2 or not ctx.leave_wishes: return None today = ctx.today for wish in ctx.leave_wishes: for start_date, end_date in _concrete_windows_near_today(wish, today): if end_date < today: continue working_days = sum( 1 for i in range((end_date - start_date).days + 1) if (start_date + timedelta(days=i)).weekday() < 5 ) if working_days == 0: continue for parent in ctx.parents: covered = sum( 1 for d in ctx.planned_days if d.parent_personal_data_uuid == parent.uuid and start_date <= d.date <= end_date ) if covered < working_days * 0.5: other = next(p for p in ctx.parents if p.uuid != parent.uuid) return Tip( id="long_break_parent_swap", severity="info", message_key="tips.long_break_parent_swap", data={ "break_name": wish.name, "start_date": start_date.isoformat(), "end_date": end_date.isoformat(), "parent_name": parent.name, "other_parent_name": other.name, }, anchor_range=(start_date, end_date), ) return None async def _rule_reserve_for_preschool_planning_days(ctx: TipsContext) -> Tip | None: if not ctx.children: return None child = ctx.children[0] buffer_needed = preschool_buffer_needed(child, settings.rules.preschool_planning_days_per_year) if buffer_needed == 0: return None for parent in ctx.parents: leave = ctx.leave_by_parent.get(parent.uuid) if leave is None: continue if leave.days_left < buffer_needed * 2: return Tip( id="reserve_for_preschool_planning_days", severity="info", message_key="tips.reserve_for_preschool_planning_days", data={ "parent_name": parent.name, "buffer_days": buffer_needed, "days_left": leave.days_left, }, ) return None # --------------------------------------------------------------------------- # Registry — order matters (earlier rules shown first) # --------------------------------------------------------------------------- RULE_REGISTRY: list[RuleFn] = [ _rule_use_vacation_before_expiry, _rule_balance_between_parents, _rule_preserve_for_age_4, _rule_age4_pooling_loss, _rule_use_dubbeldagar_strategically, _rule_lagsta_niva_coverage, _rule_stretch_time_at_home, _rule_sgi_minimum_after_age_1, _rule_wrap_weekend_with_vacation, _rule_wrap_red_days_with_vacation, _rule_long_break_parent_swap, _rule_reserve_for_preschool_planning_days, ] # Parental-leave entitlement (`parental_leave_fetch`), and every planned day # that draws on it, is tracked per CHILD (see `ParentalLeaveRepository. # get_latest_by_parent` / `PlannedLeaveRepository.list_by_workspace_child`). # `TipsService.get_tips` therefore runs most rules once per child, against a # context scoped to that child's own leave/planned-day data (see # `_build_context`). `use_vacation_before_expiry` is the one rule that reads # neither — vacation days belong to the parent, not any child — so running it # per child would just print the same warning N times for an N-child # workspace; it runs once instead, against a childless workspace-level # context. _WORKSPACE_SCOPED_RULES: frozenset[RuleFn] = frozenset({_rule_use_vacation_before_expiry}) class TipsService: def __init__( self, leave_repo: ParentalLeaveRepository, planned_repo: PlannedLeaveRepository, child_repo: ChildRepository, pd_repo: PersonalDataRepository, vacation_repo: VacationDaysRepository, leave_wish_repo: LeaveWishRepository, dismissal_repo: TipDismissalRepository, ) -> None: self._leave = leave_repo self._planned = planned_repo self._children = child_repo self._parents = pd_repo self._vacation = vacation_repo self._leave_wishes = leave_wish_repo self._dismissals = dismissal_repo async def get_tips(self, workspace_id: UUID) -> list[Tip]: """Load workspace context(s), run all rules, drop stale-dismissed tips. One workspace-level context feeds `_WORKSPACE_SCOPED_RULES`; one per-child context feeds every other rule (see `_build_context` and `_WORKSPACE_SCOPED_RULES` for why). A child-scoped tip's `id` is suffixed with its child's id so that e.g. `preserve_for_age_4` firing for two different children never collides — neither in the returned list nor in `TipDismissal` (dismissing child A's warning must not dismiss child B's). """ workspace_ctx, child_contexts = await self._build_context(workspace_id) tips: list[Tip] = [] workspace_rules = [rule for rule in RULE_REGISTRY if rule in _WORKSPACE_SCOPED_RULES] child_rules = [rule for rule in RULE_REGISTRY if rule not in _WORKSPACE_SCOPED_RULES] workspace_results = await asyncio.gather(*(rule(workspace_ctx) for rule in workspace_rules)) tips.extend(tip for tip in workspace_results if tip is not None) for ctx in child_contexts: child = ctx.children[0] results = await asyncio.gather(*(rule(ctx) for rule in child_rules)) for tip in results: if tip is None: continue tip.data = {**tip.data, "child_id": str(child.id), "child_name": child.name} tip.id = f"{tip.id}:{child.id}" tips.append(tip) return await self._drop_stale_dismissed(workspace_id, tips) async def dismiss(self, workspace_id: UUID, tip_id: str) -> None: """Record that `tip_id` is dismissed, fingerprinted to its current `data`. Recomputes the live tip list (rather than trusting a client-supplied `data` payload) so the stored fingerprint is always the engine's own truth. Raises NotFoundError if `tip_id` doesn't match any currently-firing tip — there is nothing meaningful to dismiss. """ tips = await self.get_tips(workspace_id) tip = next((t for t in tips if t.id == tip_id), None) if tip is None: raise NotFoundError(f"Tip {tip_id!r} is not currently active for this workspace") data_hash = hash_tip_data(tip.data) existing = await self._dismissals.get_by_workspace_tip(workspace_id, tip_id) if existing is not None: existing.data_hash = data_hash await self._dismissals.update(existing.id, existing) return await self._dismissals.create( TipDismissal( id=uuid4(), workspace_id=workspace_id, tip_id=tip_id, data_hash=data_hash, ) ) async def _drop_stale_dismissed(self, workspace_id: UUID, tips: list[Tip]) -> list[Tip]: """Filter out tips whose dismissal is still fresh. A dismissal only silences the *exact* situation the user acknowledged (same tip id, same `data`). If the rule now fires with different data — a bigger `expiring_days` count, a moved `fourth_birthday`, ... — the old dismissal describes a fact that's no longer true, so the tip resurfaces instead of staying silenced forever. See module docstring on `app.entities.tip_dismissal.TipDismissal` for why an unconditional (non-expiring) dismissal would be dangerous for warnings like `preserve_for_age_4`. """ dismissals = await self._dismissals.list_by_workspace(workspace_id) if not dismissals: return tips dismissed_hash_by_id = {d.tip_id: d.data_hash for d in dismissals} return [tip for tip in tips if dismissed_hash_by_id.get(tip.id) != hash_tip_data(tip.data)] async def _build_context(self, workspace_id: UUID) -> tuple[TipsContext, list[TipsContext]]: """Build the workspace-level context plus one context per child. Two bugs used to live here: 1. **Single-child only.** Only `children[0]` was ever fetched, so a workspace with two children got advice about one and silence about the other — despite `list_by_workspace` already returning every child. Now every child gets its own context (see the loop below). 2. **Doubled planned days.** `list_by_workspace_child` already returns every parent's days for a given child; the old code called it again *inside* the per-parent loop and kept appending to the same list, so an N-parent workspace saw each planned day N times. Every count-sensitive rule (intensity, SGI, age-4 projections) was silently working from doubled input. Fetched once per child now, outside any per-parent loop. """ parents = await self._parents.list_by_workspace(workspace_id) children = await self._children.list_by_workspace(workspace_id) leave_wishes = await self._leave_wishes.get_by_workspace(workspace_id) today = date.today() # Vacation days belong to the parent, not any child — fetched once, # shared by the workspace context and every per-child context. vacation_by_parent: dict[UUID, list[VacationDays]] = {} for pd in parents: vacation_by_parent[pd.uuid] = await self._vacation.list( personal_data_uuid=pd.uuid, workspace_id=workspace_id ) workspace_ctx = TipsContext( workspace_id=workspace_id, children=[], parents=parents, leave_by_parent={}, planned_days=[], vacation_by_parent=vacation_by_parent, leave_wishes=leave_wishes, today=today, ) child_contexts: list[TipsContext] = [] for child in children: leave_by_parent: dict[UUID, ParentalLeaveFetch | None] = {} for pd in parents: leave_by_parent[pd.uuid] = await self._leave.get_latest_by_parent(pd.uuid, child.id) planned_days = await self._planned.list_by_workspace_child( workspace_id, child.id, from_date=today ) child_contexts.append( TipsContext( workspace_id=workspace_id, children=[child], parents=parents, leave_by_parent=leave_by_parent, planned_days=planned_days, vacation_by_parent=vacation_by_parent, leave_wishes=leave_wishes, today=today, ) ) return workspace_ctx, child_contexts