Upload folder using huggingface_hub

Dockerfile

@@ -0,0 +1,13 @@
+FROM python:3.10-slim
+
+WORKDIR /app
+
+COPY requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+
+COPY . .
+
+EXPOSE 8000
+
+ENV ENABLE_WEB_INTERFACE=true
+CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "8000"]

README.md

@@ -1,10 +1,255 @@
 ---
-title: Workflowops
-emoji: 🦀
-colorFrom: yellow
-colorTo: green
+title: Workflowops Environment Server
+emoji: 🗜️
+colorFrom: green
+colorTo: blue
 sdk: docker
 pinned: false
+app_port: 8000
+base_path: /web
+tags:
+  - openenv
 ---
 
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+# Workflowops Environment
+
+A simple test environment that echoes back messages. Perfect for testing the env APIs as well as demonstrating environment usage patterns.
+
+## Quick Start
+
+The simplest way to use the Workflowops environment is through the `WorkflowopsEnv` class:
+
+```python
+from workflowops import WorkflowopsAction, WorkflowopsEnv
+
+try:
+    # Create environment from Docker image
+    workflowopsenv = WorkflowopsEnv.from_docker_image("workflowops-env:latest")
+
+    # Reset
+    result = workflowopsenv.reset()
+    print(f"Reset: {result.observation.echoed_message}")
+
+    # Send multiple messages
+    messages = ["Hello, World!", "Testing echo", "Final message"]
+
+    for msg in messages:
+        result = workflowopsenv.step(WorkflowopsAction(message=msg))
+        print(f"Sent: '{msg}'")
+        print(f"  → Echoed: '{result.observation.echoed_message}'")
+        print(f"  → Length: {result.observation.message_length}")
+        print(f"  → Reward: {result.reward}")
+
+finally:
+    # Always clean up
+    workflowopsenv.close()
+```
+
+That's it! The `WorkflowopsEnv.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 project root
+docker build -t workflowops-env:latest -f server/Dockerfile .
+```
+
+## Deploying to Hugging Face Spaces
+
+You can easily deploy your OpenEnv environment to Hugging Face Spaces using the `openenv push` command:
+
+```bash
+# From the environment directory (where openenv.yaml is located)
+openenv push
+
+# Or specify options
+openenv push --namespace my-org --private
+```
+
+The `openenv push` command will:
+1. Validate that the directory is an OpenEnv environment (checks for `openenv.yaml`)
+2. Prepare a custom build for Hugging Face Docker space (enables web interface)
+3. Upload to Hugging Face (ensuring you're logged in)
+
+### Prerequisites
+
+- Authenticate with Hugging Face: The command will prompt for login if not already authenticated
+
+### Options
+
+- `--directory`, `-d`: Directory containing the OpenEnv environment (defaults to current directory)
+- `--repo-id`, `-r`: Repository ID in format 'username/repo-name' (defaults to 'username/env-name' from openenv.yaml)
+- `--base-image`, `-b`: Base Docker image to use (overrides Dockerfile FROM)
+- `--private`: Deploy the space as private (default: public)
+
+### Examples
+
+```bash
+# Push to your personal namespace (defaults to username/env-name from openenv.yaml)
+openenv push
+
+# Push to a specific repository
+openenv push --repo-id my-org/my-env
+
+# Push with a custom base image
+openenv push --base-image ghcr.io/meta-pytorch/openenv-base:latest
+
+# Push as a private space
+openenv push --private
+
+# Combine options
+openenv push --repo-id my-org/my-env --base-image custom-base:latest --private
+```
+
+After deployment, your space will be available at:
+`https://huggingface.co/spaces/<repo-id>`
+
+The deployed space includes:
+- **Web Interface** at `/web` - Interactive UI for exploring the environment
+- **API Documentation** at `/docs` - Full OpenAPI/Swagger interface
+- **Health Check** at `/health` - Container health monitoring
+- **WebSocket** at `/ws` - Persistent session endpoint for low-latency interactions
+
+## Environment Details
+
+### Action
+**WorkflowopsAction**: Contains a single field
+- `message` (str) - The message to echo back
+
+### Observation
+**WorkflowopsObservation**: Contains the echo response and metadata
+- `echoed_message` (str) - The message echoed back
+- `message_length` (int) - Length of the message
+- `reward` (float) - Reward based on message length (length × 0.1)
+- `done` (bool) - Always False for echo environment
+- `metadata` (dict) - Additional info like step count
+
+### Reward
+The reward is calculated as: `message_length × 0.1`
+- "Hi" → reward: 0.2
+- "Hello, World!" → reward: 1.3
+- Empty message → reward: 0.0
+
+## Advanced Usage
+
+### Connecting to an Existing Server
+
+If you already have a Workflowops environment server running, you can connect directly:
+
+```python
+from workflowops import WorkflowopsEnv
+
+# Connect to existing server
+workflowopsenv = WorkflowopsEnv(base_url="<ENV_HTTP_URL_HERE>")
+
+# Use as normal
+result = workflowopsenv.reset()
+result = workflowopsenv.step(WorkflowopsAction(message="Hello!"))
+```
+
+Note: When connecting to an existing server, `workflowopsenv.close()` will NOT stop the server.
+
+### Using the Context Manager
+
+The client supports context manager usage for automatic connection management:
+
+```python
+from workflowops import WorkflowopsAction, WorkflowopsEnv
+
+# Connect with context manager (auto-connects and closes)
+with WorkflowopsEnv(base_url="http://localhost:8000") as env:
+    result = env.reset()
+    print(f"Reset: {result.observation.echoed_message}")
+    # Multiple steps with low latency
+    for msg in ["Hello", "World", "!"]:
+        result = env.step(WorkflowopsAction(message=msg))
+        print(f"Echoed: {result.observation.echoed_message}")
+```
+
+The client uses WebSocket connections for:
+- **Lower latency**: No HTTP connection overhead per request
+- **Persistent session**: Server maintains your environment state
+- **Efficient for episodes**: Better for many sequential steps
+
+### Concurrent WebSocket Sessions
+
+The server supports multiple concurrent WebSocket connections. To enable this,
+modify `server/app.py` to use factory mode:
+
+```python
+# In server/app.py - use factory mode for concurrent sessions
+app = create_app(
+    WorkflowopsEnvironment,  # Pass class, not instance
+    WorkflowopsAction,
+    WorkflowopsObservation,
+    max_concurrent_envs=4,  # Allow 4 concurrent sessions
+)
+```
+
+Then multiple clients can connect simultaneously:
+
+```python
+from workflowops import WorkflowopsAction, WorkflowopsEnv
+from concurrent.futures import ThreadPoolExecutor
+
+def run_episode(client_id: int):
+    with WorkflowopsEnv(base_url="http://localhost:8000") as env:
+        result = env.reset()
+        for i in range(10):
+            result = env.step(WorkflowopsAction(message=f"Client {client_id}, step {i}"))
+        return client_id, result.observation.message_length
+
+# Run 4 episodes concurrently
+with ThreadPoolExecutor(max_workers=4) as executor:
+    results = list(executor.map(run_episode, range(4)))
+```
+
+## Development & Testing
+
+### Direct Environment Testing
+
+Test the environment logic directly without starting the HTTP server:
+
+```bash
+# From the server directory
+python3 server/workflowops_environment.py
+```
+
+This verifies that:
+- Environment resets correctly
+- Step executes actions properly
+- State tracking works
+- Rewards are calculated correctly
+
+### Running Locally
+
+Run the server locally for development:
+
+```bash
+uvicorn server.app:app --reload
+```
+
+## Project Structure
+
+```
+workflowops/
+├── .dockerignore         # Docker build exclusions
+├── __init__.py            # Module exports
+├── README.md              # This file
+├── openenv.yaml           # OpenEnv manifest
+├── pyproject.toml         # Project metadata and dependencies
+├── uv.lock                # Locked dependencies (generated)
+├── client.py              # WorkflowopsEnv client
+├── models.py              # Action and Observation models
+└── server/
+    ├── __init__.py        # Server module exports
+    ├── workflowops_environment.py  # Core environment logic
+    ├── app.py             # FastAPI application (HTTP + WebSocket endpoints)
+    └── Dockerfile         # Container image definition
+```

__init__.py

@@ -0,0 +1,16 @@
+# 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.
+
+"""Workflowops Environment."""
+
+from .client import WorkflowopsEnv
+from .models import WorkflowopsAction, WorkflowopsObservation
+
+__all__ = [
+    "WorkflowopsAction",
+    "WorkflowopsObservation",
+    "WorkflowopsEnv",
+]

app.py

@@ -0,0 +1,67 @@
+from fastapi import FastAPI, HTTPException
+from pydantic import BaseModel
+from typing import Dict, Any
+import uvicorn
+
+from tasks.email_triage import EmailTriageTask
+from tasks.support_ticket import SupportTicketTask
+from tasks.schedule_conflict import ScheduleConflictTask
+
+app = FastAPI(title="WorkflowOps OpenEnv", version="1.0.0")
+
+TASKS = {
+    "email_triage": EmailTriageTask,
+    "support_ticket": SupportTicketTask,
+    "schedule_conflict": ScheduleConflictTask
+}
+
+active_envs: Dict[str, Any] = {}
+
+
+class ActionRequest(BaseModel):
+    episode_id: str
+    action: Dict[str, Any]
+
+
+@app.get("/health")
+async def health():
+    return {"status": "healthy", "service": "workflowops"}
+
+
+@app.get("/ready")
+async def ready():
+    return {"ready": True, "tasks": list(TASKS.keys())}
+
+
+@app.post("/env/{task_name}/reset")
+async def reset_environment(task_name: str):
+    if task_name not in TASKS:
+        raise HTTPException(status_code=404, detail="Task not found")
+    
+    env = TASKS[task_name]()
+    observation = env.reset()
+    
+    active_envs[env.state.episode_id] = env
+    
+    return {
+        "episode_id": env.state.episode_id,
+        "observation": observation.dict()
+    }
+
+
+@app.post("/env/step")
+async def step_environment(request: ActionRequest):
+    if request.episode_id not in active_envs:
+        raise HTTPException(status_code=404, detail="Episode not found")
+    
+    env = active_envs[request.episode_id]
+    observation = env.step(request.action)
+    
+    if observation.done:
+        del active_envs[request.episode_id]
+    
+    return observation.dict()
+
+
+if __name__ == "__main__":
+    uvicorn.run(app, host="0.0.0.0", port=8000)

