afterhours-shift-manager/src/slack-bot/app.py

"""Slack Bolt app — /oncall command handlers and interactive actions.

The handler functions are module-level (not closures over ``create_app``) so they
can be unit-tested directly. ``schedule`` (a ``ShiftSchedule``) and
``schedule_channel`` are threaded through as explicit parameters. ``create_app``
is a thin wiring layer that registers the Bolt routes and delegates to them.
"""

import functools
import logging
import os
import re
import time
from datetime import datetime, timedelta
from zoneinfo import ZoneInfo

import boto3
from slack_bolt import App

from shared.blocks import (
    HOLIDAY_ADD_MODAL_CALLBACK,
    OPEN_HOLIDAY_MODAL_ACTION,
    OPEN_OVERRIDE_MODAL_ACTION,
    OVERRIDE_MODAL_CALLBACK,
    build_help_blocks,
    build_holiday_add_modal,
    build_holiday_added_blocks,
    build_home_view,
    build_override_modal,
    build_pay_summary_blocks,
    build_pickup_request_blocks,
    build_pickup_resolved_blocks,
    build_roster_blocks,
    build_shift_change_message,
    build_swap_request_blocks,
    build_swap_resolved_blocks,
    build_week_schedule,
)
from shared.changelog import latest_entry
from shared.ring_scheduler import update_queue_routing
from shared.schedule import (
    FALLBACK_EXTENSION,
    WEEKEND_DAYS,
    ExtensionAlreadyRegistered,
    ShiftSchedule,
    determine_shift_type,
    week_start,
)
from shared.secrets import get_secret
from shared.three_cx_client import ThreeCXClient

logger = logging.getLogger(__name__)
EASTERN = ZoneInfo("America/New_York")

# Activity-bump debounce: don't re-post the schedule to the bottom of the
# channel more than once per this many seconds, so a burst of chatter triggers
# at most one delete+repost. The window is stored as ``last_bump_ts`` on the
# schedule-post record (avoids reading channel history).
SCHEDULE_BUMP_DEBOUNCE_SECONDS = 180

DAY_NAMES = [
    "monday",
    "tuesday",
    "wednesday",
    "thursday",
    "friday",
    "saturday",
    "sunday",
]


# Formats carrying an explicit year — parsed as-is, never rolled forward.
_DATED_FORMATS = (
    "%Y-%m-%d",
    "%m/%d/%Y",
    "%m/%d/%y",
    "%m-%d-%Y",
    "%m-%d-%y",
    "%b %d %Y",
    "%B %d %Y",
    "%b %d %y",
    "%B %d %y",
)
# Year-less formats — stamped with the current year, then rolled to next year
# if already more than a day in the past.
_UNDATED_FORMATS = ("%m/%d", "%m-%d", "%b %d", "%B %d")

_ORDINAL_RE = re.compile(r"(\d+)(st|nd|rd|th)\b")


def parse_date(text: str) -> datetime | None:
    """Parse flexible date input: today, tomorrow, day names, m/d, m/d/yy,
    YYYY-MM-DD, and month-name forms (e.g. ``jul 3``, ``july 3rd 2026``)."""
    now = datetime.now(EASTERN)
    text = text.strip().lower()

    if text == "today":
        return now
    if text == "tomorrow":
        return now + timedelta(days=1)

    if text in DAY_NAMES:
        target = DAY_NAMES.index(text)
        current = now.weekday()
        delta = (target - current) % 7
        if delta == 0:
            delta = 7
        return now + timedelta(days=delta)

    candidate = _ORDINAL_RE.sub(r"\1", text)

    for fmt in _DATED_FORMATS:
        try:
            return datetime.strptime(candidate, fmt).replace(tzinfo=EASTERN)
        except ValueError:
            continue

    for fmt in _UNDATED_FORMATS:
        try:
            parsed = datetime.strptime(candidate, fmt).replace(year=now.year)
            if parsed.replace(tzinfo=EASTERN) < now - timedelta(days=1):
                parsed = parsed.replace(year=now.year + 1)
            return parsed.replace(tzinfo=EASTERN)
        except ValueError:
            continue

    return None


def _update_3cx_routing(extension: str) -> None:
    """Update the 3CX queue to forward calls to the given extension."""
    queue_number = os.environ.get("QUEUE_NUMBER")
    secret_prefix = os.environ.get("TCX_SECRET_PREFIX")
    if not queue_number or not secret_prefix:
        logger.warning("3CX env vars not set — skipping queue update")
        return
    try:
        update_queue_routing(
            extension=extension,
            queue_number=queue_number,
            domain=get_secret(f"{secret_prefix}domain"),
            client_id=get_secret(f"{secret_prefix}client-id"),
            client_secret=get_secret(f"{secret_prefix}client-secret"),
        )
    except Exception:
        logger.exception("Failed to update 3CX queue")


def _make_3cx_client() -> ThreeCXClient | None:
    """Build an OAuth ThreeCXClient from the configured secret prefix, or None.

    Mirrors ``_update_3cx_routing``'s tolerance of a missing config: returns
    None (and logs) rather than raising when ``TCX_SECRET_PREFIX`` is unset, so
    holiday flows degrade gracefully in environments without 3CX wired up.
    """
    secret_prefix = os.environ.get("TCX_SECRET_PREFIX")
    if not secret_prefix:
        logger.warning("3CX env vars not set — skipping 3CX call")
        return None
    return ThreeCXClient(
        domain=get_secret(f"{secret_prefix}domain"),
        auth_mode="oauth",
        client_id=get_secret(f"{secret_prefix}client-id"),
        client_secret=get_secret(f"{secret_prefix}client-secret"),
    )


def _set_holiday_queue_agents(schedule, date_str: str) -> None:
    """Point the holiday queue (802) at the current holiday assignees.

    Used by the inline-activation path and the late-pickup approval path when
    the holiday window is open — the membership must reflect the live assignee
    set. Falls back to ``[FALLBACK_EXTENSION]`` ("100") when no slot is filled.
    Best-effort: any failure is logged and swallowed so the Slack flow still
    completes.
    """
    holiday = schedule.get_holiday(date_str)
    if holiday is None:
        return
    assignees = holiday.get("assignees", {}) or {}
    extensions = list(assignees.keys()) or [FALLBACK_EXTENSION]
    try:
        client = _make_3cx_client()
        if client is None:
            return
        queue_number = schedule.get_holiday_queue()
        queue = client.get_queue(queue_number)
        client.set_queue_agents(queue["Id"], extensions)
    except Exception:
        logger.exception("Failed to set holiday queue agents for %s", date_str)


def _activate_holiday_inline(schedule, date_str: str) -> None:
    """Activate a late-added holiday in-process or via the holiday-router Lambda."""
    from shared.side_effects import activate_holiday_inline

    activate_holiday_inline(schedule, date_str)


def _holiday_schedule_names(date_str: str) -> tuple[str, str]:
    """The (activate, deactivate) one-off schedule names for a holiday date."""
    from shared.side_effects import holiday_schedule_names

    return holiday_schedule_names(date_str)


def _create_holiday_schedules(date_str: str) -> list[str]:
    """Create the two one-off EventBridge schedules for a holiday and return names."""
    from shared.side_effects import create_holiday_schedules

    return create_holiday_schedules(date_str)


def _delete_holiday_schedules(schedule_names: list[str]) -> None:
    """Delete any still-outstanding one-off holiday schedules (best-effort).

    A schedule that has already fired self-deletes (ActionAfterCompletion=DELETE),
    so a ResourceNotFound on delete is expected and ignored.
    """
    if not schedule_names:
        return
    group = os.environ.get("HOLIDAY_SCHEDULE_GROUP", "default")
    try:
        client = boto3.client("scheduler")
    except Exception:
        logger.exception("Could not create scheduler client to delete schedules")
        return
    for name in schedule_names:
        try:
            client.delete_schedule(Name=name, GroupName=group)
        except client.exceptions.ResourceNotFoundException:
            logger.info("Holiday schedule %s already gone — nothing to delete", name)
        except Exception:
            logger.exception("Failed to delete holiday schedule %s", name)


def is_today(date_str: str) -> bool:
    return date_str == datetime.now(EASTERN).strftime("%Y-%m-%d")


def _is_active_shift_type(shift_type: str) -> bool:
    return determine_shift_type() == shift_type


def _shift_start(date_str: str, shift_type: str) -> datetime:
    """The datetime (ET) a shift begins — weekend day at 08:00, otherwise 17:00."""
    d = datetime.strptime(date_str, "%Y-%m-%d").replace(tzinfo=EASTERN)
    return d.replace(hour=8 if shift_type == "day" else 17)


def _shift_started(date_str: str, shift_type: str) -> bool:
    return datetime.now(EASTERN) >= _shift_start(date_str, shift_type)


def _shift_end(date_str: str, shift_type: str) -> datetime:
    """The datetime (ET) a shift ends.

    A day shift (weekend day or holiday, 08:00–17:00 ET) ends at 17:00 the same
    day. A night shift (17:00–08:00 ET) ends at 08:00 the *next* day.
    """
    d = datetime.strptime(date_str, "%Y-%m-%d").replace(tzinfo=EASTERN)
    if shift_type == "day":
        return d.replace(hour=17)
    return d.replace(hour=8) + timedelta(days=1)


def _shift_ended(date_str: str, shift_type: str) -> bool:
    return datetime.now(EASTERN) >= _shift_end(date_str, shift_type)


def _holiday_window_active(date_str: str) -> bool:
    """True when a holiday day-shift window (08:00–17:00 ET) is currently open.

    The 3CX call flow is only repointed to the holiday queue during this window,
    so inline activation (admin add) and late-pickup membership refreshes are
    gated on it.
    """
    now = datetime.now(EASTERN)
    return _shift_start(date_str, "day") <= now < _shift_end(date_str, "day")


