Fix conversation history duplication and optimize file embedding

This major refactoring addresses critical bugs in conversation history management
and significantly improves token efficiency through intelligent file embedding:

**Key Improvements:**
• Fixed conversation history duplication bug by centralizing reconstruction in server.py
• Added intelligent file filtering to prevent re-embedding files already in conversation history
• Centralized file processing logic in BaseTool._prepare_file_content_for_prompt()
• Enhanced log monitoring with better categorization and file embedding visibility
• Updated comprehensive test suite to verify new architecture and edge cases

**Architecture Changes:**
• Removed duplicate conversation history reconstruction from tools/base.py
• Conversation history now handled exclusively by server.py:reconstruct_thread_context
• All tools now use centralized file processing with automatic deduplication
• Improved token efficiency by embedding unique files only once per conversation

**Performance Benefits:**
• Reduced token usage through smart file filtering
• Eliminated redundant file embeddings in continued conversations
• Better observability with detailed debug logging for file operations

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-Authored-By: Claude <noreply@anthropic.com>

README.md

@@ -10,6 +10,8 @@
 
 The ultimate development partner for Claude - a Model Context Protocol server that gives Claude access to Google's Gemini 2.5 Pro for extended thinking, code analysis, and problem-solving. **Automatically reads files and directories, passing their contents to Gemini for analysis within its 1M token context.**
 
+**Features true AI orchestration with conversation continuity across tool usage** - start a task with one tool, continue with another, and maintain full context throughout. Claude and Gemini can collaborate seamlessly across multiple interactions and different tools, creating a unified development experience.
+
 **Think of it as Claude Code _for_ Claude Code.**
 
 ## Quick Navigation

log_monitor.py

@@ -73,6 +73,14 @@ def monitor_mcp_activity():
                                 print(f"[{datetime.now().strftime('%H:%M:%S')}] ❌ {line}")
                             elif "WARNING" in line:
                                 print(f"[{datetime.now().strftime('%H:%M:%S')}] ⚠️  {line}")
+                            elif "DEBUG" in line:
+                                # Highlight file embedding debug logs
+                                if "📄" in line or "📁" in line:
+                                    print(f"[{datetime.now().strftime('%H:%M:%S')}] 📂 FILE: {line}")
+                                else:
+                                    print(f"[{datetime.now().strftime('%H:%M:%S')}] 🔍 {line}")
+                            elif "INFO" in line and ("Gemini API" in line or "Tool" in line or "Conversation" in line):
+                                print(f"[{datetime.now().strftime('%H:%M:%S')}] ℹ️  {line}")
                             elif "Gemini API" in line and ("Sending" in line or "Received" in line):
                                 print(f"[{datetime.now().strftime('%H:%M:%S')}] API: {line}")
 

tests/test_conversation_history_bug.py

@@ -68,13 +68,12 @@ class TestConversationHistoryBugFix:
     def setup_method(self):
         self.tool = FileContextTool()
 
-    @patch("tools.base.get_thread")
     @patch("tools.base.add_turn")
-    async def test_conversation_history_included_with_continuation_id(self, mock_add_turn, mock_get_thread):
+    async def test_conversation_history_included_with_continuation_id(self, mock_add_turn):
         """Test that conversation history (including file context) is included when using continuation_id"""
 
         # Create a thread context with previous turns including files