client.py

@@ -0,0 +1,99 @@
+# 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.
+
+"""Workflowops Environment Client."""
+
+from typing import Dict
+
+from openenv.core import EnvClient
+from openenv.core.client_types import StepResult
+from openenv.core.env_server.types import State
+
+from .models import WorkflowopsAction, WorkflowopsObservation
+
+
+class WorkflowopsEnv(
+    EnvClient[WorkflowopsAction, WorkflowopsObservation, State]
+):
+    """
+    Client for the Workflowops Environment.
+
+    This client maintains a persistent WebSocket connection to the environment server,
+    enabling efficient multi-step interactions with lower latency.
+    Each client instance has its own dedicated environment session on the server.
+
+    Example:
+        >>> # Connect to a running server
+        >>> with WorkflowopsEnv(base_url="http://localhost:8000") as client:
+        ...     result = client.reset()
+        ...     print(result.observation.echoed_message)
+        ...
+        ...     result = client.step(WorkflowopsAction(message="Hello!"))
+        ...     print(result.observation.echoed_message)
+
+    Example with Docker:
+        >>> # Automatically start container and connect
+        >>> client = WorkflowopsEnv.from_docker_image("workflowops-env:latest")
+        >>> try:
+        ...     result = client.reset()
+        ...     result = client.step(WorkflowopsAction(message="Test"))
+        ... finally:
+        ...     client.close()
+    """
+
+    def _step_payload(self, action: WorkflowopsAction) -> Dict:
+        """
+        Convert WorkflowopsAction to JSON payload for step message.
+
+        Args:
+            action: WorkflowopsAction instance
+
+        Returns:
+            Dictionary representation suitable for JSON encoding
+        """
+        return {
+            "message": action.message,
+        }
+
+    def _parse_result(self, payload: Dict) -> StepResult[WorkflowopsObservation]:
+        """
+        Parse server response into StepResult[WorkflowopsObservation].
+
+        Args:
+            payload: JSON response data from server
+
+        Returns:
+            StepResult with WorkflowopsObservation
+        """
+        obs_data = payload.get("observation", {})
+        observation = WorkflowopsObservation(
+            echoed_message=obs_data.get("echoed_message", ""),
+            message_length=obs_data.get("message_length", 0),
+            done=payload.get("done", False),
+            reward=payload.get("reward"),
+            metadata=obs_data.get("metadata", {}),
+        )
+
+        return StepResult(
+            observation=observation,
+            reward=payload.get("reward"),
+            done=payload.get("done", False),
+        )
+
+    def _parse_state(self, payload: Dict) -> State:
+        """
+        Parse server response into State object.
+
+        Args:
+            payload: JSON response from state request
+
+        Returns:
+            State object with episode_id and step_count
+        """
+        return State(
+            episode_id=payload.get("episode_id"),
+            step_count=payload.get("step_count", 0),
+        )

environment/base.py

@@ -0,0 +1,71 @@
+from typing import Dict, Any, Optional, List
+from uuid import uuid4
+from pydantic import BaseModel, Field
+from openenv.core.env_server.interfaces import Environment
+from openenv.core.env_server.types import State
+
+
+class Observation(BaseModel):
+    done: bool = False
+    reward: float = 0.0
+    observation: Dict[str, Any] = Field(default_factory=dict)
+    metadata: Dict[str, Any] = Field(default_factory=dict)
+
+
+class BaseWorkflowEnvironment(Environment):
+    SUPPORTS_CONCURRENT_SESSIONS: bool = True
+    
+    def __init__(self, seed: Optional[int] = None):
+        self._state = State(episode_id=str(uuid4()), step_count=0)
+        self.seed = seed
+        self.history: List[Dict[str, Any]] = []
+        self.max_steps: int = 20
+        self.task_state: Dict[str, Any] = {}
+        
+    def reset(self) -> Observation:
+        self._state = State(episode_id=str(uuid4()), step_count=0)
+        self.history = []
+        self.task_state = {}
+        return Observation(
+            done=False,
+            reward=0.0,
+            observation={"status": "ready", "episode_id": self._state.episode_id},
+            metadata={"reset_count": len(self.history)}
+        )
+    
+    def step(self, action: Dict[str, Any]) -> Observation:
+        self._state.step_count += 1
+        
+        # Validate action
+        if not isinstance(action, dict):
+            return Observation(
+                done=True,
+                reward=-0.5,
+                observation={"error": "Invalid action format"},
+                metadata={"step": self._state.step_count}
+            )
+        
+        # Record history
+        self.history.append({
+            "step": self._state.step_count,
+            "action": action,
+            "timestamp": self._state.episode_id
+        })
+        
+        # Check max steps
+        if self._state.step_count >= self.max_steps:
+            return Observation(
+                done=True,
+                reward=0.0,
+                observation={"status": "max_steps_reached"},
+                metadata={"step": self._state.step_count}
+            )
+        
+        return self._execute_action(action)
+    
+    def _execute_action(self, action: Dict[str, Any]) -> Observation:
+        raise NotImplementedError("Subclasses must implement _execute_action")
+    
+    @property
+    def state(self) -> State:
+        return self._state

graders/base.py

@@ -0,0 +1,26 @@
+from typing import Dict, Any, Tuple
+
+
+class BaseGrader:
+    """Base grader class with standard scoring logic"""
+    
+    def __init__(self, max_score: float = 1.0):
+        self.max_score = max_score
+        self.efficiency_penalty_per_step = 0.02
+        self.invalid_action_penalty = 0.15
+    
+    def grade(self, state: Dict[str, Any], action: Dict[str, Any], step_count: int) -> Tuple[float, Dict[str, Any]]:
+        """
+        Grade an action against current state
+        Returns: (score [0,1], metadata)
+        """
+        raise NotImplementedError("Subclasses must implement grade method")
+    
+    def apply_efficiency_penalty(self, base_score: float, step_count: int) -> float:
+        """Apply penalty for taking too many steps"""
+        penalty = step_count * self.efficiency_penalty_per_step
+        return max(0.0, base_score - penalty)
+    
+    def apply_invalid_action_penalty(self, score: float) -> float:
+        """Apply penalty for invalid actions"""
+        return max(0.0, score - self.invalid_action_penalty)

inference.py

@@ -0,0 +1,74 @@
+#!/usr/bin/env python3
+import os
+import sys
+import json
+import time
+from typing import Dict, Any
+from openai import OpenAI
+import requests
+
+
+def main():
+    print("[START]")
+    
+    api_base = os.getenv("API_BASE_URL", "https://api.openai.com/v1")
+    model_name = os.getenv("MODEL_NAME", "gpt-3.5-turbo")
+    env_url = os.getenv("ENV_URL", "http://localhost:8000")
+    task = os.getenv("TASK_NAME", "email_triage")
+    
+    client = OpenAI(base_url=api_base, api_key=os.getenv("OPENAI_API_KEY"))
+    
+    # Reset environment
+    resp = requests.post(f"{env_url}/env/{task}/reset")
+    resp.raise_for_status()
+    data = resp.json()
+    
+    episode_id = data["episode_id"]
+    observation = data["observation"]
+    
+    step = 0
+    
+    while not observation.get("done", False):
+        step += 1
+        print(f"[STEP] {step}")
+        
+        try:
+            # Call LLM
+            response = client.chat.completions.create(
+                model=model_name,
+                messages=[
+                    {"role": "system", "content": "You are an operations assistant. Respond ONLY with valid JSON action."},
+                    {"role": "user", "content": json.dumps(observation["observation"])}
+                ],
+                temperature=0.0,
+                max_tokens=256
+            )
+            
+            action_text = response.choices[0].message.content.strip()
+            action = json.loads(action_text)
+            
+            # Execute step
+            resp = requests.post(f"{env_url}/env/step", json={
+                "episode_id": episode_id,
+                "action": action
+            })
+            resp.raise_for_status()
+            observation = resp.json()
+            
+        except Exception as e:
+            print(f"[ERROR] {str(e)}", file=sys.stderr)
+            # Fallback action
+            resp = requests.post(f"{env_url}/env/step", json={
+                "episode_id": episode_id,
+                "action": {}
+            })
+            observation = resp.json()
+        
+        time.sleep(0.1)
+    
+    print(f"[FINAL SCORE] {observation.get('reward', 0.0)}")
+    print("[END]")
+
+
+if __name__ == "__main__":
+    main()

models.py

