Add multimodal perception, quaternary encoding, and Qalb-aware analysis

backend/api/main.py

@@ -449,6 +449,17 @@ class WebhookCreateRequest(BaseModel):
 class SkillInstallRequest(BaseModel):
     name: str = Field(..., min_length=1, max_length=200)
 
+
+class MultimodalInput(BaseModel):
+    """Input for multimodal perception analysis (text + image + audio)."""
+
+    text: str = Field(default="", max_length=50000)
+    image_base64: str | None = None  # base64-encoded image
+    audio_base64: str | None = None  # base64-encoded audio
+    media_type: str = Field(default="image/png", max_length=50)
+    agent_id: str | None = None
+    qalb_state: str = Field(default="", max_length=50)
+
     # NOTE: Startup and shutdown logic is handled by the lifespan context manager above.
 
 
@@ -1082,6 +1093,46 @@ async def list_memories(memory_type: str | None = None, limit: int = 30):
     return {"results": results, "total": len(results)}
 
 
+# === PERCEPTION (Sam' + Basar - سَمْع + بَصَر) ===
+
+
+@app.post("/api/perception/analyze")
+async def analyze_multimodal(
+    req: MultimodalInput,
+    user: TokenPayload | None = Depends(get_current_user),
+):
+    """Analyze multimodal input through the QCA perception pipeline.
+
+    Accepts text, base64-encoded images, and base64-encoded audio.
+    Follows Quranic ordering: Sam' (hearing) before Basar (sight).
+    """
+    import base64 as b64
+
+    agent = None
+    if req.agent_id and req.agent_id in active_agents:
+        agent = active_agents[req.agent_id]
+    elif active_agents:
+        agent = next(iter(active_agents.values()))
+
+    if not agent:
+        raise HTTPException(503, "No agents available for perception analysis")
+
+    image_bytes = b64.b64decode(req.image_base64) if req.image_base64 else None
+    audio_bytes = b64.b64decode(req.audio_base64) if req.audio_base64 else None
+
+    if not req.text and not image_bytes and not audio_bytes:
+        raise HTTPException(400, "At least one of text, image_base64, or audio_base64 is required")
+
+    result = await agent.qca.process_input_multimodal(
+        text=req.text,
+        image_bytes=image_bytes,
+        audio_bytes=audio_bytes,
+        media_type=req.media_type,
+        qalb_state=req.qalb_state,
+    )
+    return {"result": result}
+
+
 # === KNOWLEDGE INGESTION (Ilm - عِلْم) ===
 
 
@@ -2580,6 +2631,51 @@ async def task_stream(chunk):
                         },
                     )
 
+            elif msg_type == "multimodal":
+                import base64 as b64
+
+                session_id = data.get("session_id", client_id)
+                text = data.get("content", "")
+                image_b64 = data.get("image_base64")
+                audio_b64 = data.get("audio_base64")
+                media_type = data.get("media_type", "image/png")
+                qalb_state = data.get("qalb_state", "")
+
+                image_bytes = b64.b64decode(image_b64) if image_b64 else None
+                audio_bytes = b64.b64decode(audio_b64) if audio_b64 else None
+
+                agent = next(iter(active_agents.values()), None) if active_agents else None
+                if agent:
+                    try:
+                        result = await agent.qca.process_input_multimodal(
+                            text=text,
+                            image_bytes=image_bytes,
+                            audio_bytes=audio_bytes,
+                            media_type=media_type,
+                            qalb_state=qalb_state,
+                        )
+                        await manager.send(
+                            client_id,
+                            {
+                                "type": "perception_result",
+                                "session_id": session_id,
+                                "result": result,
+                            },
+                        )
+                    except Exception as e:
+                        await manager.send(
+                            client_id,
+                            {
+                                "type": "error",
+                                "message": f"Multimodal processing failed: {e}",
+                            },
+                        )
+                else:
+                    await manager.send(
+                        client_id,
+                        {"type": "error", "message": "No agents available"},
+                    )
+
             elif msg_type == "command":
                 cmd = data.get("command", "").strip()
                 if cmd == "/status" or cmd == "status":

backend/memory/knowledge_graph.py

@@ -178,6 +178,50 @@ async def query_entity(self, name: str) -> dict:
             "incoming": incoming,
         }
 
+    async def search_entities(
+        self, query: str, limit: int = 10, entity_type: str | None = None
+    ) -> list[dict]:
+        """Search entities by name (LIKE query) with optional type filtering.
+
+        Used by MemoryPyramid layer 4 for graph-based recall.
+        """
+        conn = sqlite3.connect(self.db_path)
+        c = conn.cursor()
+
+        results = []
+        seen_ids: set[str] = set()
+        words = query.lower().split()
+
+        for word in words[:5]:
+            if len(word) < 2:
+                continue
+            if entity_type:
+                c.execute(
+                    "SELECT id, name, type, properties FROM kg_entities "
+                    "WHERE name LIKE ? AND type = ? LIMIT ?",
+                    (f"%{word}%", entity_type, limit),
+                )
+            else:
+                c.execute(
+                    "SELECT id, name, type, properties FROM kg_entities "
+                    "WHERE name LIKE ? LIMIT ?",
+                    (f"%{word}%", limit),
+                )
+            for row in c.fetchall():
+                if row[0] not in seen_ids:
+                    seen_ids.add(row[0])
+                    results.append(
+                        {
+                            "id": row[0],
+                            "name": row[1],
+                            "type": row[2],
+                            "properties": json.loads(row[3]) if row[3] else {},
+                        }
+                    )
+
+        conn.close()
+        return results[:limit]
+
     async def get_stats(self) -> dict:
         """Get knowledge graph statistics"""
         conn = sqlite3.connect(self.db_path)

backend/memory/lawh_mahfuz.py

@@ -37,6 +37,7 @@ class LawhEntry:
     length: int       # len(content) in bytes
     certainty: float = 1.0
     category: str = "general"
+    quaternary_checksum: str = ""  # DNA-inspired quaternary checksum (ACGT)
 
     def to_dict(self) -> dict:
         return {
@@ -46,6 +47,7 @@ def to_dict(self) -> dict:
             "certainty": self.certainty,
             "category": self.category,
             "stored_at": self.stored_at,
+            "quaternary_checksum": self.quaternary_checksum,
         }
 
 
@@ -108,6 +110,11 @@ def _init_db(self):
                     quarantined_at REAL
                 )
             """)
+            # Migration: add quaternary_checksum column if not present
+            try:
+                c.execute("ALTER TABLE lawh_entries ADD COLUMN quaternary_checksum TEXT DEFAULT ''")
+            except sqlite3.OperationalError:
+                pass  # Column already exists
             conn.commit()
         finally:
             conn.close()
@@ -119,11 +126,13 @@ def _load_into_cache(self):
             c = conn.cursor()
             c.execute("SELECT * FROM lawh_entries")
             for row in c.fetchall():
-                key, content, source, stored_at, sha256, crc32, length, certainty, category = row
+                key, content, source, stored_at, sha256, crc32, length, certainty, category = row[:9]
+                quat_checksum = row[9] if len(row) > 9 else ""
                 entry = LawhEntry(
                     key=key, content=content, source=source, stored_at=stored_at,
                     sha256=sha256, crc32=crc32, length=length,
                     certainty=certainty, category=category,
+                    quaternary_checksum=quat_checksum or "",
                 )
                 self._cache[key] = entry
             logger.info("[LAWH] Loaded %d entries from preserved tablet", len(self._cache))
@@ -152,27 +161,36 @@ def store_immutable(
         length = len(content.encode("utf-8"))
         now = time.time()
 
+        # Compute quaternary checksum (DNA-inspired integrity layer)
+        quat_checksum = ""
+        try:
+            from memory.quaternary import quaternary_checksum
+            quat_checksum = quaternary_checksum(content)
+        except ImportError:
+            pass
+
         entry = LawhEntry(
             key=key, content=content, source=source, stored_at=now,
             sha256=sha, crc32=crc, length=length,
             certainty=certainty, category=category,
+            quaternary_checksum=quat_checksum,
         )
 
         conn = self._get_conn()
         try:
             c = conn.cursor()
             c.execute(
                 """INSERT OR IGNORE INTO lawh_entries
-                   (key, content, source, stored_at, sha256, crc32, length, certainty, category)
-                   VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)""",
-                (key, content, source, now, sha, crc, length, certainty, category),
+                   (key, content, source, stored_at, sha256, crc32, length, certainty, category, quaternary_checksum)
+                   VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
+                (key, content, source, now, sha, crc, length, certainty, category, quat_checksum),
             )
             conn.commit()
         finally:
             conn.close()
 
         self._cache[key] = entry
-        logger.debug("[LAWH] Stored '%s' (len=%d sha=%s...)", key, length, sha[:8])
+        logger.debug("[LAWH] Stored '%s' (len=%d sha=%s... quat=%s...)", key, length, sha[:8], quat_checksum[:8] if quat_checksum else "n/a")
         return key
 
     def verify_integrity(self, key: str) -> bool:
@@ -203,6 +221,20 @@ def verify_integrity(self, key: str) -> bool:
             self._quarantine_entry(key, f"Length mismatch: {actual_len}≠{entry.length}")
             return False
 
+        # 4th check: Quaternary checksum (DNA-inspired, when present)
+        if entry.quaternary_checksum:
+            try:
+                from memory.quaternary import hamming_distance, quaternary_checksum
+                actual_quat = quaternary_checksum(entry.content)
+                dist = hamming_distance(actual_quat, entry.quaternary_checksum)
+                if dist > 0:
+                    self._quarantine_entry(
+                        key, f"Quaternary checksum drift: hamming_distance={dist}"
+                    )
+                    return False
+            except ImportError:
+                pass  # Quaternary module not available; skip this check
+
         return True
 
     def get(self, key: str) -> LawhEntry | None:

backend/memory/living_memory.py

@@ -167,7 +167,7 @@ class LivingMemorySystem:
     context-dependent recall, and Dhikr maintenance daemon.
     """
 
