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())