@@ -0,0 +1,27 @@
+# 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 Workflowops Environment.
+
+The workflowops environment is a simple test environment that echoes back messages.
+"""
+
+from openenv.core.env_server.types import Action, Observation
+from pydantic import Field
+
+
+class WorkflowopsAction(Action):
+    """Action for the Workflowops environment - just a message to echo."""
+
+    message: str = Field(..., description="Message to echo back")
+
+
+class WorkflowopsObservation(Observation):
+    """Observation from the Workflowops environment - the echoed message."""
+
+    echoed_message: str = Field(default="", description="The echoed message")
+    message_length: int = Field(default=0, description="Length of the echoed message")

openenv.yaml

@@ -0,0 +1,37 @@
+name: workflowops
+version: 1.0.0
+description: Workflow Operations OpenEnv Environment
+author: WorkflowOps Team
+
+environment:
+  python_version: "3.10"
+  concurrent_sessions: true
+  max_episode_steps: 20
+  timeout: 300
+
+tasks:
+  - name: email_triage
+    endpoint: /env/email_triage/reset
+    max_steps: 8
+    description: "Categorize incoming emails into correct priority buckets"
+  
+  - name: support_ticket
+    endpoint: /env/support_ticket/reset
+    max_steps: 6
+    description: "Prioritize and assign incoming support tickets"
+  
+  - name: schedule_conflict
+    endpoint: /env/schedule_conflict/reset
+    max_steps: 1
+    description: "Resolve calendar scheduling conflicts"
+
+deployment:
+  platform: huggingface
+  port: 8000
+  health_check: /health
+  ready_check: /ready
+
+validation:
+  deterministic: true
+  reward_range: [0.0, 1.0]
+  requires_api_key: false

pyproject.toml

@@ -0,0 +1,45 @@
+# 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-workflowops"
+version = "0.1.0"
+description = "Workflowops environment for OpenEnv"
+requires-python = ">=3.10"
+dependencies = [
+    # Core OpenEnv runtime (provides FastAPI server + HTTP client types)
+    # install from github
+    # "openenv-core[core] @ git+https://github.com/meta-pytorch/OpenEnv.git",
+    "openenv-core[core]>=0.2.2",
+    # Environment-specific dependencies
+    # Add all dependencies needed for your environment here
+    # Examples:
+    # "numpy>=1.19.0",
+    # "torch>=2.0.0",
+    # "gymnasium>=0.29.0",
+    # "openspiel>=1.0.0",
+    # "smolagents>=1.22.0,<2",
+]
+
+[project.optional-dependencies]
+dev = [
+    "pytest>=8.0.0",
+    "pytest-cov>=4.0.0",
+]
+
+[project.scripts]
+# Server entry point - enables running via: uv run --project . server
+# or: python -m workflowops.server.app
+server = "workflowops.server.app:main"
+
+[tool.setuptools]
+include-package-data = true
+packages = ["workflowops", "workflowops.server"]
+package-dir = { "workflowops" = ".", "workflowops.server" = "server" }

requirements.txt

@@ -0,0 +1,7 @@
+fastapi>=0.104.0
+uvicorn>=0.24.0
+pydantic>=2.5.0
+requests>=2.31.0
+openai>=1.3.0
+openenv>=0.1.0
+python-multipart>=0.0.6

server/Dockerfile

@@ -0,0 +1,80 @@
+# 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 is flexible and works for both:
+# - In-repo environments (with local OpenEnv sources)
+# - Standalone environments (with openenv from PyPI/Git)
+# The build script (openenv build) handles context detection and sets appropriate build args.
+
+ARG BASE_IMAGE=ghcr.io/meta-pytorch/openenv-base:latest
+FROM ${BASE_IMAGE} AS builder
+
+WORKDIR /app
+
+# Ensure git is available (required for installing dependencies from VCS)
+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=workflowops
+
+# 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"
+
+# Health check
+HEALTHCHECK --interval=30s --timeout=3s --start-period=5s --retries=3 \
+    CMD curl -f http://localhost:8000/health || 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"]

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.
+
+"""Workflowops environment server components."""
+
+from .workflowops_environment import WorkflowopsEnvironment
+
+__all__ = ["WorkflowopsEnvironment"]

server/app.py

@@ -0,0 +1,84 @@
+# 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 Workflowops Environment.
+
+This module creates an HTTP server that exposes the WorkflowopsEnvironment
+over HTTP and WebSocket endpoints, compatible with EnvClient.
+
+Endpoints:
+    - POST /reset: Reset the environment
+    - POST /step: Execute an action
+    - GET /state: Get current environment state
+    - GET /schema: Get action/observation schemas
+    - WS /ws: WebSocket endpoint for persistent sessions
+
+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 --workers 4
+
+    # Or run directly:
+    python -m server.app
+"""
+
+try:
+    from openenv.core.env_server.http_server import create_app
+except Exception as e:  # pragma: no cover
+    raise ImportError(
+        "openenv is required for the web interface. Install dependencies with '\n    uv sync\n'"
+    ) from e
+
+try:
+    from ..models import WorkflowopsAction, WorkflowopsObservation
+    from .workflowops_environment import WorkflowopsEnvironment
+except ModuleNotFoundError:
+    from models import WorkflowopsAction, WorkflowopsObservation
+    from server.workflowops_environment import WorkflowopsEnvironment
+
+
+# Create the app with web interface and README integration
+app = create_app(
+    WorkflowopsEnvironment,
+    WorkflowopsAction,
+    WorkflowopsObservation,
+    env_name="workflowops",
+    max_concurrent_envs=1,  # increase this number to allow more concurrent WebSocket sessions
+)
+
+
+def main(host: str = "0.0.0.0", port: int = 8000):
+    """
+    Entry point for direct execution via uv run or python -m.
+
+    This function enables running the server without Docker:
+        uv run --project . server
+        uv run --project . server --port 8001
+        python -m workflowops.server.app
+
+    Args:
+        host: Host address to bind to (default: "0.0.0.0")
+        port: Port number to listen on (default: 8000)
+
+    For production deployments, consider using uvicorn directly with
+    multiple workers:
+        uvicorn workflowops.server.app:app --workers 4
+    """
+    import uvicorn
+
+    uvicorn.run(app, host=host, port=port)
+
+
+if __name__ == "__main__":
+    import argparse
+
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--port", type=int, default=8000)
+    args = parser.parse_args()
+    main(port=args.port)

