$COMMIT_MESSAGE

Dockerfile

@@ -0,0 +1,22 @@
+FROM python:3.11-slim
+
+WORKDIR /app
+
+# Copy requirements first for better caching
+COPY requirements.txt .
+
+# Install dependencies
+RUN pip install --no-cache-dir -r requirements.txt
+
+# Copy application code
+COPY . .
+
+# Expose port for HuggingFace Spaces
+EXPOSE 7860
+
+# Set environment variables
+ENV PYTHONPATH=/app
+ENV PYTHONUNBUFFERED=1
+
+# Run the application (supports both inference and Space deployment)
+CMD ["python", "app.py"]

app.py

@@ -0,0 +1,195 @@
+#!/usr/bin/env python3
+
+import os
+import sys
+import gradio as gr
+import json
+from typing import Dict, Any, Optional, List
+import logging
+
+# Add current directory to Python path
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+
+from env.environment import CodeGuardianEnv
+from env.models import Action, Observation
+
+# Setup logging
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+class CodeGuardianSpace:
+    def __init__(self):
+        self.env = CodeGuardianEnv()
+        self.current_obs = None
+        self.reset_environment()
+    
+    def reset_environment(self):
+        """Reset the environment and return initial observation."""
+        try:
+            self.current_obs = self.env.reset()
+            return self.format_observation(self.current_obs)
+        except Exception as e:
+            logger.error(f"Failed to reset environment: {e}")
+            return f"Error: {str(e)}"
+    
+    def format_observation(self, obs: Observation) -> str:
+        """Format observation for display."""
+        return f"""
+**Current Task:** {obs.task_id} ({obs.difficulty})
+**Step:** {obs.step_index}/3
+
+**Code:**
+```python
+{obs.code}
+```
+
+**Intent:** {obs.intent}
+
+**Constraints:**
+{chr(10).join(f"- {c}" for c in obs.constraints)}
+"""
+    
+    def step_environment(self, action_type: str, detected_issues: str = "", risk_level: str = "low", suggested_fix: str = ""):
+        """Execute a step in the environment."""
+        try:
+            if not self.current_obs or self.env.done:
+                return "Environment is done. Please reset to continue.", "", 0.0
+            
+            # Parse detected issues
+            issues_list = []
+            if detected_issues.strip():
+                issues_list = [issue.strip() for issue in detected_issues.split(',')]
+            
+            # Create action
+            action = Action(
+                action_type=action_type,
+                detected_issues=issues_list if action_type == "detect" else None,
+                risk_level=risk_level if action_type == "risk" else None,
+                suggested_fix=suggested_fix if action_type == "fix" else None
+            )
+            
+            # Execute step
+            next_obs, reward, done, info = self.env.step(action)
+            self.current_obs = next_obs
+            
+            # Format results
+            obs_text = self.format_observation(next_obs) if next_obs else "Environment completed!"
+            feedback_text = f"Feedback: {info['feedback']}" if info['feedback'] else "No feedback"
+            current_score = self.env.state()['partial_score']
+            score_text = f"Current Score: {current_score:.2f}"
+            
+            if done:
+                obs_text += f"\n\n**Final Score: {current_score:.2f}**"
+            
+            return obs_text, feedback_text, reward
+            
+        except Exception as e:
+            logger.error(f"Failed to execute step: {e}")
+            return f"Error: {str(e)}", "", 0.0
+    
+    def get_expected_step(self):
+        """Get the expected action type for current step."""
+        if not self.current_obs or self.env.done:
+            return "reset"
+        step_types = ["detect", "risk", "fix"]
+        return step_types[self.env.state()['current_step']]
+
+# Initialize the space
+space = CodeGuardianSpace()
+
+def reset_interface():
+    """Reset the environment interface."""
+    return space.reset_environment()
+
+def step_interface(action_type, detected_issues, risk_level, suggested_fix):
+    """Step the environment interface."""
+    return space.step_environment(action_type, detected_issues, risk_level, suggested_fix)
+
+def update_ui():
+    """Update UI based on current state."""
+    expected = space.get_expected_step()
+    if expected == "reset":
+        return gr.update(visible=True), gr.update(visible=False), gr.update(visible=False), gr.update(visible=False), gr.update(visible=False)
+    else:
+        return gr.update(visible=False), gr.update(visible=True), gr.update(visible=True), gr.update(visible=True), gr.update(visible=True)
+
+# Create Gradio interface
+with gr.Blocks(title="CodeGuardian Environment") as demo:
+    gr.Markdown("# CodeGuardian Environment")
+    gr.Markdown("Test your AI agent on code review tasks: detect issues, assess risks, and suggest fixes.")
+    
+    with gr.Row():
+        reset_btn = gr.Button("Reset Environment", variant="primary")
+    
+    obs_display = gr.Markdown("Click 'Reset Environment' to start")
+    
+    with gr.Group(visible=False) as action_group:
+        gr.Markdown("## Take Action")
+        
+        with gr.Row():
+            with gr.Column():
+                action_type = gr.Radio(
+                    choices=["detect", "risk", "fix"],
+                    label="Action Type",
+                    value="detect"
+                )
+                
+                with gr.Group(visible=True) as detect_group:
+                    detected_issues = gr.Textbox(
+                        label="Detected Issues (comma-separated)",
+                        placeholder="e.g., wrong operator, security issue, performance problem"
+                    )
+                
+                with gr.Group(visible=False) as risk_group:
+                    risk_level = gr.Radio(
+                        choices=["low", "medium", "high"],
+                        label="Risk Level",
+                        value="low"
+                    )
+                
+                with gr.Group(visible=False) as fix_group:
+                    suggested_fix = gr.Textbox(
+                        label="Suggested Fix",
+                        lines=3,
+                        placeholder="Describe or provide the fix..."
+                    )
+            
+            with gr.Column():
+                step_btn = gr.Button("Execute Step", variant="primary")
+                feedback = gr.Textbox(label="Feedback", interactive=False)
+                reward = gr.Number(label="Reward", interactive=False)
+    
+    # Event handlers
+    reset_btn.click(
+        reset_interface,
+        outputs=obs_display
+    ).then(
+        update_ui,
+        outputs=[reset_btn, action_group, detect_group, risk_group, fix_group]
+    )
+    
+    def toggle_action_inputs(action):
+        """Show/hide inputs based on action type."""
+        return (
+            gr.update(visible=(action == "detect")),
+            gr.update(visible=(action == "risk")),
+            gr.update(visible=(action == "fix"))
+        )
+    
+    action_type.change(
+        toggle_action_inputs,
+        inputs=action_type,
+        outputs=[detect_group, risk_group, fix_group]
+    )
+    
+    step_btn.click(
+        step_interface,
+        inputs=[action_type, detected_issues, risk_level, suggested_fix],
+        outputs=[obs_display, feedback, reward]
+    ).then(
+        update_ui,
+        outputs=[reset_btn, action_group, detect_group, risk_group, fix_group]
+    )
+
+if __name__ == "__main__":
+    demo.launch(server_name="0.0.0.0", server_port=7860)

env/environment.py

@@ -1,6 +1,6 @@
 from env.models import Observation, Action
 from env.tasks import TASKS
-from env.grader import grade_action
+from env.grader import grade_detect, grade_risk, grade_fix
 
 class CodeGuardianEnv:
     def __init__(self):
@@ -22,51 +22,51 @@ class CodeGuardianEnv:
         return Observation(
             code=task["code"],
             intent=task["intent"],
-            constraints=task["constraints"]
+            constraints=task["constraints"],
+            task_id=task["task_id"],
+            difficulty=task["difficulty"],
+            step_index=self.current_step
         )
 
     def step(self, action: Action):
+        # Early guard to prevent IndexError when environment is done
+        if self.done:
+            return None, 0.0, True, {
+                "feedback": "Environment is complete. Reset to continue.",
+                "partial_scores": {}
+            }
+        
         task = self.tasks[self.current_task_index]
-        expected = task["expected"]
-        reward = 0
-        feedback = []
+        weights = [0.3, 0.2, 0.5]  # detect, risk, fix weights
+        step_types = ['detect', 'risk', 'fix']
         
         # Check if action order is valid
-        if (self.current_step == 0 and action.action_type != "detect") or \
-           (self.current_step == 1 and action.action_type != "risk") or \
-           (self.current_step == 2 and action.action_type != "fix"):
-            reward = -0.5
-            feedback.append(f"Invalid action order: expected {['detect', 'risk', 'fix'][self.current_step]}, got {action.action_type}")
+        if action.action_type != step_types[self.current_step]:
+            reward = -0.3
+            feedback = f"Invalid action order: expected {step_types[self.current_step]}, got {action.action_type}"
+            partial_scores = {}
         else:
-            # Step 0: detect issues
-            if self.current_step == 0:
-                if action.detected_issues and any(expected["issue"] in issue.lower() for issue in action.detected_issues):
-                    reward = 0.3
-                else:
-                    reward = -0.2
-                    feedback.append("Issue not correctly identified")
-                self.current_step += 1
+            # Use deterministic graders
+            original_code = task["code"]
+            
+            if action.action_type == "detect":
+                score = grade_detect(action.detected_issues or [], task["expected_issues"])
+            elif action.action_type == "risk":
+                score = grade_risk(action.risk_level or "", task["expected_risk"])
+            elif action.action_type == "fix":
+                score = grade_fix(action.suggested_fix or "", task["canonical_fix"], original_code)
+            else:
+                score = 0.0
             
-            # Step 1: assign risk level
-            elif self.current_step == 1:
-                if action.risk_level and action.risk_level.lower() == expected["risk"]:
-                    reward = 0.2
-                else:
-                    reward = -0.2
-                    feedback.append("Incorrect risk level")
-                self.current_step += 1
+            # Weighted reward
+            reward = score * weights[self.current_step]
+            feedback = str(score)
+            partial_scores = {action.action_type: score}
             
-            # Step 2: suggest fix
-            elif self.current_step == 2:
-                if action.suggested_fix and expected["fix_contains"] in action.suggested_fix.lower():
-                    reward = 0.5
-                else:
-                    reward = -0.3
-                    feedback.append("Fix is incorrect or missing")
-                
-                # Move to next task and reset current_step
+            # Advance step_index % 3, task if done
+            self.current_step = (self.current_step + 1) % 3
+            if self.current_step == 0:  # Completed full cycle
                 self.current_task_index += 1
-                self.current_step = 0
                 if self.current_task_index >= len(self.tasks):
                     self.done = True
         
@@ -76,7 +76,7 @@ class CodeGuardianEnv:
         
         info = {
             "feedback": feedback,
-            "total_score": self.partial_score
+            "partial_scores": partial_scores
         }
         
         return next_obs, reward, self.done, info

env/grader.py

@@ -1,23 +1,184 @@
-def grade_action(action, expected):
-    score = 0
-    feedback = []
+from typing import List, Dict, Any, Optional
+from env.models import Action
+
+def grade_detect(issues_pred: List[str], expected: List[str]) -> float:
+    """
+    F1 score for issue detection (normalize lowercase, token overlap).
+    """
+    if not issues_pred or not expected:
+        return 0.0
+    
+    # Normalize to lowercase and extract tokens
+    pred_tokens = set(' '.join(issues_pred).lower().split())
+    exp_tokens = set(' '.join(expected).lower().split())
+    
+    # Calculate precision and recall
+    if not pred_tokens:
+        return 0.0
+    
+    intersection = pred_tokens & exp_tokens
+    precision = len(intersection) / len(pred_tokens)
+    recall = len(intersection) / len(exp_tokens) if exp_tokens else 0
+    
+    # F1 score
+    if precision + recall == 0:
+        return 0.0
+    return 2 * (precision * recall) / (precision + recall)
+
+def grade_risk(pred: str, expected: str) -> float:
+    """
+    1.0 exact, 0.5 adjacent, 0.0 mismatch (ordinal {'low':0,'medium':1,'high':2}).
+    """
+    if not pred or not expected:
+        return 0.0
+    
+    risk_order = {'low': 0, 'medium': 1, 'high': 2}
+    
+    try:
+        pred_val = risk_order[pred.lower()]
+        exp_val = risk_order[expected.lower()]
+        
+        if pred_val == exp_val:
+            return 1.0
+        elif abs(pred_val - exp_val) == 1:
+            return 0.5
+        else:
+            return 0.0
+    except KeyError:
+        return 0.0
 
-    # Issue detection
-    if any(expected["issue"] in issue.lower() for issue in action.detected_issues):
-        score += 0.3
+def levenshtein_distance(s1: str, s2: str) -> int:
+    """Calculate Levenshtein distance between two strings."""
+    if len(s1) < len(s2):
+        return levenshtein_distance(s2, s1)
+    
+    if len(s2) == 0:
+        return len(s1)
+    
+    previous_row = list(range(len(s2) + 1))
+    for i, c1 in enumerate(s1):
+        current_row = [i + 1]
+        for j, c2 in enumerate(s2):
+            insertions = previous_row[j + 1] + 1
+            deletions = current_row[j] + 1
+            substitutions = previous_row[j] + (c1 != c2)
+            current_row.append(min(insertions, deletions, substitutions))
+        previous_row = current_row
+    
+    return previous_row[-1]
+
+def grade_fix(fix: str, canonical: str, orig: str) -> float:
+    """
+    Deterministic fix grading with bounded components and normalized output.
+    Returns score in [0.0, 1.0].
+    
+    Components:
+    - syntax_ok: 1.0 if syntax is valid, 0.0 otherwise
+    - lev_norm: Normalized Levenshtein distance (0.0 to 1.0)
+    - bad_pattern_absent: 0.1 if no bad patterns, 0.0 otherwise
+    
+    Final score: syntax_ok - lev_norm + bad_pattern_absent, clamped to [0.0, 1.0]
+    """
+    if not fix:
+        return 0.0
+    
+    # Syntax check (basic) - bounded component
+    syntax_ok = 1.0
+    try:
+        # Check for balanced parentheses and brackets
+        brackets = {'(': ')', '[': ']', '{': '}'}
+        stack = []
+        for char in fix:
+            if char in brackets:
+                stack.append(brackets[char])
+            elif char in brackets.values():
+                if not stack or stack.pop() != char:
+                    syntax_ok = 0.0
+                    break
+        if stack:
+            syntax_ok = 0.0
+    except:
+        syntax_ok = 0.0
+    
+    # Normalized Levenshtein distance - bounded component [0.0, 1.0]
+    max_len = max(len(fix), len(canonical))
+    if max_len == 0:
+        lev_norm = 0.0
     else:
-        feedback.append("Issue not correctly identified")
+        distance = levenshtein_distance(fix.lower(), canonical.lower())
+        lev_norm = min(distance / max_len, 1.0)  # Ensure bounded to [0.0, 1.0]
+    
+    # Check for bad patterns (simple heuristic) - bounded component [0.0, 0.1]
+    bad_patterns = ['pass', 'TODO', 'FIXME', 'XXX', 'print(']
+    bad_pattern_absent = 0.1
+    for pattern in bad_patterns:
+        if pattern in fix:
+            bad_pattern_absent = 0.0
+            break
+    
+    # Calculate raw score and clamp to [0.0, 1.0]
+    raw_score = syntax_ok - lev_norm + bad_pattern_absent
+    return max(0.0, min(1.0, raw_score))
 
-    # Risk level
-    if action.risk_level.lower() == expected["risk"]:
-        score += 0.2
+def grade_step(action: Action, expected: Dict[str, Any], original_code: str = "") -> Dict[str, Any]:
+    """
+    Grade a single step with deterministic partial-credit scoring.
+    Returns score (0-1) and feedback.
+    """
+    if action.action_type == "detect":
+        score = grade_detect(action.detected_issues, expected["issues"])
+        feedback = f"Issue detection score: {score:.2f}"
+        if score < 0.5:
+            feedback += f" (expected issues: {expected['issues']})"
+        
+    elif action.action_type == "risk":
+        score = grade_risk(action.risk_level, expected["risk"])
+        feedback = f"Risk assessment score: {score:.2f}"
+        if score < 1.0:
+            feedback += f" (expected risk: {expected['risk']})"
+        
+    elif action.action_type == "fix":
+        score = grade_fix(action.suggested_fix, expected["fix_contains"], original_code)
+        feedback = f"Fix suggestion score: {score:.2f}"
+        if score < 0.7:
+            feedback += f" (expected to contain: {expected['fix_contains']})"
+        
     else:
-        feedback.append("Incorrect risk level")
+        score = 0.0
+        feedback = f"Unknown action type: {action.action_type}"
+    
+    return {
+        "score": score,
+        "feedback": feedback
+    }
 
-    # Fix correctness
-    if expected["fix_contains"] in action.suggested_fix.lower():
-        score += 0.5
+def map_score_to_reward(score: float, action_type: str) -> float:
+    """
+    Map grader score (0-1) to reward values with appropriate weighting.
+    """
+    # Base weights for each action type
+    weights = {
+        "detect": 0.3,
+        "risk": 0.2,
+        "fix": 0.5
+    }
+    
+    base_weight = weights.get(action_type, 0.1)
+    
+    # Apply score to weight, with small penalty for very low scores
+    if score >= 0.8:
+        return base_weight
+    elif score >= 0.6:
+        return base_weight * 0.8
+    elif score >= 0.4:
+        return base_weight * 0.6
+    elif score >= 0.2:
+        return base_weight * 0.4
     else:
-        feedback.append("Fix is incorrect or missing")
+        return -base_weight * 0.3  # Small penalty for very poor performance
 
-    return min(score, 1.0), "; ".join(feedback)
+# Legacy function for backward compatibility
+def grade_action(action, expected):
+    """Legacy grading function - use grade_step instead."""
+    result = grade_step(action, expected)
+    return result["score"], result["feedback"]

env/tasks.py

@@ -1,42 +1,60 @@
 TASKS = [
     {
-        "id": "easy",
-        "code": "def add(a, b): return a - b",
-        "intent": "Return sum of two numbers",
-        "constraints": ["correct logic"],
-        "expected": {
-            "issue": "wrong operator",
-            "risk": "low",
-            "fix_contains": "+"
-        }
+        "task_id": "easy_off_by_one",
+        "difficulty": "easy",
+        "code": "def sum_list(arr): return sum(arr[:-1]) # missing last",
+        "intent": "Sum all elements",
+        "constraints": ["O(n) time"],
+        "expected_issues": ["off-by-one error"],
+        "expected_risk": "low",
+        "canonical_fix": "return sum(arr)"
     },
     {
-        "id": "medium",
-        "code": """def login(user, password):
-    if user == "admin" and password == "1234":
-        return True""",
-        "intent": "Authenticate user securely",
-        "constraints": ["no hardcoding", "security"],
-        "expected": {
-            "issue": "hardcoded credentials",
-            "risk": "high",
-            "fix_contains": "hashed"
-        }
+        "task_id": "medium_sql_injection", 
+        "difficulty": "medium",
+        "code": "def get_user(user_id): query = f'SELECT * FROM users WHERE id={user_id}'; cursor.execute(query); return cursor.fetchone()",
+        "intent": "Fetch user by ID safely",
+        "constraints": ["SQL injection prevention", "parameterized queries"],
+        "expected_issues": ["SQL injection"],
+        "expected_risk": "high",
+        "canonical_fix": "cursor.execute('SELECT * FROM users WHERE id=%s', (user_id,))"
     },
     {
-        "id": "hard",
-        "code": """def process(data):
-    result = []
-    for i in range(len(data)):
-        if data[i] % 2 == 0:
-            result.append(data[i])
-    return result""",
-        "intent": "Return even numbers efficiently",
-        "constraints": ["efficient", "clean code"],
-        "expected": {
-            "issue": "inefficient loop",
-            "risk": "low",
-            "fix_contains": "list comprehension"
-        }
+        "task_id": "hard_async_race",
+        "difficulty": "hard", 
+        "code": """import asyncio
+import requests
+
+async def fetch(url):
+    data = requests.get(url).json()  # no await
+    return data
+
+class DataCache:
+    def __init__(self):
+        self.cache = {}  # mutable shared state
+    
+    async def update(self, key, value):
+        self.cache[key] = value  # race condition
+        await asyncio.sleep(0.1)
+        return self.cache[key]""",
+        "intent": "Fetch data asynchronously with thread-safe caching",
+        "constraints": ["async safety", "no blocking calls", "thread safety"],
+        "expected_issues": ["missing await", "race condition", "blocking I/O"],
+        "expected_risk": "high",
+        "canonical_fix": """async def fetch(url):
+    async with aiohttp.ClientSession() as session:
+        async with session.get(url) as response:
+            return await response.json()
+
+class DataCache:
+    def __init__(self):
+        self.cache = {}
+        self._lock = asyncio.Lock()
+    
+    async def update(self, key, value):
+        async with self._lock:
+            self.cache[key] = value
+            await asyncio.sleep(0.1)
+            return self.cache[key]"""
     }
 ]

