scripts/propose_code_mutation.py

from __future__ import annotations

import argparse
import ast
import json
import socket
import sys
import urllib.error
import urllib.request
from pathlib import Path
from typing import Any

ROOT = Path(__file__).resolve().parents[1]
if str(ROOT) not in sys.path:
    sys.path.insert(0, str(ROOT))

from life_game.code_mutation import (  # noqa: E402
    CodeMutationSpecError,
    apply_code_mutation_spec,
    apply_method_replacement_spec,
    apply_unified_diff_patch,
    code_mutation_diff_stats,
    code_mutation_spec_to_dict,
    extract_unified_diff_patch,
    method_replacement_spec_to_dict,
    parse_code_mutation_spec,
    parse_method_replacement_spec,
    parse_unified_diff_patch_spec,
    render_method_replacement_diff,
    render_code_mutation_diff,
)
from life_game.code_extensions import format_extension_briefs, select_extension_brief  # noqa: E402
from life_game.game import GAME_REGISTRY  # noqa: E402


def main() -> int:
    parser = argparse.ArgumentParser(description="Ask the LLM for a constrained game-code mutation proposal.")
    parser.add_argument("--mode", choices=sorted(GAME_REGISTRY), help="Game mode whose code should be modified.")
    parser.add_argument("--intent", help="Developer intent for the code change.")
    parser.add_argument("--base-url", default="http://127.0.0.1:8080/v1")
    parser.add_argument("--model", default="signal-garden-qwen35-code-mutation")
    parser.add_argument("--timeout-seconds", type=float, default=120.0)
    parser.add_argument("--max-tokens", type=int, default=2048)
    parser.add_argument("--common-chars", type=int, default=1200)
    parser.add_argument("--output", default="annotations/code_mutation_proposals.jsonl")
    parser.add_argument("--from-record", help="Replay a JSONL proposal record instead of querying the LLM.")
    parser.add_argument("--record-index", type=int, default=-1)
    parser.add_argument("--list-extensions", action="store_true", help="Print detailed extension briefs for --mode.")
    parser.add_argument("--extension-id", help="Use one detailed extension brief as the mutation target.")
    parser.add_argument("--ambitious", action="store_true", help="Bias the prompt toward 30+ changed-line patches.")
    parser.add_argument("--min-changed-lines", type=int, default=0, help="Reject proposals below this diff size.")
    parser.add_argument("--repair-attempts", type=int, default=0, help="Ask the LLM to repair invalid patch output.")
    parser.add_argument(
        "--proposal-format",
        choices=("auto", "line-ranges", "method-replacements", "json-patch", "unified-diff"),
        default="auto",
        help="Use JSON line ranges for small edits or whole-method replacement JSON for larger patches.",
    )
    parser.add_argument("--print-diff", action=argparse.BooleanOptionalAction, default=True)
    parser.add_argument("--apply", action="store_true", help="Write the proposed exact edits into this worktree.")
    args = parser.parse_args()

    if args.list_extensions:
        if not args.mode:
            print("Choose --mode to list detailed extension briefs for a game.", file=sys.stderr)
            return 1
        try:
            print(format_extension_briefs(args.mode, args.extension_id))
        except KeyError as exc:
            print(f"code mutation failed: {exc}", file=sys.stderr)
            return 1
        return 0

    min_changed_lines = max(0, int(args.min_changed_lines))
    if args.ambitious:
        args.max_tokens = max(args.max_tokens, 2560)
        args.common_chars = max(args.common_chars, 1200)
    proposal_format = args.proposal_format
    if proposal_format == "auto":
        proposal_format = "method-replacements" if args.ambitious or args.extension_id or min_changed_lines >= 30 else "line-ranges"
    repair_attempts = max(0, int(args.repair_attempts))
    if args.ambitious and proposal_format in {"json-patch", "method-replacements"}:
        repair_attempts = max(repair_attempts, 1)

    raw = ""
    prompt_messages: Any = []
    mode = args.mode or ""
    intent = args.intent or ""
    model = args.model
    base_url = args.base_url
    patch_spec: dict[str, Any] | None = None
    method_spec = None
    try:
        if args.from_record:
            source_record = _load_jsonl_record(Path(args.from_record), args.record_index)
            raw = str(source_record.get("raw_response") or "")
            prompt_messages = source_record.get("prompt_messages") or []
            proposal_format = str(source_record.get("proposal_format") or source_record.get("format") or proposal_format)
            mode = str(source_record.get("mode") or "")
            intent = str(source_record.get("intent") or "")
            model = str(source_record.get("model") or args.model)
            base_url = str(source_record.get("base_url") or args.base_url)
            if proposal_format == "method-replacements":
                method_spec_payload = source_record.get("method_spec")
                if method_spec_payload:
                    method_spec = parse_method_replacement_spec(json.dumps(method_spec_payload))
                else:
                    method_spec = parse_method_replacement_spec(raw)
                _validate_required_method_replacements(method_spec, mode, args.extension_id)
                preview_diff = render_method_replacement_diff(method_spec, ROOT)
                targets = apply_method_replacement_spec(method_spec, ROOT, dry_run=True)
            elif proposal_format in {"json-patch", "unified-diff"}:
                patch_spec = source_record.get("patch_spec") if isinstance(source_record.get("patch_spec"), dict) else None
                preview_diff = str(source_record.get("patch") or source_record.get("diff") or raw)
                targets = apply_unified_diff_patch(preview_diff, ROOT, dry_run=True)
            else:
                spec_payload = source_record.get("parsed_spec")
                spec = parse_code_mutation_spec(json.dumps(spec_payload))
                preview_diff = render_code_mutation_diff(spec, ROOT)
                targets = apply_code_mutation_spec(spec, ROOT, dry_run=True)
        else:
            if not args.mode or not args.intent:
                raise RuntimeError("--mode and --intent are required unless --from-record is used.")
            mode = args.mode
            intent = args.intent
            model = args.model
            base_url = args.base_url
            prompt_messages = _prompt_messages(
                mode,
                intent,
                args.common_chars,
                args.extension_id,
                args.ambitious,
                min_changed_lines,
                proposal_format,
            )
            last_error = ""
            for attempt in range(repair_attempts + 1):
                if attempt > 0:
                    prompt_messages = _repair_prompt_messages(proposal_format, raw, last_error)
                raw = _post_chat_completion(
                    args.base_url,
                    args.model,
                    prompt_messages,
                    args.timeout_seconds,
                    args.max_tokens,
                    json_response=proposal_format in {"line-ranges", "method-replacements", "json-patch"},
                )
                try:
                    if proposal_format == "method-replacements":
                        method_spec = parse_method_replacement_spec(raw)
                        _validate_required_method_replacements(method_spec, mode, args.extension_id)
                        preview_diff = render_method_replacement_diff(method_spec, ROOT)
                        targets = apply_method_replacement_spec(method_spec, ROOT, dry_run=True)
                    elif proposal_format == "json-patch":
                        patch_spec = parse_unified_diff_patch_spec(raw)
                        preview_diff = str(patch_spec["patch"])
                        targets = apply_unified_diff_patch(preview_diff, ROOT, dry_run=True)
                    elif proposal_format == "unified-diff":
                        preview_diff = extract_unified_diff_patch(raw)
                        targets = apply_unified_diff_patch(preview_diff, ROOT, dry_run=True)
                    else:
                        spec = parse_code_mutation_spec(raw)
                        preview_diff = render_code_mutation_diff(spec, ROOT)
                        targets = apply_code_mutation_spec(spec, ROOT, dry_run=True)
                except (CodeMutationSpecError, RuntimeError) as exc:
                    last_error = str(exc)
                    if attempt >= repair_attempts:
                        raise
                    continue
                break
        diff_stats = code_mutation_diff_stats(preview_diff)
        if diff_stats.changed_lines < min_changed_lines:
            raise RuntimeError(
                f"proposal changed {diff_stats.changed_lines} lines, below requested minimum {min_changed_lines}"
            )
        if args.apply:
            if proposal_format == "method-replacements":
                targets = apply_method_replacement_spec(method_spec, ROOT, dry_run=False)
            elif proposal_format in {"json-patch", "unified-diff"}:
                targets = apply_unified_diff_patch(preview_diff, ROOT, dry_run=False)
            else:
                targets = apply_code_mutation_spec(spec, ROOT, dry_run=False)
    except (CodeMutationSpecError, RuntimeError, KeyError) as exc:
        if raw or prompt_messages:
            _record_failure(args.output, mode, intent, model, base_url, proposal_format, prompt_messages, raw, str(exc))
        print(f"code mutation failed: {exc}", file=sys.stderr)
        return 1

    record = {
        "schema": "signal-garden-code-mutation-proposal/v1",
        "mode": mode,
        "intent": intent,
        "model": model,
        "base_url": base_url,
        "proposal_format": proposal_format,
        "prompt_messages": prompt_messages,
        "raw_response": raw,
        "parsed_spec": code_mutation_spec_to_dict(spec) if proposal_format == "line-ranges" else None,
        "method_spec": method_replacement_spec_to_dict(method_spec) if proposal_format == "method-replacements" else None,
        "patch_spec": patch_spec if proposal_format == "json-patch" else None,
        "patch": preview_diff if proposal_format in {"json-patch", "unified-diff"} else None,
        "diff": preview_diff,
        "diff_stats": {
            "added_lines": diff_stats.added_lines,
            "removed_lines": diff_stats.removed_lines,
            "changed_lines": diff_stats.changed_lines,
        },
        "applied": bool(args.apply),
        "targets": [str(path.relative_to(ROOT)) for path in targets],
    }
    output_path = ROOT / args.output
    output_path.parent.mkdir(parents=True, exist_ok=True)
    with output_path.open("a", encoding="utf-8") as handle:
        handle.write(json.dumps(record, sort_keys=True) + "\n")

    action = "Applied" if args.apply else "Validated"
    if proposal_format == "line-ranges":
        summary = spec.summary
    elif proposal_format == "method-replacements":
        summary = method_spec.summary
    else:
        summary = "unified diff patch"
    print(f"{action} proposal: {summary}")
    print(
        f"Diff size: +{diff_stats.added_lines}/-{diff_stats.removed_lines} "
        f"({diff_stats.changed_lines} changed lines)"
    )
    for target in targets:
        print(f"- {target.relative_to(ROOT)}")
    if args.print_diff and preview_diff:
        print()
        print(preview_diff.rstrip())
    print(f"Recorded proposal in {_display_path(output_path)}")
    if not args.apply:
        print("Rerun with --apply in this worktree to write the exact edits.")
    return 0