server/requirements.txt

@@ -0,0 +1,6 @@
+openenv[core]>=0.2.0
+fastapi>=0.115.0
+uvicorn>=0.24.0
+
+
+

server/workflowops_environment.py

@@ -0,0 +1,104 @@
+# 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.
+
+"""
+Workflowops Environment Implementation.
+
+A simple test environment that echoes back messages sent to it.
+Perfect for testing HTTP server infrastructure.
+"""
+
+from uuid import uuid4
+
+from openenv.core.env_server.interfaces import Environment
+from openenv.core.env_server.types import State
+
+try:
+    from ..models import WorkflowopsAction, WorkflowopsObservation
+except ImportError:
+    from models import WorkflowopsAction, WorkflowopsObservation
+
+
+class WorkflowopsEnvironment(Environment):
+    """
+    A simple echo environment that echoes back messages.
+
+    This environment is designed for testing the HTTP server infrastructure.
+    It maintains minimal state and simply echoes back whatever message it receives.
+
+    Example:
+        >>> env = WorkflowopsEnvironment()
+        >>> obs = env.reset()
+        >>> print(obs.echoed_message)  # "Workflowops environment ready!"
+        >>>
+        >>> obs = env.step(WorkflowopsAction(message="Hello"))
+        >>> print(obs.echoed_message)  # "Hello"
+        >>> print(obs.message_length)  # 5
+    """
+
+    # Enable concurrent WebSocket sessions.
+    # Set to True if your environment isolates state between instances.
+    # When True, multiple WebSocket clients can connect simultaneously, each
+    # getting their own environment instance (when using factory mode in app.py).
+    SUPPORTS_CONCURRENT_SESSIONS: bool = True
+
+    def __init__(self):
+        """Initialize the workflowops environment."""
+        self._state = State(episode_id=str(uuid4()), step_count=0)
+        self._reset_count = 0
+
+    def reset(self) -> WorkflowopsObservation:
+        """
+        Reset the environment.
+
+        Returns:
+            WorkflowopsObservation with a ready message
+        """
+        self._state = State(episode_id=str(uuid4()), step_count=0)
+        self._reset_count += 1
+
+        return WorkflowopsObservation(
+            echoed_message="Workflowops environment ready!",
+            message_length=0,
+            done=False,
+            reward=0.0,
+        )
+
+    def step(self, action: WorkflowopsAction) -> WorkflowopsObservation:  # type: ignore[override]
+        """
+        Execute a step in the environment by echoing the message.
+
+        Args:
+            action: WorkflowopsAction containing the message to echo
+
+        Returns:
+            WorkflowopsObservation with the echoed message and its length
+        """
+        self._state.step_count += 1
+
+        message = action.message
+        length = len(message)
+
+        # Simple reward: longer messages get higher rewards
+        reward = length * 0.1
+
+        return WorkflowopsObservation(
+            echoed_message=message,
+            message_length=length,
+            done=False,
+            reward=reward,
+            metadata={"original_message": message, "step": self._state.step_count},
+        )
+
+    @property
+    def state(self) -> State:
+        """
+        Get the current environment state.
+
+        Returns:
+            Current State with episode_id and step_count
+        """
+        return self._state

tasks/email_triage.py

