diff --git a/alembic/versions/0043_event_duration_replaces_end_dt.py b/alembic/versions/0043_event_duration_replaces_end_dt.py
new file mode 100644
index 0000000..3fac32a
--- /dev/null
+++ b/alembic/versions/0043_event_duration_replaces_end_dt.py
@@ -0,0 +1,73 @@
+"""replace events.end_dt with duration_minutes (Fable #160)
+
+Revision ID: 0043
+Revises: 0042
+Create Date: 2026-04-29
+
+Structural fix for the "end before start" bug class observed on prod
+2026-04-29: an event landed with end_dt 32 days before start_dt due
+to a tool-call mishap, then disappeared from upcoming-list filters.
+Storing duration instead of end_dt makes the invalid state
+inexpressible at the schema level (duration_minutes >= 0).
+
+Backfill rules:
+  - end_dt valid (end_dt > start_dt) → duration_minutes = total minutes
+  - end_dt == start_dt → duration_minutes = 0 (zero-duration point)
+  - end_dt NULL OR end_dt < start_dt → duration_minutes = NULL (corrupt
+    or open-ended; treated as a point event from here on)
+
+Existing API consumers continue to receive `end_dt` in responses — the
+field is now derived from `start_dt + duration_minutes` in
+``Event.to_dict()`` rather than stored.
+"""
+from alembic import op
+import sqlalchemy as sa
+
+
+revision = "0043"
+down_revision = "0042"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.add_column(
+        "events",
+        sa.Column("duration_minutes", sa.Integer(), nullable=True),
+    )
+    op.create_check_constraint(
+        "events_duration_minutes_non_negative",
+        "events",
+        "duration_minutes IS NULL OR duration_minutes >= 0",
+    )
+    # Backfill: convert valid end_dt into a minute count; leave NULL for
+    # corrupt or absent end_dt. Bad rows (end_dt <= start_dt) collapse
+    # cleanly to point events instead of forcing a recovery guess.
+    op.execute(
+        """
+        UPDATE events
+        SET duration_minutes = CAST(
+            EXTRACT(EPOCH FROM (end_dt - start_dt)) / 60 AS INTEGER
+        )
+        WHERE end_dt IS NOT NULL AND end_dt >= start_dt
+        """
+    )
+    op.drop_column("events", "end_dt")
+
+
+def downgrade() -> None:
+    op.add_column(
+        "events",
+        sa.Column("end_dt", sa.DateTime(timezone=True), nullable=True),
+    )
+    # Restore end_dt = start_dt + duration_minutes minutes for rows that
+    # had a duration. NULL duration → NULL end_dt (point event).
+    op.execute(
+        """
+        UPDATE events
+        SET end_dt = start_dt + (duration_minutes || ' minutes')::interval
+        WHERE duration_minutes IS NOT NULL
+        """
+    )
+    op.drop_constraint("events_duration_minutes_non_negative", "events")
+    op.drop_column("events", "duration_minutes")
diff --git a/src/fabledassistant/models/event.py b/src/fabledassistant/models/event.py
index 672f6a8..a357467 100644
--- a/src/fabledassistant/models/event.py
+++ b/src/fabledassistant/models/event.py
@@ -1,4 +1,4 @@
-from datetime import datetime, timezone
+from datetime import datetime, timedelta, timezone
 
 from sqlalchemy import Boolean, DateTime, ForeignKey, Integer, Text
 from sqlalchemy.orm import Mapped, mapped_column
@@ -20,7 +20,11 @@ class Event(Base):
     uid: Mapped[str] = mapped_column(Text)
     title: Mapped[str] = mapped_column(Text, default="")
     start_dt: Mapped[datetime] = mapped_column(DateTime(timezone=True))
-    end_dt: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
+    # Duration in minutes; NULL = point event with no end specified.
+    # Replaces the prior `end_dt` column (Fable #160 / migration 0043).
+    # The DB has a CHECK constraint that this is NULL or >= 0, so an
+    # event whose end is before its start is structurally inexpressible.
+    duration_minutes: Mapped[int | None] = mapped_column(Integer, nullable=True)
     all_day: Mapped[bool] = mapped_column(Boolean, default=False)
     description: Mapped[str] = mapped_column(Text, default="")
     location: Mapped[str] = mapped_column(Text, default="")
@@ -38,7 +42,21 @@ class Event(Base):
         onupdate=lambda: datetime.now(timezone.utc),
     )
 
