F-r-ldraledigplannerare/backend/app/services/planning_service.py

"""Business logic for planned leave days (calendar planning)."""

from dataclasses import dataclass, field
from datetime import date, datetime
from uuid import UUID, uuid4

from dateutil.relativedelta import relativedelta

from app.config import settings
from app.entities.day_assertion import DayAssertion, partition_assertion_keys
from app.entities.planned_leave import PlannedLeaveDay
from app.repositories.day_assertion_repository import DayAssertionRepository
from app.repositories.day_override_repository import DayOverrideRepository
from app.repositories.parental_leave_repository import ParentalLeaveRepository
from app.repositories.planned_leave_repository import PlannedLeaveRepository
from app.repositories.unit_of_work import NoOpUnitOfWork, UnitOfWork
from app.repositories.vacation_days_repository import VacationDaysRepository
from app.repositories.workspace_repository import WorkspaceRepository
from app.entities.leave_wish import LeaveWish
from app.services.allocation_service import PlanningEngineService, PlanProposal, WindowOutcome
from app.services.analytics_service import AnalyticsService
from app.services.day_guards import reject_illegal_free_day
from app.services.leave_wish_service import LeaveWishService
from app.utils.errors import DubbeldagError, NotFoundError, ValidationError
from app.utils.tip_predicates import default_tier_pool_share

# Day types that can actually be persisted as a `planned_leave_day` row. A
# proposal may also contain 'free' entries (automatic weekend/red-day
# non-workdays, see allocation_service.ProposedDay) purely for explainability
# — those never become rows here, see `apply_proposed_days`.
_PERSISTABLE_DAY_TYPES = {"sjukpenning", "lagsta", "vacation"}


# ---------------------------------------------------------------------------
# POST /api/plan/apply — bulk accept of a (previously proposed) set of days.
# ---------------------------------------------------------------------------


@dataclass(frozen=True)
class ProposedDayInput:
    """One day out of a batch handed to `apply_proposed_days`.

    Deliberately a plain domain dataclass (not a Pydantic model) so the
    service stays independent of the wire schema in `app.models.plan` — the
    router is responsible for that translation. Shape mirrors
    `allocation_service.ProposedDay` minus the explain-only `reason` field,
    which carries no persistence meaning. `wish_id` DOES carry persistence
    meaning (unlike `reason`): it links the persisted `planned_leave_day` row
    back to the `LeaveWish` that produced it (nullable — ad-hoc/legacy days
    have none), which is what lets the UI later treat a multi-day band as one
    editable period instead of one day at a time.
    """

    date: date
    parent_personal_data_uuid: UUID
    child_id: UUID
    day_type: str
    is_dubbeldag: bool = False
    wish_id: UUID | None = None


@dataclass(frozen=True)
class DayCollision:
    """A date already planned differently than what the batch proposes."""

    date: date
    parent_personal_data_uuid: UUID
    child_id: UUID
    existing_day_type: str
    proposed_day_type: str


@dataclass(frozen=True)
class Rejection:
    """A single reason a batch (or part of it) was refused.

    §3.5: mirrors `tips_service.Tip`'s `message_key` + `data` shape rather
    than a pre-rendered string, so `PlanApplyResponse.rejections` can be
    rendered in the caller's active locale instead of leaking backend-
    generated (previously English-only, regardless of locale) text.
    """

    message_key: str
    data: dict[str, object]


@dataclass(frozen=True)
class PlanApplyResult:
    accepted: bool
    created: list[PlannedLeaveDay] = field(default_factory=list)
    created_count: int = 0
    skipped_identical_count: int = 0
    skipped_free_count: int = 0
    replaced_count: int = 0
    deleted_count: int = 0
    collisions: list[DayCollision] = field(default_factory=list)
    rejections: list[Rejection] = field(default_factory=list)


# ---------------------------------------------------------------------------
# POST /api/plan/sync — propose + apply in one step, with a structured diff.
# ---------------------------------------------------------------------------


@dataclass(frozen=True)
class PlanChange:
    """One (child, parent, date) slot's delta from `sync_plan`'s pre-write

    snapshot to what's actually persisted after the call. Structured data
    only — house style is `message_key` + `data`, never backend-rendered
    prose, so the frontend composes the toast copy ("Sommar 2028 flyttades 2
    dagar") itself. `kind`:

    - 'added' — no row existed for (child, parent, date); one does now.
    - 'removed' — a row existed and no longer does (old_* populated, new_*
      all `None`).
    - 'retyped' — the row survived but `day_type`/`is_dubbeldag` changed.
    - 'reattributed' — same day and type, but `wish_id` changed (the day now
      belongs to a different period).

    `evicted` can be `True` on a 'removed' OR a 'reattributed' change — both
    are ways a wish can lose a date it still wants to a *different* wish's
    higher-priority claim (see `PlanningService._genuinely_evicted_rows`):
    'removed' when the two wishes land on different (child, parent, date)
    keys so the losing wish's row is deleted outright, 'reattributed' when
    they land on the EXACT SAME key so the row survives and is merely
    repointed at the winning wish's `wish_id` — that repointing is still a
    genuine displacement of the OLD wish, not the ordinary "wish itself was
    edited/retargeted" case, and must be called out the same way. Step 2 of
    the intent-vs-derived-output restructure made displacement an
    informational outcome rather than a refusal — `sync_plan` always writes,
    and this flag is what lets the UI call out "N days were freed up for a
    higher-priority period" instead of folding it into the same silent
    "N days changed" count as a self-caused edit.
    """

    kind: str
    date: date
    parent_personal_data_uuid: UUID
    child_id: UUID
    old_day_type: str | None
    new_day_type: str | None
    old_is_dubbeldag: bool | None
    new_is_dubbeldag: bool | None
    old_wish_id: UUID | None
    new_wish_id: UUID | None
    evicted: bool = False


@dataclass(frozen=True)
class PlanSyncResult:
    proposal: PlanProposal
    apply_result: PlanApplyResult
    changes: list[PlanChange] = field(default_factory=list)


# ---------------------------------------------------------------------------
# POST /api/plan/sync-with-wish — the calendar's four create/edit-then-sync
# paths (drag-select, "promote" an ad-hoc band, edit a wish-backed band, pick
# a preset), each collapsed into ONE atomic backend call. See
# `PlanningService.sync_with_wish`'s docstring for the shape and the
# atomicity guarantee.
# ---------------------------------------------------------------------------


@dataclass(frozen=True)
class WishCreateInput:
    """Everything `LeaveWishService.create` needs, minus `workspace_id`

    (carried once on `sync_with_wish` itself). Mirrors `LeaveWishCreate`
    field-for-field — kept as a separate domain dataclass (not the Pydantic
    model) for the same reason `ProposedDayInput` is: the service stays
    independent of the wire schema.
    """

    name: str
    window_start_month: int
    window_start_day: int
    window_end_month: int
    window_end_day: int
    target_weeks: int
    first_year: int
    child_id: UUID | None = None
    name_message_key: str | None = None
    target_days: int | None = None
    recurrence: str = "once"
    last_year: int | None = None
    priority: int = 1
    parent_mode: str = "either"
    parent_personal_data_uuid: UUID | None = None
    window_mode: str = "fixed"
    series_id: UUID | None = None
    wants_dubbeldag: bool = False


@dataclass(frozen=True)
class WishUpdateInput:
    """Everything `LeaveWishService.update` needs for one occurrence/series edit.

    Mirrors `LeaveWishUpdate` field-for-field, including its `clear_*`
    "explicitly set to null" flags (see that model's docstring) — the
    router computes those from the patch's `model_fields_set`, exactly as
    `PUT /api/leave-wishes/{id}` already does.
    """

    leave_wish_id: UUID
    scope: str = "occurrence"
    child_id: UUID | None = None
    name: str | None = None
    name_message_key: str | None = None
    window_start_month: int | None = None
    window_start_day: int | None = None
    window_end_month: int | None = None
    window_end_day: int | None = None
    target_weeks: int | None = None
    target_days: int | None = None
    recurrence: str | None = None
    first_year: int | None = None
    last_year: int | None = None
    priority: int | None = None
    parent_mode: str | None = None
    parent_personal_data_uuid: UUID | None = None
    window_mode: str | None = None
    wants_dubbeldag: bool | None = None
    clear_child_id: bool = False
    clear_last_year: bool = False
    clear_parent_personal_data_uuid: bool = False
    clear_name_message_key: bool = False
    clear_target_days: bool = False


