refactor: move generate_concolic_tests and normalize_code into LanguageSupport protocol

- Add generate_concolic_tests as optional protocol method (default no-op,
  Python overrides with CrossHair). Delete concolic_testing.py.
- Remove normalize_code indirection: callers use protocol directly,
  PythonSupport calls PythonNormalizer without going through deduplicate_code.

Author: KRRT7

codeflash/code_utils/deduplicate_code.py

@@ -11,20 +11,15 @@
 
 from codeflash.code_utils.normalizers import get_normalizer
 from codeflash.languages import current_language
-from codeflash.languages.base import Language
 
 
-def normalize_code(
-    code: str, remove_docstrings: bool = True, return_ast_dump: bool = False, language: str | None = None
-) -> str:
+def normalize_code(code: str, language: str | None = None) -> str:
     """Normalize code by parsing, cleaning, and normalizing variable names.
 
     Function names, class names, and parameters are preserved.
 
     Args:
         code: Source code as string
-        remove_docstrings: Whether to remove docstrings (Python only)
-        return_ast_dump: Return AST dump instead of unparsed code (Python only)
         language: Language of the code. If None, uses the current session language.
 
     Returns:
@@ -35,30 +30,10 @@ def normalize_code(
         language = current_language().value
 
     try:
-        normalizer = get_normalizer(language)
-
-        # Python has additional options
-        if language == Language.PYTHON:
-            if return_ast_dump:
-                return normalizer.normalize_for_hash(code)
-            return normalizer.normalize(code, remove_docstrings=remove_docstrings)
-
-        # For other languages, use standard normalization
-        return normalizer.normalize(code)
+        return get_normalizer(language).normalize(code)
     except ValueError:
-        # Unknown language - fall back to basic normalization
         return _basic_normalize(code)
     except Exception:
-        # Parsing error - try other languages or fall back
-        if language == Language.PYTHON:
-            # Try JavaScript as fallback
-            try:
-                js_normalizer = get_normalizer("javascript")
-                js_result = js_normalizer.normalize(code)
-                if js_result != _basic_normalize(code):
-                    return js_result
-            except Exception:
-                pass
         return _basic_normalize(code)
 
 

codeflash/languages/base.py

@@ -750,6 +750,20 @@ def parse_line_profile_results(self, line_profiler_output_file: Path) -> dict:
 
     # === Test Execution ===
 
+    def generate_concolic_tests(
+        self,
+        test_cfg: TestConfig,
+        project_root: Path,
+        function_to_optimize: FunctionToOptimize,
+        function_to_optimize_ast: Any,
+    ) -> tuple[dict, str]:
+        """Generate concolic tests for a function.
+
+        Default implementation returns empty results. Override for languages
+        that support concolic testing (e.g. Python via CrossHair).
+        """
+        return {}, ""
+
     def run_behavioral_tests(
         self,
         test_paths: Any,

codeflash/languages/python/support.py

@@ -574,21 +574,10 @@ def validate_syntax(self, source: str) -> bool:
             return False
 
     def normalize_code(self, source: str) -> str:
-        """Normalize Python code for deduplication.
-
-        Removes comments, normalizes whitespace, and replaces variable names.
-
-        Args:
-            source: Source code to normalize.
-
-        Returns:
-            Normalized source code.
-
-        """
-        from codeflash.code_utils.deduplicate_code import normalize_code
+        from codeflash.code_utils.normalizers import get_normalizer
 
         try:
-            return normalize_code(source, remove_docstrings=True, language=Language.PYTHON)
+            return get_normalizer("python").normalize(source, remove_docstrings=True)
         except Exception:
             return source
 
@@ -1077,3 +1066,111 @@ def run_line_profile_tests(
             timeout=600,
         )
         return result_file_path, results
+
+    def generate_concolic_tests(
+        self, test_cfg: Any, project_root: Path, function_to_optimize: FunctionToOptimize, function_to_optimize_ast: Any
+    ) -> tuple[dict, str]:
+        import ast
+        import importlib.util
+        import subprocess
+        import tempfile
+        import time
+
+        from codeflash.cli_cmds.console import console
+        from codeflash.code_utils.compat import SAFE_SYS_EXECUTABLE
+        from codeflash.code_utils.shell_utils import make_env_with_project_root
+        from codeflash.discovery.discover_unit_tests import discover_unit_tests
+        from codeflash.languages.python.static_analysis.concolic_utils import (
+            clean_concolic_tests,
+            is_valid_concolic_test,
+        )
+        from codeflash.languages.python.static_analysis.static_analysis import has_typed_parameters
+        from codeflash.lsp.helpers import is_LSP_enabled
+        from codeflash.telemetry.posthog_cf import ph
+        from codeflash.verification.verification_utils import TestConfig
+
+        crosshair_available = importlib.util.find_spec("crosshair") is not None
+
+        start_time = time.perf_counter()
+        function_to_concolic_tests: dict = {}
+        concolic_test_suite_code = ""
+
+        if not crosshair_available:
+            logger.debug("Skipping concolic test generation (crosshair-tool is not installed)")
+            return function_to_concolic_tests, concolic_test_suite_code
+
+        if is_LSP_enabled():
+            logger.debug("Skipping concolic test generation in LSP mode")
+            return function_to_concolic_tests, concolic_test_suite_code
+
+        if (
+            test_cfg.concolic_test_root_dir
+            and isinstance(function_to_optimize_ast, ast.FunctionDef)
+            and has_typed_parameters(function_to_optimize_ast, function_to_optimize.parents)
+        ):
+            logger.info("Generating concolic opcode coverage tests for the original code…")
+            console.rule()
+            try:
+                env = make_env_with_project_root(project_root)
+                cover_result = subprocess.run(
+                    [
+                        SAFE_SYS_EXECUTABLE,
+                        "-m",
+                        "crosshair",
+                        "cover",
+                        "--example_output_format=pytest",
+                        "--per_condition_timeout=20",
+                        ".".join(
+                            [
+                                function_to_optimize.file_path.relative_to(project_root)
+                                .with_suffix("")
+                                .as_posix()
+                                .replace("/", "."),
+                                function_to_optimize.qualified_name,
+                            ]
+                        ),
+                    ],
+                    capture_output=True,
+                    text=True,
+                    cwd=project_root,
+                    check=False,
+                    timeout=600,
+                    env=env,
+                )
+            except subprocess.TimeoutExpired:
+                logger.debug("CrossHair Cover test generation timed out")
+                return function_to_concolic_tests, concolic_test_suite_code
+
+            if cover_result.returncode == 0:
+                generated_concolic_test: str = cover_result.stdout
+                if not is_valid_concolic_test(generated_concolic_test, project_root=str(project_root)):
+                    logger.debug("CrossHair generated invalid test, skipping")
+                    console.rule()
+                    return function_to_concolic_tests, concolic_test_suite_code
+                concolic_test_suite_code = clean_concolic_tests(generated_concolic_test)
+                concolic_test_suite_dir = Path(tempfile.mkdtemp(dir=test_cfg.concolic_test_root_dir))
+                concolic_test_suite_path = concolic_test_suite_dir / "test_concolic_coverage.py"
+                concolic_test_suite_path.write_text(concolic_test_suite_code, encoding="utf8")
+
+                concolic_test_cfg = TestConfig(
+                    tests_root=concolic_test_suite_dir,
+                    tests_project_rootdir=test_cfg.concolic_test_root_dir,
+                    project_root_path=project_root,
+                )
+                function_to_concolic_tests, num_discovered_concolic_tests, _ = discover_unit_tests(concolic_test_cfg)
+                logger.info(
+                    "Created %d concolic unit test case%s ",
+                    num_discovered_concolic_tests,
+                    "s" if num_discovered_concolic_tests != 1 else "",
+                )
+                console.rule()
+                ph("cli-optimize-concolic-tests", {"num_tests": num_discovered_concolic_tests})
+
+            else:
+                logger.debug(
+                    "Error running CrossHair Cover%s", ": " + cover_result.stderr if cover_result.stderr else "."
+                )
+                console.rule()
+        end_time = time.perf_counter()
+        logger.debug("Generated concolic tests in %.2f seconds", end_time - start_time)
+        return function_to_concolic_tests, concolic_test_suite_code

codeflash/optimization/function_optimizer.py

@@ -58,7 +58,6 @@
     EffortLevel,
     get_effort_value,
 )
-from codeflash.code_utils.deduplicate_code import normalize_code
 from codeflash.code_utils.env_utils import get_pr_number
 from codeflash.code_utils.formatter import format_code, format_generated_code, sort_imports
 from codeflash.code_utils.git_utils import git_root_dir
@@ -104,7 +103,6 @@
 )
 from codeflash.result.explanation import Explanation
 from codeflash.telemetry.posthog_cf import ph
-from codeflash.verification.concolic_testing import generate_concolic_tests
 from codeflash.verification.equivalence import compare_test_results
 from codeflash.verification.parse_test_output import parse_concurrency_metrics, parse_test_results
 from codeflash.verification.verification_utils import get_test_file_path
@@ -965,7 +963,9 @@ def select_best_optimization(
         runtimes_list = []
 
         for valid_opt in eval_ctx.valid_optimizations:
-            valid_opt_normalized_code = normalize_code(valid_opt.candidate.source_code.flat.strip())
+            valid_opt_normalized_code = self.language_support.normalize_code(
+                valid_opt.candidate.source_code.flat.strip()
+            )
             new_candidate_with_shorter_code = OptimizedCandidate(
                 source_code=eval_ctx.ast_code_to_id[valid_opt_normalized_code]["shorter_source_code"],
                 optimization_id=valid_opt.candidate.optimization_id,
@@ -1072,7 +1072,7 @@ def process_single_candidate(
 
         candidate = candidate_node.candidate
 
-        normalized_code = normalize_code(candidate.source_code.flat.strip())
+        normalized_code = self.language_support.normalize_code(candidate.source_code.flat.strip())
 
         if normalized_code == normalized_original:
             logger.info(f"h3|Candidate {candidate_index}/{total_candidates}: Identical to original code, skipping.")
@@ -1284,7 +1284,7 @@ def determine_best_candidate(
             self.future_adaptive_optimizations,
         )
         candidate_index = 0
-        normalized_original = normalize_code(code_context.read_writable_code.flat.strip())
+        normalized_original = self.language_support.normalize_code(code_context.read_writable_code.flat.strip())
 
         # Process candidates using queue-based approach
         while not processor.is_done():
@@ -1679,9 +1679,9 @@ def generate_tests(
             future_concolic_tests = None
         else:
             future_concolic_tests = self.executor.submit(
-                generate_concolic_tests,
+                self.language_support.generate_concolic_tests,
                 self.test_cfg,
-                self.args,
+                self.args.project_root,
                 self.function_to_optimize,
                 self.function_to_optimize_ast,
             )