feat(schedule): implement Agent schedule application service

backend/pyproject.toml

@@ -15,6 +15,7 @@ dependencies = [
     "psycopg[binary]>=3.2,<4",
     "pydantic>=2.13,<3",
     "python-dotenv>=1.1,<2",
+    "python-dateutil>=2.9,<3",
     "sqlalchemy>=2.0,<3",
     "uvicorn>=0.51,<1",
     "websockets>=15,<16",
@@ -28,6 +29,7 @@ dev = [
     "pytest-asyncio>=1,<2",
     "pytest-cov>=7,<8",
     "ruff>=0.15,<1",
+    "types-python-dateutil>=2.9,<3",
 ]
 
 [tool.hatch.build.targets.wheel]

backend/src/timeflow/business/calendar/__init__.py

@@ -22,7 +22,7 @@
     ScheduleUpdatePatch,
     UpdateScheduleCommand,
 )
-from timeflow.business.calendar.service import ScheduleAgentService
+from timeflow.business.calendar.service import ScheduleAgentService, ScheduleApplicationService
 
 __all__ = [
     "CreateScheduleCommand",
@@ -35,6 +35,7 @@
     "ReminderStrength",
     "ReminderType",
     "ScheduleAgentService",
+    "ScheduleApplicationService",
     "ScheduleBusinessError",
     "ScheduleErrorCode",
     "ScheduleKind",

backend/src/timeflow/business/calendar/ports.py

@@ -0,0 +1,111 @@
+"""Persistence abstractions owned by the schedule business layer."""
+
+from collections.abc import Callable
+from datetime import datetime
+from types import TracebackType
+from typing import Protocol, Self
+
+from timeflow.business.calendar.contracts import (
+    ScheduleOccurrenceOverrideSnapshot,
+    ScheduleSnapshot,
+)
+
+
+class ScheduleRevisionConflictError(RuntimeError):
+    """An account-owned schedule no longer has the expected revision."""
+
+    __slots__ = ("actual_revision", "expected_revision", "schedule_id")
+
+    def __init__(
+        self,
+        *,
+        schedule_id: str,
+        expected_revision: int,
+        actual_revision: int,
+    ) -> None:
+        super().__init__(
+            f"Schedule {schedule_id!r} revision conflict: "
+            f"expected {expected_revision}, found {actual_revision}"
+        )
+        self.schedule_id = schedule_id
+        self.expected_revision = expected_revision
+        self.actual_revision = actual_revision
+
+
+class ScheduleRepositoryPort(Protocol):
+    """Account-scoped persistence operations required by the application service."""
+
+    def add_schedule(self, snapshot: ScheduleSnapshot) -> ScheduleSnapshot: ...
+
+    def get_schedule(
+        self,
+        *,
+        account_id: str,
+        schedule_id: str,
+        include_deleted: bool = False,
+    ) -> ScheduleSnapshot | None: ...
+
+    def list_schedules(
+        self,
+        *,
+        account_id: str,
+        include_deleted: bool = False,
+    ) -> tuple[ScheduleSnapshot, ...]: ...
+
+    def list_schedule_candidates(
+        self,
+        *,
+        account_id: str,
+        starts_at_or_after: datetime | None,
+        starts_before: datetime | None,
+        include_deleted: bool = False,
+    ) -> tuple[ScheduleSnapshot, ...]: ...
+
+    def update_schedule(
+        self,
+        *,
+        snapshot: ScheduleSnapshot,
+        expected_revision: int,
+    ) -> ScheduleSnapshot | None: ...
+
+    def add_occurrence_override(
+        self,
+        *,
+        account_id: str,
+        snapshot: ScheduleOccurrenceOverrideSnapshot,
+    ) -> ScheduleOccurrenceOverrideSnapshot | None: ...
+
+    def list_occurrence_overrides(
+        self,
+        *,
+        account_id: str,
+        schedule_id: str | None = None,
+    ) -> tuple[ScheduleOccurrenceOverrideSnapshot, ...]: ...
+
+
+class ScheduleUnitOfWork(Protocol):
+    """One atomic schedule use-case transaction."""
+
+    schedules: ScheduleRepositoryPort
+
+    def __enter__(self) -> Self: ...
+
+    def __exit__(
+        self,
+        exc_type: type[BaseException] | None,
+        exc_value: BaseException | None,
+        traceback: TracebackType | None,
+    ) -> None: ...
+
+    def commit(self) -> None: ...
+
+
+ScheduleUnitOfWorkFactory = Callable[[], ScheduleUnitOfWork]
+
+
+__all__ = [
+    "ScheduleRepositoryPort",
+    "ScheduleRevisionConflictError",
+    "ScheduleUnitOfWork",
+    "ScheduleUnitOfWorkFactory",
+]

backend/src/timeflow/business/calendar/recurrence.py

@@ -0,0 +1,160 @@
+"""RRULE validation and occurrence selection for schedule use cases."""
+
+from datetime import UTC, datetime, time
+from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
+
+from dateutil.rrule import rrule, rrulebase, rrulestr
+
+from timeflow.business.calendar.contracts import (
+    ScheduleOccurrenceOverrideSnapshot,
+    ScheduleSnapshot,
+)
+
+
+class InvalidRecurrenceRuleError(ValueError):
+    """A recurrence rule cannot be expanded from its schedule start."""
+
+
+class InvalidTimezoneKeyError(ValueError):
+    """A schedule timezone is not a valid IANA key."""
+
+
+def get_schedule_timezone(key: str) -> ZoneInfo:
+    """Return an IANA timezone while normalizing known invalid-key failures."""
+    try:
+        return ZoneInfo(key)
+    except (ZoneInfoNotFoundError, ValueError) as exc:
+        raise InvalidTimezoneKeyError(key) from exc
+
+
+def normalize_recurrence_rule(rule: str) -> str:
+    """Return the canonical body of exactly one RRULE.
+
+    The shared contract stores an RRULE body such as ``FREQ=WEEKLY;BYDAY=MO``.
+    A single legacy-style ``RRULE:`` prefix is accepted and removed, while
+    recurrence sets, content lines, and embedded DTSTART/RDATE/EXDATE data are
+    rejected before dateutil parses the rule body.
+    """
+    normalized = rule.strip()
+    if not normalized or "\n" in normalized or "\r" in normalized:
+        raise InvalidRecurrenceRuleError("recurrence_rule must contain one RRULE")
+    if normalized[:6].upper() == "RRULE:":
+        normalized = normalized[6:]
+    if not normalized or ":" in normalized:
+        raise InvalidRecurrenceRuleError("recurrence_rule must contain only one RRULE body")
+    return normalized
+
+
+def parse_recurrence_rule(rule: str, *, start_time: datetime) -> rrulebase:
+    """Parse one RFC 5545 RRULE using the schedule start as DTSTART."""
+    normalized = normalize_recurrence_rule(rule)
+    try:
+        parsed = rrulestr(normalized, dtstart=start_time)
+        if not isinstance(parsed, rrule):
+            raise InvalidRecurrenceRuleError("recurrence_rule must contain exactly one RRULE")
+        first = parsed.after(start_time, inc=True)
+    except (TypeError, ValueError, OverflowError) as exc:
+        raise InvalidRecurrenceRuleError("recurrence_rule is not a valid RFC 5545 RRULE") from exc
+    if first is None:
+        raise InvalidRecurrenceRuleError("recurrence_rule has no occurrence at or after start_time")
+    return parsed
+
+
+def first_active_occurrence_on_or_after_local_date(
+    schedule: ScheduleSnapshot,
+    *,
+    now: datetime,
+    overrides: tuple[ScheduleOccurrenceOverrideSnapshot, ...],
+) -> datetime | None:
+    """Return the first non-overridden occurrence on/after today's local date."""
+    if schedule.start_time is None or schedule.recurrence_rule is None:
+        return None
+    timezone = get_schedule_timezone(schedule.timezone)
+    local_start = schedule.start_time.astimezone(timezone)
+    local_date = now.astimezone(timezone).date()
+    boundary = datetime.combine(local_date, time.min, tzinfo=timezone)
+    rule = parse_recurrence_rule(schedule.recurrence_rule, start_time=local_start)
+    overridden = {override.occurrence_start for override in overrides}
+    occurrence = rule.after(boundary, inc=True)
+    while occurrence is not None and occurrence in overridden:
+        occurrence = rule.after(occurrence, inc=False)
+    return None if occurrence is None else occurrence.astimezone(UTC)
+
+
+def first_occurrence_in_window(
+    schedule: ScheduleSnapshot,
+    *,
+    starts_at_or_after: datetime | None,
+    starts_before: datetime | None,
+    excluded_occurrence_starts: frozenset[datetime],
+) -> datetime | None:
+    """Find one non-overridden occurrence in a half-open query window.
+
+    ``after``/``before`` jump directly to the window edge. Iteration advances
+    only when that occurrence has an override, rather than replaying history
+    from the schedule start.
+    """
+    if schedule.start_time is None or schedule.recurrence_rule is None:
+        return None
+    timezone = get_schedule_timezone(schedule.timezone)
+    local_start = schedule.start_time.astimezone(timezone)
+    rule = parse_recurrence_rule(schedule.recurrence_rule, start_time=local_start)
+    local_lower = None if starts_at_or_after is None else starts_at_or_after.astimezone(timezone)
+    local_upper = None if starts_before is None else starts_before.astimezone(timezone)
+
+    if local_lower is not None:
+        occurrence = rule.after(local_lower, inc=True)
+        while occurrence is not None and (local_upper is None or occurrence < local_upper):
+            utc_occurrence = occurrence.astimezone(UTC)
+            if utc_occurrence not in excluded_occurrence_starts:
+                return utc_occurrence
+            occurrence = rule.after(occurrence, inc=False)
+        return None
+
+    if local_upper is None:
+        return None
+    occurrence = rule.before(local_upper, inc=False)
+    while occurrence is not None and occurrence >= local_start:
+        utc_occurrence = occurrence.astimezone(UTC)
+        if utc_occurrence not in excluded_occurrence_starts:
+            return utc_occurrence
+        occurrence = rule.before(occurrence, inc=False)
+    return None
+
+
+def truncate_rule_before_occurrence(
+    schedule: ScheduleSnapshot,
+    occurrence: datetime,
+) -> str | None:
+    """Return an RRULE ending at the prior occurrence, or None for the first one."""
+    if schedule.start_time is None or schedule.recurrence_rule is None:
+        return None
+    timezone = get_schedule_timezone(schedule.timezone)
+    local_start = schedule.start_time.astimezone(timezone)
+    local_occurrence = occurrence.astimezone(timezone)
+    rule = parse_recurrence_rule(schedule.recurrence_rule, start_time=local_start)
+    previous = rule.before(local_occurrence, inc=False)
+    if previous is None:
+        return None
+
+    components = [
+        component
+        for component in normalize_recurrence_rule(schedule.recurrence_rule).split(";")
+        if not component.upper().startswith(("UNTIL=", "COUNT="))
+    ]
+    until = previous.astimezone(UTC).strftime("%Y%m%dT%H%M%SZ")
+    truncated = ";".join((*components, f"UNTIL={until}"))
+    parse_recurrence_rule(truncated, start_time=local_start)
+    return truncated
+
+
+__all__ = [
+    "InvalidRecurrenceRuleError",
+    "InvalidTimezoneKeyError",
+    "first_active_occurrence_on_or_after_local_date",
+    "first_occurrence_in_window",
+    "get_schedule_timezone",
+    "normalize_recurrence_rule",
+    "parse_recurrence_rule",
+    "truncate_rule_before_occurrence",
+]