fix mobile stuff and sessionization bug

baml_src/aggregation.baml

@@ -52,7 +52,6 @@ template_string FormatMobileAppUsageLog(log: MobileAppUsageEvent) #"
 Mobile App Usage Log
 Time: {{ log.timestamp.minutes}}m{{ log.timestamp.seconds}}s
 App Label: {{ log.app_label }}
-Package Name: {{ log.package_name }}
 {% if log.activity_name %}Activity Name: {{ log.activity_name }}{% endif %}
 {% if log.media_summary %}Media: {{ log.media_summary }}{% endif %}
 {% if log.notification_text %}Notification: {{ log.notification_text }}{% endif %}
@@ -82,19 +81,19 @@ Text from OCR extraction: {{ log.image_text }}
 
 
 
-template_string FormattedLogsList( logs: TimelineLog[] ) #"
-{%- for log in logs %}
-{{ loop.index }}. {%- if log.kind == "mobile_app_usage" %}
-{{ FormatMobileAppUsageLog(log) }}
-{%- elif log.kind == "desktop_screenshot_vlm" %}
-{{ FormatDesktopCheckScreenshotVLMLog(log) }}
-{%- elif log.kind == "desktop_screenshot_ocr" %}
-{{ FormatDesktopCheckScreenshotOCRLog(log) }}
-{%- endif %}
+template_string FormattedLogsList(logs: TimelineLog[]) #"
+{% for log in logs %}
+{{ loop.index }}.
+{% if log.kind == "mobile_app_usage" %}
+  {{ FormatMobileAppUsageLog(log) }}
+{% elif log.kind == "desktop_screenshot_vlm" %}
+  {{ FormatDesktopCheckScreenshotVLMLog(log) }}
+{% elif log.kind == "desktop_screenshot_ocr" %}
+  {{ FormatDesktopCheckScreenshotOCRLog(log) }}
+{% endif %}
 
-{%- endfor %}
+{% endfor %}
 "#
