refactor: modernize indexing and error handling

- Improved Indexer: Refactored to use GitPython (git.Repo) for commit/diff operations.
- Enhanced doctor.py: Added OSError handling and logging in _path_size.
- MCP Server Improvements: Added type annotations and removed type: ignores in call_tool.
- Service Layer Cleanup: Standardized top-level imports and improved reload error handling.
- Maintenance: Updated .gitignore for SQLite temporary files.

Author: deepakdgupta1

.gitignore

@@ -213,6 +213,8 @@ docs_test/
 .knowcode/
 knowcode_index/
 knowledge.db
+knowledge.db-shm
+knowledge.db-wal
 .DS_Store
 tests/test_sample-prose
 tests/.DS_Store

src/knowcode/doctor.py

@@ -686,10 +686,16 @@ def _path_size(path: Path) -> int:
         return path.stat().st_size
 
     total = 0
+    import logging
+    logger = logging.getLogger(__name__)
+    
     for child in path.rglob("*"):
-        with contextlib.suppress(OSError):
+        try:
             if child.is_file():
                 total += child.stat().st_size
+        except OSError as e:
+            logger.warning("Skipped file size calculation for %s: %s", child, e)
+            
     return total
 
 

src/knowcode/indexing/indexer.py

@@ -87,13 +87,12 @@ def index_directory(self, root_dir: str | Path) -> int:
                 total_chunks += 1
 
         # Store current commit hash for future incremental indexing
-        import subprocess
         try:
-            res = subprocess.run(["git", "rev-parse", "HEAD"], cwd=str(root_path), capture_output=True, text=True, check=True)
-            self.manifest["last_indexed_commit"] = res.stdout.strip()
+            import git
+            repo = git.Repo(str(root_path), search_parent_directories=True)
+            self.manifest["last_indexed_commit"] = repo.head.commit.hexsha
         except Exception as e:
-            import logging
-            logging.getLogger(__name__).warning("Ignored exception: %s", e)
+            logger.warning("Failed to get git commit for manifest: %s", e)
 
         return total_chunks
 
@@ -106,16 +105,16 @@ def index_incremental(self, root_dir: str | Path) -> int:
         Returns:
             Number of new chunks added.
         """
-        import subprocess
         root_path = Path(root_dir).resolve()
 
         # Determine last indexed commit
         last_commit = self.manifest.get("last_indexed_commit")
 
         # Get current commit
         try:
-            res = subprocess.run(["git", "rev-parse", "HEAD"], cwd=str(root_path), capture_output=True, text=True, check=True)
-            current_commit = res.stdout.strip()
+            import git
+            repo = git.Repo(str(root_path), search_parent_directories=True)
+            current_commit = repo.head.commit.hexsha
         except Exception as e:
             logger.warning(f"Failed to get current git commit: {e}. Incremental indexer falling back to full index.")
             return self.index_directory(root_dir)
@@ -131,20 +130,20 @@ def index_incremental(self, root_dir: str | Path) -> int:
 
         # Get changed files
         try:
-            diff_res = subprocess.run(
-                ["git", "diff", "--name-only", last_commit, current_commit],
-                cwd=str(root_path), capture_output=True, text=True, check=True
-            )
-            untracked_res = subprocess.run(
-                ["git", "ls-files", "--others", "--exclude-standard"],
-                cwd=str(root_path), capture_output=True, text=True, check=True
-            )
+            changed_files_rel = []
+            diff = repo.commit(last_commit).diff(current_commit)
+            for d in diff:
+                if d.b_path:
+                    changed_files_rel.append(d.b_path)
+            
+            untracked = repo.untracked_files
+            changed_files_rel.extend(untracked)
         except Exception as e:
             logger.warning(f"Failed to get git diff: {e}. Falling back to full index.")
             self.manifest["last_indexed_commit"] = current_commit
             return self.index_directory(root_dir)
 
-        changed_files_rel = diff_res.stdout.splitlines() + untracked_res.stdout.splitlines()
+        changed_files_rel = list(set(changed_files_rel))
         changed_files = [str(root_path / f) for f in set(changed_files_rel) if f.strip()]
 
         if not changed_files:

src/knowcode/mcp/server.py

@@ -330,6 +330,8 @@ def handle_tool_call(self, name: str, arguments: dict[str, Any]) -> str:
             logging.getLogger(__name__).warning("Ignored telemetry OS error: %s", e)
 
         try:
+            result: dict[str, Any] | list[dict[str, Any]]
+            
             if name == "search_codebase":
                 if "query" not in arguments or not isinstance(arguments["query"], str):
                     raise ValueError("search_codebase requires 'query' as a string")
@@ -343,7 +345,7 @@ def handle_tool_call(self, name: str, arguments: dict[str, Any]) -> str:
             elif name == "get_entity_context":
                 if "entity_id" not in arguments or not isinstance(arguments["entity_id"], str):
                     raise ValueError("get_entity_context requires 'entity_id' as a string")
-                result = self.get_entity_context(  # type: ignore
+                result = self.get_entity_context(
                     entity_id=arguments["entity_id"],
                     task_type=str(arguments.get("task_type", "general")),
                     max_tokens=int(arguments.get("max_tokens", 2000)),
@@ -359,7 +361,7 @@ def handle_tool_call(self, name: str, arguments: dict[str, Any]) -> str:
             elif name == "retrieve_context_for_query":
                 if "query" not in arguments or not isinstance(arguments["query"], str):
                     raise ValueError("retrieve_context_for_query requires 'query' as a string")
-                result = self.retrieve_context_for_query(  # type: ignore
+                result = self.retrieve_context_for_query(
                     query=arguments["query"],
                     task_type=str(arguments.get("task_type", "auto")),
                     max_tokens=int(arguments.get("max_tokens", 4000)),
@@ -368,7 +370,7 @@ def handle_tool_call(self, name: str, arguments: dict[str, Any]) -> str:
                     verbosity=str(arguments.get("verbosity", "minimal")),
                 )
             else:
-                result = {"error": f"Unknown tool: {name}"}  # type: ignore
+                result = {"error": f"Unknown tool: {name}"}
 
             return json.dumps(result, separators=(',', ':'))
         except KnowCodePrerequisiteError as e:

src/knowcode/service.py

@@ -2,7 +2,10 @@
 
 from __future__ import annotations
 
+import os
 import re
+import shutil
+import logging
 from pathlib import Path
 from typing import TYPE_CHECKING, Any, Optional
 
@@ -284,8 +287,6 @@ def retrieve_context_for_query(
 
     def get_freshness_metadata(self) -> dict[str, Any]:
         """Compute freshness metadata for the knowledge store and index."""
-        import os
-
         store_file = self._store_file()
         last_store_rebuild = 0.0
         if store_file.exists():
@@ -346,7 +347,6 @@ def _build_index(self, directory: str | Path, index_path: str | Path, incrementa
 
         if not incremental:
             # Clear/initialize directory
-            import shutil
             if resolved_index_path.exists():
                 shutil.rmtree(resolved_index_path)
 
@@ -688,9 +688,9 @@ def reload(self) -> None:
         try:
             # Force reload by accessing the property
             _ = self.store
-        except FileNotFoundError:
+        except FileNotFoundError as e:
             # If the file is gone, keep _store as None
-            pass
+            logging.getLogger(__name__).warning("Knowledge store file not found during reload: %s", e)
 
     def get_entity_details(self, entity_id: str) -> Optional[dict[str, Any]]:
         """Get detailed information about an entity as a dictionary.