From d9ab538ef4871bde76510e50181d21a93d97d87a Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Sat, 25 Apr 2026 22:37:39 -0400 Subject: [PATCH] feat(journal): backend services (moments, search, prep, scheduler, pipeline) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - services/moments.py — CRUD with embedding sync + entity-link helpers - services/journal_search.py — three-mode search (temporal / entity / semantic) with notes-RAG isolation; cosine-similarity helper unit-tested - services/journal_prep.py — gathers tasks/events/weather/news/projects/ recent-moments/open-threads into a structured prep block - services/journal_scheduler.py — per-user APScheduler cron for daily prep, follows the BackgroundScheduler + threadsafe-async pattern from event_scheduler - services/journal_pipeline.py — system prompt (persona + calibration rules) with last-48h ambient moments injection - app.py: wire journal scheduler start/stop hooks - routes/settings.py: re-add live-reschedule on timezone change (now via journal_scheduler.update_user_schedule) Co-Authored-By: Claude Opus 4.7 --- src/fabledassistant/app.py | 6 +- src/fabledassistant/routes/settings.py | 6 +- .../services/journal_pipeline.py | 117 ++++++++++ src/fabledassistant/services/journal_prep.py | 208 ++++++++++++++++++ .../services/journal_scheduler.py | 140 ++++++++++++ .../services/journal_search.py | 204 +++++++++++++++++ src/fabledassistant/services/moments.py | 185 ++++++++++++++++ 7 files changed, 863 insertions(+), 3 deletions(-) create mode 100644 src/fabledassistant/services/journal_pipeline.py create mode 100644 src/fabledassistant/services/journal_prep.py create mode 100644 src/fabledassistant/services/journal_scheduler.py create mode 100644 src/fabledassistant/services/journal_search.py create mode 100644 src/fabledassistant/services/moments.py diff --git a/src/fabledassistant/app.py b/src/fabledassistant/app.py index 102f215..4b879c9 100644 --- a/src/fabledassistant/app.py +++ b/src/fabledassistant/app.py @@ -331,7 +331,9 @@ def create_app() -> Quart: asyncio.create_task(_delayed_backfill()) - # Journal scheduler is wired up in Stage C (services/journal_scheduler). + # Start journal scheduler (per-user daily prep generation) + from fabledassistant.services.journal_scheduler import start_journal_scheduler + start_journal_scheduler(asyncio.get_running_loop()) # Start event scheduler (reminders + CalDAV pull sync) from fabledassistant.services.event_scheduler import start_event_scheduler @@ -345,6 +347,8 @@ def create_app() -> Quart: @app.after_serving async def shutdown(): + from fabledassistant.services.journal_scheduler import stop_journal_scheduler + stop_journal_scheduler() from fabledassistant.services.event_scheduler import stop_event_scheduler stop_event_scheduler() diff --git a/src/fabledassistant/routes/settings.py b/src/fabledassistant/routes/settings.py index 62a2398..7f6b179 100644 --- a/src/fabledassistant/routes/settings.py +++ b/src/fabledassistant/routes/settings.py @@ -94,8 +94,10 @@ async def update_settings_route(): if to_save: await set_settings_batch(uid, to_save) - # Journal scheduler live-reschedule on timezone change is wired up in - # services/journal_scheduler.update_user_schedule (Stage C). + # Live-reschedule the journal daily-prep job when the timezone changes. + if "user_timezone" in to_save: + from fabledassistant.services.journal_scheduler import update_user_schedule as _update_journal_schedule + await _update_journal_schedule(uid) if "default_model" in to_save and to_save["default_model"]: asyncio.create_task(_prime_kv_cache_bg(uid, to_save["default_model"])) diff --git a/src/fabledassistant/services/journal_pipeline.py b/src/fabledassistant/services/journal_pipeline.py new file mode 100644 index 0000000..b65e4e2 --- /dev/null +++ b/src/fabledassistant/services/journal_pipeline.py @@ -0,0 +1,117 @@ +"""System prompt + ambient context injection for journal conversations. + +Distinct from the chat pipeline (services/llm.py) in three ways: +1. New persona: warm, curious listener; not a task manager. +2. Calibration rules: ask before structural changes; record_moment freely. +3. Auto-injects last ~48h of moments for ambient continuity (no notes-RAG). + +Notes-RAG auto-injection MUST be disabled for journal sessions — the +journal's ambient context replaces it. Failing to disable would let notes +leak into journal sessions, violating the design's isolation invariant. +""" +from __future__ import annotations + +import datetime + +from fabledassistant.services.journal_search import search_journal +from fabledassistant.services.user_profile import build_profile_context + +JOURNAL_PERSONA = ( + "You are a warm, curious listener helping the user record their day. " + "You are NOT a task manager. Your role is to listen, ask gentle " + "follow-up questions when something seems significant or underspecified, " + "and quietly maintain structure as a byproduct of conversation." +) + +JOURNAL_CALIBRATION = """\ +CALIBRATION (very important): +- When the user mentions a person you don't know, ask: "Is someone I should remember?" Confirm before calling update_person. +- When a person reference is ambiguous (multiple matches), ask which one. Never guess. +- When the user says something action-implying (e.g., "I finished X"), ASK before calling update_task. Silent updates feel derailing. +- Use the existing tool's `confirmed=false` -> user-confirms -> `confirmed=true` pattern. The frontend renders the inline confirm UI automatically. +- record_moment is the EXCEPTION: call it freely and silently when the user mentions a meaningful beat. Moments are cheap and user-correctable in the timeline view; gating them would kill the flow. +- Do NOT call set_rag_scope. The journal scope is implicit. +- Notes are not auto-retrieved here. If you need to reference notes, call search_notes explicitly. +""" + +PHASE_GREETINGS = { + "morning": "Morning. What's the day looking like for you?", + "midday": "How's it going so far?", + "evening": "Wrapping up — how'd the day shake out?", +} + + +def determine_phase( + *, + now_local: datetime.datetime, + day_rollover_hour: int = 4, + morning_end_hour: int = 12, + midday_end_hour: int = 18, +) -> str: + """Return 'morning' | 'midday' | 'evening' for a local datetime.""" + h = now_local.hour + if h < day_rollover_hour: + return "evening" + if h < morning_end_hour: + return "morning" + if h < midday_end_hour: + return "midday" + return "evening" + + +def phase_greeting(phase: str) -> str: + return PHASE_GREETINGS.get(phase, PHASE_GREETINGS["morning"]) + + +async def build_journal_system_prompt( + *, + user_id: int, + day_date: datetime.date, + user_timezone: str, +) -> str: + """Static-then-dynamic system prompt. + + Static prefix (persona + calibration) is identical on every request, + preserving Ollama KV cache. Dynamic suffix changes per-day. + """ + static_block = f"{JOURNAL_PERSONA}\n\n{JOURNAL_CALIBRATION}" + + today_iso = day_date.isoformat() + tz_block = f"Today is {today_iso} ({user_timezone})." + + profile_context = await build_profile_context(user_id) + profile_section = f"\n\n{profile_context}" if profile_context else "" + + ambient = await _ambient_moments_block(user_id=user_id, day_date=day_date) + ambient_section = ( + f"\n\nRECENT JOURNAL CONTEXT (last 48h):\n{ambient}" if ambient else "" + ) + + return f"{static_block}\n\n{tz_block}{profile_section}{ambient_section}" + + +async def _ambient_moments_block( + *, user_id: int, day_date: datetime.date +) -> str: + """Render last 48h of moments as a compact text block. + + Capped at 20 moments / 1500 chars total. Distinct from RAG retrieval. + """ + moments = await search_journal( + user_id=user_id, + date_from=day_date - datetime.timedelta(days=2), + date_to=day_date, + limit=20, + ) + if not moments: + return "" + + lines: list[str] = [] + total_chars = 0 + for m in moments: + line = f"- [{m['occurred_at']}] {m['content']}" + if total_chars + len(line) > 1500: + break + lines.append(line) + total_chars += len(line) + return "\n".join(lines) diff --git a/src/fabledassistant/services/journal_prep.py b/src/fabledassistant/services/journal_prep.py new file mode 100644 index 0000000..69a5421 --- /dev/null +++ b/src/fabledassistant/services/journal_prep.py @@ -0,0 +1,208 @@ +"""Daily prep generator for the Journal. + +Runs once per day per user (scheduled, or lazy on first journal-open of a +new day). Gathers raw data into a structured prep block that becomes the +first message of the day's journal Conversation. + +The output shape lives on Message.msg_metadata as: + { kind: 'daily_prep', sections: { tasks, events, weather, news, + projects, recent_moments, open_threads } } + +Deliberately deterministic — no LLM call here. The journal LLM weaves +narrative into the conversation later, when the user actually starts +talking. Generating prose at scheduled-prep time would burn model time +on something the user may never read. +""" +from __future__ import annotations + +import datetime +import logging + +from sqlalchemy import select + +from fabledassistant.models import Conversation, Message, async_session +from fabledassistant.services.events import list_events +from fabledassistant.services.journal_search import search_journal +from fabledassistant.services.notes import list_notes +from fabledassistant.services.projects import list_projects +from fabledassistant.services.rss import get_recent_items +from fabledassistant.services.weather import get_cached_weather_rows + +logger = logging.getLogger(__name__) + + +async def generate_daily_prep( + *, + user_id: int, + day_date: datetime.date, + user_timezone: str, +) -> dict: + """Gather all daily-prep sections and return them as a dict.""" + sections: dict = {} + + try: + tasks_today, _ = await list_notes( + user_id=user_id, + is_task=True, + status=["todo", "in_progress"], + due_before=day_date, + limit=20, + sort="due_date", + order="asc", + ) + sections["tasks"] = [ + { + "id": t.id, + "title": t.title, + "status": t.status, + "priority": t.priority, + "due_date": t.due_date.isoformat() if t.due_date else None, + } + for t in tasks_today + ] + except Exception: + logger.exception("daily_prep tasks section failed for user %d", user_id) + sections["tasks"] = [] + + try: + day_start = datetime.datetime.combine(day_date, datetime.time.min) + day_end = datetime.datetime.combine(day_date, datetime.time.max) + # list_events already returns dicts. + sections["events"] = await list_events( + user_id=user_id, + date_from=day_start, + date_to=day_end, + ) + except Exception: + logger.exception("daily_prep events section failed for user %d", user_id) + sections["events"] = [] + + try: + weather_rows = await get_cached_weather_rows(user_id) + sections["weather"] = [w.to_dict() for w in weather_rows] + except Exception: + logger.exception("daily_prep weather section failed for user %d", user_id) + sections["weather"] = [] + + try: + news = await get_recent_items(user_id=user_id, limit=5) + sections["news"] = news + except Exception: + logger.exception("daily_prep news section failed for user %d", user_id) + sections["news"] = [] + + try: + projects = await list_projects(user_id=user_id, status="active") + sections["projects"] = [ + { + "id": p.id, + "title": p.title, + "auto_summary": p.auto_summary, + } + for p in projects[:5] + ] + except Exception: + logger.exception("daily_prep projects section failed for user %d", user_id) + sections["projects"] = [] + + try: + sections["recent_moments"] = await search_journal( + user_id=user_id, + date_from=day_date - datetime.timedelta(days=3), + date_to=day_date - datetime.timedelta(days=1), + limit=10, + ) + except Exception: + logger.exception("daily_prep recent_moments section failed for user %d", user_id) + sections["recent_moments"] = [] + + try: + sections["open_threads"] = await _open_threads(user_id=user_id, day_date=day_date) + except Exception: + logger.exception("daily_prep open_threads section failed for user %d", user_id) + sections["open_threads"] = [] + + return sections + + +async def _open_threads(*, user_id: int, day_date: datetime.date) -> list[dict]: + """Heuristic: moments from the last 7 days that look unresolved. + + Treated as 'unresolved' when they have no linked tasks/notes and aren't + pinned. Starting heuristic — refine empirically. + """ + candidates = await search_journal( + user_id=user_id, + date_from=day_date - datetime.timedelta(days=7), + date_to=day_date - datetime.timedelta(days=1), + limit=50, + ) + return [ + m for m in candidates + if not m.get("task_ids") + and not m.get("note_ids") + and not m.get("pinned") + ] + + +async def ensure_daily_prep_message( + *, + user_id: int, + day_date: datetime.date, + user_timezone: str, + force: bool = False, +) -> Message: + """Get or create today's journal Conversation, then ensure the prep message exists. + + With force=True, regenerate even if a prep message already exists today + (used by the manual /api/journal/trigger-prep endpoint). + """ + async with async_session() as session: + result = await session.execute( + select(Conversation).where( + Conversation.user_id == user_id, + Conversation.conversation_type == "journal", + Conversation.day_date == day_date, + ) + ) + conv = result.scalar_one_or_none() + if conv is None: + conv = Conversation( + user_id=user_id, + conversation_type="journal", + day_date=day_date, + title=day_date.isoformat(), + ) + session.add(conv) + await session.flush() + + prep_stmt = select(Message).where( + Message.conversation_id == conv.id, + Message.role == "system", + ) + existing_prep = None + for msg in (await session.execute(prep_stmt)).scalars(): + if msg.msg_metadata and msg.msg_metadata.get("kind") == "daily_prep": + existing_prep = msg + break + + if existing_prep and not force: + return existing_prep + + sections = await generate_daily_prep( + user_id=user_id, day_date=day_date, user_timezone=user_timezone + ) + if existing_prep: + existing_prep.msg_metadata = {"kind": "daily_prep", "sections": sections} + await session.commit() + return existing_prep + + prep_msg = Message( + conversation_id=conv.id, + role="system", + content="", + msg_metadata={"kind": "daily_prep", "sections": sections}, + ) + session.add(prep_msg) + await session.commit() + return prep_msg diff --git a/src/fabledassistant/services/journal_scheduler.py b/src/fabledassistant/services/journal_scheduler.py new file mode 100644 index 0000000..8130207 --- /dev/null +++ b/src/fabledassistant/services/journal_scheduler.py @@ -0,0 +1,140 @@ +"""APScheduler instance for the Journal — daily prep generation. + +One per-user cron job: generate today's daily prep at the configured +prep_time in the user's local timezone. Replaces briefing_scheduler. + +Mirrors the BackgroundScheduler + threadsafe-async-call pattern used by +event_scheduler.py. +""" +from __future__ import annotations + +import asyncio +import datetime +import json +import logging +from zoneinfo import ZoneInfo + +from apscheduler.schedulers.background import BackgroundScheduler +from apscheduler.triggers.cron import CronTrigger +from sqlalchemy import select + +from fabledassistant.models import User, async_session +from fabledassistant.services.journal_prep import ensure_daily_prep_message +from fabledassistant.services.settings import get_setting + +logger = logging.getLogger(__name__) + +_scheduler: BackgroundScheduler | None = None +_loop: asyncio.AbstractEventLoop | None = None + + +DEFAULT_CONFIG = { + "prep_enabled": True, + "prep_hour": 5, + "prep_minute": 0, + "day_rollover_hour": 4, + "morning_end_hour": 12, + "midday_end_hour": 18, +} + + +async def get_journal_config(user_id: int) -> dict: + """Load a user's journal_config (JSON in settings) merged with defaults.""" + raw = await get_setting(user_id, "journal_config", "") + config: dict = {} + if raw: + try: + parsed = json.loads(raw) if isinstance(raw, str) else raw + if isinstance(parsed, dict): + config = parsed + except Exception: + logger.warning("Invalid journal_config for user %d; using defaults", user_id) + return {**DEFAULT_CONFIG, **config} + + +async def get_user_timezone(user_id: int) -> str: + return await get_setting(user_id, "user_timezone", "UTC") or "UTC" + + +def _resolve_tz(tz_str: str) -> ZoneInfo: + try: + return ZoneInfo(tz_str) + except Exception: + return ZoneInfo("UTC") + + +async def _do_daily_prep(user_id: int) -> None: + try: + tz_str = await get_user_timezone(user_id) + tz = _resolve_tz(tz_str) + config = await get_journal_config(user_id) + rollover = int(config.get("day_rollover_hour", 4)) + now = datetime.datetime.now(tz) + # Today is the day after the most recent rollover. + if now.hour < rollover: + today = (now - datetime.timedelta(days=1)).date() + else: + today = now.date() + await ensure_daily_prep_message( + user_id=user_id, day_date=today, user_timezone=tz_str + ) + except Exception: + logger.exception("Daily prep failed for user %d", user_id) + + +def _run_daily_prep_threadsafe(user_id: int) -> None: + if _loop is None: + return + asyncio.run_coroutine_threadsafe(_do_daily_prep(user_id), _loop) + + +async def update_user_schedule(user_id: int) -> None: + """Add or replace this user's daily-prep job using their current config.""" + if _scheduler is None: + return + job_id = f"journal_prep_{user_id}" + if _scheduler.get_job(job_id): + _scheduler.remove_job(job_id) + + config = await get_journal_config(user_id) + if not config.get("prep_enabled", True): + return + + tz_str = await get_user_timezone(user_id) + tz = _resolve_tz(tz_str) + prep_hour = int(config.get("prep_hour", 5)) + prep_minute = int(config.get("prep_minute", 0)) + + _scheduler.add_job( + _run_daily_prep_threadsafe, + trigger=CronTrigger(hour=prep_hour, minute=prep_minute, timezone=tz), + args=[user_id], + id=job_id, + replace_existing=True, + ) + + +async def _register_all_user_jobs() -> None: + async with async_session() as session: + users = (await session.execute(select(User))).scalars().all() + for user in users: + await update_user_schedule(user.id) + + +def start_journal_scheduler(loop: asyncio.AbstractEventLoop) -> None: + global _scheduler, _loop + if _scheduler is not None: + return + _loop = loop + _scheduler = BackgroundScheduler() + _scheduler.start() + asyncio.run_coroutine_threadsafe(_register_all_user_jobs(), loop) + logger.info("Journal scheduler started") + + +def stop_journal_scheduler() -> None: + global _scheduler + if _scheduler is not None: + _scheduler.shutdown(wait=False) + _scheduler = None + logger.info("Journal scheduler stopped") diff --git a/src/fabledassistant/services/journal_search.py b/src/fabledassistant/services/journal_search.py new file mode 100644 index 0000000..ca12740 --- /dev/null +++ b/src/fabledassistant/services/journal_search.py @@ -0,0 +1,204 @@ +"""Search across Moments and (optionally) journal transcripts. + +Three modes, expressed through one tool surface: +1. Pure temporal: no query, just date range -> ordered by occurred_at DESC. +2. Pure entity: person_id or place_id filter -> junction lookup. +3. Semantic: query string -> embedding similarity, optionally constrained. + +This module ONLY queries Moments and (optionally) journal-conversation +messages. It MUST NOT touch notes or note_embeddings. The notes-RAG and +journal-RAG isolation is a hard invariant of the journal design. +""" +from __future__ import annotations + +import datetime +import math +from typing import Sequence + +from sqlalchemy import select + +from fabledassistant.models import ( + Conversation, + Message, + Moment, + MomentEmbedding, + async_session, + moment_people, + moment_places, +) +from fabledassistant.services.embeddings import get_embedding + +DEFAULT_LIMIT = 10 +DEFAULT_THRESHOLD = 0.55 + + +def _cosine(a: Sequence[float], b: Sequence[float]) -> float: + if not a or not b: + return 0.0 + dot = sum(x * y for x, y in zip(a, b)) + norm_a = math.sqrt(sum(x * x for x in a)) + norm_b = math.sqrt(sum(y * y for y in b)) + if norm_a == 0 or norm_b == 0: + return 0.0 + return dot / (norm_a * norm_b) + + +async def search_journal( + *, + user_id: int, + query: str | None = None, + person_id: int | None = None, + place_id: int | None = None, + tag: str | None = None, + date_from: datetime.date | None = None, + date_to: datetime.date | None = None, + include_transcripts: bool = False, + limit: int = DEFAULT_LIMIT, + threshold: float = DEFAULT_THRESHOLD, +) -> list[dict]: + """Return Moments (and optional transcript snippets) matching the filters. + + Result rows: dicts from Moment.to_dict() plus an optional `score` when + `query` is set. Transcript snippets carry `kind='transcript'` and `id=None` + to distinguish them. + """ + moment_rows = await _search_moments( + user_id=user_id, + query=query, + person_id=person_id, + place_id=place_id, + tag=tag, + date_from=date_from, + date_to=date_to, + limit=limit, + threshold=threshold, + ) + + if not include_transcripts: + return moment_rows + + transcript_rows = await _search_transcripts( + user_id=user_id, + query=query, + date_from=date_from, + date_to=date_to, + limit=limit, + ) + return moment_rows + transcript_rows + + +async def _search_moments( + *, + user_id: int, + query: str | None, + person_id: int | None, + place_id: int | None, + tag: str | None, + date_from: datetime.date | None, + date_to: datetime.date | None, + limit: int, + threshold: float, +) -> list[dict]: + async with async_session() as session: + stmt = select(Moment).where(Moment.user_id == user_id) + + if person_id is not None: + stmt = stmt.join(moment_people, moment_people.c.moment_id == Moment.id).where( + moment_people.c.person_id == person_id + ) + if place_id is not None: + stmt = stmt.join(moment_places, moment_places.c.moment_id == Moment.id).where( + moment_places.c.place_id == place_id + ) + if tag is not None: + stmt = stmt.where(Moment.tags.any(tag)) + if date_from is not None: + stmt = stmt.where(Moment.day_date >= date_from) + if date_to is not None: + stmt = stmt.where(Moment.day_date <= date_to) + + if query is None: + stmt = stmt.order_by(Moment.occurred_at.desc()).limit(limit) + moments = (await session.execute(stmt)).scalars().all() + return [m.to_dict() for m in moments] + + # Semantic mode. Pull a wider candidate set, then rank in Python. + candidate_stmt = stmt.order_by(Moment.occurred_at.desc()).limit(limit * 5) + candidates = (await session.execute(candidate_stmt)).scalars().all() + if not candidates: + return [] + + candidate_ids = [m.id for m in candidates] + emb_stmt = select(MomentEmbedding).where( + MomentEmbedding.moment_id.in_(candidate_ids) + ) + embeddings = { + e.moment_id: e.embedding + for e in (await session.execute(emb_stmt)).scalars() + } + + query_vec = await get_embedding(query) + scored = [] + for m in candidates: + vec = embeddings.get(m.id) + if vec is None: + continue + score = _cosine(query_vec, vec) + if score >= threshold: + row = m.to_dict() + row["score"] = score + scored.append(row) + + scored.sort(key=lambda r: r["score"], reverse=True) + return scored[:limit] + + +async def _search_transcripts( + *, + user_id: int, + query: str | None, + date_from: datetime.date | None, + date_to: datetime.date | None, + limit: int, +) -> list[dict]: + """Fallback substring search over raw journal Messages. + + Substring is intentional: transcripts catch what the LLM didn't extract + as Moments. Semantic search over messages would require a third embedding + index, which we deliberately don't maintain. + """ + if query is None: + return [] + async with async_session() as session: + stmt = ( + select(Message, Conversation.day_date) + .join(Conversation, Message.conversation_id == Conversation.id) + .where( + Conversation.user_id == user_id, + Conversation.conversation_type == "journal", + Message.content.ilike(f"%{query}%"), + ) + .order_by(Message.created_at.desc()) + .limit(limit) + ) + if date_from is not None: + stmt = stmt.where(Conversation.day_date >= date_from) + if date_to is not None: + stmt = stmt.where(Conversation.day_date <= date_to) + rows = (await session.execute(stmt)).all() + return [ + { + "id": None, + "kind": "transcript", + "message_id": msg.id, + "conversation_id": msg.conversation_id, + "day_date": day_date.isoformat() if day_date else None, + "occurred_at": msg.created_at.isoformat(), + "content": msg.content[:400], + "raw_excerpt": None, + "tags": [], + "people": [], + "places": [], + } + for msg, day_date in rows + ] diff --git a/src/fabledassistant/services/moments.py b/src/fabledassistant/services/moments.py new file mode 100644 index 0000000..0079b6a --- /dev/null +++ b/src/fabledassistant/services/moments.py @@ -0,0 +1,185 @@ +"""Moment CRUD, entity linking, and embedding sync. + +Moments are small structured extractions from journal conversations. +Stored separately from Notes — they have their own embedding index and +the isolation between notes-RAG and journal-RAG is enforced at the API +boundary, not just the schema. +""" +from __future__ import annotations + +import datetime +from typing import Sequence + +from sqlalchemy import delete, select +from sqlalchemy.ext.asyncio import AsyncSession + +from fabledassistant.models import ( + Moment, + MomentEmbedding, + async_session, + moment_notes, + moment_people, + moment_places, + moment_tasks, +) +from fabledassistant.services.embeddings import get_embedding + + +async def create_moment( + *, + user_id: int, + content: str, + occurred_at: datetime.datetime, + day_date: datetime.date, + conversation_id: int | None = None, + source_message_id: int | None = None, + raw_excerpt: str | None = None, + tags: Sequence[str] = (), + person_ids: Sequence[int] = (), + place_ids: Sequence[int] = (), + task_ids: Sequence[int] = (), + note_ids: Sequence[int] = (), +) -> Moment: + """Insert a Moment, write its embedding, create junction rows.""" + async with async_session() as session: + moment = Moment( + user_id=user_id, + conversation_id=conversation_id, + source_message_id=source_message_id, + day_date=day_date, + occurred_at=occurred_at, + content=content, + raw_excerpt=raw_excerpt, + tags=list(tags), + ) + session.add(moment) + await session.flush() + + await _set_links( + session, + moment_id=moment.id, + person_ids=person_ids, + place_ids=place_ids, + task_ids=task_ids, + note_ids=note_ids, + ) + + embedding_vector = await get_embedding(content) + session.add( + MomentEmbedding( + moment_id=moment.id, + user_id=user_id, + embedding=embedding_vector, + ) + ) + + await session.commit() + await session.refresh(moment, ["people", "places", "tasks", "notes"]) + return moment + + +async def get_moment(*, user_id: int, moment_id: int) -> Moment | None: + async with async_session() as session: + result = await session.execute( + select(Moment).where(Moment.id == moment_id, Moment.user_id == user_id) + ) + return result.scalar_one_or_none() + + +async def update_moment( + *, + user_id: int, + moment_id: int, + content: str | None = None, + tags: Sequence[str] | None = None, + pinned: bool | None = None, + person_ids: Sequence[int] | None = None, + place_ids: Sequence[int] | None = None, + task_ids: Sequence[int] | None = None, + note_ids: Sequence[int] | None = None, +) -> Moment | None: + """Update fields and (optionally) replace junction sets. + + None for a junction parameter ⇒ leave unchanged. Empty sequence ⇒ clear. + Mirrors the Notes update convention. + """ + async with async_session() as session: + result = await session.execute( + select(Moment).where(Moment.id == moment_id, Moment.user_id == user_id) + ) + moment = result.scalar_one_or_none() + if moment is None: + return None + + if content is not None: + moment.content = content + embedding_vector = await get_embedding(content) + await session.execute( + delete(MomentEmbedding).where(MomentEmbedding.moment_id == moment_id) + ) + session.add( + MomentEmbedding( + moment_id=moment_id, + user_id=user_id, + embedding=embedding_vector, + ) + ) + if tags is not None: + moment.tags = list(tags) + if pinned is not None: + moment.pinned = pinned + + await _set_links( + session, + moment_id=moment_id, + person_ids=person_ids, + place_ids=place_ids, + task_ids=task_ids, + note_ids=note_ids, + ) + + await session.commit() + await session.refresh(moment, ["people", "places", "tasks", "notes"]) + return moment + + +async def delete_moment(*, user_id: int, moment_id: int) -> bool: + async with async_session() as session: + result = await session.execute( + select(Moment).where(Moment.id == moment_id, Moment.user_id == user_id) + ) + moment = result.scalar_one_or_none() + if moment is None: + return False + await session.delete(moment) + await session.commit() + return True + + +async def _set_links( + session: AsyncSession, + *, + moment_id: int, + person_ids: Sequence[int] | None, + place_ids: Sequence[int] | None, + task_ids: Sequence[int] | None, + note_ids: Sequence[int] | None, +) -> None: + """Replace junction-table rows. + + None ⇒ leave existing rows alone. Empty sequence ⇒ clear all rows. + """ + for ids, table, fk_col in [ + (person_ids, moment_people, "person_id"), + (place_ids, moment_places, "place_id"), + (task_ids, moment_tasks, "task_id"), + (note_ids, moment_notes, "note_id"), + ]: + if ids is None: + continue + await session.execute(delete(table).where(table.c.moment_id == moment_id)) + if ids: + await session.execute( + table.insert(), + [{"moment_id": moment_id, fk_col: i} for i in ids], + )