From 72936b389707b9ffd9f5083d75af32833bb4c8f4 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 24 Jun 2026 17:18:29 +0000 Subject: [PATCH 1/2] fix: close leaked sqlite connections in scheduled_tasks, message_history, conversations with sqlite3.connect(...) as conn only commits/rolls back the transaction on exit, it never closes the connection or releases its file descriptor. ScheduledTasks.run() opens fresh connections every 30s forever, so over enough uptime these leak until the process hits its fd ulimit, causing "unable to open database file" everywhere (scheduled tasks, /list, and likely the Telegram poller's sockets too). Co-Authored-By: Claude Sonnet 4.6 Claude-Session: https://claude.ai/code/session_01UfuBc1EqfevpENRBAmTMrA --- app/core/scheduled_tasks.py | 40 +++++++++++++++++++++++------------- app/infra/conversations.py | 10 ++++++--- app/infra/message_history.py | 5 ++++- 3 files changed, 37 insertions(+), 18 deletions(-) diff --git a/app/core/scheduled_tasks.py b/app/core/scheduled_tasks.py index cb8dff2..d83cc73 100755 --- a/app/core/scheduled_tasks.py +++ b/app/core/scheduled_tasks.py @@ -50,6 +50,7 @@ def _init_tasks_db(self): """) conn.commit() + conn.close() def load_tasks(self) -> list[dict]: @@ -58,6 +59,7 @@ def load_tasks(self) -> list[dict]: FROM tasks""" with sqlite3.connect(APP_DB) as conn: rows = conn.execute(query).fetchall() + conn.close() return [{"name": n, "prompt": p, "enabled": e, "repeat": rpt, "interval_mins": i, "last_run": lr, "next_run": nr, @@ -68,31 +70,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: - 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() - except sqlite3.IntegrityError: - raise ValueError(f"Task '{name}' already exists") + try: + with sqlite3.connect(APP_DB) as conn: + 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() + 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() + try: + 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() + finally: + conn.close() def save_output(self, name: str, prompt: str, output: str, status: str = "success", duration_secs: float = None): @@ -102,6 +111,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 +120,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 +160,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) diff --git a/app/infra/conversations.py b/app/infra/conversations.py index b002190..bedca85 100644 --- a/app/infra/conversations.py +++ b/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.execute("PRAGMA foreign_keys = ON") - yield conn + conn = sqlite3.connect(db_path) + conn.execute("PRAGMA foreign_keys = ON") + try: + with conn: + yield conn + finally: + conn.close() class ConversationStore: diff --git a/app/infra/message_history.py b/app/infra/message_history.py index 9ab15ca..321af9d 100755 --- a/app/infra/message_history.py +++ b/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,6 +60,7 @@ 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]: @@ -66,6 +68,7 @@ def get_history(self, limit: int = 100) -> list[dict]: 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)] From 5fd9cab9bc0fcd3a01210bf771af1105b4f457d6 Mon Sep 17 00:00:00 2001 From: Rikul Patel Date: Thu, 25 Jun 2026 13:59:15 -0500 Subject: [PATCH 2/2] Apply suggestions from code review Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> --- app/core/scheduled_tasks.py | 20 +++++++++++--------- app/infra/conversations.py | 2 +- 2 files changed, 12 insertions(+), 10 deletions(-) diff --git a/app/core/scheduled_tasks.py b/app/core/scheduled_tasks.py index d83cc73..ff7c4f5 100755 --- a/app/core/scheduled_tasks.py +++ b/app/core/scheduled_tasks.py @@ -57,9 +57,11 @@ 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() - conn.close() + finally: + conn.close() return [{"name": n, "prompt": p, "enabled": e, "repeat": rpt, "interval_mins": i, "last_run": lr, "next_run": nr, @@ -70,16 +72,16 @@ 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() + conn = sqlite3.connect(APP_DB) try: - with sqlite3.connect(APP_DB) as conn: - try: + try: + 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)) - conn.commit() - except sqlite3.IntegrityError: - raise ValueError(f"Task '{name}' already exists") + except sqlite3.IntegrityError: + raise ValueError(f"Task '{name}' already exists") finally: conn.close() @@ -94,12 +96,12 @@ def update_task(self, name: str, **fields): return set_clause = ", ".join(f"{col} = ?" for col in fields) values = list(fields.values()) + [name] + conn = sqlite3.connect(APP_DB) try: - with sqlite3.connect(APP_DB) as conn: + 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) - conn.commit() finally: conn.close() diff --git a/app/infra/conversations.py b/app/infra/conversations.py index bedca85..38cc5c8 100644 --- a/app/infra/conversations.py +++ b/app/infra/conversations.py @@ -15,8 +15,8 @@ def _fk_conn(db_path: Path): """sqlite3 connection with FK constraints enforced.""" conn = sqlite3.connect(db_path) - conn.execute("PRAGMA foreign_keys = ON") try: + conn.execute("PRAGMA foreign_keys = ON") with conn: yield conn finally: