void create_digest_persist_table();
void flush_digest_to_sqlite();
void load_persisted_digests();

void Discovery_Schema::create_digest_persist_table() {
	if (!vector_db) return;
	const char* ddl =
		"CREATE TABLE IF NOT EXISTS mcp_query_digest_persist ("
		"tool_name VARCHAR NOT NULL,"
		"run_id INTEGER NOT NULL,"
		"digest VARCHAR NOT NULL,"
		"digest_text VARCHAR NOT NULL,"
		"count_star INTEGER NOT NULL,"
		"first_seen INTEGER NOT NULL,"
		"last_seen INTEGER NOT NULL,"
		"sum_time INTEGER NOT NULL,"
		"min_time INTEGER NOT NULL,"
		"max_time INTEGER NOT NULL,"
		"PRIMARY KEY(tool_name, run_id, digest))";
	vector_db->execute(ddl);
}

void Discovery_Schema::flush_digest_to_sqlite() {
	if (!digest_persist_db) return;

	pthread_rwlock_rdlock(&mcp_digest_rwlock);

	digest_persist_db->execute("BEGIN IMMEDIATE");
	for (auto const& [key1, inner_map] : mcp_digest_umap) {
		for (auto const& [digest, stats_ptr] : inner_map) {
			MCP_Query_Digest_Stats* stats = (MCP_Query_Digest_Stats*)stats_ptr;
			char* sql = sqlite3_mprintf(
				"INSERT OR REPLACE INTO mcp_query_digest_persist "
				"(tool_name, run_id, digest, digest_text, count_star, "
				"first_seen, last_seen, sum_time, min_time, max_time) "
				"VALUES (%Q, %d, '%llu', %Q, %llu, %ld, %ld, %llu, %llu, %llu)",
				stats->tool_name.c_str(), stats->run_id,
				(unsigned long long)stats->digest, stats->digest_text.c_str(),
				(unsigned long long)stats->count_star,
				(long)stats->first_seen, (long)stats->last_seen,
				(unsigned long long)stats->sum_time,
				(unsigned long long)stats->min_time,
				(unsigned long long)stats->max_time);
			if (sql) {
				digest_persist_db->execute(sql);
				sqlite3_free(sql);
			}
		}
	}
	digest_persist_db->execute("COMMIT");

	pthread_rwlock_unlock(&mcp_digest_rwlock);
}

void Discovery_Schema::load_persisted_digests() {
	if (!digest_persist_db) return;

	char* errmsg = NULL;
	sqlite3_stmt* stmt = NULL;
	int rc = sqlite3_prepare_v2(digest_persist_db->get_db(),
		"SELECT tool_name, run_id, digest, digest_text, count_star, "
		"first_seen, last_seen, sum_time, min_time, max_time "
		"FROM mcp_query_digest_persist", -1, &stmt, NULL);
	if (rc != SQLITE_OK) return;

	pthread_rwlock_wrlock(&mcp_digest_rwlock);

	while (sqlite3_step(stmt) == SQLITE_ROW) {
		std::string tool_name = (const char*)sqlite3_column_text(stmt, 0);
		int run_id = sqlite3_column_int(stmt, 1);
		uint64_t digest = (uint64_t)sqlite3_column_int64(stmt, 2);
		std::string digest_text = (const char*)sqlite3_column_text(stmt, 3);

		std::string key = tool_name + "|" + std::to_string(run_id);
		auto& tool_map = mcp_digest_umap[key];

		MCP_Query_Digest_Stats* stats = new MCP_Query_Digest_Stats();
		stats->tool_name = tool_name;
		stats->run_id = run_id;
		stats->digest = digest;
		stats->digest_text = digest_text;
		stats->count_star = (unsigned long long)sqlite3_column_int64(stmt, 4);
		stats->first_seen = (time_t)sqlite3_column_int64(stmt, 5);
		stats->last_seen = (time_t)sqlite3_column_int64(stmt, 6);
		stats->sum_time = (unsigned long long)sqlite3_column_int64(stmt, 7);
		stats->min_time = (unsigned long long)sqlite3_column_int64(stmt, 8);
		stats->max_time = (unsigned long long)sqlite3_column_int64(stmt, 9);

		tool_map[digest] = stats;
	}

	pthread_rwlock_unlock(&mcp_digest_rwlock);
	sqlite3_finalize(stmt);
}

	if (++update_count % 100 == 0) {
		flush_digest_to_sqlite();
	}

	if (digest_persist_db) {
		digest_persist_db->execute("DELETE FROM mcp_query_digest_persist");
	}

	ds->create_digest_persist_table();
	ds->load_persisted_digests();