# What `sync_with_wish` may do to one already-persisted `planned_leave_day`
# row before running the sync — see that method's docstring for when the
# calendar uses which. Restructure step 3b: "pin"/"unpin" are `DayAssertion`
# writes now (`PlanningService._assert_pin`/`_clear_pin`), not a column flip
# — there is no `is_pinned` column left on `planned_leave_day`.
# - 'attach' — re-point the row at this call's wish AND assert it (the
#   "promote, locked" path: keep these exact days, just under the new wish).
# - 'pin' / 'unpin' — assert/retract at the row's own key, `wish_id`
#   untouched (editing an existing wish-backed band's Planering ladder).
# - 'detach' — delete the row outright and retract any assertion at its key
#   (hand it back to the planner to refill from scratch).
_DAY_LOCK_ACTIONS = frozenset({"attach", "pin", "unpin", "detach"})


@dataclass(frozen=True)
class DayLockAction:
    planned_day_id: UUID
    action: str  # one of `_DAY_LOCK_ACTIONS`


@dataclass(frozen=True)
class WishSyncResult:
    """`sync_with_wish`'s return value: the wish as created/updated, plus the

    same `PlanSyncResult` `sync_plan` itself returns. `wish` reflects the
    fields the caller asked for even when `sync.apply_result.accepted` is
    `False` — the whole point of the atomic call is that in that case the
    wish (and every day-lock action) was rolled back and does NOT exist in
    the database, but the caller (the calendar's toast/undo UI) still needs
    the would-be id/fields to render the outcome, e.g. matching `sync.
    proposal.outcomes` by `wish_id`. Always check `sync.apply_result.
    accepted` before treating `wish` as persisted.
    """

    wish: LeaveWish
    sync: PlanSyncResult


class _SyncWithWishRejected(Exception):
    """Internal control-flow only — never escapes `sync_with_wish`.

    Raised inside `sync_with_wish`'s `unit_of_work()` block the moment the
    inner `sync_plan` call comes back `accepted=False`, so the surrounding
    `unit_of_work()` rolls back the wish create/update and every day-lock
    action along with it — without this, `sync_plan` reporting a rejection
    is an ordinary return value, not an exception, and the transaction would
    otherwise commit the wish anyway (exactly the orphan-wish bug this
    endpoint exists to make structurally impossible).
    """

    def __init__(self, wish: LeaveWish, result: PlanSyncResult) -> None:
        super().__init__("sync_plan rejected the batch; rolling back the wish/day changes too")
        self.wish = wish
        self.result = result


