Add memora DB agent project - #166
Open
SPARKEDIX wants to merge 1 commit into
Open
Conversation
Reviewer's GuideAdds a new Memora memory agent package that implements a standalone, production-style cognitive memory store (CCDB v8) using SQLite, FAISS, BM25, and domain clustering, along with packaging, docs, benchmarks, and validation tests. Sequence diagram for Memora Memory.add flow with dedup and domain assignmentsequenceDiagram
actor User
participant Memory
participant SQLite
participant FAISSIndex
participant BM25Cache
User->>Memory: add(text, user, session, ttl)
Memory->>Memory: _embed([text])
Memory->>Memory: _check_duplicate(emb, user)
alt duplicate found
Memory->>SQLite: UPDATE crystals SET strength, last_accessed
else new memory
Memory->>SQLite: INSERT INTO crystals(..., domain="unassigned", ...)
Memory->>Memory: _assign_domain(conn, crystal_id, emb, user, text)
Memory->>FAISSIndex: add(emb)
Memory->>BM25Cache: append text tokens
Memory->>BM25Cache: _save_bm25()
end
Memory->>Memory: _maybe_rebuild_index()
Memory-->>User: crystal_id
Sequence diagram for Memora Memory.get flow with hybrid search and domain filteringsequenceDiagram
actor User
participant Memory
participant FAISSIndex
participant BM25Cache
participant SQLite
User->>Memory: get(query, user, domain, top_k)
Memory->>Memory: _auto_compress()
Memory->>Memory: _embed([query])
Memory->>FAISSIndex: search(q_embs, sk)
Memory->>BM25Cache: _get_bm25()
BM25Cache-->>Memory: BM25Okapi
Memory->>BM25Cache: get_scores(tokens)
Memory->>Memory: _detect_domain(q_emb)
Memory->>SQLite: SELECT id,text,domain,level FROM crystals WHERE id IN (...) AND domain = target_domain
SQLite-->>Memory: primary rows
alt include_bonded
Memory->>SQLite: SELECT text,level FROM crystals WHERE domain = target_domain AND id NOT IN primary_ids
SQLite-->>Memory: bonded rows
end
alt mode == "latent"
Memory-->>User: projected vectors via adapter.project()
else mode == "text"
Memory-->>User: formatted context string
end
File-Level Changes
Tips and commandsInteracting with Sourcery
Customizing Your ExperienceAccess your dashboard to:
Getting Help
|
Author
|
done i give you a advance agentic Database |
![sourcery-ai[bot]](https://avatars.githubusercontent.com/in/48477?s=60&v=4){kind=small}
There was a problem hiding this comment.
Hey - I've found 9 security issues, 7 other issues, and left some high level feedback:
Security issues:
- Avoiding SQL string concatenation: untrusted input concatenated with raw SQL query can result in SQL Injection. In order to execute raw query safely, prepared statement should be used. SQLAlchemy provides TextualSQL to easily used prepared statement with named parameters. For complex SQL composition, use SQL Expression Language or Schema Definition Language. In most cases, SQLAlchemy ORM will be a better option. (link)
- Avoiding SQL string concatenation: untrusted input concatenated with raw SQL query can result in SQL Injection. In order to execute raw query safely, prepared statement should be used. SQLAlchemy provides TextualSQL to easily used prepared statement with named parameters. For complex SQL composition, use SQL Expression Language or Schema Definition Language. In most cases, SQLAlchemy ORM will be a better option. (link)
- Avoiding SQL string concatenation: untrusted input concatenated with raw SQL query can result in SQL Injection. In order to execute raw query safely, prepared statement should be used. SQLAlchemy provides TextualSQL to easily used prepared statement with named parameters. For complex SQL composition, use SQL Expression Language or Schema Definition Language. In most cases, SQLAlchemy ORM will be a better option. (link)
- Avoiding SQL string concatenation: untrusted input concatenated with raw SQL query can result in SQL Injection. In order to execute raw query safely, prepared statement should be used. SQLAlchemy provides TextualSQL to easily used prepared statement with named parameters. For complex SQL composition, use SQL Expression Language or Schema Definition Language. In most cases, SQLAlchemy ORM will be a better option. (link)
- Avoiding SQL string concatenation: untrusted input concatenated with raw SQL query can result in SQL Injection. In order to execute raw query safely, prepared statement should be used. SQLAlchemy provides TextualSQL to easily used prepared statement with named parameters. For complex SQL composition, use SQL Expression Language or Schema Definition Language. In most cases, SQLAlchemy ORM will be a better option. (link)
- Avoiding SQL string concatenation: untrusted input concatenated with raw SQL query can result in SQL Injection. In order to execute raw query safely, prepared statement should be used. SQLAlchemy provides TextualSQL to easily used prepared statement with named parameters. For complex SQL composition, use SQL Expression Language or Schema Definition Language. In most cases, SQLAlchemy ORM will be a better option. (link)
- Avoiding SQL string concatenation: untrusted input concatenated with raw SQL query can result in SQL Injection. In order to execute raw query safely, prepared statement should be used. SQLAlchemy provides TextualSQL to easily used prepared statement with named parameters. For complex SQL composition, use SQL Expression Language or Schema Definition Language. In most cases, SQLAlchemy ORM will be a better option. (link)
- Avoiding SQL string concatenation: untrusted input concatenated with raw SQL query can result in SQL Injection. In order to execute raw query safely, prepared statement should be used. SQLAlchemy provides TextualSQL to easily used prepared statement with named parameters. For complex SQL composition, use SQL Expression Language or Schema Definition Language. In most cases, SQLAlchemy ORM will be a better option. (link)
- Avoiding SQL string concatenation: untrusted input concatenated with raw SQL query can result in SQL Injection. In order to execute raw query safely, prepared statement should be used. SQLAlchemy provides TextualSQL to easily used prepared statement with named parameters. For complex SQL composition, use SQL Expression Language or Schema Definition Language. In most cases, SQLAlchemy ORM will be a better option. (link)
General comments:
- FAISS index usage with
remove_idslooks incorrect since you never wrap the index in anIndexIDMapand pass crystal IDs instead of FAISS internal IDs; this will either no-op or corrupt the index, so consider usingIndexIDMapwith explicit IDs or rebuilding the index instead of deleting by ID. - The
deleteandoptimizeflows rebuild_faiss_to_crystal_idfrom SQL but do not re-add embeddings to the FAISS index, which can desynchronize the mapping from the actual index contents over time; you may want to rebuild the FAISS index from scratch (reload all embeddings and re-add) after bulk deletions/merges. - The README examples for
older_thanand TTL ("30d","1y") suggest string-based durations, but the implementation ofdelete(older_than)expects a timestamp float, so either updatedeleteto accept duration strings via_parse_ttlor adjust the examples to avoid misleading usage.
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- FAISS index usage with `remove_ids` looks incorrect since you never wrap the index in an `IndexIDMap` and pass crystal IDs instead of FAISS internal IDs; this will either no-op or corrupt the index, so consider using `IndexIDMap` with explicit IDs or rebuilding the index instead of deleting by ID.
- The `delete` and `optimize` flows rebuild `_faiss_to_crystal_id` from SQL but do not re-add embeddings to the FAISS index, which can desynchronize the mapping from the actual index contents over time; you may want to rebuild the FAISS index from scratch (reload all embeddings and re-add) after bulk deletions/merges.
- The README examples for `older_than` and TTL (`"30d"`, `"1y"`) suggest string-based durations, but the implementation of `delete(older_than)` expects a timestamp float, so either update `delete` to accept duration strings via `_parse_ttl` or adjust the examples to avoid misleading usage.
## Individual Comments
### Comment 1
<location path="agents/memora/core.py" line_range="289-298" />
<code_context>
+ ids.append(dup_id)
+ continue
+
+ emb_bytes = emb.astype(np.float32).tobytes()
+ cur = conn.execute(
+ "INSERT INTO crystals (user,text,embedding,domain,strength,created_at,last_accessed,level,is_session,session_id,expires_at) VALUES (?,?,?,?,1,?,?,0,?,?,?)",
+ (user, text, emb_bytes, "unassigned", now, now, int(session), sid, exp))
+ nid = cur.lastrowid
+ ids.append(nid)
+
+ domain = self._assign_domain(conn, nid, emb, user, text)
+
+ self._ensure_trained()
</code_context>
<issue_to_address>
**issue (bug_risk):** Newly added crystals always persist with domain='unassigned' even when a matching domain is detected.
`domain` is always inserted as "unassigned", and although `_assign_domain` is called, its result never updates the row. When `_detect_domain` finds a matching domain, the DB still stores "unassigned", breaking immediate domain-based filtering in `get()` and distorting stats. Please either insert using the detected domain or issue an update on `crystals.domain` right after `_assign_domain` when it returns a concrete domain.
</issue_to_address>
### Comment 2
<location path="agents/memora/core.py" line_range="528-124" />
<code_context>
+ dead = conn.execute(
+ "SELECT id FROM crystals WHERE expires_at IS NOT NULL AND expires_at < ?", (now,)
+ ).fetchall()
+ if dead:
+ ids = [r[0] for r in dead]
+ conn.execute("DELETE FROM crystals WHERE id IN " +
+ f"({','.join('?' for _ in ids)})", ids)
+ try:
+ self._global_index.remove_ids(np.array(ids, dtype=np.int64))
+ self._save_index()
</code_context>
<issue_to_address>
**issue (bug_risk):** FAISS `remove_ids` is called with crystal IDs, but index entries are added without explicit IDs, causing mismatches.
The indices are `IndexFlatIP` / `IndexIVFFlat` populated via `add`, so FAISS uses implicit IDs (0..ntotal-1). In `_clean_expired`, `delete`, and `optimize`, `remove_ids` is called with DB `crystals.id`, which do not correspond to those implicit IDs. Unless the index is wrapped in `IndexIDMap` and vectors were inserted with `add_with_ids` using these same IDs, `remove_ids` will delete the wrong entries or fail. To support deletion by crystal ID, use `IndexIDMap`/`IndexIDMap2` with `add_with_ids`, or consistently delete by FAISS’s internal IDs (e.g., via a maintained mapping).
</issue_to_address>
### Comment 3
<location path="agents/memora/core.py" line_range="364-373" />
<code_context>
+ # This avoids creating fragmented domains from diverse but related memories
+ pass
+
+ def _cluster_unassigned(self, conn):
+ """Greedy clustering on all unassigned embeddings to form new domains."""
+ if len(self.unassigned_embeddings) < MIN_DOMAIN_SIZE:
+ return
+
+ embeddings = [u[1] for u in self.unassigned_embeddings]
+ n = len(embeddings)
+ used = [False] * n
+
+ for i in range(n):
+ if used[i]:
+ continue
+ # Start a new cluster with embedding i
+ cluster = [i]
+ used[i] = True
+
+ for j in range(i + 1, n):
+ if used[j]:
+ continue
+ # Check similarity to cluster centroid
+ cluster_embs = [embeddings[k] for k in cluster]
+ centroid = np.mean(cluster_embs, axis=0)
+ centroid = centroid / np.linalg.norm(centroid)
</code_context>
<issue_to_address>
**suggestion (performance):** Unassigned clustering recomputes centroids for each candidate, leading to quadratic–cubic time complexity.
Within `_cluster_unassigned`, `cluster_embs` and the centroid are recomputed inside the inner loop for every `j`, making clustering O(n^3) as clusters grow and potentially slowing or blocking adds for large unassigned pools. Consider maintaining a running centroid per cluster and updating it incrementally when adding a member, or at minimum computing and caching the centroid once per outer iteration instead of per candidate.
Suggested implementation:
```python
for i in range(n):
if used[i]:
continue
# Start a new cluster with embedding i
cluster = [i]
used[i] = True
# Initialize running centroid and cluster size
cluster_size = 1
centroid = embeddings[i]
centroid_norm = np.linalg.norm(centroid)
if centroid_norm != 0:
centroid = centroid / centroid_norm
for j in range(i + 1, n):
```
```python
for j in range(i + 1, n):
if used[j]:
continue
# Check similarity to current cluster centroid (maintained incrementally)
candidate = embeddings[j]
candidate_norm = np.linalg.norm(candidate)
if candidate_norm == 0:
continue
candidate_unit = candidate / candidate_norm
sim_to_centroid = float(np.dot(centroid, candidate_unit))
if sim_to_centroid >= UNASSIGNED_SIM_THRESHOLD:
cluster.append(j)
used[j] = True
# Update running centroid as mean of cluster embeddings
cluster_size += 1
centroid = centroid * ((cluster_size - 1) / cluster_size) + candidate_unit / cluster_size
# Renormalize centroid
centroid_norm = np.linalg.norm(centroid)
if centroid_norm != 0:
centroid = centroid / centroid_norm
```
```python
if len(cluster) >= MIN_DOMAIN_SIZE:
# Form domain from this cluster
# Centroid has been maintained incrementally; ensure it's normalized
centroid_norm = np.linalg.norm(centroid)
if centroid_norm != 0:
centroid = centroid / centroid_norm
```
</issue_to_address>
### Comment 4
<location path="agents/memora/core.py" line_range="134-138" />
<code_context>
+
+ def _init_indices(self):
+ self._crystal_count = self._count_crystals()
+ self._faiss_to_crystal_id = [] # Maps FAISS index -> crystal_id
+ if os.path.exists(self.ip):
+ self._global_index = faiss.read_index(self.ip)
+ self._trained = True
+ self._load_faiss_mapping()
+ else:
+ self._create_new_index()
</code_context>
<issue_to_address>
**issue (bug_risk):** FAISS-to-crystal ID mapping is kept in-memory only and can become inconsistent after selective deletions.
Selective deletions call `remove_ids` with DB IDs, but `self._faiss_to_crystal_id` is keyed by FAISS positions built from an ordered DB snapshot. After partial removals, this mapping no longer reflects the actual index, so `_check_duplicate` and `get_many` can return incorrect results when translating FAISS indices. Consider using `IndexIDMap` with `add_with_ids` so FAISS stores the DB IDs directly and the manual mapping can be removed, or ensure deletions are done by FAISS position and that the mapping is updated in lockstep with any adds/removes.
</issue_to_address>
### Comment 5
<location path="agents/memora/tests/validation_test.py" line_range="1-10" />
<code_context>
+import memora, time, os
</code_context>
<issue_to_address>
**issue (testing):** Convert this script-style validation into proper automated tests with assertions
This file currently relies on `print`-based checks, which won’t cause CI failures on regressions. Please rework this into automated tests (e.g., `pytest`/`unittest`) with explicit assertions, such as: duplicates don’t increase `total_memories`, `dup_id1 == dup_id2`, and query results respecting domain isolation. That way regressions in dedup, domain handling, and retrieval will be caught automatically.
</issue_to_address>
### Comment 6
<location path="agents/memora/tests/validation_test.py" line_range="46-55" />
<code_context>
+for mem in memories:
</code_context>
<issue_to_address>
**suggestion (testing):** Add targeted tests for key behaviors that are currently untested (TTL/session, domain filters, latent mode, deletion/import)
The current validation covers deduplication, domain formation, query accuracy, storage, and speed, but several key `Memory` behaviors remain untested: (1) TTLs and session memories (entries expiring and no longer returned). (2) Domain-scoped `get`/`get_many` (requesting a specific domain and ensuring other domains are excluded). (3) `mode="latent"` output shape and stability. (4) `delete`, `import_data`, and `optimize` correctly updating FAISS/BM25 indices and DB state. Please add focused tests with assertions for each of these behaviors to verify the implementation matches the intended feature set and to prevent regressions.
Suggested implementation:
```python
info = m.info()
print("=== DEDUP TEST ===")
print(f"Total crystals: {info['total_memories']} (expected: <=30)")
if info['total_memories'] > 30:
print("FAIL: Dedup broken")
else:
print("PASS: Dedup working")
print("\n=== TTL / SESSION TEST ===")
# Add a short-lived session memory and ensure it expires and is not returned
session_mem = "Temporary session memory that should expire"
session_id = m.add(
session_mem,
user="rahul",
session="validation_session_ttl",
ttl=1, # 1 second TTL
)
session_results_initial = m.get(
"Temporary session",
user="rahul",
session="validation_session_ttl",
)
if any(r.get("id") == session_id for r in session_results_initial):
print("PASS: Session memory returned before TTL expiry")
else:
print("FAIL: Session memory missing before TTL expiry")
# Wait for TTL to expire
import time
time.sleep(2)
session_results_expired = m.get(
"Temporary session",
user="rahul",
session="validation_session_ttl",
)
if any(r.get("id") == session_id for r in session_results_expired):
print("FAIL: Expired session memory still being returned")
else:
print("PASS: Expired session memory no longer returned")
print("\n=== DOMAIN FILTER TEST ===")
# Create domain-scoped memories and ensure cross-domain isolation
domain_a = "gaming"
domain_b = "work"
mem_a1 = "Playing Valorant ranked"
mem_a2 = "Grinding Apex Legends"
mem_b1 = "Preparing quarterly report"
mem_b2 = "Writing project documentation"
id_a1 = m.add(mem_a1, user="rahul", domain=domain_a)
id_a2 = m.add(mem_a2, user="rahul", domain=domain_a)
id_b1 = m.add(mem_b1, user="rahul", domain=domain_b)
id_b2 = m.add(mem_b2, user="rahul", domain=domain_b)
results_domain_a = m.get_many(
queries=["Valorant", "Apex"],
user="rahul",
domain=domain_a,
)
results_domain_b = m.get_many(
queries=["report", "documentation"],
user="rahul",
domain=domain_b,
)
ids_domain_a = {r.get("id") for res in results_domain_a for r in res}
ids_domain_b = {r.get("id") for res in results_domain_b for r in res}
if {id_a1, id_a2}.issubset(ids_domain_a) and id_b1 not in ids_domain_a and id_b2 not in ids_domain_a:
print("PASS: Domain A filter correctly includes only gaming memories")
else:
print("FAIL: Domain A filter returned incorrect memories")
if {id_b1, id_b2}.issubset(ids_domain_b) and id_a1 not in ids_domain_b and id_a2 not in ids_domain_b:
print("PASS: Domain B filter correctly includes only work memories")
else:
print("FAIL: Domain B filter returned incorrect memories")
print("\n=== LATENT MODE TEST ===")
# Validate latent mode output shape and stability for repeated queries
latent_query = "gaming preferences and habits"
latent_results_1 = m.get(
latent_query,
user="rahul",
mode="latent",
)
latent_results_2 = m.get(
latent_query,
user="rahul",
mode="latent",
)
def _latent_ids(results):
return [r.get("id") for r in results]
latent_ids_1 = _latent_ids(latent_results_1)
latent_ids_2 = _latent_ids(latent_results_2)
if isinstance(latent_results_1, list) and all(isinstance(r, dict) for r in latent_results_1):
print("PASS: Latent mode returns list of dicts")
else:
print("FAIL: Latent mode output shape invalid")
if latent_ids_1 == latent_ids_2:
print("PASS: Latent mode results are stable across repeated queries")
else:
print("FAIL: Latent mode results are not stable")
print("\n=== DELETE / IMPORT / OPTIMIZE TEST ===")
# 1. DELETE behavior: ensure memory is removed from indices and DB
delete_mem = "This memory will be deleted"
delete_id = m.add(delete_mem, user="rahul", domain="delete_test")
delete_results_before = m.get("deleted", user="rahul", domain="delete_test")
if any(r.get("id") == delete_id for r in delete_results_before):
print("PASS: Deletion candidate present before delete")
else:
print("FAIL: Deletion candidate missing before delete")
m.delete(delete_id, user="rahul")
delete_results_after = m.get("deleted", user="rahul", domain="delete_test")
if any(r.get("id") == delete_id for r in delete_results_after):
print("FAIL: Deleted memory still present in search results")
else:
print("PASS: Deleted memory no longer returned and indices updated")
# 2. IMPORT behavior: ensure imported data is searchable and correctly scoped
import_domain = "import_test"
import_payload = [
{
"text": "Imported FAISS-backed memory",
"user": "rahul",
"domain": import_domain,
},
{
"text": "Imported BM25-backed memory",
"user": "rahul",
"domain": import_domain,
},
]
import_ids = m.import_data(import_payload)
import_results = m.get_many(
queries=["FAISS-backed", "BM25-backed"],
user="rahul",
domain=import_domain,
)
import_result_ids = {r.get("id") for res in import_results for r in res}
if set(import_ids).issubset(import_result_ids):
print("PASS: Imported memories are queryable and correctly indexed")
else:
print("FAIL: Imported memories not fully visible in search results")
# 3. OPTIMIZE behavior: ensure optimize does not change logical results
opt_query = "gaming"
pre_opt_results = m.get(opt_query, user="rahul", domain=domain_a)
pre_opt_ids = _latent_ids(pre_opt_results)
m.optimize(domain=domain_a)
post_opt_results = m.get(opt_query, user="rahul", domain=domain_a)
post_opt_ids = _latent_ids(post_opt_results)
if pre_opt_ids == post_opt_ids:
print("PASS: Optimize preserves logical search results")
else:
print("FAIL: Optimize changed logical search results unexpectedly")
```
1. At the top of `agents/memora/tests/validation_test.py`, ensure `import time` is present (if not, add it).
2. Adjust parameter names if the actual `Memory` API differs:
- If `ttl`/`session`/`domain`/`mode` use different argument names or are encapsulated in options, update the calls to `m.add` and `m.get`/`m.get_many` accordingly.
- If `get`/`get_many` return a different structure (e.g., objects instead of dicts, nested fields for `id`), update the result accessors (`r.get("id")`) to match.
3. If `import_data` returns a different shape (e.g., list of objects or a mapping), adapt the `import_ids` handling to extract IDs correctly.
4. If `optimize` is a global operation (no `domain` argument) or lives on a different object, update the call `m.optimize(domain=domain_a)` to the correct invocation.
5. If a test framework (e.g., `pytest`) is preferred over print-based validation in this file, convert these checks into proper test functions with assertions following the existing conventions in the rest of the file.
</issue_to_address>
### Comment 7
<location path="agents/memora/README.md" line_range="167" />
<code_context>
+# Temporary chat
+memory.add("Aaj mausam accha hai", user="rahul", ttl="7d")
+
+# Session-only (disappears after session)
+memory.add("OTP is 123456", user="rahul", session=True)
+
</code_context>
<issue_to_address>
**nitpick (typo):** Minor grammar: add "the" to "after session".
You could also phrase it as: `# Session-only (disappears after the session)` for smoother readability.
Suggested implementation:
```
# Temporary chat
memory.add("Aaj mausam accha hai", user="rahul", ttl="7d")
# Session-only (disappears after the session)
memory.add("OTP is 123456", user="rahul", session=True)
```
If this example is referenced elsewhere in the README (e.g., in a bullet list or prose description), update any similar phrasing there to `disappears after the session` for consistency.
</issue_to_address>
### Comment 8
<location path="agents/memora/core.py" line_range="125" />
<code_context>
conn.execute(f"ALTER TABLE crystals ADD COLUMN {col} {dt}")
</code_context>
<issue_to_address>
**security (python.sqlalchemy.security.sqlalchemy-execute-raw-query):** Avoiding SQL string concatenation: untrusted input concatenated with raw SQL query can result in SQL Injection. In order to execute raw query safely, prepared statement should be used. SQLAlchemy provides TextualSQL to easily used prepared statement with named parameters. For complex SQL composition, use SQL Expression Language or Schema Definition Language. In most cases, SQLAlchemy ORM will be a better option.
*Source: opengrep*
</issue_to_address>
### Comment 9
<location path="agents/memora/core.py" line_range="235" />
<code_context>
rows = conn.execute(f"SELECT embedding FROM crystals WHERE id IN ({placeholders})", ids).fetchall()
</code_context>
<issue_to_address>
**security (python.sqlalchemy.security.sqlalchemy-execute-raw-query):** Avoiding SQL string concatenation: untrusted input concatenated with raw SQL query can result in SQL Injection. In order to execute raw query safely, prepared statement should be used. SQLAlchemy provides TextualSQL to easily used prepared statement with named parameters. For complex SQL composition, use SQL Expression Language or Schema Definition Language. In most cases, SQLAlchemy ORM will be a better option.
*Source: opengrep*
</issue_to_address>
### Comment 10
<location path="agents/memora/core.py" line_range="530-531" />
<code_context>
conn.execute("DELETE FROM crystals WHERE id IN " +
f"({','.join('?' for _ in ids)})", ids)
</code_context>
<issue_to_address>
**security (python.sqlalchemy.security.sqlalchemy-execute-raw-query):** Avoiding SQL string concatenation: untrusted input concatenated with raw SQL query can result in SQL Injection. In order to execute raw query safely, prepared statement should be used. SQLAlchemy provides TextualSQL to easily used prepared statement with named parameters. For complex SQL composition, use SQL Expression Language or Schema Definition Language. In most cases, SQLAlchemy ORM will be a better option.
*Source: opengrep*
</issue_to_address>
### Comment 11
<location path="agents/memora/core.py" line_range="678-681" />
<code_context>
rows = con.execute(
f"SELECT id, text, domain, level FROM crystals WHERE id IN ({ph}){w}{df} ORDER BY strength DESC, last_accessed DESC LIMIT ?",
params + dp + [top_k]
).fetchall()
</code_context>
<issue_to_address>
**security (python.sqlalchemy.security.sqlalchemy-execute-raw-query):** Avoiding SQL string concatenation: untrusted input concatenated with raw SQL query can result in SQL Injection. In order to execute raw query safely, prepared statement should be used. SQLAlchemy provides TextualSQL to easily used prepared statement with named parameters. For complex SQL composition, use SQL Expression Language or Schema Definition Language. In most cases, SQLAlchemy ORM will be a better option.
*Source: opengrep*
</issue_to_address>
### Comment 12
<location path="agents/memora/core.py" line_range="698-701" />
<code_context>
br = con.execute(
f"SELECT text, level FROM crystals WHERE user = ? AND domain = ? {ex} AND (expires_at IS NULL OR expires_at > ?) ORDER BY strength DESC, last_accessed DESC LIMIT ?",
[user, target_dom] + bp + [now, top_k]
).fetchall()
</code_context>
<issue_to_address>
**security (python.sqlalchemy.security.sqlalchemy-execute-raw-query):** Avoiding SQL string concatenation: untrusted input concatenated with raw SQL query can result in SQL Injection. In order to execute raw query safely, prepared statement should be used. SQLAlchemy provides TextualSQL to easily used prepared statement with named parameters. For complex SQL composition, use SQL Expression Language or Schema Definition Language. In most cases, SQLAlchemy ORM will be a better option.
*Source: opengrep*
</issue_to_address>
### Comment 13
<location path="agents/memora/core.py" line_range="703-706" />
<code_context>
br = con.execute(
f"SELECT text, level FROM crystals WHERE domain = ? {ex} AND (expires_at IS NULL OR expires_at > ?) ORDER BY strength DESC, last_accessed DESC LIMIT ?",
[target_dom] + bp + [now, top_k]
).fetchall()
</code_context>
<issue_to_address>
**security (python.sqlalchemy.security.sqlalchemy-execute-raw-query):** Avoiding SQL string concatenation: untrusted input concatenated with raw SQL query can result in SQL Injection. In order to execute raw query safely, prepared statement should be used. SQLAlchemy provides TextualSQL to easily used prepared statement with named parameters. For complex SQL composition, use SQL Expression Language or Schema Definition Language. In most cases, SQLAlchemy ORM will be a better option.
*Source: opengrep*
</issue_to_address>
### Comment 14
<location path="agents/memora/core.py" line_range="756" />
<code_context>
rows = con.execute(f"SELECT id FROM crystals {w}", params).fetchall()
</code_context>
<issue_to_address>
**security (python.sqlalchemy.security.sqlalchemy-execute-raw-query):** Avoiding SQL string concatenation: untrusted input concatenated with raw SQL query can result in SQL Injection. In order to execute raw query safely, prepared statement should be used. SQLAlchemy provides TextualSQL to easily used prepared statement with named parameters. For complex SQL composition, use SQL Expression Language or Schema Definition Language. In most cases, SQLAlchemy ORM will be a better option.
*Source: opengrep*
</issue_to_address>
### Comment 15
<location path="agents/memora/core.py" line_range="757" />
<code_context>
con.execute(f"DELETE FROM crystals {w}", params)
</code_context>
<issue_to_address>
**security (python.sqlalchemy.security.sqlalchemy-execute-raw-query):** Avoiding SQL string concatenation: untrusted input concatenated with raw SQL query can result in SQL Injection. In order to execute raw query safely, prepared statement should be used. SQLAlchemy provides TextualSQL to easily used prepared statement with named parameters. For complex SQL composition, use SQL Expression Language or Schema Definition Language. In most cases, SQLAlchemy ORM will be a better option.
*Source: opengrep*
</issue_to_address>
### Comment 16
<location path="agents/memora/core.py" line_range="860-861" />
<code_context>
con.execute("DELETE FROM crystals WHERE id IN " +
f"({','.join('?' for _ in rem_ids)})", rem_ids)
</code_context>
<issue_to_address>
**security (python.sqlalchemy.security.sqlalchemy-execute-raw-query):** Avoiding SQL string concatenation: untrusted input concatenated with raw SQL query can result in SQL Injection. In order to execute raw query safely, prepared statement should be used. SQLAlchemy provides TextualSQL to easily used prepared statement with named parameters. For complex SQL composition, use SQL Expression Language or Schema Definition Language. In most cases, SQLAlchemy ORM will be a better option.
*Source: opengrep*
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
Comment on lines
+289
to
+298
| emb_bytes = emb.astype(np.float32).tobytes() | ||
| conn.execute( | ||
| "INSERT INTO unassigned (crystal_id, user, text, embedding, timestamp) VALUES (?, ?, ?, ?, ?)", | ||
| (crystal_id, user, text, emb_bytes, time.time()) | ||
| ) | ||
|
|
||
| def _remove_unassigned(self, conn, crystal_id: int): | ||
| conn.execute("DELETE FROM unassigned WHERE crystal_id = ?", (crystal_id,)) | ||
| self.unassigned_embeddings = [u for u in self.unassigned_embeddings if u[0] != crystal_id] | ||
|
|
There was a problem hiding this comment.
issue (bug_risk): Newly added crystals always persist with domain='unassigned' even when a matching domain is detected.
domain is always inserted as "unassigned", and although _assign_domain is called, its result never updates the row. When _detect_domain finds a matching domain, the DB still stores "unassigned", breaking immediate domain-based filtering in get() and distorting stats. Please either insert using the detected domain or issue an update on crystals.domain right after _assign_domain when it returns a concrete domain.
| for col, dt in [("created_at", "REAL DEFAULT 0"), ("level", "INTEGER DEFAULT 0"), | ||
| ("compressed_to", "INTEGER"), ("is_session", "INTEGER DEFAULT 0"), | ||
| ("session_id", "TEXT"), ("expires_at", "REAL")]: | ||
| try: |
There was a problem hiding this comment.
issue (bug_risk): FAISS remove_ids is called with crystal IDs, but index entries are added without explicit IDs, causing mismatches.
The indices are IndexFlatIP / IndexIVFFlat populated via add, so FAISS uses implicit IDs (0..ntotal-1). In _clean_expired, delete, and optimize, remove_ids is called with DB crystals.id, which do not correspond to those implicit IDs. Unless the index is wrapped in IndexIDMap and vectors were inserted with add_with_ids using these same IDs, remove_ids will delete the wrong entries or fail. To support deletion by crystal ID, use IndexIDMap/IndexIDMap2 with add_with_ids, or consistently delete by FAISS’s internal IDs (e.g., via a maintained mapping).
Comment on lines
+364
to
+373
| def _cluster_unassigned(self, conn): | ||
| """Greedy clustering on all unassigned embeddings to form new domains.""" | ||
| if len(self.unassigned_embeddings) < MIN_DOMAIN_SIZE: | ||
| return | ||
|
|
||
| embeddings = [u[1] for u in self.unassigned_embeddings] | ||
| n = len(embeddings) | ||
| used = [False] * n | ||
|
|
||
| for i in range(n): |
There was a problem hiding this comment.
suggestion (performance): Unassigned clustering recomputes centroids for each candidate, leading to quadratic–cubic time complexity.
Within _cluster_unassigned, cluster_embs and the centroid are recomputed inside the inner loop for every j, making clustering O(n^3) as clusters grow and potentially slowing or blocking adds for large unassigned pools. Consider maintaining a running centroid per cluster and updating it incrementally when adding a member, or at minimum computing and caching the centroid once per outer iteration instead of per candidate.
Suggested implementation:
for i in range(n):
if used[i]:
continue
# Start a new cluster with embedding i
cluster = [i]
used[i] = True
# Initialize running centroid and cluster size
cluster_size = 1
centroid = embeddings[i]
centroid_norm = np.linalg.norm(centroid)
if centroid_norm != 0:
centroid = centroid / centroid_norm
for j in range(i + 1, n): for j in range(i + 1, n):
if used[j]:
continue
# Check similarity to current cluster centroid (maintained incrementally)
candidate = embeddings[j]
candidate_norm = np.linalg.norm(candidate)
if candidate_norm == 0:
continue
candidate_unit = candidate / candidate_norm
sim_to_centroid = float(np.dot(centroid, candidate_unit))
if sim_to_centroid >= UNASSIGNED_SIM_THRESHOLD:
cluster.append(j)
used[j] = True
# Update running centroid as mean of cluster embeddings
cluster_size += 1
centroid = centroid * ((cluster_size - 1) / cluster_size) + candidate_unit / cluster_size
# Renormalize centroid
centroid_norm = np.linalg.norm(centroid)
if centroid_norm != 0:
centroid = centroid / centroid_norm if len(cluster) >= MIN_DOMAIN_SIZE:
# Form domain from this cluster
# Centroid has been maintained incrementally; ensure it's normalized
centroid_norm = np.linalg.norm(centroid)
if centroid_norm != 0:
centroid = centroid / centroid_norm
Comment on lines
+134
to
+138
| self._faiss_to_crystal_id = [] # Maps FAISS index -> crystal_id | ||
| if os.path.exists(self.ip): | ||
| self._global_index = faiss.read_index(self.ip) | ||
| self._trained = True | ||
| self._load_faiss_mapping() |
There was a problem hiding this comment.
issue (bug_risk): FAISS-to-crystal ID mapping is kept in-memory only and can become inconsistent after selective deletions.
Selective deletions call remove_ids with DB IDs, but self._faiss_to_crystal_id is keyed by FAISS positions built from an ordered DB snapshot. After partial removals, this mapping no longer reflects the actual index, so _check_duplicate and get_many can return incorrect results when translating FAISS indices. Consider using IndexIDMap with add_with_ids so FAISS stores the DB IDs directly and the manual mapping can be removed, or ensure deletions are done by FAISS position and that the mapping is updated in lockstep with any adds/removes.
Comment on lines
+1
to
+10
| import memora, time, os | ||
|
|
||
| # Clean start | ||
| for f in ["fix_test.db", "fix_test_faiss.bin", "fix_test_bm25.pkl"]: | ||
| if os.path.exists(f): os.remove(f) | ||
|
|
||
| m = memora.Memory(db_path="fix_test.db") | ||
|
|
||
| # Test data: 30 memories (10 health, 10 work, 10 gaming) | ||
| memories = [ |
There was a problem hiding this comment.
issue (testing): Convert this script-style validation into proper automated tests with assertions
This file currently relies on print-based checks, which won’t cause CI failures on regressions. Please rework this into automated tests (e.g., pytest/unittest) with explicit assertions, such as: duplicates don’t increase total_memories, dup_id1 == dup_id2, and query results respecting domain isolation. That way regressions in dedup, domain handling, and retrieval will be caught automatically.
Comment on lines
+698
to
+701
| br = con.execute( | ||
| f"SELECT text, level FROM crystals WHERE user = ? AND domain = ? {ex} AND (expires_at IS NULL OR expires_at > ?) ORDER BY strength DESC, last_accessed DESC LIMIT ?", | ||
| [user, target_dom] + bp + [now, top_k] | ||
| ).fetchall() |
There was a problem hiding this comment.
security (python.sqlalchemy.security.sqlalchemy-execute-raw-query): Avoiding SQL string concatenation: untrusted input concatenated with raw SQL query can result in SQL Injection. In order to execute raw query safely, prepared statement should be used. SQLAlchemy provides TextualSQL to easily used prepared statement with named parameters. For complex SQL composition, use SQL Expression Language or Schema Definition Language. In most cases, SQLAlchemy ORM will be a better option.
Source: opengrep
Comment on lines
+703
to
+706
| br = con.execute( | ||
| f"SELECT text, level FROM crystals WHERE domain = ? {ex} AND (expires_at IS NULL OR expires_at > ?) ORDER BY strength DESC, last_accessed DESC LIMIT ?", | ||
| [target_dom] + bp + [now, top_k] | ||
| ).fetchall() |
There was a problem hiding this comment.
security (python.sqlalchemy.security.sqlalchemy-execute-raw-query): Avoiding SQL string concatenation: untrusted input concatenated with raw SQL query can result in SQL Injection. In order to execute raw query safely, prepared statement should be used. SQLAlchemy provides TextualSQL to easily used prepared statement with named parameters. For complex SQL composition, use SQL Expression Language or Schema Definition Language. In most cases, SQLAlchemy ORM will be a better option.
Source: opengrep
| if older_than: | ||
| where.append("created_at < ?"); params.append(older_than) | ||
| w = "WHERE " + " AND ".join(where) if where else "" | ||
| rows = con.execute(f"SELECT id FROM crystals {w}", params).fetchall() |
There was a problem hiding this comment.
security (python.sqlalchemy.security.sqlalchemy-execute-raw-query): Avoiding SQL string concatenation: untrusted input concatenated with raw SQL query can result in SQL Injection. In order to execute raw query safely, prepared statement should be used. SQLAlchemy provides TextualSQL to easily used prepared statement with named parameters. For complex SQL composition, use SQL Expression Language or Schema Definition Language. In most cases, SQLAlchemy ORM will be a better option.
Source: opengrep
| where.append("created_at < ?"); params.append(older_than) | ||
| w = "WHERE " + " AND ".join(where) if where else "" | ||
| rows = con.execute(f"SELECT id FROM crystals {w}", params).fetchall() | ||
| con.execute(f"DELETE FROM crystals {w}", params) |
There was a problem hiding this comment.
security (python.sqlalchemy.security.sqlalchemy-execute-raw-query): Avoiding SQL string concatenation: untrusted input concatenated with raw SQL query can result in SQL Injection. In order to execute raw query safely, prepared statement should be used. SQLAlchemy provides TextualSQL to easily used prepared statement with named parameters. For complex SQL composition, use SQL Expression Language or Schema Definition Language. In most cases, SQLAlchemy ORM will be a better option.
Source: opengrep
Comment on lines
+860
to
+861
| con.execute("DELETE FROM crystals WHERE id IN " + | ||
| f"({','.join('?' for _ in rem_ids)})", rem_ids) |
There was a problem hiding this comment.
security (python.sqlalchemy.security.sqlalchemy-execute-raw-query): Avoiding SQL string concatenation: untrusted input concatenated with raw SQL query can result in SQL Injection. In order to execute raw query safely, prepared statement should be used. SQLAlchemy provides TextualSQL to easily used prepared statement with named parameters. For complex SQL composition, use SQL Expression Language or Schema Definition Language. In most cases, SQLAlchemy ORM will be a better option.
Source: opengrep
Author
|
ok thanks i will fix it |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Brief description of what this PR adds or changes.
Type of Change
Agent Details (if adding new agent)
agents/your-agent-name/How to Run (if adding code)
Expected output:
Checklist
requirements.txtwith pinned versions.env.examplewith required env vars (no real keys!)metadata.yamladdedRelated Issues
Closes #
Summary by Sourcery
Add a Memora memory agent providing an auto-domain, deduplicated hybrid search memory layer for LLM applications.
New Features:
memora-memory.Enhancements:
Build:
setup.pyto package the Memora agent and declare dependencies for distribution asmemora-memory.Documentation:
Tests: