fix: block recursive BrainLayer ingest

src/brainlayer/drain.py

@@ -28,6 +28,7 @@
     merge_existing_chunk_content,
     merge_existing_chunk_seen,
 )
+from .ingest_guard import recursive_mcp_output_reason
 from .paths import get_db_path
 
 logger = logging.getLogger(__name__)
@@ -159,6 +160,10 @@ def _apply_store(conn: apsw.Connection, event: dict[str, Any]) -> ApplyResult:
     if not content:
         logger.warning("Skipping malformed store event with empty content")
         return ApplyResult()
+    recursive_reason = recursive_mcp_output_reason(content)
+    if recursive_reason:
+        logger.warning("Skipping recursive MCP store event: %s", recursive_reason)
+        return ApplyResult()
     now = datetime.now(timezone.utc).isoformat()
     metadata = {"memory_type": event.get("memory_type", "note")}
     raw_metadata = event.get("metadata")
@@ -239,6 +244,10 @@ def _apply_watcher(conn: apsw.Connection, event: dict[str, Any]) -> None:
     if not content:
         logger.warning("Skipping malformed watcher event with empty content")
         return
+    recursive_reason = recursive_mcp_output_reason(content)
+    if recursive_reason:
+        logger.warning("Skipping recursive MCP watcher event: %s", recursive_reason)
+        return
     tags = event.get("tags")
     _insert_or_merge_chunk(
         conn,
@@ -270,6 +279,10 @@ def _apply_hook(conn: apsw.Connection, event: dict[str, Any]) -> None:
     if not content:
         logger.warning("Skipping malformed hook event with empty content")
         return
+    recursive_reason = recursive_mcp_output_reason(content)
+    if recursive_reason:
+        logger.warning("Skipping recursive MCP hook event: %s", recursive_reason)
+        return
     content_hash = event.get("content_hash") or hashlib.sha256(content.encode()).hexdigest()[:16]
     session_id = event.get("session_id") or "unknown"
     chunk_id = event.get("chunk_id") or f"rt-{str(session_id)[:8]}-{content_hash}"

src/brainlayer/engine.py

@@ -178,6 +178,7 @@ def think(
     embed_fn: Any,
     project: str | None = None,
     max_results: int = 10,
+    include_audit: bool = False,
 ) -> ThinkResult:
     """Given current task context, retrieve relevant past knowledge.
 
@@ -206,6 +207,7 @@ def think(
         n_results=max_results,
         project_filter=project,
         importance_min=3.0,  # Skip low-importance noise
+        include_audit=include_audit,
     )
 
     if not results["documents"][0]:
@@ -239,6 +241,7 @@ def recall(
     topic: str | None = None,
     project: str | None = None,
     max_results: int = 10,
+    include_audit: bool = False,
 ) -> RecallResult:
     """Proactive smart retrieval based on file or topic.
 
@@ -278,6 +281,7 @@ def recall(
                 query_text=fname,
                 n_results=max_results,
                 project_filter=project,
+                include_audit=include_audit,
             )
             for doc, meta in zip(search_results["documents"][0], search_results["metadatas"][0]):
                 result.related_chunks.append(
@@ -299,6 +303,7 @@ def recall(
             query_text=topic,
             n_results=max_results,
             project_filter=project,
+            include_audit=include_audit,
         )
         for doc, meta in zip(search_results["documents"][0], search_results["metadatas"][0]):
             result.related_chunks.append(

src/brainlayer/ingest_guard.py

@@ -0,0 +1,40 @@
+"""Write-side guards for content that must never enter BrainLayer."""
+
+from __future__ import annotations
+
+import re
+
+_JSONRPC_MESSAGE_RE = re.compile(r'"jsonrpc"\s*:\s*"2\.0"', re.IGNORECASE)
+_INVALID_JSONRPC_MARKER = "mcp brainlayer memory: invalid json-rpc message"
+_BRAIN_SEARCH_BOX_PREFIX = "┌─ brain_search:"
+_BRAINLAYER_BOX_PREFIX_RE = re.compile(
+    r"^┌─\s*(?:brain_[a-z_]+|entity(?:\s+search)?):",
+    re.IGNORECASE,
+)
+
+
+def recursive_mcp_output_reason(content: str | None) -> str | None:
+    """Return a reason when content is BrainLayer MCP output being re-ingested."""
+    if not content:
+        return None
+
+    stripped = str(content).lstrip()
+    if stripped.startswith(_BRAIN_SEARCH_BOX_PREFIX):
+        return "brain_search_output"
+    if _BRAINLAYER_BOX_PREFIX_RE.match(stripped):
+        return "brainlayer_mcp_output"
+
+    folded = stripped.casefold()
+    if _INVALID_JSONRPC_MARKER in folded:
+        return "invalid_jsonrpc_mcp_output"
+    if _JSONRPC_MESSAGE_RE.search(stripped):
+        return "jsonrpc_message"
+
+    return None
+
+
+def reject_recursive_mcp_output(content: str | None) -> None:
+    """Raise ValueError when content is recursive BrainLayer MCP output."""
+    reason = recursive_mcp_output_reason(content)
+    if reason:
+        raise ValueError(f"recursive MCP output is not stored in BrainLayer: {reason}")

src/brainlayer/kg_repo.py

@@ -480,17 +480,27 @@ def set_entity_parent(self, entity_id: str, parent_id: str) -> None:
             (parent_id, entity_id),
         )
 
-    def get_entity_chunks(self, entity_id: str, limit: int = 20) -> List[Dict[str, Any]]:
+    def get_entity_chunks(
+        self,
+        entity_id: str,
+        limit: int = 20,
+        *,
+        include_audit: bool = False,
+    ) -> List[Dict[str, Any]]:
         """Get chunks linked to an entity, ordered by relevance."""
         cursor = self._read_cursor()
+        where_clauses = ["ec.entity_id = ?"]
+        if not include_audit:
+            where_clauses.append(self._audit_recursion_exclusion_sql("c.id", "c.tags", "c.content"))
+        where_sql = " AND ".join(where_clauses)
         rows = list(
             cursor.execute(
-                """
+                f"""
                 SELECT ec.chunk_id, ec.relevance, ec.context, ec.mention_type,
                        c.content, c.source_file, c.project, c.content_type, c.created_at
                 FROM kg_entity_chunks ec
                 JOIN chunks c ON ec.chunk_id = c.id
-                WHERE ec.entity_id = ?
+                WHERE {where_sql}
                 ORDER BY ec.relevance DESC
                 LIMIT ?
                 """,
@@ -982,6 +992,7 @@ def kg_search(
         relation_type: Optional[str] = None,
         limit: int = 20,
         include_checkpoints: bool = False,
+        include_audit: bool = False,
     ) -> List[Dict[str, Any]]:
         """Structured KG fact retrieval. Excludes co_occurs_with noise."""
         results: List[Dict[str, Any]] = []
@@ -990,19 +1001,32 @@ def kg_search(
         if entity:
             cursor = self._read_cursor()
 
-            checkpoint_join = ""
+            source_chunk_join = ""
             checkpoint_filter = ""
+            audit_filter = ""
             checkpoint_params: list[str] = []
+            needs_source_chunk = (
+                not include_checkpoints and getattr(self, "_has_chunk_origin", True)
+            ) or not include_audit
+            if needs_source_chunk:
+                source_chunk_join = "LEFT JOIN chunks source_chunk ON r.source_chunk_id = source_chunk.id"
             if not include_checkpoints and getattr(self, "_has_chunk_origin", True):
-                checkpoint_join = "LEFT JOIN chunks source_chunk ON r.source_chunk_id = source_chunk.id"
-                checkpoint_filter = """
+                checkpoint_clause = self._checkpoint_exclusion_clause("source_chunk")
+                checkpoint_filter = f"""
+                          AND (
+                              r.source_chunk_id IS NULL
+                              OR source_chunk.id IS NULL
+                              OR ({checkpoint_clause})
+                          )
+                """
+            if not include_audit:
+                audit_filter = f"""
                           AND (
                               r.source_chunk_id IS NULL
                               OR source_chunk.id IS NULL
-                              OR COALESCE(source_chunk.chunk_origin, 'unknown') != ?
+                              OR {self._audit_recursion_exclusion_sql("source_chunk.id", "source_chunk.tags", "source_chunk.content")}
                           )
                 """
-                checkpoint_params.append("precompact_checkpoint")
 
             if relation_type:
                 type_filter_src = "AND r.relation_type = ?"
@@ -1024,10 +1048,11 @@ def kg_search(
                     FROM kg_current_facts r
                     JOIN kg_entities se ON r.source_id = se.id
                     JOIN kg_entities te ON r.target_id = te.id
-                    {checkpoint_join}
+                    {source_chunk_join}
                     WHERE ((r.source_id = ? {type_filter_src})
                        OR (r.target_id = ? {type_filter_tgt}))
                     {checkpoint_filter}
+                    {audit_filter}
                     ORDER BY r.importance DESC, r.confidence DESC
                     LIMIT ?
                     """,
@@ -1088,6 +1113,7 @@ def kg_hybrid_search(
             relation_type=relation_type,
             limit=n_results,
             include_checkpoints=bool(kwargs.get("include_checkpoints", False)),
+            include_audit=bool(kwargs.get("include_audit", False)),
         )
 
         scored_facts = []

src/brainlayer/mcp/__init__.py

 elif name == "brain_recall": 
 elif name == "brain_recall": 
@@ -511,6 +511,11 @@ async def list_tools() -> list[Tool]:
                             "default": False,
                             "description": "Include PreCompact checkpoint chunks in search results. Defaults to false; use brain_resume for explicit session recovery.",
                         },
+                        "include_audit": {
+                            "type": "boolean",
+                            "default": False,
+                            "description": "Opt in to audit/eval and recursive MCP-output memories. Defaults false to prevent audit-recursion pollution.",
+                        },
                         "detail": {
                             "type": "string",
                             "enum": ["compact", "full"],
@@ -828,6 +833,11 @@ async def list_tools() -> list[Tool]:
                             "default": False,
                             "description": "Include PreCompact checkpoint chunks in mode=search results. Defaults to false; use brain_resume for explicit session recovery.",
                         },
+                        "include_audit": {
+                            "type": "boolean",
+                            "default": False,
+                            "description": "Opt in to audit/eval and recursive MCP-output memories in mode=search. Defaults false to prevent audit-recursion pollution.",
+                        },
                     },
                 }
             ),
@@ -934,6 +944,11 @@ async def list_tools() -> list[Tool]:
                             "minimum": 0,
                             "description": "Pagination offset for list action.",
                         },
+                        "include_audit": {
+                            "type": "boolean",
+                            "default": False,
+                            "description": "Opt in to audit/eval and recursive MCP-output evidence. Defaults false to prevent audit-recursion pollution.",
+                        },
                     },
                     "required": [],
                 }
@@ -1255,6 +1270,7 @@ async def call_tool(name: str, arguments: dict[str, Any]):
                 source_filter=resolved_source_filter,
                 correction_category=arguments.get("correction_category"),
                 include_checkpoints=arguments.get("include_checkpoints", False),
+                include_audit=arguments.get("include_audit", False),
             )
         )
 
@@ -1343,6 +1359,7 @@ async def call_tool(name: str, arguments: dict[str, Any]):
                 detail=arguments.get("detail", "compact"),
                 entity_type=arguments.get("entity_type"),
                 include_checkpoints=arguments.get("include_checkpoints", False),
+                include_audit=arguments.get("include_audit", False),
             )
         )
 
@@ -1395,6 +1412,7 @@ async def call_tool(name: str, arguments: dict[str, Any]):
                 mode="entity",
                 query=query,
                 entity_type=arguments.get("entity_type"),
+                include_audit=arguments.get("include_audit", False),
             )
         )
 

src/brainlayer/mcp/entity_handler.py

@@ -16,6 +16,7 @@
 async def _brain_entity(
     query: str,
     entity_type: str | None = None,
+    include_audit: bool = False,
 ) -> CallToolResult:
     """Handle brain_entity tool call."""
     from ..pipeline.digest import entity_lookup
@@ -32,6 +33,7 @@ async def _brain_entity(
                 store=store,
                 embed_fn=model.embed_query,
                 entity_type=entity_type,
+                include_audit=include_audit,
             ),
         )
     except Exception as e: