jacobh2/F-r-ldraledigplannerare
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
8ebfcb9462dfd711b0de715eb13bc4fc4b94b015
F-r-ldraledigplannerare/backend/app/services/allocation_service.py
T
jacobh2 87b3eb5d0c Fix four bugs found by browser verification 
First real browser walkthrough of the app, against seeded demo data. All 1140
tests were passing; none of these four were visible to them.

Age-4 entitlement (high severity, domain bug). The no-snapshot fallback gave a
single parent rules.sjukpenning_days + rules.lagsta_niva_days = 390 + 90 = 480
days, but those config values are the whole CHILD's combined tier totals across
both parents, not one parent's share. The correct fallback is days_per_parent
(240 = 195 + 45), which only AnalyticsService used. So a parent without an FK
snapshot was credited with roughly double their real entitlement, and the
Dashboard and Plan tab disagreed about expiring_at_b4 for the same committed
state. The wrong default existed in three places; allocation_service.py and
planning_service.py are corrected, and analytics_service.py now calls the shared
pure days_left_at_age_4_projection instead of re-deriving the arithmetic, so
there is one source of truth. A test that had encoded the buggy 390/90 default
as expected behaviour is corrected too.

The Dashboard stat is workspace-wide while the Plan tab breaks down per parent,
which is legitimate but was unlabelled and contributed to the confusion; the
dashboard label now says it is a combined all-parents total.

Nonsensical remedy text. lower_priority_wish computed
max(other.target_weeks // 2, 1), which for a 1-week wish clamped back to 1 and
rendered "Lowering Var from 1 to 1 weeks frees up 32 days". target_weeks has a
domain floor of 1 and the planner works in whole days, so there is no smaller
whole-week suggestion to make: the remedy is now suppressed for 1-week wishes.
Wishes of 2+ weeks still get a genuine, strictly smaller suggestion.

Day override invisible until reload. Calendar refetched the overrides list but
not GET /api/calendar/days, which backs the grid's shading and badges, so a
just-applied override did not render. useCalendarDays now exposes refetch and
Calendar calls it after a successful upsert or remove.

Dashboard showed the previous user's data after a login swap. WorkspaceContext
sits above the router and outlives login/logout, and its refetch was gated on
!workspaceId || children.length === 0 - so precisely because the old session's
data was still present, the guard blocked the refetch. It now keys off an
explicit isAuthenticated transition and clears state on the logout edge itself,
including the persisted localStorage workspaceId, so one account's figures can
never be briefly visible under another's session. useAnalytics, useParentalLeave
and usePlannedDays clear cached data on identity loss as defence in depth.

Backend 435 -> 439 tests, frontend 705 -> 710.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01RXMhc9nT6sNz1SL4j8dznL
2026-09-14 08:21:18 +02:00

2039 lines
83 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, has_vacation_near_expiry
# ---------------------------------------------------------------------------
# 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
_MAX_SAVABLE_AT_B4 = settings.rules.max_savable_days_at_age_4
_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_goal: str # 'max_time' | 'max_money' | 'balanced'
planning_mode: str # 'solo' | 'couple'
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, max_savable_days_at_age_4)
expiring_at_b4: int # max(projected_left_at_b4 - max_savable, 0) — LOST if not used
@dataclass(frozen=True)
class WindowOutcome:
wish_id: UUID
wish_name: str
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))
# ---------------------------------------------------------------------------
# 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_preschool_days()
# -- setup ----------------------------------------------------------
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, 96). 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, _MAX_SAVABLE_AT_B4)
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,
resolved_goal: str,
early: bool,
) -> tuple[str, str] | None:
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 resolved_goal == "max_money":
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
if resolved_goal == "max_time":
if d < child.birth_date + relativedelta(months=12):
# skip_fp_first_12_months: SGI is protected regardless, so
# don't spend the scarce FP pool now — save it to stretch
# later. Vacation only, or leave unfunded.
if self._spend_vacation(parent.uuid, d):
return "vacation", REASON_VACATION_WRAP_WEEKEND
return None
# stretch_time_at_home: draw the flat lagsta tier first to
# preserve the higher-value sjukpenning tier for later.
fp = self._spend_fp(pool_key, prefer="lagsta")
if fp:
return fp, (REASON_FP_LAGSTA_STRETCH if fp == "lagsta" else REASON_FP_INTERIOR)
if self._spend_vacation(parent.uuid, d):
return "vacation", REASON_VACATION_WRAP_WEEKEND
return None
# 'balanced': semester early in the window, FP for the remainder.
if early:
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],
resolved_goal: str,
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
half_index = (len(dates) + 1) // 2
for i, d in enumerate(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, resolved_goal, early=(i < half_index)
)
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],
resolved_goal: str,
responsible: _Responsible,
) -> tuple[list[ProposedDay], dict[str, int]]:
if not dates:
return [], _new_cost()
mid_index = (len(dates) + 1) // 2 # first half gets the extra day when odd
first_half, second_half = dates[:mid_index], dates[mid_index:]
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, resolved_goal, parent_a, REASON_SWAP_FIRST_HALF
)
days_b, cost_b = self._fund_solo(
wish, second_half, red_cluster, resolved_goal, parent_b, REASON_SWAP_SECOND_HALF
)
merged_cost = {k: cost_a[k] + cost_b[k] for k in cost_a}
return days_a + days_b, merged_cost
def _fund_both(
self,
wish: LeaveWish,
w: _WishWindow,
dates: list[date],
red_cluster: set[date],
resolved_goal: str,
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, resolved_goal, 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, resolved_goal, 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
resolved_goal = wish.goal_override or self._input.planning_goal
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, resolved_goal, responsible
)
strategy = "dubbeldag"
elif responsible.mode == "swap":
days, cost = self._fund_swap(
wish, w, all_dates, red_cluster, resolved_goal, 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, resolved_goal, 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,
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,
"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] = []
remedy_child = self._resolve_wish_child(wish)
if remedy_child is not None:
reserved = self._preschool_reserved.get(remedy_child.id, 0)
if reserved > 0:
remedies.append(
ShortfallRemedy(
kind=REMEDY_REDUCE_PRESCHOOL_RESERVE,
message_key=f"planRemedies.{REMEDY_REDUCE_PRESCHOOL_RESERVE}",
data={"reserved": reserved},
days_freed=reserved,
)
)
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:
remedies.append(
ShortfallRemedy(
kind=REMEDY_CARRY_VACATION_FORWARD,
message_key=f"planRemedies.{REMEDY_CARRY_VACATION_FORWARD}",
data={"spare_total": spare_total},
days_freed=spare_total,
)
)
if wish.child_id is not None:
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 siblings:
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:
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:
remedies.append(
ShortfallRemedy(
kind=REMEDY_LOWER_PRIORITY_WISH,
message_key=f"planRemedies.{REMEDY_LOWER_PRIORITY_WISH}",
data={
"wish_name": other.name,
"from_weeks": other.target_weeks,
"to_weeks": half_weeks,
"freed_days": freed_estimate,
},
days_freed=freed_estimate,
)
)
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:
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, "shift_gain": shift_gain},
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,
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_weeks * 7
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_weeks * 7
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,
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,
"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,
)
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, _MAX_SAVABLE_AT_B4)
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,
)
)
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) -> AllocationInput:
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
)
existing_planned.extend(planned_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: `rules.sjukpenning_days`/`lagsta_niva_days`
# are the WHOLE CHILD's combined tier totals (390 + 90, split
# across both parents) — assuming that full amount for a single
# parent double-counts the pool. Halve it to get this parent's
# default share, matching AnalyticsService's no-snapshot
# fallback (`rules.days_per_parent` == 195 + 45 == 240) and the
# manual-entry convention (see scripts/seed_demo.py, and
# `fk_scraper.py`'s "= rules.lagsta_niva_days/2" comment).
sjuk_left = rules.sjukpenning_days // 2
lagsta_left = rules.lagsta_niva_days // 2
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.
committed = [
d
for d in planned_for_child
if d.parent_personal_data_uuid == parent.uuid
and d.day_type in ("sjukpenning", "lagsta")
]
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:
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
)
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_goal=ws.planning_goal,
planning_mode=ws.planning_mode,
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) -> PlanProposal:
alloc_input = await self._loader.load(workspace_id, today=today)
return self._allocator.propose(alloc_input)
Reference in New Issue View Git Blame Copy Permalink