def _within_drop_lock(date_str: str, shift_type: str) -> bool:
    """True inside the 24h-before-start window where dropping a shift is locked.

    Within this window a shift can't be abandoned via drop — it must be handed
    off through a verified swap (target accepts) or opened by an admin.
    """
    return datetime.now(EASTERN) >= _shift_start(date_str, shift_type) - timedelta(
        hours=24
    )


def _shift_type_label(day_name: str, shift_type: str) -> str:
    if day_name not in WEEKEND_DAYS:
        return ""
    label = "Day" if shift_type == "day" else "Night"
    return f" ({label})"


def _find_employee_shift(schedule, date_str, day_name, employee_ext):
    """Find which shift type an employee is assigned to on a given date.

    On weekends, checks both day and night shifts. Returns (ext, name, source, shift_type)
    or None if not found on any shift.
    """
    if day_name in WEEKEND_DAYS:
        for st in ("day", "night"):
            ext, name, source = schedule.resolve_shift(date_str, day_name, st)
            if ext == employee_ext:
                return ext, name, source, st
        return None
    ext, name, source = schedule.resolve_shift(date_str, day_name, "night")
    if ext == employee_ext:
        return ext, name, source, "night"
    return None


_DROP_SHIFT_LABELS = {
    "holiday": "holiday",
    "day": "day (8am–5pm)",
    "night": "night (5pm–8am)",
}


def _droppable_shifts(schedule, date_str, day_name, employee_ext) -> list[str]:
    """Return the shift types the employee holds on a date and could drop.

    Possible values, in display priority: ``holiday``, ``day``, ``night``. A
    holiday day-slot supersedes a regular weekend day shift on its date.
    """
    held = []
    holiday_ctx = schedule.get_shift_context(date_str, day_name, "day")
    if holiday_ctx["kind"] == "holiday":
        if any(a["extension"] == employee_ext for a in holiday_ctx["assignees"]):
            held.append("holiday")
    elif day_name in WEEKEND_DAYS:
        ext, _name, _source = schedule.resolve_shift(date_str, day_name, "day")
        if ext == employee_ext:
            held.append("day")
    ext, _name, _source = schedule.resolve_shift(date_str, day_name, "night")
    if ext == employee_ext:
        held.append("night")
    return held


def _schedule_fallback_text() -> str:
    """Notification fallback text for the two-week schedule post."""
    now = datetime.now(EASTERN)
    start = week_start(now)
    end_date = start + timedelta(days=13)
    return (
        f"After-Hours Schedule — {start.strftime('%b %-d')} "
        f"to {end_date.strftime('%b %-d')}"
    )


def _refresh_schedule_post(schedule, schedule_channel, client):
    """Update the tracked schedule message in-place after a shift change."""
    channel = schedule_channel
    if not channel:
        return
    post = schedule.get_schedule_post(channel)
    if not post or not post.get("message_ts"):
        return
    try:
        client.chat_update(
            channel=channel,
            ts=post["message_ts"],
            blocks=build_week_schedule(schedule),
            text=_schedule_fallback_text(),
        )
    except Exception:
        logger.warning("Could not update schedule post", exc_info=True)


def _slack_error_code(exc) -> str | None:
    """Best-effort Slack API error code (e.g. ``not_in_channel``) from an exc."""
    response = getattr(exc, "response", None)
    if response is None:
        return None
    try:
        return response.get("error")
    except Exception:
        return None


def handle_channel_message(event, client, schedule, schedule_channel, retry_num=None):
    """Bump the schedule post to the bottom of the channel on new activity.

    Debounced delete+repost of the tracked post so it stays at the bottom of
    the channel timeline. Idempotent/safe: skips retried Slack deliveries
    (``retry_num``), thread replies, the bot's own / edit-delete messages, when
    there is no stored post, when debounced inside the window, or when the bot
    isn't in the channel.
    """
    # A retry means our first delivery already (likely) did the work but the
    # 200 ack didn't reach Slack in time. Never bump again on a retry — the next
    # real channel message bumps anyway — so a slow run can't double-post.
    if retry_num:
        logger.info("Ignoring retried message event (retry %s)", retry_num)
        return
    if not schedule_channel or event.get("channel") != schedule_channel:
        logger.info(
            "Skipping bump — channel mismatch (expected %s, got %s)",
            schedule_channel,
            event.get("channel"),
        )
        return
    # Ignore the bot's own posts and non-user message events (edits, deletes,
    # joins, …), plus thread replies — a threaded reply doesn't push the
    # schedule down the main timeline, so it isn't worth a delete+repost.
    if event.get("bot_id") or event.get("subtype") or event.get("thread_ts"):
        logger.info(
            "Skipping bump — bot message, subtype %s, or thread reply",
            event.get("subtype"),
        )
        return

    post = schedule.get_schedule_post(schedule_channel)
    if not post or not post.get("message_ts"):
        logger.info("Skipping bump — no stored schedule post for %s", schedule_channel)
        return

    now = time.time()
    last_bump = post.get("last_bump_ts")
    if (
        last_bump is not None
        and now - float(last_bump) < SCHEDULE_BUMP_DEBOUNCE_SECONDS
    ):
        logger.info(
            "Skipping bump — debounced (last bump %.0fs ago, window %ds)",
            now - float(last_bump),
            SCHEDULE_BUMP_DEBOUNCE_SECONDS,
        )
        return

    # Optimistically stamp the debounce window *before* the delete/repost, so a
    # retry fired while this run is still in flight (or after it dies mid-bump)
    # is suppressed and can't orphan a duplicate schedule message.
    schedule.save_schedule_post(
        schedule_channel,
        post["message_ts"],
        post.get("week_start", ""),
        last_bump_ts=now,
    )

    blocks = build_week_schedule(schedule)
    text = _schedule_fallback_text()
    try:
        client.chat_delete(channel=schedule_channel, ts=post["message_ts"])
    except Exception as exc:
        if _slack_error_code(exc) == "not_in_channel":
            logger.info("Bot not in schedule channel; skipping bump")
            return
        # The old message may already be gone — keep going and repost a fresh
        # one so the channel still ends with the schedule.
        logger.info("Could not delete old schedule post for bump", exc_info=True)

    try:
        result = client.chat_postMessage(
            channel=schedule_channel, blocks=blocks, text=text
        )
    except Exception as exc:
        if _slack_error_code(exc) == "not_in_channel":
            logger.info("Bot not in schedule channel; skipping bump")
            return
        logger.warning("Could not repost schedule for bump", exc_info=True)
        return

    schedule.save_schedule_post(
        schedule_channel,
        result["ts"],
        post.get("week_start", ""),
        last_bump_ts=now,
    )
    logger.info(
        "Bumped schedule post to bottom of %s (ts=%s)", schedule_channel, result["ts"]
    )


# ── Command dispatch ────────────────────────────────────────────────────


def dispatch_oncall(command, respond, client, schedule, schedule_channel):
    """Route a /oncall slash command to the appropriate subcommand handler."""
    text = (command.get("text") or "").strip()
    user_id = command["user_id"]
    channel_id = command["channel_id"]

    is_admin = user_id in schedule.get_admin_users()
    post_channel = schedule_channel or channel_id

    if not text or text == "schedule":
        _show_schedule(respond, schedule)
    elif text == "next":
        _show_next_week(respond, schedule)
    elif text == "help":
        respond(blocks=build_help_blocks(is_admin=is_admin))
    elif text == "roster":
        respond(blocks=build_roster_blocks(schedule.get_roster()))
    elif text == "pay":
        _show_pay(respond, schedule)
    elif text.startswith("rate"):
        _handle_rate(respond, schedule, text, is_admin)
    elif text.startswith("register"):
        _handle_register(respond, schedule, user_id, text)
    elif text.startswith("pick"):
        _handle_pick(
            respond, schedule, user_id, text, post_channel, client, schedule_channel
        )
    elif text.startswith("drop"):
        _handle_drop(
            respond, schedule, user_id, text, post_channel, client, schedule_channel
        )
    elif text.startswith("swap"):
        _handle_swap(
            respond, schedule, user_id, text, post_channel, client, schedule_channel
        )
    elif text.startswith("admin"):
        _handle_admin(
            respond, schedule, user_id, text, is_admin, client, schedule_channel
        )
    else:
        respond(text="Unknown command. Try `/oncall help`")


def handle_pickup(body, respond, client, schedule, schedule_channel):
    """Handle the interactive "pick up open shift" button.

    Routes holiday day-shift buttons to an atomic slot claim and gates pickups
    of an already-started shift behind admin approval (same flow as ``pick``).
    """
    action_id = body["actions"][0]["action_id"]
    remainder = action_id[len("pickup_") :]
    if remainder.endswith("_day"):
        date_str = remainder[:-4]
        shift_type = "day"
    else:
        date_str = remainder
        shift_type = "night"

    user_id = body["user"]["id"]
    channel_id = body["channel"]["id"]

    today_str = datetime.now(EASTERN).strftime("%Y-%m-%d")
    if date_str < today_str:
        client.chat_postEphemeral(
            channel=channel_id,
            user=user_id,
            text="That shift has already passed and can't be picked up.",
        )
        return

    employee = schedule.get_employee_by_slack_id(user_id)
    if not employee:
        client.chat_postEphemeral(
            channel=channel_id,
            user=user_id,
            text="You're not registered. Use `/oncall register <extension>` first.",
        )
        return

    day_name = datetime.strptime(date_str, "%Y-%m-%d").strftime("%A")
    ctx = schedule.get_shift_context(date_str, day_name, shift_type)
    is_holiday = ctx["kind"] == "holiday"

    def _ephemeral(text):
        client.chat_postEphemeral(channel=channel_id, user=user_id, text=text)

    # A pickup after the shift has ended is rejected outright; after it has
    # started (but before it ends) it needs admin approval.
    if _shift_ended(date_str, shift_type):
        _ephemeral("That shift has already ended and can't be picked up.")
        return
    if _shift_started(date_str, shift_type):
        _request_late_pickup(
            _ephemeral_respond(_ephemeral),
            schedule,
            employee,
            date_str,
            shift_type,
            is_holiday=is_holiday,
            client=client,
        )
        return

    if is_holiday:
        claimed = schedule.claim_holiday_slot(
            date_str, employee["extension"], employee["name"]
        )
    else:
        claimed = schedule.claim_open_shift(
            date_str, employee["extension"], employee["name"], shift_type
        )
    if not claimed:
        _ephemeral(f"That shift on *{date_str}* was already picked up by someone else.")
        return

    if not is_holiday and is_today(date_str) and _is_active_shift_type(shift_type):
        _update_3cx_routing(employee["extension"])

    blocks = build_shift_change_message(
        user_id,
        date_str,
        "picked_up",
        employee["extension"],
        employee["name"],
        shift_type=shift_type,
    )
    respond(
        response_type="in_channel",
        replace_original=False,
        blocks=blocks,
        text=f"Shift picked up for {date_str}",
    )
    _refresh_schedule_post(schedule, schedule_channel, client)