@@ -0,0 +1,118 @@
+from typing import Dict, Any, Tuple
+from graders.base import BaseGrader
+from environment.base import BaseWorkflowEnvironment, Observation
+
+
+class EmailTriageGrader(BaseGrader):
+    CATEGORIES = ["urgent", "important", "normal", "spam"]
+    
+    def grade(self, state: Dict[str, Any], action: Dict[str, Any], step_count: int) -> Tuple[float, Dict[str, Any]]:
+        email = state.get("current_email", {})
+        correct_category = email.get("correct_category")
+        
+        selected_category = action.get("category")
+        
+        if not selected_category or selected_category not in self.CATEGORIES:
+            return self.apply_invalid_action_penalty(0.0), {"error": "Invalid category"}
+        
+        if selected_category == correct_category:
+            base_score = 1.0
+        else:
+            # Partial credit for near correct
+            if correct_category == "urgent" and selected_category == "important":
+                base_score = 0.6
+            elif correct_category == "important" and selected_category == "urgent":
+                base_score = 0.6
+            else:
+                base_score = 0.2
+        
+        final_score = self.apply_efficiency_penalty(base_score, step_count)
+        
+        return final_score, {
+            "base_score": base_score,
+            "correct": selected_category == correct_category,
+            "step_count": step_count
+        }
+
+
+class EmailTriageTask(BaseWorkflowEnvironment):
+    TEST_EMAILS = [
+        {
+            "id": 1,
+            "subject": "SERVER OUTAGE - All systems down",
+            "body": "Production servers are not responding. Immediate action required.",
+            "correct_category": "urgent",
+            "sender": "sysadmin@company.com"
+        },
+        {
+            "id": 2,
+            "subject": "Quarterly Performance Review",
+            "body": "Please schedule your performance review meeting.",
+            "correct_category": "important",
+            "sender": "hr@company.com"
+        },
+        {
+            "id": 3,
+            "subject": "Team lunch tomorrow",
+            "body": "We are having team lunch at 1pm. Let us know if you are coming.",
+            "correct_category": "normal",
+            "sender": "team@company.com"
+        },
+        {
+            "id": 4,
+            "subject": "YOU WON $1,000,000!!!",
+            "body": "Click here to claim your prize now!",
+            "correct_category": "spam",
+            "sender": "scam@fake.com"
+        }
+    ]
+    
+    def reset(self) -> Observation:
+        super().reset()
+        self.task_state["email_index"] = 0
+        self.task_state["current_email"] = self.TEST_EMAILS[0]
+        self.task_state["graded"] = 0
+        self.task_state["score"] = 0.0
+        
+        return Observation(
+            done=False,
+            reward=0.0,
+            observation={
+                "task": "email_triage",
+                "email": self.task_state["current_email"],
+                "categories": EmailTriageGrader.CATEGORIES,
+                "remaining": len(self.TEST_EMAILS)
+            }
+        )
+    
+    def _execute_action(self, action: Dict[str, Any]) -> Observation:
+        grader = EmailTriageGrader()
+        score, meta = grader.grade(self.task_state, action, self._state.step_count)
+        
+        self.task_state["score"] += score
+        self.task_state["graded"] += 1
+        
+        # Move to next email
+        self.task_state["email_index"] += 1
+        
+        if self.task_state["email_index"] >= len(self.TEST_EMAILS):
+            final_score = self.task_state["score"] / len(self.TEST_EMAILS)
+            return Observation(
+                done=True,
+                reward=final_score,
+                observation={"final_score": final_score, "total_graded": self.task_state["graded"]},
+                metadata=meta
+            )
+        
+        self.task_state["current_email"] = self.TEST_EMAILS[self.task_state["email_index"]]
+        
+        return Observation(
+            done=False,
+            reward=score,
+            observation={
+                "email": self.task_state["current_email"],
+                "current_score": self.task_state["score"],
+                "remaining": len(self.TEST_EMAILS) - self.task_state["email_index"]
+            },
+            metadata=meta
+        )

tasks/schedule_conflict.py

@@ -0,0 +1,81 @@
+from typing import Dict, Any, Tuple
+from graders.base import BaseGrader
+from environment.base import BaseWorkflowEnvironment, Observation
+
+
+class ScheduleConflictGrader(BaseGrader):
+    def grade(self, state: Dict[str, Any], action: Dict[str, Any], step_count: int) -> Tuple[float, Dict[str, Any]]:
+        events = state.get("events", [])
+        solution = action.get("schedule", [])
+        
+        if not isinstance(solution, list):
+            return self.apply_invalid_action_penalty(0.0), {"error": "Invalid schedule format"}
+        
+        # Check for overlaps
+        overlaps = 0
+        for i in range(len(solution)):
+            for j in range(i+1, len(solution)):
+                e1 = solution[i]
+                e2 = solution[j]
+                if not (e1["end"] <= e2["start"] or e2["end"] <= e1["start"]):
+                    overlaps += 1
+        
+        overlap_penalty = overlaps * 0.25
+        
+        # Check all events are scheduled
+        missing_events = len(events) - len(solution)
+        missing_penalty = missing_events * 0.15
+        
+        base_score = 1.0 - overlap_penalty - missing_penalty
+        final_score = self.apply_efficiency_penalty(max(0.0, base_score), step_count)
+        
+        return final_score, {
+            "overlaps": overlaps,
+            "missing_events": missing_events,
+            "base_score": base_score,
+            "final_score": final_score
+        }
+
+
+class ScheduleConflictTask(BaseWorkflowEnvironment):
+    TEST_SCHEDULES = [
+        {
+            "events": [
+                {"id": 1, "title": "Team Standup", "duration": 15, "priority": 2},
+                {"id": 2, "title": "Client Call", "duration": 60, "priority": 1},
+                {"id": 3, "title": "Code Review", "duration": 30, "priority": 3},
+                {"id": 4, "title": "Lunch Break", "duration": 45, "priority": 2}
+            ],
+            "available_slots": [
+                {"start": 540, "end": 720},  # 9:00 - 12:00
+                {"start": 780, "end": 1020}  # 13:00 - 17:00
+            ]
+        }
+    ]
+    
+    def reset(self) -> Observation:
+        super().reset()
+        self.task_state["schedule_index"] = 0
+        self.task_state["events"] = self.TEST_SCHEDULES[0]["events"]
+        self.task_state["available_slots"] = self.TEST_SCHEDULES[0]["available_slots"]
+        
+        return Observation(
+            done=False,
+            reward=0.0,
+            observation={
+                "task": "schedule_conflict",
+                "events": self.task_state["events"],
+                "available_slots": self.task_state["available_slots"]
+            }
+        )
+    
+    def _execute_action(self, action: Dict[str, Any]) -> Observation:
+        grader = ScheduleConflictGrader()
+        score, meta = grader.grade(self.task_state, action, self._state.step_count)
+        
+        return Observation(
+            done=True,
+            reward=score,
+            observation={"final_score": score},
+            metadata=meta
+        )