git add plugins/genai/include/Discovery_Schema.h plugins/genai/src/Discovery_Schema.cpp
git commit -m "feat(genai): persist MCP query digest stats to SQLite (Feature E)"

std::vector<std::pair<std::string, std::string>> collect_status_variables();

std::vector<std::pair<std::string, std::string>> GenAI_Threads_Handler::collect_status_variables() {
	std::vector<std::pair<std::string, std::string>> vars;
	vars.push_back({"genai_threads_initialized", std::to_string(status_variables.threads_initialized)});
	vars.push_back({"genai_active_requests", std::to_string(status_variables.active_requests)});
	vars.push_back({"genai_completed_requests", std::to_string(status_variables.completed_requests)});
	vars.push_back({"genai_failed_requests", std::to_string(status_variables.failed_requests)});
	return vars;
}

std::vector<std::pair<std::string, std::string>> collect_status_variables();

std::vector<std::pair<std::string, std::string>> MCP_Threads_Handler::collect_status_variables() {
	std::vector<std::pair<std::string, std::string>> vars;
	vars.push_back({"mcp_total_requests", std::to_string(status_variables.total_requests)});
	vars.push_back({"mcp_failed_requests", std::to_string(status_variables.failed_requests)});
	vars.push_back({"mcp_active_connections", std::to_string(status_variables.active_connections)});
	return vars;
}

std::vector<std::pair<std::string, std::string>> collect_status_variables();

std::vector<std::pair<std::string, std::string>> AI_Features_Manager::collect_status_variables() {
	std::vector<std::pair<std::string, std::string>> vars;
	vars.push_back({"llm_total_requests", std::to_string(status_variables.llm_total_requests)});
	vars.push_back({"llm_cache_hits", std::to_string(status_variables.llm_cache_hits)});
	vars.push_back({"llm_cache_misses", std::to_string(status_variables.llm_cache_misses)});
	vars.push_back({"llm_local_model_calls", std::to_string(status_variables.llm_local_model_calls)});
	vars.push_back({"llm_cloud_model_calls", std::to_string(status_variables.llm_cloud_model_calls)});
	vars.push_back({"llm_total_response_time_ms", std::to_string(status_variables.llm_total_response_time_ms)});
	vars.push_back({"llm_cache_total_lookup_time_ms", std::to_string(status_variables.llm_cache_total_lookup_time_ms)});
	vars.push_back({"llm_cache_total_store_time_ms", std::to_string(status_variables.llm_cache_total_store_time_ms)});
	vars.push_back({"llm_cache_lookups", std::to_string(status_variables.llm_cache_lookups)});
	vars.push_back({"llm_cache_stores", std::to_string(status_variables.llm_cache_stores)});
	vars.push_back({"anomaly_total_checks", std::to_string(status_variables.anomaly_total_checks)});
	vars.push_back({"anomaly_blocked_queries", std::to_string(status_variables.anomaly_blocked_queries)});
	vars.push_back({"anomaly_flagged_queries", std::to_string(status_variables.anomaly_flagged_queries)});
	vars.push_back({"daily_cloud_spend_usd", std::to_string(status_variables.daily_cloud_spend_usd)});
	return vars;
}

git add plugins/genai/include/GenAI_Thread.h plugins/genai/src/GenAI_Thread.cpp \
        plugins/genai/include/MCP_Thread.h plugins/genai/src/MCP_Thread.cpp \
        plugins/genai/include/AI_Features_Manager.h plugins/genai/src/AI_Features_Manager.cpp
git commit -m "feat(genai): add collect_status_variables() to genai/mcp/ai handlers (Feature C)"

static constexpr const char* kStatsGenaiGlobal =
	"CREATE TABLE stats_genai_global ("
	"Variable_name VARCHAR NOT NULL, "
	"Value VARCHAR NOT NULL, "
	"PRIMARY KEY (Variable_name))";

