Revert "Added database queue for memory operations"

nullure · nullure · commit 62ff53df2cea · 2025-11-05T15:29:21.000+05:30
This reverts commit 4bece46.
diff --git a/.gitignore b/.gitignore
@@ -360,5 +360,3 @@ dist
 !.yarn/versions
 
 benchmark
-temp
-dashboard
diff --git a/backend/src/ai/mcp.ts b/backend/src/ai/mcp.ts
@@ -67,9 +67,6 @@ export const create_mcp_srv = () => {
         metadata: z.record(z.any()).optional().describe('Arbitrary metadata blob'),
         user_id: z.string().trim().min(1).optional().describe('Associate the memory with a specific user identifier')
     }, async ({ content, tags, metadata, user_id }) => {
-        // Note: add_hsg_memory internally queues writes to handle concurrent parallel calls
-        // This prevents SQLite "transaction within transaction" errors when clients like
-        // Gemini CLI make multiple simultaneous store requests
         const u = uid(user_id)
         const res = await add_hsg_memory(content, j(tags || []), metadata, u)
         if (u) update_user_summary(u).catch(err => console.error('[MCP] user summary update failed:', err))
diff --git a/backend/src/core/queue.ts b/backend/src/core/queue.ts
diff --git a/backend/src/memory/hsg.ts b/backend/src/memory/hsg.ts
@@ -3,7 +3,6 @@ import { canonical_token_set } from '../utils/text'
 import { inc_q, dec_q } from './decay'
 import { env, tier } from '../core/cfg'
 import { cos_sim, buf_to_vec, vec_to_buf } from '../utils/index'
-import { dbq } from '../core/queue'
 export interface sector_cfg {
     model: string
     decay_lambda: number
@@ -673,88 +672,86 @@ export async function add_hsg_memory(
     metadata?: any,
     user_id?: string
 ): Promise<{ id: string, primary_sector: string, sectors: string[], chunks?: number, deduplicated?: boolean }> {
-    return dbq.exec(async () => {
-        const simhash = compute_simhash(content)
-        const existing = await q.get_mem_by_simhash.get(simhash)
-        if (existing && hamming_dist(simhash, existing.simhash) <= 3) {
-            const now = Date.now()
-            const boosted_sal = Math.min(1, existing.salience + 0.15)
-            await q.upd_seen.run(existing.id, now, boosted_sal, now)
-            return {
-                id: existing.id,
-                primary_sector: existing.primary_sector,
-                sectors: [existing.primary_sector],
-                deduplicated: true
-            }
-        }
-        const id = crypto.randomUUID()
+    const simhash = compute_simhash(content)
+    const existing = await q.get_mem_by_simhash.get(simhash)
+    if (existing && hamming_dist(simhash, existing.simhash) <= 3) {
         const now = Date.now()
-        const chunks = chunk_text(content)
-        const use_chunking = chunks.length > 1
-        const classification = classify_content(content, metadata)
-        const all_sectors = [classification.primary, ...classification.additional]
-        await transaction.begin()
-        try {
-            const max_seg_res = await q.get_max_segment.get()
-            let cur_seg = max_seg_res?.max_seg ?? 0
-            const seg_cnt_res = await q.get_segment_count.get(cur_seg)
-            const seg_cnt = seg_cnt_res?.c ?? 0
-            if (seg_cnt >= env.seg_size) {
-                cur_seg++
-                console.log(`[HSG] Rotated to segment ${cur_seg} (previous segment full: ${seg_cnt} memories)`)
-            }
-            const stored_content = extract_essence(content, classification.primary, env.summary_max_length)
-            const sec_cfg = sector_configs[classification.primary]
-            const init_sal = Math.max(0, Math.min(1, 0.4 + 0.1 * classification.additional.length))
-            await q.ins_mem.run(
-                id,
-                user_id || null,
-                cur_seg,
-                stored_content,
-                simhash,
-                classification.primary,
-                tags || null,
-                JSON.stringify(metadata || {}),
-                now,
-                now,
-                now,
-                init_sal,
-                sec_cfg.decay_lambda,
-                1,
-                null,
-                null,
-                null,  // compressed_vec
-                0      // feedback_score
-            )
-            const emb_res = await embedMultiSector(id, content, all_sectors, use_chunking ? chunks : undefined)
-            for (const result of emb_res) {
-                const vec_buf = vectorToBuffer(result.vector)
-                await q.ins_vec.run(id, result.sector, vec_buf, result.dim)
-            }
-            const mean_vec = calc_mean_vec(emb_res, all_sectors)
-            const mean_vec_buf = vectorToBuffer(mean_vec)
-            await q.upd_mean_vec.run(id, mean_vec.length, mean_vec_buf)
+        const boosted_sal = Math.min(1, existing.salience + 0.15)
+        await q.upd_seen.run(existing.id, now, boosted_sal, now)
+        return {
+            id: existing.id,
+            primary_sector: existing.primary_sector,
+            sectors: [existing.primary_sector],
+            deduplicated: true
+        }
+    }
+    const id = crypto.randomUUID()
+    const now = Date.now()
+    const chunks = chunk_text(content)
+    const use_chunking = chunks.length > 1
+    const classification = classify_content(content, metadata)
+    const all_sectors = [classification.primary, ...classification.additional]
+    await transaction.begin()
+    try {
+        const max_seg_res = await q.get_max_segment.get()
+        let cur_seg = max_seg_res?.max_seg ?? 0
+        const seg_cnt_res = await q.get_segment_count.get(cur_seg)
+        const seg_cnt = seg_cnt_res?.c ?? 0
+        if (seg_cnt >= env.seg_size) {
+            cur_seg++
+            console.log(`[HSG] Rotated to segment ${cur_seg} (previous segment full: ${seg_cnt} memories)`)
+        }
+        const stored_content = extract_essence(content, classification.primary, env.summary_max_length)
+        const sec_cfg = sector_configs[classification.primary]
+        const init_sal = Math.max(0, Math.min(1, 0.4 + 0.1 * classification.additional.length))
+        await q.ins_mem.run(
+            id,
+            user_id || null,
+            cur_seg,
+            stored_content,
+            simhash,
+            classification.primary,
+            tags || null,
+            JSON.stringify(metadata || {}),
+            now,
+            now,
+            now,
+            init_sal,
+            sec_cfg.decay_lambda,
+            1,
+            null,
+            null,
+            null,  // compressed_vec
+            0      // feedback_score
+        )
+        const emb_res = await embedMultiSector(id, content, all_sectors, use_chunking ? chunks : undefined)
+        for (const result of emb_res) {
+            const vec_buf = vectorToBuffer(result.vector)
+            await q.ins_vec.run(id, result.sector, vec_buf, result.dim)
+        }
+        const mean_vec = calc_mean_vec(emb_res, all_sectors)
+        const mean_vec_buf = vectorToBuffer(mean_vec)
+        await q.upd_mean_vec.run(id, mean_vec.length, mean_vec_buf)
 
-            // Store compressed vector for smart tier (for future query optimization)
-            if (tier === 'smart' && mean_vec.length > 128) {
-                const comp = compress_vec_for_storage(mean_vec, 128)
-                const comp_buf = vectorToBuffer(comp)
-                await q.upd_compressed_vec.run(comp_buf, id)
-            }
+        // Store compressed vector for smart tier (for future query optimization)
+        if (tier === 'smart' && mean_vec.length > 128) {
+            const comp = compress_vec_for_storage(mean_vec, 128)
+            const comp_buf = vectorToBuffer(comp)
+            await q.upd_compressed_vec.run(comp_buf, id)
+        }
 
-            await create_single_waypoint(id, mean_vec, now)
-            await transaction.commit()
-            return {
-                id,
-                primary_sector: classification.primary,
-                sectors: all_sectors,
-                chunks: chunks.length
-            }
-        } catch (error) {
-            await transaction.rollback()
-            throw error
+        await create_single_waypoint(id, mean_vec, now)
+        await transaction.commit()
+        return {
+            id,
+            primary_sector: classification.primary,
+            sectors: all_sectors,
+            chunks: chunks.length
         }
-    })
+    } catch (error) {
+        await transaction.rollback()
+        throw error
+    }
 }
 export async function reinforce_memory(id: string, boost: number = 0.1): Promise<void> {
     const mem = await q.get_mem.get(id)
@@ -768,37 +765,35 @@ export async function update_memory(
     tags?: string[],
     metadata?: any
 ): Promise<{ id: string, updated: boolean }> {
-    return dbq.exec(async () => {
-        const mem = await q.get_mem.get(id)
-        if (!mem) throw new Error(`Memory ${id} not found`)
-        const new_content = content !== undefined ? content : mem.content
-        const new_tags = tags !== undefined ? j(tags) : (mem.tags || '[]')
-        const new_meta = metadata !== undefined ? j(metadata) : (mem.meta || '{}')
-        await transaction.begin()
-        try {
-            if (content !== undefined && content !== mem.content) {
-                const chunks = chunk_text(new_content)
-                const use_chunking = chunks.length > 1
-                const classification = classify_content(new_content, metadata)
-                const all_sectors = [classification.primary, ...classification.additional]
-                await q.del_vec.run(id)
-                const emb_res = await embedMultiSector(id, new_content, all_sectors, use_chunking ? chunks : undefined)
-                for (const result of emb_res) {
-                    const vec_buf = vectorToBuffer(result.vector)
-                    await q.ins_vec.run(id, result.sector, vec_buf, result.dim)
-                }
-                const mean_vec = calc_mean_vec(emb_res, all_sectors)
-                const mean_vec_buf = vectorToBuffer(mean_vec)
-                await q.upd_mean_vec.run(id, mean_vec.length, mean_vec_buf)
-                await q.upd_mem_with_sector.run(new_content, classification.primary, new_tags, new_meta, Date.now(), id)
-            } else {
-                await q.upd_mem.run(new_content, new_tags, new_meta, Date.now(), id)
+    const mem = await q.get_mem.get(id)
+    if (!mem) throw new Error(`Memory ${id} not found`)
+    const new_content = content !== undefined ? content : mem.content
+    const new_tags = tags !== undefined ? j(tags) : (mem.tags || '[]')
+    const new_meta = metadata !== undefined ? j(metadata) : (mem.meta || '{}')
+    await transaction.begin()
+    try {
+        if (content !== undefined && content !== mem.content) {
+            const chunks = chunk_text(new_content)
+            const use_chunking = chunks.length > 1
+            const classification = classify_content(new_content, metadata)
+            const all_sectors = [classification.primary, ...classification.additional]
+            await q.del_vec.run(id)
+            const emb_res = await embedMultiSector(id, new_content, all_sectors, use_chunking ? chunks : undefined)
+            for (const result of emb_res) {
+                const vec_buf = vectorToBuffer(result.vector)
+                await q.ins_vec.run(id, result.sector, vec_buf, result.dim)
             }
-            await transaction.commit()
-            return { id, updated: true }
-        } catch (error) {
-            await transaction.rollback()
-            throw error
+            const mean_vec = calc_mean_vec(emb_res, all_sectors)
+            const mean_vec_buf = vectorToBuffer(mean_vec)
+            await q.upd_mean_vec.run(id, mean_vec.length, mean_vec_buf)
+            await q.upd_mem_with_sector.run(new_content, classification.primary, new_tags, new_meta, Date.now(), id)
+        } else {
+            await q.upd_mem.run(new_content, new_tags, new_meta, Date.now(), id)
         }
-    })
+        await transaction.commit()
+        return { id, updated: true }
+    } catch (error) {
+        await transaction.rollback()
+        throw error
+    }
 }
diff --git a/backend/src/ops/ingest.ts b/backend/src/ops/ingest.ts
@@ -2,7 +2,6 @@ import { add_hsg_memory } from '../memory/hsg'
 import { q, transaction } from '../core/db'
 import { rid, now, j } from '../utils'
 import { extractText, ExtractionResult } from './extract'
-import { dbq } from '../core/queue'
 
 const LG = 8000, SEC = 3000
 
@@ -29,21 +28,19 @@ const split = (t: string, sz: number): string[] => {
 }
 
 const mkRoot = async (txt: string, ex: ExtractionResult, meta?: Record<string, unknown>) => {
-    return dbq.exec(async () => {
-        const sum = txt.length > 500 ? txt.slice(0, 500) + '...' : txt
-        const cnt = `[Document: ${ex.metadata.content_type.toUpperCase()}]\n\n${sum}\n\n[Full content split across ${Math.ceil(txt.length / SEC)} sections]`
-        const id = rid(), ts = now()
-        await transaction.begin()
-        try {
-            await q.ins_mem.run(id, cnt, 'reflective', j([]), j({ ...meta, ...ex.metadata, is_root: true, ingestion_strategy: 'root-child', ingested_at: ts }), ts, ts, ts, 1.0, 0.1, 1, null, null)
-            await transaction.commit()
-            return id
-        } catch (e) {
-            console.error('[ERROR] Root failed:', e)
-            await transaction.rollback()
-            throw e
-        }
-    })
+    const sum = txt.length > 500 ? txt.slice(0, 500) + '...' : txt
+    const cnt = `[Document: ${ex.metadata.content_type.toUpperCase()}]\n\n${sum}\n\n[Full content split across ${Math.ceil(txt.length / SEC)} sections]`
+    const id = rid(), ts = now()
+    await transaction.begin()
+    try {
+        await q.ins_mem.run(id, cnt, 'reflective', j([]), j({ ...meta, ...ex.metadata, is_root: true, ingestion_strategy: 'root-child', ingested_at: ts }), ts, ts, ts, 1.0, 0.1, 1, null, null)
+        await transaction.commit()
+        return id
+    } catch (e) {
+        console.error('[ERROR] Root failed:', e)
+        await transaction.rollback()
+        throw e
+    }
 }
 
 const mkChild = async (txt: string, idx: number, tot: number, rid: string, meta?: Record<string, unknown>) => {
@@ -52,19 +49,17 @@ const mkChild = async (txt: string, idx: number, tot: number, rid: string, meta?
 }
 
 const link = async (rid: string, cid: string, idx: number) => {
-    return dbq.exec(async () => {
-        const ts = now()
-        await transaction.begin()
-        try {
-            await q.ins_waypoint.run(rid, cid, 1.0, ts, ts)
-            await transaction.commit()
-            console.log(`🔗 Link: ${rid.slice(0, 8)} → ${cid.slice(0, 8)} (${idx})`)
-        } catch (e) {
-            await transaction.rollback()
-            console.error(`❌ Link failed for ${idx}:`, e)
-            throw e
-        }
-    })
+    const ts = now()
+    await transaction.begin()
+    try {
+        await q.ins_waypoint.run(rid, cid, 1.0, ts, ts)
+        await transaction.commit()
+        console.log(`🔗 Link: ${rid.slice(0, 8)} → ${cid.slice(0, 8)} (${idx})`)
+    } catch (e) {
+        await transaction.rollback()
+        console.error(`❌ Link failed for ${idx}:`, e)
+        throw e
+    }
 }
 
 export async function ingestDocument(t: string, data: string | Buffer, meta?: Record<string, unknown>, cfg?: ingestion_cfg): Promise<IngestionResult> {
