Upload folder using huggingface_hub

Dockerfile

@@ -0,0 +1,76 @@
+# 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.
+ARG BASE_IMAGE=ghcr.io/meta-pytorch/openenv-base:latest
+FROM ${BASE_IMAGE} AS builder
+
+WORKDIR /app
+
+# Ensure git is available for dependency installation.
+RUN apt-get update && \
+    apt-get install -y --no-install-recommends git && \
+    rm -rf /var/lib/apt/lists/*
+
+# Build argument to control whether we're building standalone or in-repo
+ARG BUILD_MODE=in-repo
+ARG ENV_NAME=terminus_env
+
+# Copy environment code (always at root of build context)
+COPY . /app/env
+
+# For in-repo builds, openenv is already vendored in the build context
+# For standalone builds, openenv will be installed via pyproject.toml
+WORKDIR /app/env
+
+# Ensure uv is available (for local builds where base image lacks it)
+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
+    
+# Install dependencies using uv sync
+# If uv.lock exists, use it; otherwise resolve on the fly
+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
+
+# 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"
+ENV ENABLE_WEB_INTERFACE=true
+
+# Health check
+HEALTHCHECK --interval=30s --timeout=3s --start-period=5s --retries=3 \
+    CMD python -c "import urllib.request; urllib.request.urlopen('http://localhost:8000/health', timeout=2).read()" || exit 1
+
+# Run the FastAPI server
+# The module path is constructed to work with the /app/env structure
+CMD ["sh", "-c", "cd /app/env && uvicorn server.app:app --host 0.0.0.0 --port 8000"]

README.md

@@ -1,10 +1,95 @@
 ---
-title: Terminus Env
-emoji: 🔥
-colorFrom: indigo
+title: Terminus Environment Server
+emoji: 🛠️
+colorFrom: green
 colorTo: blue
 sdk: docker
 pinned: false
+app_port: 8000
+base_path: /web
+tags:
+  - openenv
+  - terminus
+  - hf-sandbox
+  - coding
+short_description: Single-tool HF Sandbox-backed coding environment
 ---
 
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+# Terminus Environment
+
+`terminus_env` is a single-tool coding environment backed by
+[hf-sandbox](https://github.com/huggingface/hf-sandbox). Each OpenEnv episode
+creates a fresh Hugging Face Jobs-backed sandbox, runs optional
+setup commands, keeps shell state and files isolated for that episode, and runs
+optional verify commands when the agent submits a final answer.
+
+The tool shape follows the Terminus-style "one tool" idea: agents do their work
+through a single terminal entrypoint rather than a notebook/toolbox surface.
+
+## Tool
+
+- `terminal(command="", final_answer="")`: run a shell command inside the
+  session sandbox, or submit a final answer and run verification.
+
+## Quick Start
+
+```python
+from terminus_env import TerminusEnv
+
+with TerminusEnv(base_url="http://localhost:8000").sync() as env:
+    env.reset(
+        setup=["mkdir -p /home/user/work"],
+        verify=["test -f /home/user/work/answer.txt"],
+    )
+    print(env.call_tool("terminal", command="echo done > /home/user/work/answer.txt"))
+    print(env.call_tool("terminal", final_answer="done"))
+```
+
+## Local Server
+
+```bash
+cd envs/terminus_env
+HF_TOKEN=hf_... uv run --project . server
+```
+
+The API and custom terminal web UI are served on port 8000. The UI is mounted
+at `/web`.
+
+## Docker
+
+```bash
+cd envs/terminus_env
+openenv build -t terminus-env
+docker run -p 8000:8000 -e HF_TOKEN=hf_... terminus-env
+```
+
+## Configuration
+
+- `HF_TOKEN`: required by `hf-sandbox` to launch Hugging Face Jobs.
+- `HF_SANDBOX_IMAGE`: sandbox image. Defaults to `python:3.12`.
+- `HF_SANDBOX_FLAVOR`: Hugging Face Jobs flavor. Defaults to `cpu-basic`.
+- `HF_SANDBOX_TIMEOUT`: Hugging Face Jobs timeout. Defaults to `1h`.
+- `HF_SANDBOX_FORWARD_HF_TOKEN`: forward `HF_TOKEN` into the sandbox. Defaults
+  to `false`.
+- `MAX_CONCURRENT_ENVS`: maximum concurrent WebSocket sessions. Defaults to `4`.
+
+## Setup and Verify Commands
+
+`reset()` accepts either `setup` / `verify` or `setup_scripts` /
+`verify_scripts`.
+
+```python
+env.reset(
+    setup=["pip install -q pytest"],
+    verify=["pytest -q /home/user/work/tests"],
+)
+```
+
+Setup failure ends the reset response with `done=True` and returns captured
+setup results. Verify commands run when `terminal(final_answer="...")` is
+called. Reward defaults to `passed_verify_commands / total_verify_commands`. A
+verify command can override this by writing a float to:
+
+```text
+/home/user/logs/verifier/reward.txt
+```

__init__.py

@@ -0,0 +1,23 @@
+# 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.
+
+"""Terminus Environment for OpenEnv."""
+
+from openenv.core.env_server.mcp_types import CallToolAction, ListToolsAction
+
+from .client import TerminusEnv
+from .harness import TerminusSessionFactory, build_terminal_tool_call
+from .models import CommandResult, TerminusState
+
+__all__ = [
+    "TerminusEnv",
+    "TerminusSessionFactory",
+    "TerminusState",
+    "CommandResult",
+    "CallToolAction",
+    "ListToolsAction",
+    "build_terminal_tool_call",
+]

client.py

@@ -0,0 +1,43 @@
+# 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.
+
+"""Client for the Terminus environment."""
+
+from typing import Any
+
+from openenv.core.mcp_client import MCPToolClient
+
+from .models import CommandResult, TerminusState
+
+
+class TerminusEnv(MCPToolClient):
+    """MCP client for calling the Terminus single-rollout tool."""
+
+    def _parse_state(self, payload: dict[str, Any]) -> TerminusState:
+        """Convert server state payloads to the Terminus state model."""
+
+        def command_results(name: str) -> list[CommandResult]:
+            values = payload.get(name, [])
+            if not isinstance(values, list):
+                return []
+            return [
+                value if isinstance(value, CommandResult) else CommandResult(**value)
+                for value in values
+                if isinstance(value, dict) or isinstance(value, CommandResult)
+            ]
+
+        return TerminusState(
+            episode_id=payload.get("episode_id"),
+            step_count=payload.get("step_count", 0),
+            sandbox_id=payload.get("sandbox_id"),
+            setup_results=command_results("setup_results"),
+            verify_commands=list(payload.get("verify_commands", []) or []),
+            verify_results=command_results("verify_results"),
+            commands=command_results("commands"),
+            submitted_answer=payload.get("submitted_answer"),
+            last_reward=payload.get("last_reward"),
+            last_error=payload.get("last_error"),
+        )

harness.py

@@ -0,0 +1,356 @@
+# 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.
+
+"""Harness-oriented Terminus session adapter."""
+
+from __future__ import annotations
+
+import json
+from typing import Any, Callable
+
+from openenv.core.env_server.mcp_types import CallToolAction, Tool
+from openenv.core.harness import (
+    ResourceSessionFactory,
+    StepEnvSessionAdapter,
+    ToolResult,
+    VerifyResult,
+)
+
+from .client import TerminusEnv
+
+_TERMINUS_TOOLS: list[Tool] = [
+    Tool(
+        name="terminal",
+        description=(
+            "Run a shell command in the Terminus sandbox, or submit final_answer "
+            "to trigger verification."
+        ),
+        input_schema={
+            "type": "object",
+            "properties": {
+                "command": {
+                    "type": "string",
+                    "description": "Shell command to run in the sandbox.",
+                },
+                "final_answer": {
+                    "type": "string",
+                    "description": "Final answer to submit when the task is complete.",
+                },
+            },
+            "additionalProperties": False,
+        },
+    )
+]
+
+
+def _task_field(task: Any, *names: str, default: Any = None) -> Any:
+    if not isinstance(task, dict):
+        return default
+    for name in names:
+        value = task.get(name)
+        if value is not None:
+            return value
+    return default
+
+
+def _coerce_commands(value: Any) -> list[str]:
+    if value is None:
+        return []
+    if isinstance(value, str):
+        return [value] if value.strip() else []
+    return [str(item) for item in value if str(item).strip()]
+
+
+def _format_initial_prompt(result: Any, task: Any) -> str:
+    if isinstance(task, str):
+        instruction = task
+        setup_commands: list[str] = []
+        verify_commands: list[str] = []
+    elif isinstance(task, list):
+        user_messages = [
+            item.get("content")
+            for item in task
+            if isinstance(item, dict) and item.get("role") == "user"
+        ]
+        instruction = str(user_messages[-1] if user_messages else task)
+        setup_commands = []
+        verify_commands = []
+    elif isinstance(task, dict):
+        instruction = str(
+            _task_field(task, "instruction", "prompt", "question", "task", default="")
+        )
+        setup_commands = _coerce_commands(_task_field(task, "setup", "setup_scripts"))
+        verify_commands = _coerce_commands(_task_field(task, "verify", "verify_scripts"))
+    else:
+        instruction = str(task or "")
+        setup_commands = []
+        verify_commands = []
+
+    metadata = getattr(result.observation, "metadata", {}) or {}
+    verify_commands = _coerce_commands(
+        metadata.get("verify_commands") or verify_commands
+    )
+
+    parts = []
+    if instruction:
+        parts.append(f"Task:\n{instruction}")
+    else:
+        parts.append("Task:\nUse the terminal tool to solve the current task.")
+
+    reset_message = metadata.get("message")
+    if reset_message:
+        parts.append(f"Environment:\n{reset_message}")
+
+    if setup_commands:
+        parts.append(
+            "Setup commands have already run:\n"
+            + "\n".join(f"- {command}" for command in setup_commands)
+        )
+    if verify_commands:
+        parts.append(
+            "Verification commands will run after final_answer:\n"
+            + "\n".join(f"- {command}" for command in verify_commands)
+        )
+
+    parts.append(
+        "Use terminal(command=...) to inspect and modify the sandbox. "
+        "When finished, call terminal(final_answer=...) exactly once so "
+        "verification runs and emits the environment reward."
+    )
+    return "\n\n".join(parts)
+
+
+def _extract_tool_output(observation: Any) -> Any:
+    result = getattr(observation, "result", None)
+    if result is None:
+        return None
+    if hasattr(result, "data"):
+        return result.data
+    if isinstance(result, dict):
+        if "data" in result:
+            return result["data"]
+        content = result.get("content")
+        if isinstance(content, list):
+            texts = [
+                str(item.get("text"))
+                for item in content
+                if isinstance(item, dict) and item.get("text") is not None
+            ]
+            if texts:
+                return "\n".join(texts)
+        return result
+    content = getattr(result, "content", None)
+    if isinstance(content, list):
+        texts = [
+            getattr(item, "text", None)
+            for item in content
+            if getattr(item, "text", None) is not None
+        ]
+        if texts:
+            return "\n".join(texts)
+    return result
+
+
+def _tool_error_message(observation: Any) -> str | None:
+    error = getattr(observation, "error", None)
+    if error is None:
+        return None
+    message = getattr(error, "message", None)
+    if message is not None:
+        return str(message)
+    if isinstance(error, dict):
+        return str(error.get("message") or error)
+    return str(error)
+
+
+def _state_to_data(state: Any) -> Any:
+    if state is None:
+        return None
+    if hasattr(state, "model_dump"):
+        return state.model_dump()
+    return state
+
+
+def _build_tool_result(
+    tool_name: str,
+    arguments: dict[str, Any],
+    result: Any,
+    state: Any,
+) -> ToolResult:
+    output = _extract_tool_output(result.observation)
+    error = _tool_error_message(result.observation)
+    data = {
+        "tool_name": tool_name,
+        "arguments": dict(arguments),
+        "output": output,
+        "reward": result.reward,
+        "done": result.done,
+    }
+    if error:
+        data["error"] = error
+
+    return ToolResult(
+        data=data,
+        done=bool(result.done),
+        error=error,
+        metadata={
+            "reward": result.reward,
+            "state": _state_to_data(state),
+        },
+    )
+
+
+def _build_verify(
+    transcript: list[dict[str, Any]],
+    final_state: Any | None,
+    last_result: Any | None,
+    state: Any,
+) -> VerifyResult:
+    reward = None if last_result is None else last_result.reward
+    done = False if last_result is None else bool(last_result.done)
+    state_data = _state_to_data(state)
+    metrics = {
+        "done": done,
+        "step_count": getattr(state, "step_count", 0),
+        "commands": len(getattr(state, "commands", []) or []),
+        "verify_commands": len(getattr(state, "verify_commands", []) or []),
+        "setup_commands": len(getattr(state, "setup_results", []) or []),
+        "submitted_answer": getattr(state, "submitted_answer", None) is not None,
+        "sandbox_id": getattr(state, "sandbox_id", None),
+    }
+    if state is None and last_result is not None:
+        metrics["step_count"] = len(transcript)
+    return VerifyResult(
+        env_reward=reward,
+        done=done,
+        metrics=metrics,
+        artifacts={
+            "final_state": state_data,
+            "final_rollout": final_state,
+            "transcript_length": len(transcript),
+        },
+    )
+
+
+def _build_reset_kwargs(
+    task: Any,
+    default_setup: list[str],
+    default_verify: list[str],
+    default_sandbox: dict[str, Any],
+) -> dict[str, Any]:
+    reset_kwargs: dict[str, Any] = dict(default_sandbox)
+    setup = list(default_setup)
+    verify = list(default_verify)
+    if isinstance(task, dict):
+        setup = _coerce_commands(_task_field(task, "setup", "setup_scripts", default=setup))
+        verify = _coerce_commands(
+            _task_field(task, "verify", "verify_scripts", default=verify)
+        )
+        for key in (
+            "sandbox_image",
+            "sandbox_flavor",
+            "sandbox_timeout",
+            "hf_sandbox_image",
+            "hf_sandbox_flavor",
+            "hf_sandbox_timeout",
+            "forward_hf_token",
+        ):
+            if key in task:
+                reset_kwargs[key] = task[key]
+
+    if setup:
+        reset_kwargs["setup"] = setup
+    if verify:
+        reset_kwargs["verify"] = verify
+    return reset_kwargs
+
+
+class TerminusSessionFactory(ResourceSessionFactory):
+    """Create Terminus-backed resource sessions for harness rollouts."""
+
+    def __init__(
+        self,
+        client_factory: Callable[[], TerminusEnv],
+        *,
+        default_setup: list[str] | None = None,
+        default_verify: list[str] | None = None,
+        sandbox: dict[str, Any] | None = None,
+    ):
+        self._client_factory = client_factory
+        self._default_setup = list(default_setup or [])
+        self._default_verify = list(default_verify or [])
+        self._sandbox = dict(sandbox or {})
+
+    def create(
+        self,
+        task: Any = None,
+        seed: int | None = None,
+        episode_id: str | None = None,
+    ) -> StepEnvSessionAdapter:
+        reset_kwargs = _build_reset_kwargs(
+            task,
+            self._default_setup,
+            self._default_verify,
+            self._sandbox,
+        )
+
+        return StepEnvSessionAdapter(
+            client=self._client_factory(),
+            task=task,
+            seed=seed,
+            episode_id=episode_id,
+            tool_specs=list(_TERMINUS_TOOLS),
+            action_builder=lambda name, arguments: CallToolAction(
+                tool_name=name,
+                arguments=dict(arguments),
+            ),
+            initial_messages_builder=lambda result, current_task: [
+                {
+                    "role": "user",
+                    "content": _format_initial_prompt(result, current_task),
+                }
+            ],
+            tool_result_builder=_build_tool_result,
+            verify_builder=_build_verify,
+            reset_kwargs=reset_kwargs,
+        )
+
+
+def build_terminal_tool_call(response_text: str, *, call_id: str = "terminal-0"):
+    """Parse a simple JSON terminal call from model text.
+
+    The preferred format is one JSON object containing ``command`` or
+    ``final_answer``. Invalid text falls back to a no-op shell command so the
+    environment, not this parser, decides whether a rollout earns reward.
+    """
+
+    from openenv.core.llm_client import ToolCall
+
+    text = response_text.strip()
+    if text.startswith("```"):
+        text = text.strip("`").strip()
+        if text.startswith("json"):
+            text = text[4:].strip()
+    try:
+        payload = json.loads(text)
+    except json.JSONDecodeError:
+        payload = {"command": response_text}
+    if not isinstance(payload, dict):
+        payload = {"command": response_text}
+    arguments = {
+        key: str(payload[key])
+        for key in ("command", "final_answer")
+        if payload.get(key) is not None
+    }
+    if not arguments:
+        arguments = {"command": ""}
+    return ToolCall(id=call_id, name="terminal", args=arguments)
+
+
+__all__ = [
+    "TerminusSessionFactory",
+    "build_terminal_tool_call",
+]

models.py

@@ -0,0 +1,34 @@
+# 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.
+
+"""Models for the Terminus environment."""
+
+from __future__ import annotations
+
+from openenv.core.env_server.types import State
+from pydantic import BaseModel, Field
+
+
+class CommandResult(BaseModel):
+    """Outcome of one shell command run inside the HF sandbox."""
+
+    command: str
+    output: str = ""
+    error: str | None = None
+    success: bool = True
+
+
+class TerminusState(State):
+    """Per-session state for the single-tool Terminus environment."""
+
+    sandbox_id: str | None = None
+    setup_results: list[CommandResult] = Field(default_factory=list)
+    verify_commands: list[str] = Field(default_factory=list)
+    verify_results: list[CommandResult] = Field(default_factory=list)
+    commands: list[CommandResult] = Field(default_factory=list)
+    submitted_answer: str | None = None
+    last_reward: float | None = None
+    last_error: str | None = None

openenv.yaml

@@ -0,0 +1,7 @@
+spec_version: 1
+name: terminus_env
+type: space
+runtime: fastapi
+app: server.app:app
+port: 8000
+

openenv_terminus_env.egg-info/PKG-INFO

@@ -0,0 +1,16 @@
+Metadata-Version: 2.4
+Name: openenv-terminus-env
+Version: 0.1.0
+Summary: Single-tool HF Sandbox-backed coding environment for OpenEnv
+Requires-Python: >=3.10
+Requires-Dist: openenv-core[core]>=0.2.2
+Requires-Dist: hf-sandbox>=0.1.1
+Requires-Dist: fastapi>=0.115.0
+Requires-Dist: fastmcp>=3.0.0
+Requires-Dist: gradio>=4.0.0
+Requires-Dist: pydantic>=2.0.0
+Requires-Dist: requests>=2.31.0
+Requires-Dist: uvicorn>=0.24.0
+Provides-Extra: dev
+Requires-Dist: pytest>=8.0.0; extra == "dev"
+Requires-Dist: pytest-cov>=4.0.0; extra == "dev"

openenv_terminus_env.egg-info/SOURCES.txt

@@ -0,0 +1,24 @@
+README.md
+pyproject.toml
+./__init__.py
+./client.py
+./harness.py
+./models.py
+./openenv.yaml
+./openenv_terminus_env.egg-info/SOURCES.txt
+./openenv_terminus_env.egg-info/dependency_links.txt
+./openenv_terminus_env.egg-info/entry_points.txt
+./openenv_terminus_env.egg-info/requires.txt
+./openenv_terminus_env.egg-info/top_level.txt
+./server/requirements.txt
+openenv_terminus_env.egg-info/PKG-INFO
+openenv_terminus_env.egg-info/SOURCES.txt
+openenv_terminus_env.egg-info/dependency_links.txt
+openenv_terminus_env.egg-info/entry_points.txt
+openenv_terminus_env.egg-info/requires.txt
+openenv_terminus_env.egg-info/top_level.txt
+server/__init__.py
+server/app.py
+server/gradio_ui.py
+server/hf_sandbox.py
+server/terminus_env_environment.py

openenv_terminus_env.egg-info/dependency_links.txt

@@ -0,0 +1 @@
+

openenv_terminus_env.egg-info/entry_points.txt

@@ -0,0 +1,2 @@
+[console_scripts]
+server = terminus_env.server.app:main

openenv_terminus_env.egg-info/requires.txt

@@ -0,0 +1,12 @@
+openenv-core[core]>=0.2.2
+hf-sandbox>=0.1.1
+fastapi>=0.115.0
+fastmcp>=3.0.0
+gradio>=4.0.0
+pydantic>=2.0.0
+requests>=2.31.0
+uvicorn>=0.24.0
+
+[dev]
+pytest>=8.0.0
+pytest-cov>=4.0.0

openenv_terminus_env.egg-info/top_level.txt

@@ -0,0 +1 @@
+terminus_env

pyproject.toml

@@ -0,0 +1,42 @@
+# 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-terminus-env"
+version = "0.1.0"
+description = "Single-tool HF Sandbox-backed coding environment for OpenEnv"
+requires-python = ">=3.10"
+dependencies = [
+    "openenv-core[core]>=0.2.2",
+    "hf-sandbox>=0.1.1",
+    "fastapi>=0.115.0",
+    "fastmcp>=3.0.0",
+    "gradio>=4.0.0",
+    "pydantic>=2.0.0",
+    "requests>=2.31.0",
+    "uvicorn>=0.24.0",
+]
+
+[project.optional-dependencies]
+dev = [
+    "pytest>=8.0.0",
+    "pytest-cov>=4.0.0",
+]
+
+[project.scripts]
+server = "terminus_env.server.app:main"
+
+[tool.setuptools]
+include-package-data = true
+packages = ["terminus_env", "terminus_env.server"]
+package-dir = { "terminus_env" = ".", "terminus_env.server" = "server" }
+
+[tool.setuptools.package-data]
+terminus_env = ["**/*.txt", "**/*.yaml"]

server/__init__.py

@@ -0,0 +1,11 @@
+# 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.
+
+"""Terminus Env environment server components."""
+
+from .terminus_env_environment import TerminusEnvironment
+
+__all__ = ["TerminusEnvironment"]

server/app.py

@@ -0,0 +1,60 @@
+# 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 app for the Terminus environment."""
+
+from __future__ import annotations
+
+import os
+from pathlib import Path
+
+from openenv.core.env_server.http_server import create_app
+from openenv.core.env_server.mcp_types import CallToolAction, CallToolObservation
+
+try:
+    from .gradio_ui import terminus_ui_builder
+    from .terminus_env_environment import TerminusEnvironment
+except ImportError:  # pragma: no cover
+    from server.gradio_ui import terminus_ui_builder  # type: ignore
+    from server.terminus_env_environment import TerminusEnvironment  # type: ignore
+
+
+def _load_env_file() -> None:
+    candidate = Path(__file__).resolve().parents[1] / ".env"
+    if not candidate.exists():
+        return
+    for raw in candidate.read_text(encoding="utf-8").splitlines():
+        line = raw.strip()
+        if not line or line.startswith("#") or "=" not in line:
+            continue
+        key, _, value = line.partition("=")
+        key = key.strip()
+        value = value.strip().strip('"').strip("'")
+        if key and key not in os.environ:
+            os.environ[key] = value
+
+
+_load_env_file()
+os.environ.setdefault("ENABLE_WEB_INTERFACE", "true")
+
+app = create_app(
+    TerminusEnvironment,
+    CallToolAction,
+    CallToolObservation,
+    env_name="terminus_env",
+    max_concurrent_envs=int(os.getenv("MAX_CONCURRENT_ENVS", "4")),
+    gradio_builder=terminus_ui_builder,
+)
+
+
+def main(host: str = "0.0.0.0", port: int = 8000) -> None:
+    import uvicorn
+
+    uvicorn.run(app, host=host, port=port)
+
+
+if __name__ == "__main__":
+    main()

server/gradio_ui.py

@@ -0,0 +1,343 @@
+# 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.
+
+"""Custom Gradio terminal UI for the Terminus environment."""
+
+from __future__ import annotations
+
+from typing import Any, Optional
+
+import gradio as gr
+
+
+COMMAND_EXAMPLES = {
+    "Inspect workspace": "pwd && ls -la",
+    "Create answer file": "mkdir -p /home/user/work && echo done > /home/user/work/answer.txt",
+    "Run Python": "python - <<'PY'\nprint('hello from the sandbox')\nPY",
+    "Install package": "pip install -q pytest",
+    "Run tests": "pytest -q",
+}
+
+DEFAULT_SETUP = "mkdir -p /home/user/work"
+DEFAULT_VERIFY = "test -f /home/user/work/answer.txt"
+
+_STYLE = """
+<style>
+.term-shell {
+  background: #020403;
+  border: 1px solid #16a34a;
+  box-shadow: inset 0 0 0 1px #052e16, 0 0 24px rgba(22, 163, 74, 0.18);
+  color: #d1fae5;
+  font-family: ui-monospace, SFMono-Regular, Menlo, Monaco, Consolas, "Liberation Mono", monospace;
+  min-height: 520px;
+  max-height: 720px;
+  overflow-y: auto;
+  padding: 14px;
+}
+.term-empty {
+  color: #6ee7b7;
+  padding: 30px 10px;
+  text-align: center;
+}
+.term-entry {
+  border-bottom: 1px solid #064e3b;
+  padding: 10px 0 12px;
+}
+.term-entry:last-child { border-bottom: 0; }
+.term-meta {
+  color: #34d399;
+  font-size: 12px;
+  margin-bottom: 6px;
+  text-transform: uppercase;
+  letter-spacing: 0;
+}
+.term-prompt {
+  color: #22c55e;
+  white-space: pre-wrap;
+  word-break: break-word;
+}
+.term-prompt::before {
+  color: #86efac;
+}
+.term-output {
+  color: #d1fae5;
+  margin-top: 8px;
+  white-space: pre-wrap;
+  word-break: break-word;
+}
+.term-error {
+  color: #fca5a5;
+  margin-top: 8px;
+  white-space: pre-wrap;
+  word-break: break-word;
+}
+.term-success { color: #86efac; }
+.term-fail { color: #f87171; }
+.term-final {
+  border: 1px solid #16a34a;
+  background: #04130a;
+  margin-top: 12px;
+  padding: 12px;
+}
+</style>
+"""
+
+_EMPTY_TERMINAL = (
+    "<div class='term-shell'><div class='term-empty'>"
+    "Reset the environment to create an HF sandbox, then run commands."
+    "</div></div>"
+)
+
+
+def _escape(value: Any) -> str:
+    return str(value).replace("&", "&amp;").replace("<", "&lt;").replace(">", "&gt;")
+
+
+def _lines(value: str) -> list[str]:
+    return [line.strip() for line in value.splitlines() if line.strip()]
+
+
+def _extract_tool_text(result: dict[str, Any]) -> str:
+    obs = result.get("observation", {})
+    if not isinstance(obs, dict):
+        return ""
+    tool_result = obs.get("result", {})
+    if not isinstance(tool_result, dict):
+        return ""
+    content = tool_result.get("content", [])
+    if isinstance(content, list) and content:
+        first = content[0]
+        if isinstance(first, dict):
+            return str(first.get("text", ""))
+    return str(tool_result.get("data", ""))
+
+
+def _state_summary(state: dict[str, Any]) -> dict[str, Any]:
+    return {
+        "episode_id": state.get("episode_id"),
+        "sandbox_id": state.get("sandbox_id"),
+        "step_count": state.get("step_count", 0),
+        "commands": len(state.get("commands", [])),
+        "setup_commands": len(state.get("setup_results", [])),
+        "verify_commands": len(state.get("verify_commands", [])),
+        "last_reward": state.get("last_reward"),
+        "last_error": state.get("last_error"),
+    }
+
+
+def render_terminal_html(state: dict[str, Any]) -> str:
+    entries: list[str] = []
+
+    for index, item in enumerate(state.get("setup_results", []), start=1):
+        status = "ok" if item.get("success", False) else "failed"
+        status_cls = "term-success" if item.get("success", False) else "term-fail"
+        entries.append(
+            "<div class='term-entry'>"
+            f"<div class='term-meta'>setup {index} - <span class='{status_cls}'>{status}</span></div>"
+            f"<div class='term-prompt'>$ {_escape(item.get('command', ''))}</div>"
+            f"<div class='term-output'>{_escape(item.get('output', ''))}</div>"
+            f"{_error_html(item)}"
+            "</div>"
+        )
+
+    for index, item in enumerate(state.get("commands", []), start=1):
+        status = "ok" if item.get("success", False) else "failed"
+        status_cls = "term-success" if item.get("success", False) else "term-fail"
+        entries.append(
+            "<div class='term-entry'>"
+            f"<div class='term-meta'>command {index} - <span class='{status_cls}'>{status}</span></div>"
+            f"<div class='term-prompt'>$ {_escape(item.get('command', ''))}</div>"
+            f"<div class='term-output'>{_escape(item.get('output', ''))}</div>"
+            f"{_error_html(item)}"
+            "</div>"
+        )
+
+    if state.get("submitted_answer") is not None:
+        answer = _escape(state.get("submitted_answer", ""))
+        reward = _escape(state.get("last_reward", ""))
+        entries.append(
+            "<div class='term-final'>"
+            "<div class='term-meta'>final answer</div>"
+            f"<div>{answer}</div>"
+            f"<div class='term-meta'>reward: {reward}</div>"
+            "</div>"
+        )
+
+    if not entries:
+        return _STYLE + _EMPTY_TERMINAL
+    return _STYLE + "<div class='term-shell'>" + "\n".join(entries) + "</div>"
+
+
+def _closed_terminal_html() -> str:
+    return (
+        _STYLE + "<div class='term-shell'><div class='term-empty'>"
+        "Session closed. Reset the sandbox to start a new terminal."
+        "</div></div>"
+    )
+
+
+def _error_html(item: dict[str, Any]) -> str:
+    if not item.get("error"):
+        return ""
+    return f"<div class='term-error'>{_escape(item['error'])}</div>"
+
+
+def terminus_ui_builder(
+    web_manager,
+    action_fields: list[dict[str, Any]],
+    metadata: Optional[Any],
+    is_chat_env: bool,
+    title: str,
+    quick_start_md: Optional[str],
+) -> gr.Blocks:
+    def current_state() -> dict[str, Any]:
+        try:
+            return web_manager.get_state()
+        except RuntimeError:
+            return {}
+
+    with gr.Blocks(title=f"{title} - Terminal") as demo:
+        gr.Markdown(f"# {title}")
+        gr.Markdown(
+            "Single-tool terminal environment backed by an HF sandbox. "
+            "Reset creates a fresh session; commands run through `terminal(command=...)`."
+        )
+
+        with gr.Row():
+            with gr.Column(scale=1, min_width=320):
+                gr.Markdown("### Episode")
+                setup_input = gr.Textbox(
+                    label="Setup commands",
+                    value=DEFAULT_SETUP,
+                    lines=5,
+                    placeholder="One shell command per line",
+                )
+                verify_input = gr.Textbox(
+                    label="Verify commands",
+                    value=DEFAULT_VERIFY,
+                    lines=4,
+                    placeholder="One shell command per line",
+                )
+                reset_btn = gr.Button("Reset sandbox", variant="primary", size="lg")
+                close_btn = gr.Button("Stop / Close session", variant="secondary")
+                state_box = gr.JSON(label="Session state", value={})
+
+                gr.Markdown("### Command")
+                example_dd = gr.Dropdown(
+                    choices=list(COMMAND_EXAMPLES.keys()),
+                    label="Load example",
+                    value=None,
+                    interactive=True,
+                )
+                command_input = gr.Textbox(
+                    label="Terminal command",
+                    value="pwd && ls -la",
+                    lines=6,
+                    placeholder="Run shell commands in the HF sandbox",
+                )
+                run_btn = gr.Button("Run command", variant="primary")
+
+                gr.Markdown("### Submit")
+                answer_input = gr.Textbox(
+                    label="Final answer",
+                    value="done",
+                    lines=2,
+                )
+                submit_btn = gr.Button("Submit final answer", variant="secondary")
+
+            with gr.Column(scale=2):
+                gr.Markdown("### Terminal")
+                terminal_display = gr.HTML(value=_STYLE + _EMPTY_TERMINAL)
+                status_bar = gr.Textbox(
+                    label="Status",
+                    value="Reset the sandbox before running commands.",
+                    interactive=False,
+                    lines=2,
+                )
+
+        async def on_reset(setup_text: str, verify_text: str):
+            payload = {
+                "setup": _lines(setup_text),
+                "verify": _lines(verify_text),
+            }
+            result = await web_manager.reset_environment(payload)
+            state = current_state()
+            done = result.get("done", False)
+            metadata = result.get("metadata", {})
+            if done and isinstance(metadata, dict):
+                status = (
+                    metadata.get("error") or metadata.get("message") or "Reset failed."
+                )
+            else:
+                status = "Sandbox reset. Setup and verify configuration applied."
+            return render_terminal_html(state), _state_summary(state), status
+
+        async def on_close():
+            await web_manager._run_sync_in_thread_pool(web_manager.env.close)
+            return (
+                _closed_terminal_html(),
+                {},
+                "Session closed. The HF sandbox was released.",
+            )
+
+        async def on_run(command: str):
+            if not command.strip():
+                return gr.update(), gr.update(), "No command provided."
+            result = await web_manager.step_environment(
+                {
+                    "tool_name": "terminal",
+                    "arguments": {"command": command},
+                }
+            )
+            state = current_state()
+            text = _extract_tool_text(result)
+            status = text if text.startswith("Error:") else "Command completed."
+            return render_terminal_html(state), _state_summary(state), status
+
+        async def on_submit(answer: str):
+            if not answer.strip():
+                return gr.update(), gr.update(), "No final answer provided."
+            result = await web_manager.step_environment(
+                {
+                    "tool_name": "terminal",
+                    "arguments": {"final_answer": answer},
+                }
+            )
+            state = current_state()
+            status = _extract_tool_text(result) or "Final answer submitted."
+            return render_terminal_html(state), _state_summary(state), status
+
+        def on_example(choice: str):
+            if choice and choice in COMMAND_EXAMPLES:
+                return COMMAND_EXAMPLES[choice]
+            return gr.update()
+
+        reset_btn.click(
+            fn=on_reset,
+            inputs=[setup_input, verify_input],
+            outputs=[terminal_display, state_box, status_bar],
+        )
+        close_btn.click(
+            fn=on_close,
+            outputs=[terminal_display, state_box, status_bar],
+        )
+        run_btn.click(
+            fn=on_run,
+            inputs=[command_input],
+            outputs=[terminal_display, state_box, status_bar],
+        )
+        submit_btn.click(
+            fn=on_submit,
+            inputs=[answer_input],
+            outputs=[terminal_display, state_box, status_bar],
+        )
+        example_dd.change(
+            fn=on_example,
+            inputs=[example_dd],
+            outputs=[command_input],
+        )
+
+    return demo

server/hf_sandbox.py

@@ -0,0 +1,93 @@
+# 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.
+
+"""Small hf-sandbox wrapper for terminal-style environments."""
+
+from __future__ import annotations
+
+import os
+from dataclasses import dataclass
+
+_HF_SANDBOX_IMPORT_ERROR: ImportError | None = None
+
+try:
+    from hf_sandbox import Sandbox
+except ImportError as _hf_sandbox_import_error:  # pragma: no cover
+    _HF_SANDBOX_IMPORT_ERROR = _hf_sandbox_import_error
+    Sandbox = None  # type: ignore[assignment]
+
+
+DEFAULT_IMAGE = "python:3.12"
+DEFAULT_FLAVOR = "cpu-basic"
+DEFAULT_TIMEOUT = "1h"
+
+
+@dataclass
+class ShellResult:
+    """Normalized result from a command executed in an HF sandbox."""
+
+    stdout: str
+    stderr: str
+    error: str | None
+    success: bool
+
+
+class HFSandbox:
+    """Manages one hf-sandbox job for one OpenEnv episode."""
+
+    def __init__(
+        self,
+        *,
+        image: str | None = None,
+        flavor: str | None = None,
+        timeout: str | None = None,
+        forward_hf_token: bool | None = None,
+    ):
+        if Sandbox is None:
+            raise ImportError(
+                "hf-sandbox is not installed. Install the terminus_env package "
+                "dependencies to use HFSandbox. Original import error: "
+                f"{_HF_SANDBOX_IMPORT_ERROR}"
+            )
+
+        resolved_forward = _coerce_bool(
+            os.getenv("HF_SANDBOX_FORWARD_HF_TOKEN", "false")
+        )
+        if forward_hf_token is not None:
+            resolved_forward = bool(forward_hf_token)
+
+        self._sandbox = Sandbox.create(
+            image=image or os.getenv("HF_SANDBOX_IMAGE", DEFAULT_IMAGE),
+            flavor=flavor or os.getenv("HF_SANDBOX_FLAVOR", DEFAULT_FLAVOR),
+            timeout=timeout or os.getenv("HF_SANDBOX_TIMEOUT", DEFAULT_TIMEOUT),
+            forward_hf_token=resolved_forward,
+        )
+        self.sandbox_id: str = self._sandbox.job_id
+
+    def run_shell(self, command: str, timeout_s: int = 120) -> ShellResult:
+        process = self._sandbox.exec(
+            "bash",
+            "-lc",
+            command,
+            timeout=timeout_s,
+        )
+        success = process.returncode == 0
+        return ShellResult(
+            stdout=process.stdout or "",
+            stderr=process.stderr or "",
+            error=None if success else f"exit code {process.returncode}",
+            success=success,
+        )
+
+    def kill(self) -> None:
+        try:
+            self._sandbox.terminate()
+        except Exception:
+            pass
+
+
+def _coerce_bool(value: str) -> bool:
+    return value.strip().lower() in {"1", "true", "yes", "on"}

server/requirements.txt

@@ -0,0 +1,8 @@
+openenv-core[core]>=0.2.2
+hf-sandbox>=0.1.1
+fastapi>=0.115.0
+fastmcp>=3.0.0
+gradio>=4.0.0
+pydantic>=2.0.0
+requests>=2.31.0
+uvicorn>=0.24.0

server/terminus_env_environment.py

@@ -0,0 +1,263 @@
+# 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.
+
+"""HF Sandbox-backed single-tool coding environment inspired by Terminus."""
+
+from __future__ import annotations
+
+from typing import Any, Iterable, Optional
+from uuid import uuid4
+
+from fastmcp import FastMCP
+from openenv.core.env_server.mcp_environment import MCPEnvironment
+from openenv.core.env_server.types import Action, Observation
+
+try:
+    from .hf_sandbox import HFSandbox
+    from ..models import CommandResult, TerminusState
+except ImportError:  # pragma: no cover
+    from models import CommandResult, TerminusState
+    from server.hf_sandbox import HFSandbox
+
+
+REWARD_FILE = "/home/user/logs/verifier/reward.txt"
+
+
+class TerminusEnvironment(MCPEnvironment):
+    """Single-tool terminal environment with one HF Sandbox job per episode."""
+
+    SUPPORTS_CONCURRENT_SESSIONS = True
+
+    def __init__(self):
+        self._sandbox: Optional[HFSandbox] = None
+        self._state = TerminusState(episode_id=str(uuid4()), step_count=0)
+
+        mcp = FastMCP("terminus_env")
+
+        @mcp.tool
+        def terminal(command: str = "", final_answer: str = "") -> str:
+            """Run a shell command or submit a final answer inside the sandbox.
+
+            Args:
+                command: Shell command to execute in the episode's HF Sandbox.
+                final_answer: Optional answer string. When provided, stored
+                    as the final answer and any reset-time verify commands run.
+
+            Returns:
+                Command output, or final-answer verification summary.
+            """
+            if not self._sandbox:
+                return "Error: environment not reset. Call reset() first."
+            if final_answer:
+                self._state.submitted_answer = final_answer
+                if not self._state.verify_commands:
+                    return f"Answer submitted: {final_answer}"
+                summary = self._run_verify_commands()
+                return (
+                    f"Answer submitted: {final_answer}\n"
+                    f"Verification: {summary['passed']}/{summary['total']} passed; "
+                    f"reward={summary['reward']}"
+                )
+            if not command.strip():
+                return "Error: command or final_answer is required."
+            result = self._run_shell_command(command)
+            self._state.commands.append(result)
+            return result.output
+
+        super().__init__(mcp)
+
+    def reset(
+        self,
+        seed: Optional[int] = None,
+        episode_id: Optional[str] = None,
+        **kwargs: Any,
+    ) -> Observation:
+        """Create a fresh HF Sandbox job and run optional setup commands."""
+        if self._sandbox:
+            self._sandbox.kill()
+            self._sandbox = None
+
+        self._state = TerminusState(
+            episode_id=episode_id or str(uuid4()),
+            step_count=0,
+        )
+        try:
+            self._sandbox = HFSandbox(
+                image=kwargs.get("sandbox_image") or kwargs.get("hf_sandbox_image"),
+                flavor=kwargs.get("sandbox_flavor") or kwargs.get("hf_sandbox_flavor"),
+                timeout=kwargs.get("sandbox_timeout") or kwargs.get("hf_sandbox_timeout"),
+                forward_hf_token=kwargs.get("forward_hf_token"),
+            )
+        except Exception as exc:  # noqa: BLE001
+            return Observation(
+                done=True,
+                reward=None,
+                metadata={
+                    "status": "error",
+                    "error": f"failed to create HF sandbox: {type(exc).__name__}: {exc}",
+                },
+            )
+
+        self._state.sandbox_id = self._sandbox.sandbox_id
+        setup_commands = _coerce_commands(
+            kwargs.get("setup", kwargs.get("setup_scripts", []))
+        )
+        verify_commands = _coerce_commands(
+            kwargs.get("verify", kwargs.get("verify_scripts", []))
+        )
+        self._state.verify_commands = verify_commands
+
+        self._sandbox.run_shell("mkdir -p /home/user/logs/verifier")
+        if setup_commands:
+            setup_results = self._run_shell_commands(setup_commands)
+            self._state.setup_results = setup_results
+            failed = [result for result in setup_results if not result.success]
+            if failed:
+                return Observation(
+                    done=True,
+                    reward=None,
+                    metadata={
+                        "status": "error",
+                        "sandbox_id": self._state.sandbox_id,
+                        "message": "Setup command failed.",
+                        "setup_results": [
+                            result.model_dump() for result in setup_results
+                        ],
+                    },
+                )
+
+        msg = "Terminus environment ready. Use terminal(command=...) to work."
+        if setup_commands:
+            msg += f" Setup commands run: {len(setup_commands)}."
+        if verify_commands:
+            msg += f" Verify commands registered: {len(verify_commands)}."
+        return Observation(
+            done=False,
+            reward=None,
+            metadata={
+                "status": "ready",
+                "sandbox_id": self._state.sandbox_id,
+                "message": msg,
+                "setup_results": [
+                    result.model_dump() for result in self._state.setup_results
+                ],
+                "verify_commands": verify_commands,
+            },
+        )
+
+    def _step_impl(
+        self,
+        action: Action,
+        timeout_s: Optional[float] = None,
+        **_: Any,
+    ) -> Observation:
+        return Observation(
+            done=False,
+            reward=None,
+            metadata={
+                "error": (
+                    f"Unknown action type: {type(action).__name__}. "
+                    "Use ListToolsAction or CallToolAction for MCP interactions."
+                )
+            },
+        )
+
+    def step(
+        self,
+        action: Action,
+        timeout_s: Optional[float] = None,
+        **kwargs: Any,
+    ) -> Observation:
+        self._state.step_count += 1
+        obs = super().step(action, timeout_s=timeout_s, **kwargs)
+        if self._state.submitted_answer is not None and self._state.last_reward is not None:
+            obs.done = True
+            obs.reward = self._state.last_reward
+        elif obs.reward is None:
+            obs.reward = 0.0
+        return obs
+
+    async def step_async(
+        self,
+        action: Action,
+        timeout_s: Optional[float] = None,
+        **kwargs: Any,
+    ) -> Observation:
+        self._state.step_count += 1
+        obs = await super().step_async(action, timeout_s=timeout_s, **kwargs)
+        if self._state.submitted_answer is not None and self._state.last_reward is not None:
+            obs.done = True
+            obs.reward = self._state.last_reward
+        elif obs.reward is None:
+            obs.reward = 0.0
+        return obs
+
+    @property
+    def state(self) -> TerminusState:
+        return self._state
+
+    def close(self) -> None:
+        if self._sandbox:
+            self._sandbox.kill()
+            self._sandbox = None
+
+    def _run_shell_commands(self, commands: Iterable[str]) -> list[CommandResult]:
+        return [self._run_shell_command(command) for command in commands]
+
+    def _run_shell_command(self, command: str) -> CommandResult:
+        result = self._sandbox.run_shell(command)
+        output = _format_for_llm(result)
+        return CommandResult(
+            command=command,
+            output=output,
+            error=result.error,
+            success=result.success,
+        )
+
+    def _run_verify_commands(self) -> dict[str, Any]:
+        if not self._sandbox:
+            return {"passed": 0, "total": 0, "reward": None}
+
+        self._sandbox.run_shell("mkdir -p /home/user/logs/verifier")
+        verify_results = self._run_shell_commands(self._state.verify_commands)
+        self._state.verify_results = verify_results
+        passed = sum(1 for result in verify_results if result.success)
+        total = len(verify_results)
+        reward = _read_reward_override(self._sandbox)
+        if reward is None and total:
+            reward = passed / total
+        self._state.last_reward = reward
+        return {"passed": passed, "total": total, "reward": reward}
+
+
+def _coerce_commands(value: Any) -> list[str]:
+    if value is None:
+        return []
+    if isinstance(value, str):
+        return [value] if value.strip() else []
+    return [str(item) for item in value if str(item).strip()]
+
+
+def _format_for_llm(result) -> str:
+    parts = []
+    if result.stdout:
+        parts.append(result.stdout.strip())
+    if result.stderr:
+        parts.append(result.stderr.strip())
+    if result.error:
+        parts.append(f"ERROR:\n{result.error}")
+    return "\n".join(parts) if parts else "(no output)"
+
+
+def _read_reward_override(sandbox: HFSandbox) -> Optional[float]:
+    result = sandbox.run_shell(f"cat {REWARD_FILE} 2>/dev/null || true")
+    raw = (result.stdout or "").strip()
+    if not raw:
+        return None
+    try:
+        return float(raw)
+    except ValueError:
+        return None