🧠 Memória compartilhada vs privada

Memória sem escopo é vazamento esperando hora. Toda memória do seu agente cai em um e apenas um de três baldes. Decidir o balde é decisão de produto, não de engenharia — engenharia só executa.

Banco compartilhado com coluna user_id resolve 80% — mas você ainda paga risco de query mal escrita vazando linha. File-per-tenant elimina esse risco fisicamente: cada user é um memory.db separado em disco. Query errada no DB do João não pode retornar nada da Maria.

Estrutura no disco

~/.app/users/
├── 6a2f3c-uuid-joao/
│   ├── memory.db          # SQLite só do João
│   ├── sessions/
│   └── attachments/
├── 91bd-uuid-maria/
│   ├── memory.db          # SQLite só da Maria
│   └── ...
└── _system/
    └── catalog.db         # mapeia uuid → email, plano, status

Classe UserMemory — abre o DB certo, sempre

from pathlib import Path
import sqlite3, json, time

ROOT = Path.home() / ".app/users"

class UserMemory:
    def __init__(self, user_id: str):
        if not user_id:
            raise ValueError("user_id obrigatório — sem default global")
        self.user_id = user_id
        self.path = ROOT / user_id / "memory.db"
        self.path.parent.mkdir(parents=True, exist_ok=True)
        self.db = sqlite3.connect(self.path)
        self._migrate()

    def _migrate(self):
        self.db.executescript("""
            CREATE TABLE IF NOT EXISTS memories(
              id INTEGER PRIMARY KEY,
              kind TEXT NOT NULL,           -- fact | preference | event
              content TEXT NOT NULL,
              created_at INTEGER NOT NULL,
              last_used_at INTEGER,
              uses INTEGER DEFAULT 0
            );
            CREATE INDEX IF NOT EXISTS idx_kind ON memories(kind);
        """)

    def add(self, kind: str, content: str) -> int:
        cur = self.db.execute(
            "INSERT INTO memories(kind, content, created_at) VALUES(?, ?, ?)",
            (kind, content, int(time.time()))
        )
        self.db.commit()
        return cur.lastrowid

    def recall(self, kind: str | None = None) -> list[dict]:
        sql = "SELECT id, kind, content FROM memories"
        args = ()
        if kind:
            sql += " WHERE kind = ?"; args = (kind,)
        return [dict(zip(["id","kind","content"], r))
                for r in self.db.execute(sql, args).fetchall()]

SQLite resolve fato estruturado. Para memória semântica (RAG, busca por similaridade) você precisa de vetor. Manter um Chroma/Qdrant por user fica caro — a alternativa é um índice único + metadata filter por namespace. Mas a disciplina passa a viver no filter.

Convenção de namespace

# formato: <scope>:<owner_id>
user:6a2f3c-uuid-joao        # memória privada do João
team:acme-marketing          # compartilhada no time
world:public                 # pública, indexada por todos

Chroma com metadata filter — escrita e busca

import chromadb
from uuid import uuid4

client = chromadb.PersistentClient(path="./vec")
col = client.get_or_create_collection("memories")

def remember(text: str, *, scope: str, owner: str):
    ns = f"{scope}:{owner}"
    col.add(
        documents=[text],
        metadatas=[{"namespace": ns, "scope": scope, "owner": owner}],
        ids=[f"{ns}::{uuid4()}"],
    )

def search(query: str, *, scopes: tuple[str, ...] = ("user",), owner: str):
    if not scopes:
        raise ValueError("scopes vazio = busca sem filtro = vazamento")
    # monta namespaces autorizados explicitamente
    allowed = []
    for s in scopes:
        if s == "user":    allowed.append(f"user:{owner}")
        elif s == "team":  allowed += team_ns_for(owner)   # lookup de membership
        elif s == "world": allowed.append("world:public")
    return col.query(
        query_texts=[query],
        n_results=8,
        where={"namespace": {"$in": allowed}},
    )

# BAD — sem where, busca em TUDO
col.query(query_texts=[q], n_results=8)

# GOOD — wrapper search() acima força scopes obrigatório
search(q, scopes=("user",), owner=user_id)

Memória nasce user. Promover para team é decisão explícita: comando /share, confirmação humana, audit log. Nunca automático — nem "ML sugere promover". Vazamento por ML é o pior tipo: você nem sabe onde foi parar.

Comando /share com confirmação