def _ephemeral_respond(post_ephemeral):
    """Adapt an ephemeral-poster into a ``respond(text=...)`` callable.

    ``_request_late_pickup`` reports its outcome via ``respond(text=...)``; the
    button path has no ``respond`` that targets the clicker, so this wraps the
    ephemeral poster so the same helper serves both the slash command and the
    button.
    """

    def _respond(text="", **_kwargs):
        if text:
            post_ephemeral(text)

    return _respond


# ── Subcommand handlers ─────────────────────────────────────────────────


def _show_schedule(respond, schedule):
    blocks = build_week_schedule(schedule)
    respond(blocks=blocks)


def _show_next_week(respond, schedule):
    now = datetime.now(EASTERN)
    # The default view is already two weeks, so "next" starts two Sundays ahead.
    next_start = week_start(now) + timedelta(days=14)
    blocks = build_week_schedule(schedule, start_date=next_start)
    respond(blocks=blocks)


def _pay_week_bounds(pay_record: dict) -> tuple[datetime, datetime]:
    """Label bounds for a pay record.

    New rows are Sunday–Saturday. A cutover row may store a shorter
    ``window_start``/``window_end``. Legacy rows are Monday–Sunday via
    ``week_start`` plus six days.
    """
    start = datetime.strptime(
        pay_record.get("window_start") or pay_record["week_start"], "%Y-%m-%d"
    )
    if pay_record.get("window_end"):
        end = datetime.strptime(pay_record["window_end"], "%Y-%m-%d")
    else:
        end = start + timedelta(days=6)
    return start, end


def _show_pay(respond, schedule):
    now = datetime.now(EASTERN)
    # The Monday 7am close writes the Sun–Sat week that ended Saturday. On
    # Sunday, and on Monday before that close, that row is not written yet, so
    # also try the prior week's Sunday key and the legacy Monday keys.
    prev_start = week_start(now) - timedelta(days=7)
    pay_record = None
    for start in (
        prev_start,
        prev_start + timedelta(days=1),
        prev_start - timedelta(days=7),
        prev_start - timedelta(days=6),
    ):
        record = schedule.get_pay_record(start.strftime("%Y-%m-%d"))
        if record and record.get("breakdown"):
            pay_record = record
            break
    if pay_record and pay_record.get("breakdown"):
        label_start, label_end = _pay_week_bounds(pay_record)
        week_label = (
            f"{label_start.strftime('%b %-d')} to {label_end.strftime('%b %-d')}"
        )
        blocks = build_pay_summary_blocks(
            week_label, pay_record["breakdown"], pay_record["totals"]
        )
        respond(blocks=blocks)
    else:
        respond(
            text=f"No pay record found for the week of {prev_start.strftime('%b %-d')}."
        )


def _handle_rate(respond, schedule, text, is_admin):
    # Setting pay rates is an admin-only operation; reading them is gated too
    # since rates are sensitive payroll data.
    if not is_admin:
        respond(text="Rate commands are restricted. Contact an administrator.")
        return

    parts = text.split()
    # /oncall rate — show current rates
    if len(parts) == 1:
        default_rate = schedule.get_shift_rate()
        roster = schedule.get_roster()
        lines = [f"*Default rate:* ${default_rate:.2f}/shift\n"]
        custom = [
            (e["SK"], e.get("name", "Unknown"), float(e["shift_rate"]))
            for e in roster
            if e.get("shift_rate")
        ]
        if custom:
            lines.append("*Per-person rates:*")
            for ext, name, rate in sorted(custom, key=lambda x: x[0]):
                lines.append(f"• {name} (Ext {ext}) — ${rate:.2f}/shift")
        else:
            lines.append("_No per-person rates set — everyone uses the default._")
        respond(text="\n".join(lines))
        return

    # /oncall rate default <amount>
    if parts[1] == "default":
        if len(parts) < 3:
            respond(
                text="Usage: `/oncall rate default <amount>`  (e.g. `/oncall rate default 50`)"
            )
            return
        try:
            amount = float(parts[2].replace("$", ""))
        except ValueError:
            respond(text=f"Invalid amount: `{parts[2]}`")
            return
        schedule.set_default_shift_rate(amount)
        respond(text=f"Default shift rate set to *${amount:.2f}*.")
        return

    # /oncall rate <extension> <amount>
    if len(parts) < 3:
        respond(
            text="Usage: `/oncall rate <extension> <amount>`  (e.g. `/oncall rate 114 75`)"
        )
        return

    ext = parts[1]
    employee = schedule.get_employee_by_extension(ext)
    if not employee:
        respond(text=f"Extension `{ext}` not found in the roster.")
        return

    try:
        amount = float(parts[2].replace("$", ""))
    except ValueError:
        respond(text=f"Invalid amount: `{parts[2]}`")
        return

    schedule.set_employee_shift_rate(ext, amount)
    respond(
        text=f"Shift rate for *{employee['name']}* (Ext {ext}) set to *${amount:.2f}*."
    )


def _handle_register(respond, schedule, user_id, text):
    parts = text.split()
    if len(parts) < 2:
        respond(
            text="Usage: `/oncall register <extension>`  (e.g. `/oncall register 114`)"
        )
        return

    ext = parts[1].strip()
    try:
        employee = schedule.register_user(user_id, ext)
    except ExtensionAlreadyRegistered:
        respond(
            text=(
                f"Extension {ext} is already registered to another person. "
                "If this is your extension, ask an admin to clear it."
            )
        )
        return
    if not employee:
        respond(
            text=f"Extension {ext} not found in the roster. Check `/oncall roster`."
        )
        return

    respond(text=f"Linked your account to *{employee['name']}* (Ext {ext}).")


def _handle_pick(
    respond, schedule, user_id, text, channel_id, client, schedule_channel
):
    parts = text.split()
    if len(parts) < 2:
        respond(
            text="Usage: `/oncall pick <date> [day|night]`  (e.g. `/oncall pick friday`)"
        )
        return

    employee = schedule.get_employee_by_slack_id(user_id)
    if not employee:
        respond(text="You're not registered. Use `/oncall register <extension>` first.")
        return

    explicit_shift = (
        parts[2] if len(parts) > 2 and parts[2] in ("day", "night") else None
    )
    date_text = parts[1]
    date = parse_date(date_text)
    if not date:
        respond(
            text=f"Couldn't parse date: `{date_text}`. Try: today, tomorrow, friday, 4/5, 7/3/26, jul 3, 2026-04-05"
        )
        return

    date_str = date.strftime("%Y-%m-%d")
    if date_str < datetime.now(EASTERN).strftime("%Y-%m-%d"):
        respond(text="You can't pick up a shift in the past.")
        return

    day_name = date.strftime("%A")
    shift_type = _resolve_pick_shift_type(schedule, date_str, day_name, explicit_shift)
    ctx = schedule.get_shift_context(date_str, day_name, shift_type)

    if ctx["kind"] == "holiday":
        _pick_holiday(
            respond,
            schedule,
            employee,
            date,
            date_str,
            ctx,
            channel_id,
            client,
            schedule_channel,
        )
        return

    _pick_regular(
        respond,
        schedule,
        employee,
        date,
        date_str,
        day_name,
        shift_type,
        ctx,
        channel_id,
        client,
        schedule_channel,
    )


def _resolve_pick_shift_type(schedule, date_str, day_name, explicit_shift) -> str:
    """Pick the shift type a bare ``/oncall pick <date>`` should target.

    Honours an explicit ``day``/``night``. Otherwise a holiday (day-only) wins,
    then any open day/night slot on a weekend; falls back to ``night``.
    """
    if explicit_shift:
        return explicit_shift
    if schedule.get_shift_context(date_str, day_name, "day")["kind"] == "holiday":
        return "day"
    if day_name in WEEKEND_DAYS:
        for st in ("day", "night"):
            _ext, _name, source = schedule.resolve_shift(date_str, day_name, st)
            if source == "available":
                return st
    return "night"


def _pick_regular(
    respond,
    schedule,
    employee,
    date,
    date_str,
    day_name,
    shift_type,
    ctx,
    channel_id,
    client,
    schedule_channel,
):
    """Pick up a regular (non-holiday) shift, gating late pickups on approval."""
    assignees = ctx["assignees"]
    # Already assigned to someone else (not open).
    if assignees and assignees[0]["extension"] != employee["extension"]:
        name = assignees[0]["name"]
        ext = assignees[0]["extension"]
        respond(
            text=f"That shift is already covered by {name} (Ext {ext}). They'd need to drop it first."
        )
        return

    date_label = date.strftime("%A, %b %-d")
    shift_label = _shift_type_label(day_name, shift_type)

    # A pickup after the shift has started (but before it ends) needs an admin to
    # approve it. After the shift ends it's too late to pick up at all.
    if _shift_ended(date_str, shift_type):
        respond(text=f"The *{date_label}*{shift_label} shift has already ended.")
        return
    if _shift_started(date_str, shift_type):
        _request_late_pickup(
            respond,
            schedule,
            employee,
            date_str,
            shift_type,
            is_holiday=False,
            client=client,
        )
        return

    # Already this employee's own shift — nothing to claim, just confirm.
    already_mine = (
        bool(assignees) and assignees[0]["extension"] == employee["extension"]
    )
    if not already_mine:
        # Atomic conditional claim: only one of two concurrent pickers wins, so
        # the loser is told it's taken instead of silently overwriting (TOCTOU).
        claimed = schedule.claim_open_shift(
            date_str, employee["extension"], employee["name"], shift_type
        )
        if not claimed:
            respond(
                text=(
                    f"The *{date_label}*{shift_label} shift was just picked up by "
                    "someone else."
                )
            )
            return

    if is_today(date_str) and _is_active_shift_type(shift_type):
        _update_3cx_routing(employee["extension"])

    respond(text=f"You picked up the shift for *{date_label}*{shift_label}.")

    blocks = build_shift_change_message(
        user_id=employee.get("slack_user_id", ""),
        date_str=date_str,
        action="picked_up",
        ext=employee["extension"],
        name=employee["name"],
        shift_type=shift_type,
    )
    try:
        client.chat_postMessage(
            channel=channel_id,
            blocks=blocks,
            text=f"Shift picked up for {date_str}",
        )
    except Exception:
        logger.exception("Failed to post pickup notification to channel")
    _refresh_schedule_post(schedule, schedule_channel, client)


def _pick_holiday(
    respond,
    schedule,
    employee,
    date,
    date_str,
    ctx,
    channel_id,
    client,
    schedule_channel,
):
    """Pick up a holiday day-shift slot, gating late pickups on approval.

    Holidays support multiple concurrent assignees; the claim is atomic via
    ``claim_holiday_slot`` so concurrent pickers can't oversubscribe the slots.
    """
    ext = employee["extension"]
    date_label = date.strftime("%A, %b %-d")
    label = ctx.get("label") or "Holiday"

    if any(a["extension"] == ext for a in ctx["assignees"]):
        respond(text=f"You're already on the *{date_label}* ({label}) holiday shift.")
        return

    # Late pickup (window open) or too late (window closed).
    if _shift_ended(date_str, "day"):
        respond(text=f"The *{date_label}* ({label}) holiday shift has already ended.")
        return
    if _shift_started(date_str, "day"):
        _request_late_pickup(
            respond,
            schedule,
            employee,
            date_str,
            "day",
            is_holiday=True,
            client=client,
        )
        return

    claimed = schedule.claim_holiday_slot(date_str, ext, employee["name"])
    if not claimed:
        respond(
            text=(
                f"Couldn't claim a slot on the *{date_label}* ({label}) holiday — "
                "it's full or you're already on it."
            )
        )
        return

    respond(text=f"You picked up a slot on the *{date_label}* ({label}) holiday shift.")

    blocks = build_shift_change_message(
        user_id=employee.get("slack_user_id", ""),
        date_str=date_str,
        action="picked_up",
        ext=ext,
        name=employee["name"],
        shift_type="day",
    )
    try:
        client.chat_postMessage(
            channel=channel_id,
            blocks=blocks,
            text=f"Holiday shift picked up for {date_str}",
        )
    except Exception:
        logger.exception("Failed to post holiday pickup notification to channel")
    _refresh_schedule_post(schedule, schedule_channel, client)


def _handle_drop(
    respond, schedule, user_id, text, channel_id, client, schedule_channel
):
    parts = text.split()
    if len(parts) < 2:
        respond(
            text="Usage: `/oncall drop <date> [day|night|holiday]`  (e.g. `/oncall drop friday`)"
        )
        return

    employee = schedule.get_employee_by_slack_id(user_id)
    if not employee:
        respond(text="You're not registered. Use `/oncall register <extension>` first.")
        return

    # An optional trailing qualifier picks the shift to drop; the date itself may
    # be multiple words (e.g. `jul 3`), so strip the qualifier off the end first.
    tokens = parts[1:]
    explicit_shift = None
    if tokens[-1] in ("day", "night", "holiday"):
        explicit_shift = tokens[-1]
        tokens = tokens[:-1]
    if not tokens:
        respond(
            text="Usage: `/oncall drop <date> [day|night|holiday]`  (e.g. `/oncall drop friday`)"
        )
        return

    date_text = " ".join(tokens)
    date = parse_date(date_text)
    if not date:
        respond(
            text=f"Couldn't parse date: `{date_text}`. Try: today, tomorrow, friday, 4/5, 7/3/26, jul 3, 2026-04-05"
        )
        return

    date_str = date.strftime("%Y-%m-%d")
    if date_str < datetime.now(EASTERN).strftime("%Y-%m-%d"):
        respond(text="You can't drop a shift in the past.")
        return

    day_name = date.strftime("%A")
    held = _droppable_shifts(schedule, date_str, day_name, employee["extension"])
    date_label = date.strftime("%A, %b %-d")

    if explicit_shift:
        if explicit_shift not in held:
            if held:
                options = ", ".join(_DROP_SHIFT_LABELS[s] for s in held)
                respond(
                    text=(
                        f"You don't hold the {_DROP_SHIFT_LABELS[explicit_shift]} "
                        f"shift on *{date_label}*. You hold: {options}."
                    )
                )
            else:
                ext, name, _source = schedule.resolve_shift(date_str, day_name)
                respond(
                    text=f"That's not your shift — it belongs to {name} (Ext {ext})."
                )
            return
        target = explicit_shift
    elif not held:
        ext, name, _source = schedule.resolve_shift(date_str, day_name)
        respond(text=f"That's not your shift — it belongs to {name} (Ext {ext}).")
        return
    elif len(held) > 1:
        options = ", ".join(_DROP_SHIFT_LABELS[s] for s in held)
        respond(
            text=(
                f"You hold more than one shift on *{date_label}*: {options}. "
                f"Tell me which to drop: `/oncall drop {date_text} [{'|'.join(held)}]`."
            )
        )
        return
    else:
        target = held[0]

    if target == "holiday":
        _drop_holiday(
            respond,
            schedule,
            employee,
            date,
            date_str,
            channel_id,
            client,
            schedule_channel,
        )
        return

    _drop_regular(
        respond,
        schedule,
        user_id,
        date,
        date_str,
        day_name,
        target,
        channel_id,
        client,
        schedule_channel,
    )


def _drop_regular(
    respond,
    schedule,
    user_id,
    date,
    date_str,
    day_name,
    shift_type,
    channel_id,
    client,
    schedule_channel,
):
    """Release a regular (weekend day or night) shift, honouring the 24h lock."""
    if _within_drop_lock(date_str, shift_type):
        respond(
            text=(
                "This shift starts in under 24 hours — you can't drop it now. "
                "Hand it off with `/oncall swap <date> @person` (they'll need to accept), "
                "or ask an admin to open it."
            )
        )
        return

    ext, name, _source = schedule.resolve_shift(date_str, day_name, shift_type)

    # No 3CX repoint here: a same-day shift is always inside the 24h lock above,
    # so a drop that reaches this point is never today's active shift.
    schedule.mark_open(date_str, shift_type)

    date_label = date.strftime("%A, %b %-d")
    shift_label = _shift_type_label(day_name, shift_type)
    respond(
        text=(
            f"You dropped the shift for *{date_label}*{shift_label}. "
            "It's now open for pickup."
        )
    )

    blocks = build_shift_change_message(
        user_id,
        date_str,
        "dropped",
        ext,
        name,
        shift_type=shift_type,
    )
    try:
        client.chat_postMessage(
            channel=channel_id, blocks=blocks, text=f"Shift dropped for {date_str}"
        )
    except Exception:
        logger.exception("Failed to post drop notification to channel")
    _refresh_schedule_post(schedule, schedule_channel, client)


def _drop_holiday(
    respond, schedule, employee, date, date_str, channel_id, client, schedule_channel
):
    """Release the employee's holiday slot (subject to the 24h drop lock).

    The slot becomes open for someone else to pick up. The released slot does
    not repoint 3CX here: a same-day drop is always inside the 24h lock (the
    holiday starts 08:00 ET), so a drop that reaches the release is never the
    live window.
    """
    if _within_drop_lock(date_str, "day"):
        respond(
            text=(
                "This holiday shift starts in under 24 hours — you can't drop it now. "
                "Hand it off with `/oncall swap <date> @person` (they'll need to accept), "
                "or ask an admin to open it."
            )
        )
        return

    released = schedule.release_holiday_slot(date_str, employee["extension"])
    if not released:
        respond(text="You're not on that holiday shift.")
        return

    date_label = date.strftime("%A, %b %-d")
    respond(
        text=(
            f"You dropped your slot on the *{date_label}* holiday shift. "
            "It's now open for pickup."
        )
    )

    blocks = build_shift_change_message(
        user_id=employee.get("slack_user_id", ""),
        date_str=date_str,
        action="dropped",
        ext=employee["extension"],
        name=employee["name"],
        shift_type="day",
    )
    try:
        client.chat_postMessage(
            channel=channel_id,
            blocks=blocks,
            text=f"Holiday shift dropped for {date_str}",
        )
    except Exception:
        logger.exception("Failed to post holiday drop notification to channel")
    _refresh_schedule_post(schedule, schedule_channel, client)


