Files
HarborForge.Backend/app/models/calendar.py
hanghang zhang 2cbf6445eb feat(calendar): maintenance window + schedule_type special slots
## What this adds

1. **Maintenance window on ScheduleType**
   - New columns: maintenance_from / maintenance_to (UTC hours, 0-23)
   - Invariant: window is exactly 1 hour (validated in pydantic;
     maintenance_to must equal (maintenance_from + 1) % 24)
   - Default applied via additive migration: 8:00-9:00 UTC for existing
     rows so deployments don't crash on first boot

2. **ScheduleTypeSpecialSlot** — admin-managed slot template
   - New table schedule_type_special_slots
   - Admin (schedule_type.manage) CRUD via
     /schedule-types/{id}/special-slots
   - Fields: name, description, minute_in_window (0-59 inside the
     parent maintenance window), estimated_duration, priority,
     event_data (JSON merged into materialised slot), is_active
   - Unique constraint (schedule_type_id, name) — name is the stable
     human-readable identifier per cohort

3. **Per-agent materialisation**
   - New service app/services/special_slot_materialiser.py
   - GET /calendar/sync calls materialise_special_slots_for_claw
     (idempotent, one row per agent per template per date)
   - GET /calendar/day calls materialise_special_slots_for_user
   - Materialised rows are slot_type=system, event_type=system_event,
     is_admin_locked=true, special_slot_id pointing back to template
   - Plugin's runSync picks them up like any other due slot via the
     normal real-slots query path

4. **Admin-locked enforcement**
   - New TimeSlot columns: is_admin_locked, special_slot_id (FK to
     schedule_type_special_slots, ON DELETE SET NULL)
   - PATCH /calendar/slots/{id}: refuses any edit on admin-locked
     slots (423)
   - POST /calendar/slots/{id}/cancel: refuses cancel on admin-locked
     (423)
   - PATCH /calendar/slots/{id}/agent-update: admin-locked accept only
     ongoing/paused/finished/aborted statuses (423 on other transitions)

5. **Maintenance-window guard on slot creation**
   - POST /calendar/slots: rejects slot_type=system outright (only
     materialiser may create system slots) and rejects any non-system
     slot whose [scheduled_at, +duration] intersects the calling
     user's schedule_type maintenance window (422). Handles 23->0 wrap

6. **Schema response**
   - TimeSlotResponse / CalendarSlotItem now include is_admin_locked
     and special_slot_id so clients can render the lock indicator and
     trace back to the template

## Migration

Additive only — no destructive changes. Lives in _migrate_schema()
in app/main.py; the new schedule_type_special_slots table is created
by Base.metadata.create_all() on first boot.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-22 19:18:42 +01:00

348 lines
10 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""Calendar models — TimeSlot, SchedulePlan and related enums.
TimeSlot represents a single scheduled slot on a user's calendar.
Slots can be created manually or materialized from a SchedulePlan.
SchedulePlan represents a recurring schedule rule that generates
virtual slots on matching dates. Virtual slots are materialized
into real TimeSlot rows on demand (daily pre-compute, or when
edited/cancelled).
See: NEXT_WAVE_DEV_DIRECTION.md §1.1 §1.3
"""
from sqlalchemy import (
Column, Integer, String, Text, DateTime, Date, Time,
ForeignKey, Enum, Boolean, JSON, CheckConstraint,
)
from sqlalchemy.orm import relationship, validates
from sqlalchemy.sql import func
from app.core.config import Base
import enum
# ---------------------------------------------------------------------------
# Enums
# ---------------------------------------------------------------------------
class SlotType(str, enum.Enum):
"""What kind of slot this is."""
WORK = "work"
ON_CALL = "on_call"
ENTERTAINMENT = "entertainment"
SYSTEM = "system"
class SlotStatus(str, enum.Enum):
"""Lifecycle status of a slot."""
NOT_STARTED = "not_started"
ONGOING = "ongoing"
DEFERRED = "deferred"
SKIPPED = "skipped"
PAUSED = "paused"
FINISHED = "finished"
ABORTED = "aborted"
class EventType(str, enum.Enum):
"""High-level event category stored alongside the slot."""
JOB = "job"
ENTERTAINMENT = "entertainment"
SYSTEM_EVENT = "system_event"
class DayOfWeek(str, enum.Enum):
"""Day-of-week for SchedulePlan.on_day."""
SUN = "sun"
MON = "mon"
TUE = "tue"
WED = "wed"
THU = "thu"
FRI = "fri"
SAT = "sat"
class MonthOfYear(str, enum.Enum):
"""Month for SchedulePlan.on_month."""
JAN = "jan"
FEB = "feb"
MAR = "mar"
APR = "apr"
MAY = "may"
JUN = "jun"
JUL = "jul"
AUG = "aug"
SEP = "sep"
OCT = "oct"
NOV = "nov"
DEC = "dec"
# ---------------------------------------------------------------------------
# TimeSlot model
# ---------------------------------------------------------------------------
class TimeSlot(Base):
__tablename__ = "time_slots"
id = Column(Integer, primary_key=True, index=True)
user_id = Column(
Integer,
ForeignKey("users.id"),
nullable=False,
index=True,
comment="Owner of this slot",
)
date = Column(
Date,
nullable=False,
index=True,
comment="Calendar date for this slot",
)
slot_type = Column(
Enum(SlotType, values_callable=lambda x: [e.value for e in x]),
nullable=False,
comment="work | on_call | entertainment | system",
)
estimated_duration = Column(
Integer,
nullable=False,
comment="Estimated duration in minutes (1-50)",
)
scheduled_at = Column(
Time,
nullable=False,
comment="Planned start time (00:00-23:00)",
)
started_at = Column(
Time,
nullable=True,
comment="Actual start time (filled when slot begins)",
)
attended = Column(
Boolean,
default=False,
nullable=False,
comment="Whether the slot has been attended",
)
actual_duration = Column(
Integer,
nullable=True,
comment="Actual duration in minutes (0-65535), no upper design limit",
)
event_type = Column(
Enum(EventType, values_callable=lambda x: [e.value for e in x]),
nullable=True,
comment="job | entertainment | system_event",
)
event_data = Column(
JSON,
nullable=True,
comment="Event details JSON — structure depends on event_type",
)
priority = Column(
Integer,
nullable=False,
default=0,
comment="Priority 0-99, higher = more important",
)
status = Column(
Enum(SlotStatus, values_callable=lambda x: [e.value for e in x]),
nullable=False,
default=SlotStatus.NOT_STARTED,
comment="Lifecycle status of this slot",
)
wakeup_sent_at = Column(
DateTime(timezone=True),
nullable=True,
comment="When Discord wakeup was sent for this slot",
)
plan_id = Column(
Integer,
ForeignKey("schedule_plans.id"),
nullable=True,
comment="Source plan if materialized from a SchedulePlan; set NULL on edit/cancel",
)
# -----------------------------------------------------------------
# Admin-locked slots are materialised from a ScheduleTypeSpecialSlot
# template. The agent can complete / abort / pause / resume them but
# cannot edit their time, type, duration, or cancel them outright —
# the slot exists because admin decided every agent on the parent
# schedule_type should run it. See `_apply_agent_slot_update` for
# the enforcement.
# -----------------------------------------------------------------
is_admin_locked = Column(
Boolean,
nullable=False,
server_default="0",
comment="True for slots materialised from a schedule_type special slot template.",
)
# Pointer back to the template that materialised this slot. NULL for
# all user-created or plan-generated slots. Lets us cascade updates
# and surface 'why is this on my calendar' to the agent.
special_slot_id = Column(
Integer,
ForeignKey("schedule_type_special_slots.id", ondelete="SET NULL"),
nullable=True,
index=True,
)
created_at = Column(DateTime(timezone=True), server_default=func.now())
updated_at = Column(DateTime(timezone=True), onupdate=func.now())
# relationship ----------------------------------------------------------
plan = relationship("SchedulePlan", back_populates="materialized_slots")
special_slot = relationship("ScheduleTypeSpecialSlot")
# ---------------------------------------------------------------------------
# SchedulePlan model
# ---------------------------------------------------------------------------
class SchedulePlan(Base):
"""A recurring schedule rule that generates virtual TimeSlots.
Hierarchy constraint for the period parameters:
• ``at_time`` is always required.
• ``on_month`` requires ``on_week`` (which in turn requires ``on_day``).
• ``on_week`` requires ``on_day``.
Examples:
• ``--at 09:00`` → every day at 09:00
• ``--at 09:00 --on-day sun`` → every Sunday at 09:00
• ``--at 09:00 --on-day sun --on-week 1`` → 1st-week Sunday each month
• ``--at … --on-day sun --on-week 1 --on-month jan`` → Jan 1st-week Sunday
"""
__tablename__ = "schedule_plans"
__table_args__ = (
# on_month requires on_week
CheckConstraint(
"(on_month IS NULL) OR (on_week IS NOT NULL)",
name="ck_plan_month_requires_week",
),
# on_week requires on_day
CheckConstraint(
"(on_week IS NULL) OR (on_day IS NOT NULL)",
name="ck_plan_week_requires_day",
),
)
id = Column(Integer, primary_key=True, index=True)
user_id = Column(
Integer,
ForeignKey("users.id"),
nullable=False,
index=True,
comment="Owner of this plan",
)
# -- slot template fields -----------------------------------------------
slot_type = Column(
Enum(SlotType, values_callable=lambda x: [e.value for e in x]),
nullable=False,
comment="work | on_call | entertainment | system",
)
estimated_duration = Column(
Integer,
nullable=False,
comment="Estimated duration in minutes (1-50)",
)
event_type = Column(
Enum(EventType, values_callable=lambda x: [e.value for e in x]),
nullable=True,
comment="job | entertainment | system_event",
)
event_data = Column(
JSON,
nullable=True,
comment="Event details JSON — copied to materialized slots",
)
# -- period parameters --------------------------------------------------
at_time = Column(
Time,
nullable=False,
comment="Daily scheduled time (--at HH:mm), always required",
)
on_day = Column(
Enum(DayOfWeek, values_callable=lambda x: [e.value for e in x]),
nullable=True,
comment="Day of week (--on-day); NULL = every day",
)
on_week = Column(
Integer,
nullable=True,
comment="Week-of-month 1-4 (--on-week); NULL = every week",
)
on_month = Column(
Enum(MonthOfYear, values_callable=lambda x: [e.value for e in x]),
nullable=True,
comment="Month (--on-month); NULL = every month",
)
is_active = Column(
Boolean,
default=True,
nullable=False,
comment="Soft-delete / plan-cancel flag",
)
created_at = Column(DateTime(timezone=True), server_default=func.now())
updated_at = Column(DateTime(timezone=True), onupdate=func.now())
# relationship ----------------------------------------------------------
materialized_slots = relationship(
"TimeSlot",
back_populates="plan",
lazy="dynamic",
)
# -- application-level validation ---------------------------------------
@validates("on_week")
def _validate_on_week(self, _key: str, value: int | None) -> int | None:
if value is not None and not (1 <= value <= 4):
raise ValueError("on_week must be between 1 and 4")
return value
@validates("on_month")
def _validate_on_month(self, _key: str, value):
"""Enforce: on_month requires on_week (and transitively on_day)."""
if value is not None and self.on_week is None:
raise ValueError(
"on_month requires on_week to be set "
"(hierarchy: on_month → on_week → on_day)"
)
return value
@validates("estimated_duration")
def _validate_estimated_duration(self, _key: str, value: int) -> int:
if not (1 <= value <= 50):
raise ValueError("estimated_duration must be between 1 and 50")
return value