""" CodeReviewEnv — Semantic RL Environment for OpenEnv A real-world RL environment simulating software code review. Designed as an MBRL-ready benchmark: every episode is logged as a (state, action, reward, next_state) trajectory for semantic world model training. We define a Semantic Markov Decision Process (S-MDP) as a tuple: (S, A, T, R, γ) Where: S — semantic state space: structured text + metadata (not R^n) A — structured action space: typed decisions over semantic entities T — semantic transition function: T(s, a) → s' where s,s' ∈ S T is not expressible as a closed-form equation T is learned from trajectory data R — shaped reward: R(s, a, s') → [-1, 1] with trajectory-level components γ — discount factor: 0.95 (standard) CodeReviewEnv is the first concrete instantiation of an S-MDP designed for empirical study of semantic world model learning. This is distinct from: - POMDPs: partial observability, not semantic transitions - Text games (Jericho, TWC): synthetic, not real-world tasks - LLM agent benchmarks: measure success/failure, no MDP formalism - Standard MBRL benchmarks: continuous vector state, physics transitions Research context: Current MBRL benchmarks: MuJoCo, Atari, DMControl — continuous/pixel state spaces CodeReviewEnv: structured text state, semantic transitions, knowledge-work domain Gap filled: first environment enabling Model-Based RL over semantic state spaces Trajectory dataset: use export_trajectory() to build training data for world models How this helps OpenEnv: Expands ecosystem into knowledge-work domains Proves spec works for semantic observation spaces Trajectory logging makes OpenEnv useful for MBRL, not just evaluation """ import random from typing import Dict, List, Tuple, Optional, Any from ulid import ULID from env.models import Observation, Action, Reward, State from env.trajectory_logger import TrajectoryLogger from tasks.task_easy import EasyTask from tasks.task_medium import MediumTask from tasks.task_hard import HardTask from graders.grader_easy import EasyGrader from graders.grader_medium import MediumGrader from graders.grader_hard import HardGrader class CodeReviewEnv: """ Semantic RL Environment for Software Code Review. Implements the OpenEnv interface: reset() → step() → state() → export_trajectory(). Three difficulty levels (easy, medium, hard) with deterministic grading. MBRL hook: every transition is logged as (state, action, reward, next_state) for semantic world model training. See export_trajectory() and world_model/scaffold.py for the research pipeline. Usage: env = CodeReviewEnv(task="easy", seed=42) obs = env.reset() action = Action(action_type="label_severity", severity="high") obs, reward, done, info = env.step(action) trajectory = env.export_trajectory() """ def __init__(self, task: str = "easy", seed: int = 42): """ Initialize environment. Args: task: "easy" | "medium" | "hard" seed: random seed for reproducibility """ self.task_name = task self.seed = seed # Set global seed for reproducibility random.seed(seed) # Initialize task if task == "easy": self.task = EasyTask(seed=seed) elif task == "medium": self.task = MediumTask(seed=seed) elif task == "hard": self.task = HardTask(seed=seed) else: raise ValueError(f"Unknown task: {task}. Must be 'easy', 'medium', or 'hard'.") # Initialize grader if task == "easy": self.grader = EasyGrader() elif task == "medium": self.grader = MediumGrader() else: self.grader = HardGrader() # Initialize trajectory logger self.logger = TrajectoryLogger() # Episode state self.episode_id: Optional[str] = None self.current_obs: Optional[Observation] = None self.step_count: int = 0 self.total_reward: float = 0.0 self.done: bool = False self.reviewed_prs: List[str] = [] self.pending_prs: List[str] = [] self.step_rewards: List[float] = [] self._trajectory: List[Dict[str, Any]] = [] self._severity_labels: Dict[str, str] = {} # for consistency check def reset(self) -> Observation: """ Reset environment for a new episode. Generates a fresh episode using FIXED_TEST_SUITE with the configured seed. Clears trajectory log. Returns: Initial observation for the episode. """ # Generate new episode ID self.episode_id = str(ULID()) # Reset grader state self.grader.reset() # Reset episode state self.step_count = 0 self.total_reward = 0.0 self.done = False self.reviewed_prs = [] self.step_rewards = [] self._trajectory = [] self._severity_labels = {} # Start trajectory logging self.logger.start_episode(self.episode_id, self.task_name) # Get initial observation from task self.current_obs = self.task.reset() self.pending_prs = list(self.current_obs.review_queue) return self.current_obs def step(self, action: Action) -> Tuple[Observation, Reward, bool, Dict]: """ Execute one step in the environment. Validates action, computes reward using grader, logs transition, and advances state. Invalid actions receive penalty reward but never crash the environment. Args: action: Agent's action (must pass Pydantic validation) Returns: (next_observation, reward, done, info) """ if self.done: return self._terminal_step() info: Dict[str, Any] = { "task": self.task_name, "episode_id": self.episode_id, "step": self.step_count, "parse_error": None, } prev_obs = self.current_obs # ── Route to task-specific step logic ──────────────────────── try: if self.task_name == "easy": reward, grader_info = self._step_easy(action) elif self.task_name == "medium": reward, grader_info = self._step_medium(action) else: reward, grader_info = self._step_hard(action) info.update(grader_info) except Exception as e: # Never crash on agent output — return penalty reward = Reward( value=0.01, breakdown={"step_reward": 0.01, "error_penalty": -0.01}, reason=f"Action processing error: {str(e)}", ) info["parse_error"] = str(e) # ── Apply trajectory-level reward shaping ──────────────────── reward = self._apply_reward_shaping(reward) # ── Update state ───────────────────────────────────────────── self.step_rewards.append(reward.value) self.total_reward += reward.value self.step_count += 1 # Check if episode is done if self.task_name == "hard": self.done = self.task.is_done() else: self.done = self.task.is_done(self.step_count) # Get next observation if not self.done: if self.task_name == "hard": self.current_obs = self.task.get_observation() else: self.current_obs = self.task.get_observation(self.step_count) next_obs = self.current_obs # ── Log transition ─────────────────────────────────────────── self.logger.log_transition( step=self.step_count - 1, state=prev_obs, action=action, reward=reward, next_state=next_obs, done=self.done, ) # Store in internal trajectory self._trajectory.append({ "step": self.step_count - 1, "state": prev_obs.model_dump(), "action": action.model_dump(), "reward": reward.model_dump(), "next_state": next_obs.model_dump(), "done": self.done, }) # Save trajectory file when episode ends if self.done: self.logger.save() info["trajectory_path"] = f"trajectories/{self.task_name}_{self.episode_id}.jsonl" info["reward_breakdown"] = reward.breakdown info["ground_truth"] = self._get_ground_truth_for_step() return next_obs, reward, self.done, info def _step_easy(self, action: Action) -> Tuple[Reward, Dict]: """Handle one step of the easy task.""" pr_id = self.task.get_current_pr_id(self.step_count) # Track for consistency checking if action.severity: self._severity_labels[pr_id] = action.severity self.reviewed_prs.append(pr_id) reward, grader_info = self.grader.grade(action, pr_id) return reward, grader_info def _step_medium(self, action: Action) -> Tuple[Reward, Dict]: """Handle one step of the medium task.""" queue_templates = self.task.get_queue_templates(self.step_count) ground_truth_order = self.task.get_ground_truth_order(self.step_count) pr_id = self.task.get_current_pr_id(self.step_count) self.reviewed_prs.append(pr_id) reward, grader_info = self.grader.grade(action, queue_templates, ground_truth_order) return reward, grader_info def _step_hard(self, action: Action) -> Tuple[Reward, Dict]: """Handle one step of the hard task.""" pr_id = self.task.get_current_pr_id() grader_info: Dict = {} if action.action_type == "add_comment": # Track comment in grader, give decaying feedback self.grader.add_comment(pr_id, action) self.grader.consecutive_comments += 1 advanced = self.task.process_action("add_comment") # If auto-advanced due to comment limit, grade the PR if advanced: self.grader.consecutive_comments = 0 self.reviewed_prs.append(pr_id) reward, grader_info = self.grader.grade_pr(pr_id, "request_changes") return reward, grader_info # Decaying ack: consecutive comments without decision get penalized base_ack = 0.05 spam_penalty = 0.02 * max(0, self.grader.consecutive_comments - 1) ack_value = max(0.01, base_ack - spam_penalty) reward = Reward( value=ack_value, breakdown={"step_reward": ack_value}, reason=f"Comment {self.task.comments_on_current_pr} recorded — awaiting decision.", ) return reward, grader_info elif action.action_type in ("approve", "request_changes"): self.grader.consecutive_comments = 0 self.reviewed_prs.append(pr_id) reward, grader_info = self.grader.grade_pr(pr_id, action.action_type) self.task.process_action(action.action_type) return reward, grader_info else: # Invalid action type for hard task reward = Reward( value=0.01, breakdown={"step_reward": 0.01, "invalid_action_penalty": -0.01}, reason=f"Invalid action_type '{action.action_type}' for hard task.", ) return reward, grader_info def _apply_reward_shaping(self, reward: Reward) -> Reward: """ Apply trajectory-level reward shaping bonuses and penalties. Beyond per-step reward, these shape agent behavior across the episode: efficiency_bonus: +0.1 if completing under budget consistency_penalty: -0.2 if contradicting own labels coverage_bonus: +0.15 if catching all critical bugs """ breakdown = dict(reward.breakdown) # Efficiency bonus: complete in fewer steps than budget # Only applied at episode end to avoid premature termination incentive breakdown["efficiency_bonus"] = 0.0 if self.task_name != "hard" and self.done: if self.task_name == "easy": budget = EasyTask.EPISODE_LENGTH else: budget = MediumTask.EPISODE_LENGTH if self.step_count < budget: breakdown["efficiency_bonus"] = 0.1 # Consistency penalty: labeling same PR differently in same episode # This catches agents that flip-flop on severity assessments breakdown["consistency_penalty"] = 0.0 if self.task_name == "easy": # Check if we've seen this PR before with a different label pr_id = self.task.get_current_pr_id(max(0, self.step_count)) current_severity = getattr(reward, '_current_severity', None) if pr_id in self._severity_labels and current_severity: prev = self._severity_labels[pr_id] if prev != current_severity: breakdown["consistency_penalty"] = -0.2 # Coverage bonus: catch all critical bugs # Only applied at episode end to encourage thorough review breakdown["coverage_bonus"] = 0.0 if self.task_name == "hard" and self.done: # Check if the grader's coverage component averaged >= 0.9 # across all PRs reviewed in this episode coverage_scores = [ r for r in self.step_rewards if r > 0.0 # non-zero means actual review ] if coverage_scores and sum(coverage_scores) / len(coverage_scores) >= 0.7: breakdown["coverage_bonus"] = 0.15 # Compute shaping adjustment (only the bonuses/penalties added here) shaping_adjustment = ( breakdown.get("efficiency_bonus", 0.0) + breakdown.get("consistency_penalty", 0.0) + breakdown.get("coverage_bonus", 0.0) ) # Use original reward value + shaping adjustments only # (avoid double-counting grader component scores in breakdown) total = reward.value + shaping_adjustment total = max(0.01, min(0.99, total)) return Reward( value=total, breakdown=breakdown, reason=reward.reason, ) def _terminal_step(self) -> Tuple[Observation, Reward, bool, Dict]: """Handle step() calls after episode is done.""" reward = Reward( value=0.01, breakdown={"step_reward": 0.01}, reason="Episode already done.", ) return self.current_obs, reward, True, { "task": self.task_name, "episode_id": self.episode_id, "step": self.step_count, "terminal": True, } def _get_ground_truth_for_step(self) -> Dict: """Get ground truth for the current step (for info dict).""" try: if self.task_name == "easy": return self.task.get_ground_truth(max(0, self.step_count - 1)) elif self.task_name == "medium": step = max(0, self.step_count - 1) return { "pr_id": self.task.get_current_pr_id(step), "priority_order": self.task.get_ground_truth_order(step), } else: template = self.task.get_current_template() return { "pr_id": template["pr_id"], "severity": template["ground_truth_severity"], "bug_category": template["bug_category"], "bug_lines": template["bug_lines"], } except Exception: return {} def state(self) -> State: """ Return full current state including trajectory history. The trajectory list enables in-episode analysis and is the raw material for semantic world model training. """ return State( current_pr=self.current_obs, reviewed_prs=self.reviewed_prs, pending_prs=self.pending_prs, total_reward=self.total_reward, step=self.step_count, done=self.done, trajectory=self._trajectory, ) def export_trajectory(self) -> List[Dict]: """ Return full episode as list of dicts. Format: [{step, state, action, reward, next_state, done, timestamp}] Clean JSONL-ready format for world model training dataset. Usage for MBRL research: trajectory = env.export_trajectory() # Each entry is one (s, a, r, s') transition # Encode states with sentence-transformers # Train transition model: f(z_t, a_t) → (z_{t+1}, r_t) """ return self.logger.export() def get_system_prompt(self) -> str: """ Return task-specific system prompt for LLM agents. Includes: role description, exact Action JSON schema, one example per action type. Ends with: "Respond ONLY with valid JSON matching the Action schema. No explanation." """ return self.task.get_system_prompt()