refactor: enforce strict config loading, improve MCP argument validation, and update error handling to catch specific OS exceptions

Author: deepakdgupta1

src/knowcode/config.py

@@ -185,10 +185,9 @@ def _load_from_yaml(cls, path: Path, strict: bool = False) -> "AppConfig":
             )
         except Exception as e:
             message = f"Failed to load config from {path}: {e}"
-            if strict:
-                raise ValueError(message) from e
-            logger.warning(message)
-            return cls.default()
+            # RF-001-D004: NEVER swallow errors silently. 
+            # If a config file is present but malformed, fail explicitly.
+            raise ValueError(message) from e
 
     @classmethod
     def _validate_keys(cls, data: dict[str, Any], *, path: Path, strict: bool) -> None:

src/knowcode/indexing/monitor.py

@@ -12,8 +12,7 @@
     from watchdog.events import FileSystemEventHandler
 except ImportError:
     Observer = None  # type: ignore[assignment]
-    class FileSystemEventHandler: 
-        pass  # type: ignore[no-redef]
+    FileSystemEventHandler = object  # type: ignore[misc, assignment]
 
 if TYPE_CHECKING:
     from knowcode.indexing.background_indexer import BackgroundIndexer

src/knowcode/mcp/server.py

@@ -312,6 +312,9 @@ def handle_tool_call(self, name: str, arguments: dict[str, Any]) -> str:
         Returns:
             JSON string result.
         """
+        if not isinstance(arguments, dict):
+            return json.dumps({"error": "arguments must be a dictionary"})
+            
         try:
             from knowcode.telemetry import log_event
             log_event(
@@ -322,37 +325,47 @@ def handle_tool_call(self, name: str, arguments: dict[str, Any]) -> str:
                     "arguments": arguments,
                 }
             )
-        except Exception as e:
+        except OSError as e:
             import logging
-            logging.getLogger(__name__).warning("Ignored exception: %s", e)
+            logging.getLogger(__name__).warning("Ignored telemetry OS error: %s", e)
 
         try:
             if name == "search_codebase":
-
+                if "query" not in arguments or not isinstance(arguments["query"], str):
+                    raise ValueError("search_codebase requires 'query' as a string")
+                limit = arguments.get("limit", 10)
+                if not isinstance(limit, int):
+                    raise ValueError("'limit' must be an integer")
                 result = self.search_codebase(
                     query=arguments["query"],
-                    limit=arguments.get("limit", 10),
+                    limit=limit,
                 )
             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
                     entity_id=arguments["entity_id"],
-                    task_type=arguments.get("task_type", "general"),
-                    max_tokens=arguments.get("max_tokens", 2000),
+                    task_type=str(arguments.get("task_type", "general")),
+                    max_tokens=int(arguments.get("max_tokens", 2000)),
                 )
             elif name == "trace_calls":
+                if "entity_id" not in arguments or not isinstance(arguments["entity_id"], str):
+                    raise ValueError("trace_calls requires 'entity_id' as a string")
                 result = self.trace_calls(
                     entity_id=arguments["entity_id"],
-                    direction=arguments.get("direction", "callees"),
-                    depth=arguments.get("depth", 1),
+                    direction=str(arguments.get("direction", "callees")),
+                    depth=int(arguments.get("depth", 1)),
                 )
             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
                     query=arguments["query"],
-                    task_type=arguments.get("task_type", "auto"),
-                    max_tokens=arguments.get("max_tokens", 4000),
-                    limit_entities=arguments.get("limit_entities", 3),
-                    expand_deps=arguments.get("expand_deps", True),
-                    verbosity=arguments.get("verbosity", "minimal"),
+                    task_type=str(arguments.get("task_type", "auto")),
+                    max_tokens=int(arguments.get("max_tokens", 4000)),
+                    limit_entities=int(arguments.get("limit_entities", 3)),
+                    expand_deps=bool(arguments.get("expand_deps", True)),
+                    verbosity=str(arguments.get("verbosity", "minimal")),
                 )
             else:
                 result = {"error": f"Unknown tool: {name}"}  # type: ignore
@@ -363,6 +376,8 @@ def handle_tool_call(self, name: str, arguments: dict[str, Any]) -> str:
                 {"error": str(e), "code": e.code, "hint": e.hint},
                 separators=(",", ":"),
             )
+        except ValueError as e:
+            return json.dumps({"error": f"Validation Error: {e}"}, separators=(',', ':'))
         except Exception as e:
             return json.dumps({"error": str(e)}, separators=(',', ':'))
 

src/knowcode/service.py

@@ -309,9 +309,9 @@ def get_freshness_metadata(self) -> dict[str, Any]:
                 files = scanner.scan_all()
                 if files:
                     latest_source_change = max(os.path.getmtime(f.path) for f in files)
-        except Exception as e:
+        except OSError as e:
             import logging
-            logging.getLogger(__name__).warning("Failed to check source staleness: %s", e)
+            logging.getLogger(__name__).warning("Failed to check source staleness (OS error): %s", e)
 
         if last_store_rebuild == 0.0:
             is_stale = True

tests/unit/test_config_loading.py

@@ -105,15 +105,12 @@ def test_load_strict_rejects_invalid_root_type(tmp_path: Path) -> None:
         AppConfig.load(str(config_file), strict=True)
 
 
-def test_load_non_strict_invalid_file_falls_back_to_default(
-    tmp_path: Path, caplog: pytest.LogCaptureFixture
+def test_load_non_strict_invalid_file_raises_error(
+    tmp_path: Path
 ) -> None:
-    """Non-strict mode should warn and return defaults on invalid config."""
+    """Non-strict mode should still raise on invalid config to prevent silent failures."""
     config_file = tmp_path / "aimodels.yaml"
     _write(config_file, "config: [")
 
-    with caplog.at_level(logging.WARNING, logger="knowcode.config"):
-        cfg = AppConfig.load(str(config_file))
-
-    assert cfg.models  # default models loaded
-    assert any("Failed to load config" in rec.message for rec in caplog.records)
+    with pytest.raises(ValueError, match="Failed to load config from"):
+        AppConfig.load(str(config_file))