def _prompt_messages(
    mode: str,
    intent: str,
    common_chars: int,
    extension_id: str | None,
    ambitious: bool,
    min_changed_lines: int,
    proposal_format: str,
) -> list[dict[str, str]]:
    game = GAME_REGISTRY[mode]
    game_path = _module_path(game.__class__.__module__)
    game_source = game_path.read_text(encoding="utf-8")
    common_source = (ROOT / "life_game/games/common.py").read_text(encoding="utf-8")
    regions = _python_regions(game_source)
    selected_extension = select_extension_brief(mode, extension_id) if extension_id else None
    extension_briefs = _prompt_extension_briefs(mode, selected_extension)
    source_context = _source_context(game_path.relative_to(ROOT), game_source, selected_extension, proposal_format)
    helper_context = _helper_context(game_source, common_source, common_chars)
    method_scope_text = _method_scope_text(selected_extension, proposal_format)
    signatures_text = _method_signature_text(game_source, selected_extension.required_methods if selected_extension else ())
    required_method_text = (
        f"- Required replacement methods for this extension: {', '.join(selected_extension.required_methods)}.\n"
        if selected_extension and selected_extension.required_methods
        else ""
    )
    semantic_requirements = _semantic_requirement_text(mode, extension_id)
    target_lines = max(
        min_changed_lines,
        selected_extension.target_change_lines if selected_extension and ambitious else 30 if ambitious else 0,
    )
    ambition_rules = (
        f"- This is a material extension request: target at least {target_lines} changed diff lines.\n"
        "- Prefer 2-4 coordinated complete method replacements over a one-line tweak.\n"
        "- Do not answer with a tiny constant tweak; proposals below the target changed-line count are rejected.\n"
        "- If you cannot safely reach the target size, make the biggest coherent patch rather than padding with noise."
        if target_lines
        else "- Keep the patch compact unless the requested mechanic needs a larger change."
    )

    response_contract = _response_contract(proposal_format)
    user = f"""
Mode: {mode}
Objective: {game.objective}
Developer intent: {intent}

You may modify game code in this experimental git worktree.

{response_contract}

Rules:
- For material changes, replace complete methods from the replaceable regions list.
{required_method_text.rstrip()}
{semantic_requirements.rstrip()}
{method_scope_text.rstrip()}
- Only modify life_game/games/*.py, tests/test_*.py, or USAGE_hackathon-clean.md.
- Prefer changing only {game_path.relative_to(ROOT)}; include a test edit only when it is essential.
- Keep the game deterministic enough for tests and compatible with GameMode/GameModel.
- Do not add new third-party dependencies.
{ambition_rules}

Extension brief:
{extension_briefs}

{source_context}

Replaceable regions in {game_path.relative_to(ROOT)}:
{regions}

{signatures_text}

Imported helper excerpts:
```python
{helper_context}
```
"""
    return [
        {
            "role": "system",
            "content": _system_prompt(proposal_format),
        },
        {"role": "user", "content": user.strip()},
    ]


