feat(journal): backend services (moments, search, prep, scheduler, pipeline)

- 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 <noreply@anthropic.com>

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()
 

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
+    # Live-reschedule the journal daily-prep job when the timezone changes.
-    # services/journal_scheduler.update_user_schedule (Stage C).
+    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"]))

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 <Name> 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)

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

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")

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
+        ]

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],
+            )