-    def __init__(self, masalik=None, dhikr_db=None, lawh=None):
+    def __init__(self, masalik=None, dhikr_db=None, lawh=None, vector_store=None):
         # Memory stores by level
         self.sadr: list[MemoryTrace] = []       # working memory (capacity-limited)
         self.traces: dict[str, MemoryTrace] = {}  # all traces by ID
@@ -177,6 +177,7 @@ def __init__(self, masalik=None, dhikr_db=None, lawh=None):
         self._masalik = masalik     # MasalikNetwork for spread activation
         self._dhikr_db = dhikr_db  # DhikrMemorySystem for persistence
         self._lawh = lawh          # LawhMahfuz for immutable storage
+        self._vector_store = vector_store  # VectorStore for semantic similarity
 
         self._daemon_cycle = 0
         self._content_hashes: dict[str, str] = {}  # hash → trace_id for fast lookup
@@ -620,19 +621,81 @@ def _store_trace(self, trace: MemoryTrace) -> None:
             except Exception:
                 pass
 
+        # Persist to VectorStore for semantic similarity search
+        if self._vector_store and getattr(self._vector_store, "_available", False):
+            try:
+                import asyncio
+                import concurrent.futures
+
+                metadata = {"level": trace.level.value, "importance": trace.importance}
+                coro = self._vector_store.store(
+                    trace.content, memory_id=trace.trace_id, metadata=metadata
+                )
+                try:
+                    asyncio.get_running_loop()
+                    with concurrent.futures.ThreadPoolExecutor(max_workers=1) as executor:
+                        executor.submit(asyncio.run, coro).result(timeout=5)
+                except RuntimeError:
+                    asyncio.run(coro)
+            except Exception:
+                pass
+
     def _find_best_match(self, content: str) -> tuple[MemoryTrace | None, float]:
-        """Find the most similar existing trace."""
+        """Find the most similar existing trace (text + optional vector similarity)."""
         best = None
         best_sim = 0.0
+
+        # Primary: Jaccard text similarity (always available)
         for trace in self.traces.values():
             if trace.trace_id in self.archive:
                 continue
             sim = self._text_similarity(content, trace.content)
             if sim > best_sim:
                 best_sim = sim
                 best = trace
+
+        # Secondary: Vector similarity via ChromaDB (if available)
+        if self._vector_store and getattr(self._vector_store, "_available", False):
+            try:
+                # VectorStore.search() is async but underlying ChromaDB is sync.
+                # Use sync wrapper to avoid event loop issues.
+                vector_results = self._vector_search_sync(content, limit=5)
+                for vr in vector_results:
+                    distance = vr.get("distance", 1.0)
+                    # ChromaDB L2 distance → 0-1 similarity
+                    vector_sim = max(0.0, 1.0 - (distance / 2.0))
+                    vec_id = vr.get("id", "")
+                    if vec_id in self.traces and vec_id not in self.archive:
+                        trace = self.traces[vec_id]
+                        hybrid_sim = max(
+                            self._text_similarity(content, trace.content),
+                            vector_sim,
+                        )
+                        if hybrid_sim > best_sim:
+                            best_sim = hybrid_sim
+                            best = trace
+            except Exception:
+                pass  # Graceful degradation to text-only
+
         return best, best_sim
 
+    def _vector_search_sync(self, query: str, limit: int = 5) -> list[dict]:
+        """Synchronous wrapper for VectorStore.search()."""
+        import asyncio
+        import concurrent.futures
+
+        try:
+            asyncio.get_running_loop()
+            # Already in async context — run in thread to avoid nested loop
+            with concurrent.futures.ThreadPoolExecutor(max_workers=1) as executor:
+                future = executor.submit(
+                    asyncio.run, self._vector_store.search(query, limit=limit)
+                )
+                return future.result(timeout=5)
+        except RuntimeError:
+            # No running event loop — safe to run directly
+            return asyncio.run(self._vector_store.search(query, limit=limit))
+
     def _extract_novel_parts(self, new_content: str, existing: MemoryTrace) -> str:
         """Extract what's genuinely new in content vs existing trace."""
         new_words = set(new_content.lower().split())