Upload folder using huggingface_hub

Dockerfile

@@ -0,0 +1,136 @@
+# Copyright (c) Meta Platforms, Inc. and affiliates.
+# All rights reserved.
+#
+# This source code is licensed under the BSD-style license found in the
+# LICENSE file in the root directory of this source tree.
+
+# Multi-stage build using openenv-base
+# This Dockerfile works for both in-repo and standalone builds.
+
+ARG BASE_IMAGE=ghcr.io/meta-pytorch/openenv-base:latest
+FROM ${BASE_IMAGE} AS builder
+
+WORKDIR /app
+
+# Build argument to control whether we're building standalone or in-repo
+ARG BUILD_MODE=in-repo
+
+# Install system dependencies for Julia
+RUN apt-get update && apt-get install -y --no-install-recommends \
+    git \
+    curl \
+    ca-certificates \
+    && rm -rf /var/lib/apt/lists/*
+
+# Install juliaup and Julia 1.10
+RUN curl -fsSL https://install.julialang.org | sh -s -- --yes --default-channel 1.10
+
+# Add Julia to PATH
+ENV PATH="/root/.juliaup/bin:${PATH}"
+
+# Configure juliaup to prevent runtime network calls
+# Set the default channel explicitly and mark juliaup as initialized
+ENV JULIAUP_CHANNEL="1.10"
+RUN juliaup default 1.10
+
+# Verify Julia installation
+RUN julia --version
+
+# Precompile commonly used Julia packages (Test is built-in)
+RUN julia -e 'using Test; println("Julia Test module ready")'
+
+# Copy environment code (always at root of build context)
+COPY . /app/env
+
+# Ensure uv is available
+RUN if ! command -v uv >/dev/null 2>&1; then \
+        curl -LsSf https://astral.sh/uv/install.sh | sh && \
+        mv /root/.local/bin/uv /usr/local/bin/uv && \
+        mv /root/.local/bin/uvx /usr/local/bin/uvx; \
+    fi
+
+WORKDIR /app/env
+
+# Install dependencies using uv sync
+RUN --mount=type=cache,target=/root/.cache/uv \
+    if [ -f uv.lock ]; then \
+        uv sync --frozen --no-install-project --no-editable; \
+    else \
+        uv sync --no-install-project --no-editable; \
+    fi
+
+RUN --mount=type=cache,target=/root/.cache/uv \
+    if [ -f uv.lock ]; then \
+        uv sync --frozen --no-editable; \
+    else \
+        uv sync --no-editable; \
+    fi
+
+# Final runtime stage
+FROM ${BASE_IMAGE}
+
+WORKDIR /app
+
+# Install Julia runtime dependencies
+RUN apt-get update && apt-get install -y --no-install-recommends \
+    curl \
+    ca-certificates \
+    && rm -rf /var/lib/apt/lists/*
+
+# Copy Julia installation from builder
+# juliaup stores the manager in ~/.juliaup and actual Julia versions in ~/.julia/juliaup
+COPY --from=builder /root/.juliaup /root/.juliaup
+COPY --from=builder /root/.julia /root/.julia
+
+# Create a direct symlink to Julia binary, bypassing the juliaup wrapper
+# This prevents juliaup from trying to check for updates at runtime
+# The juliaup wrapper (at ~/.juliaup/bin/julia) triggers network calls;
+# the actual Julia binary is in ~/.julia/juliaup/julia-*/bin/julia
+RUN echo "Looking for Julia binary..." && \
+    ls -la /root/.julia/juliaup/ && \
+    JULIA_BIN=$(find /root/.julia/juliaup -type f -executable -name "julia" 2>/dev/null | head -1) && \
+    if [ -z "$JULIA_BIN" ]; then \
+        echo "ERROR: Could not find Julia binary. Contents of .julia/juliaup:" && \
+        find /root/.julia -name "julia*" -ls && \
+        exit 1; \
+    fi && \
+    ln -sf "$JULIA_BIN" /usr/local/bin/julia && \
+    echo "Linked Julia from: $JULIA_BIN"
+
+# Verify Julia works without network
+RUN julia --version
+
+ENV PATH="/usr/local/bin:${PATH}"
+
+# Disable juliaup's version update checks (as fallback if juliaup wrapper is used)
+ENV JULIAUP_CHANNEL="1.10"
+
+# Copy the virtual environment from builder
+COPY --from=builder /app/env/.venv /app/.venv
+
+# Copy the environment code
+COPY --from=builder /app/env /app/env
+
+# Set PATH to use the virtual environment
+ENV PATH="/app/.venv/bin:$PATH"
+
+# Set PYTHONPATH so imports work correctly
+ENV PYTHONPATH="/app/env:$PYTHONPATH"
+
+# Julia process pool configuration (can be overridden at runtime)
+ENV JULIA_MAX_WORKERS=8
+ENV JULIA_EXECUTION_TIMEOUT=120
+ENV JULIA_LOG_FILE=/tmp/julia_env.log
+ENV JULIA_LOG_LEVEL=INFO
+ENV PYTHONUNBUFFERED=1
+
+# Health check
+HEALTHCHECK --interval=30s --timeout=5s --start-period=30s --retries=3 \
+    CMD python -c "import urllib.request; urllib.request.urlopen('http://localhost:8000/health')" || exit 1
+
+# Expose port
+EXPOSE 8000
+
+# Run the FastAPI server
+ENV ENABLE_WEB_INTERFACE=true
+CMD ["sh", "-c", "cd /app/env && uvicorn server.app:app --host 0.0.0.0 --port 8000"]

README.md

@@ -1,10 +1,234 @@
 ---
-title: Julia Env Test
-emoji: 👀
-colorFrom: yellow
-colorTo: purple
+title: Julia Environment Server
+emoji: 🔬
+colorFrom: purple
+colorTo: green
 sdk: docker
 pinned: false
+app_port: 8000
+base_path: /web
+tags:
+  - openenv
+  - julia
 ---
 
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+# Julia Environment
+
+A Julia code execution environment that runs Julia code with test result tracking and reward calculation. Perfect for reinforcement learning training with Julia programming tasks.
+
+## Quick Start
+
+The simplest way to use the Julia environment is through the `JuliaEnv` class:
+
+```python
+from envs.julia_env import JuliaAction, JuliaEnv
+
+try:
+    # Create environment from Docker image
+    julia_env = JuliaEnv.from_docker_image("julia-env:latest")
+
+    # Reset
+    result = julia_env.reset()
+    print(f"Reset complete: exit_code={result.observation.exit_code}")
+
+    # Execute Julia code with tests
+    action = JuliaAction(
+        core_code="""
+        function multiply(a, b)
+            return a * b
+        end
+        """,
+        test_code="""
+        using Test
+        @test multiply(3, 4) == 12
+        @test multiply(5, 6) == 30
+        """
+    )
+
+    result = julia_env.step(action)
+    print(f"Tests passed: {result.observation.tests_passed}")
+    print(f"Tests failed: {result.observation.tests_failed}")
+    print(f"Code compiles: {result.observation.code_compiles}")
+    print(f"Reward: {result.reward}")
+
+finally:
+    # Always clean up
+    julia_env.close()
+```
+
+That's it! The `JuliaEnv.from_docker_image()` method handles:
+- Starting the Docker container
+- Waiting for the server to be ready
+- Connecting to the environment
+- Container cleanup when you call `close()`
+
+## Building the Docker Image
+
+Before using the environment, you need to build the Docker image:
+
+```bash
+# From the julia_env directory
+cd envs/julia_env
+docker build -t julia-env:latest -f server/Dockerfile .
+```
+
+## Environment Details
+
+### Action
+
+**JuliaAction**: Contains two fields for Julia code execution
+- `core_code` (str) - The main Julia code to execute (e.g., function definitions)
+- `test_code` (str) - Test code using Julia's `Test` module (e.g., `@test` statements)
+
+### Observation
+
+**JuliaObservation**: Contains the execution results and test outcomes
+- `stdout` (str) - Standard output from Julia execution
+- `stderr` (str) - Standard error from Julia execution
+- `exit_code` (int) - Exit code (0 for success, non-zero for errors)
+- `tests_passed` (int) - Number of tests that passed
+- `tests_failed` (int) - Number of tests that failed
+- `code_compiles` (bool) - Whether the core code compiled/executed successfully
+
+### State
+
+**JuliaState**: Tracks episode execution state
+- `episode_id` (str) - Unique identifier for the episode
+- `step_count` (int) - Number of steps taken in the episode
+- `last_exit_code` (int) - Exit code from the last execution
+- `last_code_compiles` (bool) - Whether the last code compiled successfully
+- `total_tests_passed` (int) - Cumulative number of tests passed in the episode
+- `total_tests_failed` (int) - Cumulative number of tests failed in the episode
+
+### Reward Calculation
+
+The environment calculates rewards based on execution success and test results:
+- Code compiles successfully: Base reward
+- Tests pass: Additional reward per test
+- Tests fail or code doesn't compile: Negative reward
+
+See `server/julia_transforms.py` for detailed reward logic.
+
+## Features
+
+- ✅ Execute Julia code in isolated subprocess
+- ✅ Parse Julia `Test` module output (tests passed/failed)
+- ✅ Calculate rewards based on execution results and test outcomes
+- ✅ Safety transforms for output truncation (prevents excessive output)
+- ✅ Docker support for reproducible execution
+- ✅ Compatible with GRPO and other RL training frameworks
+
+## Advanced Usage
+
+### Connecting to an Existing Server
+
+If you already have a Julia environment server running, you can connect directly:
+
+```python
+from envs.julia_env import JuliaEnv, JuliaAction
+
+# Connect to existing server
+julia_env = JuliaEnv(base_url="http://localhost:8000")
+
+# Use as normal
+result = julia_env.reset()
+result = julia_env.step(JuliaAction(
+    core_code="println(2 + 2)",
+    test_code=""
+))
+```
+
+Note: When connecting to an existing server, `julia_env.close()` will NOT stop the server.
+
+### Custom Timeout
+
+The Julia environment uses a longer timeout (180s) by default to accommodate Julia compilation and execution:
+
+```python
+# Custom timeout (in seconds)
+julia_env = JuliaEnv(base_url="http://localhost:8000", message_timeout_s=300.0)
+```
+
+### Running with Docker Directly
+
+```bash
+# Run with default settings (port 8000)
+docker run -d -p 8000:8000 --name julia-env julia-env:latest
+
+# Check health
+curl http://localhost:8000/health
+
+# View logs
+docker logs -f julia-env
+```
+
+## Example Code
+
+Here's a more complex example demonstrating test-driven development:
+
+```python
+from envs.julia_env import JuliaAction, JuliaEnv
+
+julia_env = JuliaEnv.from_docker_image("julia-env:latest")
+
+try:
+    # Reset the environment
+    julia_env.reset()
+
+    # Step 1: Define a function with tests
+    action = JuliaAction(
+        core_code="""
+        function fibonacci(n)
+            if n <= 1
+                return n
+            end
+            return fibonacci(n-1) + fibonacci(n-2)
+        end
+        """,
+        test_code="""
+        using Test
+        @test fibonacci(0) == 0
+        @test fibonacci(1) == 1
+        @test fibonacci(5) == 5
+        @test fibonacci(10) == 55
+        """
+    )
+
+    result = julia_env.step(action)
+    print(f"Step 1 - Tests passed: {result.observation.tests_passed}/4")
+    print(f"Step 1 - Reward: {result.reward}")
+
+    # Get current state
+    state = julia_env.state()
+    print(f"Total tests passed so far: {state.total_tests_passed}")
+    print(f"Total tests failed so far: {state.total_tests_failed}")
+
+finally:
+    julia_env.close()
+```
+
+## Compatibility
+
+This environment is compatible with:
+- **torchforge**: For GRPO training with Julia tasks
+- **TRL**: Hugging Face's RL library
+- **Any OpenEnv-compatible RL framework**
+
+## Development
+
+For local development without Docker:
+
+```bash
+# Install Julia 1.10+ from https://julialang.org/downloads/
+
+# Install Python dependencies
+cd envs/julia_env
+pip install -e .
+
+# Run the server locally
+uvicorn julia_env.server.app:app --host 0.0.0.0 --port 8000
+```
+
+## License
+
+This source code is licensed under the BSD-style license found in the LICENSE file in the root directory of this source tree.

__init__.py

