Remove apscheduler, raw dramatiq based tick scheduling

.devcontainer/devcontainer.json

@@ -11,7 +11,9 @@
 		"dramatiq-worker",
 		"valkey-exporter",
 		"prometheus",
-		"grafana"
+		"grafana",
+		"loki",
+		"promtail"
 	],
 	"features": {
 		"ghcr.io/devcontainers/features/git:1": {},

.vscode/settings.json

@@ -24,5 +24,6 @@
         "shellscript": true,
         "python": true
     },
-    "baml.enablePlaygroundProxy": false
+    "baml.enablePlaygroundProxy": false,
+    "python.languageServer": "None"
 }

docker-compose.yml

@@ -1,6 +1,8 @@
 volumes:
   db_data_dev:
   valkey_data_dev:
+  loki_data_dev:
+  promtail_data_dev:
 
 services:
   clepsy:
@@ -16,8 +18,8 @@ services:
       VALKEY_URL: redis://valkey:6379/0
       MONITORING_ENABLED: "true"
       CLEPSY_MODE: dev
-      CLEPSY_GID : "0"
-      CLEPSY_UID : "0"
+      CLEPSY_GID: "0"
+      CLEPSY_UID: "0"
 
     volumes:
       - .:/app
@@ -27,10 +29,10 @@ services:
       - .env
     ports:
       - "8000:8000"
-
     depends_on:
       valkey:
         condition: service_healthy
+    restart: no
 
   valkey:
     image: valkey/valkey:9.0.0
@@ -108,3 +110,26 @@ services:
       prometheus:
         condition: service_started
     restart: unless-stopped
+
+  loki:
+    image: grafana/loki:latest
+    ports:
+      - "3100:3100"
+    command: -config.file=/etc/loki/local-config.yml
+    volumes:
+      - ./ops/loki-config.yml:/etc/loki/local-config.yml:ro
+      - loki_data_dev:/loki
+    restart: unless-stopped
+
+  promtail:
+    image: grafana/promtail:latest
+    volumes:
+      - /var/lib/docker/containers:/var/lib/docker/containers:ro
+      - /var/run/docker.sock:/var/run/docker.sock
+      - ./ops/promtail-config.yml:/etc/promtail/config.yml:ro
+      - promtail_data_dev:/data
+    command: -config.file=/etc/promtail/config.yml
+    depends_on:
+      loki:
+        condition: service_started
+    restart: unless-stopped

migrations/00001_first.sql

@@ -298,6 +298,25 @@ ON aggregations(start_time, end_time);
 CREATE UNIQUE INDEX IF NOT EXISTS idx_activity_events_unique
 ON activity_events(activity_id, event_time, event_type);
 
+
+CREATE TABLE scheduled_jobs (
+  id INTEGER PRIMARY KEY AUTOINCREMENT,
+  schedule_key TEXT NOT NULL UNIQUE,
+  job_type TEXT NOT NULL,
+  cron_expr TEXT,
+  next_run_at DATETIME NOT NULL,
+  enabled INTEGER NOT NULL DEFAULT 1 CHECK (enabled IN (0,1)),
+  payload TEXT,
+  running_count INTEGER NOT NULL DEFAULT 0 CHECK (running_count >= 0),
+  max_concurrent INTEGER NOT NULL DEFAULT 1 CHECK (max_concurrent >= 1),
+  last_started_at DATETIME,
+  status TEXT NOT NULL DEFAULT 'idle' CHECK (status IN ('idle','error','disabled')),
+  created_at DATETIME NOT NULL DEFAULT (datetime('now'))
+);
+CREATE INDEX idx_scheduled_jobs_next_run ON scheduled_jobs(next_run_at);
+CREATE INDEX idx_scheduled_jobs_status ON scheduled_jobs(status);
+
+
 -- +goose StatementEnd
 
 -- +goose Down
