"""Calendar and event tools.""" from __future__ import annotations import re from datetime import date as _date, datetime, time as _time, timezone from fabledassistant.services.events import ( create_event as events_create_event, delete_event as events_delete_event, find_events_by_query, list_events as events_list_events, search_events as events_search_events, update_event as events_update_event, ) from fabledassistant.services.tools._helpers import resolve_project from fabledassistant.services.tools._registry import tool from fabledassistant.services.tz import get_user_tz _DATE_RE = re.compile(r"^\d{4}-\d{2}-\d{2}$") _TIME_RE = re.compile(r"^\d{2}:\d{2}(:\d{2})?$") async def _combine_local_in_user_tz( user_id: int, date_str: str, time_str: str | None ) -> datetime: """Build a UTC datetime from separate date and time strings. The whole point of this helper: a `YYYY-MM-DD` string carries no TZ metadata that a model could mis-tag, so the calendar day cannot drift across the local→UTC boundary. The wall-clock `HH:MM` likewise has no TZ; we attach the user's local zone explicitly via ``datetime.combine`` and then convert to UTC for storage. Strict shape validation rejects anything that isn't a bare date or a bare time — no `2026-05-01Z`, no `08:00 UTC` slipping through. """ if not _DATE_RE.match(date_str): raise ValueError( f"start_date / end_date must be YYYY-MM-DD with no timezone; got {date_str!r}" ) d = _date.fromisoformat(date_str) if time_str is None or time_str == "": t = _time(0, 0) else: if not _TIME_RE.match(time_str): raise ValueError( f"start_time / end_time must be HH:MM (or HH:MM:SS), no timezone; got {time_str!r}" ) t = _time.fromisoformat(time_str) user_tz = await get_user_tz(user_id) local = datetime.combine(d, t, tzinfo=user_tz) return local.astimezone(timezone.utc) async def _parse_datetime_in_user_tz( user_id: int, value: str ) -> tuple[datetime, bool]: """Legacy single-string parser. Kept as a fallback when the model emits the older `start` / `end` shape; new calls should use `start_date`+`start_time` (and `end_date`+`end_time`) which sidestep the TZ-tagging foot-gun this parser is vulnerable to. Naive inputs are interpreted in the **user's local timezone** and then converted to UTC for storage. Never default to UTC for naive inputs — that's how all-day events landed on the wrong day for non-UTC users. Returns ``(utc_datetime, was_date_only)``. """ was_date_only = "T" not in value and " " not in value if was_date_only: value = f"{value}T00:00:00" dt = datetime.fromisoformat(value.replace("Z", "+00:00")) if dt.tzinfo is None: user_tz = await get_user_tz(user_id) dt = dt.replace(tzinfo=user_tz) return dt.astimezone(timezone.utc), was_date_only async def _resolve_event_start( user_id: int, args: dict ) -> tuple[datetime, bool]: """Resolve the start datetime from either the new split fields (`start_date` + optional `start_time`) or the legacy combined `start`. Returns ``(utc_datetime, was_date_only)``.""" if "start_date" in args and args["start_date"]: date_str = args["start_date"] time_str = args.get("start_time") or None return await _combine_local_in_user_tz(user_id, date_str, time_str), time_str is None if "start" in args and args["start"]: return await _parse_datetime_in_user_tz(user_id, args["start"]) raise ValueError("Either start_date or start is required") async def _resolve_event_end( user_id: int, args: dict ) -> datetime | None: """Resolve the end datetime from either the new split fields or the legacy combined `end`. Returns ``None`` when no end fields are set.""" if "end_date" in args and args["end_date"]: date_str = args["end_date"] time_str = args.get("end_time") or None return await _combine_local_in_user_tz(user_id, date_str, time_str) if "end" in args and args["end"]: dt, _ = await _parse_datetime_in_user_tz(user_id, args["end"]) return dt return None def _candidate_summary(event) -> dict: """Compact event summary used in ambiguous-match responses. Keeps the candidate list small so the model can disambiguate from the same turn without bloating context. Includes id (for the follow-up call), title, start_dt, and location when present. """ return { "id": event.id, "title": event.title, "start_dt": event.start_dt.isoformat() if event.start_dt else None, "location": event.location or None, } async def _resolve_event_for_action( *, user_id: int, arguments: dict, action: str, ): """Pick the single event the model intends to update or delete. Resolution rules: - ``event_id`` in arguments → exact lookup (skip query). Used by the model to disambiguate after a multi-match refusal. - else ``query`` → ``find_events_by_query``: - 0 results → return error tuple ("not_found", ...) - 1 result → return that event - 2+ results → return ("ambiguous", error, candidates) so the caller can refuse the call and show candidates to the model. Returns either an Event (success) or a 2- or 3-tuple of ``(error_kind, error_dict)`` for the caller to translate into a tool-call response. """ from fabledassistant.services.events import get_event event_id = arguments.get("event_id") if event_id is not None: try: event_id_int = int(event_id) except (TypeError, ValueError): return ("invalid_id", { "success": False, "error": f"event_id must be an integer; got {event_id!r}.", }) ev = await get_event(user_id=user_id, event_id=event_id_int) if ev is None: return ("not_found", { "success": False, "error": f"No event found with id={event_id_int}.", }) return ev query = arguments.get("query", "") if not query: return ("invalid_query", { "success": False, "error": "Either query or event_id is required.", }) matches = await find_events_by_query(user_id=user_id, query=query) if not matches: return ("not_found", { "success": False, "error": f"No event found matching {query!r}.", }) if len(matches) == 1: return matches[0] # Multi-match: refuse and surface candidates so the model can # disambiguate via event_id on the next call. Prevents the silent- # picks-matches[0] failure mode that mutated the wrong event in the # 2026-04-29 dentist-appointment incident (Fable #161). return ("ambiguous", { "success": False, "error": ( f"Found {len(matches)} events matching {query!r}. " f"Pick one by passing `event_id` instead of `query`, " f"or refine the search term to match a single event." ), "action": action, "candidates": [_candidate_summary(m) for m in matches[:8]], }) def _validate_weekday(start_dt_utc: datetime, user_tz, expected: str | None) -> str | None: """Verify the resolved local date falls on the expected day of the week. Models routinely miscompute "this Friday" / "next Monday" when the system prompt only carries an ISO date without a weekday. When the model passes `expected_weekday`, the backend rejects mismatches with a self-correcting error message naming the actual weekday. Returns an error string on mismatch, or ``None`` when the check passes (or no expected weekday was supplied). """ if not expected: return None expected_norm = expected.strip().lower() valid = {"monday", "tuesday", "wednesday", "thursday", "friday", "saturday", "sunday"} if expected_norm not in valid: return f"expected_weekday must be a full English weekday name; got {expected!r}." local = start_dt_utc.astimezone(user_tz) actual = local.strftime("%A").lower() if actual == expected_norm: return None return ( f"Date {local.date().isoformat()} falls on {actual.title()}, " f"not {expected_norm.title()}. Recompute the date for " f"{expected_norm.title()} or confirm with the user before retrying." ) @tool( name="create_event", description=( "Create a calendar event for the user. Use this when the user asks " "to schedule, add, or create a meeting, appointment, or event. " "Always pass `start_date` (YYYY-MM-DD) and `start_time` (HH:MM) as " "separate fields in the user's local time — never combine them and " "never include a timezone suffix. The server attaches the user's " "configured timezone. Omit `start_time` (or set `all_day=true`) " "for all-day events like birthdays or holidays. " "When the user names a weekday ('this Friday', 'next Monday'), " "state the resolved calendar date in your reply BEFORE calling " "this tool, and pass `expected_weekday` so the server can verify " "the date falls on the day you intended.\n\n" "DON'T call this tool with placeholder values. If the user " "mentions an event without giving you concrete details (a real " "title, a specific time, and location when it applies), record " "a moment instead and ask for the missing pieces — do NOT create " "an event with a stand-in title like 'Appointment', 'Meeting', " "or 'Event' and a description that says 'details TBD'. Wait for " "the user's reply, then call create_event ONCE with the actual " "title, time, and location. Premature placeholder events pollute " "the calendar and require an immediate update_event to fix — " "both visible to the user, neither what they asked for." ), parameters={ "title": {"type": "string", "description": "A descriptive event title"}, "start_date": {"type": "string", "description": "Start calendar date as YYYY-MM-DD in the user's local time. No timezone suffix."}, "start_time": {"type": "string", "description": "Start wall-clock time as HH:MM (24-hour). Omit for all-day events. No timezone suffix."}, "end_date": {"type": "string", "description": "Optional end calendar date as YYYY-MM-DD."}, "end_time": {"type": "string", "description": "Optional end wall-clock time as HH:MM."}, "duration": {"type": "integer", "description": "Optional duration in minutes (default 60, ignored if end_date/end_time is set or all_day is true)"}, "description": {"type": "string", "description": "Optional event description"}, "location": {"type": "string", "description": "Optional event location"}, "color": {"type": "string", "description": "Optional hex color for the event (e.g. '#6366f1')"}, "all_day": {"type": "boolean", "description": "True for all-day events (birthdays, holidays, deadlines)"}, "recurrence": {"type": "string", "description": "Optional iCalendar RRULE (e.g. 'FREQ=YEARLY' for annual, 'FREQ=WEEKLY' for weekly, 'FREQ=MONTHLY' for monthly)"}, "reminder_minutes": {"type": "integer", "description": "Optional reminder N minutes before the event (e.g. 30 for 30 minutes before)"}, "attendees": {"type": "array", "items": {"type": "string"}, "description": "Optional list of attendee email addresses"}, "calendar_name": {"type": "string", "description": "Optional calendar name to create the event in. Falls back to default calendar."}, "project": {"type": "string", "description": "Optional project name to associate this event with"}, "expected_weekday": {"type": "string", "description": "Optional weekday name (e.g. 'friday') the start_date should fall on. Pass this whenever the user names a weekday so the server can verify the date is correct. Rejects with a corrective error if the date falls on a different day."}, # Legacy combined fields kept for backward compatibility with saved # tool-call payloads in conversation history. New calls should use # start_date + start_time. Hidden from typical model output via the # description above; still accepted by the resolver as a fallback. "start": {"type": "string", "description": "[Deprecated] Combined start datetime — prefer start_date + start_time."}, "end": {"type": "string", "description": "[Deprecated] Combined end datetime — prefer end_date + end_time."}, }, required=["title"], ) async def create_event_tool(*, user_id, arguments, **_ctx): all_day = arguments.get("all_day", False) try: start_dt, start_was_date_only = await _resolve_event_start(user_id, arguments) except ValueError as exc: return {"success": False, "error": str(exc)} except TypeError as exc: return {"success": False, "error": f"Invalid start: {exc}"} if start_was_date_only: all_day = True user_tz = await get_user_tz(user_id) weekday_err = _validate_weekday(start_dt, user_tz, arguments.get("expected_weekday")) if weekday_err: return {"success": False, "error": weekday_err} try: end_dt = await _resolve_event_end(user_id, arguments) except ValueError as exc: return {"success": False, "error": str(exc)} except TypeError as exc: return {"success": False, "error": f"Invalid end: {exc}"} project_id = None project_name = arguments.get("project") if project_name: proj = await resolve_project(user_id, project_name) if proj: project_id = proj.id event = await events_create_event( user_id=user_id, title=arguments.get("title", "Untitled Event"), start_dt=start_dt, end_dt=end_dt, all_day=all_day, description=arguments.get("description") or "", location=arguments.get("location") or "", color=arguments.get("color") or "", recurrence=arguments.get("recurrence"), project_id=project_id, duration=arguments.get("duration"), reminder_minutes=arguments.get("reminder_minutes"), attendees=arguments.get("attendees"), calendar_name=arguments.get("calendar_name"), ) return {"success": True, "type": "event", "data": event.to_dict()} @tool( name="list_events", description="List calendar events in a date range. Use this when the user asks what events or meetings they have. Pass plain local dates (YYYY-MM-DD) — the server interprets them in the user's timezone and expands to a full local day.", parameters={ "date_from": {"type": "string", "description": "Start of range as a local date (YYYY-MM-DD) or local datetime. Interpreted in the user's timezone."}, "date_to": {"type": "string", "description": "End of range as a local date (YYYY-MM-DD) or local datetime. A bare date is expanded to 23:59:59 local."}, }, required=["date_from", "date_to"], read_only=True, briefing=True, ) async def list_events_tool(*, user_id, arguments, **_ctx): # Bare local dates are expanded to a full local day in the user's TZ: # date_from → 00:00 local, date_to → 23:59:59 local, both converted to # UTC before the DB query. Previously the tool description told the # model to pass UTC ranges, which missed events for non-UTC users. try: date_from, _ = await _parse_datetime_in_user_tz( user_id, arguments["date_from"] ) date_to_str = arguments["date_to"] if "T" not in date_to_str and " " not in date_to_str: date_to_str = f"{date_to_str}T23:59:59" date_to, _ = await _parse_datetime_in_user_tz(user_id, date_to_str) except (ValueError, TypeError, KeyError) as exc: return {"success": False, "error": f"Invalid date range: {exc}"} events = await events_list_events(user_id=user_id, date_from=date_from, date_to=date_to) return { "success": True, "type": "events", "data": {"count": len(events), "events": events}, } @tool( name="search_events", description="Search calendar events by keyword. Use this when the user asks to find a specific event or meeting.", parameters={ "query": {"type": "string", "description": "Search keyword to match against event titles, locations, and descriptions"}, "include_past": {"type": "boolean", "description": "Set to true to include past events in results (default: future events only)"}, }, required=["query"], read_only=True, briefing=True, ) async def search_events_tool(*, user_id, arguments, **_ctx): events = await events_search_events( user_id=user_id, query=arguments.get("query", ""), include_past=arguments.get("include_past", False), ) return { "success": True, "type": "events", "data": { "query": arguments.get("query", ""), "count": len(events), "events": [e.to_dict() for e in events], }, } @tool( name="update_event", description=( "Update an existing calendar event. Use this when the user asks to " "change, move, reschedule, or modify an event. Pass `start_date` " "(YYYY-MM-DD) and `start_time` (HH:MM) as separate fields in the " "user's local time when rescheduling — never combine them, never " "include a timezone suffix. " "When the user names a weekday ('move to Friday'), state the " "resolved calendar date in your reply BEFORE calling this tool, " "and pass `expected_weekday` so the server can verify the date " "falls on the day you intended.\n\n" "Identify the event with EITHER `query` (a title substring) OR " "`event_id` (when you already have an exact id from a prior tool " "result). If `query` matches multiple events, the tool returns " "an ambiguity error with a candidate list — pick one by passing " "its `event_id` on the next call, or refine the query so it " "matches a single event." ), parameters={ "query": {"type": "string", "description": "Search term to find the event to update (matches against title). Required unless event_id is set."}, "event_id": {"type": "integer", "description": "Exact event id, used to disambiguate when a prior call returned multiple candidates. Takes precedence over query."}, "title": {"type": "string", "description": "New title for the event"}, "start_date": {"type": "string", "description": "New start calendar date as YYYY-MM-DD in the user's local time. No timezone suffix."}, "start_time": {"type": "string", "description": "New start wall-clock time as HH:MM. No timezone suffix."}, "end_date": {"type": "string", "description": "New end calendar date as YYYY-MM-DD."}, "end_time": {"type": "string", "description": "New end wall-clock time as HH:MM."}, "all_day": {"type": "boolean", "description": "Whether the event is all-day"}, "description": {"type": "string", "description": "New event description"}, "location": {"type": "string", "description": "New event location"}, "color": {"type": "string", "description": "New hex color for the event (e.g. '#6366f1')"}, "recurrence": {"type": "string", "description": "New iCalendar RRULE"}, "reminder_minutes": {"type": "integer", "description": "Reminder N minutes before the event. Pass 0 to remove an existing reminder."}, "expected_weekday": {"type": "string", "description": "Optional weekday name (e.g. 'friday') the new start_date should fall on. Pass whenever the user names a weekday."}, # Legacy combined fields kept for backcompat — see create_event. "start": {"type": "string", "description": "[Deprecated] Combined start datetime — prefer start_date + start_time."}, "end": {"type": "string", "description": "[Deprecated] Combined end datetime — prefer end_date + end_time."}, }, required=[], ) async def update_event_tool(*, user_id, arguments, **_ctx): resolved = await _resolve_event_for_action( user_id=user_id, arguments=arguments, action="update", ) if isinstance(resolved, tuple): return resolved[1] # error dict from the resolver event_to_update = resolved fields: dict = {} for str_field in ("title", "description", "location", "color", "recurrence"): if arguments.get(str_field) is not None: fields[str_field] = arguments[str_field] if arguments.get("all_day") is not None: fields["all_day"] = arguments["all_day"] if "reminder_minutes" in arguments: rm = arguments["reminder_minutes"] fields["reminder_minutes"] = None if rm == 0 else rm # Resolve start: split fields preferred, legacy `start` as fallback. if arguments.get("start_date") or arguments.get("start"): try: start_dt, _ = await _resolve_event_start(user_id, arguments) except (ValueError, TypeError) as exc: return {"success": False, "error": f"Invalid start: {exc}"} user_tz = await get_user_tz(user_id) weekday_err = _validate_weekday(start_dt, user_tz, arguments.get("expected_weekday")) if weekday_err: return {"success": False, "error": weekday_err} fields["start_dt"] = start_dt if arguments.get("end_date") or arguments.get("end"): try: end_dt = await _resolve_event_end(user_id, arguments) except (ValueError, TypeError) as exc: return {"success": False, "error": f"Invalid end: {exc}"} if end_dt is not None: fields["end_dt"] = end_dt updated = await events_update_event(user_id=user_id, event_id=event_to_update.id, **fields) if updated is None: return {"success": False, "error": "Event not found or update failed."} return {"success": True, "type": "event_updated", "data": updated.to_dict()} @tool( name="delete_event", description=( "Delete a calendar event. Use this when the user asks to cancel, " "remove, or delete an event. Identify the event with EITHER " "`query` (a title substring) OR `event_id` (when you have an " "exact id). If `query` matches multiple events, the tool returns " "an ambiguity error with a candidate list — pick one by passing " "its `event_id` on the next call, or refine the query so it " "matches a single event. Deleting the wrong event is a costly " "user error; never guess between candidates." ), parameters={ "query": {"type": "string", "description": "Search term to find the event to delete (matches against title). Required unless event_id is set."}, "event_id": {"type": "integer", "description": "Exact event id, used to disambiguate when a prior call returned multiple candidates. Takes precedence over query."}, }, required=[], ) async def delete_event_tool(*, user_id, arguments, **_ctx): resolved = await _resolve_event_for_action( user_id=user_id, arguments=arguments, action="delete", ) if isinstance(resolved, tuple): return resolved[1] event_to_delete = resolved await events_delete_event(user_id=user_id, event_id=event_to_delete.id) return {"success": True, "type": "event_deleted", "data": {"id": event_to_delete.id, "title": event_to_delete.title}} @tool( name="list_calendars", description="List all available calendars. Use this when the user asks which calendars they have.", parameters={}, read_only=True, requires="caldav", ) async def list_calendars_tool(*, user_id, arguments, **_ctx): from fabledassistant.services.caldav import list_calendars calendars = await list_calendars(user_id=user_id) return { "success": True, "type": "calendars", "data": {"count": len(calendars), "calendars": calendars}, }