@@ -0,0 +1,13 @@
+# Copyright (c) Meta Platforms, Inc. and affiliates.
+# All rights reserved.
+#
+# This source code is licensed under the BSD-style license found in the
+# LICENSE file in the root directory of this source tree.
+
+"""Julia Environment - Code execution environment for RL training."""
+
+from .client import JuliaEnv
+from .models import JuliaAction, JuliaObservation, JuliaState
+
+__all__ = ["JuliaAction", "JuliaObservation", "JuliaState", "JuliaEnv"]
+

client.py

@@ -0,0 +1,161 @@
+# Copyright (c) Meta Platforms, Inc. and affiliates.
+# All rights reserved.
+#
+# This source code is licensed under the BSD-style license found in the
+# LICENSE file in the root directory of this source tree.
+
+"""
+JuliaEnv
+--------
+Client-side wrapper for the Julia environment server.
+
+This client maintains a persistent WebSocket connection to the environment
+server, enabling efficient multi-step interactions with lower latency.
+
+- Users instantiate JuliaEnv with a base_url provided by the higher-level
+  vector/orchestration layer.
+- Environment authors ship the Docker image that serves the API.
+"""
+
+from __future__ import annotations
+
+from openenv.core.client_types import StepResult
+from openenv.core.env_client import EnvClient
+
+from .models import JuliaAction, JuliaObservation, JuliaState
+
+
+class JuliaEnv(EnvClient[JuliaAction, JuliaObservation, JuliaState]):
+    """
+    WebSocket client for the Julia Environment.
+
+    This client connects to a JuliaEnvironment server and provides
+    methods to interact with it: reset(), step(), and state access.
+
+    The default message timeout is set to 180 seconds to accommodate:
+    - Server execution timeout: 120s
+    - Process pool worker wait: 30s
+    - Network overhead: 30s buffer
+
+    Example:
+        >>> # Connect to a running server
+        >>> client = JuliaEnv(base_url="http://localhost:8000")
+        >>> result = client.reset()
+        >>> print(result.observation.stdout)
+        >>>
+        >>> # Execute Julia code
+        >>> action = JuliaAction(
+        ...     core_code='''
+        ...     function multiply(a, b)
+        ...         return a * b
+        ...     end
+        ...     ''',
+        ...     test_code='''
+        ...     using Test
+        ...     @test multiply(3, 4) == 12
+        ...     '''
+        ... )
+        >>> result = client.step(action)
+        >>> print(result.observation.tests_passed)  # 1
+        >>> print(result.reward)
+
+    Example with Docker:
+        >>> # Automatically start container and connect
+        >>> client = JuliaEnv.from_docker_image("julia-env:latest")
+        >>> result = client.reset()
+        >>> result = client.step(JuliaAction(core_code="println(2 + 2)", test_code=""))
+        >>> print(result.observation.stdout)  # "4\\n"
+        >>> client.close()
+    """
+
+    # Override default timeout to accommodate Julia execution + worker wait
+    DEFAULT_MESSAGE_TIMEOUT = 180.0  # 120s execution + 30s worker wait + 30s buffer
+
+    def __init__(
+        self,
+        base_url: str,
+        connect_timeout_s: float = 10.0,
+        message_timeout_s: float | None = None,
+        **kwargs,
+    ):
+        """
+        Initialize JuliaEnv client with appropriate timeout.
+
+        Args:
+            base_url: Base URL of the Julia environment server
+            connect_timeout_s: Timeout for establishing WebSocket connection
+            message_timeout_s: Timeout for receiving responses (default: 180.0)
+            **kwargs: Additional arguments passed to EnvClient
+        """
+        if message_timeout_s is None:
+            message_timeout_s = self.DEFAULT_MESSAGE_TIMEOUT
+        super().__init__(
+            base_url,
+            connect_timeout_s=connect_timeout_s,
+            message_timeout_s=message_timeout_s,
+            **kwargs,
+        )
+
+    # --- EnvClient abstract hooks ---
+
+    def _step_payload(self, action: JuliaAction) -> dict:
+        """
+        Convert JuliaAction to JSON payload for step request.
+
+        Args:
+            action: JuliaAction instance
+
+        Returns:
+            Dictionary representation suitable for JSON encoding
+        """
+        return {
+            "core_code": action.core_code,
+            "test_code": action.test_code,
+        }
+
+    def _parse_result(self, payload: dict) -> StepResult[JuliaObservation]:
+        """
+        Parse server response into StepResult[JuliaObservation].
+
+        Args:
+            payload: JSON response from server
+
+        Returns:
+            StepResult with JuliaObservation
+        """
+        obs_data = payload.get("observation", {})
+        observation = JuliaObservation(
+            stdout=obs_data.get("stdout", ""),
+            stderr=obs_data.get("stderr", ""),
+            exit_code=obs_data.get("exit_code", 0),
+            tests_passed=obs_data.get("tests_passed", 0),
+            tests_failed=obs_data.get("tests_failed", 0),
+            code_compiles=obs_data.get("code_compiles", True),
+            done=payload.get("done", False),
+            reward=payload.get("reward"),
+        )
+
+        return StepResult[JuliaObservation](
+            observation=observation,
+            reward=payload.get("reward"),
+            done=payload.get("done", False),
+        )
+
+    def _parse_state(self, payload: dict) -> JuliaState:
+        """
+        Parse server response into JuliaState object.
+
+        Args:
+            payload: JSON response from /state endpoint
+
+        Returns:
+            JuliaState object with episode metadata
+        """
+        return JuliaState(
+            episode_id=payload.get("episode_id"),
+            step_count=payload.get("step_count", 0),
+            last_exit_code=payload.get("last_exit_code", 0),
+            last_code_compiles=payload.get("last_code_compiles", True),
+            total_tests_passed=payload.get("total_tests_passed", 0),
+            total_tests_failed=payload.get("total_tests_failed", 0),
+        )

models.py

@@ -0,0 +1,67 @@
+# Copyright (c) Meta Platforms, Inc. and affiliates.
+# All rights reserved.
+#
+# This source code is licensed under the BSD-style license found in the
+# LICENSE file in the root directory of this source tree.
+
+"""
+Data models for the Julia Environment.
+
+The Julia environment executes Julia code and provides feedback through
+compilation and unit test results.
+"""
+
+from openenv.core.env_server import Action, Observation, State
+
+
+class JuliaAction(Action):
+    """
+    Action for the Julia environment - code to execute.
+
+    Attributes:
+        core_code: Core Julia code to execute
+        test_code: Optional test code to execute. If not provided, only core_code runs.
+    """
+
+    core_code: str
+    test_code: str | None = None
+
+
+class JuliaObservation(Observation):
+    """
+    Observation from the Julia environment - execution results.
+
+    Attributes:
+        stdout: Standard output from Julia execution
+        stderr: Standard error from Julia execution
+        exit_code: Exit code (0 = success, non-zero = error)
+        tests_passed: Number of tests passed (if tests were run)
+        tests_failed: Number of tests failed (if tests were run)
+        code_compiles: Whether the core code compiled/executed successfully
+    """
+
+    stdout: str = ""
+    stderr: str = ""
+    exit_code: int = 0
+    tests_passed: int = 0
+    tests_failed: int = 0
+    code_compiles: bool = True
+
+
+class JuliaState(State):
+    """
+    State for Julia environment.
+
+    Attributes:
+        episode_id: Unique episode identifier
+        step_count: Number of steps taken in episode
+        last_exit_code: Exit code from last execution
+        last_code_compiles: Whether the last code compiled successfully
+        total_tests_passed: Cumulative tests passed in episode
+        total_tests_failed: Cumulative tests failed in episode
+    """
+
+    last_exit_code: int = 0
+    last_code_compiles: bool = True
+    total_tests_passed: int = 0
+    total_tests_failed: int = 0

openenv.yaml

@@ -0,0 +1,5 @@
+name: julia_env
+version: "0.1.0"
+description: "Julia code execution environment for OpenEnv with test tracking and reward calculation"
+action: JuliaAction
+observation: JuliaObservation

pyproject.toml

@@ -0,0 +1,40 @@
+# Copyright (c) Meta Platforms, Inc. and affiliates.
+# All rights reserved.
+#
+# This source code is licensed under the BSD-style license found in the
+# LICENSE file in the root directory of this source tree.
+
+[build-system]
+requires = ["setuptools>=45", "wheel"]
+build-backend = "setuptools.build_meta"
+
+[project]
+name = "openenv-julia_env"
+version = "0.1.0"
+description = "Julia Environment for OpenEnv - Julia code execution with test tracking and reward calculation"
+requires-python = ">=3.10"
+dependencies = [
+    "openenv-core[core] @ git+https://github.com/meta-pytorch/OpenEnv.git@main",
+    "fastapi>=0.115.0",
+    "pydantic>=2.0.0",
+    "uvicorn[standard]>=0.24.0",
+    "requests>=2.31.0",
+]
+
+[project.optional-dependencies]
+dev = [
+    "pytest>=8.0.0",
+    "pytest-cov>=4.0.0",
+]
+
+[project.scripts]
+server = "julia_env.server.app:main"
+
+
+[tool.setuptools]
+packages = ["julia_env", "julia_env.server"]
+package-dir = { "julia_env" = ".", "julia_env.server" = "server" }
+
+[tool.setuptools.package-data]
+julia_env = ["**/*.yaml", "**/*.yml", "**/*.jl"]
+"julia_env.server" = ["*.jl"]

server/__init__.py

@@ -0,0 +1,17 @@
+# Copyright (c) Meta Platforms, Inc. and affiliates.
+# All rights reserved.
+#
+# This source code is licensed under the BSD-style license found in the
+# LICENSE file in the root directory of this source tree.
+
+"""Julia Environment Server."""
+
+# Support both in-repo and standalone imports
+try:
+    from .julia_codeact_env import JuliaCodeActEnv
+    from .julia_transforms import create_safe_julia_transform
+except ImportError:
+    from server.julia_codeact_env import JuliaCodeActEnv
+    from server.julia_transforms import create_safe_julia_transform
+
+__all__ = ["JuliaCodeActEnv", "create_safe_julia_transform"]

server/app.py

@@ -0,0 +1,219 @@
+# Copyright (c) Meta Platforms, Inc. and affiliates.
+# All rights reserved.
+#
+# This source code is licensed under the BSD-style license found in the
+# LICENSE file in the root directory of this source tree.
+
+"""
+FastAPI application for the Julia Environment with concurrent execution support.
+
+This module creates an HTTP server that exposes the JuliaCodeActEnv
+over HTTP and WebSocket endpoints with optimized async execution for handling
+multiple concurrent requests efficiently.
+
+Features:
+- WebSocket support for persistent sessions (required by OpenEnv clients)
+- Julia Process Pool for 50-100x speedup on repeated executions
+- Automatic error recovery and retry logic
+- Comprehensive logging to file and console
+
+Environment Variables:
+- JULIA_MAX_WORKERS: Number of concurrent Julia executions (default: 8)
+- JULIA_EXECUTION_TIMEOUT: Timeout in seconds (default: 120)
+- JULIA_LOG_FILE: Log file path (default: /tmp/julia_env.log)
+- JULIA_LOG_LEVEL: Log level (default: INFO)
+- ENABLE_WEB_INTERFACE: Enable web interface (default: false)
+
+Usage:
+    # Development (with auto-reload):
+    uvicorn server.app:app --reload --host 0.0.0.0 --port 8000
+
+    # Production:
+    uvicorn server.app:app --host 0.0.0.0 --port 8000
+
+    # Or run directly:
+    python -m server.app
+"""
+
+import atexit
+import logging
+import os
+import sys
+from logging.handlers import RotatingFileHandler
+
+# Support both in-repo and standalone imports
+try:
+    # In-repo imports (when running from OpenEnv repository)
+    from openenv.core.env_server.http_server import create_app
+    from ..models import JuliaAction, JuliaObservation
+    from .julia_codeact_env import JuliaCodeActEnv
+    from .julia_executor import JuliaExecutor
+except ImportError:
+    # Standalone imports (when environment is standalone)
+    from openenv.core.env_server.http_server import create_app
+    from models import JuliaAction, JuliaObservation
+    from server.julia_codeact_env import JuliaCodeActEnv
+    from server.julia_executor import JuliaExecutor
+
+# Configuration from environment variables
+MAX_WORKERS = int(os.getenv("JULIA_MAX_WORKERS", "8"))
+EXECUTION_TIMEOUT = int(os.getenv("JULIA_EXECUTION_TIMEOUT", "120"))
+LOG_FILE = os.getenv("JULIA_LOG_FILE", "/tmp/julia_env.log")
+LOG_LEVEL = os.getenv("JULIA_LOG_LEVEL", "INFO")
+
+
+def setup_logging():
+    """Configure logging to both file and console with rotation."""
+    # Configure both julia_env and openenv hierarchies to share handlers
+    julia_logger = logging.getLogger("julia_env")
+    openenv_logger = logging.getLogger("openenv")
+
+    julia_logger.setLevel(getattr(logging, LOG_LEVEL))
+    openenv_logger.setLevel(getattr(logging, LOG_LEVEL))
+
+    # Prevent duplicate handlers
+    if julia_logger.handlers:
+        return julia_logger
+
+    # Create formatters
+    detailed_formatter = logging.Formatter(
+        "%(asctime)s - %(name)s - [%(process)d:%(thread)d] - %(levelname)s - %(message)s",
+        datefmt="%Y-%m-%d %H:%M:%S",
+    )
+
+    # File handler with rotation (10MB max, keep 5 backup files)
+    try:
+        os.makedirs(os.path.dirname(LOG_FILE), exist_ok=True)
+        file_handler = RotatingFileHandler(
+            LOG_FILE, maxBytes=10 * 1024 * 1024, backupCount=5, encoding="utf-8"
+        )
+        file_handler.setLevel(logging.DEBUG)
+        file_handler.setFormatter(detailed_formatter)
+        julia_logger.addHandler(file_handler)
+        openenv_logger.addHandler(file_handler)
+    except Exception as e:
+        print(f"Warning: Could not create log file {LOG_FILE}: {e}")
+
+    # Console handler
+    console_handler = logging.StreamHandler(sys.stdout)
+    console_handler.setLevel(logging.INFO)
+    console_handler.setFormatter(detailed_formatter)
+    julia_logger.addHandler(console_handler)
+    openenv_logger.addHandler(console_handler)
+
+    return julia_logger
+
+
+# Setup logging
+logger = setup_logging()
+
+
+def initialize_julia_pool():
+    """Initialize the Julia process pool for better performance."""
+    port = int(os.getenv("PORT", "8000"))
+
+    logger.info("=" * 80)
+    logger.info("Starting Julia Environment Server")
+    logger.info(f"Container Port: {port}")
+    logger.info(f"Max Workers: {MAX_WORKERS}")
+    logger.info(f"Execution Timeout: {EXECUTION_TIMEOUT}s")
+    logger.info(f"Log File: {LOG_FILE}")
+    logger.info(f"Log Level: {LOG_LEVEL}")
+    logger.info("=" * 80)
+
+    # Enable Julia process pool for better performance
+    pool_enabled = JuliaExecutor.enable_process_pool(
+        size=MAX_WORKERS, timeout=EXECUTION_TIMEOUT
+    )
+    if pool_enabled:
+        logger.info(f"Julia process pool enabled with {MAX_WORKERS} workers")
+    else:
+        logger.warning("Julia process pool not available, using subprocess mode")
+
+    logger.info("Julia Environment Server initialized successfully")
+    print(f"Julia Environment Server started on port {port} with {MAX_WORKERS} concurrent workers")
+
+
+def shutdown_julia_pool():
+    """Shutdown the Julia process pool."""
+    logger.info("Shutting down Julia Environment Server...")
+    JuliaExecutor.shutdown_pool()
+    logger.info("Julia process pool shutdown complete")
+    print("Julia Environment Server shutdown complete")
+
+
+# Initialize the pool at module load time
+initialize_julia_pool()
+
+# Register cleanup on exit
+atexit.register(shutdown_julia_pool)
+
+
+# Create the app using OpenEnv's create_app for WebSocket support
+# Pass the class (factory) instead of an instance for session support
+app = create_app(
+    JuliaCodeActEnv,
+    JuliaAction,
+    JuliaObservation,
+    env_name="julia_env",
+    max_concurrent_envs=MAX_WORKERS,
+)
+
+
+# Add request logging middleware
+import time as time_module
+from starlette.middleware.base import BaseHTTPMiddleware
+from starlette.requests import Request
+
+class RequestLoggingMiddleware(BaseHTTPMiddleware):
+    """Log all incoming HTTP requests for debugging."""
+
+    async def dispatch(self, request: Request, call_next):
+        start_time = time_module.time()
+        path = request.url.path
+
+        # Skip health check logging to reduce noise
+        if path in ["/health", "/pool_status"]:
+            return await call_next(request)
+
+        logger.debug(f"HTTP Request: {request.method} {path}")
+
+        response = await call_next(request)
+
+        elapsed = time_module.time() - start_time
+        logger.debug(f"HTTP Response: {request.method} {path} -> {response.status_code} ({elapsed:.2f}s)")
+
+        return response
+
+app.add_middleware(RequestLoggingMiddleware)
+
+
+# Track active WebSocket connections
+_active_ws_connections = 0
+_total_ws_connections = 0
+
+
+# Add custom health endpoint with pool metrics
+@app.get("/pool_status")
+async def pool_status():
+    """Get Julia process pool status."""
+    return {
+        "max_workers": MAX_WORKERS,
+        "timeout": EXECUTION_TIMEOUT,
+        "pool_enabled": JuliaExecutor.is_pool_enabled(),
+        "pool_metrics": JuliaExecutor.get_pool_metrics(),
+        "active_ws_connections": _active_ws_connections,
+        "total_ws_connections": _total_ws_connections,
+    }
+
+
+def main():
+    """Main entry point for running the server."""
+    import uvicorn
+
+    port = int(os.getenv("PORT", "8000"))
+    uvicorn.run(app, host="0.0.0.0", port=port, log_level="info")
+
+
+if __name__ == "__main__":
+    main()

server/julia_codeact_env.py

@@ -0,0 +1,452 @@
+# Copyright (c) Meta Platforms, Inc. and affiliates.
+# All rights reserved.
+#
+# This source code is licensed under the BSD-style license found in the
+# LICENSE file in the root directory of this source tree.
+
+"""
+Julia Code Action Environment.
+
+This module provides a server-side environment implementation for executing
+Julia code actions using JuliaExecutor.
+"""
+
+import logging
+import re
+import time
+import uuid
+
+# Support both in-repo and standalone imports
+try:
+    # In-repo imports (when running from OpenEnv repository)
+    from openenv.core.env_server.interfaces import Action, Environment, Observation
+    from ..models import JuliaAction, JuliaObservation, JuliaState
+    from .julia_executor import JuliaExecutor
+    from .julia_transforms import create_safe_julia_transform
+except ImportError:
+    # Standalone imports (when environment is standalone)
+    from openenv.core.env_server.interfaces import Action, Environment, Observation
+    from models import JuliaAction, JuliaObservation, JuliaState
+    from server.julia_executor import JuliaExecutor
+    from server.julia_transforms import create_safe_julia_transform
+
+# Get logger for this module (inherits from julia_env logger)
+logger = logging.getLogger("julia_env.codeact")
+
+# Request counter for tracking
+_request_counter = 0
+
+
+def _detect_infinite_loop(code: str) -> tuple[bool, str]:
+    """
+    Detect potential infinite loops in Julia code.
+
+    This function scans for `while true` loops without break/return/error statements.
+
+    Args:
+        code: Julia code string to analyze
+
+    Returns:
+        Tuple of (has_infinite_loop: bool, reason: str)
+    """
+    # Remove comments and strings to avoid false positives
+    # Remove single-line comments
+    code_without_comments = re.sub(r'#.*', '', code)
+    # Remove multi-line strings (triple quotes)
+    code_without_comments = re.sub(r'""".*?"""', '', code_without_comments, flags=re.DOTALL)
+    # Remove single-line strings
+    code_without_comments = re.sub(r'"[^"]*"', '', code_without_comments)
+
+    # Find all while true blocks
+    while_true_pattern = r'\bwhile\s+true\b'
+    while_true_matches = list(re.finditer(while_true_pattern, code_without_comments, re.IGNORECASE))
+
+    if not while_true_matches:
+        return False, ""
+
+    # For each while true, check if there's a break/return/error in the same block
+    for match in while_true_matches:
+        start_pos = match.end()
+
+        # Find the end of this while block by counting 'while'/'end' pairs
+        # Simplified heuristic: look for break/return/error before the corresponding 'end'
+        remaining_code = code_without_comments[start_pos:]
+
+        # Extract potential loop body (up to next 'end' keyword)
+        # This is a simplified check - doesn't perfectly handle nested blocks
+        end_match = re.search(r'\bend\b', remaining_code)
+        if end_match:
+            loop_body = remaining_code[:end_match.start()]
+        else:
+            loop_body = remaining_code
+
+        # Check for loop exit mechanisms in this block
+        has_break = re.search(r'\bbreak\b', loop_body) is not None
+        has_return = re.search(r'\breturn\b', loop_body) is not None
+        has_error = re.search(r'\berror\(', loop_body) is not None
+        has_throw = re.search(r'\bthrow\(', loop_body) is not None
+        has_exit = re.search(r'\bexit\(', loop_body) is not None
+
+        if not (has_break or has_return or has_error or has_throw or has_exit):
+            loop_preview = loop_body[:100].strip()
+            return True, f"Infinite loop detected: 'while true' without break/return/error/throw. Preview: {loop_preview}"
+
+    return False, ""
+
+
+class JuliaCodeActEnv(Environment):
+    """
+    Julia Code Action Environment for executing code and tracking state.
+
+    This environment executes Julia code submitted as JuliaAction during step,
+    maintains the last exit code in its state, and returns results wrapped
+    in JuliaObservation.
+
+    Example:
+        >>> env = JuliaCodeActEnv()
+        >>> obs = env.reset()
+        >>> action = JuliaAction(core_code='println("Hello, Julia!")', test_code='')
+        >>> obs = env.step(action)
+        >>> print(obs.stdout)  # "Hello, Julia!\\n"
+        >>> print(obs.exit_code)  # 0
+        >>> print(env.state.last_exit_code)  # 0
+    """
+
+    # Allow concurrent sessions - each session has its own isolated state
+    SUPPORTS_CONCURRENT_SESSIONS = True
+
+    def __init__(self, use_process_pool: bool = True):
+        """
+        Initialize the Julia Code Act Environment.
+
+        Args:
+            use_process_pool: Use persistent Julia process pool for better performance
+                            and to avoid Juliaup lock contention (default: True)
+        """
+        self._executor = JuliaExecutor(use_process_pool=use_process_pool)
+        self._state = JuliaState()
+        self.transform = create_safe_julia_transform()
+
+    def reset(self, **kwargs) -> Observation:
+        """
+        Reset environment for a fresh Julia execution session.
+        Returns an empty JuliaObservation with exit_code=0.
+
+        Note: Executor is reused to leverage process pool.
+        """
+        self._state = JuliaState(episode_id=str(uuid.uuid4()), step_count=0)
+        self._state.last_exit_code = 0
+        self._state.last_code_compiles = True
+        # Don't recreate executor - reuse it to leverage process pool
+
+        observation = JuliaObservation(
+            stdout="",
+            stderr="",
+            exit_code=0,
+            reward=0.0,
+            metadata={"core_code": "", "test_code": ""},
+            tests_passed=0,
+            tests_failed=0,
+            code_compiles=True,
+        )
+
+        observation = self._apply_transform(observation)
+        return observation
+
+    def step(self, action: Action, **kwargs) -> Observation:
+        """
+        Execute Julia code and return the result as JuliaObservation.
+
+        Optimized single-pass execution:
+        - Runs core_code + test_code together
+        - Infers compilation status from combined execution
+        - 2x faster than double execution
+
+        Args:
+            action: JuliaAction with core_code and optional test_code
+            **kwargs: Optional parameters including:
+                - timeout: Execution timeout in seconds (default: 120)
+        """
+        global _request_counter
+        _request_counter += 1
+        request_id = _request_counter
+
+        if not isinstance(action, JuliaAction):
+            logger.error(f"[REQ-{request_id}] Invalid action type: {type(action)}")
+            raise ValueError(f"Expected JuliaAction, got {type(action)}")
+
+        # Get timeout from kwargs (default handled by executor)
+        timeout = kwargs.get("timeout")
+
+        # Log request details
+        code_preview = action.core_code[:200] + "..." if len(action.core_code) > 200 else action.core_code
+        logger.info(f"[REQ-{request_id}] === NEW EXECUTION REQUEST ===")
+        logger.info(f"[REQ-{request_id}] Session: {self._state.episode_id}, Step: {self._state.step_count}")
+        logger.info(f"[REQ-{request_id}] Code length: {len(action.core_code)} chars, Test length: {len(action.test_code or '')} chars")
+        logger.debug(f"[REQ-{request_id}] Code preview: {code_preview}")
+        logger.info(f"[REQ-{request_id}] Timeout: {timeout}s" if timeout else f"[REQ-{request_id}] Timeout: default")
+
+        start_time = time.time()
+
+        # Single execution: Run core_code + test_code together (if test_code provided)
+        if action.test_code:
+            combined_code = action.core_code + "\n\n" + action.test_code
+        else:
+            combined_code = action.core_code
+
+        # Pre-execution check: detect infinite loops to avoid timeout
+        has_infinite_loop, loop_reason = _detect_infinite_loop(action.core_code)
+        if has_infinite_loop:
+            logger.warning(f"[REQ-{request_id}] INFINITE LOOP DETECTED: {loop_reason}")
+
+            # Update environment state
+            self._state.step_count += 1
+            self._state.last_exit_code = 1
+            self._state.last_code_compiles = True  # Code compiles but has infinite loop
+            self._state.total_tests_passed = 0
+            self._state.total_tests_failed = 0
+
+            # Build observation with penalty
+            observation = JuliaObservation(
+                stdout="",
+                stderr=f"Infinite loop detected (pre-execution check): {loop_reason}",
+                exit_code=1,
+                reward=-1.0,  # Penalize infinite loops
+                metadata={
+                    "core_code": action.core_code,
+                    "test_code": action.test_code or "",
+                    "infinite_loop_detected": True,
+                    "infinite_loop_reason": loop_reason,
+                },
+                tests_passed=0,
+                tests_failed=0,
+                code_compiles=True,  # Code would compile, but not run
+            )
+
+            logger.info(
+                f"[REQ-{request_id}] RESULT: infinite_loop=True, "
+                f"tests_passed=0, tests_failed=0, reward=-1.00"
+            )
+
+            observation = self._apply_transform(observation)
+            return observation
+
+        try:
+            full_result = self._executor.run(combined_code, timeout=timeout)
+            execution_time = time.time() - start_time
+
+            logger.info(f"[REQ-{request_id}] Execution completed in {execution_time:.2f}s, exit_code={full_result.exit_code}")
+
+            # Log stderr if present (often contains errors or test output)
+            if full_result.stderr:
+                stderr_preview = full_result.stderr[:500] + "..." if len(full_result.stderr) > 500 else full_result.stderr
+                logger.debug(f"[REQ-{request_id}] Stderr: {stderr_preview}")
+
+        except Exception as e:
+            execution_time = time.time() - start_time
+            logger.error(f"[REQ-{request_id}] EXECUTION FAILED after {execution_time:.2f}s: {e}")
+            raise
+
+        # Parse test results from execution output
+        tests_passed, tests_failed = self._parse_test_results(
+            full_result.stdout, full_result.stderr
+        )
+
+        # Infer compilation status from execution
+        # If tests ran, code compiled successfully
+        # If exit_code != 0 and no tests ran, code didn't compile
+        code_compiles = (
+            full_result.exit_code == 0  # Clean execution
+            or tests_passed > 0  # Some tests passed (code must have compiled)
+            or tests_failed > 0  # Some tests failed (code compiled but tests failed)
+        )
+
+        # If no tests detected and non-zero exit, check for compilation errors
+        if not code_compiles and tests_passed == 0 and tests_failed == 0:
+            # Check stderr for compilation errors
+            stderr_lower = full_result.stderr.lower()
+            if any(
+                err in stderr_lower
+                for err in ["error", "syntax", "undefined", "loadError"]
+            ):
+                code_compiles = False
+            else:
+                # If no clear compilation error, assume it compiled
+                code_compiles = True
+
+        # Calculate reward based on compilation and test results
+        reward = self._calculate_reward(code_compiles, tests_passed, tests_failed)
+
+        # Log final results
+        logger.info(
+            f"[REQ-{request_id}] RESULT: compiles={code_compiles}, "
+            f"tests_passed={tests_passed}, tests_failed={tests_failed}, reward={reward:.2f}"
+        )
+
+        # Update environment state
+        self._state.step_count += 1
+        self._state.last_exit_code = full_result.exit_code
+        self._state.last_code_compiles = code_compiles
+        self._state.total_tests_passed = tests_passed
+        self._state.total_tests_failed = tests_failed
+
+        # Build observation
+        observation = JuliaObservation(
+            stdout=full_result.stdout,
+            stderr=full_result.stderr,
+            exit_code=full_result.exit_code,
+            reward=reward,
+            metadata={
+                "core_code": action.core_code,
+                "test_code": action.test_code or "",
+            },
+            tests_passed=tests_passed,
+            tests_failed=tests_failed,
+            code_compiles=code_compiles,
+        )
+
+        # Apply safety and quality transforms
+        observation = self._apply_transform(observation)
+
+        return observation
+
+    def _parse_test_results(self, stdout: str, stderr: str) -> tuple[int, int]:
+        """
+        Parse Julia test output to count passed/failed tests.
+
+        Julia's Test module outputs results like:
+        "Test Summary:      | Pass  Fail  Total  Time"
+        "Add function Tests |    1     1      2  1.5s"
+
+        Also checks error messages:
+        "Some tests did not pass: 1 passed, 1 failed, 0 errored, 0 broken."
+
+        Args:
+            stdout: Standard output from Julia execution
+            stderr: Standard error from Julia execution
+
+        Returns:
+            Tuple of (tests_passed, tests_failed)
+        """
+        # Combine stdout and stderr for analysis
+        passed = 0
+        failed = 0
+        output = stdout + "\n" + stderr
+
+        # Method 1: Look for "Some tests did not pass" error message
+        # Pattern: "Some tests did not pass: X passed, Y failed, Z errored, W broken."
+        error_pattern = r"Some tests did not pass:\s*(\d+)\s+passed,\s*(\d+)\s+failed,\s*(\d+)\s+errored"
+        match = re.search(error_pattern, output)
+
+        if match:
+            passed = int(match.group(1))
+            failed = int(match.group(2))
+            errored = int(match.group(3))
+            return passed, failed + errored  # Treat errors as failures
+
+        # Method 2: Look for Test Summary table
+        # Multiple possible formats:
+        # All pass:     "Test Summary: | Pass  Total  Time"
+        #               "My Tests     |    3      3  0.5s"
+        # Some fail:    "Test Summary: | Pass  Fail  Total  Time"
+        #               "My Tests     |    2     1      3  0.5s"
+        # All error:    "Test Summary: | Error  Total  Time"
+        #               "My Tests     |     3      3  0.9s"
+        # Mixed:        "Test Summary: | Pass  Fail  Error  Total  Time"
+        #               "My Tests     |    1     1      1      3  0.5s"
+        summary_lines = output.split("\n")
+        for i, line in enumerate(summary_lines):
+            if "Test Summary:" in line and i + 1 < len(summary_lines):
+                header_line = line
+                next_line = summary_lines[i + 1]
+
+                # Determine which columns are present
+                has_pass = "Pass" in header_line
+                has_fail = "Fail" in header_line
+                has_error = "Error" in header_line
+
+                # Extract all numbers from the line
+                all_numbers = re.findall(r"\d+", next_line)
+                if not all_numbers:
+                    continue
+
+                # Last number is always Total, second to last is Time (skip it)
+                # Extract based on which columns exist
+                if has_pass and has_fail and has_error:
+                    # Pass  Fail  Error  Total  Time
+                    if len(all_numbers) >= 5:
+                        passed = int(all_numbers[0])
+                        failed = int(all_numbers[1]) + int(
+                            all_numbers[2]
+                        )  # Fail + Error
+                        return passed, failed
+                elif has_pass and has_fail:
+                    # Pass  Fail  Total  Time
+                    if len(all_numbers) >= 4:
+                        passed = int(all_numbers[0])
+                        failed = int(all_numbers[1])
+                        return passed, failed
+                elif has_pass and has_error:
+                    # Pass  Error  Total  Time
+                    if len(all_numbers) >= 4:
+                        passed = int(all_numbers[0])
+                        failed = int(all_numbers[1])  # Treat errors as failures
+                        return passed, failed
+                elif has_fail and has_error:
+                    # Fail  Error  Total  Time (no passes)
+                    if len(all_numbers) >= 4:
+                        passed = 0
+                        failed = int(all_numbers[0]) + int(all_numbers[1])
+                        return passed, failed
+                elif has_pass:
+                    # Pass  Total  Time (no failures/errors)
+                    if len(all_numbers) >= 3:
+                        passed = int(all_numbers[0])
+                        failed = 0
+                        return passed, failed
+                elif has_error:
+                    # Error  Total  Time (all errors, no passes)
+                    if len(all_numbers) >= 3:
+                        passed = 0
+                        failed = int(all_numbers[0])  # Treat all errors as failures
+                        return passed, failed
+                elif has_fail:
+                    # Fail  Total  Time (all failures, no passes)
+                    if len(all_numbers) >= 3:
+                        passed = 0
+                        failed = int(all_numbers[0])
+                        return passed, failed
+
+        return passed, failed
+
+    def _calculate_reward(
+        self, code_compiles: bool, tests_passed: int, tests_failed: int
+    ) -> float:
+        """
+        Normalized percentage-based reward for Julia GRPO.
+        Returns rewards in [-1, 1.5] range for comparability across problems.
+        """
+        if not code_compiles:
+            return -1.0
+
+        total_tests = tests_passed + tests_failed
+        if total_tests == 0:
+            return 0.0  # No signal when no tests run
+
+        pass_rate = tests_passed / total_tests
+
+        # Scaled 0-1 with bonus for perfection
+        if pass_rate == 1.0:
+            return 1.5  # Bonus for passing all tests
+        return pass_rate
+
+    def _apply_transform(self, observation: JuliaObservation) -> JuliaObservation:
+        """Apply safety and quality transforms to observation."""
+        if self.transform:
+            observation = self.transform(observation)
+        return observation
+
+    @property
+    def state(self) -> JuliaState:
+        """Return current environment state."""
+        return self._state

server/julia_executor.py

@@ -0,0 +1,458 @@
+# Copyright (c) Meta Platforms, Inc. and affiliates.
+# All rights reserved.
+#
+# This source code is licensed under the BSD-style license found in the
+# LICENSE file in the root directory of this source tree.
+
+"""
+Local Julia Executor with Process Pool Support.
+
+This module provides a Julia code executor with:
+- Proper process cleanup on timeout (no zombie processes)
+- Robust error handling and logging
+- Process group management for complete cleanup
+- Automatic retry on transient failures
+- Optional process pool for 50-100x speedup on repeated executions
+
+Performance Modes:
+- Standard mode: Spawn new process for each execution (default for single executions)
+- Pool mode: Reuse persistent Julia processes (recommended for repeated executions)
+"""
+
+from __future__ import annotations
+
+import logging
+import os
+import shutil
+import signal
+import subprocess
+import tempfile
+import threading
+import time
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Optional
+
+# Use julia_env hierarchy to inherit handlers from app.py's setup_logging()
+logger = logging.getLogger("julia_env.executor")
+
+
+@dataclass
+class CodeExecResult:
+    """Result of code execution."""
+
+    stdout: str
+    stderr: str
+    exit_code: int
+
+
+# Try to import process pool (optional dependency)
+try:
+    from .julia_process_pool import JuliaProcessPool
+
+    POOL_AVAILABLE = True
+except ImportError:
+    POOL_AVAILABLE = False
+    JuliaProcessPool = None
+
+
+class JuliaExecutor:
+    """
+    Executor for running Julia code with robust process management.
+
+    This class provides a safe interface to execute Julia code in isolation
+    and capture the results including stdout, stderr, and exit code.
+
+    Features:
+    - Proper timeout handling without zombie processes
+    - Process group cleanup for nested processes
+    - Automatic retry on transient failures
+    - Comprehensive logging for debugging
+    - Optional process pool for 50-100x speedup on repeated executions
+
+    Example:
+        >>> executor = JuliaExecutor()
+        >>> result = executor.run('println("Hello, Julia!")')
+        >>> print(result.stdout)  # "Hello, Julia!\\n"
+        >>> print(result.exit_code)  # 0
+        >>>
+        >>> # With process pool (recommended for repeated executions)
+        >>> JuliaExecutor.enable_process_pool(size=4)
+        >>> executor = JuliaExecutor(use_process_pool=True)
+        >>> for i in range(100):
+        ...     result = executor.run(f'println({i})')  # 50-100x faster!
+        >>> JuliaExecutor.shutdown_pool()  # Clean up when done
+    """
+
+    # Class-level process pool (shared across all instances if enabled)
+    _shared_pool: Optional["JuliaProcessPool"] = None
+    _pool_lock = threading.Lock()
+    _pool_size: int = 0
+    _pool_timeout: int = 120
+
+    def __init__(
+        self,
+        timeout: Optional[int] = None,
+        max_retries: int = 0,
+        use_optimization_flags: bool = True,
+        use_process_pool: bool = True,
+    ):
+        """
+        Initialize the JuliaExecutor.
+
+        Args:
+            timeout: Maximum execution time in seconds. If None, reads from
+                     JULIA_EXECUTION_TIMEOUT env var (default: 120 if not set)
+            max_retries: Number of retry attempts on transient failures (default: 0)
+            use_optimization_flags: Enable Julia performance flags (default: True)
+            use_process_pool: Use process pool if available (default: True)
+
+        Raises:
+            RuntimeError: If Julia executable is not found in PATH
+        """
+        # Read timeout from env var if not explicitly provided
+        if timeout is None:
+            timeout = int(os.getenv("JULIA_EXECUTION_TIMEOUT", "120"))
+            logger.debug(f"Executor timeout from JULIA_EXECUTION_TIMEOUT env var: {timeout}s")
+        self.timeout = timeout
+        self.max_retries = max_retries
+        self.use_optimization_flags = use_optimization_flags
+        self._use_process_pool = use_process_pool
+
+        # Find Julia executable in PATH
+        self.julia_path = shutil.which("julia")
+
+        if not self.julia_path:
+            # Try common installation paths
+            common_paths = [
+                os.path.expanduser("~/.juliaup/bin/julia"),
+                os.path.expanduser("~/.julia/bin/julia"),
+                "/usr/local/bin/julia",
+                "/usr/bin/julia",
+            ]
+
+            for path in common_paths:
+                if os.path.isfile(path) and os.access(path, os.X_OK):
+                    self.julia_path = path
+                    break
+
+        if not self.julia_path:
+            logger.warning(
+                "Julia executable not found in PATH or common locations. "
+                "Please install Julia: https://julialang.org/downloads/"
+            )
+
+        # Build optimized Julia command with performance flags
+        self.base_cmd = [self.julia_path] if self.julia_path else ["julia"]
+
+        if self.use_optimization_flags:
+            self.base_cmd.extend(
+                [
+                    "--compile=min",
+                    "--optimize=2",
+                    "--startup-file=no",
+                    "--history-file=no",
+                ]
+            )
+
+        logger.debug(f"JuliaExecutor initialized with Julia at: {self.julia_path}")
+        logger.debug(f"Timeout: {self.timeout}s, Max retries: {self.max_retries}")
+
+    def _kill_process_tree(
+        self, proc: subprocess.Popen, script_file: Optional[str] = None
+    ) -> None:
+        """
+        Terminate a process and all its children.
+
+        This prevents zombie processes by ensuring complete cleanup.
+
+        Args:
+            proc: The subprocess.Popen instance to terminate
+            script_file: Optional script file path (for logging)
+        """
+        if proc.poll() is None:  # Process is still running
+            try:
+                # Try graceful termination first
+                logger.warning(f"Terminating process {proc.pid} gracefully...")
+                proc.terminate()
+
+                # Wait up to 2 seconds for graceful termination
+                try:
+                    proc.wait(timeout=2.0)
+                    logger.debug(f"Process {proc.pid} terminated gracefully")
+                    return
+                except subprocess.TimeoutExpired:
+                    logger.warning(
+                        f"Process {proc.pid} did not terminate, forcing kill..."
+                    )
+
+                # Force kill if still running
+                proc.kill()
+                try:
+                    proc.wait(timeout=2.0)
+                    logger.debug(f"Process {proc.pid} killed forcefully")
+                except subprocess.TimeoutExpired:
+                    pass
+
+            except Exception as e:
+                logger.error(f"Error killing process {proc.pid}: {e}")
+
+                # Last resort: try killing via process group
+                try:
+                    if hasattr(os, "killpg"):
+                        os.killpg(os.getpgid(proc.pid), signal.SIGKILL)
+                        logger.debug(f"Killed process group for {proc.pid}")
+                except Exception as pg_error:
+                    logger.error(f"Failed to kill process group: {pg_error}")
+
+    def run(self, code: str, timeout: Optional[int] = None) -> CodeExecResult:
+        """
+        Execute Julia code and return the result with robust error handling.
+
+        This method provides:
+        - Automatic retry on transient failures
+        - Proper timeout handling without zombie processes
+        - Process group cleanup for nested processes
+        - Comprehensive error logging
+        - Optional process pool for 50-100x speedup
+
+        Args:
+            code: Julia code string to execute
+            timeout: Override default timeout (seconds). If None, uses pool's
+                     configured timeout (when using pool) or instance timeout.
+
+        Returns:
+            CodeExecResult containing stdout, stderr, and exit_code
+        """
+        # Use process pool if enabled and available
+        # Pass timeout as-is (None means use pool's configured default)
+        if self._use_process_pool and JuliaExecutor._shared_pool is not None:
+            try:
+                return JuliaExecutor._shared_pool.execute(code, timeout=timeout)
+            except Exception as e:
+                logger.warning(
+                    f"Process pool execution failed: {e}, falling back to subprocess"
+                )
+                # Fall through to standard execution
+
+        # For subprocess fallback, apply instance default if timeout not specified
+        if timeout is None:
+            timeout = self.timeout
+
+        # Check if Julia is available
+        if not self.julia_path:
+            return CodeExecResult(
+                stdout="",
+                stderr="Julia not found in PATH. Please install Julia.",
+                exit_code=127,
+            )
+
+        code_file = None
+
+        for attempt in range(self.max_retries + 1):
+            proc = None
+
+            try:
+                # Create temporary file for Julia code
+                with tempfile.NamedTemporaryFile(
+                    mode="w", suffix=".jl", delete=False, encoding="utf-8"
+                ) as f:
+                    f.write(code)
+                    code_file = f.name
+
+                script_name = Path(code_file).name
+                logger.debug(
+                    f"[Attempt {attempt + 1}/{self.max_retries + 1}] Executing: {script_name}"
+                )
+
+                # Start process with Popen for better control
+                start_time = time.time()
+
+                # On Unix systems, use process groups for better cleanup
+                kwargs = {
+                    "stdout": subprocess.PIPE,
+                    "stderr": subprocess.PIPE,
+                    "text": True,
+                }
+
+                # Create new process group on Unix systems
+                if hasattr(os, "setpgrp"):
+                    kwargs["preexec_fn"] = os.setpgrp
+
+                proc = subprocess.Popen(self.base_cmd + [code_file], **kwargs)
+
+                logger.debug(f"Started Julia process {proc.pid}")
+
+                # Wait for process with timeout
+                try:
+                    stdout, stderr = proc.communicate(timeout=timeout)
+                    exit_code = proc.returncode
+                    elapsed = time.time() - start_time
+
+                    logger.debug(
+                        f"Julia execution completed in {elapsed:.2f}s (exit: {exit_code})"
+                    )
+
+                    # Clean up temp file
+                    self._cleanup_temp_file(code_file)
+
+                    return CodeExecResult(
+                        stdout=stdout,
+                        stderr=stderr,
+                        exit_code=exit_code,
+                    )
+
+                except subprocess.TimeoutExpired:
+                    logger.error(
+                        f"Julia execution timed out after {timeout}s "
+                        f"(attempt {attempt + 1}/{self.max_retries + 1})"
+                    )
+
+                    # CRITICAL: Kill the process AND all its children
+                    self._kill_process_tree(proc, code_file)
+
+                    # If this was our last retry, return timeout error
+                    if attempt >= self.max_retries:
+                        self._cleanup_temp_file(code_file)
+                        return CodeExecResult(
+                            stdout="",
+                            stderr=f"Execution timed out after {timeout}s",
+                            exit_code=124,  # Standard timeout exit code
+                        )
+
+                    # Wait before retry
+                    time.sleep(1.0)
+                    continue
+
+            except FileNotFoundError:
+                logger.error(f"Julia executable not found at {self.julia_path}")
+                return CodeExecResult(
+                    stdout="",
+                    stderr=f"Julia executable not found: {self.julia_path}",
+                    exit_code=127,
+                )
+
+            except Exception as e:
+                logger.error(
+                    f"Error executing Julia (attempt {attempt + 1}/{self.max_retries + 1}): {e}"
+                )
+
+                # Try to kill process if it exists
+                if proc is not None and proc.poll() is None:
+                    self._kill_process_tree(proc, code_file)
+
+                # If this was our last retry, return error
+                if attempt >= self.max_retries:
+                    self._cleanup_temp_file(code_file)
+                    return CodeExecResult(
+                        stdout="",
+                        stderr=f"Error executing Julia code: {str(e)}",
+                        exit_code=1,
+                    )
+
+                # Wait before retry
+                time.sleep(1.0)
+                continue
+
+            finally:
+                # Always ensure temp file is cleaned up
+                self._cleanup_temp_file(code_file)
+
+        # Should never reach here
+        return CodeExecResult(
+            stdout="",
+            stderr="Unexpected error: all retries exhausted",
+            exit_code=1,
+        )
+
+    def _cleanup_temp_file(self, code_file: Optional[str]) -> None:
+        """Clean up temporary file safely."""
+        if code_file and Path(code_file).exists():
+            try:
+                Path(code_file).unlink()
+            except Exception as e:
+                logger.debug(f"Could not delete temp file {code_file}: {e}")
+
+    @staticmethod
+    def enable_process_pool(size: int = 4, timeout: Optional[int] = None) -> bool:
+        """
+        Enable the shared Julia process pool for all JuliaExecutor instances.
+
+        This provides 50-100x speedup for repeated code executions by reusing
+        persistent Julia processes instead of spawning new ones.
+
+        Args:
+            size: Number of worker processes to create (default: 4)
+            timeout: Default timeout for code execution in seconds.
+                     If None, reads from JULIA_EXECUTION_TIMEOUT env var (default: 120)
+
+        Returns:
+            True if pool was created successfully, False otherwise
+        """
+        if not POOL_AVAILABLE:
+            logger.warning(
+                "Process pool not available (julia_process_pool module not found). "
+                "Falling back to subprocess execution."
+            )
+            return False
+
+        # Read timeout from env var if not explicitly provided
+        if timeout is None:
+            timeout = int(os.getenv("JULIA_EXECUTION_TIMEOUT", "120"))
+
+        with JuliaExecutor._pool_lock:
+            if JuliaExecutor._shared_pool is not None:
+                logger.debug("Process pool already enabled")
+                return True
+
+            try:
+                logger.info(f"Enabling Julia process pool with {size} workers")
+                JuliaExecutor._shared_pool = JuliaProcessPool(size=size, timeout=timeout)
+                JuliaExecutor._pool_size = size
+                JuliaExecutor._pool_timeout = timeout
+                logger.info("Julia process pool enabled successfully")
+                return True
+            except Exception as e:
+                logger.error(f"Failed to enable process pool: {e}")
+                return False
+
+    @staticmethod
+    def shutdown_pool() -> None:
+        """
+        Shutdown the shared Julia process pool.
+
+        This should be called when you're done with all Julia executions
+        to properly clean up worker processes.
+        """
+        with JuliaExecutor._pool_lock:
+            if JuliaExecutor._shared_pool is not None:
+                logger.info("Shutting down Julia process pool")
+                try:
+                    JuliaExecutor._shared_pool.shutdown()
+                except Exception as e:
+                    logger.error(f"Error shutting down pool: {e}")
+                finally:
+                    JuliaExecutor._shared_pool = None
+
+    @staticmethod
+    def is_pool_enabled() -> bool:
+        """Check if the process pool is currently enabled."""
+        with JuliaExecutor._pool_lock:
+            return JuliaExecutor._shared_pool is not None
+
+    @staticmethod
+    def get_pool_metrics() -> dict:
+        """Get metrics about the process pool."""
+        if JuliaExecutor._shared_pool is None:
+            return {
+                "enabled": False,
+                "pool_size": 0,
+                "available_workers": 0,
+            }
+
+        return {
+            "enabled": True,
+            "pool_size": JuliaExecutor._pool_size,
+            "timeout": JuliaExecutor._pool_timeout,
+            "available_workers": JuliaExecutor._pool_size,
+        }

server/julia_process_pool.py

@@ -0,0 +1,601 @@
+# Copyright (c) Meta Platforms, Inc. and affiliates.
+# All rights reserved.
+#
+# This source code is licensed under the BSD-style license found in the
+# LICENSE file in the root directory of this source tree.
+
+"""
+Julia Process Pool for high-performance code execution.
+
+This module provides a pool of persistent Julia processes that can be reused
+for multiple code executions, eliminating the overhead of spawning new processes.
+
+Expected speedup: 50-100x for repeated executions compared to spawning new processes.
+
+Features:
+- Persistent Julia processes (no startup overhead)
+- Thread-safe process allocation
+- Automatic recovery from process failures
+- Proper cleanup on shutdown
+- Timeout handling per execution
+
+Example:
+    >>> pool = JuliaProcessPool(size=4, timeout=30)
+    >>> result = pool.execute("println('Hello, Julia!')")
+    >>> print(result.stdout)  # "Hello, Julia!\\n"
+    >>> pool.shutdown()  # Clean up all processes
+"""
+
+from __future__ import annotations
+
+import atexit
+import logging
+import os
+import shutil
+import subprocess
+import threading
+import time
+from collections import deque
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Optional
+
+# Use julia_env hierarchy to inherit handlers from app.py's setup_logging()
+logger = logging.getLogger("julia_env.pool")
+
+
+@dataclass
+class CodeExecResult:
+    """Result of code execution."""
+
+    stdout: str
+    stderr: str
+    exit_code: int
+
+
+class JuliaWorkerProcess:
+    """
+    Single Julia worker process that can execute code repeatedly.
+
+    This class manages communication with a persistent Julia REPL process
+    using a delimiter-based protocol.
+    """
+
+    # Communication protocol delimiters
+    START_OUTPUT = "<<<START_OUTPUT>>>"
+    START_ERROR = "<<<START_ERROR>>>"
+    EXIT_CODE_PREFIX = "<<<EXIT_CODE:"
+    END_EXECUTION = "<<<END_EXECUTION>>>"
+    END_CODE = "<<<END_CODE>>>"
+
+    def __init__(
+        self,
+        worker_id: int,
+        julia_path: str,
+        worker_script: str,
+        optimization_flags: bool = True,
+    ):
+        """
+        Initialize a Julia worker process.
+
+        Args:
+            worker_id: Unique identifier for this worker
+            julia_path: Path to Julia executable
+            worker_script: Path to julia_repl_worker.jl script
+            optimization_flags: Enable Julia optimization flags
+        """
+        self.worker_id = worker_id
+        self.julia_path = julia_path
+        self.worker_script = worker_script
+        self.optimization_flags = optimization_flags
+        self.process: Optional[subprocess.Popen] = None
+        self.is_busy = False
+        self.is_healthy = True
+        self.lock = threading.Lock()
+
+        # Start the worker process
+        self._start_process()
+
+    def _start_process(self) -> None:
+        """Start the Julia worker process."""
+        cmd = [self.julia_path]
+
+        if self.optimization_flags:
+            cmd.extend(
+                [
+                    "--compile=min",
+                    "--optimize=2",
+                    "--startup-file=no",
+                    "--history-file=no",
+                ]
+            )
+
+        cmd.append(self.worker_script)
+
+        try:
+            self.process = subprocess.Popen(
+                cmd,
+                stdin=subprocess.PIPE,
+                stdout=subprocess.PIPE,
+                stderr=subprocess.PIPE,
+                text=True,
+                bufsize=1,  # Line buffered
+            )
+
+            # Wait for "Julia worker ready" message on stderr
+            ready_msg = self.process.stderr.readline()
+            if "ready" not in ready_msg.lower():
+                raise RuntimeError(
+                    f"Worker {self.worker_id} did not start properly: {ready_msg}"
+                )
+
+            self.is_healthy = True
+            logger.info(f"Worker {self.worker_id} started (PID: {self.process.pid})")
+
+        except Exception as e:
+            self.is_healthy = False
+            logger.error(f"Failed to start worker {self.worker_id}: {e}")
+            raise
+
+    def execute(self, code: str, timeout: Optional[int] = None) -> CodeExecResult:
+        """
+        Execute Julia code in this worker process.
+
+        Args:
+            code: Julia code to execute
+            timeout: Maximum execution time in seconds.
+                     If None, reads from JULIA_EXECUTION_TIMEOUT env var (default: 120)
+
+        Returns:
+            CodeExecResult with stdout, stderr, and exit_code
+        """
+        # Read timeout from env var if not explicitly provided
+        if timeout is None:
+            timeout = int(os.getenv("JULIA_EXECUTION_TIMEOUT", "120"))
+
+        with self.lock:
+            if not self.is_healthy or self.process is None:
+                raise RuntimeError(f"Worker {self.worker_id} is not healthy")
+
+            self.is_busy = True
+
+            try:
+                # Send code to worker
+                self.process.stdin.write(code + "\n")
+                self.process.stdin.write(self.END_CODE + "\n")
+                self.process.stdin.flush()
+
+                # Use threading for proper timeout handling
+                # The blocking readline() would otherwise prevent timeout detection
+                result_container: dict = {
+                    "stdout_lines": [],
+                    "stderr_lines": [],
+                    "exit_code": -1,
+                    "completed": False,
+                    "error": None,
+                }
+
+                def read_output():
+                    """Read output in a separate thread."""
+                    stdout_lines = []
+                    stderr_lines = []
+                    exit_code = -1
+                    current_section = None
+
+                    try:
+                        while True:
+                            line = self.process.stdout.readline()
+
+                            if not line:
+                                # EOF - process died
+                                result_container["error"] = "Worker process died unexpectedly"
+                                return
+
+                            line = line.rstrip("\n")
+
+                            # Check for delimiters
+                            if line == self.START_OUTPUT:
+                                current_section = "stdout"
+                                continue
+                            elif line == self.START_ERROR:
+                                current_section = "stderr"
+                                continue
+                            elif line.startswith(self.EXIT_CODE_PREFIX):
+                                # Parse exit code
+                                exit_code_str = line[
+                                    len(self.EXIT_CODE_PREFIX) : -3
+                                ]  # Remove prefix and ">>>"
+                                exit_code = int(exit_code_str)
+                                continue
+                            elif line == self.END_EXECUTION:
+                                # Execution complete
+                                break
+
+                            # Accumulate output
+                            if current_section == "stdout":
+                                stdout_lines.append(line)
+                            elif current_section == "stderr":
+                                stderr_lines.append(line)
+
+                        result_container["stdout_lines"] = stdout_lines
+                        result_container["stderr_lines"] = stderr_lines
+                        result_container["exit_code"] = exit_code
+                        result_container["completed"] = True
+
+                    except Exception as e:
+                        result_container["error"] = f"Error reading from worker: {str(e)}"
+
+                # Start reader thread
+                reader_thread = threading.Thread(target=read_output, daemon=True)
+                reader_thread.start()
+
+                # Wait for completion with timeout
+                reader_thread.join(timeout=timeout)
+
+                if reader_thread.is_alive():
+                    # Timeout - kill the process
+                    logger.error(f"Worker {self.worker_id} execution timed out after {timeout}s")
+                    self.is_healthy = False
+                    self._kill_process()
+                    return CodeExecResult(
+                        stdout="",
+                        stderr=f"Execution timed out after {timeout} seconds",
+                        exit_code=-1,
+                    )
+
+                # Check for errors
+                if result_container["error"]:
+                    logger.error(f"Worker {self.worker_id}: {result_container['error']}")
+                    self.is_healthy = False
+                    return CodeExecResult(
+                        stdout="".join(result_container["stdout_lines"]),
+                        stderr=result_container["error"],
+                        exit_code=-1,
+                    )
+
+                # Reconstruct output (add newlines back)
+                stdout_lines = result_container["stdout_lines"]
+                stderr_lines = result_container["stderr_lines"]
+                stdout_str = "\n".join(stdout_lines) + ("\n" if stdout_lines else "")
+                stderr_str = "\n".join(stderr_lines) + ("\n" if stderr_lines else "")
+
+                return CodeExecResult(
+                    stdout=stdout_str,
+                    stderr=stderr_str,
+                    exit_code=result_container["exit_code"],
+                )
+
+            finally:
+                self.is_busy = False
+
+    def _kill_process(self) -> None:
+        """Kill the worker process."""
+        if self.process is not None:
+            try:
+                self.process.terminate()
+                self.process.wait(timeout=2.0)
+            except Exception:
+                try:
+                    self.process.kill()
+                    self.process.wait(timeout=1.0)
+                except Exception:
+                    pass
+
+    def shutdown(self) -> None:
+        """Shutdown the worker process gracefully."""
+        with self.lock:
+            if self.process is not None:
+                logger.info(f"Shutting down worker {self.worker_id}")
+                self._kill_process()
+                self.process = None
+                self.is_healthy = False
+
+
+class JuliaProcessPool:
+    """
+    Pool of persistent Julia processes for high-performance code execution.
+
+    This class manages multiple Julia worker processes and distributes
+    code execution among them, providing significant speedup by eliminating
+    process startup overhead.
+
+    Thread-safe for concurrent access from multiple threads.
+
+    Example:
+        >>> pool = JuliaProcessPool(size=4)
+        >>>
+        >>> # Execute code
+        >>> result = pool.execute("println('Hello')")
+        >>>
+        >>> # Pool automatically manages workers
+        >>> results = [pool.execute(f"println({i})") for i in range(100)]
+        >>>
+        >>> # Cleanup when done
+        >>> pool.shutdown()
+    """
+
+    def __init__(
+        self,
+        size: int = 4,
+        timeout: Optional[int] = None,
+        julia_path: Optional[str] = None,
+        optimization_flags: bool = True,
+        auto_recover: bool = True,
+    ):
+        """
+        Initialize the Julia process pool.
+
+        Args:
+            size: Number of worker processes to create (default: 4)
+            timeout: Default timeout for code execution in seconds.
+                     If None, reads from JULIA_EXECUTION_TIMEOUT env var (default: 120)
+            julia_path: Path to Julia executable (auto-detected if None)
+            optimization_flags: Enable Julia optimization flags (default: True)
+            auto_recover: Automatically restart failed workers (default: True)
+
+        Raises:
+            RuntimeError: If Julia executable is not found
+        """
+        self.size = size
+        # Read timeout from env var if not explicitly provided
+        if timeout is None:
+            timeout = int(os.getenv("JULIA_EXECUTION_TIMEOUT", "120"))
+            logger.info(f"Pool timeout from JULIA_EXECUTION_TIMEOUT env var: {timeout}s")
+        else:
+            logger.info(f"Pool timeout explicitly set: {timeout}s")
+        self.timeout = timeout
+        self.optimization_flags = optimization_flags
+        self.auto_recover = auto_recover
+
+        # Find Julia executable
+        if julia_path is None:
+            julia_path = self._find_julia_executable()
+
+        self.julia_path = julia_path
+
+        # Find worker script
+        self.worker_script = self._find_worker_script()
+
+        # Initialize workers
+        self.workers: list[JuliaWorkerProcess] = []
+        self.available_workers: deque[JuliaWorkerProcess] = deque()
+        self.pool_lock = threading.Lock()
+        self.shutdown_flag = False
+
+        # Create worker processes
+        logger.info(f"Creating Julia process pool with {size} workers")
+        for i in range(size):
+            try:
+                worker = JuliaWorkerProcess(
+                    worker_id=i,
+                    julia_path=self.julia_path,
+                    worker_script=self.worker_script,
+                    optimization_flags=self.optimization_flags,
+                )
+                self.workers.append(worker)
+                self.available_workers.append(worker)
+            except Exception as e:
+                logger.error(f"Failed to create worker {i}: {e}")
+                # Clean up partially created pool
+                self.shutdown()
+                raise RuntimeError(f"Failed to create worker pool: {e}")
+
+        logger.info(f"Julia process pool initialized with {len(self.workers)} workers")
+
+        # Register cleanup on exit
+        atexit.register(self.shutdown)
+
+    def _find_julia_executable(self) -> str:
+        """Find Julia executable in PATH or common locations."""
+        # Try shutil.which first
+        julia_path = shutil.which("julia")
+        if julia_path:
+            return julia_path
+
+        # Try common locations
+        common_paths = [
+            os.path.expanduser("~/.juliaup/bin/julia"),
+            os.path.expanduser("~/.julia/bin/julia"),
+            "/usr/local/bin/julia",
+            "/usr/bin/julia",
+        ]
+
+        for path in common_paths:
+            if os.path.isfile(path) and os.access(path, os.X_OK):
+                return path
+
+        raise RuntimeError(
+            "Julia executable not found. Please install Julia: "
+            "https://julialang.org/downloads/"
+        )
+
+    def _find_worker_script(self) -> str:
+        """Find the julia_repl_worker.jl script."""
+        # Try relative to this file
+        this_dir = Path(__file__).parent
+        worker_script = this_dir / "julia_repl_worker.jl"
+
+        if worker_script.exists():
+            return str(worker_script)
+
+        raise RuntimeError(
+            f"Worker script not found at {worker_script}. "
+            "Please ensure julia_repl_worker.jl is in the same directory."
+        )
+
+    def _get_available_worker(
+        self, timeout: float = 30.0
+    ) -> Optional[JuliaWorkerProcess]:
+        """
+        Get an available worker from the pool.
+
+        Args:
+            timeout: Maximum time to wait for a worker (seconds)
+
+        Returns:
+            Available worker or None if timeout
+        """
+        start_time = time.time()
+
+        while time.time() - start_time < timeout:
+            with self.pool_lock:
+                # Try to get healthy worker
+                while self.available_workers:
+                    worker = self.available_workers.popleft()
+
+                    if worker.is_healthy:
+                        return worker
+
+                    # Worker is unhealthy, try to recover
+                    if self.auto_recover and not self.shutdown_flag:
+                        worker = self._recover_worker(worker)
+                        if worker.is_healthy:
+                            return worker
+                        # Recovery failed, continue to next worker
+
+            # No workers available, wait a bit
+            time.sleep(0.1)
+
+        logger.error("Timeout waiting for available worker")
+        return None
+
+    def _recover_worker(self, worker: JuliaWorkerProcess) -> JuliaWorkerProcess:
+        """
+        Attempt to recover an unhealthy worker by restarting it.
+
+        Args:
+            worker: The unhealthy worker to recover
+
+        Returns:
+            The recovered worker (new instance) or the original if recovery fails
+        """
+        logger.warning(
+            f"Worker {worker.worker_id} is unhealthy, attempting recovery"
+        )
+        try:
+            worker.shutdown()
+            new_worker = JuliaWorkerProcess(
+                worker_id=worker.worker_id,
+                julia_path=self.julia_path,
+                worker_script=self.worker_script,
+                optimization_flags=self.optimization_flags,
+            )
+            # Update in workers list
+            self.workers[new_worker.worker_id] = new_worker
+            logger.info(f"Worker {new_worker.worker_id} recovered successfully")
+            return new_worker
+        except Exception as e:
+            logger.error(
+                f"Failed to recover worker {worker.worker_id}: {e}"
+            )
+            # Return original worker - it will be retried next time
+            return worker
+
+    def _return_worker(self, worker: JuliaWorkerProcess) -> None:
+        """
+        Return a worker to the available pool.
+
+        If the worker is unhealthy and auto_recover is enabled, attempts to
+        recover the worker before returning it to the pool. This ensures
+        workers are not leaked when they fail during execution.
+        """
+        with self.pool_lock:
+            if self.shutdown_flag:
+                return
+
+            # If worker is unhealthy, try to recover it immediately
+            if not worker.is_healthy and self.auto_recover:
+                worker = self._recover_worker(worker)
+
+            # Always return the worker to the pool (healthy or not)
+            # Unhealthy workers will be recovered when next acquired
+            self.available_workers.append(worker)
+
+    def execute(self, code: str, timeout: Optional[int] = None) -> CodeExecResult:
+        """
+        Execute Julia code using an available worker from the pool.
+
+        Args:
+            code: Julia code to execute
+            timeout: Execution timeout in seconds (uses pool default if None)
+
+        Returns:
+            CodeExecResult with stdout, stderr, and exit_code
+        """
+        if self.shutdown_flag:
+            return CodeExecResult(
+                stdout="",
+                stderr="Process pool has been shut down",
+                exit_code=-1,
+            )
+
+        if timeout is None:
+            timeout = self.timeout
+
+        # Get available worker
+        worker = self._get_available_worker()
+
+        if worker is None:
+            return CodeExecResult(
+                stdout="",
+                stderr="No available worker (timeout waiting for worker)",
+                exit_code=-1,
+            )
+
+        try:
+            # Execute code in worker
+            result = worker.execute(code, timeout=timeout)
+            return result
+
+        finally:
+            # Return worker to pool
+            self._return_worker(worker)
+
+    def shutdown(self) -> None:
+        """
+        Shutdown all worker processes gracefully.
+
+        This method is automatically called on exit via atexit.
+        """
+        if self.shutdown_flag:
+            return
+
+        logger.info("Shutting down Julia process pool")
+        self.shutdown_flag = True
+
+        with self.pool_lock:
+            for worker in self.workers:
+                try:
+                    worker.shutdown()
+                except Exception as e:
+                    logger.error(f"Error shutting down worker: {e}")
+
+            self.workers.clear()
+            self.available_workers.clear()
+
+        logger.info("Julia process pool shutdown complete")
+
+    def get_stats(self) -> dict:
+        """Get pool statistics."""
+        with self.pool_lock:
+            healthy_workers = sum(1 for w in self.workers if w.is_healthy)
+            available = len(self.available_workers)
+
+        return {
+            "total_workers": self.size,
+            "healthy_workers": healthy_workers,
+            "available_workers": available,
+            "shutdown": self.shutdown_flag,
+        }
+
+    def __enter__(self):
+        """Context manager entry."""
+        return self
+
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        """Context manager exit."""
+        self.shutdown()
+
+    def __del__(self):
+        """Ensure cleanup on garbage collection."""
+        self.shutdown()