def _module_path(module_name: str) -> Path:
    return ROOT / Path(*module_name.split(".")).with_suffix(".py")


def _numbered_source(source: str) -> str:
    return "\n".join(f"{line_number:04d}: {line}" for line_number, line in enumerate(source.splitlines(), start=1))


def _source_context(
    path: Path,
    source: str,
    selected_extension: Any,
    proposal_format: str,
) -> str:
    required_methods = tuple(selected_extension.required_methods) if selected_extension else ()
    if proposal_format == "method-replacements" and required_methods:
        excerpt = _numbered_method_excerpt(source, required_methods)
        return f"Relevant numbered excerpts from {path}:\n```python\n{excerpt}\n```"
    return f"Numbered {path}:\n```python\n{_numbered_source(source)}\n```"


def _numbered_method_excerpt(source: str, method_names: tuple[str, ...]) -> str:
    lines = source.splitlines()
    try:
        tree = ast.parse(source)
    except SyntaxError:
        return _numbered_source(source)

    ranges: list[tuple[int, int]] = []
    first_method = len(lines) + 1
    for node in tree.body:
        if not isinstance(node, ast.ClassDef):
            continue
        method_nodes = [item for item in node.body if isinstance(item, (ast.FunctionDef, ast.AsyncFunctionDef))]
        if not method_nodes:
            continue
        first_method = min(first_method, min(item.lineno for item in method_nodes))
        for item in method_nodes:
            if item.name in method_names and item.end_lineno is not None:
                ranges.append((item.lineno, item.end_lineno))

    header_end = max(1, first_method - 1)
    merged = _merge_ranges([(1, header_end), *ranges])
    chunks: list[str] = []
    previous_end = 0
    for start, end in merged:
        if chunks and start > previous_end + 1:
            chunks.append(".... omitted unchanged methods ....")
        for line_number in range(start, end + 1):
            if 1 <= line_number <= len(lines):
                chunks.append(f"{line_number:04d}: {lines[line_number - 1]}")
        previous_end = end
    return "\n".join(chunks)


