diff --git a/app/config.py b/app/config.py index ed9f4d8..0f26206 100644 --- a/app/config.py +++ b/app/config.py @@ -1,6 +1,7 @@ from __future__ import annotations import os +import sqlite3 import tomllib from pathlib import Path @@ -12,6 +13,21 @@ APP_DB = PROJECT_HOME / "app.db" +def get_db_connection(db_path: Path = APP_DB, *, timeout: float = 30.0, + isolation_level: str | None = "") -> sqlite3.Connection: + """Open a connection to a shared SQLite db (e.g. APP_DB) with settings safe + for concurrent access from multiple asyncio tasks/channels. + + The timeout is passed straight to sqlite3.connect, which sets the + connection's busy timeout so writers retry instead of immediately raising + "database is locked" when another connection briefly holds the write lock. + WAL mode (readers don't block the writer) is persisted in the db file + itself, so it only needs to be enabled once by each store's schema-init + step rather than on every connection. + """ + return sqlite3.connect(db_path, timeout=timeout, isolation_level=isolation_level) + + def load(path: Path | str = HOME_CONFIG_PATH) -> None: global _config diff --git a/app/core/scheduled_tasks.py b/app/core/scheduled_tasks.py index ff7c4f5..020e4be 100755 --- a/app/core/scheduled_tasks.py +++ b/app/core/scheduled_tasks.py @@ -2,7 +2,7 @@ from datetime import datetime, timedelta from ..infra.app_logging import log -from ..config import APP_DB +from ..config import APP_DB, get_db_connection import sqlite3 import asyncio from .helper_agent import HelperAgent @@ -19,45 +19,46 @@ def __init__(self, mqs: dict = None, channels: dict = None): self._init_tasks_db() def _init_tasks_db(self): - with sqlite3.connect(APP_DB) as conn: - - conn.execute(""" - CREATE TABLE IF NOT EXISTS tasks ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - name TEXT NOT NULL UNIQUE, - prompt TEXT NOT NULL, - enabled INTEGER NOT NULL DEFAULT 1, - repeat INTEGER NOT NULL DEFAULT 0, - interval_mins INTEGER NOT NULL DEFAULT 1, - last_run TEXT, - next_run TEXT NOT NULL, - delivery_channel TEXT NOT NULL DEFAULT 'telegram', - run_count INTEGER NOT NULL DEFAULT 0, - created_at TEXT NOT NULL - ) - """) - - conn.execute(""" - CREATE TABLE IF NOT EXISTS task_outputs ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - name TEXT NOT NULL, - prompt TEXT NOT NULL, - output TEXT NOT NULL, - status TEXT NOT NULL DEFAULT 'success', - duration_secs REAL, - timestamp TEXT NOT NULL - ) - """) - - conn.commit() - conn.close() + conn = get_db_connection() + try: + conn.execute("PRAGMA journal_mode=WAL") + with conn: + conn.execute(""" + CREATE TABLE IF NOT EXISTS tasks ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + name TEXT NOT NULL UNIQUE, + prompt TEXT NOT NULL, + enabled INTEGER NOT NULL DEFAULT 1, + repeat INTEGER NOT NULL DEFAULT 0, + interval_mins INTEGER NOT NULL DEFAULT 1, + last_run TEXT, + next_run TEXT NOT NULL, + delivery_channel TEXT NOT NULL DEFAULT 'telegram', + run_count INTEGER NOT NULL DEFAULT 0, + created_at TEXT NOT NULL + ) + """) + + conn.execute(""" + CREATE TABLE IF NOT EXISTS task_outputs ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + name TEXT NOT NULL, + prompt TEXT NOT NULL, + output TEXT NOT NULL, + status TEXT NOT NULL DEFAULT 'success', + duration_secs REAL, + timestamp TEXT NOT NULL + ) + """) + finally: + 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""" - conn = sqlite3.connect(APP_DB) + conn = get_db_connection() try: rows = conn.execute(query).fetchall() finally: @@ -72,7 +73,7 @@ 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) + conn = get_db_connection() try: try: with conn: @@ -86,17 +87,19 @@ def add_task(self, name: str, prompt: str, next_run: str, interval_mins: int = 1 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() + conn = get_db_connection() + try: + with conn: + conn.execute("DELETE FROM tasks WHERE name = ?", (name,)) + finally: + 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] - conn = sqlite3.connect(APP_DB) + conn = get_db_connection() try: with conn: if not conn.execute("SELECT name FROM tasks WHERE name = ?", (name,)).fetchone(): @@ -107,22 +110,26 @@ def update_task(self, name: str, **fields): def save_output(self, name: str, prompt: str, output: str, status: str = "success", duration_secs: float = None): - with sqlite3.connect(APP_DB) as conn: - conn.execute(""" - INSERT INTO task_outputs (name, prompt, output, status, duration_secs, timestamp) - VALUES (?, ?, ?, ?, ?, ?) - """, (name, prompt, output, status, duration_secs, datetime.now().isoformat())) - conn.commit() - conn.close() + conn = get_db_connection() + try: + with conn: + conn.execute(""" + INSERT INTO task_outputs (name, prompt, output, status, duration_secs, timestamp) + VALUES (?, ?, ?, ?, ?, ?) + """, (name, prompt, output, status, duration_secs, datetime.now().isoformat())) + finally: + conn.close() def get_output(self, name: str, num_entries: int = 5) -> list[dict]: - with sqlite3.connect(APP_DB) as conn: + conn = get_db_connection() + try: rows = conn.execute(""" SELECT prompt, output, status, duration_secs, timestamp FROM task_outputs WHERE name = ? ORDER BY id DESC LIMIT ? """, (name, num_entries)).fetchall() - conn.close() + finally: + conn.close() return [{"prompt": p, "output": o, "status": s, "duration_secs": d, "timestamp": t} for p, o, s, d, t in reversed(rows)] @@ -158,11 +165,13 @@ def _after_run(self, task: dict, now: datetime): intervals_passed = int(elapsed_secs // interval.total_seconds()) next_due = original_next + (intervals_passed + 1) * interval next_run = next_due.isoformat() - with sqlite3.connect(APP_DB) as conn: - 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() + conn = get_db_connection() + try: + with conn: + conn.execute("""UPDATE tasks SET last_run = ?, next_run = ?, run_count = run_count + 1 + WHERE name = ?""", (now.isoformat(), next_run, name)) + finally: + conn.close() else: self.remove_task(name) diff --git a/app/infra/conversations.py b/app/infra/conversations.py index 38cc5c8..3e0a9c1 100644 --- a/app/infra/conversations.py +++ b/app/infra/conversations.py @@ -7,14 +7,14 @@ from datetime import datetime from pathlib import Path -from ..config import APP_DB, PROJECT_HOME +from ..config import APP_DB, PROJECT_HOME, get_db_connection from .app_logging import log @contextlib.contextmanager def _fk_conn(db_path: Path): """sqlite3 connection with FK constraints enforced.""" - conn = sqlite3.connect(db_path) + conn = get_db_connection(db_path) try: conn.execute("PRAGMA foreign_keys = ON") with conn: @@ -35,6 +35,7 @@ def _ensure_schema(self) -> None: # Phase 1: DDL — idempotent CREATE TABLE/INDEX and optional ALTER TABLE with _fk_conn(self.db_path) as conn: + conn.execute("PRAGMA journal_mode=WAL") conn.execute(""" CREATE TABLE IF NOT EXISTS conversations ( id INTEGER PRIMARY KEY AUTOINCREMENT, @@ -79,7 +80,7 @@ def _ensure_schema(self) -> None: # Phase 2: Migration — BEGIN IMMEDIATE for concurrency safety. # The WHERE conversation_id IS NULL guard makes this re-entrant. - mconn = sqlite3.connect(self.db_path, isolation_level=None) + mconn = get_db_connection(self.db_path, isolation_level=None) try: mconn.execute("PRAGMA foreign_keys = ON") mconn.execute("BEGIN IMMEDIATE") diff --git a/app/infra/message_history.py b/app/infra/message_history.py index 321af9d..0c63715 100755 --- a/app/infra/message_history.py +++ b/app/infra/message_history.py @@ -4,7 +4,7 @@ from datetime import datetime from pathlib import Path from .app_logging import log -from ..config import APP_DB +from ..config import APP_DB, get_db_connection def _est_tokens(content: str) -> int: return max(1, len(content) // 4) @@ -18,33 +18,36 @@ def __init__(self, channel_type: str, db_path: Path = APP_DB): def _ensure_db(self): try: self.db_path.parent.mkdir(parents=True, exist_ok=True) - with sqlite3.connect(self.db_path) as conn: - conn.execute(""" - CREATE TABLE IF NOT EXISTS messages ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - channel text NOT NULL, - role TEXT NOT NULL, - content TEXT NOT NULL, - timestamp TEXT NOT NULL, - est_tokens INTEGER - ) - """) - conn.execute(""" - CREATE INDEX IF NOT EXISTS idx_messages_channel ON messages(channel) - """) - # Add conversation_id if missing (ConversationStore adds it too, - # but MessageHistory must be self-consistent for standalone use) - cols = {row[1] for row in conn.execute("PRAGMA table_info(messages)")} - if "conversation_id" not in cols: - conn.execute( - "ALTER TABLE messages ADD COLUMN conversation_id INTEGER" - ) - conn.execute(""" - CREATE INDEX IF NOT EXISTS idx_messages_conversation_id - ON messages(conversation_id) - """) - conn.commit() - conn.close() + conn = get_db_connection(self.db_path) + try: + conn.execute("PRAGMA journal_mode=WAL") + with conn: + conn.execute(""" + CREATE TABLE IF NOT EXISTS messages ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + channel text NOT NULL, + role TEXT NOT NULL, + content TEXT NOT NULL, + timestamp TEXT NOT NULL, + est_tokens INTEGER + ) + """) + conn.execute(""" + CREATE INDEX IF NOT EXISTS idx_messages_channel ON messages(channel) + """) + # Add conversation_id if missing (ConversationStore adds it too, + # but MessageHistory must be self-consistent for standalone use) + cols = {row[1] for row in conn.execute("PRAGMA table_info(messages)")} + if "conversation_id" not in cols: + conn.execute( + "ALTER TABLE messages ADD COLUMN conversation_id INTEGER" + ) + conn.execute(""" + CREATE INDEX IF NOT EXISTS idx_messages_conversation_id + ON messages(conversation_id) + """) + finally: + conn.close() except sqlite3.Error as e: log.error(f"Error creating message history database: {str(e)}") @@ -53,22 +56,26 @@ def _ensure_db(self): def add_message(self, role: str, content: str, conversation_id: int = None): timestamp = datetime.now().isoformat() est = _est_tokens(content) - with sqlite3.connect(self.db_path) as conn: - conn.execute( - "INSERT INTO messages (channel, role, content, timestamp, est_tokens, conversation_id) " - "VALUES (?, ?, ?, ?, ?, ?)", - (self.channel, role, content, timestamp, est, conversation_id), - ) - conn.commit() - conn.close() + conn = get_db_connection(self.db_path) + try: + with conn: + conn.execute( + "INSERT INTO messages (channel, role, content, timestamp, est_tokens, conversation_id) " + "VALUES (?, ?, ?, ?, ?, ?)", + (self.channel, role, content, timestamp, est, conversation_id), + ) + finally: + 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: + conn = get_db_connection(self.db_path) + try: rows = conn.execute("""SELECT role, content FROM messages WHERE channel = ? ORDER BY id DESC LIMIT ?""", (self.channel, limit)).fetchall() - conn.close() + finally: + conn.close() return [{"role": row[0], "content": row[1]} for row in reversed(rows)]