server/julia_repl_worker.jl

@@ -0,0 +1,209 @@
+#!/usr/bin/env julia
+
+"""
+Julia REPL Worker for Process Pool
+
+This script runs as a persistent Julia process that accepts code via stdin,
+executes it, and returns results via stdout with delimiters.
+
+Protocol:
+- Input: Code block followed by "<<<END_CODE>>>"
+- Output: Results with status markers:
+  - "<<<START_OUTPUT>>>" - stdout begins
+  - "<<<START_ERROR>>>" - stderr begins
+  - "<<<EXIT_CODE:N>>>" - exit code (0 = success, 1 = error)
+  - "<<<END_EXECUTION>>>" - execution complete
+"""
+
+# Preload commonly used modules to reduce JIT compilation overhead
+# This is done once at worker startup, so subsequent code executions are faster
+using Test
+
+# Warm up the Test framework and common operations by running simple tests
+# This pre-compiles the @test and @testset macros plus common Julia operations
+let
+    # Warm up Test framework
+    @testset "warmup" begin
+        @test 1 + 1 == 2
+        @test "hello" == "hello"
+        @test [1, 2, 3] == [1, 2, 3]
+    end
+
+    # Warm up common string operations
+    occursin("a", "abc")
+    replace("hello", "l" => "x")
+    split("a,b,c", ",")
+    join(["a", "b"], ",")
+    lowercase("ABC")
+    uppercase("abc")
+    strip("  hello  ")
+
+    # Warm up common array operations
+    arr = [3, 1, 2]
+    push!(arr, 4)
+    pop!(arr)
+    append!(arr, [5, 6])
+    sort(arr)
+    reverse(arr)
+    length(arr)
+    isempty(Int[])
+
+    # Warm up common math operations
+    div(10, 3)
+    mod(10, 3)
+    abs(-5)
+    sqrt(4.0)
+    floor(Int, 3.7)
+    ceil(Int, 3.2)
+
+    # Warm up Dict operations
+    d = Dict("a" => 1, "b" => 2)
+    get(d, "a", 0)
+    haskey(d, "a")
+    keys(d)
+    values(d)
+end
+
+# Delimiters for communication protocol
+const START_OUTPUT = "<<<START_OUTPUT>>>"
+const START_ERROR = "<<<START_ERROR>>>"
+const EXIT_CODE_PREFIX = "<<<EXIT_CODE:"
+const END_EXECUTION = "<<<END_EXECUTION>>>"
+const END_CODE = "<<<END_CODE>>>"
+
+"""
+Execute code and capture output using pipes
+"""
+function execute_code(code::String)
+    # Initialize return values
+    stdout_str = ""
+    stderr_str = ""
+    exit_code = 0
+
+    # Create pipes for output capture
+    out_pipe = Pipe()
+    err_pipe = Pipe()
+
+    try
+        # Execute with output redirected to pipes
+        redirect_stdout(out_pipe) do
+            redirect_stderr(err_pipe) do
+                try
+                    # Execute the code using include_string which properly handles
+                    # multiple statements including 'using' statements
+                    include_string(Main, code)
+                catch e
+                    # Execution error - write to stderr
+                    exit_code = 1
+                    showerror(stderr, e, catch_backtrace())
+                    println(stderr)
+                end
+            end
+        end
+
+        # Close write ends to signal EOF to readers
+        Base.close(out_pipe.in)
+        Base.close(err_pipe.in)
+
+        # Read captured output
+        stdout_str = read(out_pipe.out, String)
+        stderr_str = read(err_pipe.out, String)
+
+        # Close read ends
+        Base.close(out_pipe.out)
+        Base.close(err_pipe.out)
+
+    catch e
+        # Worker error
+        exit_code = 1
+
+        # Try to close pipes
+        try
+            Base.close(out_pipe)
+            Base.close(err_pipe)
+        catch
+        end
+
+        stderr_str = "Worker error: " * sprint(showerror, e)
+    end
+
+    return (stdout_str, stderr_str, exit_code)
+end
+
+"""
+Main loop: read code, execute, return results
+"""
+function main()
+    # Signal that worker is ready
+    println(stderr, "Julia worker ready")
+    flush(stderr)
+
+    while true
+        try
+            # Read code until END_CODE delimiter
+            code_lines = String[]
+
+            while true
+                if eof(stdin)
+                    println(stderr, "Worker received EOF, shutting down")
+                    return
+                end
+
+                line = readline(stdin)
+
+                # Check for end of code
+                if line == END_CODE
+                    break
+                end
+
+                push!(code_lines, line)
+            end
+
+            # If no code received, continue
+            if isempty(code_lines)
+                # Send empty response
+                println(START_OUTPUT)
+                println(START_ERROR)
+                println(EXIT_CODE_PREFIX, 0, ">>>")
+                println(END_EXECUTION)
+                flush(stdout)
+                continue
+            end
+
+            code = join(code_lines, "\n")
+
+            # Execute code and capture output
+            (stdout_str, stderr_str, exit_code) = execute_code(code)
+
+            # Send results with delimiters
+            println(START_OUTPUT)
+            print(stdout_str)
+            flush(stdout)
+
+            println(START_ERROR)
+            print(stderr_str)
+            flush(stdout)
+
+            println(EXIT_CODE_PREFIX, exit_code, ">>>")
+            println(END_EXECUTION)
+            flush(stdout)
+
+        catch e
+            # Worker error - report and continue
+            println(stderr, "Worker error: ", e)
+            flush(stderr)
+
+            # Send error response
+            println(START_OUTPUT)
+            println(START_ERROR)
+            println("Worker internal error: ", e)
+            println(EXIT_CODE_PREFIX, 1, ">>>")
+            println(END_EXECUTION)
+            flush(stdout)
+        end
+    end
+end
+
+# Run main loop
+main()
+

server/julia_transforms.py

