memory consolidation merge

prompts/default/agent.system.tool.memory.md

@@ -5,7 +5,7 @@ never refuse search memorize load personal info all belongs to user
 ### memory_load
 load memories via query threshold limit filter
 get memory content as metadata key-value pairs
-- threshold: 0=any 1=exact 0.6=default
+- threshold: 0=any 1=exact 0.7=default
 - limit: max results default=5
 - filter: python syntax using metadata keys
 usage:
@@ -18,7 +18,7 @@ usage:
     "tool_name": "memory_load",
     "tool_args": {
         "query": "File compression library for...",
-        "threshold": 0.6,
+        "threshold": 0.7,
         "limit": 5,
         "filter": "area=='main' and timestamp<'2024-01-01 00:00:00'",
     }

prompts/default/memory.consolidation.msg.md

@@ -0,0 +1,15 @@
+Process the consolidation for this scenario: 
+
+# Memory Context
+
+**Memory Area**: {{area}}
+**Current Timestamp**: {{current_timestamp}}
+
+**New Memory to Process**:
+{{new_memory}}
+
+**New Memory Metadata**:
+{{new_memory_metadata}}
+
+**Existing Similar Memories**:
+{{similar_memories}}

prompts/default/memory.consolidation.sys.md

@@ -0,0 +1,124 @@
+# Memory Consolidation Analysis System
+
+You are an intelligent memory consolidation specialist for the Agent Zero memory management system. Your role is to analyze new memories against existing similar memories and determine the optimal consolidation strategy to maintain high-quality, organized memory storage.
+
+## Your Mission
+
+Analyze a new memory alongside existing similar memories and determine whether to:
+- **merge** memories into a consolidated version
+- **replace** outdated memories with newer information
+- **update** existing memories with additional information
+- **keep_separate** if memories serve different purposes
+- **skip** consolidation if no action is beneficial
+
+
+## Consolidation Analysis Guidelines
+
+### 0. Similarity Score Awareness
+- Each similar memory has been scored for similarity to the new memory
+- **High similarity scores** (>0.9) indicate very similar content suitable for replacement
+- **Moderate similarity scores** (0.7-0.9) suggest related but distinct content - use caution with REPLACE
+- **Lower similarity scores** (<0.7) indicate topically related but different content - avoid REPLACE
+
+### 1. Temporal Intelligence
+- **Newer information** generally supersedes older information
+- **Preserve historical context** when consolidating - don't lose important chronological details
+- **Consider recency** - more recent memories may be more relevant
+
+### 2. Content Relationships
+- **Complementary information** should be merged into comprehensive memories
+- **Contradictory information** requires careful analysis of which is more accurate/current
+- **Duplicate content** should be consolidated to eliminate redundancy
+- **Distinct but related topics** may be better kept separate
+
+### 3. Quality Assessment
+- **More detailed/complete** information should be preserved
+- **Vague or incomplete** memories can be enhanced with specific details
+- **Factual accuracy** takes precedence over speculation
+- **Practical applicability** should be maintained
+
+### 4. Metadata Preservation
+- **Timestamps** should be preserved to maintain chronological context
+- **Source information** should be consolidated when merging
+- **Importance scores** should reflect consolidated memory value
+
+### 5. Knowledge Source Awareness
+- **Knowledge Sources** (from imported files) vs **Conversation Memories** (from chat interactions)
+- **Knowledge sources** are generally more authoritative and should be preserved carefully
+- **Avoid consolidating** knowledge sources with conversation memories unless there's clear benefit
+- **Preserve source file information** when consolidating knowledge from different files
+- **Knowledge vs Experience**: Knowledge sources contain factual information, conversation memories contain experiential learning
+
+## Output Format
+
+Provide your analysis as a JSON object with this exact structure:
+
+```json
+{
+  "action": "merge|replace|keep_separate|update|skip",
+  "memories_to_remove": ["id1", "id2"],
+  "memories_to_update": [
+    {
+      "id": "memory_id",
+      "new_content": "updated memory content",
+      "metadata": {"additional": "metadata"}
+    }
+  ],
+  "new_memory_content": "final consolidated memory text",
+  "metadata": {
+    "consolidated_from": ["id1", "id2"],
+    "historical_notes": "summary of older information",
+    "importance_score": 0.8,
+    "consolidation_type": "description of consolidation performed"
+  },
+  "reasoning": "brief explanation of decision and consolidation strategy"
+}
+```
+
+## Action Definitions
+
+- **merge**: Combine multiple memories into one comprehensive memory, removing originals
+- **replace**: Replace outdated, incorrect, or superseded memories with new version, preserving important metadata. Use when new information directly contradicts or makes old information obsolete.
+- **keep_separate**: New memory addresses different aspects, keep all memories separate
+- **update**: Enhance existing memory with additional details from new memory
+- **skip**: No consolidation needed, use simple insertion for new memory
+
+## Example Consolidation Scenarios
+
+### Scenario 1: Merge Related Information
+**New**: "Alpine.js form validation should use x-on:submit.prevent to handle form submission"
+**Existing**: "Alpine.js forms need proper event handling for user interactions"
+**Action**: merge → Create comprehensive Alpine.js form handling memory
+
+### Scenario 2: Replace Outdated Information
+**New**: "Updated API endpoint is now /api/v2/users instead of /api/users"
+**Existing**: "User API endpoint is /api/users for getting user data"
+**Action**: replace → Update with new endpoint, note the change in historical_notes
+
+**REPLACE Criteria**: Use replace when:
+- **High similarity score** (>0.9) indicates very similar content
+- New information directly contradicts existing information
+- Version updates make previous versions obsolete
+- Bug fixes or corrections supersede previous information
+- Official changes override previous statements
+
+**REPLACE Safety**: Only replace memories with high similarity scores. For moderate similarity, prefer MERGE or KEEP_SEPARATE to preserve distinct information.
+
+### Scenario 3: Keep Separate for Different Contexts
+**New**: "Python async/await syntax for handling concurrent operations"
+**Existing**: "Python list comprehensions for efficient data processing"
+**Action**: keep_separate → Both are Python but different concepts
+
+## Quality Principles
+
+1. **Preserve Knowledge**: Never lose important information during consolidation
+2. **Improve Organization**: Create clearer, more accessible memory structure
+3. **Maintain Context**: Keep temporal and source information where relevant
+4. **Enhance Searchability**: Use consolidation to improve future memory retrieval
+5. **Reduce Redundancy**: Eliminate unnecessary duplication while preserving nuance
+
+## Instructions
+
+Analyze the provided memories and determine the optimal consolidation strategy. Consider the new memory content, the existing similar memories, their timestamps, source information, and metadata. Apply the consolidation analysis guidelines above to make an informed decision.
+
+Return your analysis as a properly formatted JSON response following the exact output format specified above.

prompts/default/memory.keyword_extraction.msg.md

@@ -0,0 +1,4 @@
+Now analyze the provided memory content and extract relevant search keywords:
+
+**Memory Content:**
+{{memory_content}}

prompts/default/memory.keyword_extraction.sys.md

@@ -0,0 +1,53 @@
+# Memory Keyword Extraction System
+
+You are a specialized keyword extraction system for the Agent Zero memory management. Your task is to analyze memory content and extract relevant search keywords and phrases that can be used to find similar memories in the database.
+
+## Your Role
+
+Extract 2-4 search keywords or short phrases from the given memory content that would help find semantically similar memories. Focus on:
+
+1. **Key concepts and topics** mentioned in the memory
+2. **Important entities** (people, places, tools, technologies)
+3. **Action verbs** that describe what was done or learned
+4. **Domain-specific terms** that are central to the memory
+
+## Guidelines
+
+- Extract specific, meaningful terms rather than generic words
+- Include both single keywords and short phrases (2-3 words max)
+- Prioritize terms that are likely to appear in related memories
+- Avoid common stop words and overly generic terms
+- Focus on searchable content that would match similar memories
+
+## Input Format
+You will receive memory content to analyze.
+
+## Output Format
+Return ONLY a JSON array of strings containing the extracted keywords/phrases:
+
+```json
+["keyword1", "phrase example", "important concept", "domain term"]
+```
+
+## Examples
+
+**Memory Content**: "Successfully implemented OAuth authentication using JWT tokens for the user login system. The solution handles token refresh and validation properly."
+
+**Output**:
+```json
+["OAuth authentication", "JWT tokens", "user login", "token refresh", "authentication implementation"]
+```
+
+**Memory Content**: "Fixed the database connection timeout issue by increasing the connection pool size and optimizing slow queries with proper indexing."
+
+**Output**:
+```json
+["database connection", "timeout issue", "connection pool", "query optimization", "indexing"]
+```
+
+**Memory Content**: "Learned that Alpine.js x-data components should use camelCase for method names and snake_case for data properties to follow best practices."
+
+**Output**:
+```json
+["Alpine.js", "x-data components", "camelCase methods", "naming conventions"]
+```

python/api/import_knowledge.py

@@ -16,12 +16,22 @@ class ImportKnowledge(ApiHandler):
         context = self.get_context(ctxid)
 
         file_list = request.files.getlist("files[]")
-        KNOWLEDGE_FOLDER = files.get_abs_path(memory.get_custom_knowledge_subdir_abs(context.agent0),"main")
+        KNOWLEDGE_FOLDER = files.get_abs_path(memory.get_custom_knowledge_subdir_abs(context.agent0), "main")
+
+        # Ensure knowledge folder exists (create if missing)
+        try:
+            os.makedirs(KNOWLEDGE_FOLDER, exist_ok=True)
+        except (OSError, PermissionError) as e:
+            raise Exception(f"Failed to create knowledge folder {KNOWLEDGE_FOLDER}: {e}")
+
+        # Verify the directory is accessible
+        if not os.access(KNOWLEDGE_FOLDER, os.W_OK):
+            raise Exception(f"Knowledge folder {KNOWLEDGE_FOLDER} is not writable")
 
         saved_filenames = []
 
         for file in file_list:
-            if file:
+            if file and file.filename:
                 filename = secure_filename(file.filename)  # type: ignore
                 file.save(os.path.join(KNOWLEDGE_FOLDER, filename))
                 saved_filenames.append(filename)
@@ -33,4 +43,4 @@ class ImportKnowledge(ApiHandler):
         return {
             "message": "Knowledge Imported",
             "filenames": saved_filenames[:5]
-        }
+        }

python/extensions/message_loop_prompts_after/_50_recall_memories.py

@@ -2,6 +2,7 @@ import asyncio
 from python.helpers.extension import Extension
 from python.helpers.memory import Memory
 from agent import LoopData
+from python.tools.memory_load import DEFAULT_THRESHOLD as DEFAULT_MEMORY_THRESHOLD
 
 DATA_NAME_TASK = "_recall_memories_task"
 
@@ -10,8 +11,8 @@ class RecallMemories(Extension):
 
     INTERVAL = 3
     HISTORY = 10000
-    RESULTS = 3
-    THRESHOLD = 0.6
+    RESULTS = 5
+    THRESHOLD = DEFAULT_MEMORY_THRESHOLD
 
     async def execute(self, loop_data: LoopData = LoopData(), **kwargs):
 
@@ -88,8 +89,7 @@ class RecallMemories(Extension):
         memories_text = ""
         for memory in memories:
             memories_text += memory.page_content + "\n\n"
-        memories_text = memories_text.strip()
-
+            
         # log the full results
         log_item.update(memories=memories_text)
 

python/extensions/message_loop_prompts_after/_51_recall_solutions.py

@@ -2,16 +2,18 @@ import asyncio
 from python.helpers.extension import Extension
 from python.helpers.memory import Memory
 from agent import LoopData
+from python.tools.memory_load import DEFAULT_THRESHOLD as DEFAULT_MEMORY_THRESHOLD
 
 DATA_NAME_TASK = "_recall_solutions_task"
 
+
 class RecallSolutions(Extension):
 
     INTERVAL = 3
     HISTORY = 10000
-    SOLUTIONS_COUNT = 2
-    INSTRUMENTS_COUNT = 2
-    THRESHOLD = 0.6
+    SOLUTIONS_COUNT = 3
+    INSTRUMENTS_COUNT = 3
+    THRESHOLD = DEFAULT_MEMORY_THRESHOLD
 
     async def execute(self, loop_data: LoopData = LoopData(), **kwargs):
 
@@ -26,11 +28,11 @@ class RecallSolutions(Extension):
 
     async def search_solutions(self, loop_data: LoopData, **kwargs):
 
-        #cleanup
+        # cleanup
         extras = loop_data.extras_persistent
         if "solutions" in extras:
             del extras["solutions"]
-        
+
         # try:
 
         # show full util message

python/extensions/monologue_end/_50_memorize_fragments.py

@@ -4,12 +4,11 @@ from python.helpers.memory import Memory
 from python.helpers.dirty_json import DirtyJson
 from agent import LoopData
 from python.helpers.log import LogItem
+from python.tools.memory_load import DEFAULT_THRESHOLD as DEFAULT_MEMORY_THRESHOLD
 
 
 class MemorizeMemories(Extension):
 
-    REPLACE_THRESHOLD = 0.9
-
     async def execute(self, loop_data: LoopData = LoopData(), **kwargs):
         # try:
 
@@ -20,7 +19,8 @@ class MemorizeMemories(Extension):
         )
 
         # memorize in background
-        asyncio.create_task(self.memorize(loop_data, log_item))
+        task = asyncio.create_task(self.memorize(loop_data, log_item))
+        return task
 
     async def memorize(self, loop_data: LoopData, log_item: LogItem, **kwargs):
 
@@ -77,37 +77,75 @@ class MemorizeMemories(Extension):
         else:
             log_item.update(heading=f"{len(memories)} entries to memorize.")
 
-        # save chat history
-        db = await Memory.get(self.agent)
-
+        # Process memories with intelligent consolidation
         memories_txt = ""
-        rem = []
+        total_processed = 0
+        total_consolidated = 0
+
         for memory in memories:
-            # solution to plain text:
+            # Convert memory to plain text
             txt = f"{memory}"
             memories_txt += "\n\n" + txt
-            log_item.update(memories=memories_txt.strip())
-
-            # remove previous fragments too similiar to this one
-            if self.REPLACE_THRESHOLD > 0:
-                rem += await db.delete_documents_by_query(
-                    query=txt,
-                    threshold=self.REPLACE_THRESHOLD,
-                    filter=f"area=='{Memory.Area.FRAGMENTS.value}'",
+
+            try:
+                # Use intelligent consolidation system
+                from python.helpers.memory_consolidation import create_memory_consolidator
+                consolidator = create_memory_consolidator(
+                    self.agent,
+                    similarity_threshold=DEFAULT_MEMORY_THRESHOLD,  # More permissive for discovery
+                    max_similar_memories=8,
+                    max_llm_context_memories=4
                 )
-                if rem:
-                    rem_txt = "\n\n".join(Memory.format_docs_plain(rem))
-                    log_item.update(replaced=rem_txt)
 
-            # insert new solution
-            await db.insert_text(text=txt, metadata={"area": Memory.Area.FRAGMENTS.value})
+                # Create memory item-specific log for detailed tracking
+                memory_log = self.agent.context.log.log(
+                    type="util",
+                    heading=f"Processing memory fragment: {txt[:50]}...",
+                    temp=False,
+                    update_progress="none"  # Don't affect status bar
+                )
+
+                # Process with intelligent consolidation
+                result_obj = await consolidator.process_new_memory(
+                    new_memory=txt,
+                    area=Memory.Area.FRAGMENTS.value,
+                    metadata={"area": Memory.Area.FRAGMENTS.value},
+                    log_item=memory_log
+                )
 
+                # Update the individual log item with completion status but keep it temporary
+                if result_obj.get("success"):
+                    total_consolidated += 1
+                    memory_log.update(
+                        result="Fragment processed successfully",
+                        heading=f"Memory fragment completed: {txt[:50]}...",
+                        temp=False,  # Show completion message
+                        update_progress="none"  # Show briefly then disappear
+                    )
+                else:
+                    memory_log.update(
+                        result="Fragment processing failed",
+                        heading=f"Memory fragment failed: {txt[:50]}...",
+                        temp=False,  # Show completion message
+                        update_progress="none"  # Show briefly then disappear
+                    )
+                total_processed += 1
+
+            except Exception as e:
+                # Log error but continue processing
+                log_item.update(consolidation_error=str(e))
+                total_processed += 1
+
+        # Update final results with structured logging
+        memories_txt = memories_txt.strip()
         log_item.update(
-            result=f"{len(memories)} entries memorized.",
-            heading=f"{len(memories)} entries memorized.",
+            heading=f"Memorization completed: {total_processed} memories processed, {total_consolidated} intelligently consolidated",
+            memories=memories_txt,
+            result=f"{total_processed} memories processed, {total_consolidated} intelligently consolidated",
+            memories_processed=total_processed,
+            memories_consolidated=total_consolidated,
+            update_progress="none"
         )
-        if rem:
-            log_item.stream(result=f"\nReplaced {len(rem)} previous memories.")
 
     # except Exception as e:
     #     err = errors.format_error(e)

python/extensions/monologue_end/_51_memorize_solutions.py

@@ -4,12 +4,11 @@ from python.helpers.memory import Memory
 from python.helpers.dirty_json import DirtyJson
 from agent import LoopData
 from python.helpers.log import LogItem
+from python.tools.memory_load import DEFAULT_THRESHOLD as DEFAULT_MEMORY_THRESHOLD
 
 
 class MemorizeSolutions(Extension):
 
-    REPLACE_THRESHOLD = 0.9
-
     async def execute(self, loop_data: LoopData = LoopData(), **kwargs):
         # try:
  
@@ -20,7 +19,8 @@ class MemorizeSolutions(Extension):
         )
 
         # memorize in background
-        asyncio.create_task(self.memorize(loop_data, log_item))
+        task = asyncio.create_task(self.memorize(loop_data, log_item))
+        return task
 
     async def memorize(self, loop_data: LoopData, log_item: LogItem, **kwargs):
         # get system message and chat history for util llm
@@ -78,13 +78,13 @@ class MemorizeSolutions(Extension):
                 heading=f"{len(solutions)} successful solutions to memorize."
             )
 
-        # save chat history
-        db = await Memory.get(self.agent)
-
+        # Process solutions with intelligent consolidation
         solutions_txt = ""
-        rem = []
+        total_processed = 0
+        total_consolidated = 0
+
         for solution in solutions:
-            # solution to plain text:
+            # Convert solution to structured text
             if isinstance(solution, dict):
                 problem = solution.get('problem', 'Unknown problem')
                 solution_text = solution.get('solution', 'Unknown solution')
@@ -94,28 +94,65 @@ class MemorizeSolutions(Extension):
                 txt = f"# Solution\n {str(solution)}"
             solutions_txt += txt + "\n\n"
 
