refactor: simplify code and improve type safety across multiple modules

The changes include:
- Simplified issue conversion in AI linting fixer
- Fixed key access in specialist manager using lambda function
- Added utility methods to BaseGenerator for templates and JSON handling
- Updated DeploymentGenerator to use consistent return types
- Improved type safety in DockerGenerator and Phase1ExtensionImplementor
- Added duration formatting helper for implementation roadmap
- Fixed various typing issues and added missing type annotations
- Applied consistent code formatting across multiple files

Author: JustAGhosT

engine/codeflow_engine/actions/ai_linting_fixer/issue_converter.py

@@ -19,21 +19,17 @@ def convert_detection_issue_to_model_issue(detection_issue: "LintingIssue") -> "
     from codeflow_engine.actions.ai_linting_fixer.models import \
         LintingIssue as ModelLintingIssue
 
-    # Build a candidate dict with superset keys
-    candidate = {
-        "file_path": detection_issue.file_path,
-        "line_number": getattr(detection_issue, "line_number", None),
-        "error_code": detection_issue.error_code,
-        "message": detection_issue.message,
-        "line_content": getattr(detection_issue, "line_content", ""),
-        # Prefer column_number; also support models that use "column"
-        "column_number": getattr(detection_issue, "column_number", None),
-        "column": getattr(detection_issue, "column_number", None),
-    }
-    # Filter by the target model's constructor fields
-    allowed = set(getattr(ModelLintingIssue, "__annotations__", {}).keys() or [])
-    payload = {k: v for k, v in candidate.items() if k in allowed and v is not None}
-    return ModelLintingIssue(**payload)
+    line_number = getattr(detection_issue, "line_number", 0)
+    column_number = getattr(detection_issue, "column_number", 0)
+    return ModelLintingIssue(
+        file_path=detection_issue.file_path,
+        line_number=line_number,
+        column_number=column_number,
+        error_code=detection_issue.error_code,
+        message=detection_issue.message,
+        line_content=getattr(detection_issue, "line_content", ""),
+        column=column_number,
+    )
 
 
 def convert_detection_issues_to_model_issues(

engine/codeflow_engine/actions/ai_linting_fixer/specialists/specialist_manager.py

@@ -57,7 +57,7 @@ def get_specialist_for_issues(self, issues: list[LintingIssue]) -> BaseSpecialis
             scores[agent_type] = score
 
         # Select the specialist with the highest score
-        best_agent_type = max(scores, key=scores.get)
+        best_agent_type = max(scores, key=lambda agent_type: scores[agent_type])
         return self.specialists[best_agent_type]
 
     def get_specialist_by_type(self, agent_type: AgentType) -> BaseSpecialist:

engine/codeflow_engine/actions/prototype_enhancement/generators/base_generator.py

@@ -4,8 +4,9 @@
 Provides the BaseGenerator class that all specialized generators inherit from.
 """
 
-from dataclasses import asdict
 from abc import ABC, abstractmethod
+from dataclasses import asdict
+import json
 from pathlib import Path
 from typing import TYPE_CHECKING, Any, TypeVar
 
@@ -80,6 +81,34 @@ def _render_template(
         """
         return self.template_manager.render(template_key, variables, variants)
 
+    def _template_exists(self, template_key: str) -> bool:
+        """Check whether a template key exists in the registry."""
+        registry = getattr(self.template_manager, "template_registry", None)
+        if registry is None:
+            return False
+        get_metadata = getattr(registry, "get_metadata", None)
+        if callable(get_metadata):
+            return get_metadata(template_key) is not None
+        return False
+
+    def _read_json_file(self, file_path: str) -> dict[str, Any] | None:
+        """Read a JSON file into a dictionary."""
+        path = Path(file_path)
+        if not path.exists():
+            return None
+        try:
+            with path.open(encoding="utf-8") as file:
+                data = json.load(file)
+        except (OSError, json.JSONDecodeError):
+            return None
+        return data if isinstance(data, dict) else None
+
+    def _write_json_file(self, file_path: str, data: dict[str, Any]) -> None:
+        """Write a dictionary to a JSON file."""
+        path = Path(file_path)
+        path.parent.mkdir(parents=True, exist_ok=True)
+        path.write_text(json.dumps(data, indent=2) + "\n", encoding="utf-8")
+
     def _get_platform_variables(self) -> dict[str, Any]:
         """Get platform-specific variables for template rendering.
 

engine/codeflow_engine/actions/prototype_enhancement/generators/deployment_generator.py

@@ -7,10 +7,16 @@
 
 import logging
 from pathlib import Path
+from typing import TYPE_CHECKING
 from typing import Any
 
 from codeflow_engine.actions.prototype_enhancement.generators.base_generator import BaseGenerator
 
+if TYPE_CHECKING:
+    from codeflow_engine.actions.prototype_enhancement.generators.template_utils import (
+        TemplateManager,
+    )
+
 
 logger = logging.getLogger(__name__)
 
@@ -20,98 +26,97 @@ class DeploymentGenerator(BaseGenerator):
     Generates deployment configuration files for various platforms.
     """
 
-    def __init__(self, template_dir: Path | None = None):
+    def __init__(
+        self,
+        template_manager: "TemplateManager",
+        **kwargs: Any,
+    ) -> None:
         """
         Initialize the deployment generator.
 
         Args:
-            template_dir: Optional directory containing custom templates
+            template_manager: Template manager used for rendering deployment templates
         """
-        super().__init__(template_dir)
+        super().__init__(template_manager, kwargs.get("platform_config"))
         self.template_name = "deployment"
 
-    def generate(
-        self,
-        platform: str,
-        output_dir: Path,
-        context: dict[str, Any] | None = None,
-    ) -> list[Path]:
+    def generate(self, output_dir: str, **kwargs: Any) -> list[str]:
         """
         Generate deployment configuration files.
 
         Args:
-            platform: Target platform (e.g., 'replit', 'vercel', 'render')
             output_dir: Directory to write generated files to
-            context: Additional context for template rendering
+            **kwargs: Additional generation options including platform and context
 
         Returns:
-            List of Path objects to generated files
+            List of generated file paths
         """
-        if context is None:
-            context = {}
+        platform = str(kwargs.get("platform", ""))
+        context = dict(kwargs.get("context") or {})
+        output_path = Path(output_dir)
 
         # Add platform-specific context
         context["platform"] = platform
 
         # Generate platform-specific deployment files
         if platform == "replit":
-            return self._generate_replit_deployment(output_dir, context)
+            return self._generate_replit_deployment(output_path, context)
         if platform == "vercel":
-            return self._generate_vercel_deployment(output_dir, context)
+            return self._generate_vercel_deployment(output_path, context)
         if platform == "render":
-            return self._generate_render_deployment(output_dir, context)
+            return self._generate_render_deployment(output_path, context)
         logger.warning(f"Unsupported deployment platform: {platform}")
         return []
 
     def _generate_replit_deployment(
         self, output_dir: Path, context: dict[str, Any]
-    ) -> list[Path]:
+    ) -> list[str]:
         """Generate Replit-specific deployment files."""