static int refresh_stats_genai_global(ProxySQL_PluginServices* services,
                                       const char* table_name,
                                       SQLite3_result** result,
                                       void* context) {
	*result = new SQLite3_result(2);
	(*result)->add_column_definition(SQLITE_TEXT, "Variable_name");
	(*result)->add_column_definition(SQLITE_TEXT, "Value");

	// Collect from all three handlers
	std::vector<std::pair<std::string, std::string>> all_vars;

	if (GloGATH) {
		auto vars = GloGATH->collect_status_variables();
		all_vars.insert(all_vars.end(), vars.begin(), vars.end());
	}
	// GloMCPH: get from genai_context().mcp (see plugin_main.cpp)
	extern GenAIPluginContext& genai_context();
	if (genai_context().mcp) {
		auto vars = genai_context().mcp->collect_status_variables();
		all_vars.insert(all_vars.end(), vars.begin(), vars.end());
	}
	if (GloAI) {
		auto vars = GloAI->collect_status_variables();
		all_vars.insert(all_vars.end(), vars.begin(), vars.end());
	}

	for (auto& [name, value] : all_vars) {
		char** pta = (char**)malloc(sizeof(char*) * 2);
		pta[0] = strdup(name.c_str());
		pta[1] = strdup(value.c_str());
		(*result)->add_row(pta);
		for (int j = 0; j < 2; j++) free(pta[j]);
		free(pta);
	}

	return 0;
}

register_stats(services, "stats_genai_global", kStatsGenaiGlobal);
services->register_runtime_view("stats_genai_global",
	refresh_stats_genai_global, nullptr, ProxySQL_PluginDBKind::admin_db);

// SHOW GENAI STATUS
{
	const char* cmd = "SHOW GENAI STATUS";
	services->register_command(cmd, [](ProxySQL_PluginServices* srv, char** err,
	                                       const char* sql) -> int {
		// The admin handler will rewrite this query
		// For now, we use a simple approach: the command handler
		// triggers a refresh of stats_genai_global and the admin
		// rewriter in core handles the rest.
		//
		// Actually, we register this as a command so the admin
		// interface recognizes it. The actual rewrite happens via
		// the query hook or admin handler.
		//
		// Simplest approach: register the command, and in the
		// handler return 0 (success). The admin interface already
		// knows how to handle "SHOW" commands by looking at tables.
		return 0;
	});
}

// SHOW MCP STATUS
{
	const char* cmd = "SHOW MCP STATUS";
	services->register_command(cmd, [](ProxySQL_PluginServices* srv,
	                                       char** err, const char* sql) -> int {
		return 0;
	});
}

git add plugins/genai/src/plugin_tables.cpp
git commit -m "feat(genai): register stats_genai_global status table (Feature C)"

struct {
	bool enabled;
	bool cache_enabled;             // NEW
	char* provider;
	char* provider_url;
	char* provider_model;
	char* provider_key;
	int cache_similarity_threshold;
	int timeout_ms;
} config;

void update_config(const char* provider, const char* provider_url, const char* provider_model,
                   const char* provider_key, int cache_threshold, int timeout, bool cache_enabled);

llm_bridge->update_config(
	vars->llm_provider, vars->llm_provider_url, vars->llm_provider_model,
	vars->llm_provider_key, vars->llm_cache_similarity_threshold,
	vars->llm_timeout_ms, vars->llm_cache_enabled);

git add plugins/genai/include/LLM_Bridge.h plugins/genai/src/LLM_Bridge.cpp plugins/genai/src/GenAI_Thread.cpp
git commit -m "feat(genai): wire genai_llm_cache_enabled to LLM_Bridge config (Feature A)"

LLMResult LLM_Bridge::check_cache(const LLMRequest& req) {
	LLMResult result;
	result.cached = false;
	result.cache_hit = false;

	if (!config.cache_enabled || !vector_db || !req.allow_cache) {
		return result;
	}

	auto start_time = std::chrono::high_resolution_clock::now();

	std::vector<float> embedding = get_text_embedding(req.prompt);
	if (embedding.empty()) {
		return result;
	}

	// Serialize embedding as blob for sqlite-vec KNN search
	size_t blob_size = embedding.size() * sizeof(float);
	std::string emb_blob(reinterpret_cast<const char*>(embedding.data()), blob_size);

	sqlite3* db = vector_db->get_db();
	sqlite3_stmt* stmt = nullptr;

	// KNN search: find the closest cached prompt
	int rc = sqlite3_prepare_v2(db,
		"SELECT lc.id, lc.response, lc.hit_count, lcv.distance "
		"FROM llm_cache_vec lcv "
		"JOIN llm_cache lc ON lc.rowid = lcv.rowid "
		"WHERE lcv.embedding MATCH ?1 AND k = 1 "
		"ORDER BY lcv.distance",
		-1, &stmt, nullptr);

	if (rc != SQLITE_OK) {
		auto end_time = std::chrono::high_resolution_clock::now();
		result.cache_lookup_time_ms = std::chrono::duration_cast<std::chrono::milliseconds>(end_time - start_time).count();
		return result;
	}

	// Bind the embedding blob as parameter
	sqlite3_bind_blob(stmt, 1, emb_blob.data(), emb_blob.size(), SQLITE_STATIC);

	if (sqlite3_step(stmt) == SQLITE_ROW) {
		int64_t cache_id = sqlite3_column_int64(stmt, 0);
		const char* response_text = (const char*)sqlite3_column_text(stmt, 1);
		int hit_count = sqlite3_column_int(stmt, 2);
		double distance = sqlite3_column_double(stmt, 3);

		// cosine distance: 0 = identical, 2 = opposite
		// similarity = 1 - distance; threshold is 0-100 (e.g. 85 = 0.85 similarity)
		double similarity = 1.0 - distance;
		double threshold = config.cache_similarity_threshold / 100.0;

		if (similarity >= threshold) {
			result.cached = true;
			result.cache_hit = true;
			result.cache_id = cache_id;
			if (response_text) result.text_response = response_text;

			// Update hit stats
			char* update_sql = sqlite3_mprintf(
				"UPDATE llm_cache SET hit_count = hit_count + 1, last_hit = unixepoch() WHERE id = %lld",
				(long long)cache_id);
			if (update_sql) {
				sqlite3_exec(db, update_sql, nullptr, nullptr, nullptr);
				sqlite3_free(update_sql);
			}

			if (GloAI) {
				GloAI->increment_llm_cache_hits();
			}
		} else {
			if (GloAI) {
				GloAI->increment_llm_cache_misses();
			}
		}
	}

	sqlite3_finalize(stmt);

	auto end_time = std::chrono::high_resolution_clock::now();
	result.cache_lookup_time_ms = std::chrono::duration_cast<std::chrono::milliseconds>(end_time - start_time).count();

	if (GloAI) {
		GloAI->increment_llm_cache_lookups();
		GloAI->add_llm_cache_lookup_time_ms(result.cache_lookup_time_ms);
	}

	return result;
}

git add plugins/genai/src/LLM_Bridge.cpp
git commit -m "feat(genai): implement LLM Bridge check_cache with vector similarity (Feature A)"