def _handle_swap(
    respond, schedule, user_id, text, channel_id, client, schedule_channel
):
    # Expected format: swap <date> @user  OR  swap <date> <extension>
    parts = text.split(maxsplit=2)
    if len(parts) < 3:
        respond(
            text="Usage: `/oncall swap <date> @person`  (e.g. `/oncall swap friday @sarah`)"
        )
        return

    employee = schedule.get_employee_by_slack_id(user_id)
    if not employee:
        respond(text="You're not registered. Use `/oncall register <extension>` first.")
        return

    date = parse_date(parts[1])
    if not date:
        respond(text=f"Couldn't parse date: `{parts[1]}`.")
        return

    date_str = date.strftime("%Y-%m-%d")
    if date_str < datetime.now(EASTERN).strftime("%Y-%m-%d"):
        respond(text="You can't swap a shift in the past.")
        return

    day_name = date.strftime("%A")

    # A holiday slot the employee holds is swappable as a day shift; holidays
    # take priority on their date.
    holiday_ctx = schedule.get_shift_context(date_str, day_name, "day")
    holiday_swap = holiday_ctx["kind"] == "holiday" and any(
        a["extension"] == employee["extension"] for a in holiday_ctx["assignees"]
    )

    if holiday_swap:
        shift_type = "day"
    else:
        found = _find_employee_shift(
            schedule, date_str, day_name, employee["extension"]
        )
        if not found:
            ext, name, _source = schedule.resolve_shift(date_str, day_name)
            respond(
                text=f"That's not your shift — it belongs to {name} (Ext {ext}). You can only swap your own shifts."
            )
            return
        _ext, _name, _source, shift_type = found

    # Resolve target user — could be <@U12345> or an extension number
    target_text = parts[2].strip()
    slack_id_match = re.match(r"<@(\w+)(?:\|[^>]*)?>", target_text)
    if slack_id_match:
        target_slack_id = slack_id_match.group(1)
        target = schedule.get_employee_by_slack_id(target_slack_id)
        if not target:
            respond(
                text=f"<@{target_slack_id}> isn't registered. They need to run `/oncall register <extension>`."
            )
            return
    else:
        target = schedule.get_employee_by_extension(target_text)
        if not target:
            respond(text=f"Extension `{target_text}` not found in the roster.")
            return

    if target["extension"] == employee["extension"]:
        respond(text="That shift is already yours — nothing to swap.")
        return

    # The target must be linked to Slack so we can DM them the request.
    if not target.get("slack_user_id"):
        respond(
            text=f"*{target['name']}* (Ext {target['extension']}) isn't linked to Slack yet — "
            "they need to run `/oncall register <extension>` before they can be swapped a shift."
        )
        return

    # Create a pending swap and DM the target Accept/Decline. The shift does NOT
    # move until they accept — the original owner stays responsible until then.
    expires_at = int(_shift_start(date_str, shift_type).timestamp())
    schedule.create_pending_swap(date_str, shift_type, employee, target, expires_at)

    date_label = date.strftime("%A, %b %-d")
    shift_label = _shift_type_label(day_name, shift_type)
    try:
        client.chat_postMessage(
            channel=target["slack_user_id"],
            blocks=build_swap_request_blocks(user_id, date_str, shift_type),
            text=f"{employee['name']} wants to swap you the {date_str} shift",
        )
    except Exception:
        logger.exception("Failed to DM swap request to target")
        respond(
            text=f"Couldn't reach *{target['name']}* on Slack to send the request. Try again later."
        )
        return

    respond(
        text=(
            f"Swap request sent to <@{target['slack_user_id']}> for "
            f"*{date_label}*{shift_label}. The shift moves to them once they accept."
        )
    )


def _parse_swap_action(action_id: str, prefix: str) -> tuple[str, str]:
    """Split a swap action_id into (date_str, shift_type)."""
    remainder = action_id[len(prefix) :]
    if remainder.endswith("_day"):
        return remainder[:-4], "day"
    return remainder, "night"


def handle_swap_accept(body, respond, client, schedule, schedule_channel):
    """Target accepted a swap — apply the override and mark it verified."""
    date_str, shift_type = _parse_swap_action(
        body["actions"][0]["action_id"], "swap_accept_"
    )
    user_id = body["user"]["id"]

    swap = schedule.get_swap(date_str, shift_type)
    if (
        not swap
        or swap.get("status") != "pending"
        or swap.get("target_slack") != user_id
    ):
        respond(
            replace_original=True,
            blocks=build_swap_resolved_blocks("This swap request is no longer valid."),
        )
        return

    if _shift_started(date_str, shift_type):
        schedule.clear_swap(date_str, shift_type)
        respond(
            replace_original=True,
            blocks=build_swap_resolved_blocks(
                "This swap request has expired — the shift has already started."
            ),
        )
        return

    # A day-shift swap on a holiday date hands off a holiday slot rather than a
    # plain override: atomically move the slot from requester to target.
    is_holiday = shift_type == "day" and schedule.get_holiday(date_str) is not None
    if is_holiday:
        moved = schedule.swap_holiday_assignee(
            date_str,
            swap["requester_ext"],
            swap["target_ext"],
            swap["target_name"],
        )
        if not moved:
            schedule.clear_swap(date_str, shift_type)
            respond(
                replace_original=True,
                blocks=build_swap_resolved_blocks(
                    "This swap request is no longer valid."
                ),
            )
            return
        if _holiday_window_active(date_str):
            _set_holiday_queue_agents(schedule, date_str)
    else:
        # Re-resolve the current holder at accept time. reassign_if_held_by
        # below trusts "no override row" as "still the requester's", but a
        # weekly-held shift also has no override row — so an admin clear or a
        # weekly-schedule edit between swap-init and accept could move the shift
        # to a third party without ever creating an override, and the bare
        # conditional write would not catch it. Confirm the requester is still
        # the resolved holder before reassigning.
        accept_day_name = datetime.strptime(date_str, "%Y-%m-%d").strftime("%A")
        current_ext, _, _ = schedule.resolve_shift(
            date_str, accept_day_name, shift_type
        )
        if current_ext != swap["requester_ext"]:
            schedule.clear_swap(date_str, shift_type)
            respond(
                replace_original=True,
                blocks=build_swap_resolved_blocks(
                    "This shift is no longer assigned to the person who "
                    "requested the swap, so it couldn't be applied."
                ),
            )
            return
        # Only move the shift if it's still the requester's (or still on the
        # weekly fallback). If it was independently claimed after the swap was
        # initiated, abort instead of silently overwriting the new holder.
        moved = schedule.reassign_if_held_by(
            date_str,
            swap["requester_ext"],
            swap["target_ext"],
            swap["target_name"],
            shift_type,
        )
        if not moved:
            schedule.clear_swap(date_str, shift_type)
            respond(
                replace_original=True,
                blocks=build_swap_resolved_blocks(
                    "This shift was already picked up by someone else, so the "
                    "swap couldn't be applied."
                ),
            )
            return
        if is_today(date_str) and _is_active_shift_type(shift_type):
            _update_3cx_routing(swap["target_ext"])
    schedule.mark_swap_verified(date_str, shift_type)

    day_name = datetime.strptime(date_str, "%Y-%m-%d").strftime("%A")
    date_label = datetime.strptime(date_str, "%Y-%m-%d").strftime("%A, %b %-d")
    shift_label = _shift_type_label(day_name, shift_type)

    respond(
        replace_original=True,
        blocks=build_swap_resolved_blocks(
            f"You're now covering the *{date_label}*{shift_label} shift. Thanks!"
        ),
    )

    if swap.get("requester_slack"):
        try:
            client.chat_postMessage(
                channel=swap["requester_slack"],
                text=f"<@{user_id}> accepted your swap — they're now on the *{date_label}*{shift_label} shift.",
            )
        except Exception:
            logger.exception("Failed to DM swap requester on accept")

    if schedule_channel:
        blocks = build_shift_change_message(
            user_id,
            date_str,
            "swapped",
            swap["target_ext"],
            swap["target_name"],
            shift_type=shift_type,
        )
        try:
            client.chat_postMessage(
                channel=schedule_channel,
                blocks=blocks,
                text=f"Shift swapped for {date_str}",
            )
        except Exception:
            logger.exception("Failed to post swap notification to channel")
    _refresh_schedule_post(schedule, schedule_channel, client)


def handle_swap_decline(body, respond, client, schedule, schedule_channel):
    """Target declined a swap — clear it and notify the requester."""
    date_str, shift_type = _parse_swap_action(
        body["actions"][0]["action_id"], "swap_decline_"
    )
    user_id = body["user"]["id"]

    swap = schedule.get_swap(date_str, shift_type)
    if (
        not swap
        or swap.get("status") != "pending"
        or swap.get("target_slack") != user_id
    ):
        respond(
            replace_original=True,
            blocks=build_swap_resolved_blocks("This swap request is no longer valid."),
        )
        return

    schedule.clear_swap(date_str, shift_type)

    day_name = datetime.strptime(date_str, "%Y-%m-%d").strftime("%A")
    date_label = datetime.strptime(date_str, "%Y-%m-%d").strftime("%A, %b %-d")
    shift_label = _shift_type_label(day_name, shift_type)

    respond(
        replace_original=True,
        blocks=build_swap_resolved_blocks(
            f"You declined the *{date_label}*{shift_label} swap. No change."
        ),
    )

    if swap.get("requester_slack"):
        try:
            client.chat_postMessage(
                channel=swap["requester_slack"],
                text=f"<@{user_id}> declined your swap for *{date_label}*{shift_label} — it's still your shift.",
            )
        except Exception:
            logger.exception("Failed to DM swap requester on decline")


# ── Late-pickup approval ─────────────────────────────────────────────────


def _request_late_pickup(
    respond, schedule, employee, date_str, shift_type, is_holiday, client
):
    """Create a PICKUP_REQUEST and DM every admin an Approve/Deny prompt.

    Used when someone tries to pick up a shift that has already *started* (but
    not ended). The shift is NOT claimed yet — the first admin to approve wins
    (a conditional claim). Mirrors the verified-swap DM pattern, but the
    approvers are the admins rather than the swap target.
    """
    schedule.create_pickup_request(
        date_str,
        shift_type,
        employee,
        expires_at=int(_shift_end(date_str, shift_type).timestamp()),
        is_holiday=is_holiday,
    )

    blocks = build_pickup_request_blocks(
        requester_slack=employee.get("slack_user_id", ""),
        requester_name=employee["name"],
        date_str=date_str,
        shift_type=shift_type,
        requester_ext=employee["extension"],
        is_holiday=is_holiday,
    )
    text = f"{employee['name']} wants to pick up the already-started {date_str} shift"
    admins = schedule.get_admin_users()
    delivered = 0
    for admin_id in admins:
        try:
            client.chat_postMessage(channel=admin_id, blocks=blocks, text=text)
            delivered += 1
        except Exception:
            logger.exception("Failed to DM late-pickup request to admin %s", admin_id)

    if delivered == 0:
        # Nobody to approve it — roll the request back so it doesn't dangle.
        schedule.clear_pickup_request(date_str, shift_type, employee["extension"])
        respond(
            text=(
                "That shift has already started and needs an admin to approve a "
                "pickup, but I couldn't reach any admin. Please contact one directly."
            )
        )
        return

    respond(
        text=(
            "That shift has already started, so a pickup needs admin approval. "
            "I've sent your request to the admins — you'll be notified once it's decided."
        )
    )


