fix: Type annotation improvements across core modules

- Fixed type annotations in 18 files for mypy compatibility
- Added explicit type hints for list/dict variables
- Fixed Optional types using | None syntax
- Added type: ignore for SDK API calls with complex overloads
- Cast values where needed for proper type narrowing

Files fixed:
- empathy_llm_toolkit: core, providers, state, session_status, code_health
- empathy_llm_toolkit/security: audit_logger
- empathy_software_plugin/wizards: base, pattern_*, code_review, ai_*
- empathy_software_plugin/wizards/testing: quality_analyzer, coverage_analyzer
- empathy_software_plugin/wizards/security: exploit_analyzer, vulnerability_scanner

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

Co-Authored-By: Claude Opus 4.5 <[email protected]>

Author: 2 people

empathy_llm_toolkit/code_health.py

@@ -315,7 +315,7 @@ async def _run_check_async(
         """Run a check handler asynchronously."""
         start_time = datetime.now()
         try:
-            result = await asyncio.to_thread(handler, config)
+            result: CheckResult = await asyncio.to_thread(handler, config)
             result.duration_ms = int((datetime.now() - start_time).total_seconds() * 1000)
             return result
         except Exception as e:

empathy_llm_toolkit/core.py

@@ -298,10 +298,10 @@ async def interact(
         context = context or {}
 
         # Initialize security tracking
-        pii_detections = []
-        secrets_detections = []
+        pii_detections: list[dict] = []
+        secrets_detections: list[dict] = []
         sanitized_input = user_input
-        security_metadata = {}
+        security_metadata: dict[str, Any] = {}
 
         # Phase 3: Security Pipeline (Step 1 - PII Scrubbing)
         if self.enable_security and self.pii_scrubber:

empathy_llm_toolkit/providers.py

@@ -161,7 +161,7 @@ async def generate(
         api_kwargs.update(kwargs)
 
         # Call Anthropic API
-        response = self.client.messages.create(**api_kwargs)
+        response = self.client.messages.create(**api_kwargs)  # type: ignore[call-overload]
 
         # Extract thinking content if present
         thinking_content = None
@@ -329,21 +329,23 @@ async def generate(
         # Call OpenAI API
         response = await self.client.chat.completions.create(
             model=self.model,
-            messages=messages,
+            messages=messages,  # type: ignore[arg-type]
             temperature=temperature,
             max_tokens=max_tokens,
             **kwargs,
         )
 
         # Convert to standardized format
+        content = response.choices[0].message.content or ""
+        usage = response.usage
         return LLMResponse(
-            content=response.choices[0].message.content,
+            content=content,
             model=response.model,
-            tokens_used=response.usage.total_tokens,
+            tokens_used=usage.total_tokens if usage else 0,
             finish_reason=response.choices[0].finish_reason,
             metadata={
-                "input_tokens": response.usage.prompt_tokens,
-                "output_tokens": response.usage.completion_tokens,
+                "input_tokens": usage.prompt_tokens if usage else 0,
+                "output_tokens": usage.completion_tokens if usage else 0,
                 "provider": "openai",
             },
         )

empathy_llm_toolkit/security/audit_logger.py

@@ -660,7 +660,7 @@ def query(
             >>> # Find patterns with high PII counts (nested filter)
             >>> events = logger.query(security__pii_detected__gt=5)
         """
-        results = []
+        results: list[dict[str, Any]] = []
 
         try:
             if not self.log_path.exists():
@@ -762,23 +762,32 @@ def get_violation_summary(self, user_id: str | None = None) -> dict[str, Any]:
         """
         violations = self.query(event_type="security_violation", user_id=user_id)
 
-        summary = {
-            "total_violations": len(violations),
-            "by_type": {},
-            "by_severity": {},
-            "by_user": {},
-        }
+        by_type: dict[str, int] = {}
+        by_severity: dict[str, int] = {}
+        by_user: dict[str, int] = {}
 
         for violation in violations:
-            vtype = violation.get("violation", {}).get("type", "unknown")
-            severity = violation.get("violation", {}).get("severity", "unknown")
-            vid = violation.get("user_id", "unknown")
+            viol_data = violation.get("violation", {})
+            vtype = (
+                str(viol_data.get("type", "unknown")) if isinstance(viol_data, dict) else "unknown"
+            )
+            severity = (
+                str(viol_data.get("severity", "unknown"))
+                if isinstance(viol_data, dict)
+                else "unknown"
+            )
+            vid = str(violation.get("user_id", "unknown"))
 
-            summary["by_type"][vtype] = summary["by_type"].get(vtype, 0) + 1
-            summary["by_severity"][severity] = summary["by_severity"].get(severity, 0) + 1
-            summary["by_user"][vid] = summary["by_user"].get(vid, 0) + 1
+            by_type[vtype] = by_type.get(vtype, 0) + 1
+            by_severity[severity] = by_severity.get(severity, 0) + 1
+            by_user[vid] = by_user.get(vid, 0) + 1
 
-        return summary
+        return {
+            "total_violations": len(violations),
+            "by_type": by_type,
+            "by_severity": by_severity,
+            "by_user": by_user,
+        }
 
     def get_compliance_report(
         self, start_date: datetime | None = None, end_date: datetime | None = None

empathy_llm_toolkit/session_status.py

@@ -528,7 +528,8 @@ def _load_previous_snapshot(self) -> dict[str, Any] | None:
             if snapshot_file.stem != today:
                 try:
                     with open(snapshot_file, encoding="utf-8") as f:
-                        return json.load(f)
+                        data = json.load(f)
+                        return dict(data) if isinstance(data, dict) else None
                 except (json.JSONDecodeError, OSError):
                     continue
 

empathy_llm_toolkit/state.py

@@ -170,7 +170,7 @@ def find_matching_pattern(self, trigger_text: str) -> UserPattern | None:
 
     def get_conversation_history(
         self, max_turns: int = 10, include_metadata: bool = False
-    ) -> list[dict[str, str]]:
+    ) -> list[dict[str, Any]]:
         """
         Get recent conversation history in LLM format.
 

empathy_software_plugin/wizards/ai_context_wizard.py

@@ -321,7 +321,7 @@ def _generate_recommendations(self, issues: list[dict], predictions: list[dict])
         # Prioritize high-impact predictions
         high_impact = sorted(
             predictions,
-            key=lambda p: {"high": 3, "medium": 2, "low": 1}.get(p.get("impact"), 0),
+            key=lambda p: {"high": 3, "medium": 2, "low": 1}.get(str(p.get("impact", "")), 0),
             reverse=True,
         )
 
@@ -385,9 +385,9 @@ def _get_model_limits(self, provider: str, model: str) -> dict:
     def _estimate_context_tokens(self, call: dict, context_sources: list[dict]) -> int:
         """Estimate tokens for an AI call"""
         # Simplified: ~4 chars per token
-        base_prompt = call.get("prompt_size", 1000)
+        base_prompt = int(call.get("prompt_size", 1000))
         dynamic_context = sum(
-            source.get("estimated_size", 0)
+            int(source.get("estimated_size", 0))
             for source in context_sources
             if source.get("call_id") == call.get("id")
         )

empathy_software_plugin/wizards/ai_documentation_wizard.py

@@ -92,7 +92,7 @@ async def _analyze_ai_documentation_quality(
 
         Checks what AI needs to give good recommendations.
         """
-        issues = []
+        issues: list[dict[str, Any]] = []
 
         # Check for missing context that AI needs
         for doc_file in doc_files:

empathy_software_plugin/wizards/base_wizard.py

@@ -115,11 +115,12 @@ def get_cached_result(self, context: dict[str, Any]) -> dict[str, Any] | None:
         Returns:
             Cached result dict, or None if not cached
         """
-        if not self.has_memory():
+        if self.short_term_memory is None:
             return None
 
         key = self._cache_key(context)
-        return self.short_term_memory.retrieve(key, self._credentials)
+        result = self.short_term_memory.retrieve(key, self._credentials)
+        return dict(result) if result else None
 
     def cache_result(self, context: dict[str, Any], result: dict[str, Any]) -> bool:
         """
@@ -132,11 +133,11 @@ def cache_result(self, context: dict[str, Any], result: dict[str, Any]) -> bool:
         Returns:
             True if cached successfully
         """
-        if not self.has_memory():
+        if self.short_term_memory is None:
             return False
 
         key = self._cache_key(context)
-        return self.short_term_memory.stash(key, result, self._credentials)
+        return bool(self.short_term_memory.stash(key, result, self._credentials))
 
     async def analyze_with_cache(self, context: dict[str, Any]) -> dict[str, Any]:
         """
@@ -183,18 +184,20 @@ def share_context(self, key: str, data: Any) -> bool:
         Returns:
             True if shared successfully
         """
-        if not self.has_memory():
+        if self.short_term_memory is None:
             return False
 
         # Use global credentials for shared context (accessible to all wizards)
         global_creds = AgentCredentials(
             agent_id="wizard_shared",
             tier=AccessTier.CONTRIBUTOR,
         )
-        return self.short_term_memory.stash(
-            f"shared:{key}",
-            data,
-            global_creds,
+        return bool(
+            self.short_term_memory.stash(
+                f"shared:{key}",
+                data,
+                global_creds,
+            )
         )
 
     def get_shared_context(self, key: str, from_wizard: str | None = None) -> Any | None:
@@ -212,7 +215,7 @@ def get_shared_context(self, key: str, from_wizard: str | None = None) -> Any |
         Returns:
             The shared data, or None if not found
         """
-        if not self.has_memory():
+        if self.short_term_memory is None:
             return None
 
         # Use global shared namespace by default, or specific wizard if requested
@@ -249,7 +252,7 @@ def stage_discovered_pattern(
         Returns:
             True if staged successfully
         """
-        if not self.has_memory():
+        if self.short_term_memory is None:
             return False
 
         pattern = StagedPattern(
@@ -263,7 +266,7 @@ def stage_discovered_pattern(
             context={"wizard": self.name, "level": self.level},
         )
 
-        return self.short_term_memory.stage_pattern(pattern, self._credentials)
+        return bool(self.short_term_memory.stage_pattern(pattern, self._credentials))
 
     def send_signal(self, signal_type: str, data: dict) -> bool:
         """
@@ -278,11 +281,13 @@ def send_signal(self, signal_type: str, data: dict) -> bool:
         Returns:
             True if sent successfully
         """
-        if not self.has_memory():
+        if self.short_term_memory is None:
             return False
 
-        return self.short_term_memory.send_signal(
-            signal_type=signal_type,
-            data={"wizard": self.name, **data},
-            credentials=self._credentials,
+        return bool(
+            self.short_term_memory.send_signal(
+                signal_type=signal_type,
+                data={"wizard": self.name, **data},
+                credentials=self._credentials,
+            )
         )

empathy_software_plugin/wizards/code_review_wizard.py

@@ -283,7 +283,7 @@ def _bug_to_rule(self, bug: dict) -> AntiPatternRule | None:
 
     def _extract_safe_patterns(self, fix_code: str) -> list[str]:
         """Extract regex patterns from fix code that indicate safety."""
-        patterns = []
+        patterns: list[str] = []
         if not fix_code:
             return patterns
 
@@ -303,7 +303,7 @@ def _extract_safe_patterns(self, fix_code: str) -> list[str]:
 
     def _review_file(self, file_path: str) -> list[ReviewFinding]:
         """Review a single file for anti-patterns."""
-        findings = []
+        findings: list[ReviewFinding] = []
 
         try:
             path = Path(file_path)
@@ -465,9 +465,9 @@ def _calculate_confidence(self, findings: list[ReviewFinding]) -> float:
         avg_conf = sum(f.confidence for f in findings) / len(findings)
         return round(avg_conf, 2)
 
-    def _generate_predictions(self, findings: list[ReviewFinding]) -> list[dict]:
+    def _generate_predictions(self, findings: list[ReviewFinding]) -> list[dict[str, Any]]:
         """Generate Level 4 predictions."""
-        predictions = []
+        predictions: list[dict[str, Any]] = []
 
         if not findings:
             predictions.append(