-            # remove previous solutions too similiar to this one
-            if self.REPLACE_THRESHOLD > 0:
-                rem += await db.delete_documents_by_query(
-                    query=txt,
-                    threshold=self.REPLACE_THRESHOLD,
-                    filter=f"area=='{Memory.Area.SOLUTIONS.value}'",
+            try:
+                # Use intelligent consolidation system
+                from python.helpers.memory_consolidation import create_memory_consolidator
+                consolidator = create_memory_consolidator(
+                    self.agent,
+                    similarity_threshold=DEFAULT_MEMORY_THRESHOLD,  # More permissive for discovery
+                    max_similar_memories=6,    # Fewer for solutions (more complex)
+                    max_llm_context_memories=3
                 )
-                if rem:
-                    rem_txt = "\n\n".join(Memory.format_docs_plain(rem))
-                    log_item.update(replaced=rem_txt)
 
-            # insert new solution
-            await db.insert_text(text=txt, metadata={"area": Memory.Area.SOLUTIONS.value})
+                # Create solution-specific log for detailed tracking
+                solution_log = self.agent.context.log.log(
+                    type="util",
+                    heading=f"Processing solution: {txt[:50]}...",
+                    temp=False,
+                    update_progress="none"  # Don't affect status bar
+                )
+
+                # Process with intelligent consolidation
+                result_obj = await consolidator.process_new_memory(
+                    new_memory=txt,
+                    area=Memory.Area.SOLUTIONS.value,
+                    metadata={"area": Memory.Area.SOLUTIONS.value},
+                    log_item=solution_log
+                )
 
+                # Update the individual log item with completion status but keep it temporary
+                if result_obj.get("success"):
+                    total_consolidated += 1
+                    solution_log.update(
+                        result="Solution processed successfully",
+                        heading=f"Solution completed: {txt[:50]}...",
+                        temp=False,  # Show completion message
+                        update_progress="none"  # Show briefly then disappear
+                    )
+                else:
+                    solution_log.update(
+                        result="Solution processing failed",
+                        heading=f"Solution failed: {txt[:50]}...",
+                        temp=False,  # Show completion message
+                        update_progress="none"  # Show briefly then disappear
+                    )
+                total_processed += 1
+
+            except Exception as e:
+                # Log error but continue processing
+                log_item.update(consolidation_error=str(e))
+                total_processed += 1
+
+        # Update final results with structured logging
         solutions_txt = solutions_txt.strip()
-        log_item.update(solutions=solutions_txt)
         log_item.update(
-            result=f"{len(solutions)} solutions memorized.",
-            heading=f"{len(solutions)} solutions memorized.",
+            heading=f"Solution memorization completed: {total_processed} solutions processed, {total_consolidated} intelligently consolidated",
+            solutions=solutions_txt,
+            result=f"{total_processed} solutions processed, {total_consolidated} intelligently consolidated",
+            solutions_processed=total_processed,
+            solutions_consolidated=total_consolidated,
+            update_progress="none"
         )
-        if rem:
-            log_item.stream(result=f"\nReplaced {len(rem)} previous solutions.")
 
     # except Exception as e:
     #     err = errors.format_error(e)

python/helpers/knowledge_import.py

@@ -1,17 +1,13 @@
 import glob
 import os
 import hashlib
-import json
 from typing import Any, Dict, Literal, TypedDict
 from langchain_community.document_loaders import (
     CSVLoader,
-    JSONLoader,
     PyPDFLoader,
     TextLoader,
     UnstructuredHTMLLoader,
-    UnstructuredMarkdownLoader,
 )
-from python.helpers import files
 from python.helpers.log import LogItem
 from python.helpers.print_style import PrintStyle
 
@@ -41,34 +37,72 @@ def load_knowledge(
     metadata: dict[str, Any] = {},
     filename_pattern: str = "**/*",
 ) -> Dict[str, KnowledgeImport]:
+    """
+    Load knowledge files from a directory with change detection and metadata enhancement.
 
-    # from python.helpers.memory import Memory
+    This function now includes enhanced error handling and compatibility with the
+    intelligent memory consolidation system.
+    """
 
     # Mapping file extensions to corresponding loader classes
+    # Note: Using TextLoader for JSON and MD to avoid parsing issues with consolidation
     file_types_loaders = {
         "txt": TextLoader,
         "pdf": PyPDFLoader,
         "csv": CSVLoader,
         "html": UnstructuredHTMLLoader,
-        # "json": JSONLoader,
-        "json": TextLoader,
-        # "md": UnstructuredMarkdownLoader,
-        "md": TextLoader,
+        "json": TextLoader,  # Use TextLoader for better consolidation compatibility
+        "md": TextLoader,    # Use TextLoader for better consolidation compatibility
     }
 
     cnt_files = 0
     cnt_docs = 0
 
-    # for area in Memory.Area:
-    #     subdir = files.get_abs_path(knowledge_dir, area.value)
-
-    # if not os.path.exists(knowledge_dir):
-    #     os.makedirs(knowledge_dir)
-    #     continue
+    # Validate and create knowledge directory if needed
+    if not knowledge_dir:
+        if log_item:
+            log_item.stream(progress="\nNo knowledge directory specified")
+        PrintStyle(font_color="yellow").print("No knowledge directory specified")
+        return index
+
+    if not os.path.exists(knowledge_dir):
+        try:
+            os.makedirs(knowledge_dir, exist_ok=True)
+            # Verify the directory was actually created and is accessible
+            if not os.path.exists(knowledge_dir) or not os.access(knowledge_dir, os.R_OK):
+                error_msg = f"Knowledge directory {knowledge_dir} was created but is not accessible"
+                if log_item:
+                    log_item.stream(progress=f"\n{error_msg}")
+                PrintStyle(font_color="red").print(error_msg)
+                return index
+
+            if log_item:
+                log_item.stream(progress=f"\nCreated knowledge directory: {knowledge_dir}")
+            PrintStyle(font_color="green").print(f"Created knowledge directory: {knowledge_dir}")
+        except (OSError, PermissionError) as e:
+            error_msg = f"Failed to create knowledge directory {knowledge_dir}: {e}"
+            if log_item:
+                log_item.stream(progress=f"\n{error_msg}")
+            PrintStyle(font_color="red").print(error_msg)
+            return index
+
+    # Final accessibility check for existing directories
+    if not os.access(knowledge_dir, os.R_OK):
+        error_msg = f"Knowledge directory {knowledge_dir} exists but is not readable"
+        if log_item:
+            log_item.stream(progress=f"\n{error_msg}")
+        PrintStyle(font_color="red").print(error_msg)
+        return index
 
     # Fetch all files in the directory with specified extensions
-    kn_files = glob.glob(knowledge_dir + "/" + filename_pattern, recursive=True)
-    kn_files = [f for f in kn_files if os.path.isfile(f)]
+    try:
+        kn_files = glob.glob(os.path.join(knowledge_dir, filename_pattern), recursive=True)
+        kn_files = [f for f in kn_files if os.path.isfile(f) and not os.path.basename(f).startswith('.')]
+    except Exception as e:
+        PrintStyle(font_color="red").print(f"Error scanning knowledge directory {knowledge_dir}: {e}")
+        if log_item:
+            log_item.stream(progress=f"\nError scanning directory: {e}")
+        return index
 
     if kn_files:
         PrintStyle.standard(
@@ -80,48 +114,96 @@ def load_knowledge(
             )
 
     for file_path in kn_files:
-        ext = file_path.split(".")[-1].lower()
-        if ext in file_types_loaders:
+        try:
+            # Get file extension safely
+            file_parts = os.path.basename(file_path).split('.')
+            if len(file_parts) < 2:
+                continue  # Skip files without extensions
+
+            ext = file_parts[-1].lower()
+            if ext not in file_types_loaders:
+                continue  # Skip unsupported file types
+
             checksum = calculate_checksum(file_path)
-            file_key = file_path  # os.path.relpath(file_path, knowledge_dir)
+            if not checksum:
+                continue  # Skip files with checksum errors
 
-            # Load existing data from the index or create a new entry
-            file_data = index.get(file_key, {})
+            file_key = file_path
 
+            # Load existing data from the index or create a new entry
+            file_data: KnowledgeImport = index.get(file_key, {
+                "file": file_key,
+                "checksum": "",
+                "ids": [],
+                "state": "changed",
+                "documents": []
+            })
+
+            # Check if file has changed
             if file_data.get("checksum") == checksum:
                 file_data["state"] = "original"
             else:
                 file_data["state"] = "changed"
 
+            # Process changed files
             if file_data["state"] == "changed":
                 file_data["checksum"] = checksum
                 loader_cls = file_types_loaders[ext]
-                loader = loader_cls(
-                    file_path,
-                    **(
-                        text_loader_kwargs
-                        if ext in ["txt", "csv", "html", "md"]
-                        else {}
-                    ),
-                )
-                file_data["documents"] = loader.load_and_split()
-                for doc in file_data["documents"]:
-                    doc.metadata = {**doc.metadata, **metadata}
-                cnt_files += 1
-                cnt_docs += len(file_data["documents"])
-                # PrintStyle.standard(f"Imported {len(file_data['documents'])} documents from {file_path}")
+
+                try:
+                    loader = loader_cls(
+                        file_path,
+                        **(
+                            text_loader_kwargs
+                            if ext in ["txt", "csv", "html", "md"]
+                            else {}
+                        ),
+                    )
+                    documents = loader.load_and_split()
+
+                    # Enhanced metadata for better consolidation compatibility
+                    enhanced_metadata = {
+                        **metadata,
+                        "source_file": os.path.basename(file_path),
+                        "source_path": file_path,
+                        "file_type": ext,
+                        "knowledge_source": True,  # Flag to distinguish from conversation memories
+                        "import_timestamp": None,  # Will be set when inserted into memory
+                    }
+
+                    # Apply metadata to all documents
+                    for doc in documents:
+                        doc.metadata = {**doc.metadata, **enhanced_metadata}
+
+                    file_data["documents"] = documents
+                    cnt_files += 1
+                    cnt_docs += len(documents)
+
+                except Exception as e:
+                    PrintStyle(font_color="red").print(f"Error loading {file_path}: {e}")
+                    if log_item:
+                        log_item.stream(progress=f"\nError loading {os.path.basename(file_path)}: {e}")
+                    continue
 
             # Update the index
-            index[file_key] = file_data  # type: ignore
+            index[file_key] = file_data
+
+        except Exception as e:
+            PrintStyle(font_color="red").print(f"Error processing {file_path}: {e}")
+            continue
 
-    # loop index where state is not set and mark it as removed
-    for file_key, file_data in index.items():
-        if not file_data.get("state", ""):
+    # Mark removed files
+    current_files = set(kn_files)
+    for file_key, file_data in list(index.items()):
+        if file_key not in current_files and not file_data.get("state"):
             index[file_key]["state"] = "removed"
 
-    PrintStyle.standard(f"Processed {cnt_docs} documents from {cnt_files} files.")
-    if log_item:
-        log_item.stream(
-            progress=f"\nProcessed {cnt_docs} documents from {cnt_files} files."
-        )
+    # Log results
+    if cnt_files > 0 or cnt_docs > 0:
+        PrintStyle.standard(f"Processed {cnt_docs} documents from {cnt_files} files.")
+        if log_item:
+            log_item.stream(
+                progress=f"\nProcessed {cnt_docs} documents from {cnt_files} files."
+            )
+
     return index

python/helpers/memory_consolidation.py

@@ -0,0 +1,791 @@
+import asyncio
+import json
+from dataclasses import dataclass, field
+from datetime import datetime, timezone
+from typing import Any, Dict, List, Optional
+from enum import Enum
+
+from langchain_core.documents import Document
+
+from python.helpers.memory import Memory
+from python.helpers.dirty_json import DirtyJson
+from python.helpers.log import LogItem
+from python.helpers.print_style import PrintStyle
+from python.tools.memory_load import DEFAULT_THRESHOLD as DEFAULT_MEMORY_THRESHOLD
+from agent import Agent
+
+
+class ConsolidationAction(Enum):
+    """Actions that can be taken during memory consolidation."""
+    MERGE = "merge"
+    REPLACE = "replace"
+    KEEP_SEPARATE = "keep_separate"
+    UPDATE = "update"
+    SKIP = "skip"
+
+
+@dataclass
+class ConsolidationConfig:
+    """Configuration for memory consolidation behavior."""
+    similarity_threshold: float = DEFAULT_MEMORY_THRESHOLD
+    max_similar_memories: int = 10
+    consolidation_sys_prompt: str = "memory.consolidation.sys.md"
+    consolidation_msg_prompt: str = "memory.consolidation.msg.md"
+    max_llm_context_memories: int = 5
+    keyword_extraction_sys_prompt: str = "memory.keyword_extraction.sys.md"
+    keyword_extraction_msg_prompt: str = "memory.keyword_extraction.msg.md"
+    processing_timeout_seconds: int = 60
+    # Add safety threshold for REPLACE actions
+    replace_similarity_threshold: float = 0.9  # Higher threshold for replacement safety
+
+
+@dataclass
+class ConsolidationResult:
+    """Result of memory consolidation analysis."""
+    action: ConsolidationAction
+    memories_to_remove: List[str] = field(default_factory=list)
+    memories_to_update: List[Dict[str, Any]] = field(default_factory=list)
+    new_memory_content: str = ""
+    metadata: Dict[str, Any] = field(default_factory=dict)
+    reasoning: str = ""
+
+
+@dataclass
+class MemoryAnalysisContext:
+    """Context for LLM memory analysis."""
+    new_memory: str
+    similar_memories: List[Document]
+    area: str
+    timestamp: str
+    existing_metadata: Dict[str, Any]
+
+
+class MemoryConsolidator:
+    """
+    Intelligent memory consolidation system that uses LLM analysis to determine
+    optimal memory organization and automatically consolidates related memories.
+    """
+
+    def __init__(self, agent: Agent, config: Optional[ConsolidationConfig] = None):
+        self.agent = agent
+        self.config = config or ConsolidationConfig()
+
+    async def process_new_memory(
+        self,
+        new_memory: str,
+        area: str,
+        metadata: Dict[str, Any],
+        log_item: Optional[LogItem] = None
+    ) -> dict:
+        """
+        Process a new memory through the intelligent consolidation pipeline.
+
+        Args:
+            new_memory: The new memory content to process
+            area: Memory area (MAIN, FRAGMENTS, SOLUTIONS, INSTRUMENTS)
+            metadata: Initial metadata for the memory
+            log_item: Optional log item for progress tracking
+
+        Returns:
+            dict: {"success": bool, "memory_ids": [str, ...]}
+        """
+        try:
+            # Start processing with timeout
+            processing_task = asyncio.create_task(
+                self._process_memory_with_consolidation(new_memory, area, metadata, log_item)
+            )
+
+            result = await asyncio.wait_for(
+                processing_task,
+                timeout=self.config.processing_timeout_seconds
+            )
+            return result
+
+        except asyncio.TimeoutError:
+            PrintStyle().error(f"Memory consolidation timeout for area {area}")
+            return {"success": False, "memory_ids": []}
+
+        except Exception as e:
+            PrintStyle().error(f"Memory consolidation error for area {area}: {str(e)}")
+            return {"success": False, "memory_ids": []}
+
+    async def _process_memory_with_consolidation(
+        self,
+        new_memory: str,
+        area: str,
+        metadata: Dict[str, Any],
+        log_item: Optional[LogItem] = None
+    ) -> dict:
+        """Execute the full consolidation pipeline."""
+
+        if log_item:
+            log_item.update(progress="Starting intelligent memory consolidation...")
+
+        # Step 1: Discover similar memories
+        similar_memories = await self._find_similar_memories(new_memory, area, log_item)
+
+        # this block always returns
+        if not similar_memories:
+            # No similar memories found, insert directly
+            if log_item:
+                log_item.update(
+                    progress="No similar memories found, inserting new memory",
+                    temp=True
+                )
+            try:
+                db = await Memory.get(self.agent)
+                if 'timestamp' not in metadata:
+                    metadata['timestamp'] = self._get_timestamp()
+                memory_id = await db.insert_text(new_memory, metadata)
+                if log_item:
+                    log_item.update(
+                        result="Memory inserted successfully",
+                        memory_ids=[memory_id],
+                        consolidation_action="direct_insert"
+                    )
+                return {"success": True, "memory_ids": [memory_id]}
+            except Exception as e:
+                PrintStyle().error(f"Direct memory insertion failed: {str(e)}")
+                if log_item:
+                    log_item.update(result=f"Memory insertion failed: {str(e)}")
+                return {"success": False, "memory_ids": []}
+
+        if log_item:
+            log_item.update(
+                progress=f"Found {len(similar_memories)} similar memories, analyzing...",
+                temp=True,
+                similar_memories_count=len(similar_memories)
+            )
+
+        # Step 2: Validate that similar memories still exist (they might have been deleted by previous consolidations)
+        if similar_memories:
+            memory_ids_to_check = [doc.metadata.get('id') for doc in similar_memories if doc.metadata.get('id')]
+            # Filter out None values and ensure all IDs are strings
+            memory_ids_to_check = [str(id) for id in memory_ids_to_check if id is not None]
+            db = await Memory.get(self.agent)
+            still_existing = db.db.get_by_ids(memory_ids_to_check)
+            existing_ids = {doc.metadata.get('id') for doc in still_existing}
+
+            # Filter out deleted memories
+            valid_similar_memories = [doc for doc in similar_memories if doc.metadata.get('id') in existing_ids]
+
+            if len(valid_similar_memories) != len(similar_memories):
+                deleted_count = len(similar_memories) - len(valid_similar_memories)
+                if log_item:
+                    log_item.update(
+                        progress=f"Filtered out {deleted_count} deleted memories, {len(valid_similar_memories)} remain for analysis",
+                        temp=True,
+                        race_condition_detected=True,
+                        deleted_similar_memories_count=deleted_count
+                    )
+                similar_memories = valid_similar_memories
+
+        # If no valid similar memories remain after filtering, insert directly
+        if not similar_memories:
+            if log_item:
+                log_item.update(
+                    progress="No valid similar memories remain, inserting new memory",
+                    temp=True
+                )
+            try:
+                db = await Memory.get(self.agent)
+                if 'timestamp' not in metadata:
+                    metadata['timestamp'] = self._get_timestamp()
+                memory_id = await db.insert_text(new_memory, metadata)
+                if log_item:
+                    log_item.update(
+                        result="Memory inserted successfully (no valid similar memories)",
+                        memory_ids=[memory_id],
+                        consolidation_action="direct_insert_filtered"
+                    )
+                return {"success": True, "memory_ids": [memory_id]}
+            except Exception as e:
+                PrintStyle().error(f"Direct memory insertion failed: {str(e)}")
+                if log_item:
+                    log_item.update(result=f"Memory insertion failed: {str(e)}")
+                return {"success": False, "memory_ids": []}
+
+        # Step 3: Analyze with LLM (now with validated memories)
+        analysis_context = MemoryAnalysisContext(
+            new_memory=new_memory,
+            similar_memories=similar_memories,
+            area=area,
+            timestamp=self._get_timestamp(),
+            existing_metadata=metadata
+        )
+
+        consolidation_result = await self._analyze_memory_consolidation(analysis_context, log_item)
+
+        if consolidation_result.action == ConsolidationAction.SKIP:
+            if log_item:
+                log_item.update(
+                    progress="LLM analysis suggests skipping consolidation",
+                    temp=True
+                )
+            try:
+                db = await Memory.get(self.agent)
+                if 'timestamp' not in metadata:
+                    metadata['timestamp'] = self._get_timestamp()
+                memory_id = await db.insert_text(new_memory, metadata)
+                if log_item:
+                    log_item.update(
+                        result="Memory inserted (consolidation skipped)",
+                        memory_ids=[memory_id],
+                        consolidation_action="skip",
+                        reasoning=consolidation_result.reasoning or "LLM analysis suggested skipping"
+                    )
+                return {"success": True, "memory_ids": [memory_id]}
+            except Exception as e:
+                PrintStyle().error(f"Skip consolidation insertion failed: {str(e)}")
+                if log_item:
+                    log_item.update(result=f"Memory insertion failed: {str(e)}")
+                return {"success": False, "memory_ids": []}
+
+        # Step 4: Apply consolidation decisions
+        memory_ids = await self._apply_consolidation_result(
+            consolidation_result,
+            area,
+            analysis_context.existing_metadata,  # Pass original metadata
+            log_item
+        )
+
+        if log_item:
+            if memory_ids:
+                log_item.update(
+                    result=f"Consolidation completed: {consolidation_result.action.value}",
+                    memory_ids=memory_ids,
+                    consolidation_action=consolidation_result.action.value,
+                    reasoning=consolidation_result.reasoning or "No specific reasoning provided",
+                    memories_processed=len(similar_memories) + 1  # +1 for new memory
+                )
+            else:
+                log_item.update(
+                    result=f"Consolidation failed: {consolidation_result.action.value}",
+                    consolidation_action=consolidation_result.action.value,
+                    reasoning=consolidation_result.reasoning or "Consolidation operation failed"
+                )
+
+        return {"success": bool(memory_ids), "memory_ids": memory_ids or []}
+
+    async def _gather_consolidated_metadata(
+        self,
+        db,
+        result: ConsolidationResult,
+        original_metadata: Dict[str, Any]
+    ) -> Dict[str, Any]:
+        """
+        Gather and merge metadata from memories being consolidated to preserve important fields.
+        This ensures critical metadata like priority, source, etc. is preserved during consolidation.
+        """
+        try:
+            # Start with the new memory's metadata as base
+            consolidated_metadata = dict(original_metadata)
+
+            # Collect all memory IDs that will be involved in consolidation
+            memory_ids = []
+
+            # Add memories to be removed (MERGE, REPLACE actions)
+            if result.memories_to_remove:
+                memory_ids.extend(result.memories_to_remove)
+
+            # Add memories to be updated (UPDATE action)
+            if result.memories_to_update:
+                for update_info in result.memories_to_update:
+                    memory_id = update_info.get('id')
+                    if memory_id:
+                        memory_ids.append(memory_id)
+
+            # Retrieve original memories to extract their metadata
+            if memory_ids:
+                original_memories = await db.aget_by_ids(memory_ids)
+
+                # Merge ALL metadata fields from original memories
+                for memory in original_memories:
+                    memory_metadata = memory.metadata
+
+                    # Process ALL metadata fields from the original memory
+                    for field_name, field_value in memory_metadata.items():
+                        if field_name not in consolidated_metadata:
+                            # Field doesn't exist in consolidated metadata, add it
+                            consolidated_metadata[field_name] = field_value
+                        elif field_name in consolidated_metadata:
+                            # Field exists in both - handle special merge cases
+                            if field_name == 'tags' and isinstance(field_value, list) and isinstance(consolidated_metadata[field_name], list):
+                                # Merge tags lists and remove duplicates
+                                merged_tags = list(set(consolidated_metadata[field_name] + field_value))
+                                consolidated_metadata[field_name] = merged_tags
+                            # For all other fields, keep the new memory's value (don't overwrite)
+                            # This preserves the new memory's metadata when there are conflicts
+
+            return consolidated_metadata
+
+        except Exception as e:
+            # If metadata gathering fails, return original metadata as fallback
+            PrintStyle(font_color="yellow").print(f"Failed to gather consolidated metadata: {str(e)}")
+            return original_metadata
+
+    async def _find_similar_memories(
+        self,
+        new_memory: str,
+        area: str,
+        log_item: Optional[LogItem] = None
+    ) -> List[Document]:
+        """
+        Find similar memories using both semantic similarity and keyword matching.
+        Now includes knowledge source awareness and similarity scores for validation.
+        """
+        db = await Memory.get(self.agent)
+
+        # Step 1: Extract keywords/queries for enhanced search
+        search_queries = await self._extract_search_keywords(new_memory, log_item)
+
+        all_similar = []
+
+        # Step 2: Semantic similarity search with scores
+        semantic_similar = await db.search_similarity_threshold(
+            query=new_memory,
+            limit=self.config.max_similar_memories,
+            threshold=self.config.similarity_threshold,
+            filter=f"area == '{area}'"
+        )
+        all_similar.extend(semantic_similar)
+
+        # Step 3: Keyword-based searches
+        for query in search_queries:
+            if query.strip():
+                # Fix division by zero: ensure len(search_queries) > 0
+                queries_count = max(1, len(search_queries))  # Prevent division by zero
+                keyword_similar = await db.search_similarity_threshold(
+                    query=query.strip(),
+                    limit=max(3, self.config.max_similar_memories // queries_count),
+                    threshold=self.config.similarity_threshold,
+                    filter=f"area == '{area}'"
+                )
+                all_similar.extend(keyword_similar)
+
+        # Step 4: Deduplicate by document ID and store similarity info
+        seen_ids = set()
+        unique_similar = []
+        for doc in all_similar:
+            doc_id = doc.metadata.get('id')
+            if doc_id and doc_id not in seen_ids:
+                seen_ids.add(doc_id)
+                unique_similar.append(doc)
+
+        # Step 5: Calculate similarity scores for replacement validation
+        # Since FAISS doesn't directly expose similarity scores, use ranking-based estimation
+        # CRITICAL: All documents must have similarity >= search_threshold since FAISS returned them
+        # FIXED: Use conservative scoring that keeps all scores in safe consolidation range
+        similarity_scores = {}
+        total_docs = len(unique_similar)
+        search_threshold = self.config.similarity_threshold
+        safety_threshold = self.config.replace_similarity_threshold
+
+        for i, doc in enumerate(unique_similar):
+            doc_id = doc.metadata.get('id')
+            if doc_id:
+                # Convert ranking to similarity score with conservative distribution
+                if total_docs == 1:
+                    ranking_similarity = 1.0  # Single document gets perfect score
+                else:
+                    # Use conservative scoring: distribute between safety_threshold and 1.0
+                    # This ensures all scores are suitable for consolidation
+                    # First document gets 1.0, last gets safety_threshold (0.9 by default)
+                    ranking_factor = 1.0 - (i / (total_docs - 1))
+                    score_range = 1.0 - safety_threshold  # e.g., 1.0 - 0.9 = 0.1
+                    ranking_similarity = safety_threshold + (score_range * ranking_factor)
+
+                    # Ensure minimum score is search_threshold for logical consistency
+                    ranking_similarity = max(ranking_similarity, search_threshold)
+
+                similarity_scores[doc_id] = ranking_similarity
+
+        # Step 6: Add similarity score to document metadata for LLM analysis
+        for doc in unique_similar:
+            doc_id = doc.metadata.get('id')
+            estimated_similarity = similarity_scores.get(doc_id, 0.7)
+            # Store for later validation
+            doc.metadata['_consolidation_similarity'] = estimated_similarity
+
+        # Step 7: Limit to max context for LLM
+        limited_similar = unique_similar[:self.config.max_llm_context_memories]
+
+        return limited_similar
+
+    async def _extract_search_keywords(
+        self,
+        new_memory: str,
+        log_item: Optional[LogItem] = None
+    ) -> List[str]:
+        """Extract search keywords/queries from new memory using utility LLM."""
+
+        try:
+            system_prompt = self.agent.read_prompt(
+                self.config.keyword_extraction_sys_prompt,
+            )
+
+            message_prompt = self.agent.read_prompt(
+                self.config.keyword_extraction_msg_prompt,
+                memory_content=new_memory
+            )
+
+            # Call utility LLM to extract search queries
+            keywords_response = await self.agent.call_utility_model(
+                system=system_prompt,
+                message=message_prompt,
+                background=True
+            )
+
+            # Parse the response - expect JSON array of strings
+            keywords_json = DirtyJson.parse_string(keywords_response.strip())
+
+            if isinstance(keywords_json, list):
+                return [str(k) for k in keywords_json if k]
+            elif isinstance(keywords_json, str):
+                return [keywords_json]
+            else:
+                return []
+
+        except Exception as e:
+            PrintStyle().warning(f"Keyword extraction failed: {str(e)}")
+            # Fallback: use intelligent truncation for search
+            # Take first 200 chars if short, or first sentence if longer, but cap at 200 chars
+            if len(new_memory) <= 200:
+                fallback_content = new_memory
+            else:
+                first_sentence = new_memory.split('.')[0]
+                fallback_content = first_sentence[:200] if len(first_sentence) <= 200 else new_memory[:200]
+            return [fallback_content.strip()]
+
+    async def _analyze_memory_consolidation(
+        self,
+        context: MemoryAnalysisContext,
+        log_item: Optional[LogItem] = None
+    ) -> ConsolidationResult:
+        """Use LLM to analyze memory consolidation options."""
+
+        try:
+            # Prepare similar memories text
+            similar_memories_text = ""
+            for i, doc in enumerate(context.similar_memories):
+                timestamp = doc.metadata.get('timestamp', 'unknown')
+                doc_id = doc.metadata.get('id', f'doc_{i}')
+                similar_memories_text += f"ID: {doc_id}\nTimestamp: {timestamp}\nContent: {doc.page_content}\n\n"
+
+            # Build system prompt
+            system_prompt = self.agent.read_prompt(
+                self.config.consolidation_sys_prompt,
+            )
+
+            # Build message prompt
+            message_prompt = self.agent.read_prompt(
+                self.config.consolidation_msg_prompt,
+                new_memory=context.new_memory,
+                similar_memories=similar_memories_text.strip(),
+                area=context.area,
+                current_timestamp=context.timestamp,
+                new_memory_metadata=json.dumps(context.existing_metadata, indent=2)
+            )
+
+            analysis_response = await self.agent.call_utility_model(
+                system=system_prompt,
+                message=message_prompt,
+                callback=None,
+                background=True
+            )
+
+            # Parse LLM response
+            result_json = DirtyJson.parse_string(analysis_response.strip())
+
+            if not isinstance(result_json, dict):
+                raise ValueError("LLM response is not a valid JSON object")
+
+            # Parse consolidation result
+            action_str = result_json.get('action', 'skip')
+            try:
+                action = ConsolidationAction(action_str.lower())
+            except ValueError:
+                action = ConsolidationAction.SKIP
+
+            # Determine appropriate fallback for new_memory_content based on action
+            if action in [ConsolidationAction.MERGE, ConsolidationAction.REPLACE]:
+                # For MERGE/REPLACE, if no content provided, it's an error - don't use original
+                default_content = ""
+            else:
+                # For KEEP_SEPARATE/UPDATE/SKIP, original memory is appropriate fallback
+                default_content = context.new_memory
+
+            return ConsolidationResult(
+                action=action,
+                memories_to_remove=result_json.get('memories_to_remove', []),
+                memories_to_update=result_json.get('memories_to_update', []),
+                new_memory_content=result_json.get('new_memory_content', default_content),
+                metadata=result_json.get('metadata', {}),
+                reasoning=result_json.get('reasoning', '')
+            )
+
+        except Exception as e:
+            PrintStyle().warning(f"LLM consolidation analysis failed: {str(e)}")
+            # Fallback: skip consolidation
+            return ConsolidationResult(
+                action=ConsolidationAction.SKIP,
+                reasoning=f"Analysis failed: {str(e)}"
+            )
+
+    async def _apply_consolidation_result(
+        self,
+        result: ConsolidationResult,
+        area: str,
+        original_metadata: Dict[str, Any],  # Add original metadata parameter
+        log_item: Optional[LogItem] = None
+    ) -> list:
+        """Apply the consolidation decisions to the memory database."""
+
+        try:
+            db = await Memory.get(self.agent)
+
+            # Retrieve metadata from memories being consolidated to preserve important fields
+            consolidated_metadata = await self._gather_consolidated_metadata(db, result, original_metadata)
+
+            # Handle each action type specifically
+            if result.action == ConsolidationAction.KEEP_SEPARATE:
+                return await self._handle_keep_separate(db, result, area, consolidated_metadata, log_item)
+
+            elif result.action == ConsolidationAction.MERGE:
+                return await self._handle_merge(db, result, area, consolidated_metadata, log_item)
+
+            elif result.action == ConsolidationAction.REPLACE:
+                return await self._handle_replace(db, result, area, consolidated_metadata, log_item)
+
+            elif result.action == ConsolidationAction.UPDATE:
+                return await self._handle_update(db, result, area, consolidated_metadata, log_item)
+
+            else:
+                # Should not reach here, but handle gracefully
+                PrintStyle().warning(f"Unknown consolidation action: {result.action}")
+                return []
+
+        except Exception as e:
+            PrintStyle().error(f"Failed to apply consolidation result: {str(e)}")
+            return []
+
+    async def _handle_keep_separate(
+        self,
+        db,
+        result: ConsolidationResult,
+        area: str,
+        original_metadata: Dict[str, Any],  # Add original metadata parameter
+        log_item: Optional[LogItem] = None
+    ) -> list:
+        """Handle KEEP_SEPARATE action: Insert new memory without touching existing ones."""
+
+        if not result.new_memory_content:
+            return []
+
+        # Prepare metadata for new memory
+        # LLM metadata takes precedence over original metadata when there are conflicts
+        final_metadata = {
+            'area': area,
+            'timestamp': self._get_timestamp(),
+            'consolidation_action': result.action.value,
+            **original_metadata,  # Original metadata first
+            **result.metadata     # LLM metadata second (wins conflicts)
+        }
+
+        if result.reasoning:
+            final_metadata['consolidation_reasoning'] = result.reasoning
+
+        new_id = await db.insert_text(result.new_memory_content, final_metadata)
+        return [new_id]
+
+    async def _handle_merge(
+        self,
+        db,
+        result: ConsolidationResult,
+        area: str,
+        original_metadata: Dict[str, Any],  # Add original metadata parameter
+        log_item: Optional[LogItem] = None
+    ) -> list:
+        """Handle MERGE action: Combine memories, remove originals, insert consolidated version."""
+
+        # Step 1: Remove original memories being merged
+        if result.memories_to_remove:
+            await db.delete_documents_by_ids(result.memories_to_remove)
+
+        # Step 2: Insert consolidated memory
+        if result.new_memory_content:
+            # LLM metadata takes precedence over original metadata when there are conflicts
+            final_metadata = {
+                'area': area,
+                'timestamp': self._get_timestamp(),
+                'consolidation_action': result.action.value,
+                'consolidated_from': result.memories_to_remove,
+                **original_metadata,  # Original metadata first
+                **result.metadata     # LLM metadata second (wins conflicts)
+            }
+
+            if result.reasoning:
+                final_metadata['consolidation_reasoning'] = result.reasoning
+
+            new_id = await db.insert_text(result.new_memory_content, final_metadata)
+            return [new_id]
+        else:
+            return []
+
+    async def _handle_replace(
+        self,
+        db,
+        result: ConsolidationResult,
+        area: str,
+        original_metadata: Dict[str, Any],  # Add original metadata parameter
+        log_item: Optional[LogItem] = None
+    ) -> list:
+        """Handle REPLACE action: Remove old memories, insert new version with similarity validation."""
+
+        # Step 1: Validate similarity scores for replacement safety
+        if result.memories_to_remove:
+            # Get the memories to be removed and check their similarity scores
+            memories_to_check = await db.aget_by_ids(result.memories_to_remove)
+
+            unsafe_replacements = []
+            for memory in memories_to_check:
+                similarity = memory.metadata.get('_consolidation_similarity', 0.7)
+                if similarity < self.config.replace_similarity_threshold:
+                    unsafe_replacements.append({
+                        'id': memory.metadata.get('id'),
+                        'similarity': similarity,
+                        'content_preview': memory.page_content[:100]
+                    })
+
+            # If we have unsafe replacements, either block them or require explicit confirmation
+            if unsafe_replacements:
+                PrintStyle().warning(
+                    f"REPLACE blocked: {len(unsafe_replacements)} memories below "
+                    f"similarity threshold {self.config.replace_similarity_threshold}, converting to KEEP_SEPARATE"
+                )
+
+                # Instead of replace, just insert the new memory (keep separate)
+                if result.new_memory_content:
+                    final_metadata = {
+                        'area': area,
+                        'timestamp': self._get_timestamp(),
+                        'consolidation_action': 'keep_separate_safety',  # Indicate safety conversion
+                        'original_action': 'replace',
+                        'safety_reason': f'Similarity below threshold {self.config.replace_similarity_threshold}',
+                        **original_metadata,
+                        **result.metadata
+                    }
+
+                    if result.reasoning:
+                        final_metadata['consolidation_reasoning'] = result.reasoning
+
+                    new_id = await db.insert_text(result.new_memory_content, final_metadata)
+                    return [new_id]
+                else:
+                    return []
+
+        # Step 2: Proceed with normal replacement if similarity checks pass
+        if result.memories_to_remove:
+            await db.delete_documents_by_ids(result.memories_to_remove)
+
+        # Step 3: Insert replacement memory
+        if result.new_memory_content:
+            # LLM metadata takes precedence over original metadata when there are conflicts
+            final_metadata = {
+                'area': area,
+                'timestamp': self._get_timestamp(),
+                'consolidation_action': result.action.value,
+                'replaced_memories': result.memories_to_remove,
+                **original_metadata,  # Original metadata first
+                **result.metadata     # LLM metadata second (wins conflicts)
+            }
+
+            if result.reasoning:
+                final_metadata['consolidation_reasoning'] = result.reasoning
+
+            new_id = await db.insert_text(result.new_memory_content, final_metadata)
+            return [new_id]
+        else:
+            return []
+
+    async def _handle_update(
+        self,
+        db,
+        result: ConsolidationResult,
+        area: str,
+        original_metadata: Dict[str, Any],  # Add original metadata parameter
+        log_item: Optional[LogItem] = None
+    ) -> list:
+        """Handle UPDATE action: Modify existing memories in place with additional information."""
+
+        updated_count = 0
+        updated_ids = []
+
+        # Step 1: Update existing memories
+        for update_info in result.memories_to_update:
+            memory_id = update_info.get('id')
+            new_content = update_info.get('new_content', '')
+
+            if memory_id and new_content:
+                # Validate that the memory exists before attempting to delete it
+                existing_docs = await db.aget_by_ids([memory_id])
+                if not existing_docs:
+                    PrintStyle().warning(f"Memory ID {memory_id} not found during update, skipping")
+                    continue
+
+                # Delete old version and insert updated version
+                await db.delete_documents_by_ids([memory_id])
+
+                # LLM metadata takes precedence over original metadata when there are conflicts
+                updated_metadata = {
+                    'area': area,
+                    'timestamp': self._get_timestamp(),
+                    'consolidation_action': result.action.value,
+                    'updated_from': memory_id,
+                    **original_metadata,                    # Original metadata first
+                    **update_info.get('metadata', {})       # LLM metadata second (wins conflicts)
+                }
+
+                new_id = await db.insert_text(new_content, updated_metadata)
+                updated_count += 1
+                updated_ids.append(new_id)
+
+        # Step 2: Insert additional new memory if provided
+        new_memory_id = None
+        if result.new_memory_content:
+            # LLM metadata takes precedence over original metadata when there are conflicts
+            final_metadata = {
+                'area': area,
+                'timestamp': self._get_timestamp(),
+                'consolidation_action': result.action.value,
+                **original_metadata,  # Original metadata first
+                **result.metadata     # LLM metadata second (wins conflicts)
+            }
+
+            if result.reasoning:
+                final_metadata['consolidation_reasoning'] = result.reasoning
+
+            new_memory_id = await db.insert_text(result.new_memory_content, final_metadata)
+            updated_ids.append(new_memory_id)
+
+        return updated_ids
+
+    def _get_timestamp(self) -> str:
+        """Get current timestamp in standard format."""
+        return datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S")
+
+
+# Factory function for easy instantiation
+def create_memory_consolidator(agent: Agent, **config_overrides) -> MemoryConsolidator:
+    """
+    Create a MemoryConsolidator with optional configuration overrides.
+
+    Available configuration options:
+    - similarity_threshold: Discovery threshold for finding related memories (default 0.7)
+    - replace_similarity_threshold: Safety threshold for REPLACE actions (default 0.9)
+    - max_similar_memories: Maximum memories to discover (default 10)
+    - max_llm_context_memories: Maximum memories to send to LLM (default 5)
+    - processing_timeout_seconds: Timeout for consolidation processing (default 30)
+    """
+    config = ConsolidationConfig(**config_overrides)
+    return MemoryConsolidator(agent, config)

python/helpers/whisper.py

@@ -68,9 +68,16 @@ async def _transcribe(model_name:str, audio_bytes_b64: str):
     audio_bytes = base64.b64decode(audio_bytes_b64)
 
     # Create temp audio file
+    import os
     with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as audio_file:
         audio_file.write(audio_bytes)
-
-    # Transcribe the audio file
-    result = _model.transcribe(audio_file.name, fp16=False) # type: ignore
-    return result
+        temp_path = audio_file.name
+    try:
+        # Transcribe the audio file
+        result = _model.transcribe(temp_path, fp16=False) # type: ignore
+        return result
+    finally:
+        try:
+            os.remove(temp_path)
+        except Exception:
+            pass # ignore errors during cleanup

python/tools/knowledge_tool._py

@@ -1,4 +1,3 @@
-import os
 import asyncio
 from python.helpers import dotenv, memory, perplexity_search, duckduckgo_search
 from python.helpers.tool import Tool, Response
@@ -13,12 +12,17 @@ SEARCH_ENGINE_RESULTS = 10
 
 class Knowledge(Tool):
     async def execute(self, question="", **kwargs):
-        # Create tasks for all three search methods
+        if not question:
+            question = kwargs.get("query", "")
+            if not question:
+                return Response(message="No question provided", break_loop=False)
+
+        # Create tasks for all search methods
         tasks = [
             self.searxng_search(question),
             # self.perplexity_search(question),
             # self.duckduckgo_search(question),
-            self.mem_search(question),
+            self.mem_search_enhanced(question),
         ]
 
         # Run all tasks concurrently
@@ -31,8 +35,6 @@ class Knowledge(Tool):
         searxng_result = await self.searxng_document_qa(searxng_result, question)
 
         # Handle exceptions and format results
-        # perplexity_result = self.format_result(perplexity_result, "Perplexity")
-        # duckduckgo_result = self.format_result(duckduckgo_result, "DuckDuckGo")
         searxng_result = self.format_result_searxng(searxng_result, "Search Engine")
         memory_result = self.format_result(memory_result, "Memory")
 
@@ -102,6 +104,134 @@ class Knowledge(Tool):
         text = memory.Memory.format_docs_plain(docs)
         return "\n\n".join(text)
 
+    async def mem_search_enhanced(self, question: str):
+        """
+        Enhanced memory search with knowledge source awareness.
+        Separates and prioritizes knowledge sources vs conversation memories.
+        """
+        try:
+            db = await memory.Memory.get(self.agent)
+
+            # Search for knowledge sources (knowledge_source=True)
+            knowledge_docs = await db.search_similarity_threshold(
+                query=question, limit=5, threshold=DEFAULT_MEMORY_THRESHOLD,
+                filter="knowledge_source == True"
+            )
+
+            # Search for conversation memories (field doesn't exist or is not True)
+            conversation_docs = await db.search_similarity_threshold(
+                query=question, limit=5, threshold=DEFAULT_MEMORY_THRESHOLD,
+                filter="not knowledge_source if 'knowledge_source' in locals() else True"
+            )
+
+            # Combine and fallback to lower threshold if needed
+            all_docs = knowledge_docs + conversation_docs
+            threshold_note = ""
+
+            # If no results with default threshold, try with lower threshold
+            if not all_docs:
+                lower_threshold = DEFAULT_MEMORY_THRESHOLD * 0.8
+                knowledge_docs = await db.search_similarity_threshold(
+                    query=question, limit=5, threshold=lower_threshold,
+                    filter="knowledge_source == True"
+                )
+                conversation_docs = await db.search_similarity_threshold(
+                    query=question, limit=5, threshold=lower_threshold,
+                    filter="not knowledge_source if 'knowledge_source' in locals() else True"
+                )
+                all_docs = knowledge_docs + conversation_docs
+                if all_docs:
+                    threshold_note = f" (threshold: {lower_threshold})"
+
+            if not all_docs:
+                return await self._get_memory_diagnostics(db, question)
+
+            # Separate knowledge sources from conversation memories
+            knowledge_sources = knowledge_docs
+            conversation_memories = conversation_docs
+            result_parts = []
+
+            # Add search summary
+            result_parts.append(f"## 🔍 Search Results for: '{question}'")
+            result_parts.append(f"**Found:** {len(knowledge_sources)} knowledge sources, {len(conversation_memories)} conversation memories{threshold_note}")
+
+            # Show knowledge sources
+            if knowledge_sources:
+                result_parts.append("")
+                result_parts.append("## 📚 Knowledge Sources:")
+                for index, doc in enumerate(knowledge_sources):
+                    source_file = doc.metadata.get('source_file', 'Unknown source')
+                    file_type = doc.metadata.get('file_type', '').upper()
+                    area = doc.metadata.get('area', 'main').upper()
+
+                    result_parts.append(f"**Source:** {source_file} ({file_type}) [{area}]")
+                    result_parts.append(f"**Content:** {doc.page_content}")
+                    if index < len(knowledge_sources) - 1:
+                        result_parts.append("-" * 80)
+
+            # Show conversation memories
+            if conversation_memories:
+                if knowledge_sources:
+                    result_parts.append("")
+                result_parts.append("## 💭 Related Experience:")
+                for index, doc in enumerate(conversation_memories):
+                    timestamp = doc.metadata.get('timestamp', 'Unknown time')
+                    area = doc.metadata.get('area', 'main').upper()
+                    consolidation_action = doc.metadata.get('consolidation_action', '')
+
+                    metadata_info = f"{timestamp} [{area}]"
+                    if consolidation_action:
+                        metadata_info += f" (consolidated: {consolidation_action})"
+
+                    result_parts.append(f"**Experience:** {metadata_info}")
+                    result_parts.append(f"**Content:** {doc.page_content}")
+                    if index < len(conversation_memories) - 1:
+                        result_parts.append("-" * 80)
+
+            return "\n".join(result_parts)
+
+        except Exception as e:
+            handle_error(e)
+            return f"Memory search failed: {str(e)}"
+
+    async def _get_memory_diagnostics(self, db, query: str):
+        """Provide memory diagnostics when no search results are found."""
+        try:
+            # Get sample of all documents to see what's in memory
+            sample_docs = await db.search_similarity_threshold(
+                query="test", limit=20, threshold=0.0
+            )
+
+            if not sample_docs:
+                return f"## 🔍 No Results for: '{query}'\n**Memory database appears to be empty.**"
+
+            # Analyze what's in memory
+            area_counts: dict[str, int] = {}
+            knowledge_count = 0
+
+            for doc in sample_docs:
+                area = doc.metadata.get('area', 'unknown')
+                area_counts[area] = area_counts.get(area, 0) + 1
+                if doc.metadata.get('knowledge_source', False):
+                    knowledge_count += 1
+
+            result_parts = [
+                f"## 🔍 No Results for: '{query}'",
+                f"**Database contains:** {len(sample_docs)} total documents",
+                f"**Areas:** {', '.join([f'{area.upper()}: {count}' for area, count in area_counts.items()])}",
+                f"**Knowledge sources:** {knowledge_count} documents",
+                "",
+                "**Suggestions:**",
+                "- Try different or more general search terms",
+                "- Check if the information was recently memorized",
+                f"- Current search threshold: {DEFAULT_MEMORY_THRESHOLD}"
+            ]
+
+            return "\n".join(result_parts)
+
+        except Exception as e:
+            return f"Memory diagnostics failed: {str(e)}"
+
     def format_result(self, result, source):
         if isinstance(result, Exception):
             handle_error(result)
@@ -113,6 +243,9 @@ class Knowledge(Tool):
             handle_error(result)
             return f"{source} search failed: {str(result)}"
 
+        if not result or "results" not in result:
+            return ""
+
         outputs = []
         for item in result["results"]:
             if "qa" in item:

run_cli.py

@@ -1,116 +0,0 @@
-import asyncio
-import sys
-import threading, time, models, os
-from ansio import application_keypad, mouse_input, raw_input
-from ansio.input import InputEvent, get_input_event
-from agent import AgentContext, UserMessage
-from python.helpers.print_style import PrintStyle
-from python.helpers.files import read_file
-from python.helpers import files
-import python.helpers.timed_input as timed_input
-from initialize import initialize_agent
-from python.helpers.dotenv import load_dotenv
-
-
-context: AgentContext = None # type: ignore
-input_lock = threading.Lock()
-
-
-# Main conversation loop
-async def chat(context: AgentContext):
-    
-    # start the conversation loop  
-    while True:
-        # ask user for message
-        with input_lock:
-            timeout = context.agent0.get_data("timeout") # how long the agent is willing to wait
-            if not timeout: # if agent wants to wait for user input forever
-                PrintStyle(background_color="#6C3483", font_color="white", bold=True, padding=True).print(f"User message ('e' to leave):")        
-                if sys.platform != "win32": import readline # this fixes arrow keys in terminal
-                user_input = input("> ")
-                PrintStyle(font_color="white", padding=False, log_only=True).print(f"> {user_input}") 
-                
-            else: # otherwise wait for user input with a timeout
-                PrintStyle(background_color="#6C3483", font_color="white", bold=True, padding=True).print(f"User message ({timeout}s timeout, 'w' to wait, 'e' to leave):")        
-                if sys.platform != "win32": import readline # this fixes arrow keys in terminal
-                # user_input = timed_input("> ", timeout=timeout)
-                user_input = timeout_input("> ", timeout=timeout)
-                                    
-                if not user_input:
-                    user_input = context.agent0.read_prompt("fw.msg_timeout.md")
-                    PrintStyle(font_color="white", padding=False).stream(f"{user_input}")        
-                else:
-                    user_input = user_input.strip()
-                    if user_input.lower()=="w": # the user needs more time
-                        user_input = input("> ").strip()
-                    PrintStyle(font_color="white", padding=False, log_only=True).print(f"> {user_input}")        
-                    
-                    
-
-        # exit the conversation when the user types 'exit'
-        if user_input.lower() == 'e': break
-
-        # send message to agent0, 
-        assistant_response = await context.communicate(UserMessage(user_input, [])).result()
-        
-        # print agent0 response
-        PrintStyle(font_color="white",background_color="#1D8348", bold=True, padding=True).print(f"{context.agent0.agent_name}: reponse:")        
-        PrintStyle(font_color="white").print(f"{assistant_response}")        
-                        
-
-# User intervention during agent streaming
-def intervention():
-    if context.streaming_agent and not context.paused:
-        context.paused = True # stop agent streaming
-        PrintStyle(background_color="#6C3483", font_color="white", bold=True, padding=True).print(f"User intervention ('e' to leave, empty to continue):")        
-
-        if sys.platform != "win32": import readline # this fixes arrow keys in terminal
-        user_input = input("> ").strip()
-        PrintStyle(font_color="white", padding=False, log_only=True).print(f"> {user_input}")        
-        
-        if user_input.lower() == 'e': os._exit(0) # exit the conversation when the user types 'exit'
-        if user_input: context.streaming_agent.intervention = UserMessage(user_input, []) # set intervention message if non-empty
-        context.paused = False # continue agent streaming 
-    
-
-# Capture keyboard input to trigger user intervention
-def capture_keys():
-        global input_lock
-        intervent=False            
-        while True:
-            if intervent: intervention()
-            intervent = False
-            time.sleep(0.1)
-            
-            if context.streaming_agent:
-                # with raw_input, application_keypad, mouse_input:
-                with input_lock, raw_input, application_keypad:
-                    event: InputEvent | None = get_input_event(timeout=0.1)
-                    if event and (event.shortcut.isalpha() or event.shortcut.isspace()):
-                        intervent=True
-                        continue
-
-# User input with timeout
-def timeout_input(prompt, timeout=10):
-    return timed_input.timeout_input(prompt=prompt, timeout=timeout)
-
-def run():
-    global context
-    PrintStyle.standard("Initializing framework...")
-
-    #load env vars
-    load_dotenv()
-
-    # initialize context
-    config = initialize_agent()
-    context = AgentContext(config)
-
-    # Start the key capture thread for user intervention during agent streaming
-    threading.Thread(target=capture_keys, daemon=True).start()
-
-    #start the chat
-    asyncio.run(chat(context))
-
-if __name__ == "__main__":
-    PrintStyle.standard("\n\n!!! run_cli.py is now discontinued. run_ui.py serves as both UI and API endpoint !!!\n\n")
-    run()

tests/mcp/stream_http_mcp_server.py

@@ -1,223 +0,0 @@
-#!/usr/bin/env python3
-"""
-Hello World MCP Server using FastMCP with Streamable HTTP Protocol
-
-This is a simple example demonstrating how to create an MCP server using
-the FastMCP framework with the streamable-http transport protocol.
-
-Features:
-- Hello world tool that greets users
-- Simple resource that provides server information
-- Basic prompt template for greeting
-- Runs using streamable-http transport for better scalability
-"""
-
-from fastmcp import FastMCP, Context
-import os
-from datetime import datetime
-
-
-# Create a FastMCP server instance
-mcp: FastMCP = FastMCP(
-    "Hello World Server 🚀",
-    dependencies=[]  # No special dependencies for this simple example
-)
-
-
-# ========== TOOLS ==========
-
-@mcp.tool()
-def hello_world(name: str = "World") -> str:
-    """Say hello to someone with a personalized greeting.
-
-    Args:
-        name: The name of the person to greet (defaults to "World")
-
-    Returns:
-        A friendly greeting message
-    """
-    current_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
-    return f"Hello, {name}! 👋 Welcome to the FastMCP Hello World Server. Current time: {current_time}"
-
-
-@mcp.tool()
-def add_numbers(a: float, b: float) -> float:
-    """Add two numbers together.
-
-    Args:
-        a: First number
-        b: Second number
-
-    Returns:
-        The sum of the two numbers
-    """
-    result = a + b
-    return result
-
-
-@mcp.tool()
-async def get_server_status(ctx: Context) -> str:
-    """Get the current server status and information.
-
-    Returns:
-        Server status information including uptime and capabilities
-    """
-    # Log that someone is checking server status
-    await ctx.info("Server status requested")
-
-    # Get basic server info
-    server_info = {
-        "status": "running",
-        "protocol": "MCP (Model Context Protocol)",
-        "transport": "streamable-http",
-        "framework": "FastMCP 2.0",
-        "capabilities": ["tools", "resources", "prompts"],
-        "timestamp": datetime.now().isoformat()
-    }
-
-    return f"""
-🟢 Server Status: {server_info['status'].upper()}
-
-📊 Server Information:
-• Protocol: {server_info['protocol']}
-• Transport: {server_info['transport']}
-• Framework: {server_info['framework']}
-• Capabilities: {', '.join(server_info['capabilities'])}
-• Last checked: {server_info['timestamp']}
-
-✅ All systems operational!
-"""
-
-
-# ========== RESOURCES ==========
-
-@mcp.resource("info://server")
-def get_server_info() -> str:
-    """Static resource providing information about this MCP server."""
-    return """
-🚀 Hello World MCP Server
-
-This is a demonstration MCP server built with FastMCP, showcasing the
-streamable-http transport protocol.
-
-Available capabilities:
-• Tools: Interactive functions the LLM can call
-• Resources: Data sources for context
-• Prompts: Reusable message templates
-
-Built with FastMCP 2.0 for production-ready MCP applications.
-"""
-
-
-@mcp.resource("greeting://{user_name}")
-def get_personal_greeting(user_name: str) -> str:
-    """Dynamic resource template that provides personalized greetings.
-
-    Args:
-        user_name: The name of the user to create a greeting for
-
-    Returns:
-        A personalized greeting message
-    """
-    greetings = [
-        f"Welcome, {user_name}! 🎉",
-        f"Hello there, {user_name}! Great to see you! 👋",
-        f"Greetings, {user_name}! Hope you're having a wonderful day! ☀️"
-    ]
-
-    # Select greeting based on name length (simple example)
-    greeting_index = len(user_name) % len(greetings)
-    return greetings[greeting_index]
-
-
-# ========== PROMPTS ==========
-
-@mcp.prompt()
-def introduction_prompt(user_name: str = "friend") -> str:
-    """Generate a friendly introduction prompt.
-
-    Args:
-        user_name: Name of the person to introduce to
-
-    Returns:
-        A prompt for introducing the MCP server capabilities
-    """
-    return f"""
-Hello {user_name}! 👋
-
-I'm your Hello World MCP Server, here to demonstrate the power of the Model Context Protocol with FastMCP!
-
-Here's what I can help you with:
-
-🔧 **Tools I can execute:**
-• hello_world - Give you personalized greetings
-• add_numbers - Perform simple math operations
-• get_server_status - Check my current status
-
-📚 **Resources I can provide:**
-• Server information and documentation
-• Personalized greeting messages
-
-💡 **How to use me:**
-Try asking me to say hello, add some numbers, or check my status!
-
-What would you like to do first?
-"""
-
-
-@mcp.prompt()
-def math_prompt(operation: str = "addition") -> str:
-    """Create a prompt for helping with math operations.
-
-    Args:
-        operation: The type of math operation to help with
-
-    Returns:
-        A simple prompt for math assistance
-    """
-    return (f"I need help with {operation}. I'd be happy to help you with {operation}! "
-            f"I can add numbers together using my add_numbers tool. "
-            f"Just tell me which numbers you'd like me to work with.")
-
-
-# ========== SERVER LIFECYCLE ==========
-
-def main():
-    """Main function to run the MCP server."""
-    print("🚀 Starting Hello World MCP Server with Streamable HTTP...")
-    print("📡 Transport: streamable-http")
-    print("🌐 Framework: FastMCP 2.0")
-    print("🔗 Protocol: Model Context Protocol (MCP)")
-    print()
-
-    # Get configuration from environment or use defaults
-    host = os.getenv("MCP_HOST", "0.0.0.0")
-    port = int(os.getenv("MCP_PORT", "8000"))
-    path = os.getenv("MCP_PATH", "/mcp")
-
-    print(f"🏠 Host: {host}")
-    print(f"🚪 Port: {port}")
-    print(f"🛤️  Path: {path}")
-    print(f"📍 Full URL: http://{host}:{port}{path}")
-    print()
-    print("✅ Server is ready to accept MCP connections!")
-    print("💡 Use this server with MCP clients that support streamable-http transport")
-    print()
-
-    # Run the server with streamable-http transport
-    try:
-        mcp.run(
-            transport="streamable-http",
-            host=host,
-            port=port,
-            path=path
-        )
-    except KeyboardInterrupt:
-        print("\n👋 Server shutting down gracefully...")
-    except Exception as e:
-        print(f"❌ Server error: {e}")
-        raise
-
-
-if __name__ == "__main__":
-    main()

tests/mcp/stream_http_mcp_server_README.md

@@ -1,208 +0,0 @@
-# FastMCP Hello World Server with Streamable HTTP
-
-A comprehensive hello world example demonstrating how to build an MCP (Model Context Protocol) server using the FastMCP framework with streamable-http transport.
-
-## 🚀 Features
-
-This server demonstrates all three core MCP primitives:
-
-### 🔧 Tools (LLM-callable functions)
-- **hello_world** - Personalized greetings with timestamps
-- **add_numbers** - Simple math operations
-- **get_server_status** - Server status and information with context logging
-
-### 📚 Resources (Data sources)
-- **info://server** - Static server information
-- **greeting://{user_name}** - Dynamic personalized greetings template
-
-### 💡 Prompts (Reusable templates)
-- **introduction_prompt** - Server capability introduction
-- **math_prompt** - Math assistance template
-
-## 📋 Prerequisites
-
-- Python 3.10+
-- pip or uv package manager
-
-## 🛠️ Installation
-
-### Option 1: Using pip
-```bash
-# Install dependencies
-pip install -r stream_http_mcp_server_requirements.txt
-
-# Or install FastMCP directly
-pip install fastmcp
-```
-
-### Option 2: Using uv (recommended)
-```bash
-# Install FastMCP with uv
-uv pip install fastmcp
-```
-
-## ▶️ Running the Server
-
-### Basic Usage
-```bash
-# Run with default settings (localhost:8000/mcp)
-python stream_http_mcp_server.py
-```
-
-### Custom Configuration via Environment Variables
-```bash
-# Set custom host, port, and path
-export MCP_HOST=0.0.0.0
-export MCP_PORT=3000
-export MCP_PATH=/hello-mcp
-
-python stream_http_mcp_server.py
-```
-
-### Expected Output
-```
-🚀 Starting Hello World MCP Server with Streamable HTTP...
-📡 Transport: streamable-http
-🌐 Framework: FastMCP 2.0
-🔗 Protocol: Model Context Protocol (MCP)
-
-🏠 Host: 127.0.0.1
-🚪 Port: 8000
-🛤️  Path: /mcp
-📍 Full URL: http://127.0.0.1:8000/mcp
-
-✅ Server is ready to accept MCP connections!
-💡 Use this server with MCP clients that support streamable-http transport
-```
-
-## 🧪 Testing the Server
-
-### Method 1: Using MCP Inspector (Recommended)
-
-1. **Install MCP Inspector**:
-   ```bash
-   npm install -g @modelcontextprotocol/inspector
-   ```
-
-2. **Run the Inspector**:
-   ```bash
-   npx @modelcontextprotocol/inspector
-   ```
-
-3. **Connect to the Server**:
-   - Choose "Streamable HTTP" transport
-   - Enter URL: `http://localhost:8000/mcp`
-   - Click "Connect"
-
-4. **Test Tools**:
-   - Go to the "Tools" tab
-   - Try `hello_world` with `{"name": "Alice"}`
-   - Try `add_numbers` with `{"a": 5, "b": 3}`
-   - Try `get_server_status` (no parameters needed)
-
-5. **Test Resources**:
-   - Go to "Resources" tab
-   - View `info://server`
-   - Try `greeting://YourName`
-
-6. **Test Prompts**:
-   - Go to "Prompts" tab
-   - Try `introduction_prompt` with `{"user_name": "Developer"}`
-   - Try `math_prompt` with `{"operation": "multiplication"}`
-
-### Method 2: Agent Zero Integration
-
-Configure Agent Zero to use this server by adding to your MCP servers configuration:
-
-```json
-[
-  {
-    "name": "hello_world_server",
-    "type": "streamable-http",
-    "url": "http://localhost:8000/mcp",
-    "description": "Hello World FastMCP Server with streamable HTTP"
-  }
-]
-```
-
-### Method 3: Custom MCP Client
-
-Example using the MCP Python SDK:
-
-```python
-from mcp.client.streamable_http import streamablehttp_client
-from mcp import ClientSession
-
-async def test_server():
-    async with streamablehttp_client("http://localhost:8000/mcp") as (read, write, get_session_id):
-        async with ClientSession(read, write) as session:
-            await session.initialize()
-
-            # Test tool
-            result = await session.call_tool("hello_world", {"name": "Test"})
-            print(f"Tool result: {result}")
-
-            # Test resource
-            resource = await session.read_resource("info://server")
-            print(f"Resource: {resource}")
-
-# Run with: asyncio.run(test_server())
-```
-
-## 🔧 Configuration Options
-
-### Environment Variables
-- `MCP_HOST` - Server host (default: 127.0.0.1)
-- `MCP_PORT` - Server port (default: 8000)
-- `MCP_PATH` - Server path (default: /mcp)
-
-### Server Capabilities
-This server supports all MCP capabilities:
-- ✅ Tools (with async support and context logging)
-- ✅ Resources (static and dynamic templates)
-- ✅ Prompts (string and message-based)
-- ✅ Streamable HTTP transport
-- ✅ Session management
-
-## 🎯 Key Concepts Demonstrated
-
-1. **FastMCP Framework**: Modern, production-ready MCP server development
-2. **Streamable HTTP Transport**: Scalable transport for web deployments
-3. **Type Safety**: Full Python type hints and docstrings
-4. **Async Support**: Proper async/await patterns with context
-5. **Dynamic Resources**: Template-based resources with parameters
-6. **Context Logging**: Using MCP context for client communication
-7. **Error Handling**: Graceful startup and shutdown
-
-## 📚 Next Steps
-
-- **Scale Up**: Use FastMCP's server composition to mount multiple apps
-- **Add Auth**: Implement OAuth authentication for production
-- **Deploy**: Use Docker or cloud platforms for production deployment
-- **Integrate**: Connect with Claude Desktop, Agent Zero, or custom clients
-- **Extend**: Add more sophisticated tools, resources, and prompts
-
-## 🐛 Troubleshooting
-
-### Server Won't Start
-- Check if port 8000 is available: `lsof -i :8000`
-- Try a different port: `MCP_PORT=8001 python stream_http_mcp_server.py`
-
-### Connection Issues
-- Verify the URL in your client matches the server output
-- Check firewall settings for the port
-- Ensure you're using "streamable-http" transport type
-
-### Import Errors
-- Install FastMCP: `pip install fastmcp`
-- Check Python version: `python --version` (requires 3.10+)
-
-## 📖 Documentation Links
-
-- [FastMCP Documentation](https://gofastmcp.com/)
-- [MCP Specification](https://spec.modelcontextprotocol.io/)
-- [Agent Zero MCP Integration](../../docs/mcp_setup.md)
-
----
-
-Built with ❤️ using FastMCP 2.0 and the Model Context Protocol

tests/mcp/stream_http_mcp_server_requirements.txt

@@ -1,9 +0,0 @@
-# FastMCP Hello World Server Requirements
-# Install with: pip install -r stream_http_mcp_server_requirements.txt
-
-# FastMCP framework for building MCP servers
-fastmcp>=2.8.0
-
-# Optional: Additional dependencies that might be useful
-# uvicorn>=0.18.0  # ASGI server (may be included with FastMCP)
-# httpx>=0.24.0    # HTTP client (may be included with FastMCP)