Smoke-train deployment update

README.md

@@ -176,7 +176,7 @@ Recommended artifacts to include here:
 
 ```powershell
 cd C:\Users\kaust\PycharmProjects\meta-rl-dsa-solver
-python -m venv .venv
+py -3.11 -m venv .venv
 .\.venv\Scripts\pip install -e .
 ```
 
@@ -186,6 +186,12 @@ For training and plotting, also install your training extras:
 .\.venv\Scripts\pip install trl unsloth matplotlib wandb
 ```
 
+Recommended training target:
+
+- Python `3.11`
+- Base model `Qwen/Qwen2.5-3B-Instruct`
+- Single NVIDIA L4 with 4-bit LoRA + Unsloth GRPO
+
 ### 2. Start the OpenEnv server
 
 ```powershell
@@ -222,13 +228,13 @@ curl "http://localhost:7860/state?session_id=<SESSION_ID>"
 python training\train_grpo.py ^
   --generator-mode reward_aware ^
   --baseline-eval ^
-  --output-dir outputs_v3
+  --output-dir outputs_l4
 ```
 
 ### 7. Plot the training curves
 
 ```powershell
-python training\plot_results.py outputs_v3\reward_curve.csv
+python training\plot_results.py outputs_l4\reward_curve.csv
 ```
 
 ## Hugging Face Space

env/__init__.py

@@ -1,4 +1,3 @@
-from env.adapt_env import AdaptEnvironment
-from env.generator import GeneratorAgent, generator_reward, validate_problem
+from env.generator import GeneratorAgent, generator_reward, normalize_problem, validate_problem
 
-__all__ = ["AdaptEnvironment", "GeneratorAgent", "generator_reward", "validate_problem"]
+__all__ = ["GeneratorAgent", "generator_reward", "normalize_problem", "validate_problem"]

env/adapt_env.py

@@ -1,12 +1,12 @@
 from __future__ import annotations
 
-import ast
 from typing import Any, Generic, TypeVar
 from uuid import uuid4
 
 from env.generator import DIFFICULTY_LABELS, GeneratorAgent, generator_reward, validate_problem
 from models import AdaptAction, AdaptObservation, AdaptState
-from verifier.metrics import compute_reward
+from verifier.metrics import compute_episode_reward
+from verifier.verifier import verify
 
 try:
     from openenv.core.env_server.interfaces import Environment
@@ -22,7 +22,6 @@ except ImportError:
             pass
 
 
-FORBIDDEN_IMPORTS = {"os", "pathlib", "shutil", "socket", "subprocess"}
 MAX_STEPS_PER_EPISODE = 3
 TARGET_EFFICIENCY_SCORE = 0.95
 
@@ -158,76 +157,27 @@ class AdaptEnvironment(Environment[AdaptAction, AdaptObservation, AdaptState]):
         previous_status = self.previous_execution_status
         previous_pass_rate = float(self._state.last_pass_rate or 0.0)
 
-        syntax_ok, syntax_error = self._check_syntax(action.code)
-        if not syntax_ok:
-            done = attempt_number >= MAX_STEPS_PER_EPISODE
-            observation = self._build_observation(
-                reward=0.0,
-                done=done,
-                feedback=self._format_static_feedback(
-                    attempt_number=attempt_number,
-                    previous_status=previous_status,
-                    execution_status="syntax_error",
-                    details=f"Syntax error: {syntax_error}",
-                ),
-                syntax_valid=False,
-                execution_status="syntax_error",
-                reward_components={
-                    "correctness": 0.0,
-                    "step_discount": 1.0 if attempt_number == 1 else (0.85 if attempt_number == 2 else 0.70),
-                    "progress_delta": 0.0,
-                },
-            )
-            self.last_results = []
-            self.previous_execution_status = observation.execution_status
-            self._record_metrics(observation)
-            if done:
-                self._finalize_episode(observation)
-            return observation
-
-        safety_ok, safety_error = self._check_safety(action.code)
-        if not safety_ok:
-            done = attempt_number >= MAX_STEPS_PER_EPISODE
-            observation = self._build_observation(
-                reward=0.0,
-                done=done,
-                feedback=self._format_static_feedback(
-                    attempt_number=attempt_number,
-                    previous_status=previous_status,
-                    execution_status="safety_violation",
-                    details=safety_error,
-                ),
-                syntax_valid=True,
-                execution_status="safety_violation",
-                reward_components={
-                    "correctness": 0.0,
-                    "step_discount": 1.0 if attempt_number == 1 else (0.85 if attempt_number == 2 else 0.70),
-                    "progress_delta": 0.0,
-                },
-            )
-            self.last_results = []
-            self.previous_execution_status = observation.execution_status
-            self._record_metrics(observation)
-            if done:
-                self._finalize_episode(observation)
-            return observation
-
-        _, metadata = self._verify_submission(action.code)
+        _, metadata = self._verify_submission(action.code, attempt_number=attempt_number)
         self.last_results = list(metadata.get("results", []))
+
         hidden_pass_rate = float(metadata.get("hidden_pass_rate", metadata.get("pass_rate", 0.0)))
         visible_pass_rate = float(metadata.get("visible_pass_rate", 0.0))
         execution_status = str(metadata.get("execution_status", "completed"))
+        syntax_valid = bool(metadata.get("syntax_valid", execution_status != "syntax_error"))
+        safety_valid = bool(metadata.get("safety_valid", execution_status != "safety_violation"))
+        error_detail = str(metadata.get("error", "")).strip()
         efficiency_score = float(metadata.get("efficiency_score", 0.0))
-        efficiency_target_met = hidden_pass_rate == 1.0 and efficiency_score >= TARGET_EFFICIENCY_SCORE
-        done = efficiency_target_met or attempt_number >= MAX_STEPS_PER_EPISODE
-        reward, reward_components = self._shape_reward(
+        optimization_target_met = hidden_pass_rate == 1.0 and efficiency_score >= TARGET_EFFICIENCY_SCORE
+        done = optimization_target_met or attempt_number >= MAX_STEPS_PER_EPISODE
+
+        reward, reward_components = compute_episode_reward(
             pass_rate=hidden_pass_rate,
             step_number=attempt_number,
             execution_status=execution_status,
             previous_pass_rate=previous_pass_rate,
             done=done,
             efficiency_score=efficiency_score,
-            optimization_target_met=efficiency_target_met,
+            optimization_target_met=optimization_target_met,
         )
         feedback = self._format_feedback(
             results=self.last_results,
@@ -238,8 +188,19 @@ class AdaptEnvironment(Environment[AdaptAction, AdaptObservation, AdaptState]):
             visible_pass_rate=visible_pass_rate,
             efficiency_score=efficiency_score,
             optimization_hints=list(metadata.get("optimization_hints", [])),
-            optimization_target_met=efficiency_target_met,
+            optimization_target_met=optimization_target_met,
+            error_detail=error_detail,
         )
+
+        reward_components.update(
+            {
+                "format_compliance": round(float(metadata.get("format_compliance", 0.0)), 4),
+                "anti_cheat_compliance": round(1.0 if safety_valid and syntax_valid else 0.0, 4),
+                "hidden_correctness": round(hidden_pass_rate, 4),
+                "visible_correctness": round(visible_pass_rate, 4),
+            }
+        )
+
         observation = self._build_observation(
             reward=reward,
             done=done,
@@ -247,7 +208,7 @@ class AdaptEnvironment(Environment[AdaptAction, AdaptObservation, AdaptState]):
             pass_rate=hidden_pass_rate,
             visible_pass_rate=visible_pass_rate,
             hidden_pass_rate=hidden_pass_rate,
-            syntax_valid=True,
+            syntax_valid=syntax_valid,
             execution_status=execution_status,
             timeout_count=int(metadata.get("timeout_count", 0)),
             runtime_error_count=int(metadata.get("runtime_error_count", 0)),
@@ -337,23 +298,15 @@ class AdaptEnvironment(Environment[AdaptAction, AdaptObservation, AdaptState]):
             raise ValueError("Generator produced an invalid problem twice in a row.")
         return fallback
 
-    def _verify_submission(self, code: str) -> tuple[float, dict[str, Any]]:
-        try:
-            from verifier.verifier import verify
-        except ImportError as exc:
-            return 0.0, {
-                "feedback": f"Verifier unavailable: {exc}",
-                "execution_status": "verifier_error",
-                "results": [],
-            }
-
+    def _verify_submission(self, code: str, *, attempt_number: int) -> tuple[float, dict[str, Any]]:
         try:
-            reward, metadata = verify(code, self.test_cases)
+            reward, metadata = verify(code, self.test_cases, step_number=attempt_number)
         except Exception as exc:
             return 0.0, {
                 "feedback": f"Verifier crashed: {exc}",
                 "execution_status": "verifier_error",
                 "results": [],
+                "error": str(exc),
             }
 
         metadata = dict(metadata or {})
@@ -367,48 +320,6 @@ class AdaptEnvironment(Environment[AdaptAction, AdaptObservation, AdaptState]):
         )
         return float(reward), metadata
 
-    def _shape_reward(
-        self,
-        pass_rate: float,
-        step_number: int,
-        execution_status: str,
-        previous_pass_rate: float,
-        done: bool,
-        efficiency_score: float,
-        optimization_target_met: bool,
-    ) -> tuple[float, dict[str, float]]:
-        step_discount = 1.0 if step_number == 1 else (0.85 if step_number == 2 else 0.70)
-        progress_delta = max(0.0, float(pass_rate) - float(previous_pass_rate))
-        efficiency_score = max(0.0, min(float(efficiency_score), 1.0))
-
-        if execution_status in {"timeout", "syntax_error", "safety_violation"}:
-            reward = 0.0
-        elif pass_rate == 1.0:
-            reward = round(
-                compute_reward(
-                pass_rate=pass_rate,
-                step_number=step_number,
-                execution_status=execution_status,
-                format_compliance=0.0,
-                )
-                * (0.6 + 0.4 * efficiency_score),
-                4,
-            )
-            if not optimization_target_met and not done:
-                reward = min(reward, 0.94)
-        elif done:
-            reward = 0.0
-        else:
-            reward = round(0.1 * progress_delta, 4)
-
-        return reward, {
-            "correctness": round(float(pass_rate), 4),
-            "efficiency_score": round(efficiency_score, 4),
-            "step_discount": round(step_discount, 4),
-            "progress_delta": round(progress_delta, 4),
-            "reward": round(float(reward), 4),
-        }
-
     def _format_feedback(
         self,
         results: list[dict[str, Any]],
@@ -420,7 +331,16 @@ class AdaptEnvironment(Environment[AdaptAction, AdaptObservation, AdaptState]):
         efficiency_score: float,
         optimization_hints: list[str],
         optimization_target_met: bool,
+        error_detail: str,
     ) -> str:
+        if execution_status in {"syntax_error", "safety_violation"}:
+            return self._format_static_feedback(
+                attempt_number=attempt_number,
+                previous_status=previous_status,
+                execution_status=execution_status,
+                details=error_detail or execution_status.replace("_", " ").title(),
+            )
+
         lines = [
             f"Attempt {attempt_number}/{MAX_STEPS_PER_EPISODE}.",
             f"Previous attempt status: {previous_status}.",
@@ -556,9 +476,7 @@ class AdaptEnvironment(Environment[AdaptAction, AdaptObservation, AdaptState]):
             return ""
         chunks = []
         for test_case in visible_cases:
-            chunks.append(
-                f"Input:\n{test_case['input']}Expected Output:\n{test_case['output']}\n"
-            )
+            chunks.append(f"Input:\n{test_case['input']}Expected Output:\n{test_case['output']}\n")
         return "\n".join(chunks).rstrip()
 
     def _diversity_bonus(self, problem_type: str) -> float:
@@ -569,29 +487,6 @@ class AdaptEnvironment(Environment[AdaptAction, AdaptObservation, AdaptState]):
             return 0.0
         return 0.1
 
-    def _check_syntax(self, code: str) -> tuple[bool, str]:
-        try:
-            ast.parse(code)
-        except SyntaxError as exc:
-            return False, str(exc)
-        return True, ""
-
-    def _check_safety(self, code: str) -> tuple[bool, str]:
-        tree = ast.parse(code)
-        for node in ast.walk(tree):
-            if isinstance(node, ast.Import):
-                for alias in node.names:
-                    root_name = alias.name.split(".", 1)[0]
-                    if root_name in FORBIDDEN_IMPORTS:
-                        return False, f"Forbidden import: {root_name}"
-
-            if isinstance(node, ast.ImportFrom):
-                root_name = (node.module or "").split(".", 1)[0]
-                if root_name in FORBIDDEN_IMPORTS:
-                    return False, f"Forbidden import: {root_name}"
-
-        return True, ""
-
     def _tier_to_difficulty(self, tier: int) -> str:
         return DIFFICULTY_LABELS.get(tier, "easy")
 

env/app.py

@@ -1,13 +1,176 @@
+from __future__ import annotations
+
+from uuid import uuid4
+
 import gradio as gr
 
-def solve(problem):
-    return "Model output will come here"
+from env.adapt_env import AdaptEnvironment
+from models import AdaptAction
+from server.runtime import SpaceTrainingManager
+
+TRAINING_MANAGER = SpaceTrainingManager()
+SESSIONS: dict[str, AdaptEnvironment] = {}
+
+
+def _get_env(session_id: str | None) -> AdaptEnvironment:
+    if not session_id or session_id not in SESSIONS:
+        session_id = str(uuid4())
+        SESSIONS[session_id] = AdaptEnvironment(session_id=session_id)
+    return SESSIONS[session_id]
+
+
+def _problem_markdown(observation: dict) -> str:
+    return (
+        f"### {observation['problem_type']} ({observation['difficulty']})\n\n"
+        f"{observation['problem']}\n\n"
+        f"**Input Format**\n{observation['input_format']}\n\n"
+        f"**Constraints**\n{observation['constraints']}"
+    )
+
+
+def sample_problem(problem_id: str, difficulty: str) -> tuple[str, str, str, str, dict]:
+    env = _get_env(None)
+    observation = env.reset(
+        problem_id=problem_id or None,
+        difficulty=difficulty or None,
+        session_id=env.session_id,
+    )
+    payload = observation.model_dump()
+    return (
+        env.session_id,
+        _problem_markdown(payload),
+        payload["feedback"],
+        "",
+        payload,
+    )
+def evaluate_submission(session_id: str, code: str) -> tuple[str, str, str, dict]:
+    env = _get_env(session_id)
+    observation = env.step(AdaptAction(session_id=env.session_id, code=code))
+    payload = observation.model_dump()
+    status = (
+        f"Reward: {payload['reward']:.2f} | Hidden pass rate: {payload['hidden_pass_rate']:.2f} | "
+        f"Visible pass rate: {payload['visible_pass_rate']:.2f} | Status: {payload['execution_status']}"
+    )
+    return payload["feedback"], status, code, payload
+
+
+def model_attempt(session_id: str) -> tuple[str, str, str, dict]:
+    env = _get_env(session_id)
+    if not env.problem:
+        observation = env.reset(session_id=env.session_id)
+    else:
+        observation = env._build_observation(
+            reward=float(env.state.last_reward or 0.0),
+            done=bool(env.episode_done),
+            feedback=env.state.last_feedback or "No attempt yet.",
+            pass_rate=float(env.state.last_pass_rate or 0.0),
+            visible_pass_rate=float(env.state.recent_metrics.get("visible_pass_rate", 0.0)),
+            hidden_pass_rate=float(env.state.last_pass_rate or 0.0),
+            syntax_valid=env.state.last_execution_status != "syntax_error",
+            execution_status=env.state.last_execution_status or "ready",
+            timeout_count=int(env.state.recent_metrics.get("timeout_count", 0)),
+            runtime_error_count=int(env.state.recent_metrics.get("runtime_error_count", 0)),
+            invalid_output_count=int(env.state.recent_metrics.get("invalid_output_count", 0)),
+            wrong_answer_count=int(env.state.recent_metrics.get("wrong_answer_count", 0)),
+            format_compliance=float(env.state.recent_metrics.get("format_compliance", 0.0)),
+            reward_components=dict(env.state.recent_metrics.get("reward_components", {})),
+            generator_reward_signal=float(env.state.generator_reward_signal or 0.0),
+        )
+
+    try:
+        generation = TRAINING_MANAGER.generate_code(
+            problem=observation.problem,
+            input_format=observation.input_format,
+            constraints=observation.constraints,
+            feedback=observation.feedback,
+            problem_id=observation.problem_id,
+            problem_type=observation.problem_type,
+            difficulty=observation.difficulty,
+            attempt_number=observation.attempt_number + 1,
+            max_steps=observation.max_steps,
+        )
+    except Exception as exc:
+        return str(exc), "Model generation unavailable", "", {"error": str(exc)}
+
+    return evaluate_submission(session_id, generation["code"])
+
+
+with gr.Blocks(
+    title="ADAPT DSA Tutor Demo",
+    css="""
+    .panel {border: 1px solid #d7d3c9; border-radius: 18px; background: #fffaf2;}
+    .hero {background: linear-gradient(135deg, #f7eedb, #f3f8ef); border-radius: 22px; padding: 18px;}
+    """,
+) as demo:
+    session_id = gr.Textbox(label="Session ID", interactive=False)
+    state_payload = gr.JSON(label="Observation Payload")
+
+    gr.Markdown(
+        """
+        # ADAPT DSA Tutor
+        Sample a problem, inspect the verifier feedback, and compare your repair attempt with the currently loaded model path.
+        """,
+        elem_classes=["hero"],
+    )
+
+    with gr.Row():
+        problem_id = gr.Dropdown(
+            choices=[
+                "",
+                "sum_even_numbers",
+                "range_span",
+                "count_vowels",
+                "max_consecutive_ones",
+                "fizzbuzz_variant",
+                "running_total",
+                "count_local_peaks",
+                "longest_non_decreasing_run",
+                "two_sum_count",
+                "max_subarray_sum",
+                "group_anagrams_count",
+                "balanced_brackets",
+                "matrix_diagonal_sum",
+                "smallest_most_frequent",
+                "reverse_words",
+                "longest_common_subsequence",
+                "word_ladder_steps",
+                "merge_intervals",
+                "min_coins",
+                "rotate_matrix_90",
+            ],
+            value="",
+            label="Problem Family",
+            info="Leave blank to sample automatically.",
+        )
+        difficulty = gr.Radio(choices=["easy", "medium", "hard"], value="easy", label="Difficulty")
+        sample_btn = gr.Button("Sample Problem", variant="primary")
+
+    problem_view = gr.Markdown(elem_classes=["panel"])
+    with gr.Row():
+        code = gr.Textbox(label="Python Submission", lines=18, max_lines=24, placeholder="Write code that reads stdin and prints stdout.")
+        with gr.Column():
+            feedback = gr.Textbox(label="Verifier Feedback", lines=14)
+            status = gr.Textbox(label="Scorecard", lines=4)
+            with gr.Row():
+                verify_btn = gr.Button("Verify Submission", variant="primary")
+                model_btn = gr.Button("Run Current Model", variant="secondary")
+
+    sample_btn.click(
+        fn=sample_problem,
+        inputs=[problem_id, difficulty],
+        outputs=[session_id, problem_view, feedback, code, state_payload],
+    )
+    verify_btn.click(
+        fn=evaluate_submission,
+        inputs=[session_id, code],
+        outputs=[feedback, status, code, state_payload],
+    )
+    model_btn.click(
+        fn=model_attempt,
+        inputs=[session_id],
+        outputs=[feedback, status, code, state_payload],
+    )
 
-demo = gr.Interface(
-    fn=solve,
-    inputs="text",
-    outputs="text",
-    title="DSA Solver"
-)
 
-demo.launch()
+if __name__ == "__main__":
+    demo.launch()

env/executor.py

@@ -1,48 +1,97 @@
 from __future__ import annotations
 
 import os
+import platform
 import shutil
 import subprocess
 import sys
+import tempfile
+import time
 from pathlib import Path
-from uuid import uuid4
+from typing import Any
 
+TIMEOUT_SECONDS = 1.0
+MEMORY_LIMIT_MB = 512
+OUTPUT_LIMIT_BYTES = 256 * 1024
 
-TIMEOUT_SECONDS = 1
 
+def _sandbox_env(tmpdir: Path) -> dict[str, str]:
+    # Run user code with a tightly scoped environment to reduce hidden state.
+    return {
+        "PYTHONIOENCODING": "utf-8",
+        "PYTHONUNBUFFERED": "1",
+        "PYTHONNOUSERSITE": "1",
+        "HOME": str(tmpdir),
+        "TMPDIR": str(tmpdir),
+        "TEMP": str(tmpdir),
+        "TMP": str(tmpdir),
+    }
 
-def run_code(code: str, input_data: str, timeout_seconds: int | float | None = None) -> dict:
+
+def _linux_preexec_fn(timeout_seconds: float) -> Any:
+    if platform.system().lower() != "linux":
+        return None
+
+    def _apply_limits() -> None:
+        import resource
+
+        memory_limit = MEMORY_LIMIT_MB * 1024 * 1024
+        cpu_limit = max(1, int(timeout_seconds) + 1)
+        file_limit = OUTPUT_LIMIT_BYTES
+
+        resource.setrlimit(resource.RLIMIT_AS, (memory_limit, memory_limit))
+        resource.setrlimit(resource.RLIMIT_CPU, (cpu_limit, cpu_limit))
+        resource.setrlimit(resource.RLIMIT_FSIZE, (file_limit, file_limit))
+        resource.setrlimit(resource.RLIMIT_NOFILE, (32, 32))
+
+    return _apply_limits
+
+
+def run_code(code: str, input_data: str, timeout_seconds: int | float | None = None) -> dict[str, Any]:
+    timeout_value = float(TIMEOUT_SECONDS if timeout_seconds is None else timeout_seconds)
     temp_parent = Path(os.getenv("ADAPT_TMP_DIR", ".adapt_tmp")).resolve()
     temp_parent.mkdir(parents=True, exist_ok=True)
-    tmpdir = temp_parent / f"run_{uuid4().hex}"
-    tmpdir.mkdir()
-    timeout_value = TIMEOUT_SECONDS if timeout_seconds is None else timeout_seconds
 
-    try:
-        file_path = Path(tmpdir) / "submission.py"
-        file_path.write_text(code, encoding="utf-8")
+    tmpdir_path = Path(tempfile.mkdtemp(prefix="run_", dir=str(temp_parent)))
+    submission_path = tmpdir_path / "submission.py"
+    submission_path.write_text(code, encoding="utf-8")
 
+    started = time.perf_counter()
+    try:
         try:
             result = subprocess.run(
-                [sys.executable, str(file_path)],
+                [sys.executable, "-I", "-S", str(submission_path)],
                 input=input_data,
                 text=True,
                 capture_output=True,
                 timeout=timeout_value,
+                cwd=str(tmpdir_path),
+                env=_sandbox_env(tmpdir_path),
+                preexec_fn=_linux_preexec_fn(timeout_value),
             )
         except subprocess.TimeoutExpired as exc:
+            duration_ms = round((time.perf_counter() - started) * 1000, 2)
             return {
-                "stdout": exc.stdout or "",
+                "stdout": str(exc.stdout or ""),
                 "stderr": "Execution timed out",
                 "exit_code": -1,
                 "timed_out": True,
+                "duration_ms": duration_ms,
+                "sandboxed": True,
+                "sandbox_mode": "linux_limited" if platform.system().lower() == "linux" else "portable",
             }
 
+        duration_ms = round((time.perf_counter() - started) * 1000, 2)
+        stdout = result.stdout[:OUTPUT_LIMIT_BYTES]
+        stderr = result.stderr[:OUTPUT_LIMIT_BYTES]
         return {
-            "stdout": result.stdout,
-            "stderr": result.stderr,
-            "exit_code": result.returncode,
+            "stdout": stdout,
+            "stderr": stderr,
+            "exit_code": int(result.returncode),
             "timed_out": False,
+            "duration_ms": duration_ms,
+            "sandboxed": True,
+            "sandbox_mode": "linux_limited" if platform.system().lower() == "linux" else "portable",
         }
     finally:
-        shutil.rmtree(tmpdir, ignore_errors=True)
+        shutil.rmtree(tmpdir_path, ignore_errors=True)

env/generator.py

@@ -104,6 +104,16 @@ def validate_problem(problem_dict: dict[str, Any]) -> bool:
     return True
 
 
+def normalize_problem(problem_dict: dict[str, Any]) -> dict[str, Any]:
+    normalized = dict(problem_dict)
+    normalized["problem"] = str(problem_dict.get("problem", "")).strip()
+    normalized["input_format"] = str(problem_dict.get("input_format", "")).strip()
+    normalized["constraints"] = str(problem_dict.get("constraints", "")).strip()
+    normalized["test_cases"] = [dict(test_case) for test_case in problem_dict.get("test_cases", [])]
+    normalized["visible_problem"] = dict(problem_dict.get("visible_problem", {}))
+    return normalized
+
+
 class GeneratorAgent:
     """Deterministic, dependency-free generator for DSA-style problems."""
 
@@ -157,8 +167,9 @@ class GeneratorAgent:
                 "generation_mode": "deterministic_fallback" if self.deterministic else "local_rule_based",
                 "validity_bonus": 0.15,
             }
-            if validate_problem(problem):
-                return problem
+            normalized_problem = normalize_problem(problem)
+            if validate_problem(normalized_problem):
+                return normalized_problem
 
         raise ValueError(f"Unable to generate a valid problem for template {template.problem_type}")
 

pyproject.toml

@@ -7,8 +7,9 @@ name = "adapt-dsa-tutor"
 version = "0.2.0"
 description = "OpenEnv-compliant adversarial DSA tutor environment for RLVR code generation."
 readme = "README.md"
-requires-python = ">=3.10"
+requires-python = ">=3.11,<3.12"
 dependencies = [
+    "gradio>=6.0.0",
     "openenv-core>=0.2.3",
     "fastapi>=0.104.0",
     "pydantic>=2.0.0",

requirements.txt

@@ -1,7 +1,8 @@
-openenv 
-fastapi 
-uvicorn 
-pydantic
-huggingface_hub
-transformers
-peft
+gradio>=6.0.0
+openenv-core>=0.2.3
+fastapi>=0.104.0
+uvicorn>=0.24.0
+pydantic>=2.0.0
+huggingface_hub>=0.30.0
+transformers>=4.48.0
+peft>=0.14.0

scripts/deploy_and_smoke_train.py

@@ -8,16 +8,79 @@ import sys
 import time
 import urllib.error
 import urllib.request
+from dataclasses import dataclass
 from pathlib import Path
 from typing import Any
 
-
 DEFAULT_REMOTE = "space"
 DEFAULT_REMOTE_BRANCH = "main"
 DEFAULT_POLL_INTERVAL_SECONDS = 10
-DEFAULT_TIMEOUT_SECONDS = 60 * 30
+DEFAULT_DEPLOY_TIMEOUT_SECONDS = 60 * 20
+DEFAULT_TRAIN_TIMEOUT_SECONDS = 60 * 30
 DEFAULT_REQUIRED_HEALTHY_CHECKS = 3
 DEFAULT_MIN_DEPLOY_WAIT_SECONDS = 30
+DEFAULT_COMMIT_MESSAGE = "Smoke-train deployment update"
+
+
+class ScriptError(RuntimeError):
+    pass
+
+
+def print_info(message: str) -> None:
+    print(f"[info] {message}", flush=True)
+
+
+def print_warn(message: str) -> None:
+    print(f"[warn] {message}", flush=True)
+
+
+def print_error(message: str) -> None:
+    print(f"[error] {message}", file=sys.stderr, flush=True)
+
+
+def pretty_json(payload: dict[str, Any]) -> str:
+    return json.dumps(payload, indent=2, sort_keys=True)
+
+
+def api_url(base_url: str, path: str) -> str:
+    return f"{base_url.rstrip('/')}/{path.lstrip('/')}"
+
+
+def http_json(
+    method: str,
+    url: str,
+    *,
+    payload: dict[str, Any] | None = None,
+    timeout_seconds: int = 60,
+) -> dict[str, Any]:
+    data = None
+    headers = {"Accept": "application/json"}
+    if payload is not None:
+        data = json.dumps(payload).encode("utf-8")
+        headers["Content-Type"] = "application/json"
+
+    request = urllib.request.Request(url=url, data=data, headers=headers, method=method.upper())
+    try:
+        with urllib.request.urlopen(request, timeout=timeout_seconds) as response:
+            raw = response.read().decode("utf-8")
+            if not raw.strip():
+                return {}
+            return json.loads(raw)
+    except urllib.error.HTTPError as exc:
+        body = exc.read().decode("utf-8", errors="replace")
+        raise ScriptError(f"HTTP {exc.code} for {url}: {body}") from exc
+    except urllib.error.URLError as exc:
+        raise ScriptError(f"Request to {url} failed: {exc}") from exc
+
+
+@dataclass(frozen=True)
+class GitOptions:
+    repo_root: Path
+    remote: str
+    remote_branch: str
+    commit_message: str
+    skip_commit: bool
+    skip_push: bool
 
 
 def run_command(command: list[str], cwd: Path) -> subprocess.CompletedProcess[str]:
@@ -34,7 +97,7 @@ def ensure_success(result: subprocess.CompletedProcess[str], action: str) -> Non
     if result.returncode == 0:
         return
     message = result.stderr.strip() or result.stdout.strip() or f"{action} failed with exit code {result.returncode}"
-    raise RuntimeError(f"{action} failed: {message}")
+    raise ScriptError(f"{action} failed: {message}")
 
 
 def has_uncommitted_changes(repo_root: Path) -> bool:
@@ -43,29 +106,6 @@ def has_uncommitted_changes(repo_root: Path) -> bool:
     return bool(result.stdout.strip())
 
 
-def commit_changes(repo_root: Path, message: str) -> str | None:
-    if not has_uncommitted_changes(repo_root):
-        print("No uncommitted changes found. Reusing the current HEAD commit.", flush=True)
-        return None
-
-    add_result = run_command(["git", "add", "-A"], repo_root)
-    ensure_success(add_result, "git add")
-
-    commit_result = run_command(["git", "commit", "-m", message], repo_root)
-    ensure_success(commit_result, "git commit")
-    print(commit_result.stdout.strip(), flush=True)
-
-    rev_result = run_command(["git", "rev-parse", "HEAD"], repo_root)
-    ensure_success(rev_result, "git rev-parse")
-    return rev_result.stdout.strip()
-
-
-def push_to_space(repo_root: Path, remote: str, remote_branch: str) -> None:
-    push_result = run_command(["git", "push", remote, f"HEAD:{remote_branch}"], repo_root)
-    ensure_success(push_result, f"git push {remote} HEAD:{remote_branch}")
-    print(push_result.stdout.strip() or push_result.stderr.strip(), flush=True)
-
-
 def current_head_sha(repo_root: Path) -> str:
     result = run_command(["git", "rev-parse", "HEAD"], repo_root)
     ensure_success(result, "git rev-parse HEAD")
@@ -81,247 +121,382 @@ def remote_branch_sha(repo_root: Path, remote: str, remote_branch: str) -> str |
     return line.split()[0]
 
 
-def deployment_needed(repo_root: Path, remote: str, remote_branch: str) -> bool:
-    local_sha = current_head_sha(repo_root)
-    remote_sha = remote_branch_sha(repo_root, remote, remote_branch)
-    if remote_sha is None:
-        print(
-            f"Remote branch {remote}/{remote_branch} does not exist yet. A deployment push is required.",
-            flush=True,
-        )
-        return True
+def commit_changes_if_needed(options: GitOptions) -> str:
+    repo_root = options.repo_root
+    head_before = current_head_sha(repo_root)
+    if not has_uncommitted_changes(repo_root):
+        print_info(f"Working tree is clean at {head_before}.")
+        return head_before
 
-    if local_sha == remote_sha:
-        print(
-            f"Skipping Space deploy because {remote}/{remote_branch} is already at local HEAD {local_sha}.",
-            flush=True,
+    if options.skip_commit:
+        raise ScriptError(
+            "Working tree has uncommitted changes, but --skip-commit was set. "
+            "Commit the changes manually or remove --skip-commit."
         )
-        return False
 
-    print(
-        f"Space deploy required: local HEAD {local_sha} differs from {remote}/{remote_branch} {remote_sha}.",
-        flush=True,
+    print_info("Staging and committing local changes before deploy.")
+    ensure_success(run_command(["git", "add", "-A"], repo_root), "git add -A")
+    commit_result = run_command(["git", "commit", "-m", options.commit_message], repo_root)
+    ensure_success(commit_result, "git commit")
+    if commit_result.stdout.strip():
+        print(commit_result.stdout.strip(), flush=True)
+    head_after = current_head_sha(repo_root)
+    print_info(f"Created commit {head_after}.")
+    return head_after
+
+
+def deployment_needed(options: GitOptions, local_head_sha: str) -> bool:
+    remote_sha = remote_branch_sha(options.repo_root, options.remote, options.remote_branch)
+    if remote_sha is None:
+        print_info(f"Remote branch {options.remote}/{options.remote_branch} does not exist yet.")
+        return True
+    if remote_sha == local_head_sha:
+        print_info(f"Remote {options.remote}/{options.remote_branch} already points to {local_head_sha}.")
+        return False
+    print_info(
+        f"Remote {options.remote}/{options.remote_branch} is at {remote_sha}; "
+        f"local HEAD is {local_head_sha}."
     )
     return True
 
 
-def http_json(method: str, url: str, payload: dict[str, Any] | None = None) -> dict[str, Any]:
-    data = None
-    headers = {}
-    if payload is not None:
-        data = json.dumps(payload).encode("utf-8")
-        headers["Content-Type"] = "application/json"
+def push_current_head(options: GitOptions) -> None:
+    print_info(f"Pushing HEAD to {options.remote}/{options.remote_branch}.")
+    result = run_command(["git", "push", options.remote, f"HEAD:{options.remote_branch}"], options.repo_root)
+    ensure_success(result, f"git push {options.remote} HEAD:{options.remote_branch}")
+    summary = result.stdout.strip() or result.stderr.strip()
+    if summary:
+        print(summary, flush=True)
 
-    request = urllib.request.Request(url=url, data=data, headers=headers, method=method)
-    try:
-        with urllib.request.urlopen(request, timeout=60) as response:
-            raw = response.read().decode("utf-8")
-            return json.loads(raw)
-    except urllib.error.HTTPError as exc:
-        body = exc.read().decode("utf-8", errors="replace")
-        raise RuntimeError(f"HTTP {exc.code} for {url}: {body}") from exc
-    except urllib.error.URLError as exc:
-        raise RuntimeError(f"Request to {url} failed: {exc}") from exc
 
+def fetch_health(base_url: str) -> dict[str, Any]:
+    return http_json("GET", api_url(base_url, "/health"))
 
-def try_get_space_health(health_url: str) -> dict[str, Any] | None:
-    try:
-        return http_json("GET", health_url)
-    except RuntimeError:
-        return None
 
-
-def wait_for_space_health(
+def wait_for_health(
     base_url: str,
+    *,
     timeout_seconds: int,
     poll_interval_seconds: int,
     required_healthy_checks: int,
     min_deploy_wait_seconds: int,
-) -> None:
-    health_url = f"{base_url.rstrip('/')}/health"
+) -> dict[str, Any]:
     deadline = time.time() + timeout_seconds
     push_started_at = time.time()
-    previous_payload = try_get_space_health(health_url)
-    deployment_transition_seen = previous_payload is None
-    fallback_notice_emitted = False
-    consecutive_healthy_checks = 0
+    prior_payload: dict[str, Any] | None = None
+    transition_observed = False
+    healthy_streak = 0
 
     while time.time() < deadline:
         try:
-            payload = http_json("GET", health_url)
-        except RuntimeError as exc:
-            deployment_transition_seen = True
-            consecutive_healthy_checks = 0
-            print(f"Waiting for Space deployment: {exc}", flush=True)
+            payload = fetch_health(base_url)
+        except ScriptError as exc:
+            healthy_streak = 0
+            transition_observed = True
+            print_info(f"Waiting for service health: {exc}")
             time.sleep(poll_interval_seconds)
             continue
 
-        if previous_payload != payload:
-            deployment_transition_seen = True
-
-        waited_long_enough = (time.time() - push_started_at) >= min_deploy_wait_seconds
+        if prior_payload is not None and payload != prior_payload:
+            transition_observed = True
 
-        if not deployment_transition_seen and not waited_long_enough:
-            print(
-                "Space is still serving the pre-push health payload; waiting for the new deployment to take over.",
-                flush=True,
-            )
-            previous_payload = payload
-            time.sleep(poll_interval_seconds)
-            continue
-
-        if not deployment_transition_seen and waited_long_enough and not fallback_notice_emitted:
-            print(
-                "No observable health payload transition detected after the minimum deploy wait. "
-                "Falling back to healthy-check stabilization.",
-                flush=True,
-            )
-            fallback_notice_emitted = True
-
-        if not waited_long_enough:
-            remaining = max(0, min_deploy_wait_seconds - int(time.time() - push_started_at))
-            print(
-                f"Deployment transition detected. Waiting {remaining}s more for the Space to stabilize.",
-                flush=True,
-            )
-            previous_payload = payload
+        elapsed = time.time() - push_started_at
+        if elapsed < min_deploy_wait_seconds:
+            remaining = max(0, int(min_deploy_wait_seconds - elapsed))
+            print_info(f"Health endpoint reachable. Waiting {remaining}s for deployment stabilization.")
+            prior_payload = payload
             time.sleep(poll_interval_seconds)
             continue
 
         if payload.get("status") == "healthy":
-            consecutive_healthy_checks += 1
-            print(
-                f"Space healthy check {consecutive_healthy_checks}/{required_healthy_checks}: {payload}",
-                flush=True,
+            healthy_streak += 1
+            print_info(
+                f"Health check {healthy_streak}/{required_healthy_checks}: "
+                f"training={payload.get('training')} model_loaded={payload.get('model_loaded')}"
             )
-            if consecutive_healthy_checks >= required_healthy_checks:
-                print("Space deployment looks stable. Starting smoke training.", flush=True)
-                return
+            if healthy_streak >= required_healthy_checks:
+                if not transition_observed and prior_payload is not None:
+                    print_warn("No payload transition was observed after deploy; continuing because health stabilized.")
+                return payload
         else:
-            consecutive_healthy_checks = 0
-            print(f"Space health not ready yet: {payload}", flush=True)
+            healthy_streak = 0
+            print_info(f"Health payload not ready yet: {pretty_json(payload)}")
 
-        previous_payload = payload
+        prior_payload = payload
         time.sleep(poll_interval_seconds)
 
-    raise TimeoutError(f"Space did not finish deploying within {timeout_seconds} seconds.")
+    raise ScriptError(f"Service did not report healthy within {timeout_seconds} seconds.")
+
+
+def fetch_training_status(base_url: str) -> dict[str, Any]:
+    return http_json("GET", api_url(base_url, "/train/status"))
+
+
+def summarize_training_status(payload: dict[str, Any]) -> str:
+    status = payload.get("status")
+    phase = payload.get("phase")
+    completed = payload.get("completed_steps")
+    total = payload.get("total_steps")
+    difficulty = payload.get("current_difficulty")
+    problem_family = payload.get("last_problem_family")
+    reward = payload.get("last_reward")
+    pieces = [
+        f"status={status}",
+        f"phase={phase}",
+        f"steps={completed}/{total}",
+    ]
+    if difficulty:
+        pieces.append(f"difficulty={difficulty}")
+    if problem_family:
+        pieces.append(f"family={problem_family}")
+    if reward is not None:
+        pieces.append(f"reward={reward}")
+    return " ".join(pieces)
+
+
+def ensure_no_active_training(base_url: str) -> None:
+    payload = fetch_training_status(base_url)
+    if payload.get("status") == "running":
+        raise ScriptError(
+            "The remote training manager already reports an active run. "
+            f"Current status: {summarize_training_status(payload)}"
+        )
 
 
-def start_smoke_training(base_url: str) -> dict[str, Any]:
-    train_url = f"{base_url.rstrip('/')}/train"
-    payload = http_json("POST", train_url, {"preset": "smoke"})
-    print(f"Started smoke training: {json.dumps(payload, indent=2)}", flush=True)
+def start_training(base_url: str, train_payload: dict[str, Any]) -> dict[str, Any]:
+    print_info(f"Starting training with payload: {pretty_json(train_payload)}")
+    payload = http_json("POST", api_url(base_url, "/train"), payload=train_payload)
+    print_info(f"Training accepted: {pretty_json(payload)}")
     return payload
 
 
-def poll_training_status(base_url: str, poll_interval_seconds: int, timeout_seconds: int) -> dict[str, Any]:
-    status_url = f"{base_url.rstrip('/')}/train/status"
+def poll_training_status(
+    base_url: str,
+    *,
+    poll_interval_seconds: int,
+    timeout_seconds: int,
+) -> dict[str, Any]:
     deadline = time.time() + timeout_seconds
+    last_signature: tuple[Any, ...] | None = None
 
     while time.time() < deadline:
-        payload = http_json("GET", status_url)
-        status = payload.get("status")
-        phase = payload.get("phase")
-        completed_steps = payload.get("completed_steps")
-        total_steps = payload.get("total_steps")
-        print(
-            f"Training status: status={status} phase={phase} completed_steps={completed_steps}/{total_steps}",
-            flush=True,
+        payload = fetch_training_status(base_url)
+        signature = (
+            payload.get("status"),
+            payload.get("phase"),
+            payload.get("completed_steps"),
+            payload.get("total_steps"),
+            payload.get("last_problem_family"),
+            payload.get("last_reward"),
+            payload.get("error"),
         )
+        if signature != last_signature:
+            print_info(summarize_training_status(payload))
+            last_signature = signature
 
+        status = payload.get("status")
         if status == "failed":
-            print("Training failed. Full error payload:", flush=True)
-            print(json.dumps(payload, indent=2), flush=True)
+            print_error("Training failed. Final payload follows.")
+            print(pretty_json(payload), flush=True)
             return payload
-
         if status == "succeeded":
-            print("Training succeeded. Final payload:", flush=True)
-            print(json.dumps(payload, indent=2), flush=True)
+            print_info("Training succeeded. Final payload follows.")
+            print(pretty_json(payload), flush=True)
             return payload
 
         time.sleep(poll_interval_seconds)
 
-    raise TimeoutError(f"Training did not finish within {timeout_seconds} seconds.")
+    raise ScriptError(f"Training did not finish within {timeout_seconds} seconds.")
+
+
+def build_train_payload(args: argparse.Namespace) -> dict[str, Any]:
+    payload: dict[str, Any] = {"preset": args.preset}
+    if args.train_payload_json:
+        extra_payload = json.loads(args.train_payload_json)
+        if not isinstance(extra_payload, dict):
+            raise ScriptError("--train-payload-json must decode to a JSON object.")
+        payload.update(extra_payload)
+    return payload
 
 
 def build_parser() -> argparse.ArgumentParser:
-    parser = argparse.ArgumentParser(description="Commit, push, trigger smoke training, and poll HF Space status.")
+    parser = argparse.ArgumentParser(
+        description="Deploy the current repo to a Hugging Face Space and run a smoke training job.",
+    )
     parser.add_argument(
         "--base-url",
         required=True,
-        help="Base URL of the deployed Hugging Face Space, for example https://<space>.hf.space",
+        help="Base URL of the running server, for example https://<space>.hf.space or http://localhost:7860",
     )
     parser.add_argument(
-        "--commit-message",
-        default="Automated smoke training update",
-        help="Commit message to use when local changes are present.",
+        "--preset",
+        default="smoke",
+        help="Training preset to send to /train. Defaults to smoke.",
     )
-    parser.add_argument("--remote", default=DEFAULT_REMOTE, help="Git remote for the Hugging Face Space.")
     parser.add_argument(
-        "--remote-branch",
-        default=DEFAULT_REMOTE_BRANCH,
-        help="Remote branch that deploys the Hugging Face Space.",
+        "--train-payload-json",
+        default=None,
+        help="Optional JSON object merged into the /train request body.",
     )
     parser.add_argument(
         "--poll-interval-seconds",
         type=int,
         default=DEFAULT_POLL_INTERVAL_SECONDS,
-        help="How often to poll /train/status.",
+        help="How often to poll /health and /train/status.",
     )
     parser.add_argument(
-        "--timeout-seconds",
+        "--deploy-timeout-seconds",
         type=int,
-        default=DEFAULT_TIMEOUT_SECONDS,
-        help="Overall timeout for Space health and smoke training completion.",
+        default=DEFAULT_DEPLOY_TIMEOUT_SECONDS,
+        help="Maximum time to wait for the service to become healthy after push.",
+    )
+    parser.add_argument(
+        "--train-timeout-seconds",
+        type=int,
+        default=DEFAULT_TRAIN_TIMEOUT_SECONDS,
+        help="Maximum time to wait for the smoke train run to finish.",
     )
     parser.add_argument(
         "--required-healthy-checks",
         type=int,
         default=DEFAULT_REQUIRED_HEALTHY_CHECKS,
-        help="How many consecutive healthy checks to require before starting training.",
+        help="Number of consecutive healthy /health checks required before training starts.",
     )
     parser.add_argument(
         "--min-deploy-wait-seconds",
         type=int,
         default=DEFAULT_MIN_DEPLOY_WAIT_SECONDS,
-        help="Minimum number of seconds to wait after the push before treating the Space as fully deployed.",
+        help="Minimum time to wait after push before treating health as stable.",
+    )
+    parser.add_argument(
+        "--remote",
+        default=DEFAULT_REMOTE,
+        help="Git remote to push to when deploying.",
+    )
+    parser.add_argument(
+        "--remote-branch",
+        default=DEFAULT_REMOTE_BRANCH,
+        help="Remote branch to push HEAD to when deploying.",
+    )
+    parser.add_argument(
+        "--commit-message",
+        default=DEFAULT_COMMIT_MESSAGE,
+        help="Commit message to use if local changes need to be committed before push.",
+    )
+    parser.add_argument(
+        "--skip-commit",
+        action="store_true",
+        help="Do not auto-commit local changes before deploy.",
+    )
+    parser.add_argument(
+        "--skip-push",
+        action="store_true",
+        help="Skip git deploy entirely and just hit the running server.",
+    )
+    parser.add_argument(
+        "--skip-health-check",
+        action="store_true",
+        help="Skip waiting on /health before training.",
+    )
+    parser.add_argument(
+        "--trigger-only",
+        action="store_true",
+        help="Start the smoke run and exit without polling to completion.",
+    )
+    parser.add_argument(
+        "--status-only",
+        action="store_true",
+        help="Do not start a new run; just print /train/status and optionally poll it.",
+    )
+    parser.add_argument(
+        "--follow-running",
+        action="store_true",
+        help="If /train/status already reports a running job, follow it instead of failing.",
     )
     return parser
 
 
+def maybe_deploy(args: argparse.Namespace, repo_root: Path) -> None:
+    if args.skip_push:
+        print_info("Skipping git deploy because --skip-push was set.")
+        return
+
+    git_options = GitOptions(
+        repo_root=repo_root,
+        remote=args.remote,
+        remote_branch=args.remote_branch,
+        commit_message=args.commit_message,
+        skip_commit=args.skip_commit,
+        skip_push=args.skip_push,
+    )
+    local_head_sha = commit_changes_if_needed(git_options)
+    if not deployment_needed(git_options, local_head_sha):
+        print_info("Skipping push because the remote is already on the current local HEAD.")
+        return
+
+    push_current_head(git_options)
+
+
 def main(argv: list[str] | None = None) -> int:
     args = build_parser().parse_args(argv)
     repo_root = Path(__file__).resolve().parents[1]
 
     try:
-        commit_hash = commit_changes(repo_root, args.commit_message)
-        if commit_hash:
-            print(f"Committed changes at {commit_hash}", flush=True)
-
-        should_deploy = deployment_needed(repo_root, args.remote, args.remote_branch)
-        if should_deploy:
-            push_to_space(repo_root, args.remote, args.remote_branch)
-            wait_for_space_health(
-                base_url=args.base_url,
-                timeout_seconds=args.timeout_seconds,
+        if args.status_only:
+            payload = fetch_training_status(args.base_url)
+            print(pretty_json(payload), flush=True)
+            if args.follow_running and payload.get("status") == "running":
+                final_status = poll_training_status(
+                    args.base_url,
+                    poll_interval_seconds=args.poll_interval_seconds,
+                    timeout_seconds=args.train_timeout_seconds,
+                )
+                return 0 if final_status.get("status") == "succeeded" else 1
+            return 0
+
+        maybe_deploy(args, repo_root)
+
+        if not args.skip_health_check:
+            wait_for_health(
+                args.base_url,
+                timeout_seconds=args.deploy_timeout_seconds,
                 poll_interval_seconds=args.poll_interval_seconds,
                 required_healthy_checks=args.required_healthy_checks,
                 min_deploy_wait_seconds=args.min_deploy_wait_seconds,
             )
         else:
-            print("Reusing the currently deployed Space revision. Skipping deploy wait.", flush=True)
-        start_smoke_training(args.base_url)
+            print_info("Skipping health wait because --skip-health-check was set.")
+
+        current_status = fetch_training_status(args.base_url)
+        if current_status.get("status") == "running":
+            if args.follow_running:
+                print_warn(
+                    "A training job is already running; following the existing run instead of starting a new one."
+                )
+                final_status = poll_training_status(
+                    args.base_url,
+                    poll_interval_seconds=args.poll_interval_seconds,
+                    timeout_seconds=args.train_timeout_seconds,
+                )
+                return 0 if final_status.get("status") == "succeeded" else 1
+            ensure_no_active_training(args.base_url)
+
+        train_payload = build_train_payload(args)
+        start_training(args.base_url, train_payload)
+        if args.trigger_only:
+            print_info("Training was triggered successfully; exiting because --trigger-only was set.")
+            return 0
+
         final_status = poll_training_status(
-            base_url=args.base_url,
+            args.base_url,
             poll_interval_seconds=args.poll_interval_seconds,
-            timeout_seconds=args.timeout_seconds,
+            timeout_seconds=args.train_timeout_seconds,
         )
+        return 0 if final_status.get("status") == "succeeded" else 1
     except Exception as exc:
-        print(f"Deployment/training loop failed: {exc}", file=sys.stderr, flush=True)
+        print_error(str(exc))
         return 1
 
-    return 0 if final_status.get("status") == "succeeded" else 1
-
 
 if __name__ == "__main__":
     raise SystemExit(main())

scripts/test_env.py

@@ -46,6 +46,7 @@ def main() -> None:
     assert correct.execution_status == "completed"
     assert correct.done is True
     assert correct.reward_components["efficiency_score"] >= 0.95
+    assert correct.reward_components["hidden_correctness"] == 1.0
 
     observation = env.reset(problem_id="running_total", difficulty="easy")
     repair_1 = env.step(
@@ -98,6 +99,7 @@ def main() -> None:
     assert less_optimized.done is False
     assert less_optimized.reward < 1.0
     assert "can still be optimized further" in less_optimized.feedback
+    assert less_optimized.reward_components["format_compliance"] == 1.0
 
     observation = env.reset(problem_id="sum_even_numbers", difficulty="easy")
     syntax = env.step(AdaptAction(code="def broken(:\n    pass"))
@@ -130,6 +132,7 @@ def main() -> None:
     assert unsafe.reward == 0.0
     assert unsafe.execution_status == "safety_violation"
     assert unsafe.done is False
+    assert unsafe.reward_components["anti_cheat_compliance"] == 0.0
 
     assert env.state.history["attempts"]
     assert_hidden_tests_are_not_exposed(timeout.model_dump())

scripts/test_training_config.py

@@ -0,0 +1,74 @@
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[1]
+if str(ROOT) not in sys.path:
+    sys.path.insert(0, str(ROOT))
+
+from training.train_grpo import build_training_config, resolve_precision_policy
+from verifier.metrics import compute_episode_reward
+
+
+class FakeCuda:
+    def __init__(self, available: bool, bf16_supported: bool) -> None:
+        self._available = available
+        self._bf16_supported = bf16_supported
+
+    def is_available(self) -> bool:
+        return self._available
+
+    def is_bf16_supported(self) -> bool:
+        return self._bf16_supported
+
+
+class FakeTorch:
+    bfloat16 = "bfloat16"
+    float16 = "float16"
+    float32 = "float32"
+
+    def __init__(self, available: bool, bf16_supported: bool) -> None:
+        self.cuda = FakeCuda(available=available, bf16_supported=bf16_supported)
+
+
+def main() -> None:
+    l4_config = build_training_config("l4")
+    smoke_config = build_training_config("smoke")
+
+    assert l4_config.model_name == "Qwen/Qwen2.5-3B-Instruct"
+    assert l4_config.load_in_4bit is True
+    assert l4_config.gradient_checkpointing is True
+    assert l4_config.num_generations == 4
+
+    assert smoke_config.load_in_4bit is False
+    assert smoke_config.gradient_checkpointing is False
+
+    bf16_policy = resolve_precision_policy(l4_config, FakeTorch(available=True, bf16_supported=True))
+    assert bf16_policy["precision_mode"] == "bf16"
+    assert bf16_policy["load_in_4bit"] is True
+
+    fp16_policy = resolve_precision_policy(l4_config, FakeTorch(available=True, bf16_supported=False))
+    assert fp16_policy["precision_mode"] == "fp16"
+    assert fp16_policy["load_in_4bit"] is True
+
+    cpu_policy = resolve_precision_policy(smoke_config, FakeTorch(available=False, bf16_supported=False))
+    assert cpu_policy["precision_mode"] == "fp32"
+    assert cpu_policy["load_in_4bit"] is False
+
+    reward, components = compute_episode_reward(
+        pass_rate=1.0,
+        step_number=1,
+        execution_status="completed",
+        previous_pass_rate=0.0,
+        done=False,
+        efficiency_score=0.94,
+        optimization_target_met=False,
+    )
+    assert reward == 0.94
+    assert components["progress_delta"] == 1.0
+    print("Training config smoke tests passed")
+
+
+if __name__ == "__main__":
+    main()

scripts/test_verifier.py

@@ -54,6 +54,11 @@ nums = list(map(int, input().split()))
 print(nums[n])
 """
 
+safety_violation_code = """
+import os
+print(os.listdir("."))
+"""
+
 for name, code in [
     ("correct", correct_code),
     ("wrong", wrong_code),
@@ -61,6 +66,7 @@ for name, code in [
     ("invalid_output", invalid_output_code),
     ("timeout", timeout_code),
     ("runtime_error", runtime_error_code),
+    ("safety_violation", safety_violation_code),
 ]:
     reward, info = verify(code, test_cases)
 
@@ -77,5 +83,10 @@ for name, code in [
 
 reward_optimal, info_optimal = verify(correct_code, test_cases)
 reward_less_optimal, info_less_optimal = verify(less_optimized_code, test_cases)
+reward_safety, info_safety = verify(safety_violation_code, test_cases)
 assert info_optimal["efficiency_score"] > info_less_optimal["efficiency_score"]
 assert info_less_optimal["complexity_signals"]["list_comprehensions"] > 0
+assert info_optimal["verifier_components"]["hidden_correctness"] == 1.0
+assert info_optimal["verifier_components"]["anti_cheat_compliance"] == 1.0
+assert reward_safety == 0.0
+assert info_safety["execution_status"] == "safety_violation"

server/app.py

@@ -46,7 +46,7 @@ class ResetRequest(BaseModel):
 
 
 class TrainRequest(BaseModel):
-    preset: str = "smoke"
+    preset: str = "l4"
     model_name: Optional[str] = None
     output_dir: Optional[str] = None
     dataset_size: Optional[int] = None
@@ -54,10 +54,13 @@ class TrainRequest(BaseModel):
     batch_size: Optional[int] = None
     gradient_accumulation_steps: Optional[int] = None
     num_generations: Optional[int] = None
+    load_in_4bit: Optional[bool] = None
+    gradient_checkpointing: Optional[bool] = None
     evaluation_episodes: Optional[int] = None
     baseline_eval: Optional[bool] = None
     generator_mode: Optional[str] = None
     disable_wandb: Optional[bool] = None
+    save_merged_model: Optional[bool] = None
 
 
 class RunTrainedPolicyRequest(BaseModel):

server/runtime.py

@@ -161,7 +161,7 @@ class SpaceModelRegistry:
         return torch, AutoPeftModelForCausalLM, (AutoModelForCausalLM, AutoTokenizer)
 
     def _base_model_name(self) -> str:
-        return os.getenv("BASE_MODEL_NAME") or os.getenv("MODEL_NAME") or "unsloth/Llama-3.2-3B-Instruct"
+        return os.getenv("BASE_MODEL_NAME") or os.getenv("MODEL_NAME") or "Qwen/Qwen2.5-3B-Instruct"
 
     def _active_generation_stack(
         self,
@@ -194,7 +194,12 @@ class SpaceModelRegistry:
                 )
                 return self.status_payload()
 
-            dtype = torch.float16 if torch.cuda.is_available() else torch.float32
+            if torch.cuda.is_available() and torch.cuda.is_bf16_supported():
+                dtype = torch.bfloat16
+            elif torch.cuda.is_available():
+                dtype = torch.float16
+            else:
+                dtype = torch.float32
             tokenizer = AutoTokenizer.from_pretrained(base_model_name)
             if tokenizer.pad_token is None and tokenizer.eos_token is not None:
                 tokenizer.pad_token = tokenizer.eos_token
@@ -232,7 +237,12 @@ class SpaceModelRegistry:
             raise RuntimeError(f"Trained artifact directory does not exist: {artifact_dir}")
 
         with self._lock:
-            dtype = torch.float16 if torch.cuda.is_available() else torch.float32
+            if torch.cuda.is_available() and torch.cuda.is_bf16_supported():
+                dtype = torch.bfloat16
+            elif torch.cuda.is_available():
+                dtype = torch.float16
+            else:
+                dtype = torch.float32
             tokenizer = AutoTokenizer.from_pretrained(str(artifact_dir))
             if tokenizer.pad_token is None and tokenizer.eos_token is not None:
                 tokenizer.pad_token = tokenizer.eos_token

test.py

@@ -2,6 +2,7 @@ from __future__ import annotations
 
 from scripts.test_env import main as run_env_smoke
 from scripts.test_space_api import main as run_space_api_smoke
+from scripts.test_training_config import main as run_training_config_smoke
 from scripts.test_trace_logging import main as run_trace_logging_smoke
 from scripts.test_verifier import test_cases
 from verifier.verifier import verify
@@ -10,6 +11,7 @@ from verifier.verifier import verify
 def main() -> None:
     run_env_smoke()
     run_space_api_smoke()
+    run_training_config_smoke()
     run_trace_logging_smoke()
 
     reward, info = verify(

training/train_grpo.py

@@ -35,20 +35,21 @@ SMOKE_PREFERRED_PRECISION = "fp16"
 
 @dataclass
 class TrainingConfig:
-    model_name: str = "unsloth/Llama-3.2-3B-Instruct"
-    output_dir: str = "outputs_v3"
+    model_name: str = "Qwen/Qwen2.5-3B-Instruct"
+    output_dir: str = "outputs_l4"
     dataset_size: int = 200
     max_steps: int = 250
     batch_size: int = 1
     gradient_accumulation_steps: int = 8
-    num_generations: int = 8
+    num_generations: int = 4
     max_seq_length: int = 2048
     max_prompt_length: int = 1024
     max_completion_length: int = 512
     learning_rate: float = 5e-6
     lora_rank: int = 16
     lora_alpha: int = 16
-    disable_4bit: bool = False
+    load_in_4bit: bool = True
+    gradient_checkpointing: bool = True
     bf16: bool = False
     baseline_eval: bool = False
     evaluation_episodes: int = 20
@@ -60,6 +61,7 @@ class TrainingConfig:
     non_deterministic_generator: bool = False
     trace_logging_enabled: bool = True
     checkpoint_log_interval_steps: int = 10
+    save_merged_model: bool = False
 
     def to_dict(self) -> dict[str, Any]:
         return asdict(self)
@@ -67,6 +69,7 @@ class TrainingConfig:
 
 TRAINING_PRESETS: dict[str, dict[str, Any]] = {
     "smoke": {
+        "model_name": "Qwen/Qwen2.5-3B-Instruct",
         "dataset_size": 12,
         "max_steps": 6,
         "batch_size": 1,
@@ -75,12 +78,40 @@ TRAINING_PRESETS: dict[str, dict[str, Any]] = {
         "evaluation_episodes": 3,
         "baseline_eval": False,
         "disable_wandb": True,
-        "disable_4bit": True,
+        "load_in_4bit": False,
+        "gradient_checkpointing": False,
         "bf16": False,
         "output_dir": "outputs_smoke",
         "checkpoint_log_interval_steps": 2,
     },
-    "default": {},
+    "l4": {
+        "model_name": "Qwen/Qwen2.5-3B-Instruct",
+        "output_dir": "outputs_l4",
+        "batch_size": 1,
+        "gradient_accumulation_steps": 8,
+        "num_generations": 4,
+        "max_seq_length": 2048,
+        "max_prompt_length": 1024,
+        "max_completion_length": 512,
+        "lora_rank": 16,
+        "lora_alpha": 16,
+        "load_in_4bit": True,
+        "gradient_checkpointing": True,
+    },
+    "default": {
+        "model_name": "Qwen/Qwen2.5-3B-Instruct",
+        "output_dir": "outputs_l4",
+        "batch_size": 1,
+        "gradient_accumulation_steps": 8,
+        "num_generations": 4,
+        "max_seq_length": 2048,
+        "max_prompt_length": 1024,
+        "max_completion_length": 512,
+        "lora_rank": 16,
+        "lora_alpha": 16,
+        "load_in_4bit": True,
+        "gradient_checkpointing": True,
+    },
 }
 
 
@@ -165,7 +196,8 @@ def namespace_to_config(args: argparse.Namespace) -> TrainingConfig:
         learning_rate=args.learning_rate,
         lora_rank=args.lora_rank,
         lora_alpha=args.lora_alpha,
-        disable_4bit=args.disable_4bit,
+        load_in_4bit=not args.disable_4bit,
+        gradient_checkpointing=not getattr(args, "disable_gradient_checkpointing", False),
         bf16=args.bf16,
         baseline_eval=args.baseline_eval,
         evaluation_episodes=args.evaluation_episodes,
@@ -177,6 +209,7 @@ def namespace_to_config(args: argparse.Namespace) -> TrainingConfig:
         non_deterministic_generator=args.non_deterministic_generator,
         trace_logging_enabled=args.trace_logging_enabled,
         checkpoint_log_interval_steps=args.checkpoint_log_interval_steps,
+        save_merged_model=getattr(args, "save_merged_model", False),
     )
 
 
@@ -786,14 +819,11 @@ def resolve_precision_policy(config: TrainingConfig, torch: Any) -> dict[str, An
     if bf16_requested and not gpu_supports_bf16:
         raise RuntimeError("bf16 was requested, but the active GPU/runtime does not report BF16 support.")
 
-    output_dir_str = str(config.output_dir)
-    is_smoke_run = "outputs_smoke" in output_dir_str
-
-    if bf16_requested:
+    if bf16_requested or (use_cuda and gpu_supports_bf16):
         precision_mode = "bf16"
         model_dtype = torch.bfloat16
     elif use_cuda:
-        precision_mode = SMOKE_PREFERRED_PRECISION if is_smoke_run else "fp16"
+        precision_mode = SMOKE_PREFERRED_PRECISION
         model_dtype = torch.float16
     else:
         precision_mode = "fp32"
@@ -801,13 +831,7 @@ def resolve_precision_policy(config: TrainingConfig, torch: Any) -> dict[str, An
 
     use_bf16 = precision_mode == "bf16"
     use_fp16 = precision_mode == "fp16"
-    load_in_4bit = not config.disable_4bit
-
-    if use_cuda and load_in_4bit and (use_bf16 or use_fp16):
-        raise RuntimeError(
-            "4-bit loading with mixed-precision GRPO is disabled for this training path. "
-            "Set disable_4bit=true or use a full-precision CPU run."
-        )
+    load_in_4bit = bool(config.load_in_4bit)
 
     return {
         "use_cuda": use_cuda,
@@ -953,6 +977,7 @@ def run_training(
         target_modules=["q_proj", "k_proj", "v_proj", "o_proj"],
         lora_alpha=config.lora_alpha,
         lora_dropout=0.0,
+        use_gradient_checkpointing="unsloth" if config.gradient_checkpointing else False,
     )
     if not load_in_4bit:
         model = model.to(model_dtype)
@@ -967,9 +992,16 @@ def run_training(
             "reason": "load_in_4bit=True",
         }
 
-    critical_precision_audit = audit_critical_module_precision(model, model_dtype)
+    if load_in_4bit:
+        critical_precision_audit = {
+            "target_dtype": str(model_dtype),
+            "skipped": True,
+            "reason": "load_in_4bit=True",
+        }
+    else:
+        critical_precision_audit = audit_critical_module_precision(model, model_dtype)
     print(f"[training] critical precision audit {json.dumps(critical_precision_audit, sort_keys=True)}")
-    if critical_precision_audit["has_mismatch"]:
+    if not load_in_4bit and critical_precision_audit["has_mismatch"]:
         raise RuntimeError(
             "Critical projection modules remain in the wrong dtype before GRPOTrainer initialization. "
             f"Audit: {json.dumps(critical_precision_audit, sort_keys=True)}"
@@ -1114,7 +1146,10 @@ def run_training(
     )
     trainer.train()
 
-    model.save_pretrained(str(output_dir))
+    if config.save_merged_model and hasattr(model, "save_pretrained_merged"):
+        model.save_pretrained_merged(str(output_dir), tokenizer, save_method="merged_16bit")
+    else:
+        model.save_pretrained(str(output_dir))
     tokenizer.save_pretrained(str(output_dir))
 
     if config.baseline_eval:
@@ -1180,13 +1215,13 @@ def run_training(
 
 def build_parser() -> argparse.ArgumentParser:
     parser = argparse.ArgumentParser(description="GRPO training entrypoint for the ADAPT DSA environment.")
-    parser.add_argument("--model-name", default="unsloth/Llama-3.2-3B-Instruct")
-    parser.add_argument("--output-dir", default="outputs_v3")
+    parser.add_argument("--model-name", default="Qwen/Qwen2.5-3B-Instruct")
+    parser.add_argument("--output-dir", default="outputs_l4")
     parser.add_argument("--dataset-size", type=int, default=200)
     parser.add_argument("--max-steps", type=int, default=250)
     parser.add_argument("--batch-size", type=int, default=1)
     parser.add_argument("--gradient-accumulation-steps", type=int, default=8)
-    parser.add_argument("--num-generations", type=int, default=8)
+    parser.add_argument("--num-generations", type=int, default=4)
     parser.add_argument("--max-seq-length", type=int, default=2048)
     parser.add_argument("--max-prompt-length", type=int, default=1024)
     parser.add_argument("--max-completion-length", type=int, default=512)
@@ -1194,6 +1229,7 @@ def build_parser() -> argparse.ArgumentParser:
     parser.add_argument("--lora-rank", type=int, default=16)
     parser.add_argument("--lora-alpha", type=int, default=16)
     parser.add_argument("--disable-4bit", action="store_true")
+    parser.add_argument("--disable-gradient-checkpointing", action="store_true")
     parser.add_argument("--bf16", action="store_true")
     parser.add_argument("--baseline-eval", action="store_true")
     parser.add_argument("--evaluation-episodes", type=int, default=20)
@@ -1201,6 +1237,7 @@ def build_parser() -> argparse.ArgumentParser:
     parser.add_argument("--disable-wandb", action="store_true")
     parser.add_argument("--wandb-project", default="adapt-dsa-tutor")
     parser.add_argument("--wandb-run-name", default=None)
+    parser.add_argument("--save-merged-model", action="store_true")
     parser.add_argument("--trace-logging-enabled", action=argparse.BooleanOptionalAction, default=True)
     parser.add_argument("--checkpoint-log-interval-steps", type=int, default=10)
     parser.add_argument(

verifier/metrics.py

@@ -2,6 +2,13 @@ from __future__ import annotations
 
 from typing import Any
 
+STEP_DISCOUNTS = {1: 1.0, 2: 0.85, 3: 0.70}
+TERMINAL_ZERO_STATUSES = {"syntax_error", "safety_violation", "timeout"}
+
+
+def step_discount(step_number: int) -> float:
+    return STEP_DISCOUNTS.get(int(step_number), 0.70)
+
 
 def compute_reward(
     pass_rate: float,
@@ -9,26 +16,21 @@ def compute_reward(
     execution_status: str,
     format_compliance: float,
 ) -> float:
-    """
-    Clean, interpretable reward signal for GRPO training.
-    """
     del format_compliance
-
-    step_discount = 1.0 if step_number == 1 else (0.85 if step_number == 2 else 0.70)
-    correctness = pass_rate
-
-    if execution_status == "timeout":
-        return 0.0
-    if execution_status == "syntax_error":
+    correctness = max(0.0, min(float(pass_rate), 1.0))
+    if execution_status in TERMINAL_ZERO_STATUSES:
         return 0.0
-
-    reward = correctness * step_discount
+    reward = correctness * step_discount(step_number)
     return round(min(max(reward, 0.0), 1.0), 4)
 
 
 def compute_pass_rate(
     results: list[dict[str, Any]],
     step_number: int = 1,
+    *,
+    syntax_ok: bool = True,
+    safety_ok: bool = True,
+    precheck_status: str | None = None,
 ) -> tuple[float, dict[str, Any]]:
     total = len(results)
     hidden_results = [result for result in results if result.get("visibility") == "hidden"]
@@ -36,15 +38,14 @@ def compute_pass_rate(
 
     hidden_total = len(hidden_results)
     visible_total = len(visible_results)
+    hidden_passed = sum(1 for result in hidden_results if result.get("passed"))
+    visible_passed = sum(1 for result in visible_results if result.get("passed"))
+    passed = sum(1 for result in results if result.get("passed"))
 
-    hidden_passed = sum(1 for result in hidden_results if result["passed"])
-    visible_passed = sum(1 for result in visible_results if result["passed"])
-    passed = sum(1 for result in results if result["passed"])
-
-    timeout_count = sum(1 for result in results if result["status"] == "timeout")
-    runtime_error_count = sum(1 for result in results if result["status"] == "runtime_error")
-    invalid_output_count = sum(1 for result in results if result["status"] == "invalid_output_format")
-    wrong_answer_count = sum(1 for result in results if result["status"] == "wrong_answer")
+    timeout_count = sum(1 for result in results if result.get("status") == "timeout")
+    runtime_error_count = sum(1 for result in results if result.get("status") == "runtime_error")
+    invalid_output_count = sum(1 for result in results if result.get("status") == "invalid_output_format")
+    wrong_answer_count = sum(1 for result in results if result.get("status") == "wrong_answer")
     format_ok_count = sum(1 for result in results if result.get("format_ok", False))
 
     hidden_pass_rate = hidden_passed / hidden_total if hidden_total else 0.0
@@ -52,7 +53,13 @@ def compute_pass_rate(
     pass_rate = hidden_pass_rate if hidden_total else (passed / total if total else 0.0)
     format_compliance = format_ok_count / total if total else 0.0
 
-    if timeout_count:
+    if not syntax_ok:
+        execution_status = "syntax_error"
+    elif not safety_ok:
+        execution_status = "safety_violation"
+    elif precheck_status and precheck_status not in {"ready", "completed"}:
+        execution_status = precheck_status
+    elif timeout_count:
         execution_status = "timeout"
     elif runtime_error_count:
         execution_status = "runtime_error"
@@ -70,6 +77,15 @@ def compute_pass_rate(
         format_compliance=format_compliance,
     )
 
+    verifier_components = {
+        "hidden_correctness": round(hidden_pass_rate, 4),
+        "visible_correctness": round(visible_pass_rate, 4),
+        "format_compliance": round(format_compliance, 4),
+        "runtime_reliability": round(0.0 if timeout_count or runtime_error_count else 1.0, 4),
+        "anti_cheat_compliance": round(1.0 if syntax_ok and safety_ok else 0.0, 4),
+        "step_discount": round(step_discount(step_number), 4),
+    }
+
     return reward, {
         "passed": passed,
         "total": total,
@@ -85,10 +101,48 @@ def compute_pass_rate(
         "invalid_output_count": invalid_output_count,
         "wrong_answer_count": wrong_answer_count,
         "format_compliance": round(format_compliance, 4),
+        "syntax_valid": bool(syntax_ok),
+        "safety_valid": bool(safety_ok),
         "execution_status": execution_status,
         "reward_components": {
             "correctness": round(float(pass_rate), 4),
-            "step_discount": 1.0 if step_number == 1 else (0.85 if step_number == 2 else 0.70),
+            "step_discount": round(step_discount(step_number), 4),
             "reward": reward,
         },
+        "verifier_components": verifier_components,
+    }
+
+
+def compute_episode_reward(
+    *,
+    pass_rate: float,
+    step_number: int,
+    execution_status: str,
+    previous_pass_rate: float,
+    done: bool,
+    efficiency_score: float,
+    optimization_target_met: bool,
+) -> tuple[float, dict[str, float]]:
+    discount = step_discount(step_number)
+    clipped_pass_rate = max(0.0, min(float(pass_rate), 1.0))
+    clipped_efficiency = max(0.0, min(float(efficiency_score), 1.0))
+    progress_delta = max(0.0, clipped_pass_rate - max(0.0, min(float(previous_pass_rate), 1.0)))
+
+    if execution_status in TERMINAL_ZERO_STATUSES:
+        reward = 0.0
+    elif clipped_pass_rate == 1.0:
+        reward = round(discount * (0.6 + 0.4 * clipped_efficiency), 4)
+        if not optimization_target_met and not done:
+            reward = min(reward, 0.94)
+    elif done:
+        reward = 0.0
+    else:
+        reward = round(0.1 * progress_delta, 4)
+
+    return reward, {
+        "correctness": round(clipped_pass_rate, 4),
+        "efficiency_score": round(clipped_efficiency, 4),
+        "step_discount": round(discount, 4),
+        "progress_delta": round(progress_delta, 4),
+        "reward": round(float(reward), 4),
     }

verifier/sandbox.py

@@ -1,7 +1,89 @@
 from __future__ import annotations
 
+import ast
+from typing import Any
+
 from env.executor import run_code as execute_submission
 
+FORBIDDEN_IMPORTS = {
+    "ctypes",
+    "os",
+    "pathlib",
+    "resource",
+    "shutil",
+    "signal",
+    "socket",
+    "subprocess",
+}
+FORBIDDEN_CALLS = {
+    "__import__",
+    "breakpoint",
+    "compile",
+    "eval",
+    "exec",
+    "open",
+}
+
+
+def _call_name(node: ast.AST) -> str:
+    if isinstance(node, ast.Name):
+        return node.id
+    if isinstance(node, ast.Attribute):
+        parent = _call_name(node.value)
+        return f"{parent}.{node.attr}" if parent else node.attr
+    return ""
+
+
+def validate_code(code: str) -> dict[str, Any]:
+    try:
+        tree = ast.parse(code)
+    except SyntaxError as exc:
+        return {
+            "syntax_ok": False,
+            "safety_ok": False,
+            "execution_status": "syntax_error",
+            "error": str(exc),
+        }
+
+    for node in ast.walk(tree):
+        if isinstance(node, ast.Import):
+            for alias in node.names:
+                root_name = alias.name.split(".", 1)[0]
+                if root_name in FORBIDDEN_IMPORTS:
+                    return {
+                        "syntax_ok": True,
+                        "safety_ok": False,
+                        "execution_status": "safety_violation",
+                        "error": f"Forbidden import: {root_name}",
+                    }
+
+        if isinstance(node, ast.ImportFrom):
+            root_name = (node.module or "").split(".", 1)[0]
+            if root_name in FORBIDDEN_IMPORTS:
+                return {
+                    "syntax_ok": True,
+                    "safety_ok": False,
+                    "execution_status": "safety_violation",
+                    "error": f"Forbidden import: {root_name}",
+                }
+
+        if isinstance(node, ast.Call):
+            fn_name = _call_name(node.func)
+            if fn_name in FORBIDDEN_CALLS:
+                return {
+                    "syntax_ok": True,
+                    "safety_ok": False,
+                    "execution_status": "safety_violation",
+                    "error": f"Forbidden call: {fn_name}",
+                }
+
+    return {
+        "syntax_ok": True,
+        "safety_ok": True,
+        "execution_status": "ready",
+        "error": "",
+    }
+
 
-def run_code(code: str, stdin: str, timeout: int = 1) -> dict[str, object]:
+def run_code(code: str, stdin: str, timeout: int | float = 1) -> dict[str, object]:
     return execute_submission(code, stdin, timeout_seconds=timeout)

verifier/verifier.py

@@ -4,13 +4,36 @@ from typing import Any
 
 from verifier.complexity import analyze_code_complexity
 from verifier.metrics import compute_pass_rate
-from verifier.sandbox import run_code
+from verifier.sandbox import run_code, validate_code
 
 
-def verify(code: str, test_cases: list[dict[str, Any]] | list[tuple[str, str]]) -> tuple[float, dict[str, Any]]:
-    results: list[dict[str, Any]] = []
+def verify(
+    code: str,
+    test_cases: list[dict[str, Any]] | list[tuple[str, str]],
+    *,
+    step_number: int = 1,
+) -> tuple[float, dict[str, Any]]:
+    precheck = validate_code(code)
     complexity = analyze_code_complexity(code)
 
+    if not precheck["syntax_ok"] or not precheck["safety_ok"]:
+        reward, metrics = compute_pass_rate(
+            [],
+            step_number=step_number,
+            syntax_ok=bool(precheck["syntax_ok"]),
+            safety_ok=bool(precheck["safety_ok"]),
+            precheck_status=str(precheck["execution_status"]),
+        )
+        feedback = _build_feedback(metrics, error=str(precheck["error"]))
+        return reward, {
+            **metrics,
+            **complexity,
+            "feedback": feedback,
+            "results": [],
+            "error": str(precheck["error"]),
+        }
+
+    results: list[dict[str, Any]] = []
     for index, test_case in enumerate(test_cases):
         if isinstance(test_case, dict):
             stdin = str(test_case.get("input", ""))
@@ -51,13 +74,15 @@ def verify(code: str, test_cases: list[dict[str, Any]] | list[tuple[str, str]])
                 "input": stdin if is_visible else "",
                 "timed_out": timed_out,
                 "exit_code": exit_code,
+                "duration_ms": execution.get("duration_ms", 0.0),
+                "sandboxed": bool(execution.get("sandboxed", False)),
+                "sandbox_mode": execution.get("sandbox_mode", "portable"),
                 "visibility": "visible" if is_visible else "hidden",
             }
         )
 
-    reward, metrics = compute_pass_rate(results)
+    reward, metrics = compute_pass_rate(results, step_number=step_number)
     feedback = _build_feedback(metrics)
-
     return reward, {
         **metrics,
         **complexity,
@@ -66,13 +91,18 @@ def verify(code: str, test_cases: list[dict[str, Any]] | list[tuple[str, str]])
     }
 
 
-def _build_feedback(metrics: dict[str, Any]) -> str:
-    if metrics["execution_status"] == "timeout":
+def _build_feedback(metrics: dict[str, Any], *, error: str = "") -> str:
+    execution_status = str(metrics.get("execution_status", "unknown"))
+    if execution_status == "syntax_error":
+        return f"Submission has a syntax error. {error}".strip()
+    if execution_status == "safety_violation":
+        return f"Submission violated the sandbox policy. {error}".strip()
+    if execution_status == "timeout":
         return "Submission timed out on one or more hidden evaluation tests."
-    if metrics["execution_status"] == "runtime_error":
+    if execution_status == "runtime_error":
         return "Submission raised a runtime error on one or more hidden evaluation tests."
-    if metrics["execution_status"] == "invalid_output_format":
+    if execution_status == "invalid_output_format":
         return "Submission completed but produced invalid output format on one or more tests."
-    if metrics["execution_status"] == "wrong_answer":
+    if execution_status == "wrong_answer":
         return "Submission ran successfully but returned an incorrect answer on one or more hidden tests."
     return f"All hidden tests passed. Pass rate: {metrics['pass_rate']:.2f}"