-
 class Event {
   activity_id string
   @description("Must match activity id. Should be a lower-snake-case slug")

docker-compose.yml

@@ -58,6 +58,8 @@ services:
       prometheus_multiproc_dir: /tmp/dramatiq-prometheus
       dramatiq_prom_db: /tmp/dramatiq-prometheus
       CLEPSY_MODE: dev
+    env_file:
+      - .env
     volumes:
       - .:/app
       - db_data_dev:/var/lib/clepsy

src/clepsy/auth/auth_middleware.py

@@ -208,7 +208,7 @@ async def __call__(self, request: Request, call_next):
             token_bytes = self._b64_to_bytes(token)
         except Exception as e:
             logger.warning(
-                f"Desktop source sent malformed bearer token to {request.url.path}: {e}"
+                f"Source sent malformed bearer token to {request.url.path}: {e}"
             )
             raise HTTPException(
                 status_code=403, detail="Invalid bearer token format"
@@ -222,14 +222,12 @@ async def __call__(self, request: Request, call_next):
 
         if source is None:
             logger.warning(
-                f"Desktop source attempted to connect with invalid token to {request.url.path}. "
-                "This usually means the desktop client needs to be re-paired."
+                f"Source attempted to connect with invalid token to {request.url.path}. "
+                "This usually means the source needs to be re-paired."
             )
             raise HTTPException(status_code=403, detail="Invalid token")
         if getattr(source, "status", None) != SourceStatus.ACTIVE:
-            logger.warning(
-                f"Desktop source {source.id} attempted to connect but is disabled"
-            )
+            logger.warning(f"Source {source.id} attempted to connect but is disabled")
             raise HTTPException(status_code=403, detail="Source disabled")
 
         # Attach to request for handlers to use

src/clepsy/jobs/mobile.py

@@ -1,7 +1,6 @@
 from __future__ import annotations
 # ruff: noqa: I001
 
-import json
 from datetime import datetime, timezone
 
 import dramatiq
@@ -19,26 +18,17 @@ def ensure_aware(ts: datetime) -> datetime:
     return ts
 
 
-def serialize_mobile_event(event: MobileAppUsageEvent) -> tuple[str, datetime, str]:
-    payload = {
-        "app_label": event.app_label,
-        "package_name": event.package_name,
-        "activity_name": event.activity_name,
-        "media_metadata": event.media_metadata,
-        "notification_text": event.notification_text,
-        "timestamp": ensure_aware(event.timestamp).isoformat(),
-    }
-    return "mobile_app_usage", event.timestamp, json.dumps(payload)
-
-
 @dramatiq.actor
 def persist_mobile_app_usage_job(event_dict: dict) -> None:
     """Dramatiq async actor: anonymize (if needed) and publish mobile app usage event to stream."""
-    evt = MobileAppUsageEvent.model_validate(event_dict)
 
-    if evt.notification_text:
-        evt.notification_text = sanitize_text(evt.notification_text)
+    if notification_text := event_dict["notification_text"]:
+        event_dict["notification_text"] = sanitize_text(notification_text)
+    evt = MobileAppUsageEvent.model_validate(event_dict)
 
-    etype, etime, payload_json = serialize_mobile_event(evt)
-    _ = xadd_source_event(event_type=etype, timestamp=etime, payload_json=payload_json)
+    etype = "mobile_app_usage"
+    etime = ensure_aware(datetime.fromisoformat(event_dict["timestamp"]))
+    _ = xadd_source_event(
+        event_type=etype, timestamp=etime, payload_json=evt.model_dump_json()
+    )
     logger.debug("Published mobile app usage source_event to stream")

src/clepsy/modules/aggregator/mobile_source/router.py

@@ -1,6 +1,5 @@
-from datetime import timezone
-
 from fastapi import APIRouter, HTTPException
+from loguru import logger
 
 from clepsy.entities import MobileAppUsageEvent
 from clepsy.jobs.mobile import persist_mobile_app_usage_job
@@ -12,20 +11,13 @@
 @router.post("/app-usage")
 async def receive_mobile_app_usage(event: MobileAppUsageEvent) -> dict | None:
     try:
-        # Validate schema, then dispatch actor (job performs anonymization)
-        MobileAppUsageEvent.model_validate(event.model_dump())
-
-        # Serialize timestamp as ISO8601 naive in UTC for message safety
-        ts = event.timestamp
-        if ts.tzinfo is None:
-            ts = ts.replace(tzinfo=timezone.utc)
-        ts_utc = ts.astimezone(timezone.utc)
-        payload = event.model_dump()
-        payload["timestamp"] = ts_utc.replace(tzinfo=None).isoformat()
-
+        payload = event.model_dump(mode="json")
+        logger.debug("Received mobile app usage event {}", payload)
         persist_mobile_app_usage_job.send(payload)
         return None
     except HTTPException:
+        logger.error("HTTPException receiving mobile app usage event")
         raise
     except Exception as e:
+        logger.error(f"Error receiving mobile app usage event: {e}")
         raise HTTPException(status_code=500, detail="Internal server error") from e

src/clepsy/modules/sessions/tasks.py

@@ -1127,26 +1127,7 @@ async def run_sessionization():
             logger.error(error_message)
             return None
 
-        new_activity_specs = await select_specs_with_tags_in_time_range(
-            conn,
-            start=candidate_creation_interval_start,
-            end=candidate_creation_interval_end,
-        )
-
-    logger.info(
-        "Candidate window {} -> {} | new activity specs: {} | unique activities: {}",
-        candidate_creation_interval_start,
-        candidate_creation_interval_end,
-        len(new_activity_specs),
-        len({spec.activity_id for spec in new_activity_specs}),
-    )
-
-    new_activity_specs_sorted = sorted(
-        new_activity_specs,
-        key=lambda spec: spec.end_time(horizon=candidate_creation_interval_end),
-    )
-
-    # Fetch specs from previous window's overlap region (if any)
+    # Fetch specs from previous window's overlap region FIRST (before early exit check)
     specs_not_finalized: list[DBActivitySpecWithTags] = []
     if previous_run and previous_run.overlap_start is not None:
         if previous_run.finalized_horizon is None:
@@ -1157,22 +1138,72 @@ async def run_sessionization():
             logger.error(error_message)
             return None
 
+        logger.info(
+            "Fetching overlap activities from previous window: {} -> {}",
+            previous_run.finalized_horizon,
+            previous_run.candidate_creation_end,
+        )
         async with get_db_connection() as conn:
             specs_not_finalized = await select_specs_with_tags_in_time_range(
                 conn,
                 start=previous_run.finalized_horizon,
                 end=previous_run.candidate_creation_end,
             )
+        logger.info(
+            "Overlap region: {} -> {} | overlap activity specs: {} | unique activities: {}",
+            previous_run.finalized_horizon,
+            previous_run.candidate_creation_end,
+            len(specs_not_finalized),
+            len({spec.activity_id for spec in specs_not_finalized}),
+        )
+    else:
+        logger.info(
+            "No overlap region from previous window (first run or no unfinalized activities)"
+        )
+
+    # Fetch new activities in current window
+    new_activity_specs = await select_specs_with_tags_in_time_range(
+        conn,
+        start=candidate_creation_interval_start,
+        end=candidate_creation_interval_end,
+    )
 
     logger.info(
-        "Overlap specs from previous window: {} | unique activities: {}",
-        len(specs_not_finalized),
-        len({spec.activity_id for spec in specs_not_finalized}),
+        "Current window: {} -> {} | new activity specs: {} | unique activities: {}",
+        candidate_creation_interval_start,
+        candidate_creation_interval_end,
+        len(new_activity_specs),
+        len({spec.activity_id for spec in new_activity_specs}),
     )
 
-    # EARLY EXIT: No new activities in current window
+    new_activity_specs_sorted = sorted(
+        new_activity_specs,
+        key=lambda spec: spec.end_time(horizon=candidate_creation_interval_end),
+    )
+
+    # EARLY EXIT: No new activities AND no overlap activities to process
+    if not new_activity_specs_sorted and not specs_not_finalized:
+        logger.info(
+            "No activities to process (new activities: 0, overlap activities: 0) - early exit"
+        )
+        await finalize_carry_over_sessions_and_save(
+            carry_over_candidate_session_specs=carry_over_candidate_session_specs,
+            specs_not_finalized=specs_not_finalized,
+            candidate_creation_interval_start=candidate_creation_interval_start,
+            candidate_creation_interval_end=candidate_creation_interval_end,
+            finalized_horizon=None,
+            overlap_start=None,
+            right_tail_end=None,
+        )
+        logger.info("Sessionization complete (no new or overlap activities)")
+        return
+
+    # EARLY EXIT: No new activities but overlap activities exist - process them
     if not new_activity_specs_sorted:
-        logger.info("No new activity specs in current window")
+        logger.info(
+            "No new activities in current window, but {} overlap activities need processing",
+            len(specs_not_finalized),
+        )
         await finalize_carry_over_sessions_and_save(
             carry_over_candidate_session_specs=carry_over_candidate_session_specs,
             specs_not_finalized=specs_not_finalized,
@@ -1182,9 +1213,19 @@ async def run_sessionization():
             overlap_start=None,
             right_tail_end=None,
         )
-        logger.info("Sessionization complete (no new activities)")
+        logger.info(
+            "Sessionization complete (processed {} overlap activities, no new activities)",
+            len(specs_not_finalized),
+        )
         return
 
+    logger.info(
+        "Processing sessionization | new activities: {} | overlap activities: {} | total to process: {}",
+        len(new_activity_specs_sorted),
+        len(specs_not_finalized),
+        len(new_activity_specs_sorted) + len(specs_not_finalized),
+    )
+
     # Extract islands from new activity specs
     islands = extract_valid_islands(
         specs_in_time_range=new_activity_specs_sorted,