README.md

---

title: Cloud Queue Env Environment Server
emoji: 🖨️
colorFrom: pink
colorTo: blue
sdk: docker
pinned: false
app_port: 8000
base_path: /web
tags:
  - openenv
---


# Cloud Queue Env Environment

A real-world queue-operations benchmark for OpenEnv.

This environment simulates service operations decisions humans make in production systems:
- Admission and rejection under load
- Queue routing and dispatching
- Priority handling for urgent traffic
- Capacity scaling under infrastructure cost constraints

The benchmark includes three deterministic tasks with partial graders in [0, 1]:
- easy: single-queue stability
- medium: multi-server priority routing
- hard: two-stage queue network with scaling

## Quick Start

Use the CloudQueueEnv client to connect to a running server or container:

```python

from cloud_queue_env import CloudQueueAction, CloudQueueEnv



try:

    env = CloudQueueEnv.from_docker_image("cloud_queue_env-env:latest")



    # Configure task + seed, then reset into that deterministic episode

    env.reset()

    env.step(CloudQueueAction(action_type="configure_task", task_id="easy", seed=11))

    result = env.reset()



    for _ in range(20):

        obs = result.observation

        if obs.incoming_job_present:

            action = CloudQueueAction(action_type="admit", target_queue=0)

        else:

            action = CloudQueueAction(action_type="dispatch", target_queue=0)



        result = env.step(action)

        print(

            f"step={obs.sim_time} queues={obs.queue_lengths} "

            f"reward={result.reward:.3f} done={result.done}"

        )

        if result.done:

            break



    final_score = result.observation.metadata.get("episode_score", 0.0)

    print(f"episode_score={final_score:.3f}")



finally:

    env.close()

```

The CloudQueueEnv.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 cloud_queue_env-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
CloudQueueAction fields:
- action_type: one of configure_task, admit, reject, route, dispatch, scale, reprioritize, noop
- target_queue: queue index for route/dispatch/admit

- target_server: optional server index
- scale_delta: server delta for scale action

- new_priority: new priority value for reprioritize
- task_id: easy/medium/hard (used with configure_task)
- seed: deterministic task seed (used with configure_task)



### Observation

CloudQueueObservation includes:

- task_id, sim_time, horizon

- queue_lengths, queue_wait_ema
- server_busy, server_remaining_service, utilization

- incoming_job_present, incoming_job_size, incoming_job_priority, incoming_job_deadline, incoming_job_type

- sla_violation_rate, abandonment_rate, throughput_recent, energy_cost_rate

- level, optional_history, action_mask

- reward, done, metadata



### Reward

Per-step reward is dense and multi-objective:



$$

r_t = 0.35R_{wait} + 0.20R_{throughput} + 0.20R_{sla} + 0.15R_{cost} + 0.05R_{fair} + 0.05R_{safe}
$$

Properties:
- Partial progress signal over the full trajectory
- Penalties for invalid actions and unsafe/noop behavior under congestion
- Bounded reward values for stability

### Deterministic Graders
Each task returns a deterministic episode_score in [0, 1], stored in observation metadata.



- easy score uses avg wait, throughput, rejection rate, and SLA violations

- medium score uses urgent/normal p95 waits, urgent SLA, throughput, and action cost

- hard score uses end-to-end p95, abandonment, SLA, throughput, infra cost, and fairness gap



If invalid action rate exceeds threshold, score is capped.



## Tasks



1. easy (single queue stability)

- one queue, one server

- objective: low wait with acceptable throughput and low rejection



2. medium (priority routing)

- two queues and multiple servers

- objective: protect urgent traffic while maintaining total performance



3. hard (queue network + scaling)

- two-stage queue network with bursty arrivals and heavy-tailed service times

- objective: balance latency/SLA/abandonment against infra cost and fairness



## Baseline Inference



Run baseline inference across easy/medium/hard:



```bash

API_KEY=your_provider_key python inference.py
```



Optional variables:

- API_KEY (OpenAI-compatible provider key for model calls)

- API_BASE_URL (default: https://router.huggingface.co/v1)

- MODEL_NAME (default: Qwen/Qwen2.5-72B-Instruct)

- BASE_URL (if using deployed space)

- IMAGE_NAME (if launching local docker image)

- USE_HEURISTIC_ONLY (true/false)

- DISABLE_MODEL_ON_FIRST_ERROR (true/false)

- MAX_STEPS_OVERRIDE (integer quick-test cap)

- TASK_SEEDS_JSON (JSON map for multi-seed runs)

- ACTION_TRACE_FILE (JSON replay file keyed by task:seed)

- REPORT_JSON_PATH (write seed/task report JSON)

- REPORT_CSV_PATH (write per-seed report CSV)



Output includes required line types:

- [START]

- [STEP]

- [END]



And final aggregate summary:

- [SUMMARY] easy=<...> medium=<...> hard=<...> final=<...>



V2 reporting also includes:

- [REPORT_SEED] task=<task_id> seed=<seed> score=<score> steps=<n> trace=<digest>

- [REPORT] task=<task_id> seeds=<n> mean=<score> std=<score> ci95=<score>



## Baseline Scores



Current reproducible heuristic-only baseline (deployed runtime, single seed per task):



| Task | Seed Count | Mean Score |

|---|---:|---:|

| easy | 1 | 0.000 |

| medium | 1 | 0.000 |

| hard | 1 | 0.000 |

| final (mean of task means) | - | 0.000 |



Notes:

- These values are from heuristic fallback mode and are expected to be low.

- Model-based scores depend on provider/model availability and should be recorded from a successful funded run.

- Keep this table updated with your latest official benchmark run before final submission.



## Advanced Usage



### Connecting to an Existing Server



If you already have a Cloud Queue Env environment server running, you can connect directly:



```python

from cloud_queue_env import CloudQueueAction, CloudQueueEnv



# Connect to existing server

cloud_queue_envenv = CloudQueueEnv(base_url="<ENV_HTTP_URL_HERE>")



# Use as normal

result = cloud_queue_envenv.reset()

result = cloud_queue_envenv.step(CloudQueueAction(action_type="dispatch", target_queue=0))

```

Note: When connecting to an existing server, `cloud_queue_envenv.close()` will NOT stop the server.

### Using the Context Manager

The client supports context manager usage for automatic connection management:

```python

from cloud_queue_env import CloudQueueAction, CloudQueueEnv



# Connect with context manager (auto-connects and closes)

with CloudQueueEnv(base_url="http://localhost:8000") as env:

    result = env.reset()

    print(f"Initial queues: {result.observation.queue_lengths}")

    # Multiple steps with low latency

    for _ in range(10):

        result = env.step(CloudQueueAction(action_type="noop"))

        print(f"Reward: {result.reward:.3f}")

```

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(

    CloudQueueEnvironment,  # Pass class, not instance

    CloudQueueAction,

    CloudQueueObservation,

    max_concurrent_envs=4,  # Allow 4 concurrent sessions

)

```

Then multiple clients can connect simultaneously:

```python

from cloud_queue_env import CloudQueueAction, CloudQueueEnv

from concurrent.futures import ThreadPoolExecutor



def run_episode(client_id: int):

    with CloudQueueEnv(base_url="http://localhost:8000") as env:

        result = env.reset()

        for i in range(10):

            result = env.step(CloudQueueAction(action_type="dispatch", target_queue=i % 2))

        return client_id, result.observation.queue_lengths



# Run 4 episodes concurrently

with ThreadPoolExecutor(max_workers=4) as executor:

    results = list(executor.map(run_episode, range(4)))

```

## Development & Testing

### Direct Environment Testing

Core files:
- models: typed action/observation schema
- server environment: queue simulation, reward shaping, grading
- inference script: task sweep and benchmark logging

### Running Locally

Run the server locally for development:

```bash

uvicorn server.app:app --reload

```

## Project Structure

```

cloud_queue_env/

├── .dockerignore

├── __init__.py

├── README.md

├── openenv.yaml

├── pyproject.toml

├── client.py

├── models.py

├── inference.py

├── IMPLEMENTATION_ROADMAP.md

└── server/

    ├── __init__.py

    ├── cloud_queue_env_environment.py

    ├── app.py

    └── Dockerfile

```

TASK A — Easy (150 steps)
  Scenario:  1 queue, 1 server (M/M/1), only admit/reject/dispatch
  Objective: Keep wait low while processing throughput
  Grader:    score = 0.40×(1-avg_wait/6) + 0.30×(throughput/70)

                   + 0.15×(1-rejection_rate/0.3) + 0.15×(1-sla_breaches/0.3)

TASK B — Medium (200 steps)

  Scenario:  2 queues, 3 servers, 28% urgent jobs → route + reprioritize

  Objective: Protect urgent SLA while not starving normal jobs

  Grader:    score = 0.35×urgent_wait_score + 0.25×urgent_sla_score

                   + 0.15×normal_wait_score + 0.15×throughput + 0.10×cost

TASK C — Hard (250 steps)

  Scenario:  2-stage pipeline, 1–6 servers, heavy-tail service, abandonments

  Objective: Maximize quality under budget with fairness

  Grader:    score = 0.25×e2e_latency + 0.20×abandonment + 0.20×sla
                   + 0.15×throughput + 0.10×cost + 0.10×fairness