class PlanningService:
    def __init__(
        self,
        repo: PlannedLeaveRepository,
        analytics_service: AnalyticsService,
        parental_leave_repo: ParentalLeaveRepository,
        vacation_repo: VacationDaysRepository,
        planning_engine_service: PlanningEngineService | None = None,
        uow: UnitOfWork | None = None,
        workspace_repo: WorkspaceRepository | None = None,
        day_override_repo: DayOverrideRepository | None = None,
        leave_wish_service: LeaveWishService | None = None,
        day_assertion_repo: DayAssertionRepository | None = None,
    ) -> None:
        self._repo = repo
        self._analytics = analytics_service
        self._parental_leave_repo = parental_leave_repo
        self._vacation_repo = vacation_repo
        # Only needed by `sync_plan` (POST /api/plan/sync) — optional so
        # existing callers (scripts/seed_demo.py, older fixtures) that only
        # ever use the per-day CRUD/`apply_proposed_days` paths don't need to
        # wire up a `PlanningEngineService` they'll never use.
        self._engine = planning_engine_service
        # Defaults to a no-op so pre-existing direct-construction call sites
        # (scripts/seed_demo.py, test fixtures) keep today's per-write-commit
        # behavior; `app.dependencies.get_planning_service` always wires a
        # real `SQLiteUnitOfWork`.
        self._uow = uow or NoOpUnitOfWork()
        # Only needed by `_reject_illegal_free_day` (the hand-pin guard: a
        # free day — weekend/red day, per workspace `DayOverride` — can never
        # be `vacation`, and may only be `sjukpenning`/`lagsta` when the
        # workspace opted in via `take_pl_on_red_days`). Optional for the
        # same reason `_engine` is: existing direct-
        # construction call sites (scripts/seed_demo.py, older test
        # fixtures) don't wire these up and must keep today's behavior
        # unchanged — the guard is a no-op when either is unset.
        self._workspace_repo = workspace_repo
        self._day_override_repo = day_override_repo
        # Only needed by `sync_with_wish` (POST /api/plan/sync-with-wish) —
        # optional for the same reason `_engine` is: existing direct-
        # construction call sites that never hit that path don't need to
        # wire one up.
        self._leave_wish_service = leave_wish_service
        # Restructure step 3b — the ONLY source of "is this day pinned"
        # (`pinned_keys`, consulted by `sync_plan`'s own pre-write snapshot
        # and by the router for display) and what `sync_with_wish`'s
        # 'attach'/'pin'/'unpin' day-lock actions write to. Optional so
        # existing direct-construction call sites that never touch those
        # paths (scripts/seed_demo.py, older test fixtures) don't need to
        # wire one up; `app.dependencies.get_planning_service` always does.
        self._day_assertion_repo = day_assertion_repo

    async def _reject_illegal_free_day(self, workspace_id: UUID, d: date, day_type: str) -> None:
        """Authoritative backend guard behind the calendar's hand-pin path.

        No-op entirely when either optional repo isn't wired, so callers
        that don't pass them (scripts/seed_demo.py, existing test fixtures)
        keep byte-identical behavior to before this guard existed. The rule
        itself lives in `app.services.day_guards.reject_illegal_free_day` —
        shared with `DayAssertionService`, which now owns the hand-pin write
        path too (restructure step 3b) — so it is enforced in exactly one
        place instead of two copies drifting apart.
        """
        if self._workspace_repo is None or self._day_override_repo is None:
            return
        overrides = {
            o.date: o.is_working_day
            for o in await self._day_override_repo.list_by_workspace(workspace_id)
        }
        workspace = await self._workspace_repo.get_by_id(workspace_id)
        reject_illegal_free_day(workspace, overrides, d, day_type)

    async def pinned_keys(self, workspace_id: UUID) -> frozenset[tuple[UUID, UUID, date]]:
        """`(parent_uuid, child_id, date)` for every locked-in-place day in

        this workspace — the ONE function every reader of "is this day
        pinned" in this class calls (`sync_plan`'s pre-write snapshot, and
        the read paths below that annotate `PlannedLeaveDayResponse.
        is_pinned` for the calendar UI's lock icon). See
        `app.entities.day_assertion.partition_assertion_keys`'s docstring
        for why step 3a's two independently-computed readers is exactly the
        bug class this single shared call prevents.
        """
        if self._day_assertion_repo is None:
            return frozenset()
        assertions = await self._day_assertion_repo.list_by_workspace(workspace_id)
        pinned, _excluded = partition_assertion_keys(assertions)
        return pinned

    async def is_pinned(self, row: PlannedLeaveDay) -> bool:
        """Display-only helper for a single row — see `pinned_keys`."""
        pinned = await self.pinned_keys(row.workspace_id)
        return (row.parent_personal_data_uuid, row.child_id, row.date) in pinned

    async def _assert_pin(self, row: PlannedLeaveDay, wish_id: UUID | None) -> None:
        """Lock `row`'s current `day_type`/`is_dubbeldag` in place via a

        `DayAssertion` at its own (workspace, child, parent, date) key,
        attached to `wish_id` (attachment only, for band naming — see
        `app.entities.day_assertion.DayAssertion`). Used by `sync_with_
        wish`'s 'attach'/'pin' day-lock actions. No-op when no
        `DayAssertionRepository` is wired (existing direct-construction call
        sites that never hit `sync_with_wish`).
        """
        if self._day_assertion_repo is None:
            return
        now = datetime.now()
        await self._day_assertion_repo.upsert(
            DayAssertion(
                id=uuid4(),
                workspace_id=row.workspace_id,
                child_id=row.child_id,
                parent_personal_data_uuid=row.parent_personal_data_uuid,
                date=row.date,
                day_type=row.day_type,
                is_dubbeldag=row.is_dubbeldag,
                wish_id=wish_id,
                created_at=now,
                updated_at=now,
            )
        )

    async def _clear_pin(self, row: PlannedLeaveDay) -> None:
        """Retract any assertion at `row`'s key — see `_assert_pin`."""
        if self._day_assertion_repo is None:
            return
        await self._day_assertion_repo.delete_by_natural_key(
            row.workspace_id, row.child_id, row.parent_personal_data_uuid, row.date
        )

    async def add_planned_day(
        self,
        workspace_id: UUID,
        child_id: UUID,
        parent_personal_data_uuid: UUID,
        date: date,
        day_type: str,
        is_dubbeldag: bool = False,
        wish_id: UUID | None = None,
    ) -> PlannedLeaveDay:
        """Add a planned leave day.

        Validates that the day is available (dubbeldagar rules, no double-booking).
        Purely a derived-row write — "pinned" is never a parameter here any
        more (restructure step 3b): a caller that wants this day locked in
        place asserts it separately, via `DayAssertionService.assert_day`
        (`POST /api/day-assertions`), which mirrors its own write into this
        same table.
        """
        # Validation 0: a free day (weekend/red day) can never be vacation,
        # and may only be parental leave when the workspace opted in — see
        # `_reject_illegal_free_day`.
        await self._reject_illegal_free_day(workspace_id, date, day_type)

        # Validation 1: Check dubbeldagar quota if this is a dubbeldag
        if is_dubbeldag:
            await self._analytics.validate_dubbeldagar(child_id, date)

        # Validation 2: Check no double-booking on the same date for the same parent
        existing = await self._repo.list_by_workspace_child(
            workspace_id=workspace_id,
            child_id=child_id,
            from_date=date,
            to_date=date,
        )
        # Only reject if the same parent already has this date planned
        parent_existing = [
            e for e in existing if e.parent_personal_data_uuid == parent_personal_data_uuid
        ]
        if parent_existing:
            raise ValidationError(f"Parent already has a planned day on {date}")

        now = datetime.now()
        entity = PlannedLeaveDay(
            id=uuid4(),
            workspace_id=workspace_id,
            child_id=child_id,
            parent_personal_data_uuid=parent_personal_data_uuid,
            date=date,
            day_type=day_type,
            is_dubbeldag=is_dubbeldag,
            wish_id=wish_id,
            created_at=now,
            updated_at=now,
        )
        return await self._repo.create(entity)

    async def remove_planned_day(
        self, planned_leave_day_id: UUID, workspace_id: UUID | None = None
    ) -> None:
        """Remove a planned leave day."""
        await self.get(
            planned_leave_day_id, workspace_id
        )  # raises NotFoundError if missing/foreign
        await self._repo.delete(planned_leave_day_id)

    async def list_by_workspace(self, workspace_id: UUID) -> list[PlannedLeaveDay]:
        """Get all planned days for a workspace."""
        return await self._repo.list(workspace_id=workspace_id)

    async def list_by_date_range(
        self,
        workspace_id: UUID,
        child_id: UUID,
        start_date: date,
        end_date: date,
    ) -> list[PlannedLeaveDay]:
        """Get planned days within a date range for a child."""
        return await self._repo.list_by_workspace_child(
            workspace_id=workspace_id,
            child_id=child_id,
            from_date=start_date,
            to_date=end_date,
        )

    async def get(
        self, planned_leave_day_id: UUID, workspace_id: UUID | None = None
    ) -> PlannedLeaveDay:
        """Get a planned day by ID. `workspace_id`, when given, must match the
        row's own — a foreign workspace's row 404s exactly like a nonexistent
        one."""
        entity = await self._repo.get_by_id(planned_leave_day_id)
        if entity is None or (workspace_id is not None and entity.workspace_id != workspace_id):
            raise NotFoundError(f"PlannedLeaveDay {planned_leave_day_id} not found")
        return entity

    async def list_by_child(self, child_id: UUID) -> list[PlannedLeaveDay]:
        """Get all planned days for a child across workspaces."""
        return await self._repo.list(child_id=child_id)

    async def update(
        self,
        planned_leave_day_id: UUID,
        day_type: str | None = None,
        is_dubbeldag: bool | None = None,
        wish_id: UUID | None = None,
        workspace_id: UUID | None = None,
        _clear_wish_id: bool = False,
    ) -> PlannedLeaveDay:
        """Update a planned day (partial update).

        `is_pinned` is not a parameter (restructure step 3b) — see
        `add_planned_day`'s docstring.

        `_clear_wish_id` follows the same convention as `LeaveWishService.
        update`'s `_clear_*` flags: a nullable field's "no change" and "set
        to None" both look like `None` on the wire, so an OMITTED `wish_id`
        (leave unchanged) and an explicit `wish_id=None` (detach from the
        wish) must be told apart by a separate flag rather than conflated —
        callers/routers pass `_clear_wish_id=True` only when the request
        payload explicitly set `wish_id: null`.
        """
        entity = await self.get(planned_leave_day_id, workspace_id)

        if day_type is not None:
            # A free day (weekend/red day) can never be vacation, and may
            # only be parental leave when the workspace opted in — checked
            # against the row's own workspace (authoritative regardless of
            # whether a caller passed `workspace_id` for the 404 check
            # above) and its own date, since only `day_type` is changing
            # here. See `_reject_illegal_free_day`.
            await self._reject_illegal_free_day(entity.workspace_id, entity.date, day_type)
            entity.day_type = day_type
        if is_dubbeldag is not None:
            entity.is_dubbeldag = is_dubbeldag
        if _clear_wish_id:
            entity.wish_id = None
        elif wish_id is not None:
            entity.wish_id = wish_id

        return await self._repo.update(planned_leave_day_id, entity)

    # -- bulk apply (POST /api/plan/apply) --------------------------------
    #
    # This intentionally does NOT reuse `add_planned_day` for the write path.
    # `add_planned_day`'s double-booking check
    # (`list_by_workspace_child(..., from_date=date, to_date=date)`) looks at
    # *any* parent planned on that child+date, not just the one being added,
    # so calling it once per incoming day would misfire on legitimate
    # multi-parent days (e.g. dubbeldagar, where both parents are
    # deliberately planned on the same date) and would abort a re-apply of an
    # already-identical plan instead of treating it as a no-op. A bulk apply
    # also needs to validate the *whole* batch before writing anything ("a
    # half-applied plan is far worse than a rejected one"), which a
    # one-day-at-a-time call cannot do. `validate_dubbeldagar` on
    # `AnalyticsService` *is* reused below, since that check (15-month window)
    # is legitimately per-day and stateless.

    async def apply_proposed_days(
        self,
        workspace_id: UUID,
        days: list[ProposedDayInput],
        replace: bool = False,
        delete_keys: set[tuple[UUID, UUID, date]] | None = None,
    ) -> PlanApplyResult:
        """Persist a batch of proposed days as `planned_leave_day` rows.

        Idempotent per (workspace, child, parent, date): re-applying an
        identical batch creates nothing and is reported as all
        `skipped_identical_count`. Identity here is `day_type`/`is_dubbeldag`
        only — a day whose only difference from what's persisted is
        `wish_id` still counts as `skipped_identical_count`, but is quietly
        written as a metadata-only update rather than left stale;
        it is never treated as a collision. A date already planned
        *differently* (in `day_type`/`is_dubbeldag`) is a collision — by
        default the whole batch is refused and nothing is written; pass
        `replace=True` to overwrite those specific collisions (every other
        collision-free day in the batch is written too).

        `delete_keys` names rows (by their `(child_id, parent_personal_
        data_uuid, date)` identity) to delete as part of this same call —
        e.g. `sync_plan` uses it for rows the freshly-computed plan no longer
        accounts for at all. A key must not also appear in `days`; that is a
        structural error (a batch cannot both write and delete the same
        slot). Deletions are folded into the same up-front validation as the
        creates/replaces: the tier (or dubbeldagar slot) a deleted row frees
        is credited back before checking whether the batch's new demand
        fits, so a sync that frees N days and re-spends N days in the same
        call is validated against the *post-sync* balance, not a stale
        pre-write snapshot. A key with no matching row (already gone) is
        silently ignored rather than treated as an error.

        Validates the entire batch up front (day types, per-day dubbeldagar
        window, batch-wide dubbeldagar quota, child eligibility window,
        remaining sjukpenning/lagsta balance, credited for anything in
        `delete_keys`) and only writes if everything passes. Stages 2-4 below
        (the classification read through the final write) run inside one
        `self._uow()` unit of work — a `BEGIN IMMEDIATE` transaction on the
        shared connection, serialized against any other unit of work via a
        per-connection lock (see `app.db.connection.unit_of_work`). That lock
        + transaction is what actually keeps a rejection, a crash mid-batch,
        or a concurrent apply/sync on the same workspace from ever leaving
        partial writes behind; the up-front validate-then-write split alone
        only ever protected against a *rejected* batch writing anything, not
        a crashed or interleaved one.
        """
        delete_keys = delete_keys or set()

        # -- stage 1: structural validation ------------------------------
        structural_errors: list[Rejection] = []
        for i, d in enumerate(days):
            if d.day_type not in _PERSISTABLE_DAY_TYPES and d.day_type != "free":
                structural_errors.append(
                    Rejection(
                        message_key="planApply.rejection.invalidDayType",
                        data={"index": i, "date": d.date.isoformat(), "day_type": d.day_type},
                    )
                )

        persistable = [d for d in days if d.day_type != "free"]
        skipped_free_count = len(days) - len(persistable)

        deduped: dict[tuple[UUID, UUID, date], ProposedDayInput] = {}
        for d in persistable:
            key = (d.child_id, d.parent_personal_data_uuid, d.date)
            prior = deduped.get(key)
            # Conflict is keyed on day_type/is_dubbeldag only — the same
            # (child, parent, date) appearing twice with a different `wish_id`
            # is not a genuine conflict (attribution metadata, not the
            # persisted day's identity); the later entry simply wins.
            if prior is not None and (
                prior.day_type != d.day_type or prior.is_dubbeldag != d.is_dubbeldag
            ):
                structural_errors.append(
                    Rejection(
                        message_key="planApply.rejection.conflictingDuplicate",
                        data={
                            "date": d.date.isoformat(),
                            "child_id": str(d.child_id),
                            "parent_personal_data_uuid": str(d.parent_personal_data_uuid),
                        },
                    )
                )
            deduped[key] = d

        for key in delete_keys:
            if key in deduped:
                child_id, parent_uuid, day = key
                structural_errors.append(
                    Rejection(
                        message_key="planApply.rejection.deleteKeyConflict",
                        data={
                            "date": day.isoformat(),
                            "child_id": str(child_id),
                            "parent_personal_data_uuid": str(parent_uuid),
                        },
                    )
                )

        if structural_errors:
            return PlanApplyResult(accepted=False, rejections=structural_errors)

        unique_days = list(deduped.values())
        if not unique_days and not delete_keys:
            return PlanApplyResult(accepted=True, skipped_free_count=skipped_free_count)

        # Stages 2-4 run inside one unit of work: a concurrent apply/sync for
        # this same workspace blocks at `BEGIN IMMEDIATE` (inside `self._uow()`)
        # until this call commits or rolls back, then re-reads `existing_by_key`
        # against the now-committed state instead of racing this call's reads.
        async with self._uow():
            # -- stage 2: classify against existing rows (collision detection) --
            by_child: dict[UUID, list[date]] = {}
            for d in unique_days:
                by_child.setdefault(d.child_id, []).append(d.date)
            for child_id, _parent_uuid, day in delete_keys:
                by_child.setdefault(child_id, []).append(day)

            existing_by_key: dict[tuple[UUID, UUID, date], PlannedLeaveDay] = {}
            for child_id, dates in by_child.items():
                existing_rows = await self._repo.list_by_workspace_child(
                    workspace_id=workspace_id,
                    child_id=child_id,
                    from_date=min(dates),
                    to_date=max(dates),
                )
                for row in existing_rows:
                    existing_by_key[(row.child_id, row.parent_personal_data_uuid, row.date)] = row

            # Rows named by `delete_keys` that actually exist — a key with no
            # matching row (already gone) is silently ignored.
            delete_rows: list[PlannedLeaveDay] = [
                existing_by_key[key] for key in delete_keys if key in existing_by_key
            ]

            to_create: list[ProposedDayInput] = []
            to_replace: list[tuple[PlannedLeaveDay, ProposedDayInput]] = []
            to_update_metadata: list[tuple[PlannedLeaveDay, ProposedDayInput]] = []
            collisions: list[DayCollision] = []
            skipped_identical_count = 0

            for d in unique_days:
                key = (d.child_id, d.parent_personal_data_uuid, d.date)
                existing = existing_by_key.get(key)
                if existing is None:
                    to_create.append(d)
                    continue
                if existing.day_type == d.day_type and existing.is_dubbeldag == d.is_dubbeldag:
                    skipped_identical_count += 1
                    # Identical-vs-collision is (and stays) keyed on
                    # day_type/is_dubbeldag only. A row whose only difference is
                    # `wish_id` is never a collision requiring `replace=True`
                    # — it's a plain metadata update, written below without
                    # touching `skipped_identical_count` or `collisions`.
                    if existing.wish_id != d.wish_id:
                        to_update_metadata.append((existing, d))
                    continue
                collisions.append(
                    DayCollision(
                        date=d.date,
                        parent_personal_data_uuid=d.parent_personal_data_uuid,
                        child_id=d.child_id,
                        existing_day_type=existing.day_type,
                        proposed_day_type=d.day_type,
                    )
                )
                if replace:
                    to_replace.append((existing, d))

            if collisions and not replace:
                return PlanApplyResult(
                    accepted=False,
                    collisions=collisions,
                    rejections=[
                        Rejection(
                            message_key="planApply.rejection.collisionsNeedReplace",
                            data={"count": len(collisions)},
                        )
                    ],
                )

            pending = to_create + [new for _, new in to_replace]
            if not pending and not to_update_metadata and not delete_rows:
                # Nothing left to write — either an empty batch or a fully
                # idempotent re-apply.
                return PlanApplyResult(
                    accepted=True,
                    skipped_identical_count=skipped_identical_count,
                    skipped_free_count=skipped_free_count,
                )

            # -- stage 3: domain validation on the batch that would actually write
            #
            # `delete_rows` is threaded into both checks below so a row this same
            # call is about to delete is credited back to its tier/quota BEFORE
            # the new demand is checked against what's left — the bug this
            # `delete_keys` parameter exists to fix was validating against a
            # pre-write snapshot that hadn't yet had that capacity freed.
            domain_errors: list[str] = []
            domain_errors.extend(
                await self._validate_dubbeldagar_batch(pending, to_replace, delete_rows)
            )
            domain_errors.extend(await self._validate_eligible_range(pending))
            domain_errors.extend(
                await self._validate_remaining_balance(pending, to_replace, delete_rows)
            )
            domain_errors.extend(
                await self._validate_remaining_vacation_balance(pending, to_replace, delete_rows)
            )

            if domain_errors:
                return PlanApplyResult(
                    accepted=False, collisions=collisions, rejections=domain_errors
                )

            # -- stage 4: write, batched (H1: one executemany per operation
            # instead of one commit per row) --------------------------------
            delete_ids = [old.id for old, _ in to_replace] + [row.id for row in delete_rows]
            if delete_ids:
                await self._repo.delete_many(delete_ids)

            now = datetime.now()
            to_insert: list[PlannedLeaveDay] = [
                PlannedLeaveDay(
                    id=uuid4(),
                    workspace_id=workspace_id,
                    child_id=d.child_id,
                    parent_personal_data_uuid=d.parent_personal_data_uuid,
                    date=d.date,
                    day_type=d.day_type,
                    is_dubbeldag=d.is_dubbeldag,
                    wish_id=d.wish_id,
                    created_at=now,
                    updated_at=now,
                )
                for d in pending
            ]
            if to_insert:
                await self._repo.create_many(to_insert)
            created = to_insert

            if to_update_metadata:
                metadata_pairs: list[tuple[UUID, UUID | None]] = []
                for existing, d in to_update_metadata:
                    existing.wish_id = d.wish_id
                    metadata_pairs.append((existing.id, d.wish_id))
                await self._repo.update_metadata_many(metadata_pairs)

            return PlanApplyResult(
                accepted=True,
                created=created,
                created_count=len(to_create),
                skipped_identical_count=skipped_identical_count,
                skipped_free_count=skipped_free_count,
                replaced_count=len(to_replace),
                deleted_count=len(delete_rows),
            )

    # -- sync (POST /api/plan/sync) ---------------------------------------
    #
    # Recomputes and persists a workspace's plan in one step, then reports
    # what changed. Runs `PlanningEngineService.propose(..., only_pinned_
    # existing=True)`, which hides every *unpinned* wish-attributed
    # `planned_leave_day` row from the allocator (see that flag's docstring
    # in `allocation_service.py`) so the auto-planned portion of the plan is
    # re-derived from scratch each call — this is what lets a re-plan
    # legitimately move or retype previously-auto-placed days, while a
    # pinned day (fed to the allocator unchanged, and debited off the top by
    # `_Ledger._reserve_pinned_days`) can never be reassigned or overwritten.
    #
    # Pinned days therefore never appear in `proposal.days` in the first
    # place (the allocator never emits anything for an already-occupied
    # key) -- "no collision ever arises" is a property of the allocator, not
    # something this method has to special-case in the write path. The one
    # extra thing this method computes beyond `proposal.days` is which rows
    # fall out of the freshly-computed plan entirely (a wish's window moved
    # away from a date it used to occupy, or the wish was deleted) --
    # `apply_proposed_days` only ever creates/replaces/updates keys present
    # in its own batch, so nothing else would identify those. Those keys are
    # handed to `apply_proposed_days` as `delete_keys` rather than deleted
    # here directly, so the capacity they free is credited back during that
    # same call's up-front validation — deleting them first (or after) would
    # let a sync that frees N days and re-spends N days read the freed days
    # as still-spent net-new demand and reject the whole batch.

    async def sync_plan(
        self,
        workspace_id: UUID,
        today: date | None = None,
    ) -> PlanSyncResult:
        """Propose, persist, and diff — the single call the direct-

        manipulation calendar UX needs (no more separate propose/apply
        steps). The diff (`changes`) is computed from a pre-write snapshot
        read here, before any write, compared against what's actually
        persisted afterwards — never reconstructed from the client side or
        guessed from `apply_proposed_days`'s own (coarser) counters.

        **Displacement is informational, never a refusal.** Not every stale
        key (see the module comment above) means the same thing: a wish the
        user just edited to claim fewer/different dates sheds its OWN old
        rows as 'removed' `changes` — self-caused, expected. A stale row
        whose wish was NOT touched, still wants exactly what it always
        wanted, but lost the priority race for a shared pool to some OTHER
        wish's edit, is different — "genuinely evicted". `_genuinely_evicted_
        rows` tells the two apart by checking, for each stale row, whether
        ITS OWN wish's fresh `WindowOutcome` for that date now reports
        `shortfall_days > 0` — the wish still wants that capacity and didn't
        get it, as opposed to no longer wanting it at all.

        Step 2 of the intent-vs-derived-output restructure removed the
        earlier behavior where a genuine eviction refused to write anything
        at all pending a caller-supplied `eviction_choice` — no caller ever
        supplied one, so that refusal was permanent and silent (a background
        sync from anywhere other than Settings had nowhere to surface it).
        The allocator's own model is already "shortfall, not failure" (a
        `WindowOutcome` reports `shortfall_days` rather than refusing to
        propose); eviction now follows the same rule. This call always
        writes every stale key's deletion alongside the fresh batch — a
        genuinely evicted row is reported back as a 'removed' `PlanChange`
        with `evicted=True` (see that field's docstring) so the UI can call
        it out distinctly, but it never blocks the write.

        Pinned days are structurally excluded from `stale_keys` itself (see
        below), so a pinned row can never be displaced. A row dated
        strictly before `resolved_today` is excluded the same way -- "the
        past is fact": the allocator's fixed/flexible window expansion clips
        every window to `>= today` (`_expand_all_windows`/
        `_expand_flexible_wish_years` in `allocation_service.py`), so a
        wish's already-elapsed dates are never present in `proposal.days`
        no matter how genuinely that wish still "owns" them -- without this
        exclusion they'd always look stale and get silently deleted the
        first time `sync_plan` runs after they elapse. `today` itself is
        NOT past (still inside the allocator's own `>= today` window and
        therefore still legitimately re-plannable in the same call), so
        only `row.date < resolved_today` is frozen, not `<=`.
        """
        if self._engine is None:
            raise ValidationError("PlanningService.sync_plan requires a PlanningEngineService")
        resolved_today = today or date.today()

        # One unit of work spans the pre-write snapshot through the post-write
        # re-read, not just `apply_proposed_days`'s own writes — otherwise a
        # concurrent sync could commit between this call's `before` read and
        # its (nested, no-op-passthrough — see `unit_of_work`'s docstring)
        # write, and this call's `changes` diff would be computed against a
        # snapshot that was already stale by the time it read `before`.
        async with self._uow():
            before = await self._repo.list(workspace_id=workspace_id)
            before_by_key: dict[tuple[UUID, UUID, date], PlannedLeaveDay] = {
                (d.child_id, d.parent_personal_data_uuid, d.date): d for d in before
            }
            # Restructure step 3b — `self.pinned_keys` (a single, shared
            # `DayAssertion` read — see its docstring) is the SAME call
            # `AllocationInputLoader.load()`'s `only_pinned_existing=True`
            # filter effectively bases its own immovable-day set on, so this
            # snapshot can never disagree with what `self._engine.propose(...,
            # only_pinned_existing=True)` just below treated as immovable —
            # the exact bug class step 3a's two independent readers caused
            # (see the restructure notes' "Bug found in 3a").
            raw_pinned = await self.pinned_keys(workspace_id)
            pinned_keys = {(child_id, parent_uuid, d) for (parent_uuid, child_id, d) in raw_pinned}

            proposal = await self._engine.propose(
                workspace_id, today=today, only_pinned_existing=True
            )

            batch = [
                ProposedDayInput(
                    date=d.date,
                    parent_personal_data_uuid=d.parent_personal_data_uuid,
                    child_id=d.child_id,
                    day_type=d.day_type,
                    is_dubbeldag=d.is_dubbeldag,
                    wish_id=d.wish_id,
                )
                for d in proposal.days
                if d.day_type != "free"
                and d.child_id is not None
                # Defense-in-depth only: the allocator never proposes anything
                # for an already-occupied key (pinned rows stay in
                # `existing_planned` — see `only_pinned_existing` above), so this
                # should never actually filter anything out.
                and (d.child_id, d.parent_personal_data_uuid, d.date) not in pinned_keys
            ]

            batch_keys = {(d.child_id, d.parent_personal_data_uuid, d.date) for d in batch}
            # Rows the freshly-computed plan no longer accounts for at all:
            # unpinned, wish-attributed, dated `>= resolved_today` (see the
            # "past is fact" note in this method's docstring), and absent
            # from this run's batch — e.g. the wish was deleted, or its
            # window moved elsewhere. Passed as `delete_keys` so
            # `apply_proposed_days` validates the deletions together with
            # the creates/replaces (see the comment above this method)
            # rather than deleting them itself afterwards.
            stale_keys = {
                key
                for key, row in before_by_key.items()
                if key not in pinned_keys
                and row.wish_id is not None
                and row.date >= resolved_today
                and key not in batch_keys
            }
            stale_rows = [before_by_key[key] for key in stale_keys]

            # Genuinely evicted: a stale row whose OWN wish still wants that
            # exact capacity (fresh `shortfall_days > 0`) but lost it to a
            # different wish's edit — see this method's docstring. Every
            # other stale row is "self-caused" (the wish itself moved/
            # shrank/was retargeted). Both kinds are always deleted — the
            # split only matters for how the resulting `PlanChange` is
            # flagged, never for whether the write happens.
            evicted_rows = self._genuinely_evicted_rows(stale_rows, proposal.outcomes)
            evicted_keys = {self._row_key(row) for row in evicted_rows}
            delete_keys = stale_keys

            apply_result = await self.apply_proposed_days(
                workspace_id=workspace_id, days=batch, replace=True, delete_keys=delete_keys
            )

            changes: list[PlanChange] = []
            if apply_result.accepted:
                after = await self._repo.list(workspace_id=workspace_id)
                after_by_key: dict[tuple[UUID, UUID, date], PlannedLeaveDay] = {
                    (d.child_id, d.parent_personal_data_uuid, d.date): d for d in after
                }

                for key in sorted(
                    set(before_by_key) | set(after_by_key),
                    key=lambda k: (k[2], str(k[0]), str(k[1])),
                ):
                    child_id, parent_uuid, d = key
                    old = before_by_key.get(key)
                    new = after_by_key.get(key)
                    if old is None and new is not None:
                        changes.append(
                            PlanChange(
                                kind="added",
                                date=d,
                                parent_personal_data_uuid=parent_uuid,
                                child_id=child_id,
                                old_day_type=None,
                                new_day_type=new.day_type,
                                old_is_dubbeldag=None,
                                new_is_dubbeldag=new.is_dubbeldag,
                                old_wish_id=None,
                                new_wish_id=new.wish_id,
                            )
                        )
                    elif old is not None and new is None:
                        changes.append(
                            PlanChange(
                                kind="removed",
                                date=d,
                                parent_personal_data_uuid=parent_uuid,
                                child_id=child_id,
                                old_day_type=old.day_type,
                                new_day_type=None,
                                old_is_dubbeldag=old.is_dubbeldag,
                                new_is_dubbeldag=None,
                                old_wish_id=old.wish_id,
                                new_wish_id=None,
                                evicted=key in evicted_keys,
                            )
                        )
                    elif old is not None and new is not None:
                        if old.day_type != new.day_type or old.is_dubbeldag != new.is_dubbeldag:
                            changes.append(
                                PlanChange(
                                    kind="retyped",
                                    date=d,
                                    parent_personal_data_uuid=parent_uuid,
                                    child_id=child_id,
                                    old_day_type=old.day_type,
                                    new_day_type=new.day_type,
                                    old_is_dubbeldag=old.is_dubbeldag,
                                    new_is_dubbeldag=new.is_dubbeldag,
                                    old_wish_id=old.wish_id,
                                    new_wish_id=new.wish_id,
                                )
                            )
                        elif old.wish_id != new.wish_id:
                            # A same-date collision between two DIFFERENT
                            # wishes never shows up in `stale_keys` (the key
                            # survives — only `wish_id` changes), so without
                            # this check it would silently look like an
                            # ordinary metadata update instead of the OLD
                            # wish genuinely losing this date to the new
                            # one. Reuses the exact same "does the row's own
                            # wish still have unmet shortfall here" test
                            # `_genuinely_evicted_rows` applies to deleted
                            # stale rows.
                            reattributed_evicted = bool(
                                self._genuinely_evicted_rows([old], proposal.outcomes)
                            )
                            changes.append(
                                PlanChange(
                                    kind="reattributed",
                                    date=d,
                                    parent_personal_data_uuid=parent_uuid,
                                    child_id=child_id,
                                    old_day_type=old.day_type,
                                    new_day_type=new.day_type,
                                    old_is_dubbeldag=old.is_dubbeldag,
                                    new_is_dubbeldag=new.is_dubbeldag,
                                    old_wish_id=old.wish_id,
                                    new_wish_id=new.wish_id,
                                    evicted=reattributed_evicted,
                                )
                            )

        return PlanSyncResult(
            proposal=proposal,
            apply_result=apply_result,
            changes=changes,
        )

    # -- sync-with-wish (POST /api/plan/sync-with-wish) --------------------
    #
    # Step 1 of the intent-vs-derived-output restructure: promoting a band,
    # drawing a new one, editing a wish-backed one, or picking a preset used
    # to be a client-orchestrated 3-phase transaction (create/patch the wish
    # -> re-point or delete the affected `planned_leave_day` rows -> sync)
    # with compensating rollback written in React. A tab close, crash, or
    # network drop between any two phases left that half-applied — no amount
    # of client-side rollback logic can fix a transaction that isn't one.
    # This method IS the fix: the wish write, every day-lock action, and the
    # sync all run inside one `self._uow()` unit of work, so the frontend
    # needs exactly one call and gets an atomic all-or-nothing result.

    async def sync_with_wish(
        self,
        workspace_id: UUID,
        wish_create: WishCreateInput | None = None,
        wish_update: WishUpdateInput | None = None,
        day_actions: list[DayLockAction] | None = None,
    ) -> WishSyncResult:
        """One atomic call: create-or-update the wish, apply the day-lock/

        detach list, and sync — the single backend call behind all four of
        the calendar's create/edit-then-sync paths:

        - drag-select's default path (`wish_create` only, no `day_actions`);
        - promoting an ad-hoc band (`wish_create` + `day_actions` of
          'attach' — keep the band's exact days, re-pointed at the new wish
          -- or 'detach' -- hand them back to the planner);
        - editing an existing wish-backed band (`wish_update` +
          `day_actions` of 'pin'/'unpin', following the Planering ladder);
        - picking a preset (`wish_create` only).

        Exactly one of `wish_create`/`wish_update` must be given -- this call
        always touches exactly one wish, never zero, never both.

        **Atomicity.** Everything below runs inside one `self._uow()` unit of
        work. `sync_plan` reporting `accepted=False` (a structural batch
        error, or a genuine balance/dubbeldagar-quota/eligibility rejection
        -- eviction alone no longer causes one, see `sync_plan`'s docstring)
        is an ordinary return value, not an exception -- left alone, the
        surrounding unit of work would still commit the wish create/update
        and every day-lock action even though the sync itself wrote nothing,
        recreating exactly the orphan-wish bug this endpoint replaces.
        `_SyncWithWishRejected` exists to force that case to roll back too:
        raised the moment `sync_plan` comes back rejected, caught right
        outside the `async with self._uow():` block (by which point the
        rollback has already happened), and translated back into a normal,
        non-throwing `WishSyncResult` -- a rejection is a structured,
        expected outcome for the UI to render (see `PlanSyncResult`), not an
        HTTP error. A genuine exception (a `ValidationError` from the wish's
        own fields, a `NotFoundError` for a stale `planned_day_id`) is NOT
        caught here -- it propagates normally, the unit of work still rolls
        everything back, and the router maps it to an HTTP error exactly as
        the granular endpoints already do.

        `day_actions` runs BEFORE `sync_plan` so the fresh proposal sees
        their final state -- an 'attach'/'pin' row must already be pinned
        (and, for 'attach', already pointing at the new wish) when `sync_plan`
        reads its pre-write snapshot, or the sync would treat it as an
        ordinary unpinned/stale row and silently retype or displace it; a
        'detach' row must already be gone so the capacity it held is credited
        back before the batch's new demand is checked (see `apply_proposed_
        days`'s `delete_keys` docstring for why order matters there too).
        """
        if self._leave_wish_service is None:
            raise ValidationError("PlanningService.sync_with_wish requires a LeaveWishService")
        if (wish_create is None) == (wish_update is None):
            raise ValidationError("sync_with_wish requires exactly one of wish_create/wish_update")
        for action in day_actions or []:
            if action.action not in _DAY_LOCK_ACTIONS:
                raise ValidationError(f"unknown day lock action: {action.action!r}")

        try:
            async with self._uow():
                if wish_create is not None:
                    wish = await self._leave_wish_service.create(
                        workspace_id=workspace_id,
                        child_id=wish_create.child_id,
                        name=wish_create.name,
                        name_message_key=wish_create.name_message_key,
                        window_start_month=wish_create.window_start_month,
                        window_start_day=wish_create.window_start_day,
                        window_end_month=wish_create.window_end_month,
                        window_end_day=wish_create.window_end_day,
                        target_weeks=wish_create.target_weeks,
                        target_days=wish_create.target_days,
                        recurrence=wish_create.recurrence,
                        first_year=wish_create.first_year,
                        last_year=wish_create.last_year,
                        priority=wish_create.priority,
                        parent_mode=wish_create.parent_mode,
                        parent_personal_data_uuid=wish_create.parent_personal_data_uuid,
                        window_mode=wish_create.window_mode,
                        series_id=wish_create.series_id,
                        wants_dubbeldag=wish_create.wants_dubbeldag,
                    )
                else:
                    assert wish_update is not None  # exactly-one check above
                    wish = await self._leave_wish_service.update(
                        leave_wish_id=wish_update.leave_wish_id,
                        workspace_id=workspace_id,
                        scope=wish_update.scope,
                        child_id=wish_update.child_id,
                        name=wish_update.name,
                        name_message_key=wish_update.name_message_key,
                        window_start_month=wish_update.window_start_month,
                        window_start_day=wish_update.window_start_day,
                        window_end_month=wish_update.window_end_month,
                        window_end_day=wish_update.window_end_day,
                        target_weeks=wish_update.target_weeks,
                        target_days=wish_update.target_days,
                        recurrence=wish_update.recurrence,
                        first_year=wish_update.first_year,
                        last_year=wish_update.last_year,
                        priority=wish_update.priority,
                        parent_mode=wish_update.parent_mode,
                        parent_personal_data_uuid=wish_update.parent_personal_data_uuid,
                        window_mode=wish_update.window_mode,
                        wants_dubbeldag=wish_update.wants_dubbeldag,
                        _clear_child_id=wish_update.clear_child_id,
                        _clear_last_year=wish_update.clear_last_year,
                        _clear_parent_personal_data_uuid=(
                            wish_update.clear_parent_personal_data_uuid
                        ),
                        _clear_name_message_key=wish_update.clear_name_message_key,
                        _clear_target_days=wish_update.clear_target_days,
                    )

                for action in day_actions or []:
                    if action.action == "attach":
                        # Re-point the row at this call's wish (still a plain
                        # `planned_leave_day` field) AND lock it in place —
                        # restructure step 3b expresses "pin" purely as a
                        # `DayAssertion` write, there is no `is_pinned` column
                        # left to flip.
                        row = await self.update(
                            action.planned_day_id, wish_id=wish.id, workspace_id=workspace_id
                        )
                        await self._assert_pin(row, wish_id=wish.id)
                    elif action.action == "pin":
                        row = await self.get(action.planned_day_id, workspace_id=workspace_id)
                        await self._assert_pin(row, wish_id=row.wish_id)
                    elif action.action == "unpin":
                        row = await self.get(action.planned_day_id, workspace_id=workspace_id)
                        await self._clear_pin(row)
                    else:  # "detach"
                        row = await self.get(action.planned_day_id, workspace_id=workspace_id)
                        await self.remove_planned_day(
                            action.planned_day_id, workspace_id=workspace_id
                        )
                        # Hand the slot back to the planner with a clean
                        # slate — a stray assertion here would keep it
                        # locked/excluded even though the row is gone.
                        await self._clear_pin(row)

                sync_result = await self.sync_plan(workspace_id)
                if not sync_result.apply_result.accepted:
                    raise _SyncWithWishRejected(wish, sync_result)
        except _SyncWithWishRejected as rejected:
            return WishSyncResult(wish=rejected.wish, sync=rejected.result)

        return WishSyncResult(wish=wish, sync=sync_result)

    @staticmethod
    def _row_key(row: PlannedLeaveDay) -> tuple[UUID, UUID, date]:
        return (row.child_id, row.parent_personal_data_uuid, row.date)

    @staticmethod
    def _genuinely_evicted_rows(
        stale_rows: list[PlannedLeaveDay], outcomes: list[WindowOutcome]
    ) -> list[PlannedLeaveDay]:
        """Split "genuinely evicted" (blame a different wish) from

        "self-caused" (the row's own wish moved/shrank and no longer wants
        it) — see `sync_plan`'s docstring. A stale row is genuinely evicted
        only when its own `wish_id` has a fresh `WindowOutcome` covering
        that exact date (`start_date <= row.date <= end_date`) that reports
        `shortfall_days > 0`: the wish still wants that day and didn't get
        it. No matching outcome at all (the wish's window moved away from
        this date entirely) is always self-caused, never eviction.
        """
        outcomes_by_wish: dict[UUID, list[WindowOutcome]] = {}
        for outcome in outcomes:
            outcomes_by_wish.setdefault(outcome.wish_id, []).append(outcome)

        evicted: list[PlannedLeaveDay] = []
        for row in stale_rows:
            if row.wish_id is None:
                continue  # defensive: stale_keys already excludes these
            matching = [
                o
                for o in outcomes_by_wish.get(row.wish_id, [])
                if o.start_date <= row.date <= o.end_date
            ]
            if any(o.shortfall_days > 0 for o in matching):
                evicted.append(row)
        return evicted

    async def _validate_dubbeldagar_batch(
        self,
        pending: list[ProposedDayInput],
        to_replace: list[tuple[PlannedLeaveDay, ProposedDayInput]],
        delete_rows: list[PlannedLeaveDay],
    ) -> list[Rejection]:
        """Per-day window check (reused from AnalyticsService) + batch-wide quota.

        `AnalyticsService.validate_dubbeldagar` only sees rows already
        committed to the DB, so it cannot by itself catch a batch that
        introduces more than `max_dubbeldagar` new dubbeldagar in one go —
        that aggregate check is done here, on top of it. `delete_rows`
        (rows this same call is about to delete, via `delete_keys`) free
        their dubbeldagar slot just like a `to_replace` old row does, so
        they're credited back the same way when projecting the total.
        """
        errors: list[Rejection] = []
        dubbeldag_days = [d for d in pending if d.is_dubbeldag]
        if not dubbeldag_days:
            return errors

        for d in dubbeldag_days:
            try:
                await self._analytics.validate_dubbeldagar(d.child_id, d.date)
            except DubbeldagError as exc:
                message_key = exc.message_key or "planApply.rejection.dubbeldagInvalid"
                data: dict[str, object] = {**exc.data, "date": d.date.isoformat()}
                errors.append(Rejection(message_key=message_key, data=data))

        replaced_out_dubbeldag_by_child: dict[UUID, int] = {}
        for old, _new in to_replace:
            if old.is_dubbeldag:
                replaced_out_dubbeldag_by_child[old.child_id] = (
                    replaced_out_dubbeldag_by_child.get(old.child_id, 0) + 1
                )
        for row in delete_rows:
            if row.is_dubbeldag:
                replaced_out_dubbeldag_by_child[row.child_id] = (
                    replaced_out_dubbeldag_by_child.get(row.child_id, 0) + 1
                )

        by_child: dict[UUID, int] = {}
        for d in dubbeldag_days:
            by_child[d.child_id] = by_child.get(d.child_id, 0) + 1

        for child_id, new_count in by_child.items():
            already = await self._repo.count_dubbeldagar(child_id)
            replaced_out = replaced_out_dubbeldag_by_child.get(child_id, 0)
            projected_total = already - replaced_out + new_count
            if projected_total > settings.rules.max_dubbeldagar:
                errors.append(
                    Rejection(
                        message_key="planApply.rejection.dubbeldagBatchQuotaExceeded",
                        data={
                            "child_id": str(child_id),
                            "projected_total": projected_total,
                            "max_dubbeldagar": settings.rules.max_dubbeldagar,
                        },
                    )
                )
        return errors

    async def _validate_eligible_range(self, pending: list[ProposedDayInput]) -> list[Rejection]:
        """Every date must fall within the child's eligible window.

        Lower bound is the child's birth date; upper bound is the hard
        expiry (`LeaveRules.hard_expiry_age_years`, default 12) after which
        no FP days may be used at all, per Swedish law.
        """
        errors: list[Rejection] = []
        birth_dates: dict[UUID, date] = {}

        for d in pending:
            if d.child_id not in birth_dates:
                projection = await self._analytics.get_days_left_at_age_4(
                    d.parent_personal_data_uuid, d.child_id
                )
                birth_dates[d.child_id] = projection.birth_date

            birth_date = birth_dates[d.child_id]
            hard_expiry = birth_date + relativedelta(years=settings.rules.hard_expiry_age_years)
            if d.date < birth_date or d.date > hard_expiry:
                errors.append(
                    Rejection(
                        message_key="planApply.rejection.outsideEligibleWindow",
                        data={
                            "date": d.date.isoformat(),
                            "child_id": str(d.child_id),
                            "birth_date": birth_date.isoformat(),
                            "hard_expiry": hard_expiry.isoformat(),
                        },
                    )
                )
        return errors

    async def _tier_pool_left(self, parent_uuid: UUID, child_id: UUID) -> tuple[int, int]:
        """Remaining (sjukpenning_left, lagsta_left), net of already-planned rows.

        Mirrors the pool computation in
        `allocation_service.AllocationInputLoader.load` (same FK-snapshot-or-
        rule-default, minus locally committed days) so this check agrees with
        what the planning engine itself would have allowed.
        """
        rules = settings.rules
        latest = await self._parental_leave_repo.get_latest_by_parent(parent_uuid, child_id)
        if latest is not None:
            sjuk_total_left = latest.tier_data.sjukpenning_left
            lagsta_total_left = latest.tier_data.lagsta_niva_left
        else:
            # No FK snapshot yet: default this parent's share via the shared
            # helper — see `default_tier_pool_share` docstring (Bug A fix:
            # this and `AllocationInputLoader.load` used to disagree, 390/90
            # vs. 195/45). Parent count follows the child's actual workspace
            # (solo parent gets the full pool, not half of it).
            parent_count = await self._analytics.get_parent_count_for_child(child_id)
            sjuk_total_left, lagsta_total_left = default_tier_pool_share(rules, parent_count)

        committed_sjuk = await self._repo.count_by_parent_type(parent_uuid, child_id, "sjukpenning")
        committed_lagsta = await self._repo.count_by_parent_type(parent_uuid, child_id, "lagsta")
        return (
            max(sjuk_total_left - committed_sjuk, 0),
            max(lagsta_total_left - committed_lagsta, 0),
        )

    async def _validate_remaining_balance(
        self,
        pending: list[ProposedDayInput],
        to_replace: list[tuple[PlannedLeaveDay, ProposedDayInput]],
        delete_rows: list[PlannedLeaveDay],
    ) -> list[Rejection]:
        """The batch must not spend more sjukpenning/lagsta days than remain.

        Per (parent, child, tier): net_new = (new days of that tier the batch
        adds) - (days of that tier being freed: replaced away, via
        `to_replace`, or deleted outright via `delete_rows`/`delete_keys`).
        Rejects if net_new exceeds what's left. Crediting `delete_rows` here
        is what lets a `sync_plan` call that frees N days and re-spends N
        days elsewhere pass validation against the balance *after* those
        deletions, instead of rejecting against a stale pre-write snapshot.
        """
        errors: list[Rejection] = []
        pairs = {(d.parent_personal_data_uuid, d.child_id) for d in pending}

        for parent_uuid, child_id in pairs:
            sjuk_left, lagsta_left = await self._tier_pool_left(parent_uuid, child_id)
            for tier, left in (("sjukpenning", sjuk_left), ("lagsta", lagsta_left)):
                added = sum(
                    1
                    for d in pending
                    if d.parent_personal_data_uuid == parent_uuid
                    and d.child_id == child_id
                    and d.day_type == tier
                )
                removed = sum(
                    1
                    for old, new in to_replace
                    if new.parent_personal_data_uuid == parent_uuid
                    and new.child_id == child_id
                    and old.day_type == tier
                ) + sum(
                    1
                    for row in delete_rows
                    if row.parent_personal_data_uuid == parent_uuid
                    and row.child_id == child_id
                    and row.day_type == tier
                )
                net_new = added - removed
                if net_new > left:
                    errors.append(
                        Rejection(
                            message_key="planApply.rejection.insufficientBalance",
                            data={
                                "parent_personal_data_uuid": str(parent_uuid),
                                "child_id": str(child_id),
                                "tier": tier,
                                "needed": net_new,
                                "remaining": left,
                            },
                        )
                    )
        return errors

    async def _vacation_days_left(self, parent_uuid: UUID, year: int) -> int:
        """Remaining vacation (semester) days for one parent in one calendar year.

        Vacation is budgeted per (parent, calendar year) — unlike the FP
        tiers above, which are per (parent, child) — because a semesterlagen
        entitlement doesn't care which child a day off is attributed to.

        Mirrors `AllocationInputLoader.load`'s vacation-budget derivation in
        `allocation_service.py` exactly: `available = total_days +
        saved_days_in - already_used`, where `total_days`/`saved_days_in`
        come from the `VacationDays` row for (parent, year) if one exists.
        A year with **no** row does NOT mean zero entitlement — the loader
        defaults `total_days` to `rules.statutory_vacation_days_per_year`
        (25) and flags the result `is_assumed=True`; this does the same
        default (there is no `is_assumed` to carry here since this is a
        plain remaining-days count, not a `VacationBudget`).
        `PersonalData.min_vacation_days_per_year` is deliberately NOT used
        here: per `ParentFacts.min_vacation_days_per_year`'s docstring and
        the allocator's own carry-forward remedy, it is a semesterlagen
        floor on days that must be *taken* each year (bounding what may be
        saved forward), not a cap on how much may be spent — turning it into
        a spending cap here would reject perfectly legal plans.

        Like `_tier_pool_left` above, "already used" counts every currently
        persisted 'vacation' row for the (parent, year), pinned or not — this
        intentionally does NOT mirror the pure `_Ledger`'s pinned-day
        exclusion (that exclusion exists there only because the `_Ledger`
        debits a pinned day's cost itself in a separate step); this check has
        no such separate step, so it must count everything actually
        persisted, the same "whatever is committed reduces what's left"
        shape the sjukpenning/lagsta check above already uses.
        """
        rules = settings.rules
        row = await self._vacation_repo.get_by_parent_year(parent_uuid, year)
        if row is not None:
            total_available = row.total_days + row.saved_days_in
        else:
            total_available = rules.statutory_vacation_days_per_year

        committed = [
            d
            for d in await self._repo.list(parent_personal_data_uuid=parent_uuid)
            if d.day_type == "vacation" and d.date.year == year
        ]
        return max(total_available - len(committed), 0)

    async def _validate_remaining_vacation_balance(
        self,
        pending: list[ProposedDayInput],
        to_replace: list[tuple[PlannedLeaveDay, ProposedDayInput]],
        delete_rows: list[PlannedLeaveDay],
    ) -> list[Rejection]:
        """The batch must not spend more vacation days than remain for a year.

        Grouped by (parent, calendar year) rather than (parent, child, tier)
        — see `_vacation_days_left`. Crediting `to_replace`/`delete_rows`
        back before checking follows the exact same shape as
        `_validate_remaining_balance` above, including for a period that
        wraps New Year (e.g. Jullov, 20 Dec - 16 Jan): each `ProposedDayInput`
        is grouped under its own `date.year`, so a band that crosses the
        boundary is automatically split into two independently-checked
        (parent, year) budgets — no special-casing needed here.
        """
        errors: list[Rejection] = []
        year_pairs = {
            (d.parent_personal_data_uuid, d.date.year) for d in pending if d.day_type == "vacation"
        }

        for parent_uuid, year in year_pairs:
            left = await self._vacation_days_left(parent_uuid, year)
            added = sum(
                1
                for d in pending
                if d.parent_personal_data_uuid == parent_uuid
                and d.day_type == "vacation"
                and d.date.year == year
            )
            removed = sum(
                1
                for old, new in to_replace
                if new.parent_personal_data_uuid == parent_uuid
                and old.day_type == "vacation"
                and new.date.year == year
            ) + sum(
                1
                for row in delete_rows
                if row.parent_personal_data_uuid == parent_uuid
                and row.day_type == "vacation"
                and row.date.year == year
            )
            net_new = added - removed
            if net_new > left:
                errors.append(
                    Rejection(
                        message_key="planApply.rejection.insufficientVacationBalance",
                        data={
                            "parent_personal_data_uuid": str(parent_uuid),
                            "year": year,
                            "needed": net_new,
                            "remaining": left,
                        },
                    )
                )
        return errors