+    @property
+    def end_dt(self) -> datetime | None:
+        """Derived end datetime: ``start_dt + duration_minutes``.
+
+        Returns ``None`` for point events (``duration_minutes is None``).
+        Computed at access time rather than stored — a stored end was
+        the source of the "end before start" corruption that motivated
+        this redesign.
+        """
+        if self.duration_minutes is None:
+            return None
+        return self.start_dt + timedelta(minutes=self.duration_minutes)
+
     def to_dict(self) -> dict:
+        end_dt = self.end_dt
         return {
             "id": self.id,
             "user_id": self.user_id,
@@ -47,7 +65,8 @@ class Event(Base):
             "project_id": self.project_id,
             "title": self.title,
             "start_dt": self.start_dt.isoformat() if self.start_dt else None,
-            "end_dt": self.end_dt.isoformat() if self.end_dt else None,
+            "end_dt": end_dt.isoformat() if end_dt else None,
+            "duration_minutes": self.duration_minutes,
             "all_day": self.all_day,
             "description": self.description,
             "location": self.location,
diff --git a/src/fabledassistant/routes/events.py b/src/fabledassistant/routes/events.py
index 63ca1b0..e0f5cfa 100644
--- a/src/fabledassistant/routes/events.py
+++ b/src/fabledassistant/routes/events.py
@@ -54,19 +54,23 @@ async def create_event():
         end_dt = _parse_dt(data["end_dt"]) if data.get("end_dt") else None
     except ValueError:
         return jsonify({"error": "Invalid datetime format"}), 400
-    event = await events_svc.create_event(
-        user_id=_get_current_user_id(),
-        title=data["title"],
-        start_dt=start_dt,
-        end_dt=end_dt,
-        all_day=data.get("all_day", False),
-        description=data.get("description", ""),
-        location=data.get("location", ""),
-        color=data.get("color", ""),
-        recurrence=data.get("recurrence"),
-        project_id=data.get("project_id"),
-        reminder_minutes=data.get("reminder_minutes"),
-    )
+    try:
+        event = await events_svc.create_event(
+            user_id=_get_current_user_id(),
+            title=data["title"],
+            start_dt=start_dt,
+            end_dt=end_dt,
+            duration_minutes=data.get("duration_minutes"),
+            all_day=data.get("all_day", False),
+            description=data.get("description", ""),
+            location=data.get("location", ""),
+            color=data.get("color", ""),
+            recurrence=data.get("recurrence"),
+            project_id=data.get("project_id"),
+            reminder_minutes=data.get("reminder_minutes"),
+        )
+    except ValueError as exc:
+        return jsonify({"error": str(exc)}), 400
     return jsonify(event.to_dict()), 201
 
 
@@ -93,7 +97,7 @@ async def update_event(event_id: int):
     for bool_field in ("all_day",):
         if bool_field in data:
             fields[bool_field] = data[bool_field]
-    for int_field in ("project_id", "reminder_minutes"):
+    for int_field in ("project_id", "reminder_minutes", "duration_minutes"):
         if int_field in data:
             fields[int_field] = data[int_field]
     for dt_field in ("start_dt", "end_dt"):
@@ -106,11 +110,14 @@ async def update_event(event_id: int):
                     fields[dt_field] = _parse_dt(data[dt_field])
                 except ValueError:
                     return jsonify({"error": f"Invalid datetime for {dt_field}"}), 400
-    event = await events_svc.update_event(
-        user_id=_get_current_user_id(),
-        event_id=event_id,
-        **fields,
-    )
+    try:
+        event = await events_svc.update_event(
+            user_id=_get_current_user_id(),
+            event_id=event_id,
+            **fields,
+        )
+    except ValueError as exc:
+        return jsonify({"error": str(exc)}), 400
     if event is None:
         return jsonify({"error": "Event not found"}), 404
     return jsonify(event.to_dict())
diff --git a/src/fabledassistant/services/caldav_sync.py b/src/fabledassistant/services/caldav_sync.py
index 9f6a40a..f35ffee 100644
--- a/src/fabledassistant/services/caldav_sync.py
+++ b/src/fabledassistant/services/caldav_sync.py
@@ -130,6 +130,17 @@ async def sync_user_events(user_id: int) -> dict:
     async with async_session() as session:
         for ev in remote_events:
             caldav_uid = ev["caldav_uid"]
+            # Storage uses duration, not end_dt. Convert here so the
+            # rest of this function can compare/upsert in one shape.
+            ev_start = ev["start_dt"]
+            ev_end = ev["end_dt"]
+            ev_duration = (
+                int((ev_end - ev_start).total_seconds() // 60)
+                if ev_end is not None and ev_start is not None and ev_end > ev_start
+                else None
+            )
+            ev["duration_minutes"] = ev_duration
+
             result = await session.execute(
                 select(Event).where(
                     Event.user_id == user_id,
@@ -145,8 +156,8 @@ async def sync_user_events(user_id: int) -> dict:
                     uid=str(uuid.uuid4()),
                     caldav_uid=caldav_uid,
                     title=ev["title"],
-                    start_dt=ev["start_dt"],
-                    end_dt=ev["end_dt"],
+                    start_dt=ev_start,
+                    duration_minutes=ev_duration,
                     all_day=ev["all_day"],
                     description=ev["description"],
                     location=ev["location"],
@@ -157,7 +168,7 @@ async def sync_user_events(user_id: int) -> dict:
             else:
                 # Update if anything changed
                 changed = False
-                for field in ("title", "start_dt", "end_dt", "all_day", "description", "location", "recurrence"):
+                for field in ("title", "start_dt", "duration_minutes", "all_day", "description", "location", "recurrence"):
                     if getattr(existing, field) != ev[field]:
                         setattr(existing, field, ev[field])
                         changed = True
diff --git a/src/fabledassistant/services/calendar_sync.py b/src/fabledassistant/services/calendar_sync.py
index fc64685..0300556 100644
--- a/src/fabledassistant/services/calendar_sync.py
+++ b/src/fabledassistant/services/calendar_sync.py
@@ -207,6 +207,14 @@ async def sync_event_to_db(user_id: int, ical_uid: str) -> Event | None:
                 d = dtend.dt
                 end_dt = datetime(d.year, d.month, d.day, tzinfo=timezone.utc)
 
+            # Storage uses duration, not end_dt. Convert iCal DTEND to a
+            # minute count anchored on DTSTART. Treat invalid (end <= start)
+            # incoming data as a point event rather than rejecting; we
+            # don't control external CalDAV writers.
+            duration_minutes = None
+            if end_dt is not None and end_dt > start_dt:
+                duration_minutes = int((end_dt - start_dt).total_seconds() // 60)
+
             async with async_session() as session:
                 result = await session.execute(
                     select(Event).where(Event.user_id == user_id, Event.uid == ical_uid)
@@ -215,7 +223,7 @@ async def sync_event_to_db(user_id: int, ical_uid: str) -> Event | None:
                 if existing:
                     existing.title = title
                     existing.start_dt = start_dt
-                    existing.end_dt = end_dt
+                    existing.duration_minutes = duration_minutes
                     existing.all_day = all_day
                     existing.description = description
                     existing.location = location
@@ -230,7 +238,7 @@ async def sync_event_to_db(user_id: int, ical_uid: str) -> Event | None:
                         uid=ical_uid,
                         title=title,
                         start_dt=start_dt,
-                        end_dt=end_dt,
+                        duration_minutes=duration_minutes,
                         all_day=all_day,
                         description=description,
                         location=location,
diff --git a/src/fabledassistant/services/events.py b/src/fabledassistant/services/events.py
index 381e655..13fb989 100644
--- a/src/fabledassistant/services/events.py
+++ b/src/fabledassistant/services/events.py
@@ -1,4 +1,13 @@
-"""Internal event store service with CalDAV push sync."""
+"""Internal event store service with CalDAV push sync.
+
+Storage model: an event is anchored at ``start_dt`` and has an optional
+``duration_minutes``. The end of the event is *derived* via
+``Event.end_dt`` (a Python property), never stored. Callers may still
+pass ``end_dt`` on writes for ergonomic compatibility — the service
+converts to ``duration_minutes`` internally. This rules out the entire
+"end before start" bug class structurally (Fable #160 / migration
+0043). Open-ended events use ``duration_minutes = None``.
+"""
 from __future__ import annotations
 
 import asyncio
@@ -7,7 +16,7 @@ import uuid
 from datetime import datetime, timedelta, timezone
 
 from dateutil.rrule import rrulestr
-from sqlalchemy import and_, or_, select
+from sqlalchemy import or_, select
 
 from fabledassistant.models import async_session
 from fabledassistant.models.event import Event
@@ -15,11 +24,56 @@ from fabledassistant.models.event import Event
 logger = logging.getLogger(__name__)
 
 
+def _normalize_duration(
+    *,
+    start_dt: datetime,
+    end_dt: datetime | None,
+    duration_minutes: int | None,
+) -> int | None:
+    """Reduce (end_dt, duration_minutes) inputs to a single canonical
+    ``duration_minutes`` value.
+
+    Resolution order:
+      1. If ``duration_minutes`` is explicit, use it (validate >= 0).
+         If ``end_dt`` is also given, validate the two agree.
+      2. Otherwise, derive from ``end_dt - start_dt``.
+      3. Otherwise None (point event with no end).
+
+    Raises ``ValueError`` for any invalid combination — duration < 0,
+    end_dt < start_dt, or end_dt and duration_minutes inconsistent.
+    """
+    if duration_minutes is not None:
+        if duration_minutes < 0:
+            raise ValueError(
+                f"duration_minutes must be >= 0, got {duration_minutes}"
+            )
+        if end_dt is not None:
+            expected = int((end_dt - start_dt).total_seconds() // 60)
+            if expected != duration_minutes:
+                raise ValueError(
+                    f"end_dt ({end_dt.isoformat()}) implies "
+                    f"{expected} minutes but duration_minutes={duration_minutes} "
+                    f"was passed; pass only one or make them agree."
+                )
+        return duration_minutes
+    if end_dt is not None:
+        delta_seconds = (end_dt - start_dt).total_seconds()
+        if delta_seconds < 0:
+            raise ValueError(
+                f"end_dt ({end_dt.isoformat()}) must be at or after "
+                f"start_dt ({start_dt.isoformat()}); pass end_dt=None "
+                f"or omit it for point events."
+            )
+        return int(delta_seconds // 60)
+    return None
+
+
 async def create_event(
     user_id: int,
     title: str,
     start_dt: datetime,
     end_dt: datetime | None = None,
+    duration_minutes: int | None = None,
     all_day: bool = False,
     description: str = "",
     location: str = "",
@@ -27,12 +81,25 @@ async def create_event(
     recurrence: str | None = None,
     project_id: int | None = None,
     reminder_minutes: int | None = None,
-    # CalDAV-only fields (not stored in DB, forwarded to push)
+    # ``duration`` is a legacy alias kept for the calendar tool layer
+    # and CalDAV pass-through callers; promotes to duration_minutes
+    # when duration_minutes isn't otherwise specified.
     duration: int | None = None,
     attendees: list[str] | None = None,
     calendar_name: str | None = None,
 ) -> Event:
-    """Create an event in the DB, then fire a CalDAV push task."""
+    """Create an event in the DB, then fire a CalDAV push task.
+
+    Either ``end_dt`` or ``duration_minutes`` may be supplied; the
+    service converts to ``duration_minutes`` internally. Raises
+    ``ValueError`` on invalid combinations (negative duration, end
+    before start, end/duration disagreement).
+    """
+    if duration is not None and duration_minutes is None:
+        duration_minutes = duration
+    duration_minutes = _normalize_duration(
+        start_dt=start_dt, end_dt=end_dt, duration_minutes=duration_minutes,
+    )
     uid = str(uuid.uuid4())
     async with async_session() as session:
         event = Event(
@@ -40,7 +107,7 @@ async def create_event(
             uid=uid,
             title=title,
             start_dt=start_dt,
-            end_dt=end_dt,
+            duration_minutes=duration_minutes,
             all_day=all_day,
             description=description,
             location=location,
@@ -54,7 +121,7 @@ async def create_event(
         await session.refresh(event)
 
     extra_fields = {
-        "duration": duration,
+        "duration": duration_minutes,
         "reminder_minutes": reminder_minutes,
         "attendees": attendees,
         "calendar_name": calendar_name,
@@ -80,66 +147,74 @@ async def list_events(
     """List events for user_id that overlap [date_from, date_to].
 
     Recurring events (with an RRULE recurrence string) are expanded into
-    individual occurrences within the range.  Non-recurring events are
-    returned as-is.  All results are sorted by start time and returned as
-    dicts (same shape as Event.to_dict()).
+    individual occurrences within the range. Non-recurring events are
+    returned as-is. All results are sorted by start time and returned as
+    dicts (same shape as ``Event.to_dict()``).
+
+    Filtering strategy: a coarse SQL prefilter (events that start on or
+    before ``date_to``), then refine in Python using the event's derived
+    end (``start_dt + duration_minutes``). Doing the end-of-event math
+    in SQL would require Postgres-specific interval arithmetic; the
+    Python-side refinement is a few row-loops over a small per-user
+    result set, which is fine for personal-scale data and avoids
+    coupling the query to a specific dialect.
     """
     async with async_session() as session:
-        # Match strategy:
-        #   - Recurring events: fetch all, expand via rrule below.
-        #   - Non-recurring with an end_dt: standard overlap — starts before
-        #     date_to and ends after date_from.
-        #   - Non-recurring with no end_dt: treat as a point event at
-        #     start_dt, include only if start_dt falls within the window.
-        #     (Previously this branch matched any event with a null end_dt,
-        #     returning all past events as "happening today".)
         result = await session.execute(
-            select(Event).where(
+            select(Event)
+            .where(
                 Event.user_id == user_id,
                 or_(
                     Event.recurrence.isnot(None),
-                    and_(
-                        Event.recurrence.is_(None),
-                        Event.start_dt <= date_to,
-                        or_(
-                            Event.end_dt >= date_from,
-                            and_(
-                                Event.end_dt.is_(None),
-                                Event.start_dt >= date_from,
-                            ),
-                        ),
-                    ),
+                    Event.start_dt <= date_to,
                 ),
-            ).order_by(Event.start_dt)
+            )
+            .order_by(Event.start_dt)
         )
         events = list(result.scalars().all())
 
     items: list[dict] = []
     for event in events:
-        if not event.recurrence:
-            items.append(event.to_dict())
+        if event.recurrence:
+            duration = (
+                timedelta(minutes=event.duration_minutes)
+                if event.duration_minutes is not None
+                else None
+            )
+            try:
+                rule = rrulestr(event.recurrence, dtstart=event.start_dt, ignoretz=False)
+                occurrences = rule.between(date_from, date_to, inc=True)
+            except Exception:
+                logger.warning("Failed to expand RRULE for event %d: %r", event.id, event.recurrence)
+                # Fall back to canonical event row; still apply the
+                # window check so a far-future canonical row doesn't
+                # leak into today's list.
+                if date_from <= event.start_dt <= date_to:
+                    items.append(event.to_dict())
+                continue
+
+            base = event.to_dict()
+            for occ in occurrences:
+                if occ.tzinfo is None:
+                    occ = occ.replace(tzinfo=timezone.utc)
+                occurrence_dict = dict(base)
+                occurrence_dict["start_dt"] = occ.isoformat()
+                if duration is not None:
+                    occurrence_dict["end_dt"] = (occ + duration).isoformat()
+                items.append(occurrence_dict)
             continue
 
-        # Expand recurring event occurrences within [date_from, date_to]
-        duration = (event.end_dt - event.start_dt) if event.end_dt else None
-        try:
-            rule = rrulestr(event.recurrence, dtstart=event.start_dt, ignoretz=False)
-            occurrences = rule.between(date_from, date_to, inc=True)
-        except Exception:
-            logger.warning("Failed to expand RRULE for event %d: %r", event.id, event.recurrence)
-            items.append(event.to_dict())
-            continue
-
-        base = event.to_dict()
-        for occ in occurrences:
-            # Ensure occurrence is UTC-aware
-            if occ.tzinfo is None:
-                occ = occ.replace(tzinfo=timezone.utc)
-            occurrence_dict = dict(base)
-            occurrence_dict["start_dt"] = occ.isoformat()
-            if duration is not None:
-                occurrence_dict["end_dt"] = (occ + duration).isoformat()
-            items.append(occurrence_dict)
+        # Non-recurring: refine the coarse prefilter in Python using the
+        # derived end_dt. A point event (duration None) is included when
+        # its start is at or after date_from. A timed event is included
+        # when its end is at or after date_from.
+        derived_end = event.end_dt
+        if derived_end is None:
+            if event.start_dt >= date_from:
+                items.append(event.to_dict())
+        else:
+            if derived_end >= date_from:
+                items.append(event.to_dict())
 
     items.sort(key=lambda x: x["start_dt"])
     return items
@@ -173,7 +248,13 @@ async def search_events(
 
 
 async def update_event(user_id: int, event_id: int, **fields) -> Event | None:
-    """Partial update. Returns updated event or None if not found."""
+    """Partial update. Returns updated event or None if not found.
+
+    Accepts ``end_dt`` or ``duration_minutes`` (or both, validated for
+    agreement). The service converts to ``duration_minutes`` before
+    persisting; ``end_dt`` is never stored. Raises ``ValueError`` for
+    invalid combinations against the post-update state.
+    """
     async with async_session() as session:
         result = await session.execute(
             select(Event).where(Event.id == event_id, Event.user_id == user_id)
@@ -182,10 +263,39 @@ async def update_event(user_id: int, event_id: int, **fields) -> Event | None:
         if event is None:
             return None
         old_title = event.title  # capture before mutation for CalDAV lookup
-        allowed = {"title", "start_dt", "end_dt", "all_day", "description",
-                   "location", "color", "recurrence", "project_id", "reminder_minutes"}
-        # Nullable fields that callers can explicitly set to None to clear
-        nullable = {"end_dt", "recurrence", "project_id", "reminder_minutes"}
+
+        # Resolve any end_dt/duration_minutes inputs against the
+        # post-update start_dt. If neither is in the patch, leave the
+        # existing duration_minutes alone.
+        post_update_start = (
+            fields["start_dt"]
+            if fields.get("start_dt") is not None
+            else event.start_dt
+        )
+        if "end_dt" in fields or "duration_minutes" in fields:
+            new_end = fields.pop("end_dt", None)
+            new_duration = fields.pop("duration_minutes", None)
+            # If end_dt is in the patch but explicitly None, that's a
+            # clear → duration_minutes = None. Same shape duration_minutes=None.
+            if new_end is None and new_duration is None:
+                fields["duration_minutes"] = None
+            else:
+                fields["duration_minutes"] = _normalize_duration(
+                    start_dt=post_update_start,
+                    end_dt=new_end,
+                    duration_minutes=new_duration,
+                )
+
+        allowed = {
+            "title", "start_dt", "duration_minutes", "all_day",
+            "description", "location", "color", "recurrence",
+            "project_id", "reminder_minutes",
+        }
+        # Nullable fields callers can explicitly clear by passing None
+        nullable = {
+            "duration_minutes", "recurrence", "project_id",
+            "reminder_minutes",
+        }
         for key, value in fields.items():
             if key in allowed and (value is not None or key in nullable):
                 setattr(event, key, value)
@@ -255,11 +365,12 @@ async def _push_create(event: Event, user_id: int, extra: dict) -> None:
         )
         if not await is_caldav_configured(user_id):
             return
+        derived_end = event.end_dt  # property: start + duration_minutes
         await caldav_create(
             user_id=user_id,
             title=event.title,
             start=event.start_dt.isoformat(),
-            end=event.end_dt.isoformat() if event.end_dt else None,
+            end=derived_end.isoformat() if derived_end else None,
             description=event.description or None,
             location=event.location or None,
             all_day=event.all_day,
@@ -296,12 +407,13 @@ async def _push_update(event: Event, user_id: int, old_title: str = "") -> None:
             return
         # Use old_title so CalDAV can find the event even if the title was changed
         query_title = old_title or event.title
+        derived_end = event.end_dt
         await caldav_update(
             user_id=user_id,
             query=query_title,
             title=event.title,
             start=event.start_dt.isoformat(),
-            end=event.end_dt.isoformat() if event.end_dt else None,
+            end=derived_end.isoformat() if derived_end else None,
             description=event.description or None,
             location=event.location or None,
         )
diff --git a/tests/test_events_service.py b/tests/test_events_service.py
index bf34538..0089bd5 100644
--- a/tests/test_events_service.py
+++ b/tests/test_events_service.py
@@ -14,7 +14,7 @@ def _make_mock_session():
 
 
 def _make_mock_event(id=1, user_id=1, uid="uid-abc", title="Meeting",
-                     caldav_uid="", color=""):
+                     caldav_uid="", color="", duration_minutes=60):
     e = MagicMock()
     e.id = id
     e.user_id = user_id
@@ -23,7 +23,14 @@ def _make_mock_event(id=1, user_id=1, uid="uid-abc", title="Meeting",
     e.caldav_uid = caldav_uid
     e.color = color
     e.start_dt = datetime(2026, 3, 25, 10, 0, tzinfo=timezone.utc)
-    e.end_dt = datetime(2026, 3, 25, 11, 0, tzinfo=timezone.utc)
+    e.duration_minutes = duration_minutes
+    # end_dt is derived; mirror the property's behavior on the mock so
+    # service code that reads `event.end_dt` gets a sensible value.
+    if duration_minutes is None:
+        e.end_dt = None
+    else:
+        from datetime import timedelta
+        e.end_dt = e.start_dt + timedelta(minutes=duration_minutes)
     e.all_day = False
     e.description = ""
     e.location = ""
@@ -32,8 +39,9 @@ def _make_mock_event(id=1, user_id=1, uid="uid-abc", title="Meeting",
     e.to_dict.return_value = {
         "id": id, "uid": uid, "title": title,
         "caldav_uid": caldav_uid, "color": color,
-        "start_dt": datetime(2026, 3, 25, 10, 0, tzinfo=timezone.utc).isoformat(),
-        "end_dt": datetime(2026, 3, 25, 11, 0, tzinfo=timezone.utc).isoformat(),
+        "start_dt": e.start_dt.isoformat(),
+        "end_dt": e.end_dt.isoformat() if e.end_dt else None,
+        "duration_minutes": duration_minutes,
     }
     return e
 
@@ -124,6 +132,195 @@ async def test_update_event_fires_caldav_push():
         assert mock_task.called
 
 
+# ── Duration-model write-side guarantees (Fable #160) ─────────────────────────
+
+
+def test_normalize_duration_from_end_dt():
+    """end_dt sugar converts to a positive minute count anchored on start."""
+    from fabledassistant.services.events import _normalize_duration
+    start = datetime(2026, 5, 1, 8, 0, tzinfo=timezone.utc)
+    end = datetime(2026, 5, 1, 9, 30, tzinfo=timezone.utc)
+    assert _normalize_duration(start_dt=start, end_dt=end, duration_minutes=None) == 90
+
+
+def test_normalize_duration_zero_is_valid_point_event():
+    """end_dt == start_dt → duration 0. The point-with-zero-duration case
+    is rare but legal (e.g. an instant marker); the duration model treats
+    it the same as duration None for display purposes."""
+    from fabledassistant.services.events import _normalize_duration
+    same = datetime(2026, 5, 1, 8, 0, tzinfo=timezone.utc)
+    assert _normalize_duration(start_dt=same, end_dt=same, duration_minutes=None) == 0
+
+
+def test_normalize_duration_rejects_end_before_start():
+    """The exact 2026-04-29 prod failure: end 32 days before start.
+    The duration model makes this inexpressible at the schema level
+    via a CHECK constraint, but write-path callers still get a
+    helpful ValueError if they construct an inconsistent (start, end)
+    pair via the end_dt sugar."""
+    from fabledassistant.services.events import _normalize_duration
+    start = datetime(2026, 5, 1, 12, 0, tzinfo=timezone.utc)
+    end_before = datetime(2026, 3, 30, 12, 0, tzinfo=timezone.utc)
+    with pytest.raises(ValueError, match="at or after start_dt"):
+        _normalize_duration(
+            start_dt=start, end_dt=end_before, duration_minutes=None,
+        )
+
+
+def test_normalize_duration_rejects_negative_duration():
+    """Direct duration_minutes < 0 is rejected. Mirrors the DB CHECK
+    constraint at the service boundary so callers get a clean error
+    rather than a constraint violation from psycopg."""
+    from fabledassistant.services.events import _normalize_duration
+    start = datetime(2026, 5, 1, 12, 0, tzinfo=timezone.utc)
+    with pytest.raises(ValueError, match="must be >= 0"):
+        _normalize_duration(start_dt=start, end_dt=None, duration_minutes=-15)
+
+
+def test_normalize_duration_rejects_inconsistent_end_and_duration():
+    """If a caller passes both end_dt AND duration_minutes that disagree,
+    the inconsistency is surfaced rather than silently picking one."""
+    from fabledassistant.services.events import _normalize_duration
+    start = datetime(2026, 5, 1, 12, 0, tzinfo=timezone.utc)
+    end = datetime(2026, 5, 1, 13, 0, tzinfo=timezone.utc)  # implies 60 min
+    with pytest.raises(ValueError, match="implies 60 minutes"):
+        _normalize_duration(
+            start_dt=start, end_dt=end, duration_minutes=30,
+        )
+
+
+def test_normalize_duration_none_for_open_ended():
+    """Both inputs None → None duration (open-ended event)."""
+    from fabledassistant.services.events import _normalize_duration
+    start = datetime(2026, 5, 1, 12, 0, tzinfo=timezone.utc)
+    assert _normalize_duration(
+        start_dt=start, end_dt=None, duration_minutes=None,
+    ) is None
+
+
+@pytest.mark.asyncio
+async def test_create_event_rejects_end_before_start():
+    """Service-level rejection — same scenario as the prod bug, surfaced
+    cleanly for tool / route callers via ValueError."""
+    from fabledassistant.services.events import create_event
+    start = datetime(2026, 5, 1, 12, 0, tzinfo=timezone.utc)
+    end_before = datetime(2026, 3, 30, 12, 0, tzinfo=timezone.utc)
+    with pytest.raises(ValueError, match="at or after start_dt"):
+        await create_event(
+            user_id=1, title="Bad",
+            start_dt=start, end_dt=end_before,
+        )
+
+
+@pytest.mark.asyncio
+async def test_update_event_preserves_duration_when_only_start_changes():
+    """Sliding semantics: when the user moves an event by changing only
+    start_dt, the existing duration_minutes is preserved as-is. The new
+    effective end_dt slides forward with the start. This is a behavioral
+    upgrade vs. the old end_dt model, where moving start past the
+    stored end made the event 'go backward in time'."""
+    mock_event = _make_mock_event(duration_minutes=60)  # start 10:00, end 11:00
+    mock_session = _make_mock_session()
+    mock_result = MagicMock()
+    mock_result.scalar_one_or_none.return_value = mock_event
+    mock_session.execute = AsyncMock(return_value=mock_result)
+
+    with patch("fabledassistant.services.events.async_session") as mock_cls, \
+         patch("fabledassistant.services.events.asyncio.create_task"):
+        mock_cls.return_value = mock_session
+        from fabledassistant.services.events import update_event
+        # Move start to 12:00; effective end becomes 13:00 automatically.
+        result = await update_event(
+            user_id=1, event_id=1,
+            start_dt=datetime(2026, 3, 25, 12, 0, tzinfo=timezone.utc),
+        )
+    assert result is not None
+    # duration_minutes was NOT touched; mock_event still has 60.
+    assert mock_event.duration_minutes == 60
+
+
+@pytest.mark.asyncio
+async def test_update_event_clearing_end_dt_clears_duration():
+    """Passing end_dt=None on update is the documented way to clear the
+    end (turn a timed event into a point event). The service must
+    translate that into duration_minutes=None, not leave the prior
+    value in place."""
+    mock_event = _make_mock_event(duration_minutes=60)
+    mock_session = _make_mock_session()
+    mock_result = MagicMock()
+    mock_result.scalar_one_or_none.return_value = mock_event
+    mock_session.execute = AsyncMock(return_value=mock_result)
+
+    with patch("fabledassistant.services.events.async_session") as mock_cls, \
+         patch("fabledassistant.services.events.asyncio.create_task"):
+        mock_cls.return_value = mock_session
+        from fabledassistant.services.events import update_event
+        await update_event(user_id=1, event_id=1, end_dt=None)
+    assert mock_event.duration_minutes is None
+
+
+@pytest.mark.asyncio
+async def test_list_events_includes_point_event_in_window():
+    """A point event (duration_minutes=None) surfaces when its start
+    is in the window. Replaces the prior 'corrupt end_dt' regression
+    test — the duration model can't represent that state, but the
+    same code path is exercised here for point events."""
+    mock_event = _make_mock_event(duration_minutes=None)
+    # Point event in the upcoming window
+    mock_event.start_dt = datetime(2026, 5, 1, 12, 0, tzinfo=timezone.utc)
+    mock_event.end_dt = None
+    mock_event.to_dict.return_value = {
+        "id": 1, "title": "Point",
+        "start_dt": mock_event.start_dt.isoformat(),
+        "end_dt": None,
+        "duration_minutes": None,
+    }
+    mock_session = _make_mock_session()
+    mock_result = MagicMock()
+    mock_result.scalars.return_value.all.return_value = [mock_event]
+    mock_session.execute = AsyncMock(return_value=mock_result)
+
+    with patch("fabledassistant.services.events.async_session") as mock_cls:
+        mock_cls.return_value = mock_session
+        from fabledassistant.services.events import list_events
+        results = await list_events(
+            user_id=1,
+            date_from=datetime(2026, 4, 29, tzinfo=timezone.utc),
+            date_to=datetime(2026, 5, 27, tzinfo=timezone.utc),
+        )
+        assert len(results) == 1
+        assert results[0]["id"] == 1
+
+
+@pytest.mark.asyncio
+async def test_list_events_excludes_timed_event_that_already_ended():
+    """A timed event whose start + duration is before the window must
+    NOT surface. Verifies the Python-side refinement actually works
+    against the coarse SQL prefilter."""
+    mock_event = _make_mock_event(duration_minutes=60)
+    # Start 4/20 12:00, end 4/20 13:00; window is 4/29 → 5/27 — fully past.
+    mock_event.start_dt = datetime(2026, 4, 20, 12, 0, tzinfo=timezone.utc)
+    mock_event.end_dt = datetime(2026, 4, 20, 13, 0, tzinfo=timezone.utc)
+    mock_event.to_dict.return_value = {
+        "id": 1, "start_dt": mock_event.start_dt.isoformat(),
+        "end_dt": mock_event.end_dt.isoformat(), "duration_minutes": 60,
+    }
+    mock_session = _make_mock_session()
+    mock_result = MagicMock()
+    mock_result.scalars.return_value.all.return_value = [mock_event]
+    mock_session.execute = AsyncMock(return_value=mock_result)
+
+    with patch("fabledassistant.services.events.async_session") as mock_cls:
+        mock_cls.return_value = mock_session
+        from fabledassistant.services.events import list_events
+        results = await list_events(
+            user_id=1,
+            date_from=datetime(2026, 4, 29, tzinfo=timezone.utc),
+            date_to=datetime(2026, 5, 27, tzinfo=timezone.utc),
+        )
+        assert results == []
+
+
 @pytest.mark.asyncio
 async def test_tools_calendar_always_available():
     """Calendar tools must appear in get_tools_for_user even without CalDAV."""