feat(timeline): add semantic episode analysis

Author: AnkushMalaker

backends/advanced/src/advanced_omi_backend/app_factory.py

@@ -37,6 +37,12 @@
 )
 from advanced_omi_backend.models.memory_audit import MemoryAuditEntry
 from advanced_omi_backend.models.system_event import SystemEvent
+from advanced_omi_backend.models.timeline import (
+    AudioEvidenceSpan,
+    TimelineAnalysisRun,
+    TimelineDay,
+    TimelineEpisode,
+)
 from advanced_omi_backend.models.waveform import WaveformData
 from advanced_omi_backend.redis_factory import create_async_redis
 from advanced_omi_backend.routers.api_router import router as api_router
@@ -90,6 +96,10 @@ async def lifespan(app: FastAPI):
                 PairingCode,
                 DeviceInputItem,
                 DeviceInputJob,
+                AudioEvidenceSpan,
+                TimelineAnalysisRun,
+                TimelineEpisode,
+                TimelineDay,
             ],
         )
         application_logger.info("Beanie initialized for all document models")
@@ -282,6 +292,9 @@ async def _init_cron_scheduler():
                 process_observation_curation,
             )
             from advanced_omi_backend.services.person_photos import sync_person_photos
+            from advanced_omi_backend.services.timeline.discovery import (
+                process_current_timeline_days,
+            )
             from advanced_omi_backend.workers.annotation_jobs import (
                 surface_error_suggestions,
             )
@@ -305,6 +318,7 @@ async def _init_cron_scheduler():
             register_cron_job("device_audio_ingest", process_device_audio)
             register_cron_job("observation_curation", process_observation_curation)
             register_cron_job("screen_context_retention", purge_screen_context)
+            register_cron_job("timeline_analysis", process_current_timeline_days)
 
             scheduler = get_scheduler()
             await scheduler.start()

backends/advanced/src/advanced_omi_backend/controllers/audio_controller.py

@@ -9,7 +9,6 @@
 
 import logging
 import os
-import time
 import uuid
 
 from fastapi import UploadFile
@@ -51,6 +50,12 @@ async def upload_and_process_audio_files(
     device_name: str = "upload",
     source: str = "upload",
     annotation_only: bool = False,
+    external_source_id: str | None = None,
+    external_source_type: str | None = None,
+    data_purpose: str | None = None,
+    memory_excluded: bool | None = None,
+    memory_exclusion_reason: str | None = None,
+    skip_post_processing: bool = False,
 ) -> dict:
     """
     Upload audio files and process them directly.
@@ -114,19 +119,17 @@ async def upload_and_process_audio_files(
                         continue
 
                 # Track external source for deduplication (Google Drive, etc.)
-                external_source_id = None
-                external_source_type = None
                 if source == "gdrive":
-                    external_source_id = getattr(file, "file_id", None) or getattr(
-                        file, "audio_uuid", None
+                    external_source_id = (
+                        external_source_id
+                        or getattr(file, "file_id", None)
+                        or getattr(file, "audio_uuid", None)
                     )
-                    external_source_type = "gdrive"
+                    external_source_type = external_source_type or "gdrive"
                     if not external_source_id:
                         audio_logger.warning(
                             f"Missing file_id for gdrive file: {filename}"
                         )
-                timestamp = int(time.time() * 1000)
-
                 # Validate and prepare audio (read format from WAV file)
                 try:
                     audio_data, sample_rate, sample_width, channels, duration = (
@@ -169,11 +172,13 @@ async def upload_and_process_audio_files(
                     ),
                     external_source_id=external_source_id,
                     external_source_type=external_source_type,
-                    data_purpose="annotation" if annotation_only else None,
-                    memory_excluded=annotation_only,
-                    memory_exclusion_reason=(
-                        "annotation_only_upload" if annotation_only else None
+                    data_purpose=data_purpose
+                    or ("annotation" if annotation_only else None),
+                    memory_excluded=(
+                        annotation_only if memory_excluded is None else memory_excluded
                     ),
+                    memory_exclusion_reason=memory_exclusion_reason
+                    or ("annotation_only_upload" if annotation_only else None),
                 )
                 await conversation.insert()
                 conversation_id = (
@@ -260,14 +265,22 @@ async def upload_and_process_audio_files(
                     )
 
                 # Enqueue post-conversation processing job chain (depends on transcription)
-                job_ids = start_post_conversation_jobs(
-                    conversation_id=conversation_id,
-                    user_id=user.user_id,
-                    transcript_version_id=version_id,  # Pass the version_id from transcription job
-                    depends_on_job=transcription_job,  # Wait for transcription to complete (or None)
-                    client_id=client_id,  # Pass client_id for UI tracking
-                    skip_memory_extraction=annotation_only,
-                )
+                if skip_post_processing:
+                    job_ids = {
+                        "speaker_recognition": None,
+                        "memory": None,
+                        "title_summary": None,
+                        "event_dispatch": None,
+                    }
+                else:
+                    job_ids = start_post_conversation_jobs(
+                        conversation_id=conversation_id,
+                        user_id=user.user_id,
+                        transcript_version_id=version_id,
+                        depends_on_job=transcription_job,
+                        client_id=client_id,
+                        skip_memory_extraction=annotation_only,
+                    )
 
                 file_result = {
                     "filename": filename,

backends/advanced/src/advanced_omi_backend/controllers/conversation_controller.py

@@ -378,6 +378,9 @@ async def get_conversations(
     """
     try:
         user_filter = {} if user.is_superuser else {"user_id": str(user.user_id)}
+        # Continuous ScreenPipe artifacts are evidence for semantic episodes, not
+        # user-facing conversations. The raw device-input API remains available.
+        user_filter["external_source_type"] = {"$ne": "screenpipe"}
 
         if starred_only:
             user_filter["starred"] = True
@@ -544,7 +547,10 @@ async def _regex_search_conversations(
     """Filter conversations by text across the selected conversation fields."""
     collection = Conversation.get_pymongo_collection()
 
-    match_filter: dict = {"deleted": False}
+    match_filter: dict = {
+        "deleted": False,
+        "external_source_type": {"$ne": "screenpipe"},
+    }
     if not user.is_superuser:
         match_filter["user_id"] = str(user.user_id)
 

backends/advanced/src/advanced_omi_backend/models/job.py

@@ -54,6 +54,12 @@ async def _ensure_beanie_initialized():
                 PairingCode,
             )
             from advanced_omi_backend.models.memory_audit import MemoryAuditEntry
+            from advanced_omi_backend.models.timeline import (
+                AudioEvidenceSpan,
+                TimelineAnalysisRun,
+                TimelineDay,
+                TimelineEpisode,
+            )
             from advanced_omi_backend.models.user import User
             from advanced_omi_backend.models.waveform import WaveformData
 
@@ -86,6 +92,10 @@ async def _ensure_beanie_initialized():
                     PairingCode,
                     DeviceInputItem,
                     DeviceInputJob,
+                    AudioEvidenceSpan,
+                    TimelineAnalysisRun,
+                    TimelineEpisode,
+                    TimelineDay,
                 ],
             )