fix: close leaked sqlite connections in scheduled_tasks, message_history, conversations

app/core/scheduled_tasks.py

@@ -50,14 +50,18 @@ def _init_tasks_db(self):
             """)
 
             conn.commit()
+        conn.close()
 
 
     def load_tasks(self) -> list[dict]:
         query = """SELECT name, prompt, enabled, repeat, interval_mins,
                           last_run, next_run, delivery_channel, run_count, created_at
                    FROM tasks"""
-        with sqlite3.connect(APP_DB) as conn:
+        conn = sqlite3.connect(APP_DB)
+        try:
             rows = conn.execute(query).fetchall()
+        finally:
+            conn.close()
 
         return [{"name": n, "prompt": p, "enabled": e, "repeat": rpt,
                  "interval_mins": i, "last_run": lr, "next_run": nr,
@@ -68,31 +72,38 @@ def load_tasks(self) -> list[dict]:
     def add_task(self, name: str, prompt: str, next_run: str, interval_mins: int = 1,
                  repeat: int = 0, delivery_channel: str = "telegram", enabled: int = 1):
         now = datetime.now().isoformat()
-        with sqlite3.connect(APP_DB) as conn:
+        conn = sqlite3.connect(APP_DB)
+        try:
             try:
-                conn.execute("""
-                    INSERT INTO tasks (name, prompt, interval_mins, repeat, next_run, delivery_channel, enabled, created_at)
-                    VALUES (?, ?, ?, ?, ?, ?, ?, ?)
-                """, (name, prompt, interval_mins, repeat, next_run or now, delivery_channel, enabled, now))
-                conn.commit()
+                with conn:
+                    conn.execute("""
+                        INSERT INTO tasks (name, prompt, interval_mins, repeat, next_run, delivery_channel, enabled, created_at)
+                        VALUES (?, ?, ?, ?, ?, ?, ?, ?)
+                    """, (name, prompt, interval_mins, repeat, next_run or now, delivery_channel, enabled, now))
             except sqlite3.IntegrityError:
                 raise ValueError(f"Task '{name}' already exists")
+        finally:
+            conn.close()
 
     def remove_task(self, name: str):
         with sqlite3.connect(APP_DB) as conn:
             conn.execute("DELETE FROM tasks WHERE name = ?", (name,))
             conn.commit()
+        conn.close()
 
     def update_task(self, name: str, **fields):
         if not fields:
             return
         set_clause = ", ".join(f"{col} = ?" for col in fields)
         values = list(fields.values()) + [name]
-        with sqlite3.connect(APP_DB) as conn:
-            if not conn.execute("SELECT name FROM tasks WHERE name = ?", (name,)).fetchone():
-                raise ValueError(f"Task '{name}' not found")
-            conn.execute(f"UPDATE tasks SET {set_clause} WHERE name = ?", values)
-            conn.commit()
+        conn = sqlite3.connect(APP_DB)
+        try:
+            with conn:
+                if not conn.execute("SELECT name FROM tasks WHERE name = ?", (name,)).fetchone():
+                    raise ValueError(f"Task '{name}' not found")
+                conn.execute(f"UPDATE tasks SET {set_clause} WHERE name = ?", values)
+        finally:
+            conn.close()
 
     def save_output(self, name: str, prompt: str, output: str,
                     status: str = "success", duration_secs: float = None):
@@ -102,6 +113,7 @@ def save_output(self, name: str, prompt: str, output: str,
                 VALUES (?, ?, ?, ?, ?, ?)
             """, (name, prompt, output, status, duration_secs, datetime.now().isoformat()))
             conn.commit()
+        conn.close()
 
     def get_output(self, name: str, num_entries: int = 5) -> list[dict]:
         with sqlite3.connect(APP_DB) as conn:
@@ -110,6 +122,7 @@ def get_output(self, name: str, num_entries: int = 5) -> list[dict]:
                 WHERE name = ?
                 ORDER BY id DESC LIMIT ?
             """, (name, num_entries)).fetchall()
+        conn.close()
         return [{"prompt": p, "output": o, "status": s, "duration_secs": d, "timestamp": t}
                 for p, o, s, d, t in reversed(rows)]
 
@@ -149,6 +162,7 @@ def _after_run(self, task: dict, now: datetime):
                 conn.execute("""UPDATE tasks SET last_run = ?, next_run = ?, run_count = run_count + 1
                                 WHERE name = ?""", (now.isoformat(), next_run, name))
                 conn.commit()
+            conn.close()
         else:
             self.remove_task(name)
 

app/infra/conversations.py

@@ -14,9 +14,13 @@
 @contextlib.contextmanager
 def _fk_conn(db_path: Path):
     """sqlite3 connection with FK constraints enforced."""
-    with sqlite3.connect(db_path) as conn:
+    conn = sqlite3.connect(db_path)
+    try:
         conn.execute("PRAGMA foreign_keys = ON")
-        yield conn
+        with conn:
+            yield conn
+    finally:
+        conn.close()
 
 
 class ConversationStore:

app/infra/message_history.py

@@ -44,6 +44,7 @@ def _ensure_db(self):
                     ON messages(conversation_id)
                 """)
                 conn.commit()
+            conn.close()
 
         except sqlite3.Error as e:
             log.error(f"Error creating message history database: {str(e)}")
@@ -59,13 +60,15 @@ def add_message(self, role: str, content: str, conversation_id: int = None):
                 (self.channel, role, content, timestamp, est, conversation_id),
             )
             conn.commit()
+        conn.close()
         log.info(f"Added message to history: role={role}, est_tokens={est}, content={content[:30]}...")
 
     def get_history(self, limit: int = 100) -> list[dict]:
         with sqlite3.connect(self.db_path) as conn:
             rows = conn.execute("""SELECT role, content FROM messages
                                     WHERE channel = ?
                                     ORDER BY id DESC LIMIT ?""", (self.channel, limit)).fetchall()
-            return [{"role": row[0], "content": row[1]} for row in reversed(rows)]
+        conn.close()
+        return [{"role": row[0], "content": row[1]} for row in reversed(rows)]