jacobh2/F-r-ldraledigplannerare
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
4cd3d3187b6acc92b1ae19949e029e4095c4d58a
F-r-ldraledigplannerare/backend/app/services/allocation_service.py
T
jacobh2 782415122d Separate user intent from planner output via day assertions 
The calendar mixed two kinds of truth in one table. `planned_leave_day` was
simultaneously the allocator's output cache, the user's override store
(`is_pinned`, plus ad-hoc rows with no `wish_id`), and the resource ledger that
analytics and the age-4 maths read. The planner therefore had to ask "am I
allowed to touch this row?" on every sync, and got it wrong at the edges.

Every user assertion is now one `day_assertion` row — a concrete `day_type`
replaces a pinned day, a row with no `wish_id` replaces an ad-hoc day, and
`day_type='none'` replaces `planned_day_exclusion`. `planned_leave_day` becomes
purely derived with a single writer, and `PlanningService.pinned_keys()` is the
single answer to "is this day pinned", used by the allocator's loader, by
`sync_plan` and by the read router alike.

On top of that, every calendar band is now a real event: an ad-hoc band opens
the same full editor as a wish-backed one and saving it promotes it to a
`LeaveWish`, so it can be named, repeated yearly, assigned to either or both
parents, and handed to the planner. "Planering" is a `moveable` toggle plus a
from/to month bound plus a separate lock/unlock action on the days themselves,
replacing a three-rung ladder whose "planner decides" rung set a search window
equal to the drawn dates and so could never move anything.

Bugs found and fixed along the way:

- The plan erased its own history. Fixed windows clip to today, so an elapsed
  auto-placed day fell into `stale_keys` and was deleted on the next sync —
  and a background sync runs on every workspace load. Past rows are now frozen
  and counted as spent.
- Migration 010's downgrade silently destroyed every pinned day and exclusion
  (`is_pinned` returned as DEFAULT 0, the exclusion table came back empty,
  exit 0, no warning). It now reconstructs both from `day_assertion`, and 009
  itemises what it cannot recover before dropping the table.
- Deleting a period left its assertions behind, locking those dates forever
  with no UI path to clear them. Wish deletion now clears them by `wish_id`.
- `DayAssertionService` and `LeaveWishService` committed multi-statement writes
  without a transaction; a crash mid-write diverged the tables or half-deleted
  a series. Both now run in a unit of work.
- A sibling year's shortfall was never surfaced: the toast only inspected the
  edited occurrence's outcome, so a fully unfunded later year rendered as an
  empty band with no warning.
- A flexible window whose bound had entirely elapsed was dropped from the
  proposal with no outcome at all. It now reports as `unplaced`.
- Unlocking a period deleted its days instead of returning them to the planner.
- The e2e global setup posted to a route deleted in this change, which would
  have failed the whole suite before its first test.

Gates: backend 938 tests / 91% coverage, frontend 1351 / 97.4 | 92.2 | 94.37 |
97.4. Migrations 009 and 010 audited against a hand-built realistic database:
the upgrade reconciles exactly and a constraint violation fails safe, leaving
the schema at 008.

Not complete. Step 4's toggle, month picker and lock/unlock action are verified
by browser walk and unit tests, but the partial-lock readout was not exercised.
`ideas.md` §7 adds a full plan for the vacation-forfeiture defect (the planner
detects expiring days and then does nothing about them) — not implemented here.
Remaining smaller items are listed in the session's restructure notes.

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

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