database/db.py

import asyncio
from datetime import datetime, timezone
import json
import sqlite3
from typing import Any, Dict, List, Optional
import uuid

from schemas.enums import AgentState, DecisionStage, MissionStatus
from schemas.models import (
    AgentMessage,
    AgentReflectionDetail,
    AgentReputationModel,
    ApprovalRequestModel,
    BlackboardEntry,
    ConversationMessageModel,
    DebateSession,
    DiscussionEntry,
    EvidenceScore,
    KnowledgeEdge,
    KnowledgeNode,
    MissionContextModel,
    NotificationModel,
    TimelineEventModel,
)


class DatabaseManager:
    """SQLite Database Engine for Spark Colony System State, Memory & Logs."""

    def __init__(self, db_path: str = "spark_colony.db"):
        self.db_path = db_path
        self._init_db()

    def _get_connection(self) -> sqlite3.Connection:
        conn = sqlite3.connect(self.db_path, check_same_thread=False)
        conn.row_factory = sqlite3.Row
        conn.execute("PRAGMA journal_mode=WAL;")
        return conn

    def _init_db(self) -> None:
        with self._get_connection() as conn:
            cursor = conn.cursor()
            cursor.execute("""
                CREATE TABLE IF NOT EXISTS missions (
                    id TEXT PRIMARY KEY,
                    topic TEXT NOT NULL,
                    status TEXT NOT NULL,
                    stage TEXT NOT NULL,
                    created_at TEXT NOT NULL,
                    updated_at TEXT NOT NULL,
                    summary TEXT,
                    total_cost REAL DEFAULT 0.0
                )
            """)
            cursor.execute("""
                CREATE TABLE IF NOT EXISTS agents (
                    id TEXT PRIMARY KEY,
                    name TEXT NOT NULL,
                    role TEXT NOT NULL,
                    state TEXT NOT NULL,
                    current_task TEXT,
                    confidence REAL NOT NULL,
                    last_active TEXT NOT NULL,
                    enabled INTEGER DEFAULT 1
                )
            """)
            cursor.execute("""
                CREATE TABLE IF NOT EXISTS memories (
                    id TEXT PRIMARY KEY,
                    agent_id TEXT NOT NULL,
                    mission_id TEXT NOT NULL,
                    content TEXT NOT NULL,
                    tags TEXT,
                    created_at TEXT NOT NULL
                )
            """)
            cursor.execute("""
                CREATE TABLE IF NOT EXISTS journals (
                    id TEXT PRIMARY KEY,
                    agent_id TEXT NOT NULL,
                    mission_id TEXT NOT NULL,
                    entry TEXT NOT NULL,
                    timestamp TEXT NOT NULL
                )
            """)
            cursor.execute("""
                CREATE TABLE IF NOT EXISTS messages (
                    id TEXT PRIMARY KEY,
                    sender TEXT NOT NULL,
                    recipient TEXT NOT NULL,
                    mission_id TEXT NOT NULL,
                    status TEXT NOT NULL,
                    summary TEXT NOT NULL,
                    next_request TEXT NOT NULL,
                    confidence REAL NOT NULL,
                    timestamp TEXT NOT NULL
                )
            """)
            cursor.execute("""
                CREATE TABLE IF NOT EXISTS evidence (
                    id TEXT PRIMARY KEY,
                    mission_id TEXT NOT NULL,
                    agent_id TEXT NOT NULL,
                    claim TEXT NOT NULL,
                    source TEXT NOT NULL,
                    credibility REAL NOT NULL,
                    freshness REAL NOT NULL,
                    authority REAL NOT NULL,
                    agreement REAL NOT NULL,
                    conflict REAL NOT NULL,
                    confidence REAL NOT NULL,
                    created_at TEXT NOT NULL
                )
            """)
            cursor.execute("""
                CREATE TABLE IF NOT EXISTS token_costs (
                    id TEXT PRIMARY KEY,
                    mission_id TEXT NOT NULL,
                    agent_id TEXT NOT NULL,
                    prompt_tokens INT DEFAULT 0,
                    completion_tokens INT DEFAULT 0,
                    reasoning_tokens INT DEFAULT 0,
                    vision_tokens INT DEFAULT 0,
                    cost_usd REAL DEFAULT 0.0,
                    timestamp TEXT NOT NULL
                )
            """)
            cursor.execute("""
                CREATE TABLE IF NOT EXISTS browser_cache (
                    id TEXT PRIMARY KEY,
                    url TEXT UNIQUE NOT NULL,
                    html TEXT NOT NULL,
                    title TEXT,
                    cached_at TEXT NOT NULL
                )
            """)
            cursor.execute("""
                CREATE TABLE IF NOT EXISTS plugins (
                    id TEXT PRIMARY KEY,
                    name TEXT UNIQUE NOT NULL,
                    version TEXT NOT NULL,
                    description TEXT NOT NULL,
                    entry_point TEXT NOT NULL,
                    permissions TEXT,
                    status TEXT NOT NULL,
                    registered_at TEXT NOT NULL
                )
            """)
            cursor.execute("""
                CREATE TABLE IF NOT EXISTS reflections (
                    id TEXT PRIMARY KEY,
                    mission_id TEXT NOT NULL,
                    agent_id TEXT NOT NULL,
                    lessons_learned TEXT NOT NULL,
                    mistakes_identified TEXT,
                    cost_usd REAL NOT NULL,
                    confidence_achieved REAL NOT NULL,
                    created_at TEXT NOT NULL,
                    FOREIGN KEY (mission_id) REFERENCES missions (id)
                )
            """)
            cursor.execute("""
                CREATE TABLE IF NOT EXISTS website_profiles (
                    domain TEXT PRIMARY KEY,
                    trust_score REAL NOT NULL,
                    authority REAL NOT NULL,
                    typical_layout TEXT NOT NULL,
                    has_captcha_history INTEGER DEFAULT 0,
                    interaction_success_rate REAL DEFAULT 100.0,
                    last_visited TEXT NOT NULL
                )
            """)
            cursor.execute("""
                CREATE TABLE IF NOT EXISTS screen_memories (
                    id TEXT PRIMARY KEY,
                    mission_id TEXT NOT NULL,
                    url TEXT NOT NULL,
                    screenshot_ref TEXT NOT NULL,
                    layout_summary TEXT NOT NULL,
                    timestamp TEXT NOT NULL
                )
            """)
            cursor.execute("""
                CREATE TABLE IF NOT EXISTS downloaded_files (
                    id TEXT PRIMARY KEY,
                    mission_id TEXT NOT NULL,
                    filename TEXT NOT NULL,
                    mime_type TEXT NOT NULL,
                    file_size INTEGER NOT NULL,
                    summary TEXT,
                    created_at TEXT NOT NULL
                )
            """)
            cursor.execute("""
                CREATE TABLE IF NOT EXISTS checkpoints (
                    id TEXT PRIMARY KEY,
                    mission_id TEXT NOT NULL,
                    stage TEXT NOT NULL,
                    data_json TEXT NOT NULL,
                    created_at TEXT NOT NULL,
                    FOREIGN KEY (mission_id) REFERENCES missions (id)
                )
            """)
            cursor.execute("""
                CREATE TABLE IF NOT EXISTS knowledge_nodes (
                    id TEXT PRIMARY KEY,
                    mission_id TEXT NOT NULL,
                    label TEXT NOT NULL,
                    entity_type TEXT NOT NULL,
                    confidence REAL NOT NULL
                )
            """)
            cursor.execute("""
                CREATE TABLE IF NOT EXISTS knowledge_edges (
                    id TEXT PRIMARY KEY,
                    mission_id TEXT NOT NULL,
                    source_node_id TEXT NOT NULL,
                    target_node_id TEXT NOT NULL,
                    relationship TEXT NOT NULL,
                    FOREIGN KEY (source_node_id) REFERENCES knowledge_nodes (id),
                    FOREIGN KEY (target_node_id) REFERENCES knowledge_nodes (id)
                )
            """)
            cursor.execute("""
                CREATE TABLE IF NOT EXISTS blackboard (
                    id TEXT PRIMARY KEY,
                    mission_id TEXT NOT NULL,
                    agent_id TEXT NOT NULL,
                    topic TEXT NOT NULL,
                    data_json TEXT NOT NULL,
                    timestamp TEXT NOT NULL
                )
            """)
            cursor.execute("""
                CREATE TABLE IF NOT EXISTS mission_contexts (
                    mission_id TEXT PRIMARY KEY,
                    topic TEXT NOT NULL,
                    priority INTEGER DEFAULT 5,
                    current_phase TEXT NOT NULL,
                    progress_percent REAL DEFAULT 0.0,
                    owner_agent TEXT NOT NULL,
                    resource_usage_json TEXT NOT NULL,
                    estimated_cost REAL DEFAULT 0.0,
                    health_status TEXT DEFAULT 'HEALTHY',
                    retry_count INTEGER DEFAULT 0,
                    checkpoint_state TEXT,
                    created_at TEXT NOT NULL
                )
            """)
            cursor.execute("""
                CREATE TABLE IF NOT EXISTS mission_timeline (
                    id TEXT PRIMARY KEY,
                    mission_id TEXT NOT NULL,
                    agent_id TEXT NOT NULL,
                    step_type TEXT NOT NULL,
                    description TEXT NOT NULL,
                    metadata_json TEXT NOT NULL,
                    timestamp TEXT NOT NULL
                )
            """)
            cursor.execute("""
                CREATE TABLE IF NOT EXISTS discussions (
                    id TEXT PRIMARY KEY,
                    mission_id TEXT NOT NULL,
                    agent_id TEXT NOT NULL,
                    agent_name TEXT NOT NULL,
                    discussion_type TEXT NOT NULL,
                    topic TEXT NOT NULL,
                    content TEXT NOT NULL,
                    evidence_ref TEXT,
                    tags_json TEXT,
                    confidence REAL NOT NULL,
                    timestamp TEXT NOT NULL
                )
            """)
            cursor.execute("""
                CREATE TABLE IF NOT EXISTS debates (
                    debate_id TEXT PRIMARY KEY,
                    mission_id TEXT NOT NULL,
                    topic TEXT NOT NULL,
                    status TEXT NOT NULL,
                    participants_json TEXT NOT NULL,
                    turns_json TEXT NOT NULL,
                    consensus_summary TEXT,
                    final_confidence REAL NOT NULL,
                    created_at TEXT NOT NULL
                )
            """)
            cursor.execute("""
                CREATE TABLE IF NOT EXISTS agent_reputations (
                    agent_id TEXT PRIMARY KEY,
                    name TEXT NOT NULL,
                    role TEXT NOT NULL,
                    experience_points INTEGER DEFAULT 100,
                    trust_score REAL DEFAULT 100.0,
                    accuracy_rate REAL DEFAULT 100.0,
                    reliability_score REAL DEFAULT 100.0,
                    speed_score REAL DEFAULT 100.0,
                    cost_efficiency REAL DEFAULT 100.0,
                    avg_confidence REAL DEFAULT 100.0,
                    success_rate REAL DEFAULT 100.0,
                    total_missions_participated INTEGER DEFAULT 0
                )
            """)
            cursor.execute("""
                CREATE TABLE IF NOT EXISTS agent_reflections_v2 (
                    reflection_id TEXT PRIMARY KEY,
                    mission_id TEXT NOT NULL,
                    agent_id TEXT NOT NULL,
                    agent_name TEXT NOT NULL,
                    what_worked TEXT NOT NULL,
                    what_failed TEXT NOT NULL,
                    what_surprised TEXT NOT NULL,
                    what_to_improve TEXT NOT NULL,
                    timestamp TEXT NOT NULL
                )
            """)
            cursor.execute("""
                CREATE TABLE IF NOT EXISTS approval_requests (
                    id TEXT PRIMARY KEY,
                    mission_id TEXT NOT NULL,
                    agent_id TEXT NOT NULL,
                    action_type TEXT NOT NULL,
                    prompt_message TEXT NOT NULL,
                    status TEXT NOT NULL,
                    input_data_json TEXT,
                    created_at TEXT NOT NULL
                )
            """)
            cursor.execute("""
                CREATE TABLE IF NOT EXISTS notifications (
                    id TEXT PRIMARY KEY,
                    level TEXT NOT NULL,
                    title TEXT NOT NULL,
                    message TEXT NOT NULL,
                    mission_id TEXT,
                    acknowledged INTEGER DEFAULT 0,
                    created_at TEXT NOT NULL
                )
            """)
            cursor.execute("""
                CREATE TABLE IF NOT EXISTS conversation_logs (
                    id TEXT PRIMARY KEY,
                    user_id TEXT NOT NULL,
                    sender TEXT NOT NULL,
                    message TEXT NOT NULL,
                    metadata_json TEXT,
                    timestamp TEXT NOT NULL
                )
            """)
            # Safe schema migration for blackboard
            try:
                cursor.execute("ALTER TABLE blackboard ADD COLUMN version INTEGER DEFAULT 1")
                cursor.execute("ALTER TABLE blackboard ADD COLUMN tags_json TEXT DEFAULT '[]'")
                cursor.execute("ALTER TABLE blackboard ADD COLUMN priority INTEGER DEFAULT 5")
                cursor.execute("ALTER TABLE blackboard ADD COLUMN confidence REAL DEFAULT 100.0")
            except sqlite3.OperationalError:
                pass  # Columns already exist
            conn.commit()

    async def save_mission(
        self,
        mission_id: str,
        topic: str,
        status: MissionStatus,
        stage: DecisionStage,
        summary: Optional[str] = None,
        total_cost: float = 0.0,
    ) -> None:
        def _exec():
            now = datetime.now(timezone.utc).isoformat()
            with self._get_connection() as conn:
                conn.execute(
                    """
                    INSERT INTO missions (id, topic, status, stage, created_at, updated_at, summary, total_cost)
                    VALUES (?, ?, ?, ?, ?, ?, ?, ?)
                    ON CONFLICT(id) DO UPDATE SET
                        status=excluded.status,
                        stage=excluded.stage,
                        updated_at=excluded.updated_at,
                        summary=COALESCE(excluded.summary, missions.summary),
                        total_cost=missions.total_cost + excluded.total_cost
                    """,
                    (mission_id, topic, status.value, stage.value, now, now, summary, total_cost),
                )
                conn.commit()

        await asyncio.to_thread(_exec)

    async def update_mission_status(self, mission_id: str, status: MissionStatus, stage: Optional[DecisionStage] = None) -> None:
        def _exec():
            now = datetime.now(timezone.utc).isoformat()
            with self._get_connection() as conn:
                if stage:
                    conn.execute("UPDATE missions SET status=?, stage=?, updated_at=? WHERE id=?", (status.value, stage.value, now, mission_id))
                else:
                    conn.execute("UPDATE missions SET status=?, updated_at=? WHERE id=?", (status.value, now, mission_id))
                conn.commit()

        await asyncio.to_thread(_exec)

    async def delete_mission(self, mission_id: str) -> None:
        def _exec():
            with self._get_connection() as conn:
                conn.execute("DELETE FROM missions WHERE id=?", (mission_id,))
                conn.execute("DELETE FROM memories WHERE mission_id=?", (mission_id,))
                conn.execute("DELETE FROM journals WHERE mission_id=?", (mission_id,))
                conn.execute("DELETE FROM messages WHERE mission_id=?", (mission_id,))
                conn.execute("DELETE FROM evidence WHERE mission_id=?", (mission_id,))
                conn.execute("DELETE FROM reflections WHERE mission_id=?", (mission_id,))
                conn.execute("DELETE FROM screen_memories WHERE mission_id=?", (mission_id,))
                conn.execute("DELETE FROM downloaded_files WHERE mission_id=?", (mission_id,))
                conn.execute("DELETE FROM checkpoints WHERE mission_id=?", (mission_id,))
                conn.execute("DELETE FROM knowledge_nodes WHERE mission_id=?", (mission_id,))
                conn.execute("DELETE FROM knowledge_edges WHERE mission_id=?", (mission_id,))
                conn.execute("DELETE FROM blackboard WHERE mission_id=?", (mission_id,))
                conn.execute("DELETE FROM mission_contexts WHERE mission_id=?", (mission_id,))
                conn.execute("DELETE FROM mission_timeline WHERE mission_id=?", (mission_id,))
                conn.execute("DELETE FROM discussions WHERE mission_id=?", (mission_id,))
                conn.execute("DELETE FROM debates WHERE mission_id=?", (mission_id,))
                conn.execute("DELETE FROM agent_reflections_v2 WHERE mission_id=?", (mission_id,))
                conn.execute("DELETE FROM approval_requests WHERE mission_id=?", (mission_id,))
                conn.execute("DELETE FROM notifications WHERE mission_id=?", (mission_id,))
                conn.commit()

        await asyncio.to_thread(_exec)

    async def get_mission(self, mission_id: str) -> Optional[Dict[str, Any]]:
        def _exec():
            with self._get_connection() as conn:
                cursor = conn.cursor()
                cursor.execute("SELECT * FROM missions WHERE id = ?", (mission_id,))
                row = cursor.fetchone()
                return dict(row) if row else None

        return await asyncio.to_thread(_exec)

    async def get_all_missions(self) -> List[Dict[str, Any]]:
        def _exec():
            with self._get_connection() as conn:
                cursor = conn.cursor()
                cursor.execute("SELECT * FROM missions ORDER BY created_at DESC")
                return [dict(row) for row in cursor.fetchall()]

        return await asyncio.to_thread(_exec)

    async def count_missions(self) -> int:
        def _exec():
            with self._get_connection() as conn:
                cursor = conn.cursor()
                cursor.execute("SELECT COUNT(*) FROM missions")
                return cursor.fetchone()[0]

        return await asyncio.to_thread(_exec)

    async def upsert_agent(
        self, agent_id: str, name: str, role: str, state: AgentState, current_task: Optional[str], confidence: float, enabled: bool = True
    ) -> None:
        def _exec():
            now = datetime.now(timezone.utc).isoformat()
            with self._get_connection() as conn:
                conn.execute(
                    """
                    INSERT INTO agents (id, name, role, state, current_task, confidence, last_active, enabled)
                    VALUES (?, ?, ?, ?, ?, ?, ?, ?)
                    ON CONFLICT(id) DO UPDATE SET
                        name=excluded.name,
                        role=excluded.role,
                        state=excluded.state,
                        current_task=excluded.current_task,
                        confidence=excluded.confidence,
                        last_active=excluded.last_active,
                        enabled=excluded.enabled
                    """,
                    (agent_id, name, role, state.value, current_task, confidence, now, 1 if enabled else 0),
                )
                conn.commit()

        await asyncio.to_thread(_exec)

    async def get_all_agents(self) -> List[Dict[str, Any]]:
        def _exec():
            with self._get_connection() as conn:
                cursor = conn.cursor()
                cursor.execute("SELECT * FROM agents")
                return [dict(row) for row in cursor.fetchall()]

        return await asyncio.to_thread(_exec)

    async def get_agent(self, agent_id: str) -> Optional[Dict[str, Any]]:
        def _exec():
            with self._get_connection() as conn:
                cursor = conn.cursor()
                cursor.execute("SELECT * FROM agents WHERE id=?", (agent_id,))
                row = cursor.fetchone()
                return dict(row) if row else None

        return await asyncio.to_thread(_exec)

    async def save_memory(self, agent_id: str, mission_id: str, content: str, tags: List[str]) -> str:
        memory_id = str(uuid.uuid4())
        tags_str = json.dumps(tags)
        now = datetime.now(timezone.utc).isoformat()

        def _exec():
            with self._get_connection() as conn:
                conn.execute(
                    "INSERT INTO memories (id, agent_id, mission_id, content, tags, created_at) VALUES (?, ?, ?, ?, ?, ?)",
                    (memory_id, agent_id, mission_id, content, tags_str, now),
                )
                conn.commit()

        await asyncio.to_thread(_exec)
        return memory_id

    async def search_memories(self, query: str, tag: Optional[str] = None) -> List[Dict[str, Any]]:
        def _exec():
            with self._get_connection() as conn:
                cursor = conn.cursor()
                if tag:
                    cursor.execute("SELECT * FROM memories WHERE content LIKE ? AND tags LIKE ?", (f"%{query}%", f"%{tag}%"))
                else:
                    cursor.execute("SELECT * FROM memories WHERE content LIKE ?", (f"%{query}%",))
                rows = cursor.fetchall()
                results = []
                for r in rows:
                    item = dict(r)
                    item["tags"] = json.loads(item["tags"]) if item["tags"] else []
                    results.append(item)
                return results

        return await asyncio.to_thread(_exec)

    async def delete_memory(self, memory_id: str) -> None:
        def _exec():
            with self._get_connection() as conn:
                conn.execute("DELETE FROM memories WHERE id=?", (memory_id,))
                conn.commit()

        await asyncio.to_thread(_exec)

    async def get_memories_for_mission(self, mission_id: str) -> List[Dict[str, Any]]:
        def _exec():
            with self._get_connection() as conn:
                cursor = conn.cursor()
                cursor.execute("SELECT * FROM memories WHERE mission_id = ? ORDER BY created_at ASC", (mission_id,))
                rows = cursor.fetchall()
                results = []
                for row in rows:
                    item = dict(row)
                    item["tags"] = json.loads(item["tags"]) if item["tags"] else []
                    results.append(item)
                return results

        return await asyncio.to_thread(_exec)

    async def count_memories(self) -> int:
        def _exec():
            with self._get_connection() as conn:
                cursor = conn.cursor()
                cursor.execute("SELECT COUNT(*) FROM memories")
                return cursor.fetchone()[0]

        return await asyncio.to_thread(_exec)

    async def save_journal(self, agent_id: str, mission_id: str, entry: str) -> str:
        journal_id = str(uuid.uuid4())
        now = datetime.now(timezone.utc).isoformat()

        def _exec():
            with self._get_connection() as conn:
                conn.execute(
                    "INSERT INTO journals (id, agent_id, mission_id, entry, timestamp) VALUES (?, ?, ?, ?, ?)",
                    (journal_id, agent_id, mission_id, entry, now),
                )
                conn.commit()

        await asyncio.to_thread(_exec)
        return journal_id

    async def get_journals_for_mission(self, mission_id: str) -> List[Dict[str, Any]]:
        def _exec():
            with self._get_connection() as conn:
                cursor = conn.cursor()
                cursor.execute("SELECT * FROM journals WHERE mission_id = ? ORDER BY timestamp ASC", (mission_id,))
                return [dict(row) for row in cursor.fetchall()]

        return await asyncio.to_thread(_exec)

    async def save_message(self, message: AgentMessage) -> None:
        def _exec():
            with self._get_connection() as conn:
                conn.execute(
                    """
                    INSERT INTO messages (id, sender, recipient, mission_id, status, summary, next_request, confidence, timestamp)
                    VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
                    """,
                    (
                        message.id,
                        message.sender,
                        message.recipient,
                        message.mission_id,
                        message.status,
                        message.summary,
                        message.next_request,
                        message.confidence,
                        message.timestamp,
                    ),
                )
                conn.commit()

        await asyncio.to_thread(_exec)

    async def get_messages_for_mission(self, mission_id: str) -> List[Dict[str, Any]]:
        def _exec():
            with self._get_connection() as conn:
                cursor = conn.cursor()
                cursor.execute("SELECT * FROM messages WHERE mission_id = ? ORDER BY timestamp ASC", (mission_id,))
                return [dict(row) for row in cursor.fetchall()]

        return await asyncio.to_thread(_exec)

    async def count_messages(self) -> int:
        def _exec():
            with self._get_connection() as conn:
                cursor = conn.cursor()
                cursor.execute("SELECT COUNT(*) FROM messages")
                return cursor.fetchone()[0]

        return await asyncio.to_thread(_exec)

    async def save_evidence(
        self,
        mission_id: str,
        agent_id: str,
        claim: str,
        source: str,
        score: EvidenceScore,
    ) -> str:
        evidence_id = str(uuid.uuid4())
        now = datetime.now(timezone.utc).isoformat()

        def _exec():
            with self._get_connection() as conn:
                conn.execute(
                    """
                    INSERT INTO evidence (id, mission_id, agent_id, claim, source, credibility, freshness, authority, agreement, conflict, confidence, created_at)
                    VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                    """,
                    (
                        evidence_id,
                        mission_id,
                        agent_id,
                        claim,
                        source,
                        score.credibility,
                        score.freshness,
                        score.authority,
                        score.agreement,
                        score.conflict,
                        score.overall_confidence,
                        now,
                    ),
                )
                conn.commit()

        await asyncio.to_thread(_exec)
        return evidence_id

    async def get_evidence_for_mission(self, mission_id: str) -> List[Dict[str, Any]]:
        def _exec():
            with self._get_connection() as conn:
                cursor = conn.cursor()
                cursor.execute("SELECT * FROM evidence WHERE mission_id = ? ORDER BY created_at ASC", (mission_id,))
                return [dict(row) for row in cursor.fetchall()]

        return await asyncio.to_thread(_exec)

    async def record_tokens(
        self,
        mission_id: str,
        agent_id: str,
        prompt_tokens: int,
        completion_tokens: int,
        reasoning_tokens: int,
        vision_tokens: int,
        cost_usd: float,
    ) -> None:
        token_id = str(uuid.uuid4())
        now = datetime.now(timezone.utc).isoformat()

        def _exec():
            with self._get_connection() as conn:
                conn.execute(
                    """
                    INSERT INTO token_costs (id, mission_id, agent_id, prompt_tokens, completion_tokens, reasoning_tokens, vision_tokens, cost_usd, timestamp)
                    VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
                    """,
                    (token_id, mission_id, agent_id, prompt_tokens, completion_tokens, reasoning_tokens, vision_tokens, cost_usd, now),
                )
                conn.commit()

        await asyncio.to_thread(_exec)

    async def get_total_system_cost(self) -> float:
        def _exec():
            with self._get_connection() as conn:
                cursor = conn.cursor()
                cursor.execute("SELECT SUM(cost_usd) FROM token_costs")
                res = cursor.fetchone()[0]
                return res if res else 0.0

        return await asyncio.to_thread(_exec)

    async def save_browser_cache(self, url: str, html: str, title: Optional[str]) -> str:
        cache_id = str(uuid.uuid4())
        now = datetime.now(timezone.utc).isoformat()

        def _exec():
            with self._get_connection() as conn:
                conn.execute(
                    """
                    INSERT INTO browser_cache (id, url, html, title, cached_at)
                    VALUES (?, ?, ?, ?, ?)
                    ON CONFLICT(url) DO UPDATE SET html=excluded.html, title=excluded.title, cached_at=excluded.cached_at
                    """,
                    (cache_id, url, html, title, now),
                )
                conn.commit()

        await asyncio.to_thread(_exec)
        return cache_id

    async def clear_browser_cache(self) -> None:
        def _exec():
            with self._get_connection() as conn:
                conn.execute("DELETE FROM browser_cache")
                conn.commit()

        await asyncio.to_thread(_exec)

    async def save_plugin(self, name: str, version: str, description: str, entry_point: str, permissions: List[str]) -> str:
        plugin_id = str(uuid.uuid4())
        now = datetime.now(timezone.utc).isoformat()
        perms_str = json.dumps(permissions)

        def _exec():
            with self._get_connection() as conn:
                conn.execute(
                    """
                    INSERT INTO plugins (id, name, version, description, entry_point, permissions, status, registered_at)
                    VALUES (?, ?, ?, ?, ?, ?, ?, ?)
                    ON CONFLICT(name) DO UPDATE SET version=excluded.version, status=excluded.status
                    """,
                    (plugin_id, name, version, description, entry_point, perms_str, "ACTIVE", now),
                )
                conn.commit()

        await asyncio.to_thread(_exec)
        return plugin_id

    async def get_all_plugins(self) -> List[Dict[str, Any]]:
        def _exec():
            with self._get_connection() as conn:
                cursor = conn.cursor()
                cursor.execute("SELECT * FROM plugins")
                rows = cursor.fetchall()
                results = []
                for r in rows:
                    item = dict(r)
                    item["permissions"] = json.loads(item["permissions"]) if item["permissions"] else []
                    results.append(item)
                return results

        return await asyncio.to_thread(_exec)

    async def save_reflection(
        self,
        mission_id: str,
        agent_id: str,
        lessons_learned: str,
        mistakes_identified: Optional[str],
        cost_usd: float,
        confidence_achieved: float,
    ) -> str:
        ref_id = str(uuid.uuid4())
        now = datetime.now(timezone.utc).isoformat()

        def _exec():
            with self._get_connection() as conn:
                conn.execute(
                    """
                    INSERT INTO reflections (id, mission_id, agent_id, lessons_learned, mistakes_identified, cost_usd, confidence_achieved, created_at)
                    VALUES (?, ?, ?, ?, ?, ?, ?, ?)
                    """,
                    (ref_id, mission_id, agent_id, lessons_learned, mistakes_identified, cost_usd, confidence_achieved, now),
                )
                conn.commit()

        await asyncio.to_thread(_exec)
        return ref_id

    async def get_reflections_for_mission(self, mission_id: str) -> List[Dict[str, Any]]:
        def _exec():
            with self._get_connection() as conn:
                cursor = conn.cursor()
                cursor.execute("SELECT * FROM reflections WHERE mission_id = ? ORDER BY created_at ASC", (mission_id,))
                return [dict(row) for row in cursor.fetchall()]

        return await asyncio.to_thread(_exec)

    async def get_all_reflections(self) -> List[Dict[str, Any]]:
        def _exec():
            with self._get_connection() as conn:
                cursor = conn.cursor()
                cursor.execute("SELECT * FROM reflections ORDER BY created_at DESC")
                return [dict(row) for row in cursor.fetchall()]

        return await asyncio.to_thread(_exec)

    async def upsert_website_profile(
        self,
        domain: str,
        trust_score: float,
        authority: float,
        typical_layout: str,
        has_captcha: bool = False,
        success_rate: float = 100.0,
    ) -> None:
        def _exec():
            now = datetime.now(timezone.utc).isoformat()
            with self._get_connection() as conn:
                conn.execute(
                    """
                    INSERT INTO website_profiles (domain, trust_score, authority, typical_layout, has_captcha_history, interaction_success_rate, last_visited)
                    VALUES (?, ?, ?, ?, ?, ?, ?)
                    ON CONFLICT(domain) DO UPDATE SET
                        trust_score=excluded.trust_score,
                        authority=excluded.authority,
                        typical_layout=excluded.typical_layout,
                        has_captcha_history=excluded.has_captcha_history,
                        interaction_success_rate=excluded.interaction_success_rate,
                        last_visited=excluded.last_visited
                    """,
                    (domain, trust_score, authority, typical_layout, 1 if has_captcha else 0, success_rate, now),
                )
                conn.commit()

        await asyncio.to_thread(_exec)

    async def get_website_profiles(self) -> List[Dict[str, Any]]:
        def _exec():
            with self._get_connection() as conn:
                cursor = conn.cursor()
                cursor.execute("SELECT * FROM website_profiles ORDER BY trust_score DESC")
                return [dict(row) for row in cursor.fetchall()]

        return await asyncio.to_thread(_exec)

    async def save_screen_memory(self, mission_id: str, url: str, screenshot_ref: str, layout_summary: str) -> str:
        mem_id = str(uuid.uuid4())
        now = datetime.now(timezone.utc).isoformat()

        def _exec():
            with self._get_connection() as conn:
                conn.execute(
                    "INSERT INTO screen_memories (id, mission_id, url, screenshot_ref, layout_summary, timestamp) VALUES (?, ?, ?, ?, ?, ?)",
                    (mem_id, mission_id, url, screenshot_ref, layout_summary, now),
                )
                conn.commit()

        await asyncio.to_thread(_exec)
        return mem_id

    async def save_downloaded_file(self, mission_id: str, filename: str, mime_type: str, file_size: int, summary: str) -> str:
        file_id = str(uuid.uuid4())
        now = datetime.now(timezone.utc).isoformat()

        def _exec():
            with self._get_connection() as conn:
                conn.execute(
                    "INSERT INTO downloaded_files (id, mission_id, filename, mime_type, file_size, summary, created_at) VALUES (?, ?, ?, ?, ?, ?, ?)",
                    (file_id, mission_id, filename, mime_type, file_size, summary, now),
                )
                conn.commit()

        await asyncio.to_thread(_exec)
        return file_id

    async def get_downloaded_files(self) -> List[Dict[str, Any]]:
        def _exec():
            with self._get_connection() as conn:
                cursor = conn.cursor()
                cursor.execute("SELECT * FROM downloaded_files ORDER BY created_at DESC")
                return [dict(row) for row in cursor.fetchall()]

        return await asyncio.to_thread(_exec)

    async def save_checkpoint(self, mission_id: str, stage: DecisionStage, data: Dict[str, Any]) -> str:
        cp_id = str(uuid.uuid4())
        now = datetime.now(timezone.utc).isoformat()
        data_str = json.dumps(data)

        def _exec():
            with self._get_connection() as conn:
                conn.execute(
                    "INSERT INTO checkpoints (id, mission_id, stage, data_json, created_at) VALUES (?, ?, ?, ?, ?)",
                    (cp_id, mission_id, stage.value, data_str, now),
                )
                conn.commit()

        await asyncio.to_thread(_exec)
        return cp_id

    async def get_latest_checkpoint(self, mission_id: str) -> Optional[Dict[str, Any]]:
        def _exec():
            with self._get_connection() as conn:
                cursor = conn.cursor()
                cursor.execute("SELECT * FROM checkpoints WHERE mission_id = ? ORDER BY created_at DESC LIMIT 1", (mission_id,))
                row = cursor.fetchone()
                if not row:
                    return None
                item = dict(row)
                item["data"] = json.loads(item["data_json"])
                return item

        return await asyncio.to_thread(_exec)

    async def get_checkpoints_for_mission(self, mission_id: str) -> List[Dict[str, Any]]:
        def _exec():
            with self._get_connection() as conn:
                cursor = conn.cursor()
                cursor.execute("SELECT * FROM checkpoints WHERE mission_id = ? ORDER BY created_at ASC", (mission_id,))
                rows = cursor.fetchall()
                results = []
                for r in rows:
                    item = dict(r)
                    item["data"] = json.loads(item["data_json"])
                    results.append(item)
                return results

        return await asyncio.to_thread(_exec)

    async def save_knowledge_node(self, node: KnowledgeNode) -> None:
        def _exec():
            with self._get_connection() as conn:
                conn.execute(
                    "INSERT INTO knowledge_nodes (id, mission_id, label, entity_type, confidence) VALUES (?, ?, ?, ?, ?)",
                    (node.id, node.mission_id, node.label, node.entity_type, node.confidence),
                )
                conn.commit()

        await asyncio.to_thread(_exec)

    async def save_knowledge_edge(self, edge: KnowledgeEdge) -> None:
        def _exec():
            with self._get_connection() as conn:
                conn.execute(
                    "INSERT INTO knowledge_edges (id, mission_id, source_node_id, target_node_id, relationship) VALUES (?, ?, ?, ?, ?)",
                    (edge.id, edge.mission_id, edge.source_node_id, edge.target_node_id, edge.relationship),
                )
                conn.commit()

        await asyncio.to_thread(_exec)

    async def get_knowledge_graph(self, mission_id: str) -> Dict[str, Any]:
        def _exec():
            with self._get_connection() as conn:
                cursor = conn.cursor()
                cursor.execute("SELECT * FROM knowledge_nodes WHERE mission_id=?", (mission_id,))
                nodes = [dict(r) for r in cursor.fetchall()]
                cursor.execute("SELECT * FROM knowledge_edges WHERE mission_id=?", (mission_id,))
                edges = [dict(r) for r in cursor.fetchall()]
                return {"mission_id": mission_id, "nodes": nodes, "edges": edges}

        return await asyncio.to_thread(_exec)

    async def save_blackboard_entry(self, entry: BlackboardEntry) -> None:
        def _exec():
            with self._get_connection() as conn:
                conn.execute(
                    "INSERT INTO blackboard (id, mission_id, agent_id, topic, data_json, timestamp) VALUES (?, ?, ?, ?, ?, ?)",
                    (entry.id, entry.mission_id, entry.agent_id, entry.topic, json.dumps(entry.data), entry.timestamp),
                )
                conn.commit()

        await asyncio.to_thread(_exec)

    async def get_blackboard_entries(self, mission_id: str, topic: Optional[str] = None) -> List[Dict[str, Any]]:
        def _exec():
            with self._get_connection() as conn:
                cursor = conn.cursor()
                if topic:
                    cursor.execute("SELECT * FROM blackboard WHERE mission_id = ? AND topic = ? ORDER BY timestamp ASC", (mission_id, topic))
                else:
                    cursor.execute("SELECT * FROM blackboard WHERE mission_id = ? ORDER BY timestamp ASC", (mission_id,))
                rows = cursor.fetchall()
                results = []
                for r in rows:
                    item = dict(r)
                    item["data"] = json.loads(item["data_json"])
                    results.append(item)
                return results

        return await asyncio.to_thread(_exec)

    async def save_mission_context(self, ctx: MissionContextModel) -> None:
        def _exec():
            with self._get_connection() as conn:
                conn.execute(
                    """
                    INSERT INTO mission_contexts 
                    (mission_id, topic, priority, current_phase, progress_percent, owner_agent, resource_usage_json, estimated_cost, health_status, retry_count, checkpoint_state, created_at)
                    VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                    ON CONFLICT(mission_id) DO UPDATE SET
                        current_phase=excluded.current_phase,
                        progress_percent=excluded.progress_percent,
                        resource_usage_json=excluded.resource_usage_json,
                        estimated_cost=excluded.estimated_cost,
                        health_status=excluded.health_status,
                        retry_count=excluded.retry_count,
                        checkpoint_state=excluded.checkpoint_state
                    """,
                    (
                        ctx.mission_id,
                        ctx.topic,
                        ctx.priority,
                        ctx.current_phase.value,
                        ctx.progress_percent,
                        ctx.owner_agent,
                        json.dumps(ctx.resource_usage),
                        ctx.estimated_cost,
                        ctx.health_status,
                        ctx.retry_count,
                        ctx.checkpoint_state,
                        ctx.created_at,
                    ),
                )
                conn.commit()

        await asyncio.to_thread(_exec)

    async def get_all_mission_contexts(self) -> List[Dict[str, Any]]:
        def _exec():
            with self._get_connection() as conn:
                cursor = conn.cursor()
                cursor.execute("SELECT * FROM mission_contexts ORDER BY created_at DESC")
                rows = cursor.fetchall()
                results = []
                for r in rows:
                    item = dict(r)
                    item["resource_usage"] = json.loads(item["resource_usage_json"])
                    results.append(item)
                return results

        return await asyncio.to_thread(_exec)

    async def save_timeline_event(self, event: TimelineEventModel) -> None:
        def _exec():
            with self._get_connection() as conn:
                conn.execute(
                    "INSERT INTO mission_timeline (id, mission_id, agent_id, step_type, description, metadata_json, timestamp) VALUES (?, ?, ?, ?, ?, ?, ?)",
                    (event.id, event.mission_id, event.agent_id, event.step_type.value, event.description, json.dumps(event.metadata), event.timestamp),
                )
                conn.commit()

        await asyncio.to_thread(_exec)

    async def get_mission_timeline(self, mission_id: str) -> List[Dict[str, Any]]:
        def _exec():
            with self._get_connection() as conn:
                cursor = conn.cursor()
                cursor.execute("SELECT * FROM mission_timeline WHERE mission_id = ? ORDER BY timestamp ASC", (mission_id,))
                rows = cursor.fetchall()
                results = []
                for r in rows:
                    item = dict(r)
                    item["metadata"] = json.loads(item["metadata_json"])
                    results.append(item)
                return results

        return await asyncio.to_thread(_exec)

    async def save_discussion_entry(self, entry: DiscussionEntry) -> None:
        def _exec():
            with self._get_connection() as conn:
                conn.execute(
                    """
                    INSERT INTO discussions (id, mission_id, agent_id, agent_name, discussion_type, topic, content, evidence_ref, tags_json, confidence, timestamp)
                    VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                    """,
                    (
                        entry.id,
                        entry.mission_id,
                        entry.agent_id,
                        entry.agent_name,
                        entry.discussion_type.value,
                        entry.topic,
                        entry.content,
                        entry.evidence_ref,
                        json.dumps(entry.tags),
                        entry.confidence,
                        entry.timestamp,
                    ),
                )
                conn.commit()

        await asyncio.to_thread(_exec)

    async def get_discussions_for_mission(self, mission_id: str, tag: Optional[str] = None) -> List[Dict[str, Any]]:
        def _exec():
            with self._get_connection() as conn:
                cursor = conn.cursor()
                if tag:
                    cursor.execute("SELECT * FROM discussions WHERE mission_id = ? AND tags_json LIKE ? ORDER BY timestamp ASC", (mission_id, f"%{tag}%"))
                else:
                    cursor.execute("SELECT * FROM discussions WHERE mission_id = ? ORDER BY timestamp ASC", (mission_id,))
                rows = cursor.fetchall()
                results = []
                for r in rows:
                    item = dict(r)
                    item["tags"] = json.loads(item["tags_json"]) if item.get("tags_json") else []
                    results.append(item)
                return results

        return await asyncio.to_thread(_exec)

    async def save_debate_session(self, debate: DebateSession) -> None:
        def _exec():
            with self._get_connection() as conn:
                turns_json = json.dumps([t.model_dump() for t in debate.turns])
                conn.execute(
                    """
                    INSERT INTO debates (debate_id, mission_id, topic, status, participants_json, turns_json, consensus_summary, final_confidence, created_at)
                    VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
                    ON CONFLICT(debate_id) DO UPDATE SET
                        status=excluded.status,
                        turns_json=excluded.turns_json,
                        consensus_summary=excluded.consensus_summary,
                        final_confidence=excluded.final_confidence
                    """,
                    (
                        debate.debate_id,
                        debate.mission_id,
                        debate.topic,
                        debate.status,
                        json.dumps(debate.participants),
                        turns_json,
                        debate.consensus_summary,
                        debate.final_confidence,
                        debate.created_at,
                    ),
                )
                conn.commit()

        await asyncio.to_thread(_exec)

    async def get_debate_session(self, debate_id: str) -> Optional[Dict[str, Any]]:
        def _exec():
            with self._get_connection() as conn:
                cursor = conn.cursor()
                cursor.execute("SELECT * FROM debates WHERE debate_id = ?", (debate_id,))
                row = cursor.fetchone()
                if not row:
                    return None
                item = dict(row)
                item["participants"] = json.loads(item["participants_json"])
                item["turns"] = json.loads(item["turns_json"])
                return item

        return await asyncio.to_thread(_exec)

    async def get_debates_for_mission(self, mission_id: str) -> List[Dict[str, Any]]:
        def _exec():
            with self._get_connection() as conn:
                cursor = conn.cursor()
                cursor.execute("SELECT * FROM debates WHERE mission_id = ? ORDER BY created_at ASC", (mission_id,))
                rows = cursor.fetchall()
                results = []
                for r in rows:
                    item = dict(r)
                    item["participants"] = json.loads(item["participants_json"])
                    item["turns"] = json.loads(item["turns_json"])
                    results.append(item)
                return results

        return await asyncio.to_thread(_exec)

    async def save_agent_reputation(self, rep: AgentReputationModel) -> None:
        def _exec():
            with self._get_connection() as conn:
                conn.execute(
                    """
                    INSERT INTO agent_reputations 
                    (agent_id, name, role, experience_points, trust_score, accuracy_rate, reliability_score, speed_score, cost_efficiency, avg_confidence, success_rate, total_missions_participated)
                    VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                    ON CONFLICT(agent_id) DO UPDATE SET
                        experience_points=excluded.experience_points,
                        trust_score=excluded.trust_score,
                        accuracy_rate=excluded.accuracy_rate,
                        reliability_score=excluded.reliability_score,
                        speed_score=excluded.speed_score,
                        cost_efficiency=excluded.cost_efficiency,
                        avg_confidence=excluded.avg_confidence,
                        success_rate=excluded.success_rate,
                        total_missions_participated=excluded.total_missions_participated
                    """,
                    (
                        rep.agent_id,
                        rep.name,
                        rep.role,
                        rep.experience_points,
                        rep.trust_score,
                        rep.accuracy_rate,
                        rep.reliability_score,
                        rep.speed_score,
                        rep.cost_efficiency,
                        rep.avg_confidence,
                        rep.success_rate,
                        rep.total_missions_participated,
                    ),
                )
                conn.commit()

        await asyncio.to_thread(_exec)

    async def get_agent_reputation(self, agent_id: str) -> Optional[Dict[str, Any]]:
        def _exec():
            with self._get_connection() as conn:
                cursor = conn.cursor()
                cursor.execute("SELECT * FROM agent_reputations WHERE agent_id = ?", (agent_id,))
                row = cursor.fetchone()
                return dict(row) if row else None

        return await asyncio.to_thread(_exec)

    async def get_all_agent_reputations(self) -> List[Dict[str, Any]]:
        def _exec():
            with self._get_connection() as conn:
                cursor = conn.cursor()
                cursor.execute("SELECT * FROM agent_reputations ORDER BY trust_score DESC")
                return [dict(r) for r in cursor.fetchall()]

        return await asyncio.to_thread(_exec)

    async def save_agent_reflection_v2(self, refl: AgentReflectionDetail) -> None:
        def _exec():
            with self._get_connection() as conn:
                conn.execute(
                    """
                    INSERT INTO agent_reflections_v2 (reflection_id, mission_id, agent_id, agent_name, what_worked, what_failed, what_surprised, what_to_improve, timestamp)
                    VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
                    """,
                    (
                        refl.reflection_id,
                        refl.mission_id,
                        refl.agent_id,
                        refl.agent_name,
                        refl.what_worked,
                        refl.what_failed,
                        refl.what_surprised,
                        refl.what_to_improve,
                        refl.timestamp,
                    ),
                )
                conn.commit()

        await asyncio.to_thread(_exec)

    async def get_agent_reflections_for_mission(self, mission_id: str) -> List[Dict[str, Any]]:
        def _exec():
            with self._get_connection() as conn:
                cursor = conn.cursor()
                cursor.execute("SELECT * FROM agent_reflections_v2 WHERE mission_id = ? ORDER BY timestamp ASC", (mission_id,))
                return [dict(r) for r in cursor.fetchall()]

        return await asyncio.to_thread(_exec)

    async def save_approval_request(self, req: ApprovalRequestModel) -> None:
        def _exec():
            with self._get_connection() as conn:
                conn.execute(
                    """
                    INSERT INTO approval_requests (id, mission_id, agent_id, action_type, prompt_message, status, input_data_json, created_at)
                    VALUES (?, ?, ?, ?, ?, ?, ?, ?)
                    ON CONFLICT(id) DO UPDATE SET status=excluded.status, input_data_json=excluded.input_data_json
                    """,
                    (req.id, req.mission_id, req.agent_id, req.action_type.value, req.prompt_message, req.status.value, json.dumps(req.input_data), req.created_at),
                )
                conn.commit()

        await asyncio.to_thread(_exec)

    async def get_approval_request(self, approval_id: str) -> Optional[Dict[str, Any]]:
        def _exec():
            with self._get_connection() as conn:
                cursor = conn.cursor()
                cursor.execute("SELECT * FROM approval_requests WHERE id = ?", (approval_id,))
                row = cursor.fetchone()
                if not row:
                    return None
                item = dict(row)
                item["input_data"] = json.loads(item["input_data_json"]) if item.get("input_data_json") else {}
                return item

        return await asyncio.to_thread(_exec)

    async def get_pending_approvals(self) -> List[Dict[str, Any]]:
        def _exec():
            with self._get_connection() as conn:
                cursor = conn.cursor()
                cursor.execute("SELECT * FROM approval_requests WHERE status = 'PENDING' ORDER BY created_at ASC")
                rows = cursor.fetchall()
                results = []
                for r in rows:
                    item = dict(r)
                    item["input_data"] = json.loads(item["input_data_json"]) if item.get("input_data_json") else {}
                    results.append(item)
                return results

        return await asyncio.to_thread(_exec)

    async def save_notification(self, notif: NotificationModel) -> None:
        def _exec():
            with self._get_connection() as conn:
                conn.execute(
                    "INSERT INTO notifications (id, level, title, message, mission_id, acknowledged, created_at) VALUES (?, ?, ?, ?, ?, ?, ?)",
                    (notif.id, notif.level.value, notif.title, notif.message, notif.mission_id, 1 if notif.acknowledged else 0, notif.created_at),
                )
                conn.commit()

        await asyncio.to_thread(_exec)

    async def get_unacknowledged_notifications(self) -> List[Dict[str, Any]]:
        def _exec():
            with self._get_connection() as conn:
                cursor = conn.cursor()
                cursor.execute("SELECT * FROM notifications WHERE acknowledged = 0 ORDER BY created_at DESC")
                return [dict(r) for r in cursor.fetchall()]

        return await asyncio.to_thread(_exec)

    async def acknowledge_notification(self, notif_id: str) -> None:
        def _exec():
            with self._get_connection() as conn:
                conn.execute("UPDATE notifications SET acknowledged = 1 WHERE id = ?", (notif_id,))
                conn.commit()

        await asyncio.to_thread(_exec)

    async def save_conversation_log(self, msg: ConversationMessageModel) -> None:
        def _exec():
            with self._get_connection() as conn:
                conn.execute(
                    "INSERT INTO conversation_logs (id, user_id, sender, message, metadata_json, timestamp) VALUES (?, ?, ?, ?, ?, ?)",
                    (msg.id, msg.user_id, msg.sender, msg.message, json.dumps(msg.metadata), msg.timestamp),
                )
                conn.commit()

        await asyncio.to_thread(_exec)

    async def get_conversation_history(self, user_id: str = "human-operator", limit: int = 50) -> List[Dict[str, Any]]:
        def _exec():
            with self._get_connection() as conn:
                cursor = conn.cursor()
                cursor.execute("SELECT * FROM conversation_logs WHERE user_id = ? ORDER BY timestamp ASC LIMIT ?", (user_id, limit))
                rows = cursor.fetchall()
                results = []
                for r in rows:
                    item = dict(r)
                    item["metadata"] = json.loads(item["metadata_json"]) if item.get("metadata_json") else {}
                    results.append(item)
                return results

        return await asyncio.to_thread(_exec)