async def cmd_share(ctx, memory_id: int, target_team: str):
    # 1. valida ownership
    mem = ctx.user_mem.get(memory_id)
    if not mem:
        return reply("❌ memória não existe ou não é sua")

    # 2. valida membership no time
    if target_team not in ctx.user.teams:
        return reply(f"❌ você não é membro de {target_team}")

    # 3. preview + confirmação síncrona
    confirmed = await ctx.confirm(
        f"Compartilhar com `{target_team}`?\n\n"
        f"> {mem.content[:200]}\n\n"
        "Membros que verão: " + ", ".join(team_members(target_team))
    )
    if not confirmed:
        return reply("cancelado")

    # 4. cópia (não move) — original fica como user
    new_id = ctx.team_mem(target_team).add(
        kind=mem.kind, content=mem.content,
        source_user=ctx.user.id, source_mem_id=memory_id,
    )

    # 5. audit log imutável
    audit.write({
        "action": "memory.promote",
        "from": f"user:{ctx.user.id}",
        "to":   f"team:{target_team}",
        "memory_id": memory_id, "new_id": new_id,
        "actor": ctx.user.id, "ts": now(),
    })
    return reply(f"✅ compartilhada como #{new_id}")

A regra é cruel e simples: toda chamada a search() declara explicitamente os escopos consultados. Se a lista vier vazia, levanta exceção. Sem default implícito — porque default implícito é como bug de privacidade nasce.

def search(query: str, *, scopes: tuple[str, ...], owner: str):
    if not scopes:
        raise ValueError(
            "scopes não pode ser vazio. "
            "Use scopes=('user',) explicitamente."
        )
    for s in scopes:
        if s not in {"user", "team", "world"}:
            raise ValueError(f"escopo inválido: {s}")
    # ... resolve namespaces e consulta com where filter ...

# Uso típico no agente
async def answer(question: str, ctx):
    # pergunta pessoal — só user
    if is_personal(question):
        hits = search(question, scopes=("user",), owner=ctx.user.id)
    # pergunta sobre processo do time — user + team
    elif is_team_topic(question):
        hits = search(question, scopes=("user", "team"), owner=ctx.user.id)
    # dúvida genérica de produto — todos
    else:
        hits = search(question, scopes=("user", "team", "world"), owner=ctx.user.id)
    return await llm.respond(question, context=hits)

Memória cresce monotonicamente. Sem GC, em 6 meses você tem 80% de lixo — duplicatas, fatos obsoletos, memórias de usuários deletados que ficaram órfãs no vector store. Performance cai, custo sobe, e auditoria fica impossível.

Schema com CASCADE — órfãs morrem sozinhas

CREATE TABLE users(
  id          TEXT PRIMARY KEY,
  email       TEXT UNIQUE NOT NULL,
  deleted_at  INTEGER
);

CREATE TABLE memories(
  id            INTEGER PRIMARY KEY,
  user_id       TEXT NOT NULL REFERENCES users(id) ON DELETE CASCADE,
  content       TEXT NOT NULL,
  content_hash  TEXT NOT NULL,    -- sha256(normalize(content))
  created_at    INTEGER NOT NULL,
  last_used_at  INTEGER,
  ttl_days      INTEGER,             -- NULL = nunca expira
  UNIQUE(user_id, content_hash)        -- dedup automático na escrita
);
CREATE INDEX idx_mem_lastused ON memories(last_used_at);

Cron de GC — 3 passadas, idempotente

# cron: 0 3 * * *   (todo dia 3h da manhã)
def gc_run(db, vec):
    now = int(time.time())
    stats = {"expired": 0, "orphan_vec": 0, "dedup": 0}

    # 1) Expiração por TTL
    cur = db.execute("""
        DELETE FROM memories
        WHERE ttl_days IS NOT NULL
          AND created_at + (ttl_days * 86400) < ?
        RETURNING id, user_id
    """, (now,))
    expired = cur.fetchall()
    stats["expired"] = len(expired)
    # sincroniza no vector store
    for mem_id, uid in expired:
        vec.delete(ids=[f"user:{uid}::{mem_id}"])

    # 2) Órfãs no vector store (user deletado, CASCADE já tirou do SQLite)
    live_users = {r[0] for r in db.execute(
        "SELECT id FROM users WHERE deleted_at IS NULL")}
    for ns in vec.list_namespaces():
        if ns.startswith("user:") and ns.split(":")[1] not in live_users:
            n = vec.delete_namespace(ns)
            stats["orphan_vec"] += n

    # 3) Dedup defensivo (UNIQUE já protege, mas histórico antigo pode ter)
    cur = db.execute("""
        DELETE FROM memories
        WHERE id NOT IN (
          SELECT MIN(id) FROM memories GROUP BY user_id, content_hash
        )
    """)
    stats["dedup"] = cur.rowcount

    db.commit()
    log.info("gc.done", extra=stats)
    return stats