🧠 Memória: SQLite FTS5 vs vector vs Markdown

import aiosqlite, uuid, time, json
from dataclasses import dataclass, field

SCHEMA = """
CREATE TABLE IF NOT EXISTS memories(
  id TEXT PRIMARY KEY,
  user_id TEXT NOT NULL,
  category TEXT NOT NULL,
  content TEXT NOT NULL,
  metadata TEXT NOT NULL DEFAULT '{}',
  created_at REAL NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_user_cat ON memories(user_id, category, created_at);

CREATE VIRTUAL TABLE IF NOT EXISTS memories_fts USING fts5(
  content,
  content='memories', content_rowid='rowid',
  tokenize='unicode61 remove_diacritics 2'
);

CREATE TRIGGER IF NOT EXISTS memories_ai AFTER INSERT ON memories BEGIN
  INSERT INTO memories_fts(rowid, content) VALUES (new.rowid, new.content);
END;
CREATE TRIGGER IF NOT EXISTS memories_ad AFTER DELETE ON memories BEGIN
  INSERT INTO memories_fts(memories_fts, rowid, content) VALUES('delete', old.rowid, old.content);
END;
"""

class SQLiteMemory:
    def __init__(self, path: str): self.path = path
    async def init(self):
        async with aiosqlite.connect(self.path) as db:
            await db.executescript(SCHEMA); await db.commit()

    async def save(self, content, category, user_id, metadata=None) -> str:
        eid = str(uuid.uuid4())
        async with aiosqlite.connect(self.path) as db:
            await db.execute(
                "INSERT INTO memories(id,user_id,category,content,metadata,created_at) VALUES(?,?,?,?,?,?)",
                (eid, user_id, category, content, json.dumps(metadata or {}), time.time())
            )
            await db.commit()
        return eid

    async def search(self, query, user_id, limit=10):
        # FTS5 MATCH + snippet + bm25 ranking, escopado por user_id
        async with aiosqlite.connect(self.path) as db:
            cur = await db.execute("""
                SELECT m.id, m.content, m.category, m.created_at, m.metadata,
                       snippet(memories_fts, 0, '<b>', '</b>', '...', 16) as snip
                FROM memories_fts
                JOIN memories m ON m.rowid = memories_fts.rowid
                WHERE memories_fts MATCH ? AND m.user_id = ?
                ORDER BY bm25(memories_fts) LIMIT ?
            """, (query, user_id, limit))
            return [dict(zip([c[0] for c in cur.description], r)) async for r in cur]

    async def forget(self, entry_id, user_id) -> bool:
        async with aiosqlite.connect(self.path) as db:
            cur = await db.execute(
                "DELETE FROM memories WHERE id=? AND user_id=?", (entry_id, user_id))
            await db.commit()
            return cur.rowcount > 0

import chromadb
from chromadb.utils import embedding_functions

# Chroma local — embarcado, sem servidor. Bom pra dev e produção pequena.
client = chromadb.PersistentClient(path="./chroma_db")
embed = embedding_functions.SentenceTransformerEmbeddingFunction(
    model_name="paraphrase-multilingual-MiniLM-L12-v2"  # bom pra pt-BR
)

collection = client.get_or_create_collection(
    name="memories", embedding_function=embed,
    metadata={"hnsw:space": "cosine"}
)

# Save: id, document, metadata
collection.add(
    ids=["mem-001"],
    documents=["Marco prefere AWS por causa do Bedrock e equipe já treinada"],
    metadatas=[{"user_id": "u_42", "category": "decision", "ts": 1715900000}]
)

# Search: top-k semantic + filter
hits = collection.query(
    query_texts=["por que escolhemos cloud provider"],
    n_results=5,
    where={"user_id": "u_42"}   # SEMPRE filtrar por user
)
for doc, dist, meta in zip(hits["documents"][0], hits["distances"][0], hits["metadatas"][0]):
    print(f"[{dist:.3f}] {meta['category']}: {doc}")

---
agent: marco-os
version: 2.4
updated: 2026-05-18
owner: marco@empresa.com
tags: [identidade, regras, escopo]
priority: always-load
---

# MEMORY.md — Marco OS

## Identidade
Você é o agente pessoal do Marco. Responde em pt-BR formal, frases curtas.
Nunca usa emoji exceto em listas de check (✓/✗).

## Stakeholders
- **Marco** (CEO) — decisão final, prefere ver número antes de prosa
- **Ana** (CTO) — discutir antes de mudar arquitetura
- **Time eng** (12 pessoas) — pode ler tudo exceto folha de pagamento

## Regras inegociáveis
1. Nunca envia email sem aprovação explícita
2. Audit log obrigatório em qualquer ação financeira
3. PII só fica em SQLite local, nunca em prompt de LLM externo

## Decisões arquiteturais ativas (2026)
- Cloud: AWS (Bedrock, RDS, S3)
- LLM default: Claude Sonnet 4.6, fallback GPT-4o
- Memory: SQLite FTS5 + Chroma local pra docs

import time, uuid, json

class ForgetService:
    """Apaga em todos os stores + grava audit imutável.
    Audit fica em tabela separada que NÃO é deletada — só ela
    prova depois que o apagamento aconteceu."""

    def __init__(self, sqlite_mem, vector_mem, audit_db):
        self.sqlite = sqlite_mem
        self.vector = vector_mem
        self.audit = audit_db

    async def forget_user(self, user_id: str, requested_by: str, reason: str) -> dict:
        req_id = str(uuid.uuid4())
        started = time.time()

        # 1. SQLite — DELETE escopado por user_id
        sql_deleted = await self.sqlite.forget_all_for_user(user_id)

        # 2. Vector store — Chroma where filter
        vec_deleted = await self.vector.delete(where={"user_id": user_id})

        # 3. MD blocks — busca arquivos com owner=user e remove
        md_files = await self._purge_md_blocks(user_id)

        # 4. Logs — anonimiza request_ids associados
        logs_anon = await self._anonymize_logs(user_id)

        # 5. Audit imutável — append-only, NÃO deletar nem editar
        await self.audit.append({
            "request_id": req_id,
            "operation": "forget_user",
            "subject_user_id": user_id,
            "requested_by": requested_by,
            "reason": reason,
            "started_at": started,
            "finished_at": time.time(),
            "results": {
                "sqlite_rows_deleted": sql_deleted,
                "vector_rows_deleted": vec_deleted,
                "md_files_purged": md_files,
                "log_entries_anonymized": logs_anon,
            }
        })
        return {"request_id": req_id, "status": "completed"}