"""Planning engine allocator. See PLANNING_ENGINE_SPEC.md §3 for the full design. Split strictly in two, and the split is the single most important property of this module: - `AllocationInputLoader` — async. The ONLY part of the planning engine that touches a repository or the wall clock. Assembles a frozen `AllocationInput`. - `Allocator` — PURE. No `await`, no repository access, and above all no `date.today()`. `today` arrives on `AllocationInput.today`. Same input must always produce byte-identical output (§5 test #11) — this is what makes the engine explainable, cheap to re-run on every edit, and unit-testable without a database. `PlanningEngineService` is a thin orchestrator gluing the two together for the router (`POST /api/plan/propose`). """ from __future__ import annotations from dataclasses import dataclass, field from datetime import date, timedelta from uuid import UUID from dateutil.relativedelta import relativedelta from app.config import settings from app.entities.child import Child from app.entities.day_assertion import partition_assertion_keys from app.entities.leave_wish import LeaveWish from app.entities.planned_leave import PlannedLeaveDay from app.repositories.child_repository import ChildRepository from app.repositories.day_assertion_repository import DayAssertionRepository from app.repositories.day_override_repository import DayOverrideRepository 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.preschool_planning_days_repository import PreschoolPlanningDaysRepository from app.repositories.vacation_days_repository import VacationDaysRepository from app.repositories.workspace_repository import WorkspaceRepository # §Task 2 (vacation-forfeiture bias) — `AllocationInputLoader` reuses # `AnalyticsService._vacation_year_series` (the FIFO carry-forward bank # simulation behind the Analytics tab's balance series) verbatim rather than # re-deriving it, so the loader's forecast of what vacation would be lost # this year can never quietly drift from what the dashboard reports for the # same (parent, year). `AnalyticsService` does not import anything from this # module, so this stays a one-way dependency. from app.services.analytics_service import AnalyticsService from app.utils.errors import NotFoundError, ValidationError from app.utils.holidays import ( SUNDAY_HOLIDAY_NAME, UncachedYearError, cached_year_bounds, get_swedish_holidays, ) from app.utils.tip_predicates import ( days_left_at_age_4_projection, default_tier_pool_share, resolve_age4_savable_cap, ) # --------------------------------------------------------------------------- # Reason codes (§3.5) — machine-readable so the UI can explain itself. # --------------------------------------------------------------------------- REASON_FREE_WEEKEND = "free_weekend" REASON_FREE_RED_DAY = "free_red_day" REASON_VACATION_WRAP_RED_DAY = "vacation_wrap_red_day" REASON_VACATION_WRAP_WEEKEND = "vacation_wrap_weekend" REASON_FP_INTERIOR = "fp_interior" REASON_FP_LAGSTA_STRETCH = "fp_lagsta_stretch" REASON_SWAP_FIRST_HALF = "swap_first_half" REASON_SWAP_SECOND_HALF = "swap_second_half" REASON_DUBBELDAG = "dubbeldag" # §todo item 2 — a free day (Saturday, Sunday or public holiday) funded with a # parental-benefit day instead of being left free, opted into per-workspace via # `AllocationInput.take_pl_on_red_days` (default False — see `_fund_solo`/ # `_fund_both`). Vacation is deliberately never used here: Swedish law does not # let you "take vacation" on a day that already isn't a working day, but FP is # a benefit payment, not a scheduling construct, so it can still be drawn on a # red day to replace the (unpaid) day off with a paid one — at the cost of # spending a day the household would otherwise have kept. REASON_FP_RED_DAY = "fp_red_day" # --------------------------------------------------------------------------- # Surplus advice kinds (§9.2) — the mirror of a shortfall remedy: emitted # when a parent is forecast to *forfeit* days at the age-4 boundary rather # than run short. # --------------------------------------------------------------------------- SURPLUS_WISH_MORE_LEAVE = "wish_more_leave" SURPLUS_SHIFT_MIX_TO_FP = "shift_mix_to_fp" SURPLUS_FRONT_LOAD_EARLY_YEARS = "front_load_early_years" SURPLUS_USE_LESS_VACATION = "use_less_vacation" # --------------------------------------------------------------------------- # Shortfall remedy kinds (§3.6, §9.1) and plan warning kinds — named so their # `message_key` (below) is derived rather than hand-typed a second time. # --------------------------------------------------------------------------- REMEDY_REDUCE_PRESCHOOL_RESERVE = "reduce_preschool_reserve" REMEDY_CARRY_VACATION_FORWARD = "carry_vacation_forward" REMEDY_LOWER_PRIORITY_WISH = "lower_priority_wish" REMEDY_SHIFT_WINDOW_INTO_FREE_DAYS = "shift_window_into_free_days" WARNING_DAYS_LOST_AT_AGE_4 = "days_lost_at_age_4" WARNING_SGI_RISK = "sgi_risk" WARNING_VACATION_EXPIRING = "vacation_expiring" # --------------------------------------------------------------------------- # §10.4 — flexible ("när det passar bäst") wish outcome strategy. Added to # the existing `WindowOutcome.strategy` value set ('solo' | 'parent_swap' | # 'dubbeldag') — a bare str on both the dataclass and the wire model, so a # new literal value is a behavior change, not a schema change. # # §Decision A/C (2026-09-18 co-leave fix) extended that set again: # 'co_leave' — a `parent_mode == 'both'` window funded as PL (to whichever # parent has the most FP left) + vacation (the other parent) on the SAME # date, the new default for 'both' whenever dubbeldagar are not requested # or not legally available. 'swap' — the true last resort within a 'both' # window: a date neither co-leave resource could cover, handed to the same # sequential-halves machinery 'parent_swap' uses. `strategy` always reports # what actually happened, never what was merely hoped for; `requested_strategy` # carries the latter, and `strategy_degraded` is true whenever they differ # (see `WindowOutcome` below). # --------------------------------------------------------------------------- STRATEGY_UNPLACED = "unplaced" STRATEGY_CO_LEAVE = "co_leave" STRATEGY_SWAP = "swap" # §3.5 — i18n message keys. Every UI-facing piece of text the (pure) allocator # used to render as a full Swedish sentence is now a `message_key` + `data` # pair instead (the pattern `tips_service.py` already uses for `Tip`), so the # frontend can translate it in the caller's active locale rather than # rendering backend-generated text verbatim regardless of locale. See # `frontend/src/i18n/sv.ts` / `en.ts` for the corresponding copy and # `frontend/src/components/planning/PlanResults.tsx` for the lookup tables # that map these opaque keys to real `TranslationKey`s (same shape as # `TipCard.tsx`'s `TIP_TRANSLATION_KEYS`). MESSAGE_KEY_WINDOW_EXPLANATION = "planOutcomes.windowExplanation" _SUNDAY_HOLIDAY_NAME = SUNDAY_HOLIDAY_NAME _DUBBELDAGAR_MONTHS_LIMIT = settings.rules.dubbeldagar_months_limit # NOTE: `max_dubbeldagar` is deliberately NOT snapshotted at module-import # time (unlike `_DUBBELDAGAR_MONTHS_LIMIT` above) — `AnalyticsService. # validate_dubbeldagar`/`_validate_dubbeldagar_batch` read # `settings.rules.max_dubbeldagar` LIVE, and a stale allocator snapshot could # propose a dubbeldag the validator then rejects if the setting changes # mid-process. `_fund_both` reads it live instead — see `max_dubbeldagar` # local below. # NOTE: no module-level "_MAX_SAVABLE_AT_B4" here on purpose — which age-4 cap # applies (96 per child vs. 48 per parent) depends purely on parent count, so # it is resolved once per `propose()` run by `AllocationInputLoader` onto # `AllocationInput.age4_savable_cap` and read from there # (`self._input.age4_savable_cap`) everywhere `_Ledger` needs it — never as a # constant computed at import time. _SGI_MIN_DAYS_PER_WEEK = settings.rules.sgi_min_days_per_week_after_age_1 # A "week off" costs ~5 spendable weekdays (Sat/Sun are already free) — see # §3.4: "a 6-week summer window is ~42 calendar days but only ~30 spendable # ones". Used to turn a raw day count into an actionable week count. _SPENDABLE_DAYS_PER_WEEK = 5 # A surplus is "large" enough to justify front-loading (spending more while # the child is small, rather than trickling it out) once it clears roughly a # month of FP days. _LARGE_SURPLUS_THRESHOLD_DAYS = 4 * _SPENDABLE_DAYS_PER_WEEK # "Small child" window for the front_load_early_years suggestion. _FRONT_LOAD_CHILD_AGE_YEARS = 2 # §todo item 13 — how far `_shift_window_into_free_days_remedy` looks either # side of a fixed window's placed dates for a cheaper placement. Sized to # reliably catch a moving public holiday (Easter): the cached calendar # (`swedish_calendar.json`, 2026-2055) shows a max year-to-year Easter Sunday # drift of ~20 days, and the weekday-preserving snap applied when a fixed # series is materialised (`project_window`) can shift a naive same-month/day # anchor by up # to another 3 days -- 28 (4 weeks) covers that worst case with a small # margin. Kept a module constant, not unbounded, so `propose()` stays cheap: # each remedy call is O(bound * window_length) `is_free_day` lookups, each # already O(1). _SHIFT_REMEDY_SEARCH_BOUND_DAYS = 28 # How many ranked alternative placements the remedy reports at most. _SHIFT_REMEDY_MAX_SUGGESTIONS = 2 # §Decision 2 (flexible-window weekday alignment, user-confirmed) — when # ranking flexible-window candidates, prefer a Monday start so a repeating # wish lands on the same weekday every year (the reported bug). This is only # a tie-break: it sits after `shortfall`/`cost` in `_rank_candidate`, so a # non-Monday start that absorbs strictly more red days always wins — red # days outrank alignment. _PREFERRED_START_WEEKDAY = 0 # Monday, per `date.weekday()` # --------------------------------------------------------------------------- # §3.1 Input # --------------------------------------------------------------------------- @dataclass(frozen=True) class ChildFacts: id: UUID birth_date: date fourth_birthday: date preschool_start_date: date | None preschool_planning_days_per_year: int @dataclass(frozen=True) class ParentFacts: uuid: UUID name: str salary_type: str # 'monthly' | 'hourly' employment_status: str # Semesterlagen floor (statutory 20; employer may set higher) — see # PersonalData.min_vacation_days_per_year and PLANNING_ENGINE_SPEC §9.2. # Only days above this per (parent, year) may be carried forward, and # `use_less_vacation` surplus advice must never suggest dropping below it. min_vacation_days_per_year: int = 20 @dataclass(frozen=True) class DayPool: """Remaining (not yet used) FP days for one (parent, child) pair.""" sjukpenning_left: int lagsta_left: int dubbeldagar_left: int @dataclass(frozen=True) class VacationBudget: total_days: int saved_days_in: int already_used: int # True when no `vacation_days` row exists for this (parent, year) and # `AllocationInputLoader` defaulted `total_days` to the statutory # entitlement (§Task 2) rather than reading an entered value. A missing # row means "not entered yet", never "zero days" — this flag is how that # assumption stays visible in the report instead of silently passing for # a real entry, so the UI can eventually prompt the user to confirm it. is_assumed: bool = False @property def available(self) -> int: return self.total_days + self.saved_days_in - self.already_used @dataclass(frozen=True) class AllocationInput: today: date horizon_end: date # The age-4 savable-days ceiling that applies to THIS workspace, resolved # once by the loader via `app.utils.tip_predicates.resolve_age4_savable_cap` # (96 per child for a genuine single parent, or 48 per parent — # non-poolable — as soon as the child has 2+ parents) — never re-derived # inside the (pure) `_Ledger`. age4_savable_cap: int children: list[ChildFacts] parents: list[ParentFacts] pools: dict[tuple[UUID, UUID], DayPool] # (parent, child) -> pool vacation: dict[tuple[UUID, int], VacationBudget] # (parent, year) -> budget wishes: list[LeaveWish] existing_planned: list[PlannedLeaveDay] holidays: dict[date, str] # Workspace-scoped calendar corrections (see DayOverride entity): # True = must work this date despite the calendar, False = off this date # despite the calendar. `is_free_day` is the single place these are # merged on top of `holidays`. day_overrides: dict[date, bool] = field(default_factory=dict) # §todo item 2 — per-workspace opt-in (default False, byte-identical to # pre-existing behaviour when unset): allow a free day (Saturday, Sunday, # public holiday) inside a funded window to be paid with an FP day # (never vacation — see `REASON_FP_RED_DAY`). Read only by `_fund_solo`/ # `_fund_both`; deliberately NOT consulted by `is_free_day` itself, which # stays the single, opt-in-agnostic definition of "is this day free" (it # is also used for costing/klämdag-cluster detection, where the day must # stay classified as free regardless of whether it ends up FP-funded). take_pl_on_red_days: bool = False # §fix (parent-reassignment moves dates, decision B) — the FULL, # never-filtered snapshot of every persisted `planned_leave_day` row for # this workspace, independent of `existing_planned` above (which # `only_pinned_existing=True` — see `AllocationInputLoader.load` — # narrows to pinned/ad-hoc rows only, so `sync_plan`'s re-plan can # legitimately re-derive an unpinned wish-attributed day). `None` (the # default every hand-built `AllocationInput` in the test suite gets # unless it opts in) means "same as `existing_planned`" — see # `_Ledger.__init__`. Read ONLY by `_Ledger._persisted_by_wish` (via # `_persisted_dates_for_wish`), which needs to find a flexible wish's # own prior placement to reuse it (`_reuse_already_funded_window`) even # when that placement is currently hidden from `existing_planned` and # even when it was funded under a parent the wish no longer resolves to # (the wish was reassigned) — matched by `wish_id`, not by parent, for # exactly that reason. Never used for resource netting or `_existing_lookup`. all_existing_planned: list[PlannedLeaveDay] | None = None # Restructure step 3b — `(parent_uuid, child_id, date)` for every # `DayAssertion` with a concrete `day_type` (`'sjukpenning'|'lagsta'| # 'vacation'`), i.e. every day the user has locked in place. This is the # ONE signal `_Ledger` consumes for "is this date immovable" — # `_reserve_pinned_days`, `_persisted_by_wish`/`_pinned_dates_by_wish`, # and `AllocationInputLoader.load`'s `only_pinned_existing` filter all # check membership here instead of reading a per-row `is_pinned` flag # (that column no longer exists on `planned_leave_day` — see migration # 010). Populated by `AllocationInputLoader.load` via # `app.entities.day_assertion.partition_assertion_keys`, the same # function `PlanningService.sync_plan`'s own pre-write snapshot uses — # see that function's docstring for why routing every reader through it # matters. Empty by default, so a hand-built `AllocationInput` in the # test suite that doesn't opt in produces byte-identical output. pinned_days: frozenset[tuple[UUID, UUID, date]] = frozenset() # §remove-this-day — every (parent_uuid, child_id, date) the user has # explicitly removed from the plan: a `DayAssertion` with `day_type= # 'none'` (restructure step 3b — previously a `PlannedDayExclusion` row, # now folded into the same table/signal as `pinned_days` above, via the # same `partition_assertion_keys` call). The allocator must never place # leave on one of these, however emphatically the wish covering it asks: # "removing a single day stays permanent, even on an auto-added period # day". Empty by default, so every hand-built `AllocationInput` in the # test suite keeps producing byte-identical output. Read only via # `_Ledger._is_excluded` / `_Ledger._excluded_window_dates`; deliberately # NOT merged into `day_overrides`/`is_free_day`, which answer the # unrelated *calendar* question "is this date a working day for this # workspace" (a removed Tuesday is still a working Tuesday, for # analytics and for everyone else in the workspace). excluded_days: frozenset[tuple[UUID, UUID, date]] = frozenset() # §Task 1 (per-year preschool reserve) — (child_id, year) -> the TRUE # reserve for that specific calendar year, sourced from a # `PreschoolPlanningDays` row: recorded `taken_days` wins over # `planned_days` when present (the closure already happened, so what was # actually taken is the true cost — see that entity's docstring), # otherwise `planned_days`. A (child, year) pair with NO row here # (including every year for a workspace whose loader wasn't given a # `PreschoolPlanningDaysRepository`) falls back to `ChildFacts. # preschool_planning_days_per_year` in `_preschool_reserve_days` — never # to 0 — matching how a missing `vacation_days` row falls back to the # statutory default rather than 0. Empty by default so every hand-built # `AllocationInput` in the test suite keeps the old flat-lump behaviour. preschool_reserve_by_year: dict[tuple[UUID, int], int] = field(default_factory=dict) # §Task 2 (vacation-forfeiture bias) — (parent_uuid, year) -> the number # of vacation days forecast to be LOST at that year's end (age-cap + # over-cap combined — same `VacationYearBalance.expiring` the Analytics # tab shows), for every year from `today.year` through `horizon_end.year` # inclusive. Computed by `AllocationInputLoader` from already-committed # usage, BEFORE this run's own placements are added — see # `_Ledger._vacation_expiry_warnings`'s docstring for what that means for # the warning it drives. Never contains a year before `today.year`: a # past year's loss already happened and must never resurface here or bias # `_assign_tier3`. Empty by default, so every hand-built `AllocationInput` # in the test suite keeps producing byte-identical output absent an # explicit forfeiture figure. vacation_forfeiting: dict[tuple[UUID, int], int] = field(default_factory=dict) # --------------------------------------------------------------------------- # §3.7 Output # --------------------------------------------------------------------------- @dataclass(frozen=True) class ProposedDay: date: date parent_personal_data_uuid: UUID child_id: UUID | None day_type: str # 'sjukpenning' | 'lagsta' | 'vacation' is_dubbeldag: bool wish_id: UUID reason: str @dataclass(frozen=True) class ShortfallRemedy: kind: str message_key: str # i18n key for frontend — see MESSAGE_KEY_WINDOW_EXPLANATION note above data: dict[str, object] days_freed: int @dataclass(frozen=True) class SurplusAdvice: """§9.2 — the mirror of a `ShortfallRemedy`: a concrete, ranked suggestion emitted when a parent is forecast to *forfeit* FP days at the age-4 boundary rather than run short. Mirrors `ShortfallRemedy`'s (kind, message_key/data, day-count) shape; `data` mirrors `PlanWarning.data` so the UI can key the suggestion to the (parent, child) it concerns. """ kind: str # 'wish_more_leave' | 'shift_mix_to_fp' | 'front_load_early_years' | 'use_less_vacation' message_key: str # i18n key for frontend days_involved: int data: dict[str, object] @dataclass(frozen=True) class Age4Projection: """§3.3 part 2 — plan-aware age-4 projection for one (parent, child) pair. The same `(projected_left_at_b4, expiring_at_b4)` numbers that drive `_age4_warnings`/`build_surplus_advice`, but exposed unconditionally on every proposal (not just when there is a warning) so the UI can show "if this plan were applied, here's what's left / would expire at the 4th birthday" *before* the plan is committed. Mirrors the shape of `app.utils.tip_predicates.days_left_at_age_4_projection`'s DB-backed sibling in `AnalyticsService`, scoped to what a pure `PlanProposal` can know (no `days_used` from persisted history beyond what's already folded into the pool). """ parent_personal_data_uuid: UUID child_id: UUID fourth_birthday: date projected_left_at_b4: int savable_after_b4: int # min(projected_left_at_b4, savable_cap_at_b4) expiring_at_b4: int # max(projected_left_at_b4 - savable_cap_at_b4, 0) — LOST if not used # The cap actually applied — 96 (single parent) or 48 (non-poolable, # 2+ parents) — see `app.utils.tip_predicates.resolve_age4_savable_cap`. savable_cap_at_b4: int @dataclass(frozen=True) class WindowOutcome: wish_id: UUID wish_name: str # i18n provenance of `wish_name` (see `app.entities.leave_wish.LeaveWish # .name_message_key`) — carried alongside the literal so the frontend can # re-translate a preset-created wish's name into the UI's current locale, # exactly like `resolveWishName` already does for the wish editor. `None` # for a free-typed/user-renamed wish, or any row predating this field. wish_name_message_key: str | None year: int start_date: date end_date: date weeks_wished: int weeks_funded: float shortfall_days: int # What actually happened: 'solo' | 'parent_swap' | 'dubbeldag' | # 'co_leave' | 'swap' | 'unplaced'. Never lies — a 'both'-mode window # that degrades from dubbeldagar/co-leave down to a sequential swap for # some (or all) of its dates reports 'swap' or 'co_leave' here, not the # requested-but-unachieved 'dubbeldag'. See `requested_strategy`. strategy: str # What `parent_mode` (+ `wants_dubbeldag`) asked for, independent of # whether it was achieved: 'solo' | 'parent_swap' | 'dubbeldag' | # 'co_leave'. Compare against `strategy` to tell "as requested" from # "fell back" — the frontend surfaces this distinction (e.g. a banner # when a 'both' wish couldn't keep both parents on every date). requested_strategy: str # True when `strategy != requested_strategy` for a 'both'-mode window: # the whole-window comparison already computed by `requested_strategy` # vs. `strategy`) sees a date that couldn't be reached this run. strategy_degraded: bool # Whether dubbeldagar were a legally available option for this window # (child inside the 15-month limit, quota left, both parents held FP) — # regardless of whether `wish.wants_dubbeldag` was set. Only meaningful # for a 'both'-mode window (always False otherwise). The frontend's # period editor uses this to decide whether to SHOW the dubbeldag # opt-in checkbox at all. dubbeldagar_legally_available: bool message_key: str # i18n key for frontend — always MESSAGE_KEY_WINDOW_EXPLANATION today data: dict[str, object] cost_breakdown: dict[str, int] remedies: list[ShortfallRemedy] # §todo item (red-day shift toast) — mirrors `_WishWindow.shift_gain_days` # (see its docstring): spendable days the flexible placement search saved # versus the naive bound-start placement. 0 for every fixed window and # for a reused flexible window (see `_WishWindow`). shift_gain_days: int = 0 # Bug 4 fix (day-assertion-subsystem review) — the SAME `series_id` every # sibling occurrence of a repeating wish shares (`None` for a standalone # wish). A materialised series is one `LeaveWish` row PER YEAR, each with # its OWN `wish_id`, so a caller that only knows the occurrence it just # edited (`sync_with_wish`'s returned `wish.id`) cannot otherwise tell # which OTHER outcomes in this same batch belong to the same series — # see `Calendar.tsx`'s `buildSyncToastMessage`, which used to check only # the edited occurrence's own outcome and so silently missed a 100% # shortfall on a sibling year (accepted, no rejection, empty band, no # warning — indistinguishable from "nothing changed"). series_id: UUID | None = None @dataclass(frozen=True) class PlanWarning: kind: str # 'sgi_risk' | 'days_lost_at_age_4' | 'vacation_expiring' message_key: str # i18n key for frontend data: dict[str, object] @dataclass(frozen=True) class PlanBalances: pools_after: dict[tuple[UUID, UUID], DayPool] vacation_after: dict[tuple[UUID, int], VacationBudget] @dataclass(frozen=True) class PlanProposal: days: list[ProposedDay] outcomes: list[WindowOutcome] balances_after: PlanBalances warnings: list[PlanWarning] # §9.2 — ranked, concrete suggestions for parents forecast to forfeit FP # days at the age-4 boundary. Empty when there is no surplus to advise on. surplus_advice: list[SurplusAdvice] = field(default_factory=list) # §3.3 part 2 — plan-aware age-4 projection per (parent, child), always # present (unlike `surplus_advice`, which is empty absent a surplus). age4_projections: list[Age4Projection] = field(default_factory=list) # --------------------------------------------------------------------------- # Pure helpers (§3.4) — free-day classification and red-day clusters # --------------------------------------------------------------------------- def _dates_in(start: date, end: date) -> list[date]: n = (end - start).days + 1 return [start + timedelta(days=i) for i in range(n)] def is_free_day( d: date, holidays_map: dict[date, str], overrides: dict[date, bool] | None = None ) -> bool: """§3.4 + DayOverride: the single definition of "is this day free". A workspace `DayOverride` (if any) wins outright: `overrides[d] = True` means the workspace must work `d` despite the calendar (removes weekend/holiday-free status, even for a Saturday); `overrides[d] = False` means the workspace is off `d` despite the calendar. Absent an override, falls back to: Saturday, or present in the holiday set (which already includes every Sunday as "Söndag" — verified against `holidays.Sweden(language="sv")`). Do NOT special-case Sunday additionally. """ if overrides and d in overrides: return not overrides[d] return d.weekday() == 5 or d in holidays_map def _is_real_red_day(d: date, holidays_map: dict[date, str]) -> bool: """A holiday-set entry that is not *only* the generic "Söndag" marker. Used to detect genuine red-day clusters (Jul, Påsk, ...) without conflating them with bare weekends, per the §3.4 gotcha. """ name = holidays_map.get(d) return name is not None and name != _SUNDAY_HOLIDAY_NAME def _red_day_cluster_dates( start: date, end: date, holidays_map: dict[date, str], overrides: dict[date, bool] | None = None, ) -> set[date]: """Dates belonging to a maximal free-day run that contains >=1 real holiday. A bare weekend (Sat+Sun with no real holiday) is a free-day run too, but is NOT a "red-day cluster" — only a run touched by an actual public holiday counts, matching §3.3's distinction between tier-2 (red-day klämdagar, goal-independent) and the tier-3 weekend-wrap choice (goal-dependent). """ scan_start = start - timedelta(days=3) scan_end = end + timedelta(days=3) scan_dates = _dates_in(scan_start, scan_end) clusters: list[list[date]] = [] current: list[date] = [] for d in scan_dates: if is_free_day(d, holidays_map, overrides): current.append(d) else: if current: clusters.append(current) current = [] if current: clusters.append(current) result: set[date] = set() for cluster in clusters: if any(_is_real_red_day(d, holidays_map) for d in cluster): result.update(cluster) return result def _preschool_reserve_days( child: ChildFacts, today: date, horizon_end: date, overrides: dict[tuple[UUID, int], int] | None = None, ) -> int: """§9.3 + §Task 1: days reserved off the top for preschool planning-day closures, summed year by year. `overrides` (`AllocationInput.preschool_reserve_by_year`) supplies the TRUE per-year figure where a `PreschoolPlanningDays` row exists for that (child, year) — see that field's docstring for the taken/planned/missing precedence. A year with no override falls back to `child.preschool_planning_days_per_year`, exactly reproducing the old flat `per_year * years` lump sum for every year this loader/test doesn't supply a recorded figure for. """ if child.preschool_start_date is None or child.preschool_start_date > horizon_end: return 0 start_year = max(child.preschool_start_date.year, today.year) end_year = horizon_end.year if end_year < start_year: return 0 overrides = overrides or {} return sum( overrides.get((child.id, year), child.preschool_planning_days_per_year) for year in range(start_year, end_year + 1) ) # --------------------------------------------------------------------------- # Window expansion (§3.2) # --------------------------------------------------------------------------- @dataclass(frozen=True) class _WishWindow: wish: LeaveWish anchor_year: int start: date end: date # §todo item (red-day shift toast) — spendable days saved by the flexible # placement search versus the naive "first slot in the bound" placement. # Only ever set by `resolve_flexible_window`'s fresh-search path; the # horizon-clipped early return and the reuse-already-funded path both # leave this at the default 0 (see their call sites for why). Zero for # every `window_mode == 'fixed'` window (`_expand_all_windows` never # passes it) — a fixed window is never auto-shifted, so there is nothing # to report here for it. shift_gain_days: int = 0 def _expand_all_windows(alloc_input: AllocationInput) -> list[_WishWindow]: """§3.2 — every `window_mode == 'fixed'` wish's concrete, horizon-clipped window. One row is one year (see `app.entities.leave_wish`): a repeating wish was materialised into per-year rows by `LeaveWishService`, each already carrying its own dates — including the fixed-mode weekday snap, which is applied once at materialisation rather than re-derived here. So this is now a filter, not an expansion. Flexible wishes (§10.2) are handled separately by `_expand_flexible_wish_years`, since their window is a *bounding range* to search inside, not the final placement. """ windows: list[_WishWindow] = [] for wish in alloc_input.wishes: if wish.window_mode != "fixed": continue start, end = wish.window() if end < alloc_input.today or start > alloc_input.horizon_end: continue clipped_start = max(start, alloc_input.today) clipped_end = min(end, alloc_input.horizon_end) if clipped_start > clipped_end: continue windows.append( _WishWindow( wish=wish, anchor_year=wish.first_year, start=clipped_start, end=clipped_end ) ) return windows @dataclass(frozen=True) class _FlexibleWishYear: """One anchor-year occurrence of a `window_mode == 'flexible'` wish, resolved down to its (horizon-clipped) bounding range — the range `Allocator.propose()`'s pass 2 (§10.2) searches inside for the best- costing placement, not the final window. """ wish: LeaveWish anchor_year: int bound_start: date bound_end: date # §silent-failure gap fix — True when the wish's RAW bound (before any # clipping) ends entirely before `today`. Unlike the horizon-future case # (excluded from this list entirely — it will come back into range on a # later re-run) or a merely-partially-past bound (clipped to # `>= today` and funded as an ordinary shortfall, unchanged), a fully # elapsed bound can never be revisited. `Allocator.propose()` reports it # via `build_unplaced_outcome` instead of running it through # `resolve_flexible_window` — `bound_start`/`bound_end` here are the # wish's original, UNCLIPPED dates (clipping them to `today` would make # `bound_start > bound_end`), which is also what a user wants to see: the # window they actually asked for, not a clipped fragment of it. fully_past: bool = False def _expand_flexible_wish_years(alloc_input: AllocationInput) -> list[_FlexibleWishYear]: """§10.2 — the flexible-wish counterpart to `_expand_all_windows`. Same horizon clip; the only difference is that the row's window is treated as a bound to search inside (§10.2 step 1), not the final placement. Each year of a repeating flexible wish is its own row carrying that year's bounding range, and the per-year placement search inside it still runs at plan time (`_Ledger.resolve_flexible_window`) — materialising the rows changed where the range comes from, not that the search happens. """ wish_years: list[_FlexibleWishYear] = [] for wish in alloc_input.wishes: if wish.window_mode != "flexible": continue bound_start, bound_end = wish.window() if bound_start > alloc_input.horizon_end: # Not yet in range -- excluded, not reported: a later re-run # (horizon advances, or the user edits the bound) can still pick # this wish-year up, so there is nothing to tell the user today. continue if bound_end < alloc_input.today: # §silent-failure gap fix -- fully elapsed, and (unlike the # horizon-future case above) this can NEVER come back into # range. Keep it, unclipped, so `Allocator.propose()` reports it # as an 'unplaced' outcome instead of dropping it with no signal. wish_years.append( _FlexibleWishYear( wish=wish, anchor_year=wish.first_year, bound_start=bound_start, bound_end=bound_end, fully_past=True, ) ) continue clipped_start = max(bound_start, alloc_input.today) clipped_end = min(bound_end, alloc_input.horizon_end) if clipped_start > clipped_end: continue wish_years.append( _FlexibleWishYear( wish=wish, anchor_year=wish.first_year, bound_start=clipped_start, bound_end=clipped_end, ) ) return wish_years def _candidate_windows( bound_start: date, bound_end: date, length_days: int ) -> list[tuple[date, date]]: """§10.2 step 5 — every full-length `(start, end)` candidate (inclusive) that fits entirely inside `[bound_start, bound_end]`, sliding one day at a time (nothing requires a candidate to start on a Monday — the generator stays exhaustive; `_rank_candidate` is what prefers a Monday start among the candidates this returns, per §Decision 2). """ candidates: list[tuple[date, date]] = [] last_start = bound_end - timedelta(days=length_days - 1) s = bound_start while s <= last_start: candidates.append((s, s + timedelta(days=length_days - 1))) s += timedelta(days=1) return candidates def _candidate_cost( start: date, end: date, holidays_map: dict[date, str], overrides: dict[date, bool] ) -> int: """§10.2 step 6 — calendar-only cost: the count of spendable days in `[start, end]`. Deliberately does not run tier-3/goal logic or simulate which resource would be spent — see the spec's proof that this is exactly "fewest paid days" for every goal and `parent_mode`. """ return sum(1 for d in _dates_in(start, end) if not is_free_day(d, holidays_map, overrides)) def _split_dates_by_spendable_cost( dates: list[date], holidays_map: dict[date, str], overrides: dict[date, bool], ) -> tuple[list[date], list[date]]: """§fix (funding-fairness) — split a contiguous run of dates into two contiguous halves balanced by *spendable* (non-free) day count rather than raw calendar-day count. Free days (weekends, red days) cost neither parent anything, so a calendar-midpoint split can hand one parent a half that is mostly free days while the other gets nearly the whole cost — an uneven split in the one dimension that actually matters. Keeps the previous tie-break convention: the first half gets the extra spendable day when the count is odd. Falls back to a calendar-midpoint split when the range has no spendable days at all (nothing to balance, and `target == 0` would otherwise degenerate to an all-or-nothing split). """ spendable_total = sum(1 for d in dates if not is_free_day(d, holidays_map, overrides)) if spendable_total == 0: mid_index = (len(dates) + 1) // 2 return dates[:mid_index], dates[mid_index:] target = (spendable_total + 1) // 2 count = 0 split_at = len(dates) for i, d in enumerate(dates): if not is_free_day(d, holidays_map, overrides): count += 1 if count >= target: split_at = i + 1 break return dates[:split_at], dates[split_at:] # --------------------------------------------------------------------------- # Mutable working ledger — internal to a single `Allocator.propose()` call. # --------------------------------------------------------------------------- @dataclass class _MutablePool: sjukpenning_left: int lagsta_left: int dubbeldagar_left: int clamped_at_b4: bool = False @dataclass class _MutableVacation: total_days: int saved_days_in: int already_used: int is_assumed: bool = False @property def available(self) -> int: return self.total_days + self.saved_days_in - self.already_used @dataclass(frozen=True) class _Responsible: mode: str # 'solo' | 'swap' | 'dubbeldag' parents: list[ParentFacts] _EMPTY_COST: dict[str, int] = {"free": 0, "vacation": 0, "sjukpenning": 0, "lagsta": 0} def _new_cost() -> dict[str, int]: return dict(_EMPTY_COST) def _shortfall_and_weeks_funded( mode: str, all_dates: list[date], days: list["ProposedDay"], cost: dict[str, int], ) -> tuple[int, float]: """Shared by `_Ledger.fund_window` (real) and `_Ledger._candidate_shortfall` (§Decision C's budget-aware probe) so the two can never silently disagree on what "funded" means for the same (mode, dates, days, cost). 'dubbeldag' mode (both parents, §Decision A) spends TWO resource units per co-funded date (one per parent — dubbeldagar and co-leave both emit 2 `ProposedDay`s per date, see `_fund_both`), so the cost-dict-sum math below (built for exactly one resource per calendar date) would double-count every co-funded date. Counted by DATE instead there: a date is "funded" once at least one parent has a non-free `ProposedDay` for it. Every other mode ('solo' | 'swap') counts via `cost` rather than `days`, deliberately: `_fund_solo` skips emitting a `ProposedDay` entirely for a date `_existing_lookup` already has (a previously persisted/applied day — see its docstring), but still increments `cost[existing_type]` for it. A `days`-based count would silently treat every already-applied date as unfunded, which is exactly wrong for `_candidate_shortfall`'s purpose: ranking a candidate that reuses already-persisted dates (§fix, decision B) must see those dates as funded, not missing. """ if mode == "dubbeldag": funded_dates = {d.date for d in days if d.day_type != "free"} free_dates = {d.date for d in days if d.day_type == "free"} spendable_total = len(all_dates) - len(free_dates) funded_spendable = len(funded_dates) shortfall_days = spendable_total - funded_spendable weeks_funded = round((len(free_dates) + len(funded_dates)) / 7, 2) else: funded_home_days = cost["free"] + cost["vacation"] + cost["sjukpenning"] + cost["lagsta"] spendable_total = len(all_dates) - cost["free"] funded_spendable = funded_home_days - cost["free"] shortfall_days = spendable_total - funded_spendable weeks_funded = round(funded_home_days / 7, 2) return shortfall_days, weeks_funded class _Ledger: """Mutable bookkeeping for a single, pure `Allocator.propose()` run.""" def __init__(self, alloc_input: AllocationInput) -> None: self._input = alloc_input self._children_by_id = {c.id: c for c in alloc_input.children} self._parents_by_id = {p.uuid: p for p in alloc_input.parents} self._pools: dict[tuple[UUID, UUID], _MutablePool] = { key: _MutablePool(pool.sjukpenning_left, pool.lagsta_left, pool.dubbeldagar_left) for key, pool in alloc_input.pools.items() } self._vacation: dict[tuple[UUID, int], _MutableVacation] = { key: _MutableVacation(vb.total_days, vb.saved_days_in, vb.already_used, vb.is_assumed) for key, vb in alloc_input.vacation.items() } self._dubbeldagar_used: dict[UUID, int] = {} self._preschool_reserved: dict[UUID, int] = {} self._existing_lookup: dict[tuple[UUID, UUID | None, date], str] = { (p.parent_personal_data_uuid, p.child_id, p.date): p.day_type for p in alloc_input.existing_planned } # §eviction-preview fix — the KEYS of `_existing_lookup` as of THIS # point only, before `record_days()` (called after every # `fund_window()`, see below) starts folding in this run's own # freshly-computed placements. `_fund_solo`'s cost-crediting (below) # consults this to tell "this date is already a genuinely pre- # existing/pinned commitment -- count it toward THIS window's # funded total" from "a DIFFERENT, earlier-processed wish just # claimed this date THIS SAME RUN" -- the latter must never be # credited to a later-processed wish's own `shortfall_days` # (crediting it was the actual bug: a lower-priority wish's window # silently reported `shortfall_days == 0` even though every one of # its dates had actually gone to a different wish, which meant # `_genuinely_evicted_rows` could never detect the eviction its own # persisted rows were about to suffer). self._pre_run_occupied: frozenset[tuple[UUID, UUID | None, date]] = frozenset( self._existing_lookup ) # §fix (double-booking on flexible re-propose; extended for §fix # parent-reassignment, decision B) — an immutable snapshot of ONLY # truly-persisted planned days, taken before `record_days()` ever # folds in this run's freshly-computed (not-yet-applied) windows. # `_existing_lookup` above is mutated by `record_days()` as pass 1/2 # proceed, so a fixed wish's just-computed placement would otherwise # be indistinguishable from a flexible wish's own prior occurrence # (see `TestFixedWishNeverDisplacedByFlexible`, which must still see # an empty view here). # # Sourced from `all_existing_planned` when the caller provided one # (production always does via `AllocationInputLoader`) — the FULL, # never-`only_pinned_existing`-filtered snapshot — so this index # still finds a wish's prior placement even when `existing_planned` # itself has been narrowed down for a `sync_plan` re-plan. Falls # back to `existing_planned` when `all_existing_planned` is `None` # (every hand-built `AllocationInput` in the test suite that doesn't # opt in) so the two stay equivalent absent that narrowing. # # Grouped by (wish_id, child_id) rather than by parent: matching by # parent would silently stop finding a wish's own prior window the # moment the wish is reassigned to a different parent — exactly the # dates-move-on-reassignment regression this index exists to # prevent. `_persisted_dates_for_wish` is the only reader. persisted_source = ( alloc_input.all_existing_planned if alloc_input.all_existing_planned is not None else alloc_input.existing_planned ) self._persisted_by_wish: dict[tuple[UUID, UUID | None], set[date]] = {} # §fix (Step 3, flexible-window realignment escape hatch) — same # grouping as `_persisted_by_wish`, narrowed to the pinned rows. Lets # `_reuse_already_funded_window` tell "this wish-year's persisted # dates are just where the search last happened to land" (safe to # release for a better-ranked candidate) from "the user pinned one of # these dates on purpose" (must never move). self._pinned_dates_by_wish: dict[tuple[UUID, UUID | None], set[date]] = {} for p in persisted_source: if p.wish_id is None: continue key = (p.wish_id, p.child_id) self._persisted_by_wish.setdefault(key, set()).add(p.date) if (p.parent_personal_data_uuid, p.child_id, p.date) in alloc_input.pinned_days: self._pinned_dates_by_wish.setdefault(key, set()).add(p.date) self._reserve_pinned_days() self._reserve_preschool_days() # -- setup ---------------------------------------------------------- def _reserve_pinned_days(self) -> None: """§Task 4 — debit a pinned day's cost off the top, before any window is funded, exactly like `_reserve_preschool_days` below reserves preschool closure days. A pinned `planned_leave_day` is a hard, already-spent fact regardless of which mode loaded this input (see `AllocationInputLoader.load`'s `only_pinned_existing`, which never nets a pinned row out of the pools/vacation budgets itself — this is the one place that happens). `_existing_lookup`/ `_persisted_lookup` (built above from the same `existing_planned`) separately keep the Allocator from ever reassigning a pinned day's type or parent — this method only handles the resource debit. """ for p in self._input.existing_planned: key = (p.parent_personal_data_uuid, p.child_id, p.date) if key not in self._input.pinned_days: continue if p.day_type == "vacation": self._spend_vacation(p.parent_personal_data_uuid, p.date) continue pool = self._pools.get((p.parent_personal_data_uuid, p.child_id)) if pool is None: continue if p.day_type == "sjukpenning" and pool.sjukpenning_left > 0: pool.sjukpenning_left -= 1 elif p.day_type == "lagsta" and pool.lagsta_left > 0: pool.lagsta_left -= 1 def _reserve_preschool_days(self) -> None: for child in self._input.children: reserve = _preschool_reserve_days( child, self._input.today, self._input.horizon_end, self._input.preschool_reserve_by_year, ) if reserve <= 0: continue self._preschool_reserved[child.id] = reserve parent_keys = sorted( (key for key in self._pools if key[1] == child.id), key=lambda k: str(k[0]), ) if not parent_keys: continue share, remainder = divmod(reserve, len(parent_keys)) for i, key in enumerate(parent_keys): take = share + (1 if i < remainder else 0) pool = self._pools[key] from_sjuk = min(take, pool.sjukpenning_left) pool.sjukpenning_left -= from_sjuk take -= from_sjuk from_lagsta = min(take, pool.lagsta_left) pool.lagsta_left -= from_lagsta # -- parent resolution (§3.3, parent_swap) --------------------------- def _resolve_parents(self, wish: LeaveWish) -> _Responsible: # Couple wish-assignment (swap/dubbeldag) kicks in purely on parent # count — whether the workspace has a second parent at all. There is # no separate "planning mode" flag any more: adding a partner to the # workspace IS the signal. parents = self._input.parents if len(parents) < 2: return _Responsible(mode="solo", parents=[self._pick_solo_parent(wish, parents)]) if wish.parent_mode == "specific": specific = next( (p for p in parents if p.uuid == wish.parent_personal_data_uuid), parents[0] ) return _Responsible(mode="solo", parents=[specific]) if wish.parent_mode == "both": return _Responsible(mode="dubbeldag", parents=list(parents[:2])) # 'either' if wish.target_weeks >= 2: return _Responsible(mode="swap", parents=list(parents[:2])) return _Responsible(mode="solo", parents=[parents[0]]) @staticmethod def _pick_solo_parent(wish: LeaveWish, parents: list[ParentFacts]) -> ParentFacts: if wish.parent_mode == "specific": for p in parents: if p.uuid == wish.parent_personal_data_uuid: return p return parents[0] # -- child resolution (root cause of the FP-fallback bug) -------------- def _resolve_wish_child(self, wish: LeaveWish) -> ChildFacts | None: """Which child's FP pool this wish draws from, if any. An explicit `wish.child_id` always wins (and, per prior behaviour, a stale/unknown id resolves to "no pool" rather than guessing). A **family-wide** wish (`child_id is None`) used to mean "no FP pool to draw from, vacation or nothing" unconditionally. That was the bug: for the overwhelmingly common case of a workspace with exactly one child, "family-wide" is not actually ambiguous — there is only one pool it could mean — yet the allocator refused to touch it and reported a shortfall while the child's föräldrapenning pool sat completely full (PLANNING_ENGINE_SPEC.md §3.5: FP is a legitimate funding source for interior days under every planning goal). With 0 or >=2 children a family-wide wish stays genuinely ambiguous (which child's pool would even apply is a product decision, not implemented here) — the historical vacation-only behaviour is preserved for that case. """ if wish.child_id is not None: return self._children_by_id.get(wish.child_id) if len(self._input.children) == 1: return self._input.children[0] return None # -- resource spending ------------------------------------------------ # -- user-removed days (§remove-this-day) ------------------------------ def _is_excluded(self, parent_uuid: UUID, child_id: UUID | None, d: date) -> bool: """Whether the user removed THIS date from THIS parent's plan. The single reader of `AllocationInput.excluded_days`. A wish with no resolved child (`child_id is None`) can never match: an exclusion is always recorded from a real `planned_leave_day`, which always has a child. """ if child_id is None or not self._input.excluded_days: return False return (parent_uuid, child_id, d) in self._input.excluded_days def _excluded_window_dates( self, wish: LeaveWish, responsible: _Responsible, dates: list[date] ) -> set[date]: """Dates that leave a window ENTIRELY because no responsible parent may take them. The exclusion itself is per parent (see `DayAssertionService` for why), but what a *window* should do about it depends on how many parents that window puts on leave per date: - 'dubbeldag' (both parents off on the same dates): removing one parent's day must leave the partner's standing, so a date only drops out when BOTH parents were removed from it. The single-parent case is handled inside `_fund_both`/`_fund_solo`, which skip just the excluded parent. - 'solo'/'swap' (exactly one parent on leave per date): ANY responsible parent's exclusion empties the date. Without the `any`, `_fund_swap`'s leftover hand-off would hand the removed date straight to the partner — silently refilling the hole the user just made, under a different name, which is precisely the "it came back" failure this whole feature exists to prevent. Dropping these dates from the list `fund_window` funds (and from the list it measures) is also the shortfall decision: a day the user removed reduces the window's target rather than counting as a day the plan failed to fund. It is not a gap the planner could do anything about, `_build_remedies` must not propose freeing resources for it, and `sync_plan`'s eviction preview (which fires on a stale row whose own wish still reports `shortfall_days > 0`) must not mistake a deliberate removal for a lost priority race. """ if not self._input.excluded_days or not responsible.parents: return set() child = self._resolve_wish_child(wish) if child is None: return set() quantifier = all if responsible.mode == "dubbeldag" else any return { d for d in dates if quantifier(self._is_excluded(p.uuid, child.id, d) for p in responsible.parents) } def _window_dates( self, wish: LeaveWish, responsible: _Responsible, start: date, end: date ) -> list[date]: """`[start, end]` minus the dates the user removed from it. Shared by `fund_window` (the real placement) and `_candidate_shortfall` (§Decision C's budget-aware probe) so the two keep measuring the same window — see `_shortfall_and_weeks_funded`'s docstring for why they must never silently disagree. """ all_dates = _dates_in(start, end) excluded = self._excluded_window_dates(wish, responsible, all_dates) if not excluded: return all_dates return [d for d in all_dates if d not in excluded] def _spend_vacation(self, parent_uuid: UUID, d: date) -> bool: key = (parent_uuid, d.year) budget = self._vacation.get(key) if budget is None or budget.available <= 0: return False budget.already_used += 1 return True def _spend_fp(self, pool_key: tuple[UUID, UUID], prefer: str) -> str | None: pool = self._pools.get(pool_key) if pool is None: return None order = ["lagsta", "sjukpenning"] if prefer == "lagsta" else ["sjukpenning", "lagsta"] for tier in order: if tier == "sjukpenning" and pool.sjukpenning_left > 0: pool.sjukpenning_left -= 1 return "sjukpenning" if tier == "lagsta" and pool.lagsta_left > 0: pool.lagsta_left -= 1 return "lagsta" return None @staticmethod def _peek_fp_tier(pool: _MutablePool) -> str | None: if pool.sjukpenning_left > 0: return "sjukpenning" if pool.lagsta_left > 0: return "lagsta" return None @staticmethod def _commit_fp_tier(pool: _MutablePool, tier: str) -> None: if tier == "sjukpenning": pool.sjukpenning_left -= 1 else: pool.lagsta_left -= 1 def _try_co_leave_pair( self, fp_parent: ParentFacts, fp_pool: _MutablePool, vacation_parent: ParentFacts, d: date, ) -> tuple[ParentFacts, str, ParentFacts] | None: """§fix (long-chunk co-leave, 2026-09-19) — peek-then-commit a single co-leave pairing (one parent on FP, the other on vacation) for `d`. Returns `None` (spending NEITHER side) unless both can actually pay — used by `_fund_both` to try its chunk-designated pairing first and the swapped pairing as a same-date fallback. """ fp_tier = self._peek_fp_tier(fp_pool) vacation_budget = self._vacation.get((vacation_parent.uuid, d.year)) vacation_can_pay = vacation_budget is not None and vacation_budget.available > 0 if fp_tier is None or not vacation_can_pay: return None self._commit_fp_tier(fp_pool, fp_tier) self._spend_vacation(vacation_parent.uuid, d) return fp_parent, fp_tier, vacation_parent def _maybe_clamp_age4(self, pool_key: tuple[UUID, UUID] | None, window_start: date) -> None: """§9.1: from the 4th birthday onward, the FP pool is hard-capped at min(remaining, `self._input.age4_savable_cap`). Applied once per pool, the first time (in priority-then-chronological processing order) a window starting on or after the child's 4th birthday is funded from it. Known limitation: because windows are funded in (priority, chronological) order rather than pure chronological order, a lower-priority window that starts *before* B4 but is processed *after* a higher-priority window starting *after* B4 could see the clamp applied earlier than strict chronology would. This does not affect any of the §5 test-matrix scenarios and is documented rather than solved with a full chronological two-pass scheduler. """ if pool_key is None: return child = self._children_by_id.get(pool_key[1]) if child is None or window_start < child.fourth_birthday: return pool = self._pools.get(pool_key) if pool is None or pool.clamped_at_b4: return remaining = pool.sjukpenning_left + pool.lagsta_left capped = min(remaining, self._input.age4_savable_cap) excess = remaining - capped if excess > 0: cut_lagsta = min(excess, pool.lagsta_left) pool.lagsta_left -= cut_lagsta excess -= cut_lagsta pool.sjukpenning_left -= excess pool.clamped_at_b4 = True # -- tier-3 goal-dependent interior resolution (§3.3) ------------------ def _assign_tier3( self, wish: LeaveWish, d: date, parent: ParentFacts, child: ChildFacts | None, pool_key: tuple[UUID, UUID] | None, ) -> tuple[str, str] | None: """§3.3 tier-3 interior resolution — the single policy: maximise the value extracted from FP + vacation days combined. `salary_type` still branches (not a goal): a monthly-salaried parent loses nothing by spending a vacation day, so vacation goes first and FP (sjukpenning preferred) fills the rest; an hourly parent would lose pay to a vacation day, so FP goes first and vacation only backstops what FP can't cover. §Task 2 (vacation-forfeiture bias) overrides that hourly ordering for one specific case: when `d`'s (parent, year) is forecast to forfeit vacation anyway (`AllocationInput.vacation_forfeiting`), spending a vacation day here costs the household NOTHING it wasn't already going to lose, while the FP day it displaces stays banked for a later year — the whole point of the bias (see `_Ledger. _assign_tier3`'s call site and `AllocationInput.vacation_forfeiting`'s docstring). Never applies to a past year: every date reaching this method already comes from a window clipped to `>= today`, and `vacation_forfeiting` itself never carries a year before `today.year`. Only ever spends MORE vacation than the hourly default would, never less, so it can't push a parent below their `min_vacation_days_per_year` floor — that floor bounds how little vacation may be taken, not how much, and `_spend_vacation` still refuses once a year's budget is exhausted regardless of this bias. """ if child is None or pool_key is None: # No resolvable child (genuinely ambiguous family-wide wish, or a # stale child_id) -> no FP pool to draw from. Vacation or nothing. if self._spend_vacation(parent.uuid, d): return "vacation", REASON_VACATION_WRAP_WEEKEND return None vacation_forfeiting = self._input.vacation_forfeiting.get((parent.uuid, d.year), 0) > 0 if parent.salary_type == "monthly" or vacation_forfeiting: if self._spend_vacation(parent.uuid, d): return "vacation", REASON_VACATION_WRAP_WEEKEND fp = self._spend_fp(pool_key, prefer="sjukpenning") if fp: return fp, (REASON_FP_INTERIOR if fp == "sjukpenning" else REASON_FP_LAGSTA_STRETCH) return None fp = self._spend_fp(pool_key, prefer="sjukpenning") if fp: return fp, (REASON_FP_INTERIOR if fp == "sjukpenning" else REASON_FP_LAGSTA_STRETCH) if self._spend_vacation(parent.uuid, d): return "vacation", REASON_VACATION_WRAP_WEEKEND return None # -- per-parent day resolution (tiers 1-3) ----------------------------- def _fund_solo( self, wish: LeaveWish, dates: list[date], red_cluster: set[date], parent: ParentFacts, free_reason_override: str | None, ) -> tuple[list[ProposedDay], dict[str, int]]: days: list[ProposedDay] = [] cost = _new_cost() child = self._resolve_wish_child(wish) child_id = child.id if child is not None else None pool_key = (parent.uuid, child_id) if child_id is not None else None for d in dates: if self._is_excluded(parent.uuid, child_id, d): # §remove-this-day — the user removed this exact date for # this exact parent. Never re-derive it, and emit nothing at # all (not even a 'free' day): `fund_window`/ # `_candidate_shortfall` have already taken the date out of # the window's target wherever no responsible parent may take # it, so it is neither funded nor a gap. This is also what # stops `_fund_swap`'s leftover hand-off from handing a # parent a date they were removed from. continue key = (parent.uuid, child_id, d) existing_type = self._existing_lookup.get(key) if existing_type is not None: # §eviction-preview fix — only a genuinely pre-existing/ # pinned commitment (present in `_existing_lookup` BEFORE # this run's own `record_days()` calls started, see # `_pre_run_occupied`) counts toward THIS window's funded # total. A date claimed by a DIFFERENT, earlier-processed # wish this same run is simply unavailable to this one -- # falling through unfunded (never `continue`d past without # emitting a day) is what lets `shortfall_days` correctly # reflect the loss instead of silently masking it. if key in self._pre_run_occupied: cost[existing_type] = cost.get(existing_type, 0) + 1 continue if is_free_day(d, self._input.holidays, self._input.day_overrides): fp_on_red_day = ( self._spend_fp(pool_key, prefer="sjukpenning") if self._input.take_pl_on_red_days and pool_key is not None else None ) if fp_on_red_day is not None: cost[fp_on_red_day] += 1 days.append( ProposedDay( d, parent.uuid, child_id, fp_on_red_day, False, wish.id, REASON_FP_RED_DAY, ) ) continue cost["free"] += 1 reason = free_reason_override or ( REASON_FREE_RED_DAY if _is_real_red_day(d, self._input.holidays) else REASON_FREE_WEEKEND ) days.append(ProposedDay(d, parent.uuid, child_id, "free", False, wish.id, reason)) continue resolved: tuple[str, str] | None = None if ( d in red_cluster or (d - timedelta(days=1)) in red_cluster or (d + timedelta(days=1)) in red_cluster ): # Tier 2 — dominant play: klämdag inside a red-day cluster. # Fires regardless of goal or salary_type (§3.3). if self._spend_vacation(parent.uuid, d): resolved = ("vacation", REASON_VACATION_WRAP_RED_DAY) if resolved is None: resolved = self._assign_tier3(wish, d, parent, child, pool_key) if resolved is None: continue # unfunded — reflected in the window's shortfall_days day_type, reason = resolved cost[day_type] += 1 days.append(ProposedDay(d, parent.uuid, child_id, day_type, False, wish.id, reason)) return days, cost def _fund_swap( self, wish: LeaveWish, w: _WishWindow, dates: list[date], red_cluster: set[date], responsible: _Responsible, ) -> tuple[list[ProposedDay], dict[str, int]]: if not dates: return [], _new_cost() # §fix (funding-fairness) — split by *spendable* days, not calendar # days: a free (weekend/red) day costs neither parent anything, so a # raw date-count split can silently hand one parent a costlier half # than the other even before either parent's own budget is checked. first_half, second_half = _split_dates_by_spendable_cost( dates, self._input.holidays, self._input.day_overrides ) parent_a, parent_b = responsible.parents[0], responsible.parents[1] resolved_child = self._resolve_wish_child(wish) if resolved_child is not None: self._maybe_clamp_age4((parent_a.uuid, resolved_child.id), w.start) self._maybe_clamp_age4((parent_b.uuid, resolved_child.id), w.start) days_a, cost_a = self._fund_solo( wish, first_half, red_cluster, parent_a, REASON_SWAP_FIRST_HALF ) days_b, cost_b = self._fund_solo( wish, second_half, red_cluster, parent_b, REASON_SWAP_SECOND_HALF ) # §fix (funding-fairness) — a parent's own half can still leave # spendable days unfunded when THEIR resources (typically a per-year # vacation budget) run out, even though the partner's half was fully # funded with resources to spare. `_fund_solo` never raises for an # unfunded day — it just omits it from the returned list (see its # trailing `continue`) — so the gap is exactly "date in the half but # missing from the funded-dates set". Rather than report those days # as a shortfall while the partner's budget sits unused, hand each # half's unfunded leftover to the OTHER parent, who has already # picked up everything their own tier-3 resolution could reach and so # cannot conflict with what's being offered. A leftover date is never # a free/weekend/red day (those are always funded on the first pass, # unconditionally, before tier-2/tier-3 even run) so the # `free_reason_override` passed here is inert — kept `None` rather # than a half-label that could never actually be stamped on anything. # One hand-off per direction is enough: each parent's resources are # independent per (parent, child)/(parent, year) key, so a second # round could not free up anything the first round didn't already see. funded_a_dates = {day.date for day in days_a} funded_b_dates = {day.date for day in days_b} leftover_from_a = [d for d in first_half if d not in funded_a_dates] leftover_from_b = [d for d in second_half if d not in funded_b_dates] if leftover_from_b: extra_a, extra_cost_a = self._fund_solo( wish, leftover_from_b, red_cluster, parent_a, None ) days_a = days_a + extra_a for k in cost_a: cost_a[k] += extra_cost_a[k] if leftover_from_a: extra_b, extra_cost_b = self._fund_solo( wish, leftover_from_a, red_cluster, parent_b, None ) days_b = days_b + extra_b for k in cost_b: cost_b[k] += extra_cost_b[k] merged_cost = {k: cost_a[k] + cost_b[k] for k in cost_a} all_days = days_a + days_b all_days.sort(key=lambda p: p.date) return all_days, merged_cost def _fund_both( self, wish: LeaveWish, w: _WishWindow, dates: list[date], red_cluster: set[date], responsible: _Responsible, ) -> tuple[list[ProposedDay], dict[str, int], dict[str, int | bool]]: """§Decision A (co-leave, 2026-09-18 correction) — both parents off on the SAME dates. Three tiers, tried in order for every date: 1. Dubbeldagar — ONLY when `wish.wants_dubbeldag` is set (opt-in, never automatic — see `LeaveWish.wants_dubbeldag`) AND the date is inside the child's 15-month window AND quota remains AND both parents currently hold FP (peeked, then committed together — never a lone half-spend). 2. Co-leave (PL + vacation) — the default for every other date: which parent PAYS with FP is decided per CHUNK, not per day — §fix (long-chunk co-leave, 2026-09-19). The window's dates are split into two contiguous halves with `_split_dates_by_ spendable_cost` (the same balanced-by-cost split `_fund_swap` uses); `responsible.parents[0]` is the designated FP-payer for the first half, `responsible.parents[1]` for the second — so a long window reads as "parent A on PL while parent B is on vacation for ~half the window, then they switch", not a daily-alternating patchwork. Both parents are still off on EVERY date regardless of chunk. If the chunk's designated FP-payer has no FP left for a date (or their partner has no vacation left), the roles are tried swapped for just that date — since the exhausted resource stays exhausted, this sticks for the rest of the chunk rather than flip-flopping day to day. Each attempted pairing is peeked-then-committed together: if either side can't pay, NEITHER is spent for that date. 3. Swap (last resort) — a date co-leave genuinely cannot fund (the higher-FP parent has none left, or the other parent has no vacation left) is handed to `_fund_swap` together with the rest of that leftover run, which splits it into sequential single-parent halves — surrendering the "both off" property for just that date, not the whole window. Returns a third element (`meta`) — dubbeldag/co_leave/swap DATE counts (not day-instances: one per calendar date, however many `ProposedDay`s it produced) plus whether dubbeldagar were legally available at all — that `fund_window` uses to derive `strategy`/`requested_strategy`/`strategy_degraded`/ `dubbeldagar_legally_available`. """ parent_a, parent_b = responsible.parents[0], responsible.parents[1] child = self._resolve_wish_child(wish) # Read live, not from a module-import-time snapshot — see NOTE near # `_DUBBELDAGAR_MONTHS_LIMIT` above. max_dubbeldagar = settings.rules.max_dubbeldagar if child is None: days, cost = self._fund_swap(wish, w, dates, red_cluster, responsible) meta: dict[str, int | bool] = { "dubbeldag_dates": 0, "co_leave_dates": 0, "swap_dates": len({d.date for d in days}), "dubbeldagar_legally_available": False, } return days, cost, meta self._maybe_clamp_age4((parent_a.uuid, child.id), w.start) self._maybe_clamp_age4((parent_b.uuid, child.id), w.start) pool_a = self._pools.get((parent_a.uuid, child.id)) pool_b = self._pools.get((parent_b.uuid, child.id)) eligible_end = child.birth_date + relativedelta(months=_DUBBELDAGAR_MONTHS_LIMIT) used = self._dubbeldagar_used.get(child.id, 0) dubbeldagar_legally_available = ( pool_a is not None and pool_b is not None and any(d <= eligible_end for d in dates) and used < max_dubbeldagar and self._peek_fp_tier(pool_a) is not None and self._peek_fp_tier(pool_b) is not None ) want_dubbeldag = wish.wants_dubbeldag # §fix (long-chunk co-leave, 2026-09-19) — decide the FP-payer per # CHUNK, not per day, so a long window reads as one switch instead # of a daily-alternating patchwork (see the tier-2 docstring above). # Reuses `_fund_swap`'s own balanced-by-spendable-cost split rather # than a second splitter. first_half, _second_half = _split_dates_by_spendable_cost( dates, self._input.holidays, self._input.day_overrides ) first_half_dates: frozenset[date] = frozenset(first_half) days: list[ProposedDay] = [] cost = _new_cost() dubbeldag_dates: set[date] = set() co_leave_dates: set[date] = set() leftover_for_swap: list[date] = [] for d in dates: existing_a = self._existing_lookup.get((parent_a.uuid, child.id, d)) existing_b = self._existing_lookup.get((parent_b.uuid, child.id, d)) if existing_a is not None or existing_b is not None: leftover_for_swap.append(d) continue if is_free_day(d, self._input.holidays, self._input.day_overrides): # §todo item 2 — each parent's own FP pool is tried # independently (never vacation, never a peeked-then- # committed-together pair like dubbeldagar/co-leave above): # one parent running out of FP must never stop the other # from still getting paid for this same red day. free_reason = ( REASON_FREE_RED_DAY if _is_real_red_day(d, self._input.holidays) else REASON_FREE_WEEKEND ) for red_parent, red_pool in ((parent_a, pool_a), (parent_b, pool_b)): if self._is_excluded(red_parent.uuid, child.id, d): # Removed for this parent — the partner still gets # their free day for this date. continue fp_on_red_day = ( self._spend_fp((red_parent.uuid, child.id), prefer="sjukpenning") if self._input.take_pl_on_red_days and red_pool is not None else None ) if fp_on_red_day is not None: cost[fp_on_red_day] += 1 days.append( ProposedDay( d, red_parent.uuid, child.id, fp_on_red_day, False, wish.id, REASON_FP_RED_DAY, ) ) else: cost["free"] += 1 days.append( ProposedDay( d, red_parent.uuid, child.id, "free", False, wish.id, free_reason ) ) continue if self._is_excluded(parent_a.uuid, child.id, d) or self._is_excluded( parent_b.uuid, child.id, d ): # §remove-this-day — one parent was removed from this date # (never both: `fund_window` already dropped those dates from # `dates` entirely). Dubbeldagar and co-leave each spend for # BOTH parents, so neither can fund a date only one parent may # take. Hand it to the swap path, which funds it for whichever # parent is still allowed to have it — `_fund_solo` skips the # excluded one — so the partner keeps their day. leftover_for_swap.append(d) continue funded = False if ( want_dubbeldag and pool_a is not None and pool_b is not None and d <= eligible_end and used < max_dubbeldagar ): tier_a = self._peek_fp_tier(pool_a) tier_b = self._peek_fp_tier(pool_b) if tier_a is not None and tier_b is not None: self._commit_fp_tier(pool_a, tier_a) self._commit_fp_tier(pool_b, tier_b) used += 1 cost[tier_a] += 1 cost[tier_b] += 1 days.append( ProposedDay( d, parent_a.uuid, child.id, tier_a, True, wish.id, REASON_DUBBELDAG ) ) days.append( ProposedDay( d, parent_b.uuid, child.id, tier_b, True, wish.id, REASON_DUBBELDAG ) ) dubbeldag_dates.add(d) funded = True if not funded and pool_a is not None and pool_b is not None: # §fix (long-chunk co-leave, 2026-09-19) — the FP-payer is # decided once per half of the window (`first_half_dates`), # not by comparing live remaining balances on every date — # that per-day comparison is exactly what made the payer # flip almost daily as spending shifted the lead back and # forth. `responsible.parents[0]` pays with FP across the # first half, `responsible.parents[1]` across the second. if d in first_half_dates: primary, primary_pool = parent_a, pool_a secondary, secondary_pool = parent_b, pool_b else: primary, primary_pool = parent_b, pool_b secondary, secondary_pool = parent_a, pool_a funded_pair = self._try_co_leave_pair(primary, primary_pool, secondary, d) if funded_pair is None: # Graceful fallback: the chunk's designated FP-payer # can't afford THIS date (or their partner can't pay # vacation) — try the pairing swapped for just this # date rather than losing "both off" for it entirely. # Because the exhausted resource (FP or vacation) stays # exhausted, once this triggers it keeps triggering for # the rest of the chunk instead of flip-flopping day to # day. funded_pair = self._try_co_leave_pair(secondary, secondary_pool, primary, d) if funded_pair is not None: fp_parent, fp_tier, vacation_parent = funded_pair cost[fp_tier] += 1 cost["vacation"] += 1 fp_reason = ( REASON_FP_INTERIOR if fp_tier == "sjukpenning" else REASON_FP_LAGSTA_STRETCH ) days.append( ProposedDay(d, fp_parent.uuid, child.id, fp_tier, False, wish.id, fp_reason) ) days.append( ProposedDay( d, vacation_parent.uuid, child.id, "vacation", False, wish.id, REASON_VACATION_WRAP_WEEKEND, ) ) co_leave_dates.add(d) funded = True if not funded: leftover_for_swap.append(d) self._dubbeldagar_used[child.id] = used swap_dates: set[date] = set() if leftover_for_swap: swap_days, swap_cost = self._fund_swap( wish, w, sorted(leftover_for_swap), red_cluster, responsible ) days.extend(swap_days) for k in cost: cost[k] += swap_cost[k] swap_dates = {sd.date for sd in swap_days} meta = { "dubbeldag_dates": len(dubbeldag_dates), "co_leave_dates": len(co_leave_dates), "swap_dates": len(swap_dates), "dubbeldagar_legally_available": dubbeldagar_legally_available, } return days, cost, meta # -- per-window orchestration (§3.3-3.6) ------------------------------- def fund_window(self, w: _WishWindow) -> tuple[list[ProposedDay], WindowOutcome]: wish = w.wish responsible = self._resolve_parents(wish) # §remove-this-day — the window's dates minus the ones the user # removed outright (see `_excluded_window_dates`, which also explains # why an excluded date shrinks the target instead of counting as a # shortfall). The red-day cluster is still scanned over the FULL # range: klämdag detection is a property of the calendar, not of what # this window happens to fund. all_dates = self._window_dates(wish, responsible, w.start, w.end) red_cluster = _red_day_cluster_dates( w.start, w.end, self._input.holidays, self._input.day_overrides ) dubbeldagar_legally_available = False strategy_degraded = False if responsible.mode == "dubbeldag": days, cost, meta = self._fund_both(wish, w, all_dates, red_cluster, responsible) dubbeldagar_legally_available = bool(meta["dubbeldagar_legally_available"]) requested_strategy = "dubbeldag" if wish.wants_dubbeldag else STRATEGY_CO_LEAVE date_counts = { "dubbeldag": int(meta["dubbeldag_dates"]), "co_leave": int(meta["co_leave_dates"]), "swap": int(meta["swap_dates"]), } if any(date_counts.values()): # Dominant category by DATE count; ties broken by how # "good" the outcome is (dubbeldag > co_leave > swap) so a # mixed window never under-reports what mostly happened. _tier_rank = {"dubbeldag": 2, "co_leave": 1, "swap": 0} strategy = max(date_counts, key=lambda k: (date_counts[k], _tier_rank[k])) else: # Nothing needed real funding (e.g. an all-free-days window) # -- nothing degraded, report the requested strategy as-is. strategy = requested_strategy strategy_degraded = bool(date_counts["swap"]) or ( requested_strategy == "dubbeldag" and date_counts["co_leave"] > 0 ) elif responsible.mode == "swap": days, cost = self._fund_swap(wish, w, all_dates, red_cluster, responsible) strategy = "parent_swap" requested_strategy = "parent_swap" else: parent = responsible.parents[0] resolved_child = self._resolve_wish_child(wish) if resolved_child is not None: self._maybe_clamp_age4((parent.uuid, resolved_child.id), w.start) days, cost = self._fund_solo(wish, all_dates, red_cluster, parent, None) strategy = "solo" requested_strategy = "solo" shortfall_days, weeks_funded = _shortfall_and_weeks_funded( responsible.mode, all_dates, days, cost ) # §todo item (red-day shift toast) — a flexible window's own bound # search already IS the shift search (surfaced via # `w.shift_gain_days`/the toast), so suggesting a placement outside # the user's stated bound would contradict the flexible contract. # A fixed window never auto-shifts, so it needs the suggestion # instead — and gets it even when fully funded (shortfall_days == 0). remedies = self._build_remedies( wish, w, shortfall_days, include_shift_remedy=(wish.window_mode == "fixed") ) outcome = WindowOutcome( wish_id=wish.id, wish_name=wish.name, wish_name_message_key=wish.name_message_key, year=w.anchor_year, start_date=w.start, end_date=w.end, weeks_wished=wish.target_weeks, weeks_funded=weeks_funded, shortfall_days=shortfall_days, strategy=strategy, requested_strategy=requested_strategy, strategy_degraded=strategy_degraded, dubbeldagar_legally_available=dubbeldagar_legally_available, message_key=MESSAGE_KEY_WINDOW_EXPLANATION, data={ "wish_name": wish.name, "wish_name_message_key": wish.name_message_key, "year": w.anchor_year, "weeks_wished": wish.target_weeks, "weeks_funded": weeks_funded, "shortfall_days": shortfall_days, }, cost_breakdown=cost, remedies=remedies, shift_gain_days=w.shift_gain_days, series_id=wish.series_id, ) return days, outcome # -- shortfall remedies (§3.6, §9.1) ------------------------------------ def _build_remedies( self, wish: LeaveWish, w: _WishWindow, shortfall_days: int, include_shift_remedy: bool = True, ) -> list[ShortfallRemedy]: remedies: list[ShortfallRemedy] = [] if shortfall_days > 0: remedies.extend(self._reduce_preschool_reserve_remedy(wish)) remedies.extend(self._carry_vacation_forward_remedy(wish, w)) remedies.extend(self._lower_priority_wish_remedy(wish)) # §10.4 — `include_shift_remedy=False` from `build_unplaced_outcome`: # there `w` is the whole bounding range, not a single placed window, # so there is no one winning edge to shift and no point running the # (now real, `_SHIFT_REMEDY_SEARCH_BOUND_DAYS`-wide) search only to # discard its result. `fund_window` also passes False for a flexible # window (its own bound search already IS the shift), so this branch # is reached only for a fixed window — deliberately NOT gated on # `shortfall_days`, so a fully-funded fixed window still surfaces # "moving this frees N days" instead of hiding it behind a shortfall. if include_shift_remedy: remedies.extend(self._shift_window_into_free_days_remedy(wish, w)) return remedies def _reduce_preschool_reserve_remedy(self, wish: LeaveWish) -> list[ShortfallRemedy]: remedy_child = self._resolve_wish_child(wish) if remedy_child is None: return [] reserved = self._preschool_reserved.get(remedy_child.id, 0) if reserved <= 0: return [] return [ ShortfallRemedy( kind=REMEDY_REDUCE_PRESCHOOL_RESERVE, message_key=f"planRemedies.{REMEDY_REDUCE_PRESCHOOL_RESERVE}", data={"reserved": reserved}, days_freed=reserved, ) ] def _carry_vacation_forward_remedy( self, wish: LeaveWish, w: _WishWindow ) -> list[ShortfallRemedy]: responsible = self._resolve_parents(wish) # §9.2 Task 1 consequence #2: only days *above* the parent's # semesterlagen minimum for that year may be carried forward — the # minimum itself must stay taken, not banked. (The 5-year expiry is # a separate constraint, applied elsewhere.) spare_total = 0 for (p_uuid, year), budget in self._vacation.items(): if year >= w.start.year or not any(p_uuid == p.uuid for p in responsible.parents): continue parent = self._parents_by_id.get(p_uuid) min_days = parent.min_vacation_days_per_year if parent else 0 bankable_ceiling = max(budget.total_days - min_days, 0) spare_total += min(max(budget.available, 0), bankable_ceiling) if spare_total <= 0: return [] return [ ShortfallRemedy( kind=REMEDY_CARRY_VACATION_FORWARD, message_key=f"planRemedies.{REMEDY_CARRY_VACATION_FORWARD}", data={"spare_total": spare_total}, days_freed=spare_total, ) ] def _lower_priority_wish_remedy(self, wish: LeaveWish) -> list[ShortfallRemedy]: if wish.child_id is None: return [] siblings = [ other for other in self._input.wishes if other.id != wish.id and other.child_id == wish.child_id and other.priority > wish.priority ] if not siblings: return [] other = min(siblings, key=lambda o: o.priority) # A whole-week target can only be halved to a genuinely # smaller whole week count when it's at least 2 weeks to # begin with (`target_weeks` has a domain floor of 1 — see # `LeaveWishCreate.target_weeks`). At exactly 1 week, # `target_weeks // 2 == 0`, which used to get clamped back # up to 1, producing a nonsensical "from 1 to 1 weeks" # no-op remedy. Suppress the suggestion entirely rather # than claim a change that isn't one — the planner works in # whole days/weeks, so there's no fractional week to offer # instead. half_weeks = other.target_weeks // 2 if half_weeks < 1: return [] other_start, other_end = other.window() other_dates = _dates_in(other_start, other_end) spendable = sum( 1 for d in other_dates if not is_free_day(d, self._input.holidays, self._input.day_overrides) ) freed_estimate = round(spendable / 2) if freed_estimate <= 0: return [] return [ ShortfallRemedy( kind=REMEDY_LOWER_PRIORITY_WISH, message_key=f"planRemedies.{REMEDY_LOWER_PRIORITY_WISH}", data={ "wish_name": other.name, "wish_name_message_key": other.name_message_key, "from_weeks": other.target_weeks, "to_weeks": half_weeks, "freed_days": freed_estimate, }, days_freed=freed_estimate, ) ] def _shift_window_into_free_days_remedy( self, wish: LeaveWish, w: _WishWindow ) -> list[ShortfallRemedy]: """§todo item 13, part 2 — a real search over the surrounding weeks for a cheaper placement of this exact-length window, not just the 3-days-either-side nudge this used to be. The motivating case is a moving public holiday (Easter): a wish fixed at the same month/day every year drifts further from Easter in some years than others, and a placement a couple of weeks away can be strictly cheaper. Reuses `_candidate_cost` — the same "fewest spendable days" cost model `_rank_candidate`/`resolve_flexible_window` already use to score a flexible wish's placement — rather than inventing a second scorer. Candidates are every same-length window inside `[w.start - _SHIFT_REMEDY_SEARCH_BOUND_DAYS, w.end + _SHIFT_REMEDY_SEARCH_BOUND_DAYS]` (via `_candidate_windows`, the same sliding-window generator flexible resolution uses), ranked by (cost, distance from the current placement, start date) and reported only when strictly cheaper than the window actually funded. """ length_days = (w.end - w.start).days + 1 current_cost = _candidate_cost( w.start, w.end, self._input.holidays, self._input.day_overrides ) if current_cost <= 0: return [] # already all free days -- nowhere cheaper to move to bound_start = w.start - timedelta(days=_SHIFT_REMEDY_SEARCH_BOUND_DAYS) bound_end = w.end + timedelta(days=_SHIFT_REMEDY_SEARCH_BOUND_DAYS) scored: list[tuple[int, int, date, date]] = [] for c_start, c_end in _candidate_windows(bound_start, bound_end, length_days): if c_start == w.start and c_end == w.end: continue c_cost = _candidate_cost( c_start, c_end, self._input.holidays, self._input.day_overrides ) if c_cost < current_cost: scored.append((c_cost, abs((c_start - w.start).days), c_start, c_end)) if not scored: return [] scored.sort(key=lambda t: (t[0], t[1], t[2])) remedies: list[ShortfallRemedy] = [] for c_cost, _distance, c_start, c_end in scored[:_SHIFT_REMEDY_MAX_SUGGESTIONS]: shift_gain = current_cost - c_cost remedies.append( ShortfallRemedy( kind=REMEDY_SHIFT_WINDOW_INTO_FREE_DAYS, message_key=f"planRemedies.{REMEDY_SHIFT_WINDOW_INTO_FREE_DAYS}", data={ "wish_name": wish.name, "wish_name_message_key": wish.name_message_key, "shift_gain": shift_gain, "new_start_date": c_start.isoformat(), "new_end_date": c_end.isoformat(), }, days_freed=shift_gain, ) ) return remedies # -- flexible ("när det passar bäst") wish resolution (§10.2) ---------- def record_days(self, days: list[ProposedDay]) -> None: """§10.2 "Occupied dates" — fold a just-funded window's proposed days into the same lookup `_fund_solo`/`_fund_both` already consult (`_existing_lookup`, today: persisted `existing_planned` rows only), so a later window (fixed or flexible) in this same `propose()` run never lands a flexible candidate on a date another wish already claimed. Called by `Allocator.propose()` after every `fund_window()` call, fixed or flexible. """ for d in days: self._existing_lookup[(d.parent_personal_data_uuid, d.child_id, d.date)] = d.day_type def _candidate_occupied( self, start: date, end: date, parents: list[ParentFacts], child_id: UUID | None, exclude_dates: frozenset[date] = frozenset(), ) -> bool: """§10.2 step 5 — whether `[start, end]` collides with an already-claimed date for any of `parents` (paired with `child_id`). `exclude_dates` (§fix) lets a caller that already knows some of those claimed dates are this very wish's own prior occurrence treat them as expected, not a collision. """ for d in _dates_in(start, end): if d in exclude_dates: continue for p in parents: if (p.uuid, child_id, d) in self._existing_lookup: return True return False def _persisted_dates_for_wish( self, wish: LeaveWish, child_id: UUID | None, bound_start: date, bound_end: date, ) -> list[date]: """§fix — dates already persisted *before this `propose()` run* (i.e. from a prior apply, via `_persisted_by_wish` — never the mutable `_existing_lookup`) for THIS wish, inside its bound. The read-only source is what keeps a fixed wish's in-session placement from being mistaken for a flexible wish's own history (§10.3, `TestFixedWishNeverDisplacedByFlexible`). Matched by `wish_id`, deliberately NOT by the wish's currently resolved parent(s) (§fix, decision B): a wish's prior placement must still be found after the user reassigns its parent, so the caller can reuse the SAME window rather than researching one from scratch purely because a different parent has different (or no) occupied dates — see `AllocationInput.all_existing_planned`'s docstring. """ dates = self._persisted_by_wish.get((wish.id, child_id), set()) return sorted(d for d in dates if bound_start <= d <= bound_end) def _reused_window_carries_manual_edit( self, wish: LeaveWish, responsible: _Responsible, child_id: UUID | None, persisted_dates: list[date], ) -> bool: """§fix (Step 3) — whether any of this wish-year's already-persisted `persisted_dates` carries a manual edit that must never be silently moved by the realignment escape hatch in `_reuse_already_funded_window`: a pinned day (`is_pinned`, locked in place by the user) or a day the user explicitly removed (§remove-this-day, `_is_excluded`). Both kinds are already respected elsewhere in the ordinary funding path (`_candidate_occupied`/`_persisted_dates_for_wish` keep this wish's own claimed dates out of collision checks, `_is_excluded` keeps a removed date out of a freshly funded window) — this is what keeps the new escape hatch from re-searching its way around either one. """ pinned = self._pinned_dates_by_wish.get((wish.id, child_id), frozenset()) if any(d in pinned for d in persisted_dates): return True return any( self._is_excluded(p.uuid, child_id, d) for d in persisted_dates for p in responsible.parents ) # -- §Decision C — budget-aware candidate ranking ---------------------- def _candidate_shortfall( self, wish: LeaveWish, responsible: _Responsible, candidate_start: date, candidate_end: date, ) -> int: """§10.2 step 6, extended — how many of a candidate window's spendable days the resolved parent(s) could NOT actually pay for, simulated against a throwaway copy of the live pool/vacation state. `_candidate_cost` alone only counts calendar days — it has no idea whether the wish's resolved parent(s) can actually AFFORD them. A window that is cheap by the calendar but leaves half its days unfunded (vacation/FP already exhausted) is worse than a slightly costlier one the parent(s) can fully pay for, so this must be compared FIRST, with `_candidate_cost` only as the tie-break — see `resolve_flexible_window`/`_reuse_already_funded_window`, which both rank candidates by `(shortfall, calendar_cost, start, wish.id)`. Runs the SAME `_fund_solo`/`_fund_swap`/`_fund_both` funding logic every real window goes through, against a scratch copy of `self._pools`/`self._vacation`/`self._dubbeldagar_used` swapped in for the duration of the call and restored in a `finally` — `propose()` must stay read-only and deterministic (§5 test #11), so no candidate probe may leave a trace on the real ledger, win or lose. The winning candidate is funded for real afterwards through the ordinary `fund_window()` path, exactly as before this fix. """ dates = self._window_dates(wish, responsible, candidate_start, candidate_end) red_cluster = _red_day_cluster_dates( candidate_start, candidate_end, self._input.holidays, self._input.day_overrides ) probe_window = _WishWindow( wish=wish, anchor_year=candidate_start.year, start=candidate_start, end=candidate_end, ) real_pools, real_vacation = self._pools, self._vacation real_dubbeldagar_used = dict(self._dubbeldagar_used) self._pools = { key: _MutablePool( p.sjukpenning_left, p.lagsta_left, p.dubbeldagar_left, p.clamped_at_b4 ) for key, p in real_pools.items() } self._vacation = { key: _MutableVacation(v.total_days, v.saved_days_in, v.already_used, v.is_assumed) for key, v in real_vacation.items() } try: if responsible.mode == "dubbeldag": days, cost, _meta = self._fund_both( wish, probe_window, dates, red_cluster, responsible ) elif responsible.mode == "swap": days, cost = self._fund_swap(wish, probe_window, dates, red_cluster, responsible) else: parent = responsible.parents[0] resolved_child = self._resolve_wish_child(wish) if resolved_child is not None: self._maybe_clamp_age4((parent.uuid, resolved_child.id), candidate_start) days, cost = self._fund_solo(wish, dates, red_cluster, parent, None) finally: self._pools, self._vacation = real_pools, real_vacation self._dubbeldagar_used = real_dubbeldagar_used shortfall_days, _weeks_funded = _shortfall_and_weeks_funded( responsible.mode, dates, days, cost ) return shortfall_days def _rank_candidate( self, wish: LeaveWish, responsible: _Responsible, candidate: tuple[date, date], ) -> tuple[int, int, int, date, int]: """Sort key shared by `resolve_flexible_window` and `_reuse_already_funded_window`: fewest unfundable days first (§Decision C), THEN calendar cost, THEN Monday-start alignment (§Decision 2), THEN start date, THEN wish id — the existing priority ordering (§10.2 step 6) is otherwise unchanged. The alignment term sits strictly after `cost` and strictly before `start`: after `cost` so red days always win a candidate — a cheaper non-Monday start beats a pricier Monday one every time; before `start` so that among equally-cheap candidates (the degenerate case where the bound is exactly `length_days` long, or any other tie) a Monday start wins over "earliest", which is what made flexible placements land on an arbitrary, year-to-year-inconsistent weekday. """ start, end = candidate shortfall = self._candidate_shortfall(wish, responsible, start, end) cost = _candidate_cost(start, end, self._input.holidays, self._input.day_overrides) misaligned = 0 if start.weekday() == _PREFERRED_START_WEEKDAY else 1 return (shortfall, cost, misaligned, start, wish.id.int) def _reuse_already_funded_window( self, wish: LeaveWish, responsible: _Responsible, child_id: UUID | None, bound_start: date, bound_end: date, length_days: int, ) -> tuple[date, date] | None: """§fix — the double-booking-on-repropose fix, product-owner approved: if this flexible wish-year already has planned days from a prior apply inside its bounding range, reuse the full-length window they belong to instead of searching for a new one. The caller then funds that window through the ordinary `fund_window()` path exactly like any other window; because `_fund_solo`/`_fund_swap`/`_fund_both` all skip emitting a new `ProposedDay` for a date `_existing_lookup` already has, this single mechanism handles both "fully funded" (every spendable day in the window is already persisted -> zero new days) and "partially funded" (only some are -> the remainder is topped up) without a separate code path for either. Known and accepted limitation (§10.2's "occupied dates" tradeoff, now extended): a planned day belonging to a *different* wish that happens to fall in this same bounding range cannot be told apart from this wish's own prior occurrence, and may be mistaken for it. Returns `None` — falls through to a fresh search — when: nothing is persisted for this wish-year yet, the persisted dates don't fit inside one `length_days` window (they can't be a single occurrence of *this* wish), or every window that would contain them now collides with something else already claimed this run (e.g. a higher-priority fixed wish added since the last apply). That last case is what keeps a genuinely-unsatisfied year free to move (§10.3) rather than frozen to a window that no longer fits. §fix (Step 3, flexible-window realignment escape hatch) — also returns `None` when the winner picked below is misaligned (does not start on `_PREFERRED_START_WEEKDAY`) AND a fresh, unoccupied search across the *whole* bound (not just windows containing the persisted dates — exactly what `resolve_flexible_window` runs next when this returns `None`) would find a candidate whose `(shortfall, cost, misaligned)` rank prefix is strictly better. This is what lets a wish-year placed before §Decision 2 existed (or before its bound moved) self-heal onto a Monday on the next `propose()`, instead of being frozen forever to whichever weekday it first landed on. Never fires when any of the persisted dates carry a manual edit — a pinned day or one the user explicitly removed (see `_reused_window_carries_manual_edit`) — so a deliberate per-day override always outlives a realignment. """ persisted_dates = self._persisted_dates_for_wish(wish, child_id, bound_start, bound_end) if not persisted_dates: return None span = (persisted_dates[-1] - persisted_dates[0]).days + 1 if span > length_days: return None own_dates = frozenset(persisted_dates) candidates = [ (s, e) for s, e in _candidate_windows(bound_start, bound_end, length_days) if s <= persisted_dates[0] and persisted_dates[-1] <= e and not self._candidate_occupied(s, e, responsible.parents, child_id, own_dates) ] if not candidates: return None winner = min(candidates, key=lambda c: self._rank_candidate(wish, responsible, c)) winner_start, _winner_end = winner if winner_start.weekday() == _PREFERRED_START_WEEKDAY: return winner if self._reused_window_carries_manual_edit(wish, responsible, child_id, persisted_dates): return winner fresh_candidates = [ (s, e) for s, e in _candidate_windows(bound_start, bound_end, length_days) if not self._candidate_occupied(s, e, responsible.parents, child_id) ] if fresh_candidates: fresh_winner = min( fresh_candidates, key=lambda c: self._rank_candidate(wish, responsible, c) ) if ( self._rank_candidate(wish, responsible, fresh_winner)[:3] < self._rank_candidate(wish, responsible, winner)[:3] ): return None return winner def resolve_flexible_window( self, wish: LeaveWish, anchor_year: int, bound_start: date, bound_end: date ) -> _WishWindow | None: """§10.2 steps 2-9 — resolve one flexible wish-year's bound to a concrete window to fund. Returns `None` when the bound is long enough but every full-length candidate collides with an already-claimed date — the 'unplaced' case (§10.4), for the caller to report via `build_unplaced_outcome`. """ length_days = wish.target_span_days() bound_length = (bound_end - bound_start).days + 1 if bound_length < length_days: # Step 4: horizon clipping ate into the bound -> there is only one # candidate, the clipped bound itself, funded as an ordinary # partial shortfall (§3.6) — NOT the 'unplaced' case. return _WishWindow(wish=wish, anchor_year=anchor_year, start=bound_start, end=bound_end) responsible = self._resolve_parents(wish) resolved_child = self._resolve_wish_child(wish) child_id = resolved_child.id if resolved_child is not None else None # §fix — an already (fully or partially) funded wish-year reuses its # own prior window rather than being searched for fresh, which is # what caused the double-booking regression. reused = self._reuse_already_funded_window( wish, responsible, child_id, bound_start, bound_end, length_days ) if reused is not None: winner_start, winner_end = reused return _WishWindow( wish=wish, anchor_year=anchor_year, start=winner_start, end=winner_end ) candidates = [ (s, e) for s, e in _candidate_windows(bound_start, bound_end, length_days) if not self._candidate_occupied(s, e, responsible.parents, child_id) ] if not candidates: return None winner_start, winner_end = min( candidates, key=lambda c: self._rank_candidate(wish, responsible, c) ) # §todo item (red-day shift toast) — how many spendable days the # search saved versus the naive "first slot in the bound" placement, # so the frontend can tell the user their window moved to catch red # days. Compared only against the bound-start placement, not the # previous placement across re-proposes — see `_WishWindow # .shift_gain_days`'s docstring for why the reuse path leaves this 0. naive_start = bound_start naive_end = bound_start + timedelta(days=length_days - 1) shift_gain_days = 0 if (winner_start, winner_end) != (naive_start, naive_end): naive_cost = _candidate_cost( naive_start, naive_end, self._input.holidays, self._input.day_overrides ) winner_cost = _candidate_cost( winner_start, winner_end, self._input.holidays, self._input.day_overrides ) shift_gain_days = max(naive_cost - winner_cost, 0) return _WishWindow( wish=wish, anchor_year=anchor_year, start=winner_start, end=winner_end, shift_gain_days=shift_gain_days, ) def build_unplaced_outcome( self, wish: LeaveWish, anchor_year: int, bound_start: date, bound_end: date ) -> WindowOutcome: """§10.4 — a flexible wish-year that could not be placed at all: every full-length candidate inside a long-enough bound collided with a date another wish already claimed. Reuses the shortfall vocabulary (`WindowOutcome`) rather than inventing a new failure mode; the only new value is `strategy = 'unplaced'`. """ length_days = wish.target_span_days() bound_window = _WishWindow( wish=wish, anchor_year=anchor_year, start=bound_start, end=bound_end ) # §10.4: every remedy `_build_remedies` can generate still makes sense # against the clipped bound EXCEPT shift_window_into_free_days — there # is no single winning window edge left to shift (`bound_window` # spans the whole bound, not one placement), so skip computing it # rather than discarding its result after the fact. remedies = self._build_remedies(wish, bound_window, length_days, include_shift_remedy=False) responsible = self._resolve_parents(wish) if responsible.mode == "dubbeldag": requested_strategy = "dubbeldag" if wish.wants_dubbeldag else STRATEGY_CO_LEAVE elif responsible.mode == "swap": requested_strategy = "parent_swap" else: requested_strategy = "solo" return WindowOutcome( wish_id=wish.id, wish_name=wish.name, wish_name_message_key=wish.name_message_key, year=anchor_year, start_date=bound_start, end_date=bound_end, weeks_wished=wish.target_weeks, weeks_funded=0.0, shortfall_days=length_days, strategy=STRATEGY_UNPLACED, requested_strategy=requested_strategy, strategy_degraded=False, dubbeldagar_legally_available=False, message_key=MESSAGE_KEY_WINDOW_EXPLANATION, data={ "wish_name": wish.name, "wish_name_message_key": wish.name_message_key, "year": anchor_year, "weeks_wished": wish.target_weeks, "weeks_funded": 0.0, "shortfall_days": length_days, }, cost_breakdown=_new_cost(), remedies=remedies, series_id=wish.series_id, ) # -- warnings ------------------------------------------------------------ def build_warnings(self, all_days: list[ProposedDay]) -> list[PlanWarning]: warnings: list[PlanWarning] = [] warnings.extend(self._age4_warnings(all_days)) warnings.extend(self._sgi_warnings(all_days)) warnings.extend(self._vacation_expiry_warnings(all_days)) return warnings def _age4_projections( self, all_days: list[ProposedDay] ) -> dict[tuple[UUID, UUID], tuple[int, int]]: """(parent, child) -> (projected_left_at_b4, expiring_at_b4). Shared by `_age4_warnings` and `build_surplus_advice` (§9.2) so the warning and the suggestions that mirror it are always computed from the same numbers. Reuses the pure arithmetic already extracted for AnalyticsService (`app.utils.tip_predicates.days_left_at_age_4_projection`) rather than reimplementing the age-4 formula here. """ result: dict[tuple[UUID, UUID], tuple[int, int]] = {} for (parent_uuid, child_id), original_pool in self._input.pools.items(): child = self._children_by_id.get(child_id) if child is None: continue b4 = child.fourth_birthday total_initial = original_pool.sjukpenning_left + original_pool.lagsta_left existing_before = sum( 1 for d in self._input.existing_planned if d.parent_personal_data_uuid == parent_uuid and d.child_id == child_id and d.day_type in ("sjukpenning", "lagsta") and d.date < b4 ) new_before = sum( 1 for d in all_days if d.parent_personal_data_uuid == parent_uuid and d.child_id == child_id and d.day_type in ("sjukpenning", "lagsta") and d.date < b4 ) result[(parent_uuid, child_id)] = days_left_at_age_4_projection( days_used=0, planned_fp_before_b4=existing_before + new_before, total_days=total_initial, max_savable=self._input.age4_savable_cap, ) return result def _age4_warnings(self, all_days: list[ProposedDay]) -> list[PlanWarning]: warnings: list[PlanWarning] = [] for (parent_uuid, child_id), ( projected_left, expiring, ) in self._age4_projections(all_days).items(): if expiring <= 0: continue child = self._children_by_id[child_id] b4 = child.fourth_birthday warnings.append( PlanWarning( kind=WARNING_DAYS_LOST_AT_AGE_4, message_key=f"planWarnings.{WARNING_DAYS_LOST_AT_AGE_4}", data={ "parent_personal_data_uuid": parent_uuid, "child_id": child_id, "expiring_days": expiring, "projected_left_at_b4": projected_left, "fourth_birthday": b4.isoformat(), }, ) ) return warnings def build_age4_projections(self, all_days: list[ProposedDay]) -> list[Age4Projection]: """§3.3 part 2 — the plan-aware age-4 projection for every (parent, child) pair, always populated (unlike `build_surplus_advice`, which is empty when there's nothing to advise on) so the UI can show what this proposal implies at the age-4 boundary before it's applied. """ projections: list[Age4Projection] = [] for (parent_uuid, child_id), (projected_left, expiring) in sorted( self._age4_projections(all_days).items(), key=lambda kv: (str(kv[0][0]), str(kv[0][1])) ): child = self._children_by_id.get(child_id) if child is None: continue savable = min(projected_left, self._input.age4_savable_cap) projections.append( Age4Projection( parent_personal_data_uuid=parent_uuid, child_id=child_id, fourth_birthday=child.fourth_birthday, projected_left_at_b4=projected_left, savable_after_b4=savable, expiring_at_b4=expiring, savable_cap_at_b4=self._input.age4_savable_cap, ) ) return projections # -- surplus advice (§9.2) — the mirror of a shortfall remedy -------- def build_surplus_advice(self, all_days: list[ProposedDay]) -> list[SurplusAdvice]: """Ranked, concrete suggestions for parents forecast to *forfeit* FP days at the age-4 boundary — the mirror of `_build_remedies`' shortfall handling. Empty when no (parent, child) pair has a surplus. """ advice: list[SurplusAdvice] = [] projections = self._age4_projections(all_days) for (parent_uuid, child_id), (_projected_left, expiring) in sorted( projections.items(), key=lambda kv: (str(kv[0][0]), str(kv[0][1])) ): if expiring <= 0: continue parent = self._parents_by_id.get(parent_uuid) child = self._children_by_id.get(child_id) if parent is None or child is None: continue advice.extend(self._wish_more_leave_advice(parent, child, expiring)) advice.extend(self._shift_mix_to_fp_advice(parent, child, expiring, all_days)) advice.extend(self._front_load_advice(parent, child, expiring)) advice.extend(self._use_less_vacation_advice(parent, child, expiring)) return advice def _wish_more_leave_advice( self, parent: ParentFacts, child: ChildFacts, expiring: int ) -> list[SurplusAdvice]: weeks = expiring // _SPENDABLE_DAYS_PER_WEEK if weeks < 1: return [] return [ SurplusAdvice( kind=SURPLUS_WISH_MORE_LEAVE, message_key=f"planSurplusAdvice.{SURPLUS_WISH_MORE_LEAVE}", days_involved=expiring, data={ "parent_personal_data_uuid": parent.uuid, "child_id": child.id, "weeks": weeks, }, ) ] def _shift_mix_to_fp_advice( self, parent: ParentFacts, child: ChildFacts, expiring: int, all_days: list[ProposedDay], ) -> list[SurplusAdvice]: # Vacation days spent in the goal-dependent interior (tier 3, §3.3) — # not a red-day-cluster wrap (tier 2, which stays vacation regardless # of goal) — before this child's 4th birthday. These are the days # that could instead have drawn on the (about-to-expire) FP pool, # freeing semester for `use_less_vacation` or later years. swappable = sum( 1 for d in all_days if d.parent_personal_data_uuid == parent.uuid and d.child_id == child.id and d.day_type == "vacation" and d.reason == REASON_VACATION_WRAP_WEEKEND and d.date < child.fourth_birthday ) days_involved = min(swappable, expiring) if days_involved <= 0: return [] return [ SurplusAdvice( kind=SURPLUS_SHIFT_MIX_TO_FP, message_key=f"planSurplusAdvice.{SURPLUS_SHIFT_MIX_TO_FP}", days_involved=days_involved, data={"parent_personal_data_uuid": parent.uuid, "child_id": child.id}, ) ] def _front_load_advice( self, parent: ParentFacts, child: ChildFacts, expiring: int ) -> list[SurplusAdvice]: if expiring < _LARGE_SURPLUS_THRESHOLD_DAYS: return [] still_small = self._input.today < child.birth_date + relativedelta( years=_FRONT_LOAD_CHILD_AGE_YEARS ) if not still_small: return [] return [ SurplusAdvice( kind=SURPLUS_FRONT_LOAD_EARLY_YEARS, message_key=f"planSurplusAdvice.{SURPLUS_FRONT_LOAD_EARLY_YEARS}", days_involved=expiring, data={"parent_personal_data_uuid": parent.uuid, "child_id": child.id}, ) ] def _use_less_vacation_advice( self, parent: ParentFacts, child: ChildFacts, expiring: int ) -> list[SurplusAdvice]: # Never advise below the semesterlagen minimum (Task 1) — only the # portion of *already-planned* vacation usage above the floor is # reducible, and only up to what the expiring FP surplus could cover. remaining = expiring total_reducible = 0 years_hit: list[int] = [] for year in sorted( y for (p_uuid, y) in self._vacation if p_uuid == parent.uuid and y < child.fourth_birthday.year ): if remaining <= 0: break if self._input.vacation_forfeiting.get((parent.uuid, year), 0) > 0: # §Task 2 reconciliation — never advise reducing vacation use # in a year where that same use is exactly what # `_assign_tier3`'s forfeiture bias just spent to PREVENT a # loss. Advising "use less" there would contradict what this # same proposal already did (and correctly did) for that # year: the two pieces of advice would otherwise both fire # for the same (parent, year), pulling in opposite # directions. continue budget = self._vacation[(parent.uuid, year)] reducible = max(budget.already_used - parent.min_vacation_days_per_year, 0) take = min(reducible, remaining) if take > 0: total_reducible += take remaining -= take years_hit.append(year) if total_reducible <= 0: return [] years_text = ", ".join(str(y) for y in years_hit) return [ SurplusAdvice( kind=SURPLUS_USE_LESS_VACATION, message_key=f"planSurplusAdvice.{SURPLUS_USE_LESS_VACATION}", days_involved=total_reducible, data={ "parent_personal_data_uuid": parent.uuid, "child_id": child.id, "years_text": years_text, "min_vacation_days_per_year": parent.min_vacation_days_per_year, }, ) ] def _sgi_warnings(self, all_days: list[ProposedDay]) -> list[PlanWarning]: warnings: list[PlanWarning] = [] by_parent_week: dict[tuple[UUID, tuple[int, int]], int] = {} for d in all_days: if d.day_type not in ("sjukpenning", "lagsta"): continue iso_year, iso_week, _ = d.date.isocalendar() key = (d.parent_personal_data_uuid, (iso_year, iso_week)) by_parent_week[key] = by_parent_week.get(key, 0) + 1 flagged: set[UUID] = set() for (parent_uuid, (iso_year, iso_week)), count in sorted(by_parent_week.items()): if parent_uuid in flagged or count >= _SGI_MIN_DAYS_PER_WEEK: continue parent = self._parents_by_id.get(parent_uuid) if parent is None or parent.employment_status in ("employed", "jobseeking"): continue week_monday = date.fromisocalendar(iso_year, iso_week, 1) relevant_child = next( ( c for c in self._input.children if week_monday >= c.birth_date + relativedelta(years=1) and (parent_uuid, c.id) in self._input.pools ), None, ) if relevant_child is None: continue flagged.add(parent_uuid) warnings.append( PlanWarning( kind=WARNING_SGI_RISK, message_key=f"planWarnings.{WARNING_SGI_RISK}", data={ "parent_personal_data_uuid": parent_uuid, "week_start": week_monday.isoformat(), "days_planned": count, "min_days_per_week": _SGI_MIN_DAYS_PER_WEEK, }, ) ) return warnings def _vacation_expiry_warnings(self, all_days: list[ProposedDay]) -> list[PlanWarning]: """§Task 2 — the real forfeiture figure (`AllocationInput. vacation_forfeiting`, sourced from the same FIFO carry-forward simulation the Analytics tab uses) replaces the old `has_vacation_near_expiry` heuristic, which only ever checked "large total + large carry-in" and could both under- and over-warn relative to what actually gets lost. `vacation_forfeiting` itself is a forecast: the loader computes it from already-committed usage, before this run's own placements (including `_assign_tier3`'s forfeiture-avoidance bias — see its docstring above) are added. Left unadjusted, that forecast double- counts — `_assign_tier3` deliberately spends at-risk vacation days first specifically to avoid this forfeiture, so the naive figure warns about days the plan is actively saving. This method corrects for that here, at the one call site that sees every placed day, rather than re-deriving `AnalyticsService`'s full carry-forward/FIFO bank algorithm inside this pure `_Ledger` (still deliberately not duplicated — see `_assign_tier3`'s docstring): for each (parent, year) forecast to forfeit, it subtracts the number of `all_days` entries with `day_type == "vacation"` that this run placed for that parent with a `date.year` matching that year, and clamps the result at 0. This is sound *because* the bank is FIFO — a vacation day spent in year Y is always drawn from the oldest entry in that parent's bank, which is exactly the entry about to expire. So each vacation day placed in year Y retires one forecast-expiring day for that year one-for-one, up to the forecast itself; the clamp stops the count going negative once placements cover (or exceed) the forecast, and any placements beyond that draw from newer, non-expiring entries and are ignored. """ placed_vacation_by_parent_year: dict[tuple[UUID, int], int] = {} for d in all_days: if d.day_type != "vacation": continue key = (d.parent_personal_data_uuid, d.date.year) placed_vacation_by_parent_year[key] = placed_vacation_by_parent_year.get(key, 0) + 1 warnings: list[PlanWarning] = [] for (parent_uuid, year), forecast in sorted( self._input.vacation_forfeiting.items(), key=lambda kv: (str(kv[0][0]), kv[0][1]) ): placed = placed_vacation_by_parent_year.get((parent_uuid, year), 0) expiring = max(0, forecast - placed) if expiring <= 0: continue warnings.append( PlanWarning( kind=WARNING_VACATION_EXPIRING, message_key=f"planWarnings.{WARNING_VACATION_EXPIRING}", data={ "parent_personal_data_uuid": parent_uuid, "year": year, "expiring_days": expiring, }, ) ) return warnings # -- final snapshot -------------------------------------------------- def snapshot_balances(self) -> PlanBalances: pools_after = { key: DayPool(pool.sjukpenning_left, pool.lagsta_left, pool.dubbeldagar_left) for key, pool in self._pools.items() } vacation_after = { key: VacationBudget(v.total_days, v.saved_days_in, v.already_used, v.is_assumed) for key, v in self._vacation.items() } return PlanBalances(pools_after=pools_after, vacation_after=vacation_after) # --------------------------------------------------------------------------- # The pure Allocator # --------------------------------------------------------------------------- class Allocator: """Pure allocation engine. No `await`, no repository access, no `date.today()`. Given the same `AllocationInput`, `propose()` always returns a byte-identical `PlanProposal` (§5 test #11). """ def propose(self, alloc_input: AllocationInput) -> PlanProposal: ledger = _Ledger(alloc_input) all_days: list[ProposedDay] = [] outcomes: list[WindowOutcome] = [] # Pass 1 (§10.2, unchanged from §3.2/§3.3): every fixed-wish window is # placed first, unconditionally — a flexible wish can never displace # a fixed one. fixed_windows = _expand_all_windows(alloc_input) fixed_windows.sort(key=lambda w: (w.wish.priority, w.start, w.wish.id.int)) for w in fixed_windows: days, outcome = ledger.fund_window(w) all_days.extend(days) outcomes.append(outcome) ledger.record_days(days) # Pass 2 (§10.2, new): flexible wish-years, processed in # (priority, bound.start, wish.id) order, each searched for its # best-costing placement and funded into whatever space pass 1 left. flexible_wish_years = _expand_flexible_wish_years(alloc_input) flexible_wish_years.sort(key=lambda fw: (fw.wish.priority, fw.bound_start, fw.wish.id.int)) for fw in flexible_wish_years: if fw.fully_past: # §silent-failure gap fix -- the bound elapsed entirely # before `today`; there is nothing left to search for. outcomes.append( ledger.build_unplaced_outcome( fw.wish, fw.anchor_year, fw.bound_start, fw.bound_end ) ) continue window = ledger.resolve_flexible_window( fw.wish, fw.anchor_year, fw.bound_start, fw.bound_end ) if window is None: outcomes.append( ledger.build_unplaced_outcome( fw.wish, fw.anchor_year, fw.bound_start, fw.bound_end ) ) continue days, outcome = ledger.fund_window(window) all_days.extend(days) outcomes.append(outcome) ledger.record_days(days) warnings = ledger.build_warnings(all_days) surplus_advice = ledger.build_surplus_advice(all_days) age4_projections = ledger.build_age4_projections(all_days) balances = ledger.snapshot_balances() return PlanProposal( days=all_days, outcomes=outcomes, balances_after=balances, warnings=warnings, surplus_advice=surplus_advice, age4_projections=age4_projections, ) # --------------------------------------------------------------------------- # AllocationInputLoader — the ONLY part that touches repositories / the clock # --------------------------------------------------------------------------- def _to_child_facts(c: Child) -> ChildFacts: return ChildFacts( id=c.id, birth_date=c.birth_date, fourth_birthday=c.birth_date + relativedelta(years=4), preschool_start_date=c.preschool_start_date, preschool_planning_days_per_year=c.preschool_planning_days_per_year, ) def _compute_horizon_end(today: date, children: list[ChildFacts], wishes: list[LeaveWish]) -> date: """§9.1: max(last wish window end, latest fourth_birthday), capped at age-12. Every wish row is one concrete year now, so the horizon is simply the latest window end across them — no recurrence projection, and no "open-ended until the child turns 4" case to resolve here: that was resolved once, when `LeaveWishService` materialised the series. """ candidates: list[date] = [c.fourth_birthday for c in children] candidates.extend(wish.window()[1] for wish in wishes) if not candidates: return today + relativedelta(years=4) candidate_end = max(candidates) if children: hard_cap = max(c.birth_date + relativedelta(years=12) for c in children) candidate_end = min(candidate_end, hard_cap) return candidate_end class AllocationInputLoader: """Assembles a frozen `AllocationInput` from the repositories. This is the ONLY part of the planning engine that touches the database or the wall clock (via the optional `today` override, which tests should always pass explicitly). """ def __init__( self, workspace_repo: WorkspaceRepository, child_repo: ChildRepository, personal_data_repo: PersonalDataRepository, parental_leave_repo: ParentalLeaveRepository, vacation_repo: VacationDaysRepository, planned_leave_repo: PlannedLeaveRepository, leave_wish_repo: LeaveWishRepository, day_override_repo: DayOverrideRepository | None = None, preschool_planning_days_repo: PreschoolPlanningDaysRepository | None = None, day_assertion_repo: DayAssertionRepository | None = None, ) -> None: self._workspace_repo = workspace_repo self._children_repo = child_repo self._personal_data_repo = personal_data_repo self._parental_leave_repo = parental_leave_repo self._vacation_repo = vacation_repo self._planned_leave_repo = planned_leave_repo self._leave_wish_repo = leave_wish_repo self._day_override_repo = day_override_repo # §Task 1 (per-year preschool reserve) — optional, like # `day_override_repo` above: `None` (every caller that hasn't been # wired up yet, including every existing DI container until it's # updated to pass one) means "no per-year rows available", so # `load()` leaves `AllocationInput.preschool_reserve_by_year` empty # and `_preschool_reserve_days` falls back to the old flat # `per_year * years` lump sum unchanged. self._preschool_planning_days_repo = preschool_planning_days_repo # Restructure step 3b — `None` (an old test/script that hand-builds # a loader without one) means "no assertions available at all", so # `load()` produces empty `pinned_days`/`excluded_days`. Production # (`app.dependencies.get_allocation_input_loader`) always wires one: # `DayAssertion` is now the ONLY source for both signals — the # `is_pinned` column and the `planned_day_exclusion` table are gone # (migration 010). self._day_assertion_repo = day_assertion_repo async def load( self, workspace_id: UUID, today: date | None = None, *, only_pinned_existing: bool = False, ) -> AllocationInput: """Assemble the pure `AllocationInput`. `only_pinned_existing` (§Task 4, see `PlanningEngineService.propose`) restricts `existing_planned` — and everything derived from it here (pool/vacation netting) — to rows that must stay immovable: pinned rows (in `pinned_days`, see below), ad-hoc rows with no `wish_id`, and rows dated strictly before `today` ("the past is fact" — see the inline comment below; an elapsed date is never re-derived by the allocator regardless of this flag, so it must stay visible here or it would be counted as available capacity it has already spent). Every other (unpinned, wish-attributed, future-or-today) row is left out entirely, so the allocator treats that date as open and re-derives it fresh. Pinned rows are *never* netted out of the pools/vacation budgets here even when `only_pinned_existing` is False — the pure `_Ledger` does that itself (`_reserve_pinned_ days`), matching how it reserves preschool planning days off the top, so the debit happens in one place regardless of which mode loaded the input. Restructure step 3b — "pinned"/"excluded" both come from ONE read of `DayAssertion` (`self._day_assertion_repo`), split via `app.entities.day_assertion.partition_assertion_keys` into `AllocationInput.pinned_days`/`excluded_days`. `planned_leave_day` no longer carries an `is_pinned` column at all (migration 010) and `PlannedDayExclusion` is gone — `DayAssertion` is the sole source for both signals now, not a fallback layered on top of them. """ today = today or date.today() ws = await self._workspace_repo.get_by_id(workspace_id) if ws is None: raise NotFoundError(f"Workspace {workspace_id} not found") child_entities = await self._children_repo.list_by_workspace(workspace_id) parent_entities = await self._personal_data_repo.list_by_workspace(workspace_id) wishes = await self._leave_wish_repo.get_by_workspace(workspace_id) assertions = ( await self._day_assertion_repo.list_by_workspace(workspace_id) if self._day_assertion_repo is not None else [] ) pinned_days, excluded_days = partition_assertion_keys(assertions) children = [_to_child_facts(c) for c in child_entities] parents = [ ParentFacts( uuid=p.uuid, name=p.name, salary_type=p.salary_type, employment_status=p.employment_status, min_vacation_days_per_year=p.min_vacation_days_per_year, ) for p in parent_entities ] rules = settings.rules pools: dict[tuple[UUID, UUID], DayPool] = {} existing_planned: list[PlannedLeaveDay] = [] # §fix (parent-reassignment moves dates, decision B) — the FULL, # never-`only_pinned_existing`-filtered snapshot, always populated # regardless of the flag; see `AllocationInput.all_existing_planned`'s # docstring. Equal to `existing_planned` when `only_pinned_existing` # is False. all_existing_planned: list[PlannedLeaveDay] = [] for child in child_entities: planned_for_child = await self._planned_leave_repo.list_by_workspace_child( workspace_id, child.id ) all_existing_planned.extend(planned_for_child) if only_pinned_existing: # Hide unpinned, wish-attributed rows entirely -- see the # docstring above -- EXCEPT ones dated strictly before # `today`. "The past is fact": `_expand_all_windows`/ # `_expand_flexible_wish_years` clip every window to # `>= today`, so an elapsed date is never re-derived no # matter how far this loader hides it -- hiding it here too # would leave it neither preserved (that's `sync_plan`'s # job, via its own `resolved_today`-gated `stale_keys`) NOR # debited from the pools/vacation budgets below, letting the # plan over-report remaining days by exactly the count of # elapsed auto-placed days. Pinned rows and ad-hoc (no # wish_id) rows stay visible/immovable either way. visible_for_child = [ d for d in planned_for_child if (d.parent_personal_data_uuid, d.child_id, d.date) in pinned_days or d.wish_id is None or d.date < today ] else: visible_for_child = planned_for_child existing_planned.extend(visible_for_child) for parent in parent_entities: latest = await self._parental_leave_repo.get_latest_by_parent(parent.uuid, child.id) if latest is not None: tier = latest.tier_data sjuk_left = tier.sjukpenning_left lagsta_left = tier.lagsta_niva_left dubbel_left = latest.dubbeldagar_left else: # No FK snapshot yet: default this parent's share via the # shared helper (see `default_tier_pool_share` docstring — # matches the manual-entry convention in # scripts/seed_demo.py and `fk_scraper.py`'s # "= rules.lagsta_niva_days/2" comment). sjuk_left, lagsta_left = default_tier_pool_share(rules, len(parent_entities)) dubbel_left = rules.max_dubbeldagar # Net out days already committed locally (planned_leave_day) # that an FK snapshot (if any) doesn't yet reflect, so the pure # allocator never double-books or over-spends a resource. # Pinned days are excluded here — the pure `_Ledger` debits # those itself (`_reserve_pinned_days`), so this loader-side # netting only ever covers non-pinned commitments (whichever # of those `visible_for_child` still includes). committed = [ d for d in visible_for_child if d.parent_personal_data_uuid == parent.uuid and d.day_type in ("sjukpenning", "lagsta") and (d.parent_personal_data_uuid, d.child_id, d.date) not in pinned_days ] sjuk_committed = sum(1 for d in committed if d.day_type == "sjukpenning") lagsta_committed = sum(1 for d in committed if d.day_type == "lagsta") pools[(parent.uuid, child.id)] = DayPool( sjukpenning_left=max(sjuk_left - sjuk_committed, 0), lagsta_left=max(lagsta_left - lagsta_committed, 0), dubbeldagar_left=dubbel_left, ) horizon_end = _compute_horizon_end(today, children, wishes) # §Task 1 (per-year preschool reserve) — recorded `taken_days` wins # over `planned_days` (see `PreschoolPlanningDays.taken_days`'s # docstring); a (child, year) with no row is simply left out of the # dict, so `_preschool_reserve_days` falls back to `ChildFacts. # preschool_planning_days_per_year` for it rather than treating the # gap as zero. preschool_reserve_by_year: dict[tuple[UUID, int], int] = {} if self._preschool_planning_days_repo is not None: for child in child_entities: rows = await self._preschool_planning_days_repo.list(child_id=child.id) for row in rows: reserve = row.taken_days if row.taken_days is not None else row.planned_days preschool_reserve_by_year[(child.id, row.year)] = reserve def _already_used_vacation(parent_uuid: UUID, year: int) -> int: # Pinned vacation days are excluded here for the same reason as # the FP pools above: `_Ledger._reserve_pinned_days` debits them # itself. return sum( 1 for d in existing_planned if d.parent_personal_data_uuid == parent_uuid and d.day_type == "vacation" and d.date.year == year and (d.parent_personal_data_uuid, d.child_id, d.date) not in pinned_days ) vacation: dict[tuple[UUID, int], VacationBudget] = {} # §Task 2 (vacation-forfeiture bias) — see `AllocationInput. # vacation_forfeiting`'s docstring. Computed inside this same loop # (not a separate pass) so `vac_entries` is only fetched once per # parent. vacation_forfeiting: dict[tuple[UUID, int], int] = {} forecast_years = list(range(today.year, horizon_end.year + 1)) for parent in parent_entities: vac_entries = await self._vacation_repo.list( personal_data_uuid=parent.uuid, workspace_id=workspace_id ) entered_years = {vd.year for vd in vac_entries} for vd in vac_entries: vacation[(parent.uuid, vd.year)] = VacationBudget( total_days=vd.total_days, saved_days_in=vd.saved_days_in, already_used=_already_used_vacation(parent.uuid, vd.year), ) # A year inside the planning horizon with no `vacation_days` row # means "the user has not told us yet", not "this person gets no # holiday" — defaulting a missing year to 0 was what let the # allocator report a shortfall for want of semester while the # child's FP pool sat untouched (the same run also needs Task 1's # fix, since a real 0-budget year must still fall through to FP). # Default to the statutory entitlement and flag it as assumed # (`is_assumed=True`) so the UI can prompt the user to confirm or # override it instead of silently treating it as an entered value. for year in range(today.year, horizon_end.year + 1): if year in entered_years: continue vacation[(parent.uuid, year)] = VacationBudget( total_days=rules.statutory_vacation_days_per_year, saved_days_in=0, already_used=_already_used_vacation(parent.uuid, year), is_assumed=True, ) # §Task 2 — reuse `AnalyticsService._vacation_year_series` # verbatim (the same FIFO carry-forward bank simulation behind # the Analytics tab's balance series) rather than re-deriving it, # so this forecast can never quietly drift from what the # dashboard reports for the same (parent, year). `use_by_year` # deliberately counts EVERY persisted 'vacation' row for this # parent across ALL of their children, unfiltered by # pinned/workspace — matching `AnalyticsService. # _compute_year_balances` exactly, since a forfeiture forecast # must reflect every already-committed day, pinned or not. # Seeded from `today.year` (that year's `saved_days_in` as the # sole carry-in, exactly like the `vacation` dict above) rather # than the child's birth year: this loader never reconstructs # full lifetime history, only "now onward" — this also means the # simulation can never see, let alone report, a year before # `today.year` (a past loss already happened and must not # resurface here or bias `_assign_tier3`). vac_by_year = {vd.year: vd for vd in vac_entries} parent_planned_days = await self._planned_leave_repo.list( parent_personal_data_uuid=parent.uuid ) use_by_year: dict[int, int] = {} for pd in parent_planned_days: if pd.day_type == "vacation": use_by_year[pd.date.year] = use_by_year.get(pd.date.year, 0) + 1 vacation_balances = AnalyticsService._vacation_year_series( forecast_years, vac_by_year, use_by_year, parent.min_vacation_days_per_year ) for year, balance in zip(forecast_years, vacation_balances, strict=True): if balance.expiring > 0: vacation_forfeiting[(parent.uuid, year)] = balance.expiring # `strict=True`: never silently fall back to the live `holidays` # library for a year outside `swedish_calendar.json`'s cached range — # that fallback is missing de facto days (Julafton/Nyårsafton/ # Midsommarafton) and previously caused those to be charged as # working days once a plan's horizon ran past the cache (see # `app.utils.holidays.UncachedYearError`). With the cache now ~30 # years deep this should never fire for a realistic horizon, but a # wish with a far-out `last_year` (uncapped when the workspace has no # children yet) can still in principle reach past it. Rather than # silently clamping the horizon (which would contradict the §6 # decision to plan past B4 and show the squeeze), surface it as a # `ValidationError` the router turns into a 422 the UI can explain, # the same way `CalendarService` handles an out-of-range `start_year`. holidays_map: dict[date, str] = {} try: for year in range(today.year, horizon_end.year + 2): holidays_map.update(get_swedish_holidays(year, strict=True)) except UncachedYearError as exc: bounds = cached_year_bounds() where = f"{bounds[0]}-{bounds[1]}" if bounds is not None else "(cache missing)" raise ValidationError( f"Planning horizon ({horizon_end.isoformat()}) extends beyond the cached " f"Swedish calendar {where}; shorten the wish's last_year or the plan's " "horizon so it stays within the cached range." ) from exc day_overrides: dict[date, bool] = {} if self._day_override_repo is not None: overrides = await self._day_override_repo.list_by_workspace(workspace_id) day_overrides = {o.date: o.is_working_day for o in overrides} # §remove-this-day — the dates the user removed by hand. Loaded # unconditionally (never narrowed by `only_pinned_existing`): a # removal is an instruction about the plan, not a planned row, so a # re-plan must see every one of them or the removed day comes back on # the very next sync — which is the bug this exists to fix. # # Restructure step 3b — `excluded_days` is exactly the `excluded` # half of `partition_assertion_keys(assertions)`, computed once above # alongside `pinned_days`. `PlannedDayExclusion` is gone (migration # 010); a `day_type='none'` `DayAssertion` is the only source now. return AllocationInput( today=today, horizon_end=horizon_end, age4_savable_cap=resolve_age4_savable_cap(len(parents)), children=children, parents=parents, pools=pools, vacation=vacation, wishes=wishes, existing_planned=existing_planned, holidays=holidays_map, day_overrides=day_overrides, take_pl_on_red_days=ws.take_pl_on_red_days, all_existing_planned=all_existing_planned, pinned_days=pinned_days, excluded_days=excluded_days, preschool_reserve_by_year=preschool_reserve_by_year, vacation_forfeiting=vacation_forfeiting, ) class PlanningEngineService: """Thin orchestrator gluing the loader and the pure allocator together. This is what the router calls. It contains no allocation logic of its own — that all lives in `Allocator`, which stays pure and independently testable. """ def __init__(self, loader: AllocationInputLoader, allocator: Allocator | None = None) -> None: self._loader = loader self._allocator = allocator or Allocator() async def propose( self, workspace_id: UUID, today: date | None = None, *, only_pinned_existing: bool = False, ) -> PlanProposal: """Load fresh input and run the pure allocator over it. `only_pinned_existing` (used by `PlanningService.sync_plan`, §Task 4) makes the loader hide every *unpinned*, wish-attributed `planned_leave_day` row from `AllocationInput.existing_planned` (and stop netting it out of the pools/vacation budgets) so the allocator re-derives that portion of the plan from scratch instead of treating it as an immovable fact — this is what lets a re-plan legitimately move/retype previously-auto-placed days. Pinned days and ad-hoc (`wish_id is None`) days are unaffected either way: always visible, always immovable. `POST /api/plan/propose` never sets this — it keeps its historical, unconditional behaviour. """ alloc_input = await self._loader.load( workspace_id, today=today, only_pinned_existing=only_pinned_existing ) return self._allocator.propose(alloc_input)