void LLM_Bridge::store_in_cache(const LLMRequest& req, const LLMResult& result) {
	if (!config.cache_enabled || !vector_db || !req.allow_cache) {
		return;
	}
	if (result.text_response.empty()) {
		return;
	}

	auto start_time = std::chrono::high_resolution_clock::now();

	std::vector<float> embedding = get_text_embedding(req.prompt);
	if (embedding.empty()) {
		return;
	}

	sqlite3* db = vector_db->get_db();

	// Insert into llm_cache
	char* insert_sql = sqlite3_mprintf(
		"INSERT INTO llm_cache (prompt, response, system_message, embedding, hit_count, created_at) "
		"VALUES (%Q, %Q, %Q, ?, 0, unixepoch())",
		req.prompt.c_str(), result.text_response.c_str(),
		req.system_message.c_str());

	if (!insert_sql) return;

	sqlite3_stmt* stmt = nullptr;
	int rc = sqlite3_prepare_v2(db, insert_sql, -1, &stmt, nullptr);
	sqlite3_free(insert_sql);

	if (rc != SQLITE_OK) return;

	// Bind embedding blob
	size_t blob_size = embedding.size() * sizeof(float);
	std::string emb_blob(reinterpret_cast<const char*>(embedding.data()), blob_size);
	sqlite3_bind_blob(stmt, sqlite3_bind_parameter_index(stmt, "?1"),
	                  emb_blob.data(), emb_blob.size(), SQLITE_STATIC);

	if (sqlite3_step(stmt) != SQLITE_DONE) {
		sqlite3_finalize(stmt);
		return;
	}
	sqlite3_finalize(stmt);

	sqlite3_int64 rowid = sqlite3_last_insert_rowid(db);

	// Insert into llm_cache_vec
	const char* vec_sql = "INSERT INTO llm_cache_vec (rowid, embedding) VALUES (?, ?)";
	rc = sqlite3_prepare_v2(db, vec_sql, -1, &stmt, nullptr);
	if (rc != SQLITE_OK) return;

	sqlite3_bind_int64(stmt, 1, rowid);
	sqlite3_bind_blob(stmt, 2, emb_blob.data(), emb_blob.size(), SQLITE_STATIC);
	sqlite3_step(stmt);
	sqlite3_finalize(stmt);

	auto end_time = std::chrono::high_resolution_clock::now();
	const_cast<LLMResult&>(result).cache_store_time_ms =
		std::chrono::duration_cast<std::chrono::milliseconds>(end_time - start_time).count();

	if (GloAI) {
		GloAI->increment_llm_cache_stores();
		GloAI->add_llm_cache_store_time_ms(result.cache_store_time_ms);
	}
}

git add plugins/genai/src/LLM_Bridge.cpp
git commit -m "feat(genai): implement LLM Bridge store_in_cache (Feature A)"

void LLM_Bridge::clear_cache() {
	if (!vector_db) {
		return;
	}

	vector_db->execute("DELETE FROM llm_cache_vec");
	vector_db->execute("DELETE FROM llm_cache");

	proxy_info("LLM_Bridge: Cache cleared\n");
}

std::string LLM_Bridge::get_cache_stats() {
	json stats;
	stats["entries"] = 0;
	stats["hits"] = 0;
	stats["misses"] = 0;

	if (!vector_db) {
		return stats.dump();
	}

	sqlite3* db = vector_db->get_db();
	sqlite3_stmt* stmt = nullptr;

	int rc = sqlite3_prepare_v2(db,
		"SELECT COUNT(*), COALESCE(SUM(hit_count), 0) FROM llm_cache",
		-1, &stmt, nullptr);

	if (rc == SQLITE_OK && sqlite3_step(stmt) == SQLITE_ROW) {
		stats["entries"] = sqlite3_column_int(stmt, 0);
		stats["hits"] = sqlite3_column_int(stmt, 1);
	}

	sqlite3_finalize(stmt);

	if (GloAI) {
		stats["misses"] = GloAI->get_status_variable_llm_cache_misses();
		stats["lookups"] = GloAI->get_status_variable_llm_cache_lookups();
		stats["stores"] = GloAI->get_status_variable_llm_cache_stores();
	}

	return stats.dump();
}

git add plugins/genai/src/LLM_Bridge.cpp
git commit -m "feat(genai): implement LLM Bridge clear_cache and get_cache_stats (Feature A)"

CREATE TABLE IF NOT EXISTS rag_documents (
    doc_id VARCHAR NOT NULL PRIMARY KEY,
    source_id INTEGER NOT NULL,
    source_name VARCHAR,
    pk_json VARCHAR,
    title VARCHAR,
    body TEXT,
    metadata_json TEXT,
    content_hash VARCHAR(64),      -- NEW: SHA-256 of content for change detection
    updated_at INTEGER DEFAULT (unixepoch()),
    deleted INTEGER DEFAULT 0
)

// Add content_hash column if it doesn't exist (migration)
db.execute("ALTER TABLE rag_documents ADD COLUMN content_hash VARCHAR(64)");
// Ignore error if column already exists

#include <openssl/sha.h>