def _parse_pickup_action(action_id: str, prefix: str) -> tuple[str, str, str]:
    """Split a pickup-approval action_id into (date_str, shift_type, ext).

    The tail is ``<date>[_day]_<ext>`` (see ``build_pickup_request_blocks``).
    """
    remainder = action_id[len(prefix) :]
    ext_start = remainder.rfind("_")
    ext = remainder[ext_start + 1 :]
    rest = remainder[:ext_start]
    if rest.endswith("_day"):
        return rest[:-4], "day", ext
    return rest, "night", ext


def handle_pickup_approve(body, respond, client, schedule, schedule_channel):
    """An admin approved a late pickup — claim the shift for the requester.

    First-approve-wins: the claim is conditional (``set_override`` after a guard,
    or atomic ``claim_holiday_slot``), so a second admin approving a
    just-resolved request gets a "no longer pending" response. On success the
    requester is DM'd and 3CX is repointed if the shift window is currently live.
    """
    date_str, shift_type, ext = _parse_pickup_action(
        body["actions"][0]["action_id"], "pickup_approve_"
    )
    admin_id = body["user"]["id"]
    # The Approve button is DM'd only to admins, but action handlers receive
    # whoever clicks — re-verify the actor is an admin (mirrors swap handlers).
    if admin_id not in schedule.get_admin_users():
        respond(
            replace_original=False,
            text="Only an admin can approve a pickup request.",
        )
        return

    req = schedule.get_pickup_request(date_str, shift_type, ext)
    if not req or req.get("status") != "pending":
        respond(
            replace_original=True,
            blocks=build_pickup_resolved_blocks(
                "This pickup request is no longer pending."
            ),
        )
        return

    if _shift_ended(date_str, shift_type):
        schedule.clear_pickup_request(date_str, shift_type, ext)
        respond(
            replace_original=True,
            blocks=build_pickup_resolved_blocks(
                "This pickup request has expired — the shift has already ended."
            ),
        )
        return

    # Atomic first-approve-wins: only the admin who flips the request
    # pending->approved proceeds; a second concurrent approver is turned away.
    if not schedule.approve_pickup_request(date_str, shift_type, ext):
        respond(
            replace_original=True,
            blocks=build_pickup_resolved_blocks(
                "This pickup request is no longer pending."
            ),
        )
        return

    requester_name = req.get("requester_name", ext)
    requester_slack = req.get("requester_slack", "")
    is_holiday = bool(req.get("is_holiday"))

    day_name = datetime.strptime(date_str, "%Y-%m-%d").strftime("%A")
    date_label = datetime.strptime(date_str, "%Y-%m-%d").strftime("%A, %b %-d")
    shift_label = _shift_type_label(day_name, shift_type)

    if is_holiday:
        claimed = schedule.claim_holiday_slot(date_str, ext, requester_name)
        if not claimed:
            schedule.clear_pickup_request(date_str, shift_type, ext)
            respond(
                replace_original=True,
                blocks=build_pickup_resolved_blocks(
                    "Couldn't assign the holiday slot — it's full or already taken."
                ),
            )
            return
        if _holiday_window_active(date_str):
            _set_holiday_queue_agents(schedule, date_str)
    else:
        # Conditional claim so two different requesters' approvals for the SAME
        # shift can't silently clobber each other — the per-request approve gate
        # only serializes a single request's SK, not the OVERRIDE row.
        claimed = schedule.claim_open_shift(date_str, ext, requester_name, shift_type)
        if not claimed:
            schedule.clear_pickup_request(date_str, shift_type, ext)
            respond(
                replace_original=True,
                blocks=build_pickup_resolved_blocks(
                    "Couldn't assign the shift — it's already covered."
                ),
            )
            return
        if is_today(date_str) and _is_active_shift_type(shift_type):
            # Best-effort: a 3CX failure must not strand the request as approved.
            try:
                _update_3cx_routing(ext)
            except Exception:
                logger.exception("3CX repoint failed after approving late pickup")

    schedule.clear_pickup_request(date_str, shift_type, ext)

    respond(
        replace_original=True,
        blocks=build_pickup_resolved_blocks(
            f"Approved — {requester_name} (Ext {ext}) is now on the "
            f"*{date_label}*{shift_label} shift."
        ),
    )

    if requester_slack:
        try:
            client.chat_postMessage(
                channel=requester_slack,
                text=(
                    f"<@{admin_id}> approved your pickup — you're now on the "
                    f"*{date_label}*{shift_label} shift."
                ),
            )
        except Exception:
            logger.exception("Failed to DM late-pickup requester on approve")

    if schedule_channel:
        try:
            client.chat_postMessage(
                channel=schedule_channel,
                blocks=build_shift_change_message(
                    user_id=requester_slack,
                    date_str=date_str,
                    action="picked_up",
                    ext=ext,
                    name=requester_name,
                    shift_type=shift_type,
                ),
                text=f"Shift picked up for {date_str}",
            )
        except Exception:
            logger.exception("Failed to post late-pickup notification to channel")
    _refresh_schedule_post(schedule, schedule_channel, client)


def handle_pickup_deny(body, respond, client, schedule, schedule_channel):
    """An admin denied a late pickup — clear the request and notify the requester."""
    date_str, shift_type, ext = _parse_pickup_action(
        body["actions"][0]["action_id"], "pickup_deny_"
    )
    admin_id = body["user"]["id"]
    if admin_id not in schedule.get_admin_users():
        respond(
            replace_original=False,
            text="Only an admin can deny a pickup request.",
        )
        return

    req = schedule.get_pickup_request(date_str, shift_type, ext)
    if not req or req.get("status") != "pending":
        respond(
            replace_original=True,
            blocks=build_pickup_resolved_blocks(
                "This pickup request is no longer pending."
            ),
        )
        return

    schedule.clear_pickup_request(date_str, shift_type, ext)

    day_name = datetime.strptime(date_str, "%Y-%m-%d").strftime("%A")
    date_label = datetime.strptime(date_str, "%Y-%m-%d").strftime("%A, %b %-d")
    shift_label = _shift_type_label(day_name, shift_type)

    respond(
        replace_original=True,
        blocks=build_pickup_resolved_blocks(
            f"Denied — the *{date_label}*{shift_label} pickup was not approved."
        ),
    )

    requester_slack = req.get("requester_slack", "")
    if requester_slack:
        try:
            client.chat_postMessage(
                channel=requester_slack,
                text=(
                    f"<@{admin_id}> denied your pickup for the "
                    f"*{date_label}*{shift_label} shift."
                ),
            )
        except Exception:
            logger.exception("Failed to DM late-pickup requester on deny")


# ── Shared admin validation + side-effect helpers ───────────────────────
#
# These are the single source of the admin mutation rules and side effects, so
# the text subcommands and the Block Kit modal ``view_submission`` handlers can't
# drift. Validators return an error string (or ``None``); apply helpers run the
# ``Schedule`` mutation plus the same 3CX / schedule-post side effects.


def _resolve_roster_employee(schedule, extension):
    """Return (employee, error). ``error`` is set when the extension is unknown."""
    employee = schedule.get_employee_by_extension(extension)
    if not employee:
        return None, f"Extension `{extension}` not found in the roster."
    return employee, None


def _validate_holiday_date(date_str: str) -> str | None:
    """Error message if a holiday date is in the past, else None."""
    if date_str < datetime.now(EASTERN).strftime("%Y-%m-%d"):
        return "You can't schedule a holiday in the past."
    return None


def _parse_slots(raw):
    """Parse a holiday slot count. Returns (slots, error)."""
    try:
        slots = int(raw)
    except (ValueError, TypeError):
        return None, f"Slots must be a whole number, got `{raw}`."
    if slots < 1:
        return None, "Slots must be at least 1."
    return slots, None


def _parse_multiplier(raw):
    """Parse an optional pay multiplier (`2`, `2.0`, or `x2`). Returns (value, error)."""
    if raw is None:
        return None, None
    token = raw.strip()
    if not token:
        return None, None
    match = re.fullmatch(r"x?([0-9]+(?:\.[0-9]+)?)", token, re.IGNORECASE)
    if not match:
        return None, f"Multiplier must be a number like `2` or `x1.5`, got `{raw}`."
    return match.group(1), None


def _validate_label(label: str) -> str | None:
    if not label.strip():
        return "A holiday label is required."
    return None


def _apply_override(schedule, date_str, employee, shift_type, client, schedule_channel):
    """Set the override + same-day 3CX repoint + post refresh.

    Returns (message, repointed_3cx). Shared by the text command and the modal.
    """
    schedule.set_override(date_str, employee["extension"], employee["name"], shift_type)
    repointed = False
    if is_today(date_str) and _is_active_shift_type(shift_type):
        _update_3cx_routing(employee["extension"])
        repointed = True
    _refresh_schedule_post(schedule, schedule_channel, client)
    date = datetime.strptime(date_str, "%Y-%m-%d")
    label = "Day" if shift_type == "day" else "Night"
    message = (
        f"Override set: *{date.strftime('%A, %b %-d')}* ({label}) → "
        f"{employee['name']} (Ext {employee['extension']})"
    )
    return message, repointed