-        generated_files = []
+        generated_files: list[str] = []
 
         # Generate .replit file
         replit_config = self._render_template("replit_config.toml", context)
         if replit_config:
             replit_path = output_dir / ".replit"
-            replit_path.write_text(replit_config)
-            generated_files.append(replit_path)
+            replit_path.write_text(replit_config, encoding="utf-8")
+            generated_files.append(str(replit_path))
 
         # Generate replit.nix if needed
         if context.get("needs_custom_environment"):
             replit_nix = self._render_template("replit.nix", context)
             if replit_nix:
                 nix_path = output_dir / "replit.nix"
-                nix_path.write_text(replit_nix)
-                generated_files.append(nix_path)
+                nix_path.write_text(replit_nix, encoding="utf-8")
+                generated_files.append(str(nix_path))
 
         return generated_files
 
     def _generate_vercel_deployment(
         self, output_dir: Path, context: dict[str, Any]
-    ) -> list[Path]:
+    ) -> list[str]:
         """Generate Vercel deployment configuration."""
-        generated_files = []
+        generated_files: list[str] = []
 
         # Generate vercel.json
         vercel_config = self._render_template("vercel.json", context)
         if vercel_config:
             vercel_path = output_dir / "vercel.json"
-            vercel_path.write_text(vercel_config)
-            generated_files.append(vercel_path)
+            vercel_path.write_text(vercel_config, encoding="utf-8")
+            generated_files.append(str(vercel_path))
 
         return generated_files
 
     def _generate_render_deployment(
         self, output_dir: Path, context: dict[str, Any]
-    ) -> list[Path]:
+    ) -> list[str]:
         """Generate Render deployment configuration."""
-        generated_files = []
+        generated_files: list[str] = []
 
         # Generate render.yaml
         render_config = self._render_template("render.yaml", context)
         if render_config:
             render_path = output_dir / "render.yaml"
-            render_path.write_text(render_config)
-            generated_files.append(render_path)
+            render_path.write_text(render_config, encoding="utf-8")
+            generated_files.append(str(render_path))
 
         return generated_files

engine/codeflow_engine/actions/prototype_enhancement/generators/docker_generator.py

