jacobh2/F-r-ldraledigplannerare
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
d2453908867dea4116fa9e668bae78d80fd96c5e
F-r-ldraledigplannerare/backend/app/services/allocation_service.py
T
jacobh2 d245390886 Derive the rule constants that are arithmetic on other constants 
The per-parent day split was written out twice, once in the allocator's input
loader and once in the persistence layer, each with a comment recalling the
time the two disagreed. They now share one derivation, so they cannot drift
apart again without the shared function moving.

Three constants that were stored and happened to equal arithmetic on their
neighbours — the child's total, the per-parent share, and the per-parent
ceiling at age four — are now computed from them. The figures are unchanged;
what changes is that a future adjustment to a tier total carries into them
instead of quietly leaving them behind.

The shortfall remedies were also split into one method per remedy, matching
the surplus advice alongside them, which had already been written that way.
2026-09-18 01:12:00 +02:00

2216 lines
91 KiB
Python
Raw Blame History

"""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.leave_wish import LeaveWish
from app.entities.planned_leave import PlannedLeaveDay
from app.repositories.child_repository import ChildRepository
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.vacation_days_repository import VacationDaysRepository
from app.repositories.workspace_repository import WorkspaceRepository
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,
has_vacation_near_expiry,
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"
# ---------------------------------------------------------------------------
# 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.
# ---------------------------------------------------------------------------
STRATEGY_UNPLACED = "unplaced"
# §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
_MAX_DUBBELDAGAR = settings.rules.max_dubbeldagar
# NOTE: no module-level "_MAX_SAVABLE_AT_B4" here on purpose — which age-4 cap
# applies (96 per child vs. 48 per parent) depends on the workspace's
# `planning_mode` and 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
# ---------------------------------------------------------------------------
# §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
planning_mode: str # 'solo' | 'couple'
# 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, or 48 per parent for a couple workspace with 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)
# ---------------------------------------------------------------------------
# §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 — 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
strategy: str
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]
@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) -> int:
"""§9.3: days reserved off the top for preschool planning-day closures."""
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
years = end_year - start_year + 1
return child.preschool_planning_days_per_year * years
# ---------------------------------------------------------------------------
# Window expansion (§3.2)
# ---------------------------------------------------------------------------
@dataclass(frozen=True)
class _WishWindow:
wish: LeaveWish
anchor_year: int
start: date
end: date
def _candidate_years(
wish: LeaveWish, alloc_input: AllocationInput, children_by_id: dict[UUID, ChildFacts]
) -> list[int]:
"""Anchor years a wish (fixed or flexible) might occur in, before
`applies_to_year`/horizon filtering. Shared by `_expand_all_windows` and
`_expand_flexible_wish_years` (§10.2) — identical for both, since
`window_mode` only changes how a year's `window_for_year()` result is
*used* (exact placement vs. bounding range), never which years are
considered.
"""
if wish.recurrence == "once":
return [wish.first_year]
if wish.last_year is not None:
end_year = wish.last_year
elif wish.child_id is not None and wish.child_id in children_by_id:
end_year = children_by_id[wish.child_id].fourth_birthday.year
else:
end_year = alloc_input.horizon_end.year
return list(range(wish.first_year, end_year + 1))
def _expand_all_windows(alloc_input: AllocationInput) -> list[_WishWindow]:
"""§3.2 — expand every `window_mode == 'fixed'` wish into concrete windows.
Flexible wishes (§10.2) are expanded separately by
`_expand_flexible_wish_years`, since their `window_for_year()` result is a
*bounding range* to search inside, not the final placement.
"""
children_by_id = {c.id: c for c in alloc_input.children}
windows: list[_WishWindow] = []
for wish in alloc_input.wishes:
if wish.window_mode != "fixed":
continue
for year in _candidate_years(wish, alloc_input, children_by_id):
if not wish.applies_to_year(year):
continue
start, end = wish.window_for_year(year)
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=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
def _expand_flexible_wish_years(alloc_input: AllocationInput) -> list[_FlexibleWishYear]:
"""§10.2 — the flexible-wish counterpart to `_expand_all_windows`.
Reuses `applies_to_year`, the horizon clip and the year-wrap rule
unchanged; the only difference is that `window_for_year()`'s result is
treated as a bound to search inside (§10.2 step 1), not the final window.
"""
children_by_id = {c.id: c for c in alloc_input.children}
wish_years: list[_FlexibleWishYear] = []
for wish in alloc_input.wishes:
if wish.window_mode != "flexible":
continue
for year in _candidate_years(wish, alloc_input, children_by_id):
if not wish.applies_to_year(year):
continue
bound_start, bound_end = wish.window_for_year(year)
if bound_end < alloc_input.today or bound_start > alloc_input.horizon_end:
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=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).
"""
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)
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
}
# §fix (double-booking on flexible re-propose) — 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). `_reuse_already_funded_window` is the only
# reader.
self._persisted_lookup: dict[tuple[UUID, UUID | None, date], str] = dict(
self._existing_lookup
)
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:
if not p.is_pinned:
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)
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:
parents = self._input.parents
if len(parents) < 2 or self._input.planning_mode != "couple":
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 ------------------------------------------------
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 _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.
"""
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
if parent.salary_type == "monthly":
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:
existing_type = self._existing_lookup.get((parent.uuid, child_id, d))
if existing_type is not None:
cost[existing_type] = cost.get(existing_type, 0) + 1
continue
if is_free_day(d, self._input.holidays, self._input.day_overrides):
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]]:
parent_a, parent_b = responsible.parents[0], responsible.parents[1]
child = self._resolve_wish_child(wish)
if child is None:
return self._fund_swap(wish, w, dates, red_cluster, responsible)
self._maybe_clamp_age4((parent_a.uuid, child.id), w.start)
self._maybe_clamp_age4((parent_b.uuid, child.id), w.start)
eligible_end = child.birth_date + relativedelta(months=_DUBBELDAGAR_MONTHS_LIMIT)
days: list[ProposedDay] = []
cost = _new_cost()
used = self._dubbeldagar_used.get(child.id, 0)
leftover_for_swap: list[date] = []
for d in dates:
if d > eligible_end:
leftover_for_swap.append(d)
continue
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):
cost["free"] += 1
reason = (
REASON_FREE_RED_DAY
if _is_real_red_day(d, self._input.holidays)
else REASON_FREE_WEEKEND
)
days.append(ProposedDay(d, parent_a.uuid, child.id, "free", False, wish.id, reason))
days.append(ProposedDay(d, parent_b.uuid, child.id, "free", False, wish.id, reason))
continue
if used >= _MAX_DUBBELDAGAR:
leftover_for_swap.append(d)
continue
pool_a = self._pools.get((parent_a.uuid, child.id))
pool_b = self._pools.get((parent_b.uuid, child.id))
if pool_a is None or pool_b is None:
leftover_for_swap.append(d)
continue
tier_a = self._peek_fp_tier(pool_a)
tier_b = self._peek_fp_tier(pool_b)
if tier_a is None or tier_b is None:
leftover_for_swap.append(d)
continue
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)
)
self._dubbeldagar_used[child.id] = used
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]
return days, cost
# -- 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)
all_dates = _dates_in(w.start, w.end)
red_cluster = _red_day_cluster_dates(
w.start, w.end, self._input.holidays, self._input.day_overrides
)
if responsible.mode == "dubbeldag":
days, cost = self._fund_both(wish, w, all_dates, red_cluster, responsible)
strategy = "dubbeldag"
elif responsible.mode == "swap":
days, cost = self._fund_swap(wish, w, all_dates, red_cluster, responsible)
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"
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)
remedies = self._build_remedies(wish, w, shortfall_days)
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,
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,
)
return days, outcome
# -- shortfall remedies (§3.6, §9.1) ------------------------------------
def _build_remedies(
self, wish: LeaveWish, w: _WishWindow, shortfall_days: int
) -> list[ShortfallRemedy]:
if shortfall_days <= 0:
return []
remedies: list[ShortfallRemedy] = []
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))
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_for_year(other.first_year)
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]:
shift_gain = 0
probe = w.start - timedelta(days=1)
for _ in range(3):
if is_free_day(probe, self._input.holidays, self._input.day_overrides):
shift_gain += 1
probe -= timedelta(days=1)
else:
break
probe = w.end + timedelta(days=1)
for _ in range(3):
if is_free_day(probe, self._input.holidays, self._input.day_overrides):
shift_gain += 1
probe += timedelta(days=1)
else:
break
if shift_gain <= 0:
return []
return [
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,
},
days_freed=shift_gain,
)
]
# -- 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,
responsible: _Responsible,
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_lookup` — never the
mutable `_existing_lookup`) for this wish's resolved parent(s) and
child, 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`).
"""
parent_uuids = {p.uuid for p in responsible.parents}
return sorted(
{
d
for (p_uuid, c_id, d) in self._persisted_lookup
if c_id == child_id and p_uuid in parent_uuids and bound_start <= d <= bound_end
}
)
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.
"""
persisted_dates = self._persisted_dates_for_wish(
responsible, 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
return min(
candidates,
key=lambda c: (
_candidate_cost(c[0], c[1], self._input.holidays, self._input.day_overrides),
c[0],
wish.id.int,
),
)
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: (
_candidate_cost(c[0], c[1], self._input.holidays, self._input.day_overrides),
c[0],
wish.id.int,
),
)
return _WishWindow(wish=wish, anchor_year=anchor_year, start=winner_start, end=winner_end)
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.
remedies = [
r
for r in self._build_remedies(wish, bound_window, length_days)
if r.kind != REMEDY_SHIFT_WINDOW_INTO_FREE_DAYS
]
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,
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,
)
# -- 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())
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
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) -> list[PlanWarning]:
warnings: list[PlanWarning] = []
for (parent_uuid, year), budget in sorted(
self._vacation.items(), key=lambda kv: (str(kv[0][0]), kv[0][1])
):
if (
has_vacation_near_expiry(budget.total_days, budget.saved_days_in)
and budget.available > 0
):
warnings.append(
PlanWarning(
kind=WARNING_VACATION_EXPIRING,
message_key=f"planWarnings.{WARNING_VACATION_EXPIRING}",
data={
"parent_personal_data_uuid": parent_uuid,
"year": year,
"saved_days_in": budget.saved_days_in,
},
)
)
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:
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 year-end, latest fourth_birthday), capped at age-12."""
candidates: list[date] = [c.fourth_birthday for c in children]
children_by_id = {c.id: c for c in children}
for wish in wishes:
if wish.recurrence == "once":
end_year = wish.first_year
elif wish.last_year is not None:
end_year = wish.last_year
elif wish.child_id is not None and wish.child_id in children_by_id:
end_year = children_by_id[wish.child_id].fourth_birthday.year
else:
end_year = today.year + 4
_, end = wish.window_for_year(end_year)
candidates.append(end)
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,
) -> 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
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 (`is_pinned`) and ad-hoc rows with no `wish_id`. Every other
(unpinned, wish-attributed) 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.
"""
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)
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] = []
for child in child_entities:
planned_for_child = await self._planned_leave_repo.list_by_workspace_child(
workspace_id, child.id
)
if only_pinned_existing:
# Hide unpinned, wish-attributed rows entirely -- see the
# docstring above. 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.is_pinned or d.wish_id is None
]
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)
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 not d.is_pinned
]
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)
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 not d.is_pinned
)
vacation: dict[tuple[UUID, int], VacationBudget] = {}
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,
)
# `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}
return AllocationInput(
today=today,
horizon_end=horizon_end,
planning_mode=ws.planning_mode,
age4_savable_cap=resolve_age4_savable_cap(ws.planning_mode, len(parents)),
children=children,
parents=parents,
pools=pools,
vacation=vacation,
wishes=wishes,
existing_planned=existing_planned,
holidays=holidays_map,
day_overrides=day_overrides,
)
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)
Reference in New Issue View Git Blame Copy Permalink