diff --git a/baml_src/aggregation.baml b/baml_src/aggregation.baml index a969bb1..49ebeda 100755 --- a/baml_src/aggregation.baml +++ b/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") diff --git a/docker-compose.yml b/docker-compose.yml index ba8ad98..5bb894c 100755 --- a/docker-compose.yml +++ b/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 diff --git a/src/clepsy/auth/auth_middleware.py b/src/clepsy/auth/auth_middleware.py index 7ed3cfe..58c5aa2 100755 --- a/src/clepsy/auth/auth_middleware.py +++ b/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 diff --git a/src/clepsy/jobs/mobile.py b/src/clepsy/jobs/mobile.py index f79203f..bbd16a1 100755 --- a/src/clepsy/jobs/mobile.py +++ b/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") diff --git a/src/clepsy/modules/aggregator/mobile_source/router.py b/src/clepsy/modules/aggregator/mobile_source/router.py index 5057da9..8b2d1cb 100755 --- a/src/clepsy/modules/aggregator/mobile_source/router.py +++ b/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 diff --git a/src/clepsy/modules/sessions/tasks.py b/src/clepsy/modules/sessions/tasks.py index ed559e2..91d6fcf 100755 --- a/src/clepsy/modules/sessions/tasks.py +++ b/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,