@@ -0,0 +1,75 @@
+# Copyright (c) Meta Platforms, Inc. and affiliates.
+# All rights reserved.
+#
+# This source code is licensed under the BSD-style license found in the
+# LICENSE file in the root directory of this source tree.
+
+"""
+envs/julia_env/julia_transforms.py
+--------------------------------
+Safety and quality transforms for Julia code.
+"""
+
+import re
+
+# Support both in-repo and standalone imports
+try:
+    # In-repo imports
+    from openenv.core.env_server.interfaces import Transform
+    from ..models import JuliaObservation
+except ImportError:
+    # Standalone imports
+    from openenv.core.env_server.interfaces import Transform
+    from models import JuliaObservation
+
+
+# -------------------------
+# Safety Transform
+# -------------------------
+class JuliaSafetyTransform(Transform):
+    """Detects dangerous Julia operations and penalizes them with a negative reward."""
+
+    def __init__(self, penalty: float = -3.0):
+        self.penalty = penalty
+        self.dangerous_patterns = [
+            r"run\(",
+            r"read\(",
+            r"write\(",
+            r"unsafe_",
+            r"ccall\(",
+            r"Base\.exit",
+            r"Base\.kill",
+            r"rm\(",      # file deletion
+            r"download\(" # downloading
+        ]
+
+    def __call__(self, observation):
+        # Only act on JuliaObservation objects
+        if not isinstance(observation, JuliaObservation):
+            return observation
+
+        # Extract executed code from metadata (core_code + test_code)
+        if observation.metadata:
+            code = observation.metadata.get("core_code", "") + "\n" + observation.metadata.get("test_code", "")
+        else:
+            code = ""
+
+        for pattern in self.dangerous_patterns:
+            if re.search(pattern, code):
+                # Apply penalty and record violation
+                observation.reward = (observation.reward or 0.0) + self.penalty
+                observation.metadata = observation.metadata or {}
+                observation.metadata["safety_violation"] = pattern
+                return observation
+
+        # Safe code gets neutral reward
+        observation.reward = observation.reward or 0.0
+        return observation
+
+
+# -------------------------
+# Factory
+# -------------------------
+def create_safe_julia_transform():
+    """Creates safety transform for Julia code."""
+    return JuliaSafetyTransform()

server/requirements.txt

@@ -0,0 +1,6 @@
+# Core OpenEnv dependencies
+fastapi>=0.115.0
+pydantic>=2.0.0
+uvicorn[standard]>=0.24.0
+requests>=2.31.0
+websockets>=12.0

test_julia_env.sh

@@ -0,0 +1,215 @@
+#!/bin/bash
+# Julia Environment Test Script
+# Tests infinite loop timeout handling and worker recovery
+#
+# Usage: ./test_julia_env.sh [HOST] [PORT]
+# Default: localhost:8000
+
+set -e
+
+HOST="${1:-localhost}"
+PORT="${2:-8000}"
+BASE_URL="http://${HOST}:${PORT}"
+
+GREEN='\033[0;32m'
+RED='\033[0;31m'
+YELLOW='\033[1;33m'
+NC='\033[0m' # No Color
+
+echo "=============================================="
+echo "Julia Environment Test Suite"
+echo "Testing: ${BASE_URL}"
+echo "=============================================="
+echo ""
+
+# Function to print test results
+pass() {
+    echo -e "${GREEN}✓ PASS${NC}: $1"
+}
+
+fail() {
+    echo -e "${RED}✗ FAIL${NC}: $1"
+    FAILED=1
+}
+
+warn() {
+    echo -e "${YELLOW}! WARN${NC}: $1"
+}
+
+FAILED=0
+
+# Test 1: Health Check
+echo "--- Test 1: Health Check ---"
+HEALTH=$(curl -s --max-time 10 "${BASE_URL}/health" 2>/dev/null || echo "TIMEOUT")
+if [[ "$HEALTH" == *"healthy"* ]]; then
+    pass "Server is healthy"
+    echo "  Response: $HEALTH"
+else
+    fail "Health check failed: $HEALTH"
+fi
+echo ""
+
+# Test 2: Pool Status
+echo "--- Test 2: Pool Status ---"
+POOL=$(curl -s --max-time 10 "${BASE_URL}/pool_status" 2>/dev/null || echo "TIMEOUT")
+if [[ "$POOL" == *"pool_enabled"* ]]; then
+    pass "Pool status retrieved"
+    echo "  Response: $POOL"
+else
+    fail "Pool status failed: $POOL"
+fi
+echo ""
+
+# Test 3: Basic Execution
+echo "--- Test 3: Basic Julia Execution ---"
+BASIC_RESULT=$(curl -s --max-time 30 -X POST "${BASE_URL}/step" \
+    -H "Content-Type: application/json" \
+    -d '{"action": {"core_code": "println(\"Hello from Julia!\")", "test_code": ""}}' 2>/dev/null || echo "TIMEOUT")
+if [[ "$BASIC_RESULT" == *"Hello from Julia"* ]]; then
+    pass "Basic execution works"
+    echo "  Response: $BASIC_RESULT"
+else
+    fail "Basic execution failed: $BASIC_RESULT"
+fi
+echo ""
+
+# Test 4: Tests Passing
+echo "--- Test 4: Test Execution (Passing) ---"
+TEST_RESULT=$(curl -s --max-time 30 -X POST "${BASE_URL}/step" \
+    -H "Content-Type: application/json" \
+    -d '{
+        "action": {
+            "core_code": "function fib(n)\n    n <= 1 ? n : fib(n-1) + fib(n-2)\nend",
+            "test_code": "using Test\n@testset \"Fibonacci\" begin\n    @test fib(0) == 0\n    @test fib(1) == 1\n    @test fib(5) == 5\n    @test fib(10) == 55\nend"
+        }
+    }' 2>/dev/null || echo "TIMEOUT")
+if [[ "$TEST_RESULT" == *"tests_passed"* ]] && [[ "$TEST_RESULT" == *"4"* ]]; then
+    pass "Test execution works (4 tests passed)"
+    echo "  Response: $TEST_RESULT"
+else
+    fail "Test execution failed: $TEST_RESULT"
+fi
+echo ""
+
+# Test 5: Syntax Error Detection
+echo "--- Test 5: Syntax Error Detection ---"
+SYNTAX_RESULT=$(curl -s --max-time 30 -X POST "${BASE_URL}/step" \
+    -H "Content-Type: application/json" \
+    -d '{"action": {"core_code": "function broken(\n    # missing end", "test_code": ""}}' 2>/dev/null || echo "TIMEOUT")
+if [[ "$SYNTAX_RESULT" == *"code_compiles"*"false"* ]] || [[ "$SYNTAX_RESULT" == *"error"* ]] || [[ "$SYNTAX_RESULT" == *"Error"* ]]; then
+    pass "Syntax error detected"
+    echo "  Response: $SYNTAX_RESULT"
+else
+    warn "Syntax error may not be properly detected: $SYNTAX_RESULT"
+fi
+echo ""
+
+# Test 6: Infinite Loop Timeout (CRITICAL TEST)
+echo "--- Test 6: Infinite Loop Timeout (CRITICAL) ---"
+echo "  This test checks if infinite loops are properly terminated."
+echo "  Expected: Request should timeout and return an error within pool timeout."
+echo "  Starting..."
+START_TIME=$(date +%s)
+TIMEOUT_RESULT=$(curl -s --max-time 300 -X POST "${BASE_URL}/step" \
+    -H "Content-Type: application/json" \
+    -d '{"action": {"core_code": "while true\n    sleep(1)\nend", "test_code": ""}}' 2>/dev/null || echo "CURL_TIMEOUT")
+END_TIME=$(date +%s)
+ELAPSED=$((END_TIME - START_TIME))
+
+if [[ "$TIMEOUT_RESULT" == "CURL_TIMEOUT" ]]; then
+    fail "Request timed out after ${ELAPSED}s (curl timeout). Server may be stuck."
+elif [[ "$TIMEOUT_RESULT" == *"timed out"* ]] || [[ "$TIMEOUT_RESULT" == *"Execution timed out"* ]]; then
+    pass "Infinite loop properly timed out after ${ELAPSED}s"
+    echo "  Response: $TIMEOUT_RESULT"
+elif [[ "$TIMEOUT_RESULT" == *"exit_code"*"-1"* ]]; then
+    pass "Infinite loop terminated with exit_code -1 after ${ELAPSED}s"
+    echo "  Response: $TIMEOUT_RESULT"
+else
+    warn "Unexpected response after ${ELAPSED}s: $TIMEOUT_RESULT"
+fi
+echo ""
+
+# Test 7: Worker Recovery After Timeout
+echo "--- Test 7: Worker Recovery After Timeout ---"
+echo "  Testing if the server can handle requests after a timeout..."
+RECOVERY_RESULT=$(curl -s --max-time 30 -X POST "${BASE_URL}/step" \
+    -H "Content-Type: application/json" \
+    -d '{"action": {"core_code": "println(\"Recovery test successful!\")", "test_code": ""}}' 2>/dev/null || echo "TIMEOUT")
+if [[ "$RECOVERY_RESULT" == *"Recovery test successful"* ]]; then
+    pass "Worker recovery successful - server still operational"
+    echo "  Response: $RECOVERY_RESULT"
+else
+    fail "Worker recovery failed: $RECOVERY_RESULT"
+fi
+echo ""
+
+# Test 8: Pool Status After Timeout
+echo "--- Test 8: Pool Status After Timeout ---"
+POOL_AFTER=$(curl -s --max-time 10 "${BASE_URL}/pool_status" 2>/dev/null || echo "TIMEOUT")
+if [[ "$POOL_AFTER" == *"pool_enabled"*"true"* ]]; then
+    pass "Pool is still enabled after timeout"
+    echo "  Response: $POOL_AFTER"
+else
+    fail "Pool status check failed: $POOL_AFTER"
+fi
+echo ""
+
+# Test 9: Multiple Consecutive Timeouts
+echo "--- Test 9: Multiple Consecutive Timeouts (3x) ---"
+echo "  Testing if multiple timeouts exhaust the worker pool..."
+for i in 1 2 3; do
+    echo "  Timeout request $i/3..."
+    curl -s --max-time 300 -X POST "${BASE_URL}/step" \
+        -H "Content-Type: application/json" \
+        -d '{"action": {"core_code": "while true; sleep(1); end", "test_code": ""}}' >/dev/null 2>&1
+done
+
+echo "  Testing normal operation after 3 timeouts..."
+AFTER_MULTI=$(curl -s --max-time 30 -X POST "${BASE_URL}/step" \
+    -H "Content-Type: application/json" \
+    -d '{"action": {"core_code": "println(\"Still working after 3 timeouts!\")", "test_code": ""}}' 2>/dev/null || echo "TIMEOUT")
+if [[ "$AFTER_MULTI" == *"Still working after 3 timeouts"* ]]; then
+    pass "Server handles multiple consecutive timeouts correctly"
+else
+    fail "Server failed after multiple timeouts: $AFTER_MULTI"
+fi
+echo ""
+
+# Test 10: Concurrent Request Handling
+echo "--- Test 10: Concurrent Request Handling ---"
+echo "  Sending 4 requests in parallel..."
+# Create temp files for results
+TMPDIR=$(mktemp -d)
+for i in 1 2 3 4; do
+    curl -s --max-time 60 -X POST "${BASE_URL}/step" \
+        -H "Content-Type: application/json" \
+        -d "{\"action\": {\"core_code\": \"println(\\\"Request $i\\\")\", \"test_code\": \"\"}}" > "${TMPDIR}/result_${i}.txt" 2>&1 &
+done
+wait
+
+CONCURRENT_SUCCESS=0
+for i in 1 2 3 4; do
+    if grep -q "Request $i" "${TMPDIR}/result_${i}.txt" 2>/dev/null; then
+        ((CONCURRENT_SUCCESS++))
+    fi
+done
+rm -rf "$TMPDIR"
+
+if [[ $CONCURRENT_SUCCESS -eq 4 ]]; then
+    pass "All 4 concurrent requests succeeded"
+else
+    fail "Only $CONCURRENT_SUCCESS/4 concurrent requests succeeded"
+fi
+echo ""
+
+# Final Summary
+echo "=============================================="
+echo "Test Summary"
+echo "=============================================="
+if [[ $FAILED -eq 0 ]]; then
+    echo -e "${GREEN}All tests passed!${NC}"
+    exit 0
+else
+    echo -e "${RED}Some tests failed. Please review the output above.${NC}"
+    exit 1
+fi