def _merge_ranges(ranges: list[tuple[int, int]]) -> list[tuple[int, int]]:
    cleaned = sorted((start, end) for start, end in ranges if start <= end)
    if not cleaned:
        return []
    merged = [cleaned[0]]
    for start, end in cleaned[1:]:
        previous_start, previous_end = merged[-1]
        if start <= previous_end + 1:
            merged[-1] = (previous_start, max(previous_end, end))
        else:
            merged.append((start, end))
    return merged


def _prompt_extension_briefs(mode: str, selected_extension: Any) -> str:
    if selected_extension is None:
        return format_extension_briefs(mode, None)
    lines = [
        f"- {selected_extension.extension_id}: {selected_extension.title}",
        f"  Goal: {selected_extension.player_facing_goal}",
        "  Implement:",
        *(f"    - {note}" for note in selected_extension.implementation_notes),
        "  Guardrails:",
        *(f"    - {risk}" for risk in selected_extension.risk_notes),
        f"  Required methods: {', '.join(selected_extension.required_methods or ('none',))}",
    ]
    return "\n".join(lines)


def _method_scope_text(selected_extension: Any, proposal_format: str) -> str:
    if proposal_format != "method-replacements" or selected_extension is None or not selected_extension.required_methods:
        return ""
    methods = ", ".join(selected_extension.required_methods)
    return (
        f"- Replace only these required methods unless the brief explicitly requires another method: {methods}.\n"
        "- Do not replace private helper methods such as names starting with `_`; call them with their existing signature.\n"
    )