inference.py

@@ -141,8 +141,7 @@ def main():
         total_reward = 0
         step_count = 0
         
-        logger.info(f"Starting inference with model: {model_name}")
-        logger.info(f"Initial task: {obs.intent}")
+        logger.info(f"[START] model={model_name} task_id={obs.task_id} difficulty={obs.difficulty} intent={obs.intent}")
         
         # Main loop
         while not env.done:
@@ -151,7 +150,7 @@ def main():
             step_type = step_types[current_step]
             
             step_count += 1
-            logger.info(f"\n--- Step {step_count}: {step_type.upper()} ---")
+            logger.info(f"[STEP] step_number={step_count} step_type={step_type} task_id={obs.task_id}")
             
             # Create prompt
             prompt = create_prompt(obs, step_type)
@@ -176,11 +175,7 @@ def main():
             total_reward += reward
             
             # Log results
-            logger.info(f"Action type: {action.action_type}")
-            logger.info(f"Reward: {reward:+.2f}")
-            logger.info(f"Feedback: {info['feedback']}")
-            logger.info(f"Running total: {total_reward:+.2f}")
-            logger.info(f"State: {env.state()}")
+            logger.info(f"action_type={action.action_type} reward={reward:+.2f} feedback={info['feedback']} total_reward={total_reward:+.2f} current_step={env.state()['current_step']} current_task={env.state()['current_task']}")
             
             # Update observation for next step
             obs = next_obs
@@ -191,10 +186,7 @@ def main():
                 break
         
         # Final output
-        logger.info(f"\n=== Inference Complete ===")
-        logger.info(f"Final score: {total_reward:.2f}")
-        logger.info(f"Total steps: {step_count}")
-        logger.info(f"Final state: {env.state()}")
+        logger.info(f"[END] final_score={total_reward:.2f} total_steps={step_count} final_state={env.state()}")
         
     except Exception as e:
         logger.error(f"Inference failed: {e}")

openenv.yaml

@@ -0,0 +1,58 @@
+name: CodeGuardian
+description: A code review environment that tests AI agents on detecting security issues, assessing risks, and suggesting fixes
+version: 1.0.0
+
+spaces:
+  - name: code_review
+    description: Code review and security analysis tasks
+    type: gymnasium
+    
+tasks:
+  - id: easy_off_by_one
+    name: "Off-by-One Error"
+    difficulty: easy
+    description: "Detect off-by-one error in array summation"
+    space: code_review
+    
+  - id: medium_sql_injection  
+    name: "SQL Injection"
+    difficulty: medium
+    description: "Identify SQL injection vulnerability in user query"
+    space: code_review
+    
+  - id: hard_async_race
+    name: "Async Race Condition"
+    difficulty: hard
+    description: "Find missing await and race condition in async code"
+    space: code_review
+
+reward:
+  type: continuous
+  min: 0.0
+  max: 1.0
+  description: "Score based on correct issue detection, risk assessment, and fix suggestions"
+
+environment:
+  class: CodeGuardianEnv
+  module: env.environment
+  observation_space:
+    type: dict
+    fields:
+      code: str
+      intent: str
+      constraints: list
+      task_id: str
+      difficulty: str
+      step_index: int
+  action_space:
+    type: dict
+    fields:
+      action_type: str
+      detected_issues: list
+      risk_level: str
+      suggested_fix: str
+
+metadata:
+  author: CodeGuardian Team
+  tags: ["code-review", "security", "ai-testing"]
+  license: MIT

requirements.txt

@@ -0,0 +1,7 @@
+pydantic>=2.0.0
+openai>=1.0.0
+gymnasium>=0.28.0
+numpy>=1.24.0
+python-dotenv>=1.0.0
+flask>=3.0.0
+gradio>=4.0.0