.github/dependabot.yml

@@ -1,12 +1,7 @@
 version: 2
 updates:
   - package-ecosystem: "pip"
-    directories:
-      - "/src/slack-bot"
-      - "/src/weekly-post"
-      - "/src/roster-sync"
-      - "/src/ring-scheduler"
-      - "/src/shared"
+    directory: "/"
     schedule:
       interval: "weekly"
     assignees:

src/ring-scheduler/app.py

@@ -11,18 +11,21 @@ from datetime import datetime
 from zoneinfo import ZoneInfo
 
 from shared.ring_scheduler import update_queue_routing
-from shared.schedule import (
-    FALLBACK_EXTENSION,
-    WEEKEND_DAYS,
-    ShiftSchedule,
-    determine_shift_type,
-)
+from shared.schedule import FALLBACK_EXTENSION, ShiftSchedule
 from shared.secrets import get_secret
 
 logger = logging.getLogger()
 logger.setLevel(logging.INFO)
 
 EASTERN = ZoneInfo("America/New_York")
+WEEKEND_DAYS = {"Saturday", "Sunday"}
+
+
+def determine_shift_type(now: datetime) -> str:
+    day_name = now.strftime("%A")
+    if day_name in WEEKEND_DAYS and now.hour < 17:
+        return "day"
+    return "night"
 
 
 def handler(event, context):
@@ -52,18 +55,13 @@ def handler(event, context):
         extension = ext if source != "available" else FALLBACK_EXTENSION
 
     secret_prefix = os.environ["TCX_SECRET_PREFIX"]
-    try:
-        result = update_queue_routing(
-            extension=extension,
-            queue_number=os.environ["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 routing")
-        return {"error": True, "date": now.strftime("%Y-%m-%d"), "day": day_name}
-
+    result = update_queue_routing(
+        extension=extension,
+        queue_number=os.environ["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"),
+    )
     result.update(
         {
             "date": now.strftime("%Y-%m-%d"),

src/shared/python/shared/blocks.py

@@ -130,11 +130,7 @@ def build_shift_change_message(
     """Build a channel notification for a shift change."""
     dt = datetime.strptime(date_str, "%Y-%m-%d")
     day_label = dt.strftime("%A, %b %-d")
-    type_label = (
-        f" ({SHIFT_LABELS.get(shift_type, shift_type)})"
-        if dt.strftime("%A") in WEEKEND_DAYS
-        else ""
-    )
+    type_label = f" ({SHIFT_LABELS.get(shift_type, shift_type)})" if shift_type == "day" else ""
 
     if action == "picked_up":
         text = f":white_check_mark: <@{user_id}> picked up the *{day_label}*{type_label} shift (Ext {ext})"

src/shared/python/shared/ring_scheduler.py

@@ -27,5 +27,7 @@ def update_queue_routing(
         closed_destination=extension,
         holiday_destination=extension,
     )
-    logger.info("Updated queue %s to forward to Ext %s", queue_number, extension)
+    logger.info(
+        "Updated queue %s to forward to Ext %s", queue_number, extension
+    )
     return {"extension": extension, "queue": queue_number}

src/shared/python/shared/schedule.py

@@ -8,30 +8,14 @@ Single-table design:
 """
 
 import os
-from datetime import datetime
-from zoneinfo import ZoneInfo
 
 import boto3
 from boto3.dynamodb.conditions import Key
 
-EASTERN = ZoneInfo("America/New_York")
-WEEKEND_DAYS = {"Saturday", "Sunday"}
+
 FALLBACK_EXTENSION = "100"
 
 
-def determine_shift_type(now: datetime | None = None) -> str:
-    """Return the currently active shift type: 'day' or 'night'.
-
-    Weekend 8am-5pm is the day shift; all other times are night.
-    """
-    if now is None:
-        now = datetime.now(EASTERN)
-    day_name = now.strftime("%A")
-    if day_name in WEEKEND_DAYS and 8 <= now.hour < 17:
-        return "day"
-    return "night"
-
-
 class ShiftSchedule:
     def __init__(self, table_name: str | None = None):
         dynamodb = boto3.resource("dynamodb")
@@ -161,7 +145,9 @@ class ShiftSchedule:
     # ── Schedule post tracking ───────────────────────────────────────────
 
     def get_schedule_post(self, channel_id: str) -> dict | None:
-        resp = self.table.get_item(Key={"PK": "SCHEDULE_POST", "SK": channel_id})
+        resp = self.table.get_item(
+            Key={"PK": "SCHEDULE_POST", "SK": channel_id}
+        )
         return resp.get("Item")
 
     def save_schedule_post(
@@ -197,22 +183,16 @@ class ShiftSchedule:
         config = self.get_config()
         return config.get("admin_users", [])
 
-    def add_roster_entry(self, extension: str, name: str) -> bool:
-        """Add a new roster entry. Returns False if extension already exists."""
-        try:
-            self.table.put_item(
-                Item={
-                    "PK": "ROSTER",
-                    "SK": extension,
-                    "name": name,
-                    "extension": extension,
-                    "slack_user_id": "",
-                },
-                ConditionExpression="attribute_not_exists(PK)",
-            )
-            return True
-        except self.table.meta.client.exceptions.ConditionalCheckFailedException:
-            return False
+    def add_roster_entry(self, extension: str, name: str) -> None:
+        self.table.put_item(
+            Item={
+                "PK": "ROSTER",
+                "SK": extension,
+                "name": name,
+                "extension": extension,
+                "slack_user_id": "",
+            }
+        )
 
     def remove_roster_entry(self, extension: str) -> None:
         self.table.delete_item(Key={"PK": "ROSTER", "SK": extension})

src/slack-bot/app.py

@@ -16,12 +16,7 @@ from shared.blocks import (
     build_week_schedule,
 )
 from shared.ring_scheduler import update_queue_routing
-from shared.schedule import (
-    FALLBACK_EXTENSION,
-    WEEKEND_DAYS,
-    ShiftSchedule,
-    determine_shift_type,
-)
+from shared.schedule import FALLBACK_EXTENSION, ShiftSchedule
 from shared.secrets import get_secret
 
 logger = logging.getLogger(__name__)
@@ -93,28 +88,6 @@ 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 _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
-
-
 def create_app(
     bot_token: str, signing_secret: str, schedule_channel: str | None = None
 ) -> App:
@@ -213,15 +186,11 @@ def create_app(
             )
             return
 
-        if is_today(date_str) and _is_active_shift_type(shift_type):
+        if is_today(date_str) and shift_type == "night":
             _update_3cx_routing(employee["extension"])
 
         blocks = build_shift_change_message(
-            user_id,
-            date_str,
-            "picked_up",
-            employee["extension"],
-            employee["name"],
+            user_id, date_str, "picked_up", employee["extension"], employee["name"],
             shift_type=shift_type,
         )
         respond(
@@ -249,7 +218,7 @@ def create_app(
                 channel=channel,
                 ts=post["message_ts"],
                 blocks=blocks,
-                text=f"After-Hours Schedule — {this_monday.strftime('%b %-d')} to {end_date.strftime('%b %-d')}",
+                text=f"After-Hours Schedule — {now.strftime('%b %-d')} to {end_date.strftime('%b %-d')}",
             )
         except Exception:
             logger.warning("Could not update schedule post", exc_info=True)
@@ -395,19 +364,7 @@ def create_app(
             return
 
         day_name = date.strftime("%A")
-
-        # On weekends, find the first available shift (day then night)
-        shift_type = "night"
-        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":
-                    shift_type = st
-                    break
-            else:
-                ext, name, source = schedule.resolve_shift(date_str, day_name)
-        else:
-            ext, name, source = schedule.resolve_shift(date_str, day_name)
+        ext, name, source = schedule.resolve_shift(date_str, day_name)
 
         # Already assigned to someone else (not open)
         if source in ("weekly", "override") and ext != FALLBACK_EXTENSION:
@@ -416,22 +373,15 @@ def create_app(
             )
             return
 
-        schedule.set_override(
-            date_str, employee["extension"], employee["name"], shift_type
-        )
+        schedule.set_override(date_str, employee["extension"], employee["name"])
 
-        if is_today(date_str) and _is_active_shift_type(shift_type):
+        if is_today(date_str):
             _update_3cx_routing(employee["extension"])
 
         respond(text=f"You picked up the shift for *{date.strftime('%A, %b %-d')}*.")
 
         blocks = build_shift_change_message(
-            user_id,
-            date_str,
-            "picked_up",
-            employee["extension"],
-            employee["name"],
-            shift_type=shift_type,
+            user_id, date_str, "picked_up", employee["extension"], employee["name"]
         )
         try:
             client.chat_postMessage(
@@ -469,28 +419,22 @@ def create_app(
             return
 
         day_name = date.strftime("%A")
-        found = _find_employee_shift(
-            schedule, date_str, day_name, employee["extension"]
-        )
+        ext, name, source = schedule.resolve_shift(date_str, day_name)
 
-        if not found:
-            ext, name, _source = schedule.resolve_shift(date_str, day_name)
+        if ext != employee["extension"]:
             respond(text=f"That's not your shift — it belongs to {name} (Ext {ext}).")
             return
 
-        ext, name, source, shift_type = found
-        schedule.mark_open(date_str, shift_type)
+        schedule.mark_open(date_str)
 
-        if is_today(date_str) and _is_active_shift_type(shift_type):
+        if is_today(date_str):
             _update_3cx_routing(FALLBACK_EXTENSION)
 
         respond(
             text=f"You dropped the shift for *{date.strftime('%A, %b %-d')}*. It's now open for pickup."
         )
 
-        blocks = build_shift_change_message(
-            user_id, date_str, "dropped", ext, name, shift_type=shift_type
-        )
+        blocks = build_shift_change_message(user_id, date_str, "dropped", ext, name)
         try:
             client.chat_postMessage(
                 channel=channel_id, blocks=blocks, text=f"Shift dropped for {date_str}"
@@ -521,24 +465,15 @@ def create_app(
             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")
-        found = _find_employee_shift(
-            schedule, date_str, day_name, employee["extension"]
-        )
+        ext, name, source = schedule.resolve_shift(date_str, day_name)
 
-        if not found:
-            ext, name, _source = schedule.resolve_shift(date_str, day_name)
+        if ext != employee["extension"]:
             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)
@@ -556,9 +491,9 @@ def create_app(
                 respond(text=f"Extension `{target_text}` not found in the roster.")
                 return
 
-        schedule.set_override(date_str, target["extension"], target["name"], shift_type)
+        schedule.set_override(date_str, target["extension"], target["name"])
 
-        if is_today(date_str) and _is_active_shift_type(shift_type):
+        if is_today(date_str):
             _update_3cx_routing(target["extension"])
 
         blocks = build_shift_change_message(
@@ -567,7 +502,6 @@ def create_app(
             "swapped",
             target["extension"],
             target["name"],
-            shift_type=shift_type,
         )
         respond(
             text=f"Swapped *{date.strftime('%A, %b %-d')}* to {target['name']} (Ext {target['extension']})."
@@ -591,9 +525,9 @@ def create_app(
             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 override <date> <ext>` — Assign shift\n"
+                    "`admin open <date>` — Mark open\n"
+                    "`admin clear <date>` — Remove override\n"
                     "`admin roster add <ext> <name>` — Add employee\n"
                     "`admin roster remove <ext>` — Remove employee\n"
                     "`admin roster rename <ext> <name>` — Rename"
@@ -605,75 +539,49 @@ def create_app(
 
         if subcmd == "override":
             if len(parts) < 4:
-                respond(
-                    text="Usage: `/oncall admin override <date> <extension> [day|night]`"
-                )
+                respond(text="Usage: `/oncall admin override <date> <extension>`")
                 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 = schedule.get_employee_by_extension(ext)
             if not employee:
                 respond(text=f"Extension `{ext}` not found in the roster.")
                 return
             date_str = date.strftime("%Y-%m-%d")
-            schedule.set_override(
-                date_str, employee["extension"], employee["name"], shift_type
-            )
-            if is_today(date_str) and _is_active_shift_type(shift_type):
-                _update_3cx_routing(employee["extension"])
-            label = "Day" if shift_type == "day" else "Night"
+            schedule.set_override(date_str, employee["extension"], employee["name"])
             respond(
-                text=f"Override set: *{date.strftime('%A, %b %-d')}* ({label}) → {employee['name']} (Ext {ext})"
+                text=f"Override set: *{date.strftime('%A, %b %-d')}* → {employee['name']} (Ext {ext})"
             )
             _refresh_schedule_post(client)
 
         elif subcmd == "open":
             if len(parts) < 3:
-                respond(text="Usage: `/oncall admin open <date> [day|night]`")
+                respond(text="Usage: `/oncall admin open <date>`")
                 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.")
+            schedule.mark_open(date_str)
+            respond(text=f"*{date.strftime('%A, %b %-d')}* marked as open.")
             _refresh_schedule_post(client)
 
         elif subcmd == "clear":
             if len(parts) < 3:
-                respond(text="Usage: `/oncall admin clear <date> [day|night]`")
+                respond(text="Usage: `/oncall admin clear <date>`")
                 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"
+            schedule.remove_override(date_str)
             respond(
-                text=f"Override cleared for *{date.strftime('%A, %b %-d')}* ({label}) — reverted to weekly schedule."
+                text=f"Override cleared for *{date.strftime('%A, %b %-d')}* — reverted to weekly schedule."
             )
             _refresh_schedule_post(client)
 
@@ -689,12 +597,7 @@ def create_app(
                     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
+                schedule.add_roster_entry(ext, name)
                 respond(text=f"Added *{name}* (Ext {ext}) to the roster.")
 
             elif roster_cmd == "remove":
@@ -727,9 +630,7 @@ def create_app(
                 )
 
             else:
-                respond(
-                    text="Unknown roster command. Use `add`, `remove`, or `rename`."
-                )
+                respond(text="Unknown roster command. Use `add`, `remove`, or `rename`.")
 
         else:
             respond(text=f"Unknown admin command: `{subcmd}`. Try `/oncall help`.")

src/weekly-post/app.py

@@ -212,9 +212,7 @@ def handler(event, context):
     schedule.save_schedule_post(
         channel_id, result["ts"], this_monday.strftime("%Y-%m-%d")
     )
-    logger.info(
-        "Posted weekly schedule to channel %s (ts=%s)", channel_id, result["ts"]
-    )
+    logger.info("Posted weekly schedule to channel %s (ts=%s)", channel_id, result["ts"])
     return {
         "posted": True,
         "channel": channel_id,

template.yaml

@@ -195,9 +195,12 @@ Resources:
           QUEUE_NUMBER: !Ref QueueNumber
           TZ: !Ref Timezone
       Policies:
-        - DynamoDBCrudPolicy:
-            TableName: !Ref ShiftTable
         - Statement:
+            - Effect: Allow
+              Action:
+                - dynamodb:GetItem
+              Resource:
+                - !GetAtt ShiftTable.Arn
             - Effect: Allow
               Action:
                 - secretsmanager:GetSecretValue