tasks/support_ticket.py

@@ -0,0 +1,115 @@
+from typing import Dict, Any, Tuple
+from graders.base import BaseGrader
+from environment.base import BaseWorkflowEnvironment, Observation
+
+
+class SupportTicketGrader(BaseGrader):
+    PRIORITIES = ["critical", "high", "medium", "low"]
+    
+    def grade(self, state: Dict[str, Any], action: Dict[str, Any], step_count: int) -> Tuple[float, Dict[str, Any]]:
+        ticket = state.get("current_ticket", {})
+        correct_priority = ticket.get("correct_priority")
+        
+        selected_priority = action.get("priority")
+        estimated_time = action.get("estimated_time", 0)
+        assignee = action.get("assignee")
+        
+        score = 0.0
+        
+        # Priority score (60%)
+        if selected_priority == correct_priority:
+            score += 0.6
+        else:
+            priority_distance = abs(self.PRIORITIES.index(selected_priority) - self.PRIORITIES.index(correct_priority))
+            score += max(0.0, 0.6 - (priority_distance * 0.2))
+        
+        # Time estimation score (30%)
+        correct_time = ticket.get("correct_time", 0)
+        time_error = abs(estimated_time - correct_time) / max(correct_time, 1)
+        score += max(0.0, 0.3 - (time_error * 0.3))
+        
+        # Assignee score (10%)
+        if assignee == ticket.get("correct_assignee"):
+            score += 0.1
+        
+        final_score = self.apply_efficiency_penalty(score, step_count)
+        
+        return final_score, {
+            "priority_score": score >= 0.6,
+            "time_score": score >= 0.3,
+            "assignee_score": score >= 0.1,
+            "total": final_score
+        }
+
+
+class SupportTicketTask(BaseWorkflowEnvironment):
+    TEST_TICKETS = [
+        {
+            "id": 101,
+            "title": "Database connection failure",
+            "description": "Cannot connect to primary database. All transactions failing.",
+            "correct_priority": "critical",
+            "correct_time": 30,
+            "correct_assignee": "database-team"
+        },
+        {
+            "id": 102,
+            "title": "User password reset request",
+            "description": "User cannot log in, needs password reset.",
+            "correct_priority": "medium",
+            "correct_time": 10,
+            "correct_assignee": "support"
+        },
+        {
+            "id": 103,
+            "title": "Feature request: Dark mode",
+            "description": "Would like dark mode option for dashboard.",
+            "correct_priority": "low",
+            "correct_time": 480,
+            "correct_assignee": "frontend"
+        }
+    ]
+    
+    def reset(self) -> Observation:
+        super().reset()
+        self.task_state["ticket_index"] = 0
+        self.task_state["current_ticket"] = self.TEST_TICKETS[0]
+        self.task_state["score"] = 0.0
+        
+        return Observation(
+            done=False,
+            reward=0.0,
+            observation={
+                "task": "support_ticket",
+                "ticket": self.task_state["current_ticket"],
+                "priorities": SupportTicketGrader.PRIORITIES
+            }
+        )
+    
+    def _execute_action(self, action: Dict[str, Any]) -> Observation:
+        grader = SupportTicketGrader()
+        score, meta = grader.grade(self.task_state, action, self._state.step_count)
+        
+        self.task_state["score"] += score
+        self.task_state["ticket_index"] += 1
+        
+        if self.task_state["ticket_index"] >= len(self.TEST_TICKETS):
+            final_score = self.task_state["score"] / len(self.TEST_TICKETS)
+            return Observation(
+                done=True,
+                reward=final_score,
+                observation={"final_score": final_score},
+                metadata=meta
+            )
+        
+        self.task_state["current_ticket"] = self.TEST_TICKETS[self.task_state["ticket_index"]]
+        
+        return Observation(
+            done=False,
+            reward=score,
+            observation={
+                "ticket": self.task_state["current_ticket"],
+                "current_score": self.task_state["score"]
+            },
+            metadata=meta
+        )