-        thread_context = ThreadContext(
+        _thread_context = ThreadContext(
             thread_id="test-history-id",
             created_at="2023-01-01T00:00:00Z",
             last_updated_at="2023-01-01T00:02:00Z",
@@ -98,8 +97,6 @@ class TestConversationHistoryBugFix:
             initial_context={"question": "Analyze authentication security"},
         )
 
-        # Mock get_thread to return our test context
-        mock_get_thread.return_value = thread_context
         # Mock add_turn to return success
         mock_add_turn.return_value = True
 
@@ -125,6 +122,9 @@ class TestConversationHistoryBugFix:
             mock_create_model.return_value = mock_model
 
             # Execute tool with continuation_id
+            # In the corrected flow, server.py:reconstruct_thread_context
+            # would have already added conversation history to the prompt
+            # This test simulates that the prompt already contains conversation history
             arguments = {
                 "prompt": "What should we fix first?",
                 "continuation_id": "test-history-id",
@@ -136,38 +136,30 @@ class TestConversationHistoryBugFix:
             response_data = json.loads(response[0].text)
             assert response_data["status"] == "success"
 
-            # Verify get_thread was called for history reconstruction
-            mock_get_thread.assert_called_with("test-history-id")
+            # Note: After fixing the duplication bug, conversation history reconstruction
+            # now happens ONLY in server.py, not in tools/base.py
+            # This test verifies that tools/base.py no longer duplicates conversation history
 
-            # Verify the prompt includes conversation history
+            # Verify the prompt is captured
             assert captured_prompt is not None
 
-            # Check that conversation history is included
-            assert "=== CONVERSATION HISTORY ===" in captured_prompt
-            assert "Turn 1 (Gemini using analyze)" in captured_prompt
-            assert "Turn 2 (Gemini using codereview)" in captured_prompt
+            # The prompt should NOT contain conversation history (since we removed the duplicate code)
+            # In the real flow, server.py would add conversation history before calling tool.execute()
+            assert "=== CONVERSATION HISTORY ===" not in captured_prompt
 
-            # Check that file context from previous turns is included
-            assert "📁 Files referenced: /src/auth.py, /src/security.py" in captured_prompt
-            assert "📁 Files referenced: /src/auth.py, /tests/test_auth.py" in captured_prompt
-
-            # Check that previous turn content is included
-            assert "I've analyzed the authentication module and found several security issues." in captured_prompt
-            assert "The code review shows these files have critical vulnerabilities." in captured_prompt
-
-            # Check that continuation instruction is included
-            assert "Continue this conversation by building on the previous context." in captured_prompt
-
-            # Check that current request is still included
+            # The prompt should contain the current request
             assert "What should we fix first?" in captured_prompt
             assert "Files in current request: /src/utils.py" in captured_prompt
 
-    @patch("tools.base.get_thread")
-    async def test_no_history_when_thread_not_found(self, mock_get_thread):
+            # This test confirms the duplication bug is fixed - tools/base.py no longer
+            # redundantly adds conversation history that server.py already added
+
+    async def test_no_history_when_thread_not_found(self):
         """Test graceful handling when thread is not found"""
 
-        # Mock get_thread to return None (thread not found)
-        mock_get_thread.return_value = None
+        # Note: After fixing the duplication bug, thread not found handling
+        # happens in server.py:reconstruct_thread_context, not in tools/base.py
+        # This test verifies tools don't try to handle missing threads themselves
 
         captured_prompt = None
 
@@ -190,17 +182,16 @@ class TestConversationHistoryBugFix:
             mock_create_model.return_value = mock_model
 
             # Execute tool with continuation_id for non-existent thread
+            # In the real flow, server.py would have already handled the missing thread
             arguments = {"prompt": "Test without history", "continuation_id": "non-existent-thread-id"}
             response = await self.tool.execute(arguments)
 
-            # Should still succeed but without history
+            # Should succeed since tools/base.py no longer handles missing threads
             response_data = json.loads(response[0].text)
             assert response_data["status"] == "success"
 
-            # Verify get_thread was called for non-existent thread
-            mock_get_thread.assert_called_with("non-existent-thread-id")
-
             # Verify the prompt does NOT include conversation history
+            # (because tools/base.py no longer tries to add it)
             assert captured_prompt is not None
             assert "=== CONVERSATION HISTORY ===" not in captured_prompt
             assert "Test without history" in captured_prompt
@@ -246,6 +237,113 @@ class TestConversationHistoryBugFix:
             # (This is the existing behavior for new conversations)
             assert "If you'd like to ask a follow-up question" in captured_prompt
 
+    @patch("tools.base.get_thread")
+    @patch("tools.base.add_turn")
+    @patch("utils.file_utils.resolve_and_validate_path")
+    async def test_no_duplicate_file_embedding_during_continuation(
+        self, mock_resolve_path, mock_add_turn, mock_get_thread
+    ):
+        """Test that files already embedded in conversation history are not re-embedded"""
+
+        # Mock file resolution to allow our test files
+        def mock_resolve(path_str):
+            from pathlib import Path
+
+            return Path(path_str)  # Just return as-is for test files
+
+        mock_resolve_path.side_effect = mock_resolve
+
+        # Create a thread context with previous turns including files
+        _thread_context = ThreadContext(
+            thread_id="test-duplicate-files-id",
+            created_at="2023-01-01T00:00:00Z",
+            last_updated_at="2023-01-01T00:02:00Z",
+            tool_name="analyze",
+            turns=[
+                ConversationTurn(
+                    role="assistant",
+                    content="I've analyzed the authentication module.",
+                    timestamp="2023-01-01T00:01:00Z",
+                    tool_name="analyze",
+                    files=["/src/auth.py", "/src/security.py"],  # These files were already analyzed
+                ),
+                ConversationTurn(
+                    role="assistant",
+                    content="Found security issues in the auth system.",
+                    timestamp="2023-01-01T00:02:00Z",
+                    tool_name="codereview",
+                    files=["/src/auth.py", "/tests/test_auth.py"],  # auth.py referenced again + new file
+                ),
+            ],
+            initial_context={"question": "Analyze authentication security"},
+        )
+
+        # Mock get_thread to return our test context
+        mock_get_thread.return_value = _thread_context
+        mock_add_turn.return_value = True
+
+        # Mock the model to capture what prompt it receives
+        captured_prompt = None
+
+        with patch.object(self.tool, "create_model") as mock_create_model:
+            mock_model = Mock()
+            mock_response = Mock()
+            mock_response.candidates = [
+                Mock(
+                    content=Mock(parts=[Mock(text="Analysis of new files complete")]),
+                    finish_reason="STOP",
+                )
+            ]
+
+            def capture_prompt(prompt):
+                nonlocal captured_prompt
+                captured_prompt = prompt
+                return mock_response
+
+            mock_model.generate_content.side_effect = capture_prompt
+            mock_create_model.return_value = mock_model
+
+            # Mock read_files to simulate file existence and capture its calls
+            with patch("tools.base.read_files") as mock_read_files:
+                # When the tool processes the new files, it should only read '/src/utils.py'
+                mock_read_files.return_value = "--- /src/utils.py ---\ncontent of utils"
+
+                # Execute tool with continuation_id and mix of already-referenced and new files
+                arguments = {
+                    "prompt": "Now check the utility functions too",
+                    "continuation_id": "test-duplicate-files-id",
+                    "files": ["/src/auth.py", "/src/utils.py"],  # auth.py already in history, utils.py is new
+                }
+                response = await self.tool.execute(arguments)
+
+                # Verify response succeeded
+                response_data = json.loads(response[0].text)
+                assert response_data["status"] == "success"
+
+                # Verify the prompt structure
+                assert captured_prompt is not None
+
+                # After fixing the duplication bug, conversation history (including file embedding)
+                # is no longer added by tools/base.py - it's handled by server.py
+                # This test verifies the file filtering logic still works correctly
+
+                # The current request should still be processed normally
+                assert "Now check the utility functions too" in captured_prompt
+                assert "Files in current request: /src/auth.py, /src/utils.py" in captured_prompt
+
+                # Most importantly, verify that the file filtering logic works correctly
+                # even though conversation history isn't built by tools/base.py anymore
+                with patch.object(self.tool, "get_conversation_embedded_files") as mock_get_embedded:
+                    # Mock that certain files are already embedded
+                    mock_get_embedded.return_value = ["/src/auth.py", "/src/security.py", "/tests/test_auth.py"]
+
+                    # Test the filtering logic directly
+                    new_files = self.tool.filter_new_files(["/src/auth.py", "/src/utils.py"], "test-duplicate-files-id")
+                    assert new_files == ["/src/utils.py"]  # Only the new file should remain
+
+                    # Verify get_conversation_embedded_files was called correctly
+                    mock_get_embedded.assert_called_with("test-duplicate-files-id")
+
 
 if __name__ == "__main__":
     pytest.main([__file__])

tests/test_conversation_memory.py

@@ -183,8 +183,13 @@ class TestConversationMemory:
         assert "Python is a programming language" in history
 
         # Test file tracking
-        assert "📁 Files referenced: /home/user/main.py, /home/user/docs/readme.md" in history
-        assert "📁 Files referenced: /home/user/examples/" in history
+        # Check that the new file embedding section is included
+        assert "=== FILES REFERENCED IN THIS CONVERSATION ===" in history
+        assert "The following files have been shared and analyzed during our conversation." in history
+
+        # Check that file context from previous turns is included (now shows files used per turn)
+        assert "📁 Files used in this turn: /home/user/main.py, /home/user/docs/readme.md" in history
+        assert "📁 Files used in this turn: /home/user/examples/" in history
 
         # Test follow-up attribution
         assert "[Gemini's Follow-up: Would you like examples?]" in history
@@ -598,9 +603,9 @@ class TestConversationFlow:
         assert "--- Turn 3 (Gemini using analyze) ---" in history
 
         # Verify all files are preserved in chronological order
-        turn_1_files = "📁 Files referenced: /project/src/main.py, /project/src/utils.py"
-        turn_2_files = "📁 Files referenced: /project/tests/, /project/test_main.py"
-        turn_3_files = "📁 Files referenced: /project/tests/test_utils.py, /project/coverage.html"
+        turn_1_files = "📁 Files used in this turn: /project/src/main.py, /project/src/utils.py"
+        turn_2_files = "📁 Files used in this turn: /project/tests/, /project/test_main.py"
+        turn_3_files = "📁 Files used in this turn: /project/tests/test_utils.py, /project/coverage.html"
 
         assert turn_1_files in history
         assert turn_2_files in history
@@ -718,6 +723,63 @@ class TestConversationFlow:
         assert len(retrieved_context.turns) == 1
         assert retrieved_context.turns[0].follow_up_question == "Want to explore scalability?"
 
+    def test_token_limit_optimization_in_conversation_history(self):
+        """Test that build_conversation_history efficiently handles token limits"""
+        import os
+        import tempfile
+
+        from utils.conversation_memory import build_conversation_history
+
+        # Create test files with known content sizes
+        with tempfile.TemporaryDirectory() as temp_dir:
+            # Create small and large test files
+            small_file = os.path.join(temp_dir, "small.py")
+            large_file = os.path.join(temp_dir, "large.py")
+
+            small_content = "# Small file\nprint('hello')\n"
+            large_content = "# Large file\n" + "x = 1\n" * 10000  # Very large file
+
+            with open(small_file, "w") as f:
+                f.write(small_content)
+            with open(large_file, "w") as f:
+                f.write(large_content)
+
+            # Create context with files that would exceed token limit
+            context = ThreadContext(
+                thread_id="test-token-limit",
+                created_at="2023-01-01T00:00:00Z",
+                last_updated_at="2023-01-01T00:01:00Z",
+                tool_name="analyze",
+                turns=[
+                    ConversationTurn(
+                        role="user",
+                        content="Analyze these files",
+                        timestamp="2023-01-01T00:00:30Z",
+                        files=[small_file, large_file],  # Large file should be truncated
+                    )
+                ],
+                initial_context={"prompt": "Analyze code"},
+            )
+
+            # Build conversation history (should handle token limits gracefully)
+            history = build_conversation_history(context)
+
+            # Verify the history was built successfully
+            assert "=== CONVERSATION HISTORY ===" in history
+            assert "=== FILES REFERENCED IN THIS CONVERSATION ===" in history
+
+            # The small file should be included, but large file might be truncated
+            # At minimum, verify no crashes and history is generated
+            assert len(history) > 0
+
+            # If truncation occurred, there should be a note about it
+            if "additional file(s) were truncated due to token limit" in history:
+                assert small_file in history or large_file in history
+            else:
+                # Both files fit within limit
+                assert small_file in history
+                assert large_file in history
+
 
 if __name__ == "__main__":
     pytest.main([__file__])

tests/test_prompt_regression.py

@@ -117,7 +117,7 @@ class TestPromptRegression:
             mock_create_model.return_value = mock_model
 
             # Mock file reading
-            with patch("tools.codereview.read_files") as mock_read_files:
+            with patch("tools.base.read_files") as mock_read_files:
                 mock_read_files.return_value = "def main(): pass"
 
                 result = await tool.execute(
@@ -205,7 +205,7 @@ class TestPromptRegression:
             mock_create_model.return_value = mock_model
 
             # Mock file reading
-            with patch("tools.analyze.read_files") as mock_read_files:
+            with patch("tools.base.read_files") as mock_read_files:
                 mock_read_files.return_value = "class UserController: ..."
 
                 result = await tool.execute(
@@ -287,7 +287,7 @@ class TestPromptRegression:
             mock_model.generate_content.return_value = mock_model_response()
             mock_create_model.return_value = mock_model
 
-            with patch("tools.analyze.read_files") as mock_read_files:
+            with patch("tools.base.read_files") as mock_read_files:
                 mock_read_files.return_value = "Content"
 
                 result = await tool.execute(

tools/analyze.py

@@ -9,7 +9,6 @@ from pydantic import Field
 
 from config import TEMPERATURE_ANALYTICAL
 from prompts import ANALYZE_PROMPT
-from utils import read_files
 
 from .base import BaseTool, ToolRequest
 from .models import ToolOutput
@@ -132,11 +131,9 @@ class AnalyzeTool(BaseTool):
         if updated_files is not None:
             request.files = updated_files
 
-        # Read all files
-        file_content = read_files(request.files)
-
-        # Check token limits
-        self._validate_token_limit(file_content, "Files")
+        # Use centralized file processing logic
+        continuation_id = getattr(request, "continuation_id", None)
+        file_content = self._prepare_file_content_for_prompt(request.files, continuation_id, "Files")
 
         # Build analysis instructions
         analysis_focus = []

tools/base.py

@@ -30,14 +30,16 @@ from utils import check_token_limit
 from utils.conversation_memory import (
     MAX_CONVERSATION_TURNS,
     add_turn,
-    build_conversation_history,
     create_thread,
+    get_conversation_file_list,
     get_thread,
 )
-from utils.file_utils import read_file_content, translate_path_for_environment
+from utils.file_utils import read_file_content, read_files, translate_path_for_environment
 
 from .models import ClarificationRequest, ContinuationOffer, FollowUpRequest, ToolOutput
 
+logger = logging.getLogger(__name__)
+
 
 class ToolRequest(BaseModel):
     """
@@ -163,6 +165,123 @@ class BaseTool(ABC):
         """
         return "medium"  # Default to medium thinking for better reasoning
 
+    def get_conversation_embedded_files(self, continuation_id: Optional[str]) -> list[str]:
+        """
+        Get list of files already embedded in conversation history.
+
+        This method returns the list of files that have already been embedded
+        in the conversation history for a given continuation thread. Tools can
+        use this to avoid re-embedding files that are already available in the
+        conversation context.
+
+        Args:
+            continuation_id: Thread continuation ID, or None for new conversations
+
+        Returns:
+            list[str]: List of file paths already embedded in conversation history
+        """
+        if not continuation_id:
+            # New conversation, no files embedded yet
+            return []
+
+        thread_context = get_thread(continuation_id)
+        if not thread_context:
+            # Thread not found, no files embedded
+            return []
+
+        return get_conversation_file_list(thread_context)
+
+    def filter_new_files(self, requested_files: list[str], continuation_id: Optional[str]) -> list[str]:
+        """
+        Filter out files that are already embedded in conversation history.
+
+        This method takes a list of requested files and removes any that have
+        already been embedded in the conversation history, preventing duplicate
+        file embeddings and optimizing token usage.
+
+        Args:
+            requested_files: List of files requested for current tool execution
+            continuation_id: Thread continuation ID, or None for new conversations
+
+        Returns:
+            list[str]: List of files that need to be embedded (not already in history)
+        """
+        if not continuation_id:
+            # New conversation, all files are new
+            return requested_files
+
+        embedded_files = set(self.get_conversation_embedded_files(continuation_id))
+
+        # Return only files that haven't been embedded yet
+        new_files = [f for f in requested_files if f not in embedded_files]
+
+        return new_files
+
+    def _prepare_file_content_for_prompt(
+        self, request_files: list[str], continuation_id: Optional[str], context_description: str = "New files"
+    ) -> str:
+        """
+        Centralized file processing for tool prompts.
+
+        This method handles the common pattern across all tools:
+        1. Filter out files already embedded in conversation history
+        2. Read content of only new files
+        3. Generate informative note about skipped files
+
+        Args:
+            request_files: List of files requested for current tool execution
+            continuation_id: Thread continuation ID, or None for new conversations
+            context_description: Description for token limit validation (e.g. "Code", "New files")
+
+        Returns:
+            str: Formatted file content string ready for prompt inclusion
+        """
+        if not request_files:
+            return ""
+
+        files_to_embed = self.filter_new_files(request_files, continuation_id)
+
+        content_parts = []
+
+        # Read content of new files only
+        if files_to_embed:
+            logger.debug(f"📁 {self.name} tool embedding {len(files_to_embed)} new files: {', '.join(files_to_embed)}")
+            try:
+                file_content = read_files(files_to_embed)
+                self._validate_token_limit(file_content, context_description)
+                content_parts.append(file_content)
+
+                # Estimate tokens for debug logging
+                from utils.token_utils import estimate_tokens
+
+                content_tokens = estimate_tokens(file_content)
+                logger.debug(
+                    f"📁 {self.name} tool successfully embedded {len(files_to_embed)} files ({content_tokens:,} tokens)"
+                )
+            except Exception as e:
+                logger.error(f"📁 {self.name} tool failed to embed files {files_to_embed}: {type(e).__name__}: {e}")
+                raise
+
+        # Generate note about files already in conversation history
+        if continuation_id and len(files_to_embed) < len(request_files):
+            embedded_files = self.get_conversation_embedded_files(continuation_id)
+            skipped_files = [f for f in request_files if f in embedded_files]
+            if skipped_files:
+                logger.debug(
+                    f"📁 {self.name} tool skipping {len(skipped_files)} files already in conversation history: {', '.join(skipped_files)}"
+                )
+                if content_parts:
+                    content_parts.append("\n\n")
+                note_lines = [
+                    "--- NOTE: Additional files referenced in conversation history ---",
+                    "The following files are already available in our conversation context:",
+                    "\n".join(f"  - {f}" for f in skipped_files),
+                    "--- END NOTE ---",
+                ]
+                content_parts.append("\n".join(note_lines))
+
+        return "".join(content_parts) if content_parts else ""
+
     def get_websearch_instruction(self, use_websearch: bool, tool_specific: Optional[str] = None) -> str:
         """
         Generate web search instruction based on the use_websearch parameter.
@@ -413,15 +532,8 @@ If any of these would strengthen your analysis, specify what Claude should searc
                     pass
             else:
                 logger.debug(f"Continuing {self.name} conversation with thread {continuation_id}")
-
-                # Add conversation history when continuing a threaded conversation
-                thread_context = get_thread(continuation_id)
-                if thread_context:
-                    conversation_history = build_conversation_history(thread_context)
-                    prompt = f"{conversation_history}\n\n{prompt}"
-                    logger.debug(f"Added conversation history to {self.name} prompt for thread {continuation_id}")
-                else:
-                    logger.warning(f"Thread {continuation_id} not found for {self.name} - continuing without history")
+                # History reconstruction is handled by server.py:reconstruct_thread_context
+                # No need to rebuild it here - prompt already contains conversation history
 
             # Extract model configuration from request or use defaults
             model_name = getattr(request, "model", None) or GEMINI_MODEL

tools/codereview.py

@@ -21,7 +21,6 @@ from pydantic import Field
 
 from config import TEMPERATURE_ANALYTICAL
 from prompts import CODEREVIEW_PROMPT
-from utils import read_files
 
 from .base import BaseTool, ToolRequest
 from .models import ToolOutput
@@ -70,6 +69,8 @@ class CodeReviewTool(BaseTool):
             "PROFESSIONAL CODE REVIEW - Comprehensive analysis for bugs, security, and quality. "
             "Supports both individual files and entire directories/projects. "
             "Use this when you need to review code, check for issues, find bugs, or perform security audits. "
+            "ALSO use this to validate claims about code, verify code flow and logic, confirm assertions, "
+            "cross-check functionality, or investigate how code actually behaves when you need to be certain. "
             "I'll identify issues by severity (Critical→High→Medium→Low) with specific fixes. "
             "Supports focused reviews: security, performance, or quick checks. "
             "Choose thinking_mode based on review scope: 'low' for small code snippets, "
@@ -185,11 +186,9 @@ class CodeReviewTool(BaseTool):
         if updated_files is not None:
             request.files = updated_files
 
-        # Read all requested files, expanding directories as needed
-        file_content = read_files(request.files)
-
-        # Validate that the code fits within model context limits
-        self._validate_token_limit(file_content, "Code")
+        # Use centralized file processing logic
+        continuation_id = getattr(request, "continuation_id", None)
+        file_content = self._prepare_file_content_for_prompt(request.files, continuation_id, "Code")
 
         # Build customized review instructions based on review type
         review_focus = []

tools/debug.py

@@ -9,7 +9,6 @@ from pydantic import Field
 
 from config import TEMPERATURE_ANALYTICAL
 from prompts import DEBUG_ISSUE_PROMPT
-from utils import read_files
 
 from .base import BaseTool, ToolRequest
 from .models import ToolOutput
@@ -159,7 +158,11 @@ class DebugIssueTool(BaseTool):
 
         # Add relevant files if provided
         if request.files:
-            file_content = read_files(request.files)
+            # Use centralized file processing logic
+            continuation_id = getattr(request, "continuation_id", None)
+            file_content = self._prepare_file_content_for_prompt(request.files, continuation_id, "Code")
+
+            if file_content:
                 context_parts.append(f"\n=== RELEVANT CODE ===\n{file_content}\n=== END CODE ===")
 
         full_context = "\n".join(context_parts)

tools/thinkdeep.py

@@ -9,7 +9,6 @@ from pydantic import Field
 
 from config import TEMPERATURE_CREATIVE
 from prompts import THINKDEEP_PROMPT
-from utils import read_files
 
 from .base import BaseTool, ToolRequest
 from .models import ToolOutput
@@ -142,7 +141,11 @@ class ThinkDeepTool(BaseTool):
 
         # Add reference files if provided
         if request.files:
-            file_content = read_files(request.files)
+            # Use centralized file processing logic
+            continuation_id = getattr(request, "continuation_id", None)
+            file_content = self._prepare_file_content_for_prompt(request.files, continuation_id, "Reference files")
+
+            if file_content:
                 context_parts.append(f"\n=== REFERENCE FILES ===\n{file_content}\n=== END FILES ===")
 
         full_context = "\n".join(context_parts)

utils/conversation_memory.py

@@ -46,6 +46,7 @@ USAGE EXAMPLE:
 This enables true AI-to-AI collaboration across the entire tool ecosystem.
 """
 
+import logging
 import os
 import uuid
 from datetime import datetime, timezone
@@ -53,6 +54,8 @@ from typing import Any, Optional
 
 from pydantic import BaseModel
 
+logger = logging.getLogger(__name__)
+
 # Configuration constants
 MAX_CONVERSATION_TURNS = 10  # Maximum turns allowed per conversation thread
 
@@ -278,45 +281,174 @@ def add_turn(
         return False
 
 
-def build_conversation_history(context: ThreadContext) -> str:
+def get_conversation_file_list(context: ThreadContext) -> list[str]:
     """
-    Build formatted conversation history for tool prompts
+    Get all unique files referenced across all turns in a conversation.
 
-    Creates a formatted string representation of the conversation history
-    that can be included in tool prompts to provide context. This is the
-    critical function that enables cross-tool continuation by reconstructing
-    the full conversation context.
+    This function extracts and deduplicates file references from all conversation
+    turns to enable efficient file embedding - files are read once and shared
+    across all turns rather than being embedded multiple times.
 
     Args:
         context: ThreadContext containing the complete conversation
 
     Returns:
-        str: Formatted conversation history ready for inclusion in prompts
+        list[str]: Deduplicated list of file paths referenced in the conversation
+    """
+    if not context.turns:
+        return []
+
+    # Collect all unique files from all turns, preserving order of first appearance
+    seen_files = set()
+    unique_files = []
+
+    for turn in context.turns:
+        if turn.files:
+            for file_path in turn.files:
+                if file_path not in seen_files:
+                    seen_files.add(file_path)
+                    unique_files.append(file_path)
+
+    return unique_files
+
+
+def build_conversation_history(context: ThreadContext, read_files_func=None) -> str:
+    """
+    Build formatted conversation history for tool prompts with embedded file contents.
+
+    Creates a formatted string representation of the conversation history that includes
+    full file contents from all referenced files. Files are embedded only ONCE at the
+    start, even if referenced in multiple turns, to prevent duplication and optimize
+    token usage.
+
+    Args:
+        context: ThreadContext containing the complete conversation
+
+    Returns:
+        str: Formatted conversation history with embedded files ready for inclusion in prompts
         Empty string if no conversation turns exist
 
     Format:
         - Header with thread metadata and turn count
-        - Each turn shows: role, tool used, files referenced, content
-        - Files from previous turns are explicitly listed
+        - All referenced files embedded once with full contents
+        - Each turn shows: role, tool used, which files were used, content
         - Clear delimiters for AI parsing
         - Continuation instruction at end
 
     Note:
-        This formatted history allows tools to "see" files and context
-        from previous tools, enabling true cross-tool collaboration.
+        This formatted history allows tools to "see" both conversation context AND
+        file contents from previous tools, enabling true cross-tool collaboration
+        while preventing duplicate file embeddings.
     """
     if not context.turns:
         return ""
 
+    # Get all unique files referenced in this conversation
+    all_files = get_conversation_file_list(context)
+
     history_parts = [
         "=== CONVERSATION HISTORY ===",
         f"Thread: {context.thread_id}",
         f"Tool: {context.tool_name}",  # Original tool that started the conversation
         f"Turn {len(context.turns)}/{MAX_CONVERSATION_TURNS}",
         "",
-        "Previous exchanges:",
     ]
 
+    # Embed all files referenced in this conversation once at the start
+    if all_files:
+        history_parts.extend(
+            [
+                "=== FILES REFERENCED IN THIS CONVERSATION ===",
+                "The following files have been shared and analyzed during our conversation.",
+                "Refer to these when analyzing the context and requests below:",
+                "",
+            ]
+        )
+
+        # Import required functions
+        from config import MAX_CONTEXT_TOKENS
+
+        if read_files_func is None:
+            from utils.file_utils import read_file_content
+
+            # Optimized: read files incrementally with token tracking
+            file_contents = []
+            total_tokens = 0
+            files_included = 0
+            files_truncated = 0
+
+            for file_path in all_files:
+                try:
+                    # Correctly unpack the tuple returned by read_file_content
+                    formatted_content, content_tokens = read_file_content(file_path)
+                    if formatted_content:
+                        # read_file_content already returns formatted content, use it directly
+                        # Check if adding this file would exceed the limit
+                        if total_tokens + content_tokens <= MAX_CONTEXT_TOKENS:
+                            file_contents.append(formatted_content)
+                            total_tokens += content_tokens
+                            files_included += 1
+                            logger.debug(
+                                f"📄 File embedded in conversation history: {file_path} ({content_tokens:,} tokens)"
+                            )
+                        else:
+                            files_truncated += 1
+                            logger.debug(
+                                f"📄 File truncated due to token limit: {file_path} ({content_tokens:,} tokens, would exceed {MAX_CONTEXT_TOKENS:,} limit)"
+                            )
+                            # Stop processing more files
+                            break
+                    else:
+                        logger.debug(f"📄 File skipped (empty content): {file_path}")
+                except Exception as e:
+                    # Skip files that can't be read but log the failure
+                    logger.warning(
+                        f"📄 Failed to embed file in conversation history: {file_path} - {type(e).__name__}: {e}"
+                    )
+                    continue
+
+            if file_contents:
+                files_content = "".join(file_contents)
+                if files_truncated > 0:
+                    files_content += (
+                        f"\n[NOTE: {files_truncated} additional file(s) were truncated due to token limit]\n"
+                    )
+                history_parts.append(files_content)
+                logger.debug(
+                    f"📄 Conversation history file embedding complete: {files_included} files embedded, {files_truncated} truncated, {total_tokens:,} total tokens"
+                )
+            else:
+                history_parts.append("(No accessible files found)")
+                logger.debug(
+                    f"📄 Conversation history file embedding: no accessible files found from {len(all_files)} requested"
+                )
+        else:
+            # Fallback to original read_files function for backward compatibility
+            files_content = read_files_func(all_files)
+            if files_content:
+                # Add token validation for the combined file content
+                from utils.token_utils import check_token_limit
+
+                within_limit, estimated_tokens = check_token_limit(files_content)
+                if within_limit:
+                    history_parts.append(files_content)
+                else:
+                    # Handle token limit exceeded for conversation files
+                    error_message = f"ERROR: The total size of files referenced in this conversation has exceeded the context limit and cannot be displayed.\nEstimated tokens: {estimated_tokens}, but limit is {MAX_CONTEXT_TOKENS}."
+                    history_parts.append(error_message)
+            else:
+                history_parts.append("(No accessible files found)")
+
+        history_parts.extend(
+            [
+                "",
+                "=== END REFERENCED FILES ===",
+                "",
+            ]
+        )
+
+    history_parts.append("Previous conversation turns:")
+
     for i, turn in enumerate(context.turns, 1):
         role_label = "Claude" if turn.role == "user" else "Gemini"
 
@@ -327,9 +459,10 @@ def build_conversation_history(context: ThreadContext) -> str:
         turn_header += ") ---"
         history_parts.append(turn_header)
 
-        # Add files context if present - critical for cross-tool file access
+        # Add files context if present - but just reference which files were used
+        # (the actual contents are already embedded above)
         if turn.files:
-            history_parts.append(f"📁 Files referenced: {', '.join(turn.files)}")
+            history_parts.append(f"📁 Files used in this turn: {', '.join(turn.files)}")
             history_parts.append("")  # Empty line for readability
 
         # Add the actual content
@@ -340,7 +473,7 @@ def build_conversation_history(context: ThreadContext) -> str:
             history_parts.append(f"\n[Gemini's Follow-up: {turn.follow_up_question}]")
 
     history_parts.extend(
-        ["", "=== END HISTORY ===", "", "Continue this conversation by building on the previous context."]
+        ["", "=== END CONVERSATION HISTORY ===", "", "Continue this conversation by building on the previous context."]
     )
 
     return "\n".join(history_parts)