Spaces:
Configuration error
Configuration error
| # Copyright (c) Meta Platforms, Inc. and affiliates. | |
| # All rights reserved. | |
| # | |
| # This source code is licensed under the BSD-style license found in the | |
| # LICENSE file in the root directory of this source tree. | |
| """ | |
| Lint Coding Agent Environment Implementation. | |
| A simple test environment that echoes back messages sent to it. | |
| Perfect for testing HTTP server infrastructure. | |
| """ | |
| import json | |
| import os | |
| import ast | |
| import re | |
| import random | |
| import logging | |
| from uuid import uuid4 | |
| from typing import Tuple, Dict, Any, List | |
| from openenv.core.env_server.interfaces import Environment | |
| from openenv.core.env_server.types import State | |
| # Setup Logging for the Logic Engine | |
| logger = logging.getLogger("ArchitectEnvironment") | |
| # --- ABSOLUTE IMPORT FIX --- | |
| try: | |
| from models import LintCodingAgentAction, LintCodingAgentObservation | |
| except ImportError: | |
| from ..models import LintCodingAgentAction, LintCodingAgentObservation | |
| class LintCodingAgentEnvironment(Environment): | |
| """ | |
| Advanced Repository Sandbox. | |
| Handles VFS synchronization for 15 levels with real-time debug telemetry. | |
| """ | |
| SUPPORTS_CONCURRENT_SESSIONS: bool = True | |
| def __init__(self): | |
| self._state = State(episode_id=str(uuid4()), step_count=0) | |
| self.level = 1 | |
| self.failed_queue = [] | |
| self.vfs: Dict[str, str] = {} | |
| # --- DYNAMIC PATH RESOLUTION --- | |
| self.current_dir = os.path.dirname(os.path.abspath(__file__)) | |
| json_path = os.path.join(self.current_dir, "QUESTIONS.json") | |
| self.templates_base = os.path.join(self.current_dir, "templates") | |
| logger.info(f"🏗️ Initializing Environment Logic from: {self.current_dir}") | |
| try: | |
| with open(json_path, "r") as f: | |
| self.curriculum = json.load(f) | |
| self.all_levels = sorted(list(self.curriculum.keys()), key=int) | |
| logger.info(f"📚 Curriculum Loaded: {len(self.all_levels)} levels found.") | |
| except Exception as e: | |
| logger.error(f"❌ CRITICAL DATA ERROR: {e}") | |
| self.curriculum = {"1": {"lang": "Python", "task": "Fallback: Init", "template_dir": "level_1", "ans": "print"}} | |
| self.all_levels = ["1"] | |
| self.max_levels = int(self.all_levels[-1]) if self.all_levels else 1 | |
| def reset(self) -> LintCodingAgentObservation: | |
| """Full state reset for Phase 2 scoring.""" | |
| logger.info("🔄 Resetting Environment State...") | |
| self._state = State(episode_id=str(uuid4()), step_count=0) | |
| self.level = 1 | |
| self.failed_queue = [] | |
| return self._load_template_state("Environment Reset: 15-Level Sprint Initialized.") | |
| def _load_template_state(self, msg: str) -> LintCodingAgentObservation: | |
| """Loads physical files into the VFS with debug tracking.""" | |
| task_data = self.curriculum.get(str(self.level), {}) | |
| template_dir_name = task_data.get("template_dir", f"level_{self.level}") | |
| template_path = os.path.join(self.templates_base, template_dir_name) | |
| logger.info(f"📂 Loading Template for Level {self.level}: {template_path}") | |
| self.vfs = {} | |
| if os.path.exists(template_path): | |
| for root, _, files in os.walk(template_path): | |
| for f in files: | |
| full_p = os.path.join(root, f) | |
| rel_p = os.path.relpath(full_p, template_path) | |
| try: | |
| with open(full_p, 'r', encoding='utf-8') as file: | |
| self.vfs[rel_p] = file.read() | |
| logger.info(f" 📄 VFS Mounted: {rel_p}") | |
| except Exception as e: | |
| logger.error(f" ⚠️ VFS Read Error {rel_p}: {e}") | |
| else: | |
| logger.error(f"❌ CRITICAL: Level path missing: {template_path}") | |
| return self._get_observation(msg, 0.0, False) | |
| def _check_hijacking(self, action: LintCodingAgentAction) -> Tuple[bool, str, float]: | |
| """Anti-adversarial monitor to prevent agent shortcuts.""" | |
| new_code = action.code_solution or "" | |
| if len(new_code.strip()) < 10: | |
| return True, "HIJACK: Solution rejected for insufficient depth.", -2.0 | |
| code_only = re.sub(r'#.*', '', new_code).strip() | |
| if len(code_only) < 5: | |
| return True, "HIJACK: Solution contains only comments.", -1.5 | |
| return False, "", 0.0 | |
| def _is_syntactically_valid(self, code: str, lang: str) -> Tuple[bool, str]: | |
| """Verifies solution validity based on language type.""" | |
| if lang.lower() == "python": | |
| try: | |
| ast.parse(code) | |
| return True, "Syntax Valid" | |
| except SyntaxError as e: | |
| return False, f"SyntaxError: {e.msg} (Line {e.lineno})" | |
| return (True, "Non-Python/No-Verify") if len(code) > 5 else (False, "Empty Solution") | |
| def step(self, action: LintCodingAgentAction) -> LintCodingAgentObservation: | |
| self._state.step_count += 1 | |
| task_data = self.curriculum.get(str(self.level)) | |
| logger.info(f"👣 STEP {self._state.step_count} | Level: {self.level}") | |
| # 1. Monitoring | |
| is_hijacked, hijack_msg, penalty = self._check_hijacking(action) | |
| if is_hijacked: | |
| logger.warning(f"🚫 {hijack_msg}") | |
| return self._get_observation(hijack_msg, penalty, False) | |
| # 2. Execution & Verification | |
| syntax_ok, error_msg = self._is_syntactically_valid(action.code_solution, task_data["lang"]) | |
| if not syntax_ok: | |
| reward = -0.2 | |
| feedback = f"Transition Failed: {error_msg}" | |
| logger.info(f"❌ Syntax Invalid: {error_msg}") | |
| else: | |
| # Check logic against the 'ans' substring in manifest | |
| logic_ok = task_data["ans"].lower() in action.code_solution.lower() | |
| if logic_ok: | |
| reward = 1.0 | |
| feedback = f"Level {self.level} Success." | |
| logger.info(f"✅ Success! Solution matches '{task_data['ans']}'") | |
| self.level += 1 | |
| else: | |
| reward = 0.1 | |
| feedback = f"Syntax OK | Logic Missing: {task_data['task']}" | |
| logger.info(f"⚠️ Logic Mismatch. Expected logic containing: '{task_data['ans']}'") | |
| if str(self.level) not in self.failed_queue: | |
| self.failed_queue.append(str(self.level)) | |
| # 3. Handle Progression | |
| done = self.level > self.max_levels | |
| if not done and reward > 0.8: | |
| return self._load_template_state(feedback) | |
| return self._get_observation(feedback, reward, done) | |
| def _get_observation(self, feedback: str, reward: float, done: bool) -> LintCodingAgentObservation: | |
| lookup_level = min(self.level, self.max_levels) | |
| task_data = self.curriculum.get(str(lookup_level), self.curriculum.get("1")) | |
| return LintCodingAgentObservation( | |
| level=self.level, | |
| language=task_data["lang"], | |
| problem_statement=f"{feedback} | Current Task: {task_data['task']}", | |
| code_context=json.dumps(self.vfs), | |
| last_test_results=feedback, | |
| reward=reward, | |
| done=done, | |
| metadata={ | |
| "step": self._state.step_count, | |
| "vfs_files": list(self.vfs.keys()), | |
| "total_levels": self.max_levels | |
| } | |
| ) | |
| def state(self) -> State: | |
| return self._state |