def _apply_holiday_add(
    schedule, date_str, slots, multiplier, label, client, schedule_channel
):
    """Create the holiday record + one-off schedules + side effects.

    Returns (created, message). Shared by the text command and the modal.
    """
    schedule_names = _create_holiday_schedules(date_str)
    created = schedule.create_holiday(
        date_str,
        slots=slots,
        label=label,
        created_by=schedule_channel or "",
        multiplier=multiplier,
        schedule_names=schedule_names,
    )
    date = datetime.strptime(date_str, "%Y-%m-%d")
    if not created:
        # Roll back the schedules we just made for a date that already has one.
        _delete_holiday_schedules(schedule_names)
        return False, (
            f"A holiday already exists on *{date.strftime('%A, %b %-d')}*. "
            "Remove it first to recreate."
        )

    # If the window is already open (admin added it mid-day), the 08:00 schedule
    # has passed, so repoint the call flow now via the holiday router.
    if _holiday_window_active(date_str):
        _activate_holiday_inline(schedule, date_str)

    holiday = schedule.get_holiday(date_str)
    multiplier_value = holiday["multiplier"] if holiday else (multiplier or "1.5")

    if schedule_channel:
        try:
            client.chat_postMessage(
                channel=schedule_channel,
                blocks=build_holiday_added_blocks(
                    date_str, label, slots, multiplier_value
                ),
                text=f"Holiday added: {label} on {date_str}",
            )
        except Exception:
            logger.exception("Failed to post holiday-added notification to channel")
    _refresh_schedule_post(schedule, schedule_channel, client)
    return True, (
        f"Scheduled *{label}* holiday on *{date.strftime('%A, %b %-d')}* — "
        f"{slots} slot{'s' if slots != 1 else ''} at {float(multiplier_value):g}x pay."
    )


def _handle_admin(respond, schedule, user_id, text, is_admin, client, schedule_channel):
    if not is_admin:
        respond(text="Admin commands are restricted. Contact an administrator.")
        return

    parts = text.split()
    if len(parts) < 2:
        respond(
            text=(
                "*Admin Commands:*\n"
                "`admin override <date> <ext> [day|night]` — Assign shift\n"
                "`admin open <date> [day|night]` — Mark open\n"
                "`admin clear <date> [day|night]` — Remove override\n"
                "`admin roster add <ext> <name>` — Add employee\n"
                "`admin roster remove <ext>` — Remove employee\n"
                "`admin roster rename <ext> <name>` — Rename\n"
                "`admin holiday add <date> <slots> [x<mult>] <label>` — Schedule holiday\n"
                "`admin holiday remove <date>` — Remove holiday\n"
                "`admin holiday list` — List upcoming holidays"
            )
        )
        return

    subcmd = parts[1]

    if subcmd == "override":
        if len(parts) < 4:
            respond(
                text="Usage: `/oncall admin override <date> <extension> [day|night]`"
            )
            return
        date = parse_date(parts[2])
        if not date:
            respond(text=f"Couldn't parse date: `{parts[2]}`")
            return
        ext = parts[3]
        shift_type = (
            parts[4] if len(parts) > 4 and parts[4] in ("day", "night") else "night"
        )
        employee, error = _resolve_roster_employee(schedule, ext)
        if error:
            respond(text=error)
            return
        date_str = date.strftime("%Y-%m-%d")
        message, _repointed = _apply_override(
            schedule, date_str, employee, shift_type, client, schedule_channel
        )
        respond(text=message)

    elif subcmd == "open":
        if len(parts) < 3:
            respond(text="Usage: `/oncall admin open <date> [day|night]`")
            return
        date = parse_date(parts[2])
        if not date:
            respond(text=f"Couldn't parse date: `{parts[2]}`")
            return
        shift_type = (
            parts[3] if len(parts) > 3 and parts[3] in ("day", "night") else "night"
        )
        date_str = date.strftime("%Y-%m-%d")
        schedule.mark_open(date_str, shift_type)
        if is_today(date_str) and _is_active_shift_type(shift_type):
            _update_3cx_routing(FALLBACK_EXTENSION)
        label = "Day" if shift_type == "day" else "Night"
        respond(text=f"*{date.strftime('%A, %b %-d')}* ({label}) marked as open.")
        _refresh_schedule_post(schedule, schedule_channel, client)

    elif subcmd == "clear":
        if len(parts) < 3:
            respond(text="Usage: `/oncall admin clear <date> [day|night]`")
            return
        date = parse_date(parts[2])
        if not date:
            respond(text=f"Couldn't parse date: `{parts[2]}`")
            return
        shift_type = (
            parts[3] if len(parts) > 3 and parts[3] in ("day", "night") else "night"
        )
        date_str = date.strftime("%Y-%m-%d")
        schedule.remove_override(date_str, shift_type)
        if is_today(date_str) and _is_active_shift_type(shift_type):
            day_name = date.strftime("%A")
            ext, _name, _source = schedule.resolve_shift(date_str, day_name, shift_type)
            _update_3cx_routing(ext)
        label = "Day" if shift_type == "day" else "Night"
        respond(
            text=f"Override cleared for *{date.strftime('%A, %b %-d')}* ({label}) — reverted to weekly schedule."
        )
        _refresh_schedule_post(schedule, schedule_channel, client)

    elif subcmd == "roster":
        if len(parts) < 3:
            respond(text="Usage: `admin roster add|remove|rename <ext> [name]`")
            return
        roster_cmd = parts[2]

        if roster_cmd == "add":
            if len(parts) < 5:
                respond(text="Usage: `/oncall admin roster add <ext> <name>`")
                return
            ext = parts[3]
            name = " ".join(parts[4:])
            added = schedule.add_roster_entry(ext, name)
            if not added:
                respond(
                    text=f"Extension `{ext}` already exists. Use `roster rename` to change the name."
                )
                return
            respond(text=f"Added *{name}* (Ext {ext}) to the roster.")

        elif roster_cmd == "remove":
            if len(parts) < 4:
                respond(text="Usage: `/oncall admin roster remove <ext>`")
                return
            ext = parts[3]
            employee = schedule.get_employee_by_extension(ext)
            if not employee:
                respond(text=f"Extension `{ext}` not found in the roster.")
                return
            schedule.remove_roster_entry(ext)
            respond(
                text=f"Removed *{employee.get('name', ext)}* (Ext {ext}) from the roster."
            )

        elif roster_cmd == "rename":
            if len(parts) < 5:
                respond(text="Usage: `/oncall admin roster rename <ext> <name>`")
                return
            ext = parts[3]
            employee = schedule.get_employee_by_extension(ext)
            if not employee:
                respond(text=f"Extension `{ext}` not found in the roster.")
                return
            new_name = " ".join(parts[4:])
            schedule.rename_roster_entry(ext, new_name)
            respond(
                text=f"Renamed Ext {ext}: {employee.get('name', '?')} → *{new_name}*"
            )

        else:
            respond(text="Unknown roster command. Use `add`, `remove`, or `rename`.")

    elif subcmd == "holiday":
        _handle_admin_holiday(respond, schedule, parts, client, schedule_channel)

    else:
        respond(text=f"Unknown admin command: `{subcmd}`. Try `/oncall help`.")


# ── Admin: holidays ──────────────────────────────────────────────────────


def _handle_admin_holiday(respond, schedule, parts, client, schedule_channel):
    """Dispatch ``admin holiday add|remove|list``.

    ``parts`` is the whitespace-split command, where ``parts[1] == "holiday"``.
    """
    if len(parts) < 3:
        respond(text="Usage: `admin holiday add|remove|list ...`")
        return
    holiday_cmd = parts[2]

    if holiday_cmd == "add":
        _admin_holiday_add(respond, schedule, parts, client, schedule_channel)
    elif holiday_cmd == "remove":
        _admin_holiday_remove(respond, schedule, parts, client, schedule_channel)
    elif holiday_cmd == "list":
        _admin_holiday_list(respond, schedule)
    else:
        respond(text="Unknown holiday command. Use `add`, `remove`, or `list`.")


def _admin_holiday_add(respond, schedule, parts, client, schedule_channel):
    """`admin holiday add <date> <slots> [x<mult>] <label>`.

    Creates the HOLIDAY record, provisions the 08:00 activate / 17:00 deactivate
    one-off schedules (storing their names on the record), inline-activates when
    the window is already open, and announces the holiday in the channel.
    """
    if len(parts) < 6:
        respond(
            text=(
                "Usage: `/oncall admin holiday add <date> <slots> [x<mult>] <label>`\n"
                "e.g. `/oncall admin holiday add 2026-07-04 2 x2 Independence Day`"
            )
        )
        return

    date = parse_date(parts[3])
    if not date:
        respond(text=f"Couldn't parse date: `{parts[3]}`")
        return
    date_str = date.strftime("%Y-%m-%d")
    date_error = _validate_holiday_date(date_str)
    if date_error:
        respond(text=date_error)
        return

    slots, slots_error = _parse_slots(parts[4])
    if slots_error:
        respond(text=slots_error)
        return

    # Optional ``x<mult>`` token before the label.
    rest = parts[5:]
    multiplier = None
    if rest and re.fullmatch(r"x[0-9]+(\.[0-9]+)?", rest[0], re.IGNORECASE):
        multiplier = rest[0][1:]
        rest = rest[1:]
    label = " ".join(rest).strip()
    label_error = _validate_label(label)
    if label_error:
        respond(text=label_error)
        return

    _created, message = _apply_holiday_add(
        schedule, date_str, slots, multiplier, label, client, schedule_channel
    )
    respond(text=message)