static std::string compute_content_hash(const std::string& input) {
	unsigned char hash[SHA256_DIGEST_LENGTH];
	SHA256(reinterpret_cast<const unsigned char*>(input.c_str()), input.size(), hash);

	char hex[SHA256_DIGEST_LENGTH * 2 + 1];
	for (int i = 0; i < SHA256_DIGEST_LENGTH; i++) {
		snprintf(hex + i * 2, 3, "%02x", hash[i]);
	}
	return std::string(hex);
}

std::string content_input = doc.title + "|" + doc.body + "|" + doc.metadata_json;
std::string content_hash = compute_content_hash(content_input);

git add RAG_POC/rag_ingest.cpp RAG_POC/Makefile
git commit -m "feat(rag): add content_hash column and SHA-256 computation (Feature D)"

// OLD:
if (doc_exists(db, doc.doc_id)) {
    g_logger.trace("Document already exists: " + doc.doc_id + ", skipping");
    continue;
}

std::string existing_hash;
bool doc_changed = false;

MySQLDB::QueryResult hash_result = db.query(
    "SELECT content_hash FROM rag_documents WHERE doc_id = '" + escape_sql(doc.doc_id) + "'");
if (!hash_result.rows.empty()) {
    existing_hash = hash_result.rows[0][0] ? hash_result.rows[0][0] : "";
    if (existing_hash == content_hash) {
        g_logger.trace("Document unchanged: " + doc.doc_id);
        seen_doc_ids.insert(doc.doc_id);
        continue;
    }
    doc_changed = true;
    g_logger.debug("Document changed: " + doc.doc_id + " (hash mismatch)");
} else {
    g_logger.debug("New document: " + doc.doc_id);
}

if (doc_changed) {
    // Soft-delete old document
    db.execute("UPDATE rag_documents SET deleted = 1, updated_at = unixepoch() WHERE doc_id = '"
               + escape_sql(doc.doc_id) + "'");
    // Delete old chunks
    db.execute("DELETE FROM rag_chunks WHERE doc_id = '" + escape_sql(doc.doc_id) + "'");
    // Delete old FTS chunks
    db.execute("DELETE FROM rag_fts_chunks WHERE doc_id = '" + escape_sql(doc.doc_id) + "'");
    // Delete old vector chunks (if embeddings exist)
    db.execute("DELETE FROM rag_vec_chunks WHERE doc_id = '" + escape_sql(doc.doc_id) + "'");
}

// In the INSERT INTO rag_documents statement, add content_hash:
sql << "INSERT INTO rag_documents(doc_id, source_id, source_name, pk_json, title, body, metadata_json, content_hash) "
    << "VALUES ('";
// ... existing values ...
sql << "', '" << content_hash << "')";

git add RAG_POC/rag_ingest.cpp
git commit -m "feat(rag): hash-based update detection with re-chunk (Feature D)"

// Delete detection: find documents that exist but were NOT seen in this batch
MySQLDB::QueryResult existing_docs = db.query(
    "SELECT doc_id FROM rag_documents WHERE source_id = " + std::to_string(source.id) +
    " AND deleted = 0");

int soft_deleted = 0;
for (auto& row : existing_docs.rows) {
    std::string existing_doc_id = row[0] ? row[0] : "";
    if (seen_doc_ids.find(existing_doc_id) == seen_doc_ids.end()) {
        g_logger.debug("Document removed from source: " + existing_doc_id + " (soft deleting)");

        db.execute("UPDATE rag_documents SET deleted = 1, updated_at = unixepoch() "
                   "WHERE doc_id = '" + escape_sql(existing_doc_id) + "'");
        db.execute("DELETE FROM rag_chunks WHERE doc_id = '" + escape_sql(existing_doc_id) + "'");
        db.execute("DELETE FROM rag_fts_chunks WHERE doc_id = '" + escape_sql(existing_doc_id) + "'");
        db.execute("DELETE FROM rag_vec_chunks WHERE doc_id = '" + escape_sql(existing_doc_id) + "'");
        soft_deleted++;
    }
}

if (soft_deleted > 0) {
    g_logger.info("Soft-deleted " + std::to_string(soft_deleted) +
                  " documents no longer in source '" + source.name + "'");
}

git add RAG_POC/rag_ingest.cpp
git commit -m "feat(rag): delete detection for removed source documents (Feature D)"

SELECT * FROM stats_genai_global;