bvandeusen/FabledScribe
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 50 Wiki Activity

refactor(scribe): retire calendar/events + person/place/list entities (backend)
CI & Build / Python lint (push) Successful in 3s
Details
CI & Build / TypeScript typecheck (push) Successful in 11s
Details
CI & Build / integration (push) Successful in 30s
Details
CI & Build / Python tests (push) Failing after 31s
Details
CI & Build / Build & push image (push) Has been skipped
Details

Browse Source 
Narrow Scribe to a Claude-Code work system-of-record (milestone #194,
decision note #1759). Wholesale removal per rule #22 — backend + schema half.

Calendar/events + CalDAV: delete models/event, services/{events,caldav,
caldav_sync}, routes/events, mcp/tools/events; strip event branches from
backup (bump v3->v4), dashboard (upcoming_events), trash, recent, and the
mcp server read-only allowlist + instructions.

Typed entities (person/place/list): delete mcp/tools/entities; drop the
notes.metadata (entity_meta) column from model/service/routes and the
knowledge browse service. note_type STAYS — it also marks 'process' notes.

Scheduler: event_scheduler -> recurrence_scheduler, keeping only the
recurring-task spawn job (drops event reminders + CalDAV sync).

Schema: migration 0069 drops the events table + notes.metadata column +
orphan caldav settings rows (faithful downgrade recreates them).

KEEP: recurrence.py (task recurrence), notifications task reminders, graph
view, and every work surface. Frontend + plugin/docs true-up follow next.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01BPtbSzA4JLMAKgFZ8VTg7Q
This commit is contained in:
bvandeusen
2026-07-19 13:29:14 -04:00
co-authored by Claude Opus 4.8
parent f6629d4bcf
commit b49efdcb11
33 changed files with 194 additions and 3453 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/scribe/services/backup.py
+9 -80
View File
@@ -4,7 +4,6 @@ from datetime import date, datetime, timezone
from sqlalchemy import or_, select
from scribe.models import async_session
from scribe.models.event import Event
from scribe.models.milestone import Milestone
from scribe.models.note import Note
from scribe.models.note_draft import NoteDraft
@@ -25,9 +24,10 @@ from scribe.models.user import User
logger = logging.getLogger(__name__)
# Backup format version. v3 (2026-06) added rulebooks/topics/rules + their
# project subscription/suppression join tables, and events — all silently
# dropped by v2. Bump when the serialized schema changes.
BACKUP_VERSION = 3
# project subscription/suppression join tables. v4 (2026-07) dropped events
# when the calendar surface was retired — old v3 events are skipped on restore.
# Bump when the serialized schema changes.
BACKUP_VERSION = 4
# Tables intentionally NOT in the backup, surfaced in the payload so the gap is
# explicit rather than silent. ACL (groups/shares) is a coherent follow-up;
@@ -80,7 +80,6 @@ async def export_full_backup() -> dict:
rulebooks = (await session.execute(select(Rulebook))).scalars().all()
topics = (await session.execute(select(RulebookTopic))).scalars().all()
rules = (await session.execute(select(Rule))).scalars().all()
events = (await session.execute(select(Event))).scalars().all()
subscriptions = (await session.execute(
select(project_rulebook_subscriptions)
)).all()
@@ -245,27 +244,6 @@ async def export_full_backup() -> dict:
"rulebook_subscriptions": _subscription_rows(subscriptions),
"rule_suppressions": _rule_suppression_rows(rule_suppressions),
"topic_suppressions": _topic_suppression_rows(topic_suppressions),
"events": [
{
"id": e.id,
"user_id": e.user_id,
"project_id": e.project_id,
"uid": e.uid,
"caldav_uid": e.caldav_uid,
"title": e.title,
"start_dt": e.start_dt.isoformat() if e.start_dt else None,
"duration_minutes": e.duration_minutes,
"all_day": e.all_day,
"description": e.description,
"location": e.location,
"color": e.color,
"recurrence": e.recurrence,
"reminder_minutes": e.reminder_minutes,
"created_at": e.created_at.isoformat(),
"updated_at": e.updated_at.isoformat(),
}
for e in events
],
}
@@ -314,9 +292,6 @@ async def export_user_backup(user_id: int) -> dict:
rules = (await session.execute(
select(Rule).where(or_(*rule_filters))
)).scalars().all() if rule_filters else []
events = (await session.execute(
select(Event).where(Event.user_id == user_id)
)).scalars().all()
if project_ids:
subscriptions = (await session.execute(
select(project_rulebook_subscriptions).where(
@@ -480,27 +455,6 @@ async def export_user_backup(user_id: int) -> dict:
"rulebook_subscriptions": _subscription_rows(subscriptions),
"rule_suppressions": _rule_suppression_rows(rule_suppressions),
"topic_suppressions": _topic_suppression_rows(topic_suppressions),
"events": [
{
"id": e.id,
"user_id": e.user_id,
"project_id": e.project_id,
"uid": e.uid,
"caldav_uid": e.caldav_uid,
"title": e.title,
"start_dt": e.start_dt.isoformat() if e.start_dt else None,
"duration_minutes": e.duration_minutes,
"all_day": e.all_day,
"description": e.description,
"location": e.location,
"color": e.color,
"recurrence": e.recurrence,
"reminder_minutes": e.reminder_minutes,
"created_at": e.created_at.isoformat(),
"updated_at": e.updated_at.isoformat(),
}
for e in events
],
}
@@ -591,17 +545,17 @@ async def _restore_v1(data: dict) -> dict:
async def _restore_v2(data: dict) -> dict:
"""Restore v2/v3 backup with full FK re-mapping.
Conversations + push subscriptions in pre-pivot backups are silently
skipped — those subsystems were removed in the MCP-first pivot. v3-only
sections (rulebooks/topics/rules/join-tables/events) are guarded by
data.get so a v2 payload restores without them.
Conversations, push subscriptions, and (as of v4) events in older backups
are silently skipped — those subsystems were removed. v3+ sections
(rulebooks/topics/rules/join-tables) are guarded by data.get so a v2
payload restores without them.
"""
stats: dict[str, int] = {
"users": 0, "projects": 0, "milestones": 0, "notes": 0,
"task_logs": 0, "note_drafts": 0, "note_versions": 0,
"settings": 0, "rulebooks": 0, "rulebook_topics": 0, "rules": 0,
"rulebook_subscriptions": 0, "rule_suppressions": 0,
"topic_suppressions": 0, "events": 0,
"topic_suppressions": 0,
}
async with async_session() as session:
@@ -860,31 +814,6 @@ async def _restore_v2(data: dict) -> dict:
))
stats["topic_suppressions"] += 1
# 15. Events (v3)
for e_data in data.get("events", []):
mapped_uid = user_id_map.get(e_data.get("user_id", 0))
if mapped_uid is None:
continue
ev = Event(
user_id=mapped_uid,
project_id=project_id_map.get(e_data["project_id"]) if e_data.get("project_id") else None,
uid=e_data.get("uid", ""),
caldav_uid=e_data.get("caldav_uid", ""),
title=e_data.get("title", ""),
start_dt=_dt(e_data.get("start_dt")),
duration_minutes=e_data.get("duration_minutes"),
all_day=e_data.get("all_day", False),
description=e_data.get("description", ""),
location=e_data.get("location", ""),
color=e_data.get("color", ""),
recurrence=e_data.get("recurrence"),
reminder_minutes=e_data.get("reminder_minutes"),
created_at=_dt(e_data.get("created_at")),
updated_at=_dt(e_data.get("updated_at")),
)
session.add(ev)
stats["events"] += 1
await session.commit()
logger.info("Restored v2/v3 backup: %s", stats)
src/scribe/services/caldav.py
-770
View File
@@ -1,770 +0,0 @@
"""CalDAV calendar integration service."""
import asyncio
import logging
from datetime import date as date_type, datetime, timedelta
from zoneinfo import ZoneInfo
import caldav
import icalendar
from scribe.services.settings import get_all_settings
logger = logging.getLogger(__name__)
CALDAV_SETTING_KEYS = ["caldav_url", "caldav_username", "caldav_password", "caldav_calendar_name", "caldav_timezone"]
# Sentinel: distinguishes "leave the RRULE untouched" from "clear it" (None/"")
# in update_event, since None is a meaningful value for recurrence.
_RECURRENCE_UNSET = object()
async def get_caldav_config(user_id: int) -> dict[str, str]:
"""Return the user's CalDAV config from their settings."""
all_settings = await get_all_settings(user_id)
return {k: all_settings.get(k, "") for k in CALDAV_SETTING_KEYS}
async def is_caldav_configured(user_id: int) -> bool:
"""Check if the user has configured an external CalDAV server."""
config = await get_caldav_config(user_id)
return bool(config.get("caldav_url") and config.get("caldav_username") and config.get("caldav_password"))
def _get_calendar(client: caldav.DAVClient, calendar_name: str) -> caldav.Calendar:
"""Get a named calendar or the first available one (synchronous)."""
principal = client.principal()
calendars = principal.calendars()
if not calendars:
raise ValueError("No calendars found on the CalDAV server.")
if calendar_name:
for cal in calendars:
if cal.name == calendar_name:
return cal
names = [c.name for c in calendars]
raise ValueError(f"Calendar '{calendar_name}' not found. Available: {', '.join(names)}")
return calendars[0]
def _get_all_calendars(client: caldav.DAVClient) -> list[caldav.Calendar]:
"""Get all calendars for the user (synchronous)."""
principal = client.principal()
calendars = principal.calendars()
if not calendars:
raise ValueError("No calendars found on the CalDAV server.")
return calendars
def _make_client(config: dict[str, str]) -> caldav.DAVClient:
"""Create a CalDAV client from config dict."""
return caldav.DAVClient(
url=config["caldav_url"],
username=config.get("caldav_username") or None,
password=config.get("caldav_password") or None,
)
def _parse_vevent(component) -> dict | None:
"""Extract event data from a VEVENT component."""
if component.name != "VEVENT":
return None
title = str(component.get("SUMMARY", ""))
dtstart = component.get("DTSTART")
dtend = component.get("DTEND")
location = str(component.get("LOCATION", ""))
description = str(component.get("DESCRIPTION", ""))
uid = str(component.get("UID", ""))
start_str = dtstart.dt.isoformat() if dtstart else ""
end_str = dtend.dt.isoformat() if dtend else ""
result = {
"uid": uid,
"title": title,
"start": start_str,
"end": end_str,
"location": location,
"description": description,
}
# Extract recurrence rule
rrule = component.get("RRULE")
if rrule:
result["recurrence"] = rrule.to_ical().decode("utf-8")
# Extract alarms
alarms = []
for sub in component.subcomponents:
if sub.name == "VALARM":
trigger = sub.get("TRIGGER")
if trigger and trigger.dt:
minutes = abs(int(trigger.dt.total_seconds() // 60))
alarms.append({"minutes_before": minutes})
if alarms:
result["alarms"] = alarms
# Extract attendees
attendees = component.get("ATTENDEE")
if attendees:
if not isinstance(attendees, list):
attendees = [attendees]
result["attendees"] = [str(a).replace("mailto:", "") for a in attendees]
return result
def _parse_vtodo(component) -> dict | None:
"""Extract todo data from a VTODO component."""
if component.name != "VTODO":
return None
uid = str(component.get("UID", ""))
summary = str(component.get("SUMMARY", ""))
description = str(component.get("DESCRIPTION", ""))
status = str(component.get("STATUS", ""))
due = component.get("DUE")
due_str = due.dt.isoformat() if due else ""
priority = component.get("PRIORITY")
priority_val = int(priority) if priority else None
return {
"uid": uid,
"summary": summary,
"description": description,
"due": due_str,
"status": status,
"priority": priority_val,
}
def _apply_timezone(dt: datetime, timezone: str | None) -> datetime:
"""Apply a timezone to a naive datetime. Returns dt unchanged if already aware."""
if dt.tzinfo is not None:
return dt
if timezone:
return dt.replace(tzinfo=ZoneInfo(timezone))
return dt
def _build_valarm(minutes_before: int) -> icalendar.Alarm:
"""Create a DISPLAY alarm component triggered N minutes before the event."""
alarm = icalendar.Alarm()
alarm.add("action", "DISPLAY")
alarm.add("description", "Reminder")
alarm.add("trigger", timedelta(minutes=-minutes_before))
return alarm
def _add_attendees(event: icalendar.Event, attendees: list[str]) -> None:
"""Add mailto: attendees to an iCalendar event."""
for email in attendees:
attendee = icalendar.vCalAddress(f"mailto:{email}")
event.add("attendee", attendee)
def _check_config(config: dict[str, str]) -> None:
"""Raise if CalDAV is not configured."""
if not config.get("caldav_url"):
raise ValueError("CalDAV is not configured. Go to Settings → Calendar to enter your server URL.")
async def create_event(
user_id: int,
title: str,
start: str,
end: str | None = None,
duration: int | None = None,
description: str | None = None,
location: str | None = None,
all_day: bool = False,
recurrence: str | None = None,
timezone: str | None = None,
reminder_minutes: int | None = None,
attendees: list[str] | None = None,
calendar_name: str | None = None,
uid: str | None = None,
) -> dict:
"""Create a calendar event.
start/end are ISO date (YYYY-MM-DD) or datetime strings.
If all_day is True, DTSTART/DTEND use DATE values.
recurrence is an iCalendar RRULE string (e.g. "FREQ=YEARLY").
"""
config = await get_caldav_config(user_id)
_check_config(config)
tz = timezone or config.get("caldav_timezone") or None
cal = icalendar.Calendar()
cal.add("prodid", "-//Scribe//EN")
cal.add("version", "2.0")
event = icalendar.Event()
if uid:
# Remove auto-generated UID if the library added one, then inject ours
if "UID" in event:
del event["UID"]
event.add("uid", uid)
event.add("summary", title)
if all_day:
# All-day events use DATE values (no time component)
d_start = datetime.fromisoformat(start).date() if "T" in start else date_type.fromisoformat(start)
if end:
d_end = datetime.fromisoformat(end).date() if "T" in end else date_type.fromisoformat(end)
else:
d_end = d_start + timedelta(days=1)
event.add("dtstart", d_start)
event.add("dtend", d_end)
result_start = d_start.isoformat()
result_end = d_end.isoformat()
else:
dt_start = _apply_timezone(datetime.fromisoformat(start), tz)
event.add("dtstart", dt_start)
result_start = dt_start.isoformat()
if end:
dt_end = _apply_timezone(datetime.fromisoformat(end), tz)
elif duration:
dt_end = dt_start + timedelta(minutes=duration)
else:
dt_end = None
if dt_end is not None:
event.add("dtend", dt_end)
result_end = dt_end.isoformat()
else:
# Point event (no end, no duration): emit DTSTART only. Fabricating
# a 60-min DTEND here would round-trip back on the next pull as
# duration_minutes=60, silently lengthening a point event.
result_end = None
if description:
event.add("description", description)
if location:
event.add("location", location)
if recurrence:
# Parse RRULE string like "FREQ=YEARLY" into a vRecur dict
rrule_parts = {}
for part in recurrence.split(";"):
if "=" in part:
key, value = part.split("=", 1)
rrule_parts[key.strip().lower()] = value.strip()
event.add("rrule", rrule_parts)
if reminder_minutes is not None:
event.add_component(_build_valarm(reminder_minutes))
if attendees:
_add_attendees(event, attendees)
cal.add_component(event)
ical_str = cal.to_ical().decode("utf-8")
def _save():
client = _make_client(config)
cal_name = calendar_name or config.get("caldav_calendar_name", "")
calendar = _get_calendar(client, cal_name)
calendar.save_event(ical_str)
await asyncio.to_thread(_save)
result = {
"title": title,
"start": result_start,
"end": result_end,
"all_day": all_day,
}
if recurrence:
result["recurrence"] = recurrence
return result
async def list_events(user_id: int, date_from: str, date_to: str) -> list[dict]:
"""List calendar events in a date range. Dates are ISO datetime strings.
Searches all calendars unless caldav_calendar_name is configured.
"""
config = await get_caldav_config(user_id)
_check_config(config)
dt_from = datetime.fromisoformat(date_from)
dt_to = datetime.fromisoformat(date_to)
def _search():
client = _make_client(config)
cal_name = config.get("caldav_calendar_name", "")
if cal_name:
calendars = [_get_calendar(client, cal_name)]
else:
calendars = _get_all_calendars(client)
all_results = []
for calendar in calendars:
try:
all_results.extend(calendar.date_search(dt_from, dt_to))
except Exception:
logger.warning("Failed to search calendar '%s'", getattr(calendar, 'name', '?'))
return all_results
results = await asyncio.to_thread(_search)
events = []
for result in results:
cal = icalendar.Calendar.from_ical(result.data)
for component in cal.walk():
parsed = _parse_vevent(component)
if parsed:
events.append(parsed)
return events
async def search_events(user_id: int, query: str, days_ahead: int = 90) -> list[dict]:
"""Search events by keyword in the next N days."""
now = datetime.now()
date_from = now.isoformat()
date_to = (now + timedelta(days=days_ahead)).isoformat()
all_events = await list_events(user_id, date_from, date_to)
q = query.lower()
return [
e for e in all_events
if q in e["title"].lower() or q in e.get("location", "").lower() or q in e.get("description", "").lower()
]
async def update_event(
user_id: int,
query: str,
title: str | None = None,
start: str | None = None,
end: str | None = None,
description: str | None = None,
location: str | None = None,
timezone: str | None = None,
calendar_name: str | None = None,
recurrence: str | None | object = _RECURRENCE_UNSET,
) -> dict:
"""Update a calendar event matching the query.
``recurrence``: leave at the sentinel to keep the existing RRULE; pass an
RRULE string to set it, or None/"" to remove it. The push path passes the
local event's recurrence so RRULE edits propagate to the server.
"""
config = await get_caldav_config(user_id)
_check_config(config)
tz = timezone or config.get("caldav_timezone") or None
def _do_update():
client = _make_client(config)
cal_name = calendar_name or config.get("caldav_calendar_name", "")
now = datetime.now()
if cal_name:
calendars = [_get_calendar(client, cal_name)]
else:
calendars = _get_all_calendars(client)
results = []
for cal in calendars:
try:
results.extend(cal.date_search(now - timedelta(days=30), now + timedelta(days=365)))
except Exception:
logger.warning("Failed to search calendar '%s'", getattr(cal, 'name', '?'))
q = query.lower()
matches = []
for r in results:
cal_obj = icalendar.Calendar.from_ical(r.data)
for component in cal_obj.walk():
if component.name == "VEVENT":
event_title = str(component.get("SUMMARY", ""))
if q in event_title.lower():
matches.append((r, component))
if not matches:
raise ValueError(f"No event found matching '{query}'.")
if len(matches) > 3:
titles = [str(m[1].get("SUMMARY", "")) for m in matches]
raise ValueError(f"Too many matches ({len(matches)}) for '{query}'. Be more specific. Found: {', '.join(titles[:10])}")
event_obj, component = matches[0]
if title:
component["SUMMARY"] = title
if start:
dt_start = _apply_timezone(datetime.fromisoformat(start), tz)
del component["DTSTART"]
component.add("dtstart", dt_start)
if end:
dt_end = _apply_timezone(datetime.fromisoformat(end), tz)
if "DTEND" in component:
del component["DTEND"]
component.add("dtend", dt_end)
if description is not None:
if "DESCRIPTION" in component:
del component["DESCRIPTION"]
component.add("description", description)
if location is not None:
if "LOCATION" in component:
del component["LOCATION"]
component.add("location", location)
if recurrence is not _RECURRENCE_UNSET:
# Authoritatively sync the RRULE to the local event: drop the old
# rule, then re-add if a non-empty rule was provided (else clear it).
if "RRULE" in component:
del component["RRULE"]
if recurrence:
rrule_parts = {}
for part in str(recurrence).split(";"):
if "=" in part:
key, value = part.split("=", 1)
rrule_parts[key.strip().lower()] = value.strip()
component.add("rrule", rrule_parts)
# Rebuild ical data and save
cal_data = icalendar.Calendar()
cal_data.add("prodid", "-//Scribe//EN")
cal_data.add("version", "2.0")
cal_data.add_component(component)
event_obj.data = cal_data.to_ical().decode("utf-8")
event_obj.save()
return _parse_vevent(component)
return await asyncio.to_thread(_do_update)
async def delete_event(
user_id: int,
query: str,
calendar_name: str | None = None,
) -> dict:
"""Delete a calendar event matching the query."""
config = await get_caldav_config(user_id)
_check_config(config)
def _do_delete():
client = _make_client(config)
cal_name = calendar_name or config.get("caldav_calendar_name", "")
now = datetime.now()
if cal_name:
calendars = [_get_calendar(client, cal_name)]
else:
calendars = _get_all_calendars(client)
results = []
for cal in calendars:
try:
results.extend(cal.date_search(now - timedelta(days=30), now + timedelta(days=365)))
except Exception:
logger.warning("Failed to search calendar '%s'", getattr(cal, 'name', '?'))
q = query.lower()
matches = []
for r in results:
cal_obj = icalendar.Calendar.from_ical(r.data)
for component in cal_obj.walk():
if component.name == "VEVENT":
event_title = str(component.get("SUMMARY", ""))
if q in event_title.lower():
matches.append((r, component))
if not matches:
raise ValueError(f"No event found matching '{query}'.")
if len(matches) > 3:
titles = [str(m[1].get("SUMMARY", "")) for m in matches]
raise ValueError(f"Too many matches ({len(matches)}) for '{query}'. Be more specific. Found: {', '.join(titles[:10])}")
event_obj, component = matches[0]
parsed = _parse_vevent(component)
event_obj.delete()
return parsed
return await asyncio.to_thread(_do_delete)
async def list_calendars(user_id: int) -> list[dict]:
"""List all calendars for the user."""
config = await get_caldav_config(user_id)
_check_config(config)
def _list():
client = _make_client(config)
principal = client.principal()
calendars = principal.calendars()
return [{"name": c.name, "url": str(c.url)} for c in calendars]
return await asyncio.to_thread(_list)
async def create_todo(
user_id: int,
summary: str,
due: str | None = None,
description: str | None = None,
priority: int | None = None,
reminder_minutes: int | None = None,
timezone: str | None = None,
calendar_name: str | None = None,
) -> dict:
"""Create a CalDAV todo (VTODO)."""
config = await get_caldav_config(user_id)
_check_config(config)
tz = timezone or config.get("caldav_timezone") or None
def _create():
client = _make_client(config)
cal_name = calendar_name or config.get("caldav_calendar_name", "")
calendar = _get_calendar(client, cal_name)
kwargs = {"summary": summary}
if due:
dt_due = datetime.fromisoformat(due)
dt_due = _apply_timezone(dt_due, tz)
kwargs["due"] = dt_due
todo = calendar.save_todo(**kwargs)
# Modify component for extra fields
cal_obj = icalendar.Calendar.from_ical(todo.data)
modified = False
for component in cal_obj.walk():
if component.name == "VTODO":
if description:
component.add("description", description)
modified = True
if priority is not None:
component.add("priority", priority)
modified = True
if reminder_minutes is not None:
component.add_component(_build_valarm(reminder_minutes))
modified = True
if modified:
todo.data = cal_obj.to_ical().decode("utf-8")
todo.save()
return _parse_vtodo(component)
return {"summary": summary}
return await asyncio.to_thread(_create)
async def list_todos(
user_id: int,
include_completed: bool = False,
calendar_name: str | None = None,
) -> list[dict]:
"""List CalDAV todos."""
config = await get_caldav_config(user_id)
_check_config(config)
def _list():
client = _make_client(config)
cal_name = calendar_name or config.get("caldav_calendar_name", "")
calendar = _get_calendar(client, cal_name)
todos = calendar.todos(include_completed=include_completed)
results = []
for t in todos:
cal_obj = icalendar.Calendar.from_ical(t.data)
for component in cal_obj.walk():
parsed = _parse_vtodo(component)
if parsed:
results.append(parsed)
return results
return await asyncio.to_thread(_list)
async def search_todos(
user_id: int,
query: str,
include_completed: bool = False,
calendar_name: str | None = None,
) -> list[dict]:
"""Search CalDAV todos by keyword in summary or description."""
todos = await list_todos(user_id, include_completed=include_completed, calendar_name=calendar_name)
q = query.lower()
return [
t for t in todos
if q in t.get("summary", "").lower() or q in (t.get("description") or "").lower()
]
async def complete_todo(
user_id: int,
query: str,
calendar_name: str | None = None,
) -> dict:
"""Complete a CalDAV todo matching the query."""
config = await get_caldav_config(user_id)
_check_config(config)
def _complete():
client = _make_client(config)
cal_name = calendar_name or config.get("caldav_calendar_name", "")
calendar = _get_calendar(client, cal_name)
todos = calendar.todos(include_completed=False)
q = query.lower()
matches = []
for t in todos:
cal_obj = icalendar.Calendar.from_ical(t.data)
for component in cal_obj.walk():
if component.name == "VTODO":
s = str(component.get("SUMMARY", ""))
if q in s.lower():
matches.append((t, component))
if not matches:
raise ValueError(f"No todo found matching '{query}'.")
if len(matches) > 3:
titles = [str(m[1].get("SUMMARY", "")) for m in matches]
raise ValueError(f"Too many matches ({len(matches)}) for '{query}'. Be more specific. Found: {', '.join(titles[:10])}")
todo_obj, component = matches[0]
todo_obj.complete()
# Re-parse after completing
cal_obj = icalendar.Calendar.from_ical(todo_obj.data)
for comp in cal_obj.walk():
parsed = _parse_vtodo(comp)
if parsed:
return parsed
return {"summary": str(component.get("SUMMARY", "")), "status": "COMPLETED"}
return await asyncio.to_thread(_complete)
async def update_todo(
user_id: int,
query: str,
summary: str | None = None,
due: str | None = None,
description: str | None = None,
priority: int | None = None,
timezone: str | None = None,
calendar_name: str | None = None,
) -> dict:
"""Update a CalDAV todo matching the query."""
config = await get_caldav_config(user_id)
_check_config(config)
tz = timezone or config.get("caldav_timezone") or None
def _do_update():
client = _make_client(config)
cal_name = calendar_name or config.get("caldav_calendar_name", "")
calendar = _get_calendar(client, cal_name)
todos = calendar.todos(include_completed=True)
q = query.lower()
matches = []
for t in todos:
cal_obj = icalendar.Calendar.from_ical(t.data)
for component in cal_obj.walk():
if component.name == "VTODO":
s = str(component.get("SUMMARY", ""))
if q in s.lower():
matches.append((t, component))
if not matches:
raise ValueError(f"No todo found matching '{query}'.")
if len(matches) > 3:
titles = [str(m[1].get("SUMMARY", "")) for m in matches]
raise ValueError(
f"Too many matches ({len(matches)}) for '{query}'. "
f"Be more specific. Found: {', '.join(titles[:10])}"
)
todo_obj, component = matches[0]
if summary:
component["SUMMARY"] = summary
if description is not None:
if "DESCRIPTION" in component:
del component["DESCRIPTION"]
component.add("description", description)
if priority is not None:
if "PRIORITY" in component:
del component["PRIORITY"]
component.add("priority", priority)
if due:
if "DUE" in component:
del component["DUE"]
try:
dt = datetime.fromisoformat(due)
dt = _apply_timezone(dt, tz)
component.add("due", dt)
except ValueError:
component.add("due", date_type.fromisoformat(due))
# Rebuild ical data and save
cal_data = icalendar.Calendar()
cal_data.add("prodid", "-//Scribe//EN")
cal_data.add("version", "2.0")
cal_data.add_component(component)
todo_obj.data = cal_data.to_ical().decode("utf-8")
todo_obj.save()
return _parse_vtodo(component)
return await asyncio.to_thread(_do_update)
async def delete_todo(
user_id: int,
query: str,
calendar_name: str | None = None,
) -> dict:
"""Delete a CalDAV todo matching the query."""
config = await get_caldav_config(user_id)
_check_config(config)
def _delete():
client = _make_client(config)
cal_name = calendar_name or config.get("caldav_calendar_name", "")
calendar = _get_calendar(client, cal_name)
todos = calendar.todos(include_completed=True)
q = query.lower()
matches = []
for t in todos:
cal_obj = icalendar.Calendar.from_ical(t.data)
for component in cal_obj.walk():
if component.name == "VTODO":
s = str(component.get("SUMMARY", ""))
if q in s.lower():
matches.append((t, component))
if not matches:
raise ValueError(f"No todo found matching '{query}'.")
if len(matches) > 3:
titles = [str(m[1].get("SUMMARY", "")) for m in matches]
raise ValueError(f"Too many matches ({len(matches)}) for '{query}'. Be more specific. Found: {', '.join(titles[:10])}")
todo_obj, component = matches[0]
parsed = _parse_vtodo(component)
todo_obj.delete()
return parsed
return await asyncio.to_thread(_delete)
async def test_connection(user_id: int) -> dict:
"""Test the CalDAV connection and return status."""
config = await get_caldav_config(user_id)
if not config.get("caldav_url"):
return {"success": False, "error": "CalDAV is not configured."}
def _test():
client = _make_client(config)
principal = client.principal()
calendars = principal.calendars()
return [c.name for c in calendars]
try:
calendar_names = await asyncio.to_thread(_test)
return {
"success": True,
"calendars": calendar_names,
"message": f"Connected successfully. Found {len(calendar_names)} calendar(s).",
}
except Exception as e:
error_msg = str(e)
if "401" in error_msg or "403" in error_msg or "Unauthorized" in error_msg:
error_msg = "Authentication failed. Check your username and password."
elif "404" in error_msg or "Not Found" in error_msg:
error_msg = "CalDAV endpoint not found. Check your URL."
elif "Connection" in error_msg or "resolve" in error_msg:
error_msg = f"Connection failed: {error_msg}"
return {"success": False, "error": error_msg}
src/scribe/services/caldav_sync.py
-247
View File
@@ -1,247 +0,0 @@
"""CalDAV pull sync — imports remote events into the internal event store.
Runs as a scheduled job (hourly) and is also callable via the API.
Only syncs events in a rolling 30-day-past / 180-day-future window.
"""
from __future__ import annotations
import asyncio
import logging
import uuid
from datetime import datetime, timedelta, timezone
from typing import Any
from sqlalchemy import select, update
from scribe.models import async_session
from scribe.models.event import Event
logger = logging.getLogger(__name__)
_SYNC_PAST_DAYS = 30
_SYNC_FUTURE_DAYS = 180
# Wall-clock cap on the blocking CalDAV fetch so a hung/slow server can't
# wedge the hourly sweep indefinitely.
_SYNC_TIMEOUT_SECONDS = 120
def _parse_dt(val: Any) -> datetime | None:
"""Convert a date or datetime from an iCal component to a UTC-aware datetime."""
if val is None:
return None
import datetime as _dt_mod
if isinstance(val, _dt_mod.datetime):
if val.tzinfo is None:
return val.replace(tzinfo=timezone.utc)
return val.astimezone(timezone.utc)
if isinstance(val, _dt_mod.date):
# All-day date: treat as midnight UTC
return datetime(val.year, val.month, val.day, tzinfo=timezone.utc)
return None
def _sync_one_user(config: dict[str, str], user_id: int) -> list[dict]:
"""Synchronous CalDAV fetch — runs in a thread executor."""
import caldav # noqa: PLC0415
now = datetime.now(timezone.utc)
range_start = now - timedelta(days=_SYNC_PAST_DAYS)
range_end = now + timedelta(days=_SYNC_FUTURE_DAYS)
client = caldav.DAVClient(
url=config["caldav_url"],
username=config.get("caldav_username") or None,
password=config.get("caldav_password") or None,
)
principal = client.principal()
calendars = principal.calendars()
if not calendars:
return []
cal_name = config.get("caldav_calendar_name", "")
if cal_name:
calendars = [c for c in calendars if c.name == cal_name] or calendars
events: list[dict] = []
for calendar in calendars:
try:
results = calendar.date_search(start=range_start, end=range_end, expand=False)
except Exception:
logger.warning("CalDAV date_search failed for calendar %s", getattr(calendar, "name", "?"), exc_info=True)
continue
for vevent_obj in results:
try:
ical = vevent_obj.icalendar_instance
for component in ical.walk():
if component.name != "VEVENT":
continue
dtstart = component.get("DTSTART")
dtend = component.get("DTEND")
uid = str(component.get("UID", ""))
if not uid:
continue
start_dt = _parse_dt(dtstart.dt if dtstart else None)
end_dt = _parse_dt(dtend.dt if dtend else None)
if start_dt is None:
continue
import datetime as _dt_mod
all_day = dtstart and isinstance(dtstart.dt, _dt_mod.date) and not isinstance(dtstart.dt, _dt_mod.datetime)
rrule = component.get("RRULE")
recurrence = rrule.to_ical().decode("utf-8") if rrule else None
events.append({
"caldav_uid": uid,
"title": str(component.get("SUMMARY", "")),
"start_dt": start_dt,
"end_dt": end_dt,
"all_day": bool(all_day),
"description": str(component.get("DESCRIPTION", "")),
"location": str(component.get("LOCATION", "")),
"recurrence": recurrence,
})
except Exception:
logger.debug("Failed to parse CalDAV event", exc_info=True)
return events
async def sync_user_events(user_id: int) -> dict:
"""Pull CalDAV events for one user and upsert into the DB.
Returns a summary dict: {created, updated, unchanged}.
"""
from scribe.services.caldav import get_caldav_config, is_caldav_configured # noqa: PLC0415
if not await is_caldav_configured(user_id):
return {"skipped": True, "reason": "CalDAV not configured"}
config = await get_caldav_config(user_id)
started = datetime.now(timezone.utc)
range_start = started - timedelta(days=_SYNC_PAST_DAYS)
range_end = started + timedelta(days=_SYNC_FUTURE_DAYS)
loop = asyncio.get_running_loop()
try:
remote_events: list[dict] = await asyncio.wait_for(
loop.run_in_executor(None, _sync_one_user, config, user_id),
timeout=_SYNC_TIMEOUT_SECONDS,
)
except asyncio.TimeoutError:
logger.warning("CalDAV pull sync timed out for user %d after %ds", user_id, _SYNC_TIMEOUT_SECONDS)
return {"error": "CalDAV fetch timed out"}
except Exception:
logger.warning("CalDAV pull sync failed for user %d", user_id, exc_info=True)
return {"error": "CalDAV fetch failed"}
created = updated = unchanged = skipped = deleted = 0
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,
Event.caldav_uid == caldav_uid,
)
)
existing = result.scalar_one_or_none()
if existing is not None and existing.deleted_at is not None:
# The user trashed this event locally. Don't resurrect it by
# updating, and don't create a duplicate live copy — leave it
# in the trash. (Propagating the delete to the remote server is
# tracked separately.)
skipped += 1
continue
if existing is None:
# Create new event
new_ev = Event(
user_id=user_id,
uid=str(uuid.uuid4()),
caldav_uid=caldav_uid,
title=ev["title"],
start_dt=ev_start,
duration_minutes=ev_duration,
all_day=ev["all_day"],
description=ev["description"],
location=ev["location"],
recurrence=ev["recurrence"],
)
session.add(new_ev)
created += 1
else:
# Update if anything changed
changed = False
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
if changed:
updated += 1
else:
unchanged += 1
# Reconcile deletions: a previously-synced event (has a caldav_uid)
# that no longer appears remotely within the synced window is
# soft-deleted, so a delete on the remote propagates locally instead
# of orphaning forever. Guarded on a non-empty fetch so a spurious
# empty result can't wipe every local copy.
if remote_events:
remote_uids = {e["caldav_uid"] for e in remote_events}
orphan_batch = str(uuid.uuid4())
orphan_res = await session.execute(
update(Event)
.where(
Event.user_id == user_id,
Event.caldav_uid.isnot(None),
Event.caldav_uid.notin_(remote_uids),
Event.deleted_at.is_(None),
Event.start_dt >= range_start,
Event.start_dt <= range_end,
)
.values(deleted_at=datetime.now(timezone.utc), deleted_batch_id=orphan_batch)
)
deleted = orphan_res.rowcount or 0
await session.commit()
elapsed = (datetime.now(timezone.utc) - started).total_seconds()
logger.info(
"CalDAV sync user %d: %d created, %d updated, %d unchanged, %d skipped (trashed), "
"%d deleted (orphaned) in %.1fs",
user_id, created, updated, unchanged, skipped, deleted, elapsed,
)
return {"created": created, "updated": updated, "unchanged": unchanged,
"skipped": skipped, "deleted": deleted}
async def sync_all_users() -> None:
"""Pull CalDAV events for all users with CalDAV configured."""
from sqlalchemy import select as sa_select # noqa: PLC0415
from scribe.models.user import User # noqa: PLC0415
async with async_session() as session:
result = await session.execute(sa_select(User.id))
user_ids = [row[0] for row in result.all()]
for user_id in user_ids:
try:
await sync_user_events(user_id)
except Exception:
logger.warning("CalDAV sync failed for user %d", user_id, exc_info=True)
src/scribe/services/dashboard.py
+2 -15
View File
@@ -1,7 +1,7 @@
"""Dashboard aggregation — assembles the /dashboard landing payload.
One call: most-recently-active projects (each -> active milestones -> open
tasks), recently-completed tasks, upcoming events, week stats. Owner-scoped,
tasks), recently-completed tasks, week stats. Owner-scoped,
trashed rows excluded. Each section is independent — a failure returns its
empty value rather than blanking the page.
"""
@@ -23,7 +23,7 @@ N_PROJECTS = 3 # most-recently-active projects shown
TASKS_PER_GROUP = 5 # open-task cap per milestone / no-milestone group
RECENT_DONE_LIMIT = 8 # recently-completed tasks shown
OPEN_ISSUES_LIMIT = 10 # open issues shown on the dashboard
WINDOW_DAYS = 7 # look-back (done) / look-ahead (events) window
WINDOW_DAYS = 7 # look-back window (done items, week stats)
_OPEN = ["todo", "in_progress"]
@@ -84,7 +84,6 @@ async def build_dashboard(user_id: int) -> dict:
return {
"active_projects": await _safe(_active_projects(user_id), []),
"recently_completed": await _safe(_recently_completed(user_id), []),
"upcoming_events": await _safe(_upcoming_events(user_id), []),
"open_issues": await _safe(_open_issues(user_id), []),
"week_stats": await _safe(_week_stats(user_id), {}),
}
@@ -177,18 +176,6 @@ async def _recently_completed(user_id: int) -> list[dict]:
"completed_at": n.completed_at.isoformat()} for n, ptitle in rows]
async def _upcoming_events(user_id: int) -> list[dict]:
from scribe.services import events as events_svc
now = datetime.now(timezone.utc)
rows = await events_svc.list_events(user_id, now, now + timedelta(days=WINDOW_DAYS))
out = []
for e in rows:
d = e if isinstance(e, dict) else e.to_dict()
out.append({"id": d["id"], "title": d["title"],
"start_dt": d.get("start_dt"), "all_day": d.get("all_day", False)})
return out
async def _week_stats(user_id: int) -> dict:
cutoff = datetime.now(timezone.utc) - timedelta(days=WINDOW_DAYS)
async with async_session() as session:
src/scribe/services/event_scheduler.py
-198
View File
@@ -1,198 +0,0 @@
"""Scheduler jobs for background maintenance tasks.
- Reminder notifications: checks every 5 minutes for due event reminders and
delivers them to the in-app notification feed.
- CalDAV pull sync: runs every hour for all users with CalDAV configured.
- Recurring-task spawn: every 15 minutes, creates the next occurrence of any
recurring task whose spawn time has arrived.
Uses the BackgroundScheduler pattern shared with the other *_scheduler modules.
"""
from __future__ import annotations
import asyncio
import logging
from datetime import datetime, timedelta, timezone
from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.triggers.interval import IntervalTrigger
from dateutil.rrule import rrulestr
from sqlalchemy import and_, or_, select
from scribe.models import async_session
from scribe.models.event import Event
logger = logging.getLogger(__name__)
_scheduler: BackgroundScheduler | None = None
_loop: asyncio.AbstractEventLoop | None = None
# ---------------------------------------------------------------------------
# Reminder job
# ---------------------------------------------------------------------------
async def _fire_reminders() -> None:
"""Fire in-app reminders for events whose reminder time has arrived.
One-shot events fire once (gated on reminder_sent_at IS NULL). Recurring
events fire once PER OCCURRENCE: reminder_sent_at stores the start of the
occurrence we last reminded about, so each new occurrence re-arms the
reminder instead of the whole series firing only once.
"""
now = datetime.now(timezone.utc)
window_end = now + timedelta(minutes=5)
async with async_session() as session:
result = await session.execute(
select(Event).where(
Event.reminder_minutes.isnot(None),
Event.deleted_at.is_(None),
or_(
# Recurring events are evaluated every sweep against their
# next occurrence (the base start_dt is long past).
Event.recurrence.isnot(None),
# One-shot events: classic gate.
and_(Event.reminder_sent_at.is_(None), Event.start_dt > now),
),
)
)
candidates = list(result.scalars().all())
# (event_id, occurrence_start) — occurrence_start is also the dedup marker
# written to reminder_sent_at, so a given occurrence reminds exactly once.
to_notify: list[tuple[int, datetime]] = []
for event in candidates:
if event.recurrence:
try:
rule = rrulestr(event.recurrence, dtstart=event.start_dt, ignoretz=False)
occ = rule.after(now, inc=True)
except Exception:
logger.warning("Failed to expand RRULE for event %d reminder", event.id, exc_info=True)
continue
if occ is None:
continue
reminder_dt = occ - timedelta(minutes=event.reminder_minutes)
if reminder_dt <= window_end and event.reminder_sent_at != occ:
to_notify.append((event.id, occ))
else:
reminder_dt = event.start_dt - timedelta(minutes=event.reminder_minutes)
if reminder_dt <= window_end:
to_notify.append((event.id, event.start_dt))
if not to_notify:
return
# Deliver via the in-app notification feed (push was removed in Phase 8).
from scribe.services.notifications import create_in_app_notification
async with async_session() as session:
for event_id, occurrence_start in to_notify:
ev = (await session.execute(
select(Event).where(Event.id == event_id)
)).scalar_one_or_none()
# Skip if this exact occurrence was already reminded (covers a
# concurrent sweep and the one-shot already-sent case).
if ev is None or ev.reminder_sent_at == occurrence_start:
continue
await create_in_app_notification(ev.user_id, "event_reminder", {
"event_id": ev.id,
"title": ev.title,
"start_dt": occurrence_start.isoformat(),
"url": "/calendar",
})
# Stamp the occurrence marker only after the notification is
# created, so a delivery failure leaves it eligible to retry.
ev.reminder_sent_at = occurrence_start
await session.commit()
def _run_reminders(loop: asyncio.AbstractEventLoop) -> None:
asyncio.run_coroutine_threadsafe(_fire_reminders(), loop)
# ---------------------------------------------------------------------------
# CalDAV pull sync job
# ---------------------------------------------------------------------------
async def _run_caldav_sync() -> None:
from scribe.services.caldav_sync import sync_all_users # noqa: PLC0415
try:
await sync_all_users()
except Exception:
logger.warning("CalDAV pull sync job failed", exc_info=True)
def _run_caldav_sync_threadsafe(loop: asyncio.AbstractEventLoop) -> None:
asyncio.run_coroutine_threadsafe(_run_caldav_sync(), loop)
# ---------------------------------------------------------------------------
# Recurring-task spawn job
# ---------------------------------------------------------------------------
async def _run_recurrence_spawn() -> None:
from scribe.services.recurrence import spawn_recurring_tasks # noqa: PLC0415
try:
await spawn_recurring_tasks()
except Exception:
logger.warning("Recurring-task spawn job failed", exc_info=True)
def _run_recurrence_spawn_threadsafe(loop: asyncio.AbstractEventLoop) -> None:
asyncio.run_coroutine_threadsafe(_run_recurrence_spawn(), loop)
# ---------------------------------------------------------------------------
# Lifecycle
# ---------------------------------------------------------------------------
def start_event_scheduler(loop: asyncio.AbstractEventLoop) -> None:
global _scheduler, _loop
if _scheduler is not None:
return
_loop = loop
_scheduler = BackgroundScheduler()
# Check reminders every 5 minutes
_scheduler.add_job(
_run_reminders,
trigger=IntervalTrigger(minutes=5),
args=[loop],
id="event_reminders",
replace_existing=True,
)
# CalDAV pull sync every hour
_scheduler.add_job(
_run_caldav_sync_threadsafe,
trigger=IntervalTrigger(hours=1),
args=[loop],
id="caldav_pull_sync",
replace_existing=True,
)
# Spawn the next occurrence of due recurring tasks every 15 minutes.
# Without this job, recurrence_next_spawn_at is armed on completion but
# never drained, so recurring tasks never recur.
_scheduler.add_job(
_run_recurrence_spawn_threadsafe,
trigger=IntervalTrigger(minutes=15),
args=[loop],
id="recurrence_spawn",
replace_existing=True,
)
_scheduler.start()
logger.info(
"Event scheduler started (reminders every 5m, CalDAV sync every 1h, "
"recurring-task spawn every 15m)"
)
def stop_event_scheduler() -> None:
global _scheduler
if _scheduler is not None:
_scheduler.shutdown(wait=False)
_scheduler = None
logger.info("Event scheduler stopped")
src/scribe/services/events.py
-477
View File
@@ -1,477 +0,0 @@
"""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
import logging
import uuid
from datetime import datetime, timedelta, timezone
from dateutil.rrule import rrulestr
from sqlalchemy import or_, select
from scribe.models import async_session
from scribe.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 _localize_naive(user_id: int, dt: datetime | None) -> datetime | None:
"""Anchor a naive datetime in the user's timezone; pass tz-aware through.
Naive datetimes are the user's local wall-clock time (the MCP create/update
tools combine date+time without a zone). Attaching the user's tzinfo lets
asyncpg store the correct UTC instant, matching the REST/UI path.
"""
if dt is not None and dt.tzinfo is None:
from scribe.services.tz import get_user_tz # noqa: PLC0415
return dt.replace(tzinfo=await get_user_tz(user_id))
return dt
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 = "",
color: str = "",
recurrence: str | None = None,
project_id: int | None = None,
reminder_minutes: int | None = None,
# ``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.
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
# Canonical localization point: a naive datetime (e.g. from the MCP tool's
# date+time split) is the user's wall-clock time, so anchor it in their
# timezone before storage. tz-aware inputs (REST, CalDAV pass-through) are
# left untouched. Without this, MCP-created events landed at the same
# wall-clock numerals in UTC and drifted from UI-created ones by the offset.
start_dt = await _localize_naive(user_id, start_dt)
end_dt = await _localize_naive(user_id, end_dt)
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(
user_id=user_id,
uid=uid,
title=title,
start_dt=start_dt,
duration_minutes=duration_minutes,
all_day=all_day,
description=description,
location=location,
color=color,
recurrence=recurrence,
project_id=project_id,
reminder_minutes=reminder_minutes,
)
session.add(event)
await session.commit()
await session.refresh(event)
extra_fields = {
"duration": duration_minutes,
"reminder_minutes": reminder_minutes,
"attendees": attendees,
"calendar_name": calendar_name,
}
asyncio.create_task(_push_create(event, user_id, extra_fields))
return event
async def get_event(user_id: int, event_id: int) -> Event | None:
"""Return event owned by user_id, or None."""
async with async_session() as session:
result = await session.execute(
select(Event).where(
Event.id == event_id, Event.user_id == user_id, Event.deleted_at.is_(None)
)
)
return result.scalar_one_or_none()
async def list_events(
user_id: int,
date_from: datetime,
date_to: datetime,
) -> list[dict]:
"""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()``).
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:
result = await session.execute(
select(Event)
.where(
Event.user_id == user_id,
Event.deleted_at.is_(None),
or_(
Event.recurrence.isnot(None),
Event.start_dt <= date_to,
),
)
.order_by(Event.start_dt)
)
events = list(result.scalars().all())
items: list[dict] = []
for event in events:
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
# 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
async def search_events(
user_id: int,
query: str,
days_ahead: int = 90,
include_past: bool = False,
) -> list[Event]:
"""Search events by keyword in title, description, or location."""
now = datetime.now(timezone.utc)
q = f"%{query}%"
async with async_session() as session:
where = [
Event.user_id == user_id,
Event.deleted_at.is_(None),
or_(
Event.title.ilike(q),
Event.description.ilike(q),
Event.location.ilike(q),
),
]
if not include_past:
date_to = now + timedelta(days=days_ahead)
where.extend([Event.start_dt >= now, Event.start_dt <= date_to])
result = await session.execute(
select(Event).where(*where).order_by(Event.start_dt)
)
return result.scalars().all()
async def update_event(user_id: int, event_id: int, **fields) -> Event | None:
"""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,
Event.deleted_at.is_(None),
)
)
event = result.scalar_one_or_none()
if event is None:
return None
old_title = event.title # capture before mutation for CalDAV lookup
# Localize a naive start_dt patch to the user's timezone (same canonical
# rule as create_event) before it's used or persisted.
if fields.get("start_dt") is not None:
fields["start_dt"] = await _localize_naive(user_id, fields["start_dt"])
# 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)
# Re-arm the reminder when the timing changes, so an event moved to a
# new (future) time — or given a new lead time — fires again instead of
# being permanently suppressed by a stale reminder_sent_at.
if "start_dt" in fields or "reminder_minutes" in fields:
event.reminder_sent_at = None
await session.commit()
await session.refresh(event)
asyncio.create_task(_push_update(event, user_id, old_title=old_title))
return event
async def delete_event(user_id: int, event_id: int) -> None:
"""Delete event. Fires CalDAV delete push if caldav_uid is set."""
async with async_session() as session:
result = await session.execute(
select(Event).where(Event.id == event_id, Event.user_id == user_id)
)
event = result.scalar_one_or_none()
if event is None:
return
caldav_uid = event.caldav_uid
event_title = event.title # needed to find the event on CalDAV by title
await session.delete(event)
await session.commit()
if caldav_uid:
asyncio.create_task(_push_delete(caldav_uid, event_title, user_id))
async def find_events_by_query(user_id: int, query: str) -> list[Event]:
"""ILIKE search on title — used by AI update/delete tools.
Returns upcoming events first (start_dt >= now), falling back to
past events so the AI operates on the most relevant match.
"""
q = f"%{query}%"
now = datetime.now(timezone.utc)
async with async_session() as session:
# Prefer events at or after now; fall back to past events
upcoming = (await session.execute(
select(Event).where(
Event.user_id == user_id,
Event.deleted_at.is_(None),
Event.title.ilike(q),
Event.start_dt >= now,
).order_by(Event.start_dt)
)).scalars().all()
if upcoming:
return list(upcoming)
past = (await session.execute(
select(Event).where(
Event.user_id == user_id,
Event.deleted_at.is_(None),
Event.title.ilike(q),
Event.start_dt < now,
).order_by(Event.start_dt.desc())
)).scalars().all()
return list(past)
# ---------------------------------------------------------------------------
# CalDAV push helpers (fire-and-forget)
# ---------------------------------------------------------------------------
async def _push_create(event: Event, user_id: int, extra: dict) -> None:
try:
from scribe.services.caldav import (
create_event as caldav_create,
is_caldav_configured,
)
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=derived_end.isoformat() if derived_end else None,
description=event.description or None,
location=event.location or None,
all_day=event.all_day,
recurrence=event.recurrence,
uid=event.uid,
duration=extra.get("duration"),
reminder_minutes=extra.get("reminder_minutes"),
attendees=extra.get("attendees"),
calendar_name=extra.get("calendar_name"),
)
# Mark as synced
async with async_session() as session:
result = await session.execute(
select(Event).where(Event.id == event.id)
)
ev = result.scalar_one_or_none()
if ev:
ev.caldav_uid = event.uid
await session.commit()
except Exception:
logger.warning("CalDAV push (create) failed for event %d", event.id, exc_info=True)
async def _push_update(event: Event, user_id: int, old_title: str = "") -> None:
"""Push an update to CalDAV. Uses old_title to locate the event by its pre-rename SUMMARY."""
if not event.caldav_uid:
return
try:
from scribe.services.caldav import (
update_event as caldav_update,
is_caldav_configured,
)
if not await is_caldav_configured(user_id):
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=derived_end.isoformat() if derived_end else None,
description=event.description or None,
location=event.location or None,
# Propagate the (possibly cleared) RRULE so a local recurrence edit
# isn't overwritten by the stale remote rule on the next pull.
recurrence=event.recurrence,
)
except Exception:
logger.warning("CalDAV push (update) failed for event %d", event.id, exc_info=True)
async def _push_delete(caldav_uid: str, event_title: str, user_id: int) -> None:
"""Push a delete to CalDAV. Uses event_title to locate the event by SUMMARY."""
try:
from scribe.services.caldav import (
delete_event as caldav_delete,
is_caldav_configured,
)
if not await is_caldav_configured(user_id):
return
await caldav_delete(user_id=user_id, query=event_title)
except Exception:
logger.warning("CalDAV push (delete) failed for uid %s", caldav_uid, exc_info=True)
src/scribe/services/knowledge.py
+6 -75
View File
@@ -1,4 +1,4 @@
"""Knowledge service — unified query across notes, people, places, and lists."""
"""Knowledge service — unified query across notes, tasks, plans, and processes."""
import logging
from sqlalchemy import func, select
@@ -12,46 +12,16 @@ _SNIPPET_LEN = 200
def _note_to_item(note: Note) -> dict:
meta = note.entity_meta or {}
item: dict = {
"id": note.id,
"note_type": note.entity_type,
"note_type": note.note_type or "note",
"title": note.title,
"snippet": (note.body or "")[:_SNIPPET_LEN],
"tags": note.tags or [],
"project_id": note.project_id,
"metadata": meta,
"created_at": note.created_at.isoformat(),
"updated_at": note.updated_at.isoformat(),
}
# Type-specific convenience fields
if note.entity_type == "person":
item["relationship"] = meta.get("relationship", "")
item["email"] = meta.get("email", "")
item["phone"] = meta.get("phone", "")
item["birthday"] = meta.get("birthday", "")
item["organization"] = meta.get("organization", "")
item["address"] = meta.get("address", "")
elif note.entity_type == "place":
item["address"] = meta.get("address", "")
item["phone"] = meta.get("phone", "")
item["hours"] = meta.get("hours", "")
item["website"] = meta.get("website", "")
item["category"] = meta.get("category", "")
elif note.entity_type == "list":
# Parse markdown task list syntax into structured items
body = note.body or ""
list_items = []
for line in body.split("\n"):
stripped = line.strip()
if stripped.startswith("- [ ] ") or stripped.startswith("- [x] ") or stripped.startswith("- [X] "):
checked_item = not stripped.startswith("- [ ] ")
list_items.append({"text": stripped[6:], "checked": checked_item})
item["list_items"] = list_items
item["item_count"] = len(list_items)
item["checked_count"] = sum(1 for i in list_items if i["checked"])
item["body"] = body
# Task fields — override note_type and add status/priority/due_date
if note.is_task:
item["note_type"] = "task"
@@ -190,7 +160,7 @@ async def _semantic_knowledge_search(
continue
elif note_type == "plan" and (not note.is_task or note.task_kind != "plan"):
continue
elif note_type and note_type not in ("task", "plan") and note.entity_type != note_type:
elif note_type and note_type not in ("task", "plan") and note.note_type != note_type:
continue
if tags and not all(t in (note.tags or []) for t in tags):
continue
@@ -237,7 +207,7 @@ async def get_knowledge_counts(user_id: int, tags: list[str] | None = None) -> d
.where(Note.user_id == user_id)
.where(Note.status.is_(None))
.where(Note.deleted_at.is_(None))
.where(Note.note_type.in_(["note", "person", "place", "list", "process"]))
.where(Note.note_type.in_(["note", "process"]))
.group_by(Note.note_type)
)
if tags:
@@ -273,9 +243,9 @@ async def get_knowledge_counts(user_id: int, tags: list[str] | None = None) -> d
plan_stmt = plan_stmt.where(Note.tags.contains([tag]))
counts["plan"] = (await session.execute(plan_stmt)).scalar_one()
for t in ("note", "person", "place", "list", "task", "plan", "process"):
for t in ("note", "task", "plan", "process"):
counts.setdefault(t, 0)
counts["total"] = sum(counts[t] for t in ("note", "person", "place", "list", "task", "process"))
counts["total"] = sum(counts[t] for t in ("note", "task", "process"))
return counts
@@ -335,42 +305,3 @@ async def get_knowledge_by_ids(user_id: int, ids: list[int]) -> list[dict]:
rows = list((await session.execute(stmt)).scalars().all())
by_id = {n.id: n for n in rows}
return [_note_to_item(by_id[i]) for i in ids if i in by_id]
async def get_people_and_places_context(user_id: int) -> str:
"""Return a compact summary of known people and places for LLM system prompt injection."""
async with async_session() as session:
stmt = (
select(Note)
.where(Note.user_id == user_id)
.where(Note.note_type.in_(["person", "place"]))
.where(Note.status.is_(None))
.where(Note.deleted_at.is_(None))
.order_by(Note.title.asc())
.limit(50)
)
rows = list((await session.execute(stmt)).scalars().all())
if not rows:
return ""
people = [n for n in rows if n.entity_type == "person"]
places = [n for n in rows if n.entity_type == "place"]
lines = []
if people:
parts = []
for p in people:
meta = p.entity_meta or {}
rel = meta.get("relationship", "")
parts.append(f"{p.title}" + (f" ({rel})" if rel else ""))
lines.append("Known people: " + ", ".join(parts))
if places:
parts = []
for p in places:
meta = p.entity_meta or {}
addr = meta.get("address", "")
parts.append(f"{p.title}" + (f" – {addr}" if addr else ""))
lines.append("Known places: " + "; ".join(parts))
return "\n".join(lines)
src/scribe/services/notes.py
-2
View File
@@ -63,7 +63,6 @@ async def create_note(
due_date: date | None = None,
recurrence_rule: dict | None = None,
note_type: str = "note",
entity_meta: dict | None = None,
task_kind: str = "work",
arose_from_id: int | None = None,
) -> Note:
@@ -107,7 +106,6 @@ async def create_note(
due_date=due_date,
recurrence_rule=recurrence_rule,
note_type=note_type,
entity_meta=entity_meta,
task_kind=task_kind,
arose_from_id=arose_from_id,
)
src/scribe/services/recurrence_scheduler.py
+64
View File
@@ -0,0 +1,64 @@
"""Scheduler for recurring-task spawning.
Every 15 minutes, creates the next occurrence of any recurring task whose spawn
time has arrived — draining `recurrence_next_spawn_at`, which is armed on task
completion. Without this job, recurring tasks would never recur.
Uses the BackgroundScheduler pattern shared with the other *_scheduler modules.
(Formerly event_scheduler.py, which also ran event reminders + CalDAV sync;
those were removed when the calendar surface was retired.)
"""
from __future__ import annotations
import asyncio
import logging
from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.triggers.interval import IntervalTrigger
logger = logging.getLogger(__name__)
_scheduler: BackgroundScheduler | None = None
_loop: asyncio.AbstractEventLoop | None = None
async def _run_recurrence_spawn() -> None:
from scribe.services.recurrence import spawn_recurring_tasks # noqa: PLC0415
try:
await spawn_recurring_tasks()
except Exception:
logger.warning("Recurring-task spawn job failed", exc_info=True)
def _run_recurrence_spawn_threadsafe(loop: asyncio.AbstractEventLoop) -> None:
asyncio.run_coroutine_threadsafe(_run_recurrence_spawn(), loop)
def start_recurrence_scheduler(loop: asyncio.AbstractEventLoop) -> None:
global _scheduler, _loop
if _scheduler is not None:
return
_loop = loop
_scheduler = BackgroundScheduler()
# Spawn the next occurrence of due recurring tasks every 15 minutes.
# Without this job, recurrence_next_spawn_at is armed on completion but
# never drained, so recurring tasks never recur.
_scheduler.add_job(
_run_recurrence_spawn_threadsafe,
trigger=IntervalTrigger(minutes=15),
args=[loop],
id="recurrence_spawn",
replace_existing=True,
)
_scheduler.start()
logger.info("Recurrence scheduler started (recurring-task spawn every 15m)")
def stop_recurrence_scheduler() -> None:
global _scheduler
if _scheduler is not None:
_scheduler.shutdown(wait=False)
_scheduler = None
logger.info("Recurrence scheduler stopped")
src/scribe/services/trash.py
+3 -24
View File
@@ -14,7 +14,6 @@ from sqlalchemy import or_, select, update
from scribe.models import async_session
from scribe.models.note import Note
from scribe.models.event import Event
from scribe.models.project import Project
from scribe.models.milestone import Milestone
from scribe.models.rulebook import Rulebook, RulebookTopic, Rule
@@ -23,7 +22,6 @@ from scribe.models.rulebook import Rulebook, RulebookTopic, Rule
_MODEL_FOR = {
"note": Note,
"task": Note,
"event": Event,
"project": Project,
"milestone": Milestone,
"rulebook": Rulebook,
@@ -59,7 +57,7 @@ def _owner_clause(model, user_id: int):
select(Project.id).where(Project.user_id == user_id)
),
)
# Note, Event, Project, Milestone all carry user_id directly.
# Note, Project, Milestone all carry user_id directly.
return model.user_id == user_id
@@ -121,8 +119,6 @@ async def _cascade(session, user_id: int, etype: str, eid: int, batch: str, now)
frontier = [c for c in children if c not in ids]
ids.extend(frontier)
await _set(session, Note, [Note.user_id == user_id, Note.id.in_(ids)], batch, now)
elif etype == "event":
await _set(session, Event, [Event.user_id == user_id, Event.id == eid], batch, now)
elif etype == "rulebook":
topic_ids = (await session.execute(
select(RulebookTopic.id)
@@ -149,35 +145,18 @@ async def delete(user_id: int, entity_type: str, entity_id: int) -> str | None:
"""
batch = str(uuid.uuid4())
now = datetime.now(timezone.utc)
caldav_event: tuple[str, str] | None = None
async with async_session() as session:
if not await _exists_alive(session, user_id, entity_type, entity_id):
return None
# Capture CalDAV linkage before soft-deleting so we can propagate the
# deletion to the external server (the row stays present locally).
if entity_type == "event":
row = (await session.execute(
select(Event.caldav_uid, Event.title).where(
Event.id == entity_id, Event.user_id == user_id
)
)).first()
if row and row[0]:
caldav_event = (row[0], row[1])
await _cascade(session, user_id, entity_type, entity_id, batch, now)
await session.commit()
# Without this the soft-delete only hides the event locally and the remote
# copy lingers forever (and re-appears on any client syncing that server).
if caldav_event:
import asyncio
from scribe.services.events import _push_delete
asyncio.create_task(_push_delete(caldav_event[0], caldav_event[1], user_id))
return batch
# All soft-deletable models, and their trash-listing type label.
_ALL = [Note, Event, Project, Milestone, Rulebook, RulebookTopic, Rule]
_ALL = [Note, Project, Milestone, Rulebook, RulebookTopic, Rule]
_TYPE = {
Note: "note", Event: "event", Project: "project", Milestone: "milestone",
Note: "note", Project: "project", Milestone: "milestone",
Rulebook: "rulebook", RulebookTopic: "topic", Rule: "rule",
}