def _admin_holiday_remove(respond, schedule, parts, client, schedule_channel):
    """`admin holiday remove <date>` — delete the record + outstanding schedules."""
    if len(parts) < 4:
        respond(text="Usage: `/oncall admin holiday remove <date>`")
        return
    date = parse_date(parts[3])
    if not date:
        respond(text=f"Couldn't parse date: `{parts[3]}`")
        return
    date_str = date.strftime("%Y-%m-%d")

    holiday = schedule.get_holiday(date_str)
    if not holiday:
        respond(text=f"No holiday scheduled on *{date.strftime('%A, %b %-d')}*.")
        return

    _delete_holiday_schedules(holiday.get("schedule_names", []))
    schedule.remove_holiday(date_str)
    respond(
        text=(
            f"Removed the *{holiday.get('label', 'holiday')}* holiday on "
            f"*{date.strftime('%A, %b %-d')}*."
        )
    )
    _refresh_schedule_post(schedule, schedule_channel, client)


def _admin_holiday_list(respond, schedule):
    """`admin holiday list` — show today-and-future scheduled holidays."""
    today_str = datetime.now(EASTERN).strftime("%Y-%m-%d")
    items = schedule.list_holidays(today_str)
    if not items:
        respond(text="No upcoming holidays scheduled.")
        return

    lines = ["*Upcoming Holidays*\n"]
    for item in items:
        date_str = item["SK"]
        dt = datetime.strptime(date_str, "%Y-%m-%d")
        slots = int(item.get("slots", 0))
        filled = len(item.get("assignees", {}) or {})
        mult = f"{float(item.get('multiplier', 1.5)):g}x"
        lines.append(
            f"• {dt.strftime('%a %b %-d')} — _{item.get('label', 'Holiday')}_ "
            f"({filled}/{slots} filled, {mult})"
        )
    respond(text="\n".join(lines))


# ── App Home ────────────────────────────────────────────────────────────


@functools.lru_cache(maxsize=1)
def _changelog_text() -> str:
    """Read the CHANGELOG shipped next to this module.

    Lazy (never at import) and tolerant of a missing file, so the App Home tab
    degrades to "no What's New section" rather than erroring. Lambda zips and
    the Fargate image both keep ``CHANGELOG.md`` beside ``app.py``.
    ``LAMBDA_TASK_ROOT`` remains a fallback for the zip layout.
    """
    tried = []
    for path in _changelog_paths():
        tried.append(path)
        try:
            with open(path, encoding="utf-8") as fh:
                return fh.read()
        except OSError:
            continue
    logger.warning("CHANGELOG.md not found at %s — App Home omits What's New", tried)
    return ""


def _changelog_paths() -> list[str]:
    paths = [os.path.join(os.path.dirname(os.path.abspath(__file__)), "CHANGELOG.md")]
    task_root = os.environ.get("LAMBDA_TASK_ROOT", "").strip()
    if task_root:
        paths.append(os.path.join(task_root, "CHANGELOG.md"))
    return paths


_HOME_OVERVIEW_DAYS = 60


def publish_home(client, user_id: str, changelog_text: str, schedule=None) -> None:
    """Render and publish the App Home view for ``user_id``.

    When ``schedule`` is provided and the viewer is an admin (looked up via
    ``get_admin_users()`` — never trust the surface), the admin section with the
    override/holiday modal buttons and an upcoming overview is included.
    """
    entry = latest_entry(changelog_text)
    is_admin = False
    upcoming_overrides = None
    upcoming_holidays = None
    if schedule is not None and user_id in schedule.get_admin_users():
        is_admin = True
        today = datetime.now(EASTERN)
        start = today.strftime("%Y-%m-%d")
        end = (today + timedelta(days=_HOME_OVERVIEW_DAYS)).strftime("%Y-%m-%d")
        upcoming_overrides = schedule.list_overrides(start, end)
        upcoming_holidays = schedule.list_holidays(start)
    view = build_home_view(
        version=entry.version if entry else None,
        notes=entry.body if entry else "",
        date_label=entry.date_label if entry else "",
        is_admin=is_admin,
        upcoming_overrides=upcoming_overrides,
        upcoming_holidays=upcoming_holidays,
    )
    client.views_publish(user_id=user_id, view=view)


# ── Admin modals (views.open + view_submission) ─────────────────────────


def _notify_admin(client, user_id: str, text: str) -> None:
    """DM the admin a confirmation of a modal-driven change (best-effort)."""
    try:
        client.chat_postMessage(channel=user_id, text=text)
    except Exception:
        logger.exception("Failed to send admin modal confirmation DM")


def open_override_modal(body, client, schedule) -> None:
    """Open the override modal — admins only (re-check, never trust the surface)."""
    user_id = body["user"]["id"]
    if user_id not in schedule.get_admin_users():
        return
    client.views_open(
        trigger_id=body["trigger_id"],
        view=build_override_modal(schedule.get_roster()),
    )


def open_holiday_add_modal(body, client, schedule) -> None:
    """Open the holiday-add modal — admins only (re-check, never trust surface)."""
    user_id = body["user"]["id"]
    if user_id not in schedule.get_admin_users():
        return
    client.views_open(trigger_id=body["trigger_id"], view=build_holiday_add_modal())


def handle_override_submission(ack, body, view, client, schedule, schedule_channel):
    """``view_submission`` for the override modal."""
    user_id = body["user"]["id"]
    if user_id not in schedule.get_admin_users():
        ack(response_action="errors", errors={"date": "Admin access required."})
        return

    values = view["state"]["values"]
    date_str = values["date"]["date"]["selected_date"]
    ext = values["extension"]["extension"]["selected_option"]["value"]
    shift_type = values["shift_type"]["shift_type"]["selected_option"]["value"]

    employee, error = _resolve_roster_employee(schedule, ext)
    if error:
        ack(response_action="errors", errors={"extension": error})
        return

    ack()
    message, repointed = _apply_override(
        schedule, date_str, employee, shift_type, client, schedule_channel
    )
    if repointed:
        message += (
            "\n:telephone_receiver: This is today — the phone routing has been "
            "repointed now."
        )
    _notify_admin(client, user_id, message)


def handle_holiday_add_submission(ack, body, view, client, schedule, schedule_channel):
    """``view_submission`` for the holiday-add modal."""
    user_id = body["user"]["id"]
    if user_id not in schedule.get_admin_users():
        ack(response_action="errors", errors={"date": "Admin access required."})
        return

    values = view["state"]["values"]
    date_str = values["date"]["date"]["selected_date"]
    slots_raw = values["slots"]["slots"].get("value")
    mult_raw = values["multiplier"]["multiplier"].get("value")
    label = values["label"]["label"].get("value") or ""

    errors = {}
    if date_error := _validate_holiday_date(date_str):
        errors["date"] = date_error
    slots, slots_error = _parse_slots(slots_raw)
    if slots_error:
        errors["slots"] = slots_error
    multiplier, mult_error = _parse_multiplier(mult_raw)
    if mult_error:
        errors["multiplier"] = mult_error
    if label_error := _validate_label(label):
        errors["label"] = label_error
    if errors:
        ack(response_action="errors", errors=errors)
        return

    ack()
    _created, message = _apply_holiday_add(
        schedule, date_str, slots, multiplier, label.strip(), client, schedule_channel
    )
    _notify_admin(client, user_id, message)


# ── App factory ─────────────────────────────────────────────────────────


def create_app(
    bot_token: str, signing_secret: str, schedule_channel: str | None = None
) -> App:
    app = App(
        token=bot_token,
        signing_secret=signing_secret,
        process_before_response=True,
    )
    schedule = ShiftSchedule()

    @app.command("/oncall")
    def handle_oncall(ack, command, respond, client):
        ack()
        dispatch_oncall(command, respond, client, schedule, schedule_channel)

    # Open-shift pickup buttons (``pickup_<date>[_day]``). The negative
    # lookahead keeps this from also matching the late-pickup approval buttons
    # (``pickup_approve_…`` / ``pickup_deny_…``), which have their own handlers.
    @app.action(re.compile(r"^pickup_(?!approve_|deny_)"))
    def handle_pickup_button(ack, body, client, respond):
        ack()
        handle_pickup(body, respond, client, schedule, schedule_channel)

    @app.action(re.compile(r"^pickup_approve_"))
    def handle_pickup_approve_button(ack, body, client, respond):
        ack()
        handle_pickup_approve(body, respond, client, schedule, schedule_channel)

    @app.action(re.compile(r"^pickup_deny_"))
    def handle_pickup_deny_button(ack, body, client, respond):
        ack()
        handle_pickup_deny(body, respond, client, schedule, schedule_channel)

    @app.action(re.compile(r"^swap_accept_"))
    def handle_swap_accept_button(ack, body, client, respond):
        ack()
        handle_swap_accept(body, respond, client, schedule, schedule_channel)

    @app.action(re.compile(r"^swap_decline_"))
    def handle_swap_decline_button(ack, body, client, respond):
        ack()
        handle_swap_decline(body, respond, client, schedule, schedule_channel)

    # Admin App Home buttons → open the corresponding Block Kit modal.
    @app.action(OPEN_OVERRIDE_MODAL_ACTION)
    def handle_open_override_modal(ack, body, client):
        ack()
        open_override_modal(body, client, schedule)

    @app.action(OPEN_HOLIDAY_MODAL_ACTION)
    def handle_open_holiday_modal(ack, body, client):
        ack()
        open_holiday_add_modal(body, client, schedule)

    # Admin modal submissions (ack is handled inside, with field errors).
    @app.view(OVERRIDE_MODAL_CALLBACK)
    def handle_override_view(ack, body, view, client):
        handle_override_submission(ack, body, view, client, schedule, schedule_channel)

    @app.view(HOLIDAY_ADD_MODAL_CALLBACK)
    def handle_holiday_add_view(ack, body, view, client):
        handle_holiday_add_submission(
            ack, body, view, client, schedule, schedule_channel
        )

    @app.event("app_home_opened")
    def handle_app_home_opened(event, client):
        # Fires for the Messages tab too; only (re)publish the Home tab.
        if event.get("tab") != "home":
            return
        publish_home(client, event["user"], _changelog_text(), schedule)

    @app.event("message")
    def handle_message_event(event, client, request):
        retry_num = request.headers.get("x-slack-retry-num") if request else None
        handle_channel_message(
            event, client, schedule, schedule_channel, retry_num=retry_num
        )

    return app