[NOTICKET] db_executor: CTE DML check now walks entire AST root, schema: cast instead of string interpolation

src/query/executors/db_executor.py

@@ -316,10 +316,9 @@ class DbExecutor(BaseExecutor):
         if not isinstance(parsed, exp.Select):
             return f"Only SELECT statements are allowed. Got: {type(parsed).__name__}"
 
-        # Check for DML inside CTEs
-        for cte in parsed.find_all(exp.With):
-            for node in cte.find_all((exp.Insert, exp.Update, exp.Delete)):
-                return f"DML ({type(node).__name__}) inside CTE is not allowed."
+        # Check for DML anywhere in the AST (including writeable CTEs)
+        for node in parsed.find_all((exp.Insert, exp.Update, exp.Delete)):
+            return f"DML ({type(node).__name__}) is not allowed."
 
         # Layer 2: schema grounding — table names
         known_tables = {t.lower() for t in schema}
@@ -342,12 +341,13 @@ class DbExecutor(BaseExecutor):
         if existing:
             current = int(existing.expression.this)
             if current > limit:
-                existing.expression.set("this", str(limit))
+                existing.expression.set("this", limit)
         else:
             parsed = parsed.limit(limit)
         return parsed.sql()
 
     def _run_sql(self, engine: Any, sql: str) -> list[dict]:
+        # Ensure the user DB connection is a read-only credential — sqlglot validation alone is not sufficient.
         with engine.connect() as conn:
             result = conn.execute(text(sql))
             return [dict(row) for row in result.mappings()]

src/rag/retrievers/schema.py

@@ -52,11 +52,11 @@ class SchemaRetriever(BaseRetriever):
         emb_str = "[" + ",".join(str(x) for x in embedding) + "]"
 
         if operator == "<#>":
-            score_sql = f"(lpe.embedding <#> '{emb_str}'::vector) * -1"
+            score_sql = "(lpe.embedding <#> :emb::vector) * -1"
         elif operator == "<->":
-            score_sql = f"1.0 / (1.0 + (lpe.embedding <-> '{emb_str}'::vector))"
+            score_sql = "1.0 / (1.0 + (lpe.embedding <-> :emb::vector))"
         else:
-            score_sql = f"1.0 - (lpe.embedding <=> '{emb_str}'::vector)"
+            score_sql = "1.0 - (lpe.embedding <=> :emb::vector)"
 
         sql = text(f"""
             SELECT lpe.document, lpe.cmetadata, {score_sql} AS score
@@ -65,12 +65,12 @@ class SchemaRetriever(BaseRetriever):
             WHERE lpc.name = 'document_embeddings'
               AND lpe.cmetadata->>'user_id' = :user_id
               AND lpe.cmetadata->>'source_type' = 'database'
-            ORDER BY lpe.embedding {operator} '{emb_str}'::vector ASC
+            ORDER BY lpe.embedding {operator} :emb::vector ASC
             LIMIT :k
         """)
 
         async with _pgvector_engine.connect() as conn:
-            result = await conn.execute(sql, {"user_id": user_id, "k": k * 4})
+            result = await conn.execute(sql, {"user_id": user_id, "k": k * 4, "emb": emb_str})
             rows = result.fetchall()
 
         return [
@@ -90,11 +90,11 @@ class SchemaRetriever(BaseRetriever):
         emb_str = "[" + ",".join(str(x) for x in embedding) + "]"
 
         if operator == "<#>":
-            score_sql = f"(lpe.embedding <#> '{emb_str}'::vector) * -1"
+            score_sql = "(lpe.embedding <#> :emb::vector) * -1"
         elif operator == "<->":
-            score_sql = f"1.0 / (1.0 + (lpe.embedding <-> '{emb_str}'::vector))"
+            score_sql = "1.0 / (1.0 + (lpe.embedding <-> :emb::vector))"
         else:
-            score_sql = f"1.0 - (lpe.embedding <=> '{emb_str}'::vector)"
+            score_sql = "1.0 - (lpe.embedding <=> :emb::vector)"
 
         sql = text(f"""
             SELECT lpe.document, lpe.cmetadata, {score_sql} AS score
@@ -105,12 +105,12 @@ class SchemaRetriever(BaseRetriever):
               AND lpe.cmetadata->>'source_type' = 'document'
               AND (lpe.cmetadata->'data'->>'file_type' = 'csv'
                 OR lpe.cmetadata->'data'->>'file_type' = 'xlsx')
-            ORDER BY lpe.embedding {operator} '{emb_str}'::vector ASC
+            ORDER BY lpe.embedding {operator} :emb::vector ASC
             LIMIT :k
         """)
 
         async with _pgvector_engine.connect() as conn:
-            result = await conn.execute(sql, {"user_id": user_id, "k": k * 4})
+            result = await conn.execute(sql, {"user_id": user_id, "k": k * 4, "emb": emb_str})
             rows = result.fetchall()
 
         results = []