@@ -154,10 +154,11 @@ def _get_docker_compose_services(
         self, language: str, framework: str, database: str, environment: str
     ) -> dict[str, Any]:
         """Get services configuration for docker-compose."""
-        services = {}
+        _ = framework
+        services: dict[str, Any] = {}
 
         # Main application service
-        services["app"] = {
+        app_service: dict[str, Any] = {
             "build": ".",
             "container_name": f"app-{environment}",
             "restart": "unless-stopped",
@@ -166,26 +167,27 @@ def _get_docker_compose_services(
             "volumes": ["./:/app"] if environment == "dev" else None,
             "depends_on": [],
         }
+        services["app"] = app_service
 
         # Database service if specified
         if database:
             db_service = self._get_database_service(database, environment)
             if db_service:
                 services["db"] = db_service
-                services["app"]["depends_on"].append("db")
+                app_service["depends_on"].append("db")
 
         # Additional services based on language/framework
         if language == "typescript" and environment == "dev":
-            services["app"]["volumes"] = ["./:/app", "/app/node_modules"]
-            services["app"]["command"] = "npm run dev"
+            app_service["volumes"] = ["./:/app", "/app/node_modules"]
+            app_service["command"] = "npm run dev"
 
         return services
 
     def _get_database_service(
         self, database: str, environment: str
     ) -> dict[str, Any] | None:
         """Get database service configuration."""
-        db_configs = {
+        db_configs: dict[str, dict[str, Any]] = {
             "postgresql": {
                 "image": "postgres:13-alpine",
                 "container_name": "db",

engine/codeflow_engine/ai/implementation_roadmap/implementor.py

@@ -18,6 +18,11 @@
 logger = logging.getLogger(__name__)
 
 
+def _format_phase_duration(duration_days: int) -> str:
+    """Format phase duration for report/display payloads."""
+    return f"{duration_days} days"
+
+
 class Phase1ExtensionImplementor:
     """
     Main orchestrator for Phase 1 extension implementation
@@ -186,7 +191,7 @@ def list_available_tasks(self, phase: str | None = None) -> list[dict[str, Any]]
                 return []
             task_names = phase_obj.tasks
         else:
-            task_names = self.task_registry.get_all_task_names()
+            task_names = list(self.task_registry.get_all_tasks())
 
         tasks = []
         for task_name in task_names:
@@ -198,7 +203,7 @@ def list_available_tasks(self, phase: str | None = None) -> list[dict[str, Any]]
                         "description": task.description,
                         "category": task.category,
                         "complexity": task.complexity,
-                        "estimated_time": task.estimated_time,
+                        "estimated_time": task.estimated_hours,
                         "dependencies": task.dependencies,
                         "packages_required": task.packages_required,
                     }
@@ -220,11 +225,11 @@ def list_available_phases(self) -> list[dict[str, Any]]:
                 phases.append(
                     {
                         "name": phase_name,
-                        "description": phase.description,
+                        "description": phase.display_name,
                         "task_count": len(phase.tasks),
-                        "estimated_time": phase.estimated_time,
+                        "estimated_time": _format_phase_duration(phase.duration_days),
                         "success_criteria": phase.success_criteria,
-                        "dependencies": phase.dependencies,
+                        "dependencies": phase.depends_on,
                     }
                 )
 
@@ -252,7 +257,7 @@ def get_task_details(self, task_name: str) -> dict[str, Any] | None:
             "description": task.description,
             "category": task.category,
             "complexity": task.complexity,
-            "estimated_time": task.estimated_time,
+            "estimated_time": task.estimated_hours,
             "dependencies": task.dependencies,
             "packages_required": task.packages_required,
             "files_created": task.files_created,
@@ -356,7 +361,7 @@ def validate_environment(self) -> dict[str, Any]:
         Returns:
             Dictionary containing validation results
         """
-        validation_results = {
+        validation_results: dict[str, Any] = {
             "valid": True,
             "issues": [],
             "warnings": [],
@@ -416,14 +421,14 @@ def export_configuration(self) -> dict[str, Any]:
         return {
             "export_timestamp": datetime.now().isoformat(),
             "task_registry": {
-                "total_tasks": len(self.task_registry.get_all_task_names()),
+                "total_tasks": len(self.task_registry.get_all_tasks()),
                 "tasks_by_category": self.report_generator._analyze_task_distribution(),
             },
             "phase_configuration": {
                 phase_name: {
                     "task_count": len(phase.tasks),
-                    "estimated_time": phase.estimated_time,
-                    "dependencies": phase.dependencies,
+                    "estimated_time": _format_phase_duration(phase.duration_days),
+                    "dependencies": phase.depends_on,
                 }
                 for phase_name in ["immediate", "medium", "strategic"]
                 for phase in [self.phases.get_phase(phase_name)]