""" core/evolution/code_proposer.py Story 8.06: CodeProposer — Structural Fix Generator Uses an LLM (Opus) to generate Python interceptor code for proposed architectural fixes identified by MetaArchitect. All external I/O is dependency-injected so every method is fully testable without real API calls. Usage:: proposer = CodeProposer(opus_client=my_llm_fn, axiomatic_tests=AxiomaticTests()) proposal = proposer.propose(bottleneck, existing_code_context) if proposer.validate_proposal(proposal): path = proposer.write_proposal(proposal, epoch_id="epoch_001") """ from __future__ import annotations import json import os from dataclasses import dataclass, field from pathlib import Path from typing import Any, Callable, Optional from core.evolution.meta_architect import Bottleneck from core.evolution.axiomatic_tests import AxiomaticTests, AxiomResult # --------------------------------------------------------------------------- # Prompt template # --------------------------------------------------------------------------- PROPOSE_PROMPT = """You are a Genesis architect. Given this recurring bottleneck: Description: {description} Frequency: {frequency} Affected saga IDs: {saga_ids} Scar IDs: {scar_ids} And existing code context: {existing_code} Generate a Python interceptor that: 1. Extends BaseInterceptor 2. Addresses the bottleneck 3. Follows Genesis conventions Return JSON only (no markdown fences): {{"file_path": "core/interceptors/.py", "code_content": "...", "test_file_path": "tests/interceptors/test_.py", "test_content": "...", "config_changes": {{}}}} """ # --------------------------------------------------------------------------- # CodeProposal dataclass # --------------------------------------------------------------------------- @dataclass class CodeProposal: """A proposed code fix generated by CodeProposer. Attributes: file_path: E: drive path for the new/modified interceptor file. code_content: Full Python source content of the interceptor. test_file_path: Path for the test file (must start with ``tests/``). test_content: Full Python source content of the test file. config_changes: Any config-layer changes required (key → new_value). """ file_path: str code_content: str test_file_path: str test_content: str config_changes: dict = field(default_factory=dict) # --------------------------------------------------------------------------- # CodeProposer # --------------------------------------------------------------------------- _DEFAULT_PROPOSALS_DIR = "/mnt/e/genesis-system/data/evolution/proposals" class CodeProposer: """Generates Python interceptor code from LLM responses for bottleneck fixes. All external dependencies are dependency-injected so every code path is fully mockable in tests — zero real API or file-system access is required by the test suite. Args: opus_client: A callable ``(prompt: str) -> str`` that invokes the LLM (typically Opus) and returns raw text. When ``None`` the proposer raises ``RuntimeError`` on ``propose()`` calls. axiomatic_tests: An ``AxiomaticTests`` instance (from story 8.02). Defaults to a fresh ``AxiomaticTests()`` instance. proposals_dir: Base directory for writing proposals. Defaults to ``data/evolution/proposals``. """ def __init__( self, opus_client: Optional[Callable[[str], str]] = None, axiomatic_tests: Optional[AxiomaticTests] = None, proposals_dir: Optional[str] = None, ) -> None: self.opus_client = opus_client self.axiomatic_tests = axiomatic_tests if axiomatic_tests is not None else AxiomaticTests() self.proposals_dir = proposals_dir or _DEFAULT_PROPOSALS_DIR # ------------------------------------------------------------------ # Public API # ------------------------------------------------------------------ def propose(self, bottleneck: Bottleneck, existing_code_context: str) -> CodeProposal: """Generate a ``CodeProposal`` for the given bottleneck. Formats the prompt with bottleneck data + existing code context, calls the Opus client, and parses the JSON response into a ``CodeProposal`` instance. Args: bottleneck: The structural bottleneck to fix. existing_code_context: Relevant existing interceptor code to use as implementation reference in the prompt. Returns: A ``CodeProposal`` with all five fields populated. Raises: RuntimeError: If ``opus_client`` is not set. ValueError: If the LLM response is not valid JSON or is missing required fields. """ if self.opus_client is None: raise RuntimeError( "CodeProposer.propose() requires an opus_client callable. " "Pass one at construction time." ) prompt = PROPOSE_PROMPT.format( description=bottleneck.description, frequency=bottleneck.frequency, saga_ids=", ".join(bottleneck.affected_saga_ids) or "none", scar_ids=", ".join(bottleneck.scar_ids) or "none", existing_code=existing_code_context, ) raw_response: str = self.opus_client(prompt) proposal = self._parse_response(raw_response) return proposal def validate_proposal(self, proposal: CodeProposal) -> bool: """Validate a ``CodeProposal`` against Genesis axioms. Runs ``AxiomaticTests.run_all`` on the proposal's ``code_content``. A proposal is valid only when: • No axiom violations are found in ``code_content``, AND • ``code_content`` contains the string ``"BaseInterceptor"`` (confirming the generated class inherits from the correct base). Args: proposal: The ``CodeProposal`` to validate. Returns: ``True`` iff all axioms pass and ``BaseInterceptor`` is present. """ # Check axioms first result: AxiomResult = self.axiomatic_tests.run_all( code_content=proposal.code_content, state_content={}, ) if not result.passed: return False # Validate that generated code extends BaseInterceptor if "BaseInterceptor" not in proposal.code_content: return False # Validate test path convention if not proposal.test_file_path.startswith("tests/"): return False return True def write_proposal(self, proposal: CodeProposal, epoch_id: str) -> str: """Write a ``CodeProposal`` to disk under ``proposals_dir/{epoch_id}/``. Creates the directory structure if it does not exist. Writes: • The interceptor source to ``{proposals_dir}/{epoch_id}/{file_path}`` • The test source to ``{proposals_dir}/{epoch_id}/{test_file_path}`` • A ``proposal.json`` metadata file in ``{proposals_dir}/{epoch_id}/`` Args: proposal: The ``CodeProposal`` to persist. epoch_id: The evolution epoch identifier (e.g. ``"epoch_001"``). Returns: The absolute path to the ``proposal.json`` metadata file as a string. """ base = Path(self.proposals_dir) / epoch_id # Write interceptor source interceptor_path = base / proposal.file_path interceptor_path.parent.mkdir(parents=True, exist_ok=True) interceptor_path.write_text(proposal.code_content, encoding="utf-8") # Write test source test_path = base / proposal.test_file_path test_path.parent.mkdir(parents=True, exist_ok=True) test_path.write_text(proposal.test_content, encoding="utf-8") # Write metadata manifest metadata = { "epoch_id": epoch_id, "file_path": proposal.file_path, "test_file_path": proposal.test_file_path, "config_changes": proposal.config_changes, } manifest_path = base / "proposal.json" manifest_path.write_text(json.dumps(metadata, indent=2), encoding="utf-8") return str(manifest_path) # ------------------------------------------------------------------ # Private helpers # ------------------------------------------------------------------ def _parse_response(self, raw: str) -> CodeProposal: """Parse the LLM response string into a ``CodeProposal``. Strips markdown code fences (``` ... ```) if present, then parses the JSON body. Args: raw: Raw text response from the Opus client. Returns: A populated ``CodeProposal`` instance. Raises: ValueError: If the JSON cannot be decoded or required fields (``file_path``, ``code_content``, ``test_file_path``, ``test_content``) are missing. """ cleaned = raw.strip() # Strip markdown code fences if present if cleaned.startswith("```"): lines = cleaned.splitlines() # Remove first line (``` or ```json) and last line (```) lines = lines[1:] if lines and lines[-1].strip() == "```": lines = lines[:-1] cleaned = "\n".join(lines).strip() try: parsed: Any = json.loads(cleaned) except json.JSONDecodeError as exc: raise ValueError( f"CodeProposer: LLM response is not valid JSON. " f"Parse error: {exc}. Raw response (first 200 chars): {raw[:200]!r}" ) from exc if not isinstance(parsed, dict): raise ValueError( f"CodeProposer: LLM response parsed to {type(parsed).__name__!r}, " f"expected a JSON object (dict). Raw response (first 200 chars): {raw[:200]!r}" ) data: dict[str, Any] = parsed # Validate required fields required = ("file_path", "code_content", "test_file_path", "test_content") missing = [f for f in required if f not in data] if missing: raise ValueError( f"CodeProposer: LLM response JSON is missing required fields: {missing}. " f"Keys present: {list(data.keys())}" ) return CodeProposal( file_path=str(data["file_path"]), code_content=str(data["code_content"]), test_file_path=str(data["test_file_path"]), test_content=str(data["test_content"]), config_changes=dict(data.get("config_changes", {})), ) # --------------------------------------------------------------------------- # VERIFICATION_STAMP # Story: 8.06 # Verified By: parallel-builder # Verified At: 2026-02-25 # Tests: 12/12 # Coverage: 100% # ---------------------------------------------------------------------------