def _method_signature_text(source: str, method_names: tuple[str, ...]) -> str:
    if not method_names:
        return ""
    signatures = _method_signatures(source, method_names)
    if not signatures:
        return ""
    return "Required method signatures:\n" + "\n".join(f"- {signature}" for signature in signatures)


def _method_signatures(source: str, method_names: tuple[str, ...]) -> list[str]:
    lines = source.splitlines()
    try:
        tree = ast.parse(source)
    except SyntaxError:
        return []
    wanted = set(method_names)
    signatures: list[str] = []
    for node in ast.walk(tree):
        if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) and node.name in wanted:
            line = lines[node.lineno - 1].strip()
            current = node.lineno
            while not line.rstrip().endswith(":") and len(line) < 240 and current < len(lines):
                line += " " + lines[current].strip()
                current += 1
            signatures.append(line)
    return signatures


def _helper_context(game_source: str, common_source: str, common_chars: int) -> str:
    names = _common_import_names(game_source)
    excerpt = _common_symbol_excerpt(common_source, names, max(0, int(common_chars)))
    if excerpt:
        return excerpt
    return common_source[: max(0, int(common_chars))]


def _common_import_names(source: str) -> tuple[str, ...]:
    try:
        tree = ast.parse(source)
    except SyntaxError:
        return ()
    names: list[str] = []
    for node in tree.body:
        if not isinstance(node, ast.ImportFrom) or node.module != "common":
            continue
        for alias in node.names:
            if alias.name not in names:
                names.append(alias.name)
    return tuple(names)


def _common_symbol_excerpt(source: str, names: tuple[str, ...], limit: int) -> str:
    if limit <= 0 or not names:
        return ""
    lines = source.splitlines()
    try:
        tree = ast.parse(source)
    except SyntaxError:
        return source[:limit]
    wanted = set(names)
    chunks: list[str] = []
    for node in tree.body:
        node_names: tuple[str, ...] = ()
        if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
            node_names = (node.name,)
        elif isinstance(node, ast.Assign):
            node_names = tuple(target.id for target in node.targets if isinstance(target, ast.Name))
        elif isinstance(node, ast.AnnAssign) and isinstance(node.target, ast.Name):
            node_names = (node.target.id,)
        if not wanted.intersection(node_names) or node.end_lineno is None:
            continue
        chunk = "\n".join(lines[line_number - 1] for line_number in range(node.lineno, node.end_lineno + 1))
        candidate = "\n\n".join((*chunks, chunk))
        if len(candidate) <= limit:
            chunks.append(chunk)
            continue
        if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
            signature = _function_signature_excerpt(lines, node)
            candidate = "\n\n".join((*chunks, signature))
            if len(candidate) <= limit:
                chunks.append(signature)
        if len("\n\n".join(chunks)) >= limit:
            break
    return "\n\n".join(chunks).strip()


def _function_signature_excerpt(lines: list[str], node: ast.FunctionDef | ast.AsyncFunctionDef) -> str:
    signature = lines[node.lineno - 1]
    current = node.lineno
    while not signature.rstrip().endswith(":") and current < len(lines):
        signature += "\n" + lines[current]
        current += 1
    return signature + "\n    ..."


def _response_contract(proposal_format: str) -> str:
    if proposal_format == "unified-diff":
        return """Return only a unified diff patch. Do not wrap it in JSON. Do not include prose.

The patch must:
- use `diff --git a/path b/path` file headers,
- include enough context for `git apply --check`,
- preserve Python indentation,
- include only allowed paths."""
    if proposal_format == "method-replacements":
        return """Return JSON only with this schema:
{
  "summary": "short imperative summary",
  "rationale": "why the code change helps the game",
  "replacements": [
    {
      "path": "life_game/games/example.py",
      "class_name": "ExampleGame",
      "method_name": "advance",
      "content": "    def advance(self, model, game, rule):\n        ..."
    }
  ],
  "tests": ["uv run pytest tests/test_game.py"]
}

For method replacements:
- content is the complete replacement method, including the `def method_name(...)` line.
- Preserve the exact existing signature for every replaced method.
- Unindented method content is accepted and will be indented into the class.
- Replace required methods from the prompt; avoid private helpers unless explicitly required.
- Use existing imports/helpers only; do not invent `self.some_new_method(...)` or new imports.
- Store new state in `game.data` and existing GameModel fields such as bullets, enemies, towers, boss, claimed, or pickups."""
    if proposal_format == "json-patch":
        return """Return JSON only with this schema:
{
  "summary": "short imperative summary",
  "rationale": "why the code change helps the game",
  "patch": "unified diff patch text with escaped newline characters",
  "tests": ["uv run pytest tests/test_game.py"]
}

For the patch string:
- use `diff --git a/path b/path` file headers,
- include enough context for `git apply --check`,
- preserve Python indentation,
- include only allowed paths,
- make the patch material when requested, usually by replacing complete methods."""
    return """Return JSON only with this schema:
{
  "summary": "short imperative summary",
  "rationale": "why the code change helps the game",
  "edits": [
    {
      "path": "life_game/games/example.py",
      "purpose": "why this exact edit changes behavior",
      "start_line": 10,
      "end_line": 12,
      "new_lines": ["replacement line 1", "replacement line 2"]
    }
  ],
  "tests": ["uv run pytest tests/test_game.py"]
}

For line-range JSON:
- start_line and end_line are 1-based inclusive line numbers from the numbered source.
- Use JSON string arrays for new_lines.
- Preserve indentation inside each line string.
- The apply tool joins new_lines with newline characters and a final newline."""


def _system_prompt(proposal_format: str) -> str:
    addendum = (
        " Build material game mechanics as deterministic local changes: store new state in game.data or existing "
        "GameModel masks, update player-visible messages/progress, prefer complete new()/command()/advance()/progress() "
        "method replacements, and avoid app UI, renderer, replay, dependency, or multi-file architecture work unless "
        "the prompt explicitly requests it."
    )
    if proposal_format == "unified-diff":
        return "You are an expert Python game engineer. Return only a valid unified diff patch." + addendum
    if proposal_format == "method-replacements":
        return "You are an expert Python game engineer. Return only JSON with complete Python method replacements." + addendum
    if proposal_format == "json-patch":
        return "You are an expert Python game engineer. Return only JSON with a valid unified diff patch string." + addendum
    return "You are an expert Python game engineer. Return only validated JSON for exact code edits." + addendum


def _semantic_requirement_text(mode: str, extension_id: str | None) -> str:
    if mode == "Firewall Defender" and extension_id == "spread_charge":
        return (
            "- Semantic requirements: new() must initialize charge/max_charge/charge_cooldown; "
            "command() must create a real multi-bullet spread shot when charged by directly assigning "
            "a new bullets tuple on the returned GameModel, not by calling an invented helper; "
            "advance() must increment charge over time and maintain cooldown state.\n"
        )
    if mode == "Drone Swarm" and extension_id in {"anchor_modes", "overclock_pulse"}:
        return (
            "- Semantic requirements: command() must intentionally switch formation state; "
            "advance() must preserve or update towers according to that explicit state.\n"
        )
    if mode == "Gatekeeper" and extension_id == "typed_gates":
        return (
            "- Semantic requirements: new() must initialize typed gate state; command() must create gates with "
            "a gate_type such as blocker, zapper, or slow; advance() must branch on gate_type and implement "
            "distinct blocker, zapper, and slow behavior.\n"
        )
    if mode == "Data Vampire" and extension_id == "safe_feeding_zones":
        return (
            "- Semantic requirements: new() must initialize safe_zone or poison node state; command() or advance() "
            "must create or spend safe feeding zone state; advance() must prevent enemies from occupying active safe "
            "zones and must not use model.grid as food.\n"
        )
    if mode == "Gravity Well" and extension_id == "polarity_charge_heavies":
        return (
            "- Semantic requirements: new() must initialize polarity_cooldown, charge_meter, heavy_enemy, or "
            "orbit_shield state; command() must gate polarity toggles through cooldown or charge; advance() must "
            "make heavy enemies or orbit shield behavior mechanically different from normal enemies/core damage.\n"
        )
    return ""


def _repair_prompt_messages(proposal_format: str, raw_response: str, error: str) -> list[dict[str, str]]:
    if proposal_format == "json-patch":
        contract = _response_contract("json-patch")
    elif proposal_format == "unified-diff":
        contract = _response_contract("unified-diff")
    elif proposal_format == "method-replacements":
        contract = _response_contract("method-replacements")
    else:
        contract = _response_contract("line-ranges")
    return [
        {
            "role": "system",
            "content": _system_prompt(proposal_format),
        },
        {
            "role": "user",
            "content": f"""
Your previous code mutation output failed validation.

Validation error:
{error}

Previous output:
{raw_response[:12000]}

Return a corrected response that satisfies this contract:
{contract}

Important correction rules:
- Return the full corrected proposal, not only the method or hunk related to the validation error.
- Preserve every required replacement method from the original task; if the error names missing methods, include them.
- The patch must be accepted by `git apply --check`.
- In unified diff hunks, added Python code lines must start with exactly one `+` diff marker.
- The Python code after that marker must not contain an accidental extra leading `+`.
- Do not call invented helper methods. Method-replacement responses must be complete using existing helpers and GameModel fields.
- Preserve the exact existing method signature for every replaced method.
- Preserve indentation and enough context lines.
- Do not include prose outside the requested response format.
""".strip(),
        },
    ]


def _validate_required_method_replacements(method_spec: Any, mode: str, extension_id: str | None) -> None:
    if not extension_id:
        return
    brief = select_extension_brief(mode, extension_id)
    if brief is None or not brief.required_methods:
        return
    replaced = {replacement.method_name for replacement in method_spec.replacements}
    missing = [method for method in brief.required_methods if method not in replaced]
    if missing:
        raise RuntimeError(f"method replacement proposal is missing required method(s): {', '.join(missing)}")
    _validate_extension_method_content(method_spec, mode, extension_id)


def _validate_extension_method_content(method_spec: Any, mode: str, extension_id: str) -> None:
    replacements = {replacement.method_name: replacement.content for replacement in method_spec.replacements}
    if mode == "Firewall Defender" and extension_id == "spread_charge":
        _require_content("new", replacements.get("new", ""), ("charge", "max_charge", "charge_cooldown"))
        _require_content("command", replacements.get("command", ""), ("charge", "max_charge", "bullets"))
        _require_content("advance", replacements.get("advance", ""), ("charge", "max_charge", "charge_cooldown", "min("))
    if mode == "Drone Swarm" and extension_id in {"anchor_modes", "overclock_pulse"}:
        _require_content("command", replacements.get("command", ""), ("drone", "rotation"))
        _require_content("advance", replacements.get("advance", ""), ("towers", "events"))
    if mode == "Gatekeeper" and extension_id == "typed_gates":
        _require_content("new", replacements.get("new", ""), ("gate_type",))
        _require_content("command", replacements.get("command", ""), ("gate_type", "blocker", "zapper", "slow"))
        _require_content("advance", replacements.get("advance", ""), ("gate_type", "blocker", "zapper", "slow"))
    if mode == "Data Vampire" and extension_id == "safe_feeding_zones":
        _require_content("new", replacements.get("new", ""), ("safe", "zone"))
        _require_content("advance", replacements.get("advance", ""), ("safe", "zone", "poison"))
    if mode == "Gravity Well" and extension_id == "polarity_charge_heavies":
        _require_content("new", replacements.get("new", ""), ("polarity_cooldown", "charge_meter"))
        _require_content("command", replacements.get("command", ""), ("polarity_cooldown", "charge_meter"))
        _require_content("advance", replacements.get("advance", ""), ("heavy", "orbit_shield"))


def _require_content(method_name: str, content: str, required_terms: tuple[str, ...]) -> None:
    missing = [term for term in required_terms if term not in content]
    if missing:
        raise RuntimeError(f"{method_name} replacement is missing required content: {', '.join(missing)}")


def _python_regions(source: str) -> str:
    try:
        tree = ast.parse(source)
    except SyntaxError:
        return "- unavailable: source did not parse"

    regions: list[str] = []
    for node in tree.body:
        if isinstance(node, ast.ClassDef):
            for item in node.body:
                if isinstance(item, (ast.FunctionDef, ast.AsyncFunctionDef)):
                    regions.append(
                        f"- {node.name}.{item.name}: start_line {item.lineno}, end_line {item.end_lineno}"
                    )
        elif isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
            regions.append(f"- {node.name}: start_line {node.lineno}, end_line {node.end_lineno}")
    return "\n".join(regions) if regions else "- no functions detected"


def _post_chat_completion(
    base_url: str,
    model: str,
    messages: list[dict[str, str]],
    timeout_seconds: float,
    max_tokens: int,
    json_response: bool,
) -> str:
    payload: dict[str, Any] = {
        "model": model,
        "messages": messages,
        "temperature": 0.2,
        "top_p": 0.9,
        "max_tokens": max(256, int(max_tokens)),
    }
    if json_response:
        payload["response_format"] = {"type": "json_object"}
    endpoint = f"{base_url.rstrip('/')}/chat/completions"
    request = urllib.request.Request(
        endpoint,
        data=json.dumps(payload).encode("utf-8"),
        headers={"Content-Type": "application/json"},
        method="POST",
    )
    try:
        with urllib.request.urlopen(request, timeout=max(1.0, float(timeout_seconds))) as response:
            response_body = response.read().decode("utf-8")
    except urllib.error.HTTPError as exc:
        detail = exc.read().decode("utf-8", errors="replace")
        raise RuntimeError(f"HTTP {exc.code} from {endpoint}: {detail[:240]}") from exc
    except (socket.timeout, TimeoutError) as exc:
        raise RuntimeError(f"timeout after {timeout_seconds:g}s contacting {endpoint}") from exc
    except urllib.error.URLError as exc:
        raise RuntimeError(f"could not connect to {endpoint}: {exc.reason}") from exc

    try:
        parsed = json.loads(response_body)
        return str(parsed["choices"][0]["message"].get("content") or "")
    except (KeyError, IndexError, TypeError, json.JSONDecodeError) as exc:
        raise RuntimeError("LLM server returned an invalid chat completion response.") from exc


def _display_path(path: Path) -> str:
    try:
        return str(path.relative_to(ROOT))
    except ValueError:
        return str(path)


def _load_jsonl_record(path: Path, record_index: int) -> dict[str, Any]:
    records: list[dict[str, Any]] = []
    with path.open("r", encoding="utf-8") as handle:
        for line in handle:
            stripped = line.strip()
            if not stripped:
                continue
            parsed = json.loads(stripped)
            if not isinstance(parsed, dict):
                raise RuntimeError(f"Record in {path} is not a JSON object.")
            records.append(parsed)
    if not records:
        raise RuntimeError(f"No records found in {path}.")
    try:
        return records[record_index]
    except IndexError as exc:
        raise RuntimeError(f"Record index {record_index} is outside {path}.") from exc


def _record_failure(
    output: str,
    mode: str,
    intent: str,
    model: str,
    base_url: str,
    proposal_format: str,
    prompt_messages: Any,
    raw: str,
    error: str,
) -> None:
    output_path = ROOT / output
    output_path.parent.mkdir(parents=True, exist_ok=True)
    record = {
        "schema": "signal-garden-code-mutation-proposal/v1",
        "status": "failed",
        "mode": mode,
        "intent": intent,
        "model": model,
        "base_url": base_url,
        "proposal_format": proposal_format,
        "prompt_messages": prompt_messages,
        "raw_response": raw,
        "error": error,
    }
    with output_path.open("a", encoding="utf-8") as handle:
        handle.write(json.dumps(record, sort_keys=True) + "\n")


if __name__ == "__main__":
    raise SystemExit(main())