@@ -372,4 +391,9 @@ DROP TABLE IF EXISTS sessionization_run;
 DROP INDEX IF EXISTS idx_activity_events_unique;
 DROP INDEX IF EXISTS idx_aggregations_window_unique;
 
+
+DROP TABLE IF EXISTS scheduled_jobs;
+DROP index IF EXISTS idx_scheduled_jobs_next_run;
+DROP index IF EXISTS idx_scheduled_jobs_status;
+
 -- +goose StatementEnd

pyproject.toml

@@ -29,17 +29,17 @@ dependencies = [
     "baml-py==0.213.0",
     "loguru>=0.7.3",
     "argon2-cffi>=25.1.0",
-    "sqlalchemy>=2.0.43",
     "torch>=2.0.0",
     "gliner>=0.2.22",
     "valkey[libvalkey]>=6.1.1",
     "dramatiq[redis,watch]>=1.18.0",
-    "apscheduler[sqlalchemy]>=4.0.0.0a6",
     "prometheus-client>=0.20.0",
     "prometheus-fastapi-instrumentator>=7.1.0",
     "opencv-contrib-python-headless>=4.12.0.88",
     "paddleocr>=3.3.1",
     "paddlepaddle>=3.2.1",
+    "croniter>=6.0.0",
+    "aiosqlitepool>=1.0.0",
 ]
 
 [dependency-groups]

scripts/worker_entrypoint.sh

@@ -11,4 +11,4 @@ PGID="${CLEPSY_GID}"
 /app/scripts/fix_permissions.sh
 
 log "Starting Dramatiq worker..."
-exec gosu "${PUID}:${PGID}" bash -lc "mkdir -p /tmp/dramatiq-prometheus && rm -rf /tmp/dramatiq-prometheus/* && uv run dramatiq clepsy.jobs.desktop clepsy.jobs.mobile clepsy.jobs.aggregation clepsy.jobs.sessions clepsy.jobs.goals --processes 2 --threads 4"
+exec gosu "${PUID}:${PGID}" bash -lc "mkdir -p /tmp/dramatiq-prometheus && rm -rf /tmp/dramatiq-prometheus/* && uv run dramatiq clepsy.jobs.desktop clepsy.jobs.mobile clepsy.jobs.aggregation clepsy.jobs.sessions clepsy.jobs.goals clepsy.jobs.scheduler_tick clepsy.jobs.scheduled_job_dispatch --processes 4 --threads 4"

src/clepsy/aggregator_worker.py

@@ -4,7 +4,8 @@
 import asyncio
 from collections import defaultdict
 from datetime import datetime
-from typing import NamedTuple
+from typing import Any, NamedTuple
+import sqlite3
 
 import aiosqlite
 from baml_py import Collector
@@ -29,6 +30,7 @@
     select_tags,
     select_user_settings,
     update_activity,
+    update_scheduled_job_next_run_at,
 )
 import clepsy.entities as E
 from clepsy.llm import create_client_registry
@@ -37,6 +39,10 @@
 )
 from clepsy.modules.aggregator.stitching import stitch_timeline
 
+logger = logger.patch(
+    lambda record: record.update(message=f"[aggregator] {record['message']}")
+)
+
 
 async def llm_productivity_level_activity(
     activity_name: str, activity_description: str, user_settings: E.UserSettings
@@ -483,6 +489,10 @@ async def aggregator_core(
     text_model_config: E.LLMConfig,
     collector: Collector,
 ) -> E.AggregatorCoreOutput:
+    specific_aggregation_interval_seconds = int(
+        aggregation_time_span.duration.total_seconds()
+    )
+
     (
         generated_timeline_activities,
         generated_timeline_events,
@@ -494,7 +504,7 @@ async def aggregator_core(
         qc_policy=E.IsolatedTimelineQCPolicy.ALWAYS,
         max_desktop_screenshot_log_interval_seconds=config.max_desktop_screenshot_log_interval_seconds,
         max_pause_time_seconds=config.max_pause_time_seconds,
-        aggregation_interval_seconds=config.aggregation_interval_minutes * 60,
+        aggregation_interval_seconds=specific_aggregation_interval_seconds,
     )
 
     (
@@ -642,6 +652,7 @@ async def aggregator(
 async def do_aggregation(
     input_logs: list[E.AggregationInputEvent],
     aggregation_time_span: E.TimeSpan,
+    schedule_update: E.ScheduleUpdate | None = None,
 ):
     collector = Collector(name="Aggregation")
 
@@ -683,59 +694,15 @@ async def do_aggregation(
             )
         )
 
-    logger.info(
-        "[aggregator] Starting DEFERRED transaction for persisting aggregation results"
+    await persist_aggregation_results(
+        aggregation_time_span=aggregation_time_span,
+        aggregation_events_time_span=aggregation_events_time_span,
+        core_output=core_output,
+        activity_extras=activity_extras,
+        schedule_update=schedule_update,
     )
-    async with get_db_connection(
-        start_transaction=True, commit_on_exit=True, transaction_type="DEFERRED"
-    ) as conn:
-        aggregation_id = await insert_aggregation(
-            aggregation=E.Aggregation(
-                start_time=aggregation_time_span.start_time,
-                end_time=aggregation_time_span.end_time,
-                first_timestamp=aggregation_events_time_span.start_time,
-                last_timestamp=aggregation_events_time_span.end_time,
-            ),
-            conn=conn,
-        )
-
-        events_to_insert: list[E.ActivityEventInsert] = []
-        for ev in (
-            core_output.stitched_activities_events
-            + core_output.unstitched_activities_close_events
-        ):
-            events_to_insert.append(
-                E.ActivityEventInsert(
-                    activity_id=ev.activity_id,
-                    event_time=ev.event_time,
-                    event_type=ev.event_type,
-                    aggregation_id=aggregation_id,
-                    last_manual_action_time=None,
-                )
-            )
 
-        if core_output.new_activities:
-            await persist_new_activities(
-                new_activities=core_output.new_activities,
-                new_events=core_output.new_activity_events,
-                conn=conn,
-                activity_extras=activity_extras,
-                events_to_insert=events_to_insert,
-                aggregation_time_span=aggregation_time_span,
-                aggregation_id=aggregation_id,
-            )
-        elif events_to_insert:
-            await insert_activity_events(conn, events_to_insert)
-
-        if core_output.activities_to_update:
-            await asyncio.gather(
-                *(
-                    update_activity(conn, activity_id, kv_pairs)
-                    for activity_id, kv_pairs in core_output.activities_to_update
-                )
-            )
-
-    logger.info("[aggregator] DEFERRED transaction committed successfully")
+    logger.info("Aggregation results persisted")
 
     formatted_function_logs = "\n".join(
         [utils.format_function_log(x) for x in collector.logs]
@@ -747,12 +714,94 @@ async def do_aggregation(
     )
 
 
-async def do_empty_aggregation():
+async def persist_aggregation_results(
+    *,
+    aggregation_time_span: E.TimeSpan,
+    aggregation_events_time_span: E.TimeSpan,
+    core_output: Any,
+    activity_extras: list[Any],
+    schedule_update: E.ScheduleUpdate | None,
+) -> None:
+    logger.info("Starting IMMEDIATE transaction for persisting aggregation results")
+    try:
+        async with get_db_connection(
+            start_transaction=True, commit_on_exit=True, transaction_type="IMMEDIATE"
+        ) as conn:
+            aggregation_id = await insert_aggregation(
+                aggregation=E.Aggregation(
+                    start_time=aggregation_time_span.start_time,
+                    end_time=aggregation_time_span.end_time,
+                    first_timestamp=aggregation_events_time_span.start_time,
+                    last_timestamp=aggregation_events_time_span.end_time,
+                ),
+                conn=conn,
+            )
+
+            events_to_insert: list[E.ActivityEventInsert] = []
+            for ev in (
+                core_output.stitched_activities_events
+                + core_output.unstitched_activities_close_events
+            ):
+                events_to_insert.append(
+                    E.ActivityEventInsert(
+                        activity_id=ev.activity_id,
+                        event_time=ev.event_time,
+                        event_type=ev.event_type,
+                        aggregation_id=aggregation_id,
+                        last_manual_action_time=None,
+                    )
+                )
+
+            if core_output.new_activities:
+                await persist_new_activities(
+                    new_activities=core_output.new_activities,
+                    new_events=core_output.new_activity_events,
+                    conn=conn,
+                    activity_extras=activity_extras,
+                    events_to_insert=events_to_insert,
+                    aggregation_time_span=aggregation_time_span,
+                    aggregation_id=aggregation_id,
+                )
+            elif events_to_insert:
+                await insert_activity_events(conn, events_to_insert)
+
+            if core_output.activities_to_update:
+                await asyncio.gather(
+                    *(
+                        update_activity(conn, activity_id, kv_pairs)
+                        for activity_id, kv_pairs in core_output.activities_to_update
+                    )
+                )
+
+            if schedule_update:
+                await update_scheduled_job_next_run_at(
+                    conn,
+                    schedule_id=schedule_update.schedule_id,
+                    next_run_at=schedule_update.next_run_at,
+                )
+
+    except (sqlite3.OperationalError, aiosqlite.OperationalError) as exc:
+        if "database is locked" in str(exc).lower():
+            logger.warning("Database locked while persisting aggregation; will retry")
+        raise
+
+
+async def do_empty_aggregation(schedule_update: E.ScheduleUpdate | None = None):
     # Open a connection and perform both read and optional write within the same context
-    async with get_db_connection(start_transaction=False, commit_on_exit=True) as conn:
+    async with get_db_connection(
+        start_transaction=False,
+        commit_on_exit=True,
+        transaction_type="DEFERRED",
+    ) as conn:
         previous_aggregation = await select_latest_aggregation(conn)
         close_events = await get_interrupted_activity_close_events(
             conn, previous_aggregation=previous_aggregation
         )
         if close_events:
             await insert_activity_events(conn, close_events)
+        if schedule_update:
+            await update_scheduled_job_next_run_at(
+                conn,
+                schedule_id=schedule_update.schedule_id,
+                next_run_at=schedule_update.next_run_at,
+            )

src/clepsy/config.py

@@ -130,6 +130,8 @@ class Config(BaseSettings):
     source_enrollment_code_ttl: timedelta = timedelta(minutes=30)
     log_level: str | None = None
 
+    scheduled_job_timeout: timedelta = timedelta(minutes=15)
+
     max_session_gap: timedelta = timedelta(minutes=10)
     min_session_length: timedelta = timedelta(minutes=15)
     min_activities_per_session: int = 3
@@ -141,12 +143,8 @@ class Config(BaseSettings):
     gliner_pii_threshold: float = 0.5
     gliner_cache_dir: Path = cache_dir / "gliner"
     valkey_url: str
-    ap_scheduler_sqlite_db_path: Path = Path("/var/lib/clepsy/apscheduler.sqlite3")
     monitoring_enabled: bool = False
-
-    @property
-    def ap_scheduler_db_connection_string(self) -> str:
-        return f"sqlite:////{self.ap_scheduler_sqlite_db_path.as_posix()}"
+    preserve_dev_streams: bool = False
 
     @property
     def is_dev(self) -> bool:

src/clepsy/db/db.py

@@ -22,7 +22,7 @@ async def get_db_connection(
     commit_on_exit: bool = True,
     parse_decltypes: bool = True,
     busy_timeout: int = 5000,
-    transaction_type: TransactionTypes = "IMMEDIATE",
+    transaction_type: TransactionTypes = "DEFERRED",
     pragma_synchronous: str = "NORMAL",
     cache_size: int = 2000,
 ) -> AsyncGenerator[aiosqlite.Connection, None]: