feat: Implement COEnv Executor and Graders for task management

- Added `executor.py` to handle action execution and manage task states.
- Created grading modules for different tasks: `grader_autoscaling.py`, `grader_incident.py`, and `grader_pod_recovery.py`.
- Introduced Pydantic models in `models.py` for structured data representation.
- Removed outdated `requirements.txt`.
- Developed task implementations for autoscaling, incident handling, and pod recovery in `tasks` directory.
- Added utility functions in `utils.py` for realistic simulation of network and resource conditions.
- Implemented action validation in `validator.py` to ensure action integrity before execution.
- Created `worker.py` to manage episode execution and integrate with the executor and validator.
- Added tests for environment initialization in `test_environment.py`.

.opencode/agents/executor-agent.ts

@@ -0,0 +1,11 @@
+import { Agent } from "@opencode-ai/plugin"
+
+export const ExecutorAgent: Agent = {
+  name: "executor",
+  description: "Handles general tasks",
+
+  async run(ctx) {
+    return `Executing task: ${ctx.input}`
+  }
+}
+

.opencode/agents/memory-agent.ts

@@ -0,0 +1,20 @@
+import { Agent } from "@opencode-ai/plugin"
+
+export const MemoryAgent: Agent = {
+  name: "memory",
+  description: "Handles memory storage and retrieval",
+
+  async run(ctx) {
+    const input = ctx.input
+
+    if (input.includes("remember")) {
+      return "Saved to memory"
+    }
+
+    if (input.includes("recall")) {
+      return "Here is your memory"
+    }
+
+    return null
+  }
+}

.opencode/agents/planner-agent.ts

@@ -0,0 +1,20 @@
+import { Agent } from "@opencode-ai/plugin"
+
+export const PlannerAgent: Agent = {
+  name: "planner",
+  description: "Decides which agent to call",
+
+  async run(ctx) {
+    const input = ctx.input
+
+    if (input.includes("remember")) {
+      return ctx.call("memory-agent", input)
+    }
+
+    if (input.includes("search")) {
+      return ctx.call("web-agent", input)
+    }
+
+    return ctx.call("executor-agent", input)
+  }
+}

.opencode/agents/web-agent.ts

@@ -0,0 +1,10 @@
+import { Agent } from "@opencode-ai/plugin"
+
+export const WebAgent: Agent = {
+  name: "web",
+  description: "Fetches web data",
+
+  async run(ctx) {
+    return `Searching web for: ${ctx.input}`
+  }
+}

COEnv_Project_Documentation.md

@@ -0,0 +1,704 @@
+# COEnv — Project Documentation
+### Meta × Hugging Face OpenEnv RL Hackathon
+
+---
+
+## Table of Contents
+
+1. [What Is This Project?](#1-what-is-this-project)
+2. [Why Kubernetes?](#2-why-kubernetes)
+3. [How It Works — The Big Picture](#3-how-it-works--the-big-picture)
+4. [The Three Layers Explained](#4-the-three-layers-explained)
+5. [Team Ownership](#5-team-ownership)
+6. [Full Project Directory Structure](#6-full-project-directory-structure)
+7. [The Three Tasks (Easy → Medium → Hard)](#7-the-three-tasks-easy--medium--hard)
+8. [Reward & Grading Design](#8-reward--grading-design)
+9. [The Complete Episode Flow](#9-the-complete-episode-flow)
+10. [OpenEnv Spec Compliance Checklist](#10-openenv-spec-compliance-checklist)
+11. [Submission Checklist](#11-submission-checklist)
+12. [Key Technical Decisions](#12-key-technical-decisions)
+
+---
+
+## 1. What Is This Project?
+
+**COEnv** is a Reinforcement Learning environment that simulates real-world Kubernetes cluster operations. An AI agent (LLM) is placed inside a broken or degraded Kubernetes cluster and must figure out the right sequence of operations to fix it — just like a real Site Reliability Engineer (SRE) would.
+
+This is built for the **Meta × Hugging Face OpenEnv RL Hackathon**, which requires:
+- A real-world task simulation (not games or toys)
+- Full OpenEnv interface implementation (`step()`, `reset()`, `state()`)
+- At least 3 tasks with programmatic graders (easy → medium → hard)
+- A meaningful reward function that gives partial credit throughout the episode
+- A working `inference.py` that runs an LLM agent and logs structured output
+- Deployment on Hugging Face Spaces with a working Dockerfile
+
+**In simple terms:** We fake a Kubernetes cluster in Python memory, break it in specific ways, and then let an LLM try to fix it step by step — scoring it on how well it does.
+
+---
+
+## 2. Why Kubernetes?
+
+Kubernetes (k8s) is the industry-standard container orchestration system used by virtually every tech company running production software. Managing it is genuinely difficult and is a daily job for SREs and DevOps engineers worldwide.
+
+**Why it's a perfect RL environment:**
+
+| RL Concept | Kubernetes Equivalent |
+|---|---|
+| State | Cluster state (pod statuses, node health, resource usage) |
+| Action | kubectl commands (scale, patch, delete, restart) |
+| Reward | How close the cluster is to a healthy target state |
+| Episode | One incident recovery scenario |
+| Done | All SLOs restored / all pods healthy |
+
+**Why it's novel for OpenEnv:** None of Meta's reference environments (calendar, REPL, browser, CARLA, reasoning gym) touch infrastructure operations. This fills a real gap.
+
+**Why it's practical:** Companies would immediately use an environment like this to train or evaluate agents that assist SREs — the real-world utility score (30% of judging) is very high.
+
+---
+
+## 3. How It Works — The Big Picture
+
+Think of the project as three concentric layers:
+
+```
+┌─────────────────────────────────────────────────────────┐
+│                   LAYER 1 — RL ENVIRONMENT               │
+│  inference.py  ←→  main.py (FastAPI)  ←→  tasks/graders │
+│                      (Sandeep)                           │
+├─────────────────────────────────────────────────────────┤
+│                LAYER 2 — SIMULATION ENGINE               │
+│       world.py  ←→  models.py  ←→  conditions/          │
+│                        (You)                             │
+├─────────────────────────────────────────────────────────┤
+│                LAYER 3 — ACTION SPACE                    │
+│    worker.py  ←→  executor.py  ←→  actions/  ←→  validator│
+│                    (Third Person)                        │
+└─────────────────────────────────────────────────────────┘
+```
+
+**Layer 1 (Sandeep)** is what the judges see — the API endpoints, the inference script, the task definitions, the graders, the README.
+
+**Layer 2 (You)** is the fake Kubernetes cluster. It holds the state of the cluster, knows how pods transition between statuses, and can inject failures. Everything sits in Python dictionaries — no real Kubernetes cluster runs.
+
+**Layer 3 (Third Person)** is the action space — the specific operations the LLM agent is allowed to perform, and the validation/execution bridge that translates those actions into state changes in the simulator.
+
+---
+
+## 4. The Three Layers Explained
+
+### Layer 1 — RL Environment (Sandeep)
+
+This layer is the **public contract** of the project. It's what OpenEnv's `validate` command checks, what the judges' scripts call, and what the LLM agent talks to.
+
+**`main.py` — FastAPI application**
+
+The central API server. It exposes exactly three mandatory endpoints:
+
+- `POST /reset` — Starts a new episode. Sets up a broken cluster using one of the condition injectors. Returns the initial `ClusterObservation` (what the agent sees first).
+- `POST /step` — Receives an action from the agent. Validates it, executes it on the simulated cluster, advances time by one tick, and returns the new observation + reward + done flag + info.
+- `GET /state` — Returns the full current cluster state. Used for debugging and grading.
+
+**`inference.py` — LLM agent runner**
+
+This is the script the hackathon validators actually run. It:
+1. Reads `API_BASE_URL`, `MODEL_NAME`, `HF_TOKEN` from environment variables
+2. Calls `/reset` to start an episode
+3. Feeds the observation to the LLM using the OpenAI client
+4. Parses the LLM's response as a structured action
+5. Calls `/step` with that action
+6. Prints structured stdout logs after every step:
+   ```
+   [START] task=pod-recovery env=coenv model=Qwen3-VL-30B
+   [STEP] step=1 action=delete_pod('frontend-7d9f-xkp2') reward=0.20 done=false error=null
+   [STEP] step=2 action=scale('frontend',3) reward=0.60 done=false error=null
+   [END] success=true steps=2 rewards=0.20,0.60
+   ```
+7. Repeats until `done=true` or `max_steps` is reached
+
+**`openenv.yaml` — Spec metadata**
+
+Required for `openenv validate` to pass. Contains:
+- Environment name, version, description
+- List of task IDs with difficulty labels
+- References to the action schema and observation schema
+
+**`classes/tasks/` — Task definitions**
+
+Three Python files, each defining one task:
+- What the broken state looks like (which condition to inject)
+- What the agent's objective is (in plain English, passed to the LLM as a prompt)
+- What counts as success
+- Maximum number of steps allowed
+
+**`classes/graders/` — Reward graders**
+
+Three Python files, each implementing a `grade(world_state) -> float` function. Graders must be fully deterministic — same world state always returns same score. They implement partial credit: a grader doesn't just say "fixed or not fixed" but scores partial progress (e.g., 2 out of 5 pods fixed = 0.4).
+
+**`Dockerfile`**
+
+Single-stage Python container. Installs `requirements.txt`, copies the project, exposes port 8000, runs `uvicorn main:app`. Must build and run cleanly — this is a hard pass/fail gate.
+
+**`README.md`**
+
+Mandatory documentation. Must include: environment overview, motivation, action space definition, observation space definition, task descriptions with difficulty labels, setup instructions, baseline scores table.
+
+---
+
+### Layer 2 — Simulation Engine (You)
+
+This is the **most important layer technically**. It's what makes the environment believable. Since we cannot run a real Kubernetes cluster inside a 2 vCPU / 8 GB HF Space container, the entire cluster is simulated as an in-memory Python object.
+
+**`classes/world.py` — The cluster simulator**
+
+This is the brain of the project. It maintains the complete cluster state as a Python dictionary, structured like a real Kubernetes API response:
+
+```python
+cluster_state = {
+    "nodes": [
+        {"name": "node-1", "status": "Ready", "cpu_capacity": 4, "mem_capacity": 8192},
+        {"name": "node-2", "status": "NotReady", "cpu_capacity": 4, "mem_capacity": 8192}
+    ],
+    "deployments": [
+        {"name": "frontend", "desired_replicas": 3, "available_replicas": 1, "image": "nginx:1.21"}
+    ],
+    "pods": [
+        {"name": "frontend-7d9f-xkp2", "status": "CrashLoopBackOff", "node": "node-1", "restarts": 7},
+        {"name": "frontend-7d9f-ab3c", "status": "Running", "node": "node-1", "restarts": 0},
+        {"name": "frontend-7d9f-mn8x", "status": "Pending", "node": None, "restarts": 0}
+    ],
+    "services": [...],
+    "configmaps": [...],
+    "hpa": [...]
+}
+```
+
+Key methods:
+- `reset(condition)` — Wipes state, injects a failure condition, returns initial observation
+- `get_pods(namespace, selector)` — Returns filtered pod list (mimics `kubectl get pods`)
+- `apply_patch(resource_type, name, patch)` — Applies a patch to a resource
+- `scale(deployment_name, replicas)` — Changes replica count
+- `delete_pod(pod_name)` — Removes a pod (it gets recreated by the deployment controller on next tick)
+- `tick()` — Advances simulated time by one step. Pods in `CrashLoopBackOff` increment their restart counter. Pending pods on ready nodes eventually transition to `Running`. Dead nodes stay dead unless drained.
+- `get_observation()` — Serialises the current state into a `ClusterObservation` Pydantic model
+
+**`classes/models.py` — Pydantic typed models**
+
+All data structures are defined here. This is mandatory for OpenEnv spec compliance — typed models enforce the action/observation contract.
+
+```python
+class PodStatus(BaseModel):
+    name: str
+    status: Literal["Running", "Pending", "CrashLoopBackOff", "OOMKilled", "Terminating", "Unknown"]
+    node: Optional[str]
+    restarts: int
+    cpu_usage: float
+    mem_usage: float
+
+class NodeStatus(BaseModel):
+    name: str
+    status: Literal["Ready", "NotReady", "SchedulingDisabled"]
+    cpu_capacity: float
+    mem_capacity: float
+    cpu_usage: float
+    mem_usage: float
+
+class ClusterObservation(BaseModel):
+    nodes: List[NodeStatus]
+    pods: List[PodStatus]
+    deployments: List[DeploymentStatus]
+    services: List[ServiceStatus]
+    events: List[ClusterEvent]          # recent k8s events (error messages, warnings)
+    step: int
+    objective: str                      # plain English description of what to fix
+
+class RewardSignal(BaseModel):
+    reward: float                       # 0.0 to 1.0 incremental reward this step
+    cumulative: float                   # total reward so far
+    done: bool
+    info: Dict[str, Any]               # breakdown: why this reward was given
+```
+
+**`classes/conditions/` — Failure injectors**
+
+Each condition is a Python class with a single `inject(cluster_state) -> cluster_state` method that takes a healthy cluster and returns a broken one. This is how each task starts with a specific failure scenario:
+
+- `crash_loop.py` — Sets 3 pods to `CrashLoopBackOff` with high restart counts. Simulates a bad image tag or missing environment variable.
+- `oom_kill.py` — Sets pods to `OOMKilled`. Memory limits are set too low in the deployment spec. Pods keep restarting.
+- `node_failure.py` — Sets one node to `NotReady`. All pods on that node go to `Unknown`. New pods are `Pending` (no space to schedule).
+- `cascade_failure.py` — Combines multiple failures: one OOMKilled service causes downstream 503s in two dependent services, creating a cascading failure across 3 deployments.
+
+**`classes/utils.py` — Probability and simulation helpers**
+
+Utility functions that make the simulation feel realistic:
+- `sample_cpu_usage(base_load, noise_factor)` — Returns a slightly randomised CPU % (real clusters are never exactly at baseline)
+- `sample_latency(healthy_latency, degradation_factor)` — Simulates p95 request latency under load
+- `should_pod_recover(restarts, backoff_seconds)` — Determines if a `CrashLoopBackOff` pod would naturally recover (it usually won't — that's the point)
+- `generate_cluster_events(pod_list)` — Creates realistic k8s event messages like `"Back-off restarting failed container"` or `"OOMKilled: container exceeded memory limit"`
+
+**`config.json` — Cluster defaults**
+
+Single source of truth for all simulation parameters:
+
+```json
+{
+  "cluster": {
+    "num_nodes": 3,
+    "cpu_per_node": 4,
+    "mem_per_node_gb": 8
+  },
+  "tasks": {
+    "pod_recovery": { "max_steps": 15, "success_threshold": 0.9 },
+    "autoscaling":  { "max_steps": 20, "success_threshold": 0.85 },
+    "incident":     { "max_steps": 30, "success_threshold": 0.80 }
+  },
+  "simulation": {
+    "tick_interval_seconds": 30,
+    "crash_backoff_max_seconds": 300,
+    "hpa_cooldown_seconds": 180
+  }
+}
+```
+
+---
+
+### Layer 3 — Action Space & Workers (Third Person)
+
+This layer defines what the LLM is allowed to do, makes sure it's valid, and executes it against the simulator.
+
+**`classes/actions/` — Typed action definitions**
+
+Each action is a Pydantic model. The LLM must output one of these (Sandeep's inference.py prompts it to respond in JSON matching one of these schemas):
+
+```python
+class ScaleAction(BaseModel):
+    action_type: Literal["scale"]
+    deployment: str          # e.g. "frontend"
+    replicas: int            # e.g. 3
+
+class DeletePodAction(BaseModel):
+    action_type: Literal["delete_pod"]
+    pod_name: str            # e.g. "frontend-7d9f-xkp2"
+
+class PatchAction(BaseModel):
+    action_type: Literal["patch"]
+    resource_type: str       # "deployment" | "configmap" | "service"
+    name: str
+    patch: Dict[str, Any]   # the fields to update
+
+class RolloutRestartAction(BaseModel):
+    action_type: Literal["rollout_restart"]
+    deployment: str
+
+class SetHPAAction(BaseModel):
+    action_type: Literal["set_hpa"]
+    deployment: str
+    min_replicas: int
+    max_replicas: int
+    cpu_target_percent: int
+
+class DrainNodeAction(BaseModel):
+    action_type: Literal["drain_node"]
+    node_name: str
+
+class DescribeAction(BaseModel):
+    action_type: Literal["describe"]
+    resource_type: str
+    name: str               # "investigation" action — no state change, returns detail
+```
+
+**`classes/validator.py` — Action validation**
+
+Before any action touches the world state, the validator checks it:
+- Does the target resource exist? (Can't delete a pod that doesn't exist)
+- Is the scale value sane? (Can't scale to 0 or to 1000 replicas)
+- Is the node already drained? (Can't drain twice)
+- Is the deployment name a real deployment?
+
+If validation fails, it returns an error string. This flows directly into the `[STEP] error=` field in stdout logs. The step still counts against the agent's limit �� bad actions are penalised by wasting steps.
+
+**`classes/executor.py` — Action execution bridge**
+
+Maps each validated action type to the correct `world.py` method call:
+
+```python
+def execute(action: KubeAction, world: World) -> ExecutionResult:
+    if action.action_type == "scale":
+        world.scale(action.deployment, action.replicas)
+    elif action.action_type == "delete_pod":
+        world.delete_pod(action.pod_name)
+    elif action.action_type == "rollout_restart":
+        world.rollout_restart(action.deployment)
+    ...
+    world.tick()   # always advance time after an action
+    return ExecutionResult(observation=world.get_observation(), ...)
+```
+
+**`classes/worker.py` — Agent episode loop**
+
+Manages the full lifecycle of a single episode. Sandeep's `inference.py` calls this:
+
+```python
+class Worker:
+    def run_episode(self, task_id, world, max_steps) -> EpisodeResult:
+        obs = world.reset(task=task_id)
+        rewards = []
+        for step in range(1, max_steps + 1):
+            action = self.get_action(obs)          # calls LLM
+            result = executor.execute(action, world)
+            rewards.append(result.reward)
+            if result.done:
+                break
+        return EpisodeResult(rewards=rewards, steps=step, success=result.done)
+```
+
+---
+
+## 5. Team Ownership
+
+| Module | Owner | Why It's Their Responsibility |
+|---|---|---|
+| `main.py` | Sandeep | He owns the public API contract |
+| `inference.py` | Sandeep | He owns the hackathon submission script |
+| `openenv.yaml` | Sandeep | He owns spec compliance |
+| `Dockerfile` | Sandeep | He owns deployment |
+| `README.md` | Sandeep | He owns documentation |
+| `classes/tasks/` | Sandeep | He defines what success looks like |
+| `classes/graders/` | Sandeep | He owns the scoring logic |
+| `classes/world.py` | You | You own the cluster simulator |
+| `classes/models.py` | You | You own all typed data models |
+| `classes/utils.py` | You | You own simulation helpers |
+| `classes/conditions/` | You | You own failure injection |
+| `config.json` | You | You own all parameters |
+| `classes/worker.py` | Third person | They own the episode loop |
+| `classes/actions/` | Third person | They own the action space |
+| `classes/executor.py` | Third person | They own action execution |
+| `classes/validator.py` | Third person | They own action validation |
+| `tests/` | All three | Each writes tests for their own module |
+
+---
+
+## 6. Full Project Directory Structure
+
+```text
+COEnv/
+├── .dockerignore                  # Docker build exclusions
+├── __init__.py                    # Module exports
+├── README.md                      # Project documentation
+├── openenv.yaml                   # OpenEnv manifest
+├── pyproject.toml                 # Project metadata and dependencies
+├── uv.lock                        # Locked dependencies
+├── client.py                      # CoenvEnv client / inference-side runner
+├── models.py                      # Shared action and observation models
+├── config.json                    # Cluster defaults and simulation params
+├── mkdocs.yml                     # Docs site configuration
+├── tests/                         # End-to-end and unit tests
+│   ├── test_environment.py        # From test_world.py
+│   ├── test_conditions.py         # From test_conditions.py
+│   ├── test_models.py             # From test_models.py
+│   ├── test_actions.py            # From test_actions.py
+│   ├── test_executor.py           # From test_executor.py
+│   ├── test_graders.py            # From test_graders.py
+│   ├── test_tasks.py              # From test_tasks.py
+│   └── test_integration.py        # End-to-end reset→step→state flow
+└── server/
+    ├── __init__.py                # Server module exports
+    ├── COEnv_environment.py       # Core environment logic
+    ├── app.py                     # FastAPI app exposing /reset /step /state
+    ├── Dockerfile                 # Container image definition
+    ├── utils.py                   # Simulation helpers
+    ├── validator.py               # Action validation
+    ├── executor.py                # Action execution bridge
+    ├── worker.py                  # Episode loop manager
+    ├── tasks/
+    │   ├── __init__.py
+    │   ├── task_pod_recovery.py
+    │   ├── task_autoscaling.py
+    │   └── task_incident.py
+    ├── graders/
+    │   ├── __init__.py
+    │   ├── grader_pod_recovery.py
+    │   ├── grader_autoscaling.py
+    │   └── grader_incident.py
+    ├── conditions/
+    │   ├── __init__.py
+    │   ├── crash_loop.py
+    │   ├── oom_kill.py
+    │   ├── node_failure.py
+    │   └── cascade_failure.py
+    └── actions/
+        ├── __init__.py
+        ├── scale_action.py
+        ��── patch_action.py
+        ├── delete_pod_action.py
+        ├── rollout_action.py
+        ├── hpa_action.py
+        ├── drain_action.py
+        └── describe_action.py
+```
+
+---
+
+## 7. The Three Tasks (Easy → Medium → Hard)
+
+### Task 1 — Pod Recovery (Easy)
+
+**What's broken:** A frontend deployment has 3 pods stuck in `CrashLoopBackOff`. The restart count is climbing. The root cause is a wrong environment variable in the deployment spec pointing to a database host that doesn't exist.
+
+**What the agent must do:**
+1. Observe the broken pods and read the k8s events (which mention a connection refused error)
+2. Identify the bad `DB_HOST` environment variable using a `describe` or `patch` inspect action
+3. Patch the deployment with the correct `DB_HOST` value
+4. Optionally delete the crash-looping pods to speed up recovery (they'll get recreated with the new config)
+5. Verify all 3 pods reach `Running` state
+
+**Objective string shown to agent:** *"The frontend deployment is crash-looping. Diagnose and fix the root cause so that all pods reach Running state."*
+
+**Max steps:** 15  
+**Success threshold:** All 3 pods in `Running` state (score ≥ 0.9)
+
+**Partial rewards:**
+- +0.1 for each pod that stops crash-looping
+- +0.2 for correctly patching the environment variable
+- +0.3 bonus for all pods Running within 10 steps
+
+---
+
+### Task 2 — HPA Autoscaling Under Traffic Spike (Medium)
+
+**What's broken:** The cluster is healthy but receiving 10× normal traffic. The deployment has no HPA configured, is running on fixed 2 replicas, and is already at 95% CPU. Request latency is climbing past the SLO threshold.
+
+**What the agent must do:**
+1. Observe high CPU usage and rising latency in the observation
+2. Immediately scale up the deployment to handle current load
+3. Configure a HorizontalPodAutoscaler (HPA) with appropriate min/max replicas and CPU target
+4. Set correct CPU resource requests/limits on the deployment so HPA has a baseline to work with
+5. Verify that latency drops back below the SLO threshold
+
+**Objective string shown to agent:** *"Traffic has spiked 10×. The api-server deployment is overloaded. Configure autoscaling and ensure p95 latency stays below 500ms."*
+
+**Max steps:** 20  
+**Success threshold:** p95 latency < 500ms, HPA configured, replicas ≥ 4 (score ≥ 0.85)
+
+**Partial rewards:**
+- +0.15 for scaling up replicas immediately (within 3 steps)
+- +0.20 for configuring HPA correctly
+- +0.25 for latency dropping below 1000ms
+- +0.30 for latency dropping below 500ms (SLO met)
+- -0.10 penalty for scaling beyond 12 replicas unnecessarily (resource waste)
+
+---
+
+### Task 3 — Multi-Service Cascading Incident (Hard)
+
+**What's broken:** The `auth-service` deployment has pods getting OOMKilled because memory limits are set 4× too low relative to actual usage. This causes the `api-gateway` to fail authentication checks and return 503s. Downstream, the `data-processor` service is also throwing errors because it depends on the gateway. Three services are degraded simultaneously.
+
+**What the agent must do:**
+1. Identify the blast radius — which services are affected and why
+2. Investigate `auth-service` to find the OOMKill root cause (memory limits too low)
+3. Patch `auth-service` deployment with correct memory limits
+4. Rollout restart `auth-service` so new pods come up with correct limits
+5. Drain the partially-failed node where most OOMKilled pods were running, to force clean rescheduling
+6. Verify `api-gateway` 503 errors stop (automatically once auth recovers)
+7. Verify `data-processor` error rate drops (automatically once gateway recovers)
+8. Confirm all three services are fully healthy
+
+**Objective string shown to agent:** *"A cascading incident has degraded auth-service, api-gateway, and data-processor. Identify the root cause and restore all three services to healthy state without data loss."*
+
+**Max steps:** 30  
+**Success threshold:** All 3 services healthy, error rate < 0.1% (score ≥ 0.80)
+
+**Partial rewards:**
+- +0.10 for correctly identifying `auth-service` as the root cause (within 5 steps)
+- +0.15 for patching memory limits correctly
+- +0.15 for auth-service pods reaching Running
+- +0.20 for api-gateway 503s stopping
+- +0.20 for data-processor errors resolving
+- +0.10 for draining the bad node cleanly
+- -0.15 penalty for deleting services or breaking healthy components
+
+---
+
+## 8. Reward & Grading Design
+
+The grading philosophy follows what the PS requires: reward signal over the **full trajectory**, not just at the end.
+
+### Reward Principles
+
+**Partial progress is always rewarded.** If the agent fixes 1 out of 3 broken pods, it gets 1/3 of the maximum reward for that milestone — not zero.
+
+**Speed bonus.** Fixing the issue in fewer steps earns a small bonus. This incentivises efficient reasoning.
+
+**Waste penalty.** Unnecessary destructive actions (scaling to 0, deleting healthy pods, draining a healthy node) subtract from the reward. This teaches the agent to be surgical.
+
+**Idempotency.** Repeating the same correct action doesn't give extra reward but doesn't penalise either (except for wasted steps).
+
+### Grader Implementation Pattern
+
+Each grader implements:
+
+```python
+def grade(world_state: dict, step: int, max_steps: int) -> float:
+    score = 0.0
+
+    # Milestone 1: Partial progress
+    running_pods = [p for p in world_state["pods"] if p["status"] == "Running"]
+    score += (len(running_pods) / total_expected_pods) * 0.5
+
+    # Milestone 2: Full success
+    if all(p["status"] == "Running" for p in world_state["pods"]):
+        score += 0.4
+
+    # Speed bonus
+    efficiency = 1.0 - (step / max_steps)
+    score += efficiency * 0.1
+
+    return min(score, 1.0)  # always clamp to [0, 1]
+```
+
+---
+
+## 9. The Complete Episode Flow
+
+Here is the full step-by-step flow of one complete episode, from start to finish:
+
+```
+1. JUDGE / VALIDATOR runs:
+   python inference.py
+
+2. inference.py reads env vars:
+   API_BASE_URL, MODEL_NAME, HF_TOKEN
+
+3. inference.py calls:
+   POST /reset  { "task": "pod_recovery" }
+
+4. main.py receives /reset:
+   → Calls task_pod_recovery.get_condition()  →  crash_loop.inject(cluster_state)
+   → world.reset(broken_state)
+   → Returns ClusterObservation (3 CrashLoopBackOff pods, events, objective string)
+
+5. stdout prints:
+   [START] task=pod-recovery env=coenv model=Qwen3-30B
+
+6. inference.py builds LLM prompt:
+   "You are an SRE. Current cluster state: [observation JSON].
+    Objective: Fix the frontend deployment crash loop.
+    Respond with a JSON action from the available action types."
+
+7. LLM responds:
+   { "action_type": "describe", "resource_type": "deployment", "name": "frontend" }
+
+8. inference.py calls:
+   POST /step  { action }
+
+9. main.py receives /step:
+   → validator.validate(action, world)  →  OK
+   → executor.execute(action, world)
+   → world.tick()
+   → grader.grade(world.state, step=1) → reward=0.00 (just investigating)
+   → Returns observation, reward=0.00, done=false, info={...}
+
+10. stdout prints:
+    [STEP] step=1 action=describe('deployment','frontend') reward=0.00 done=false error=null
+
+11. LLM sees deployment spec, notices DB_HOST=wrong-host.internal
+    LLM responds: { "action_type": "patch", "resource_type": "deployment",
+                    "name": "frontend",
+                    "patch": {"env": [{"name": "DB_HOST", "value": "db.prod.internal"}]} }
+
+12. POST /step  { patch action }
+    → executor patches deployment in world state
+    → world.tick() — pods begin restarting with new config
+    → grader → reward=0.20 (correct patch applied)
+
+13. [STEP] step=2 action=patch('frontend',{env...}) reward=0.20 done=false error=null
+
+14. LLM responds: { "action_type": "delete_pod", "pod_name": "frontend-7d9f-xkp2" }
+    → world deletes pod, recreates with correct env, status → Running
+    → grader → reward=0.40
+
+15. Repeat for remaining 2 pods...
+
+16. All 3 pods Running. grader → reward=1.0, done=true
+
+17. stdout prints:
+    [END] success=true steps=8 rewards=0.00,0.20,0.40,0.55,0.70,0.85,0.95,1.00
+```
+
+---
+
+## 10. OpenEnv Spec Compliance Checklist
+
+| Requirement | File | Status |
+|---|---|---|
+| Typed Observation model | `classes/models.py` → `ClusterObservation` | Required |
+| Typed Action model | `classes/models.py` → `KubeAction` | Required |
+| Typed Reward model | `classes/models.py` → `RewardSignal` | Required |
+| `step(action) → (obs, reward, done, info)` | `main.py` → `POST /step` | Required |
+| `reset() → initial_observation` | `main.py` → `POST /reset` | Required |
+| `state() → current_state` | `main.py` → `GET /state` | Required |
+| `openenv.yaml` with metadata | `openenv.yaml` | Required |
+| `openenv validate` passes | Tested via pre-validation script | Required |
+| Min 3 tasks | `classes/tasks/` — 3 files | Required |
+| Easy → medium → hard difficulty | task_pod_recovery / task_autoscaling / task_incident | Required |
+| Graders return 0.0–1.0 | `classes/graders/` — 3 graders | Required |
+| Graders are deterministic | Pure functions, no randomness | Required |
+| Partial reward signals | All 3 graders implement milestone scoring | Required |
+| Penalise bad actions | validator.py + grader penalty terms | Required |
+| `inference.py` in root | `inference.py` | Required |
+| `[START]` log line | `inference.py` → `log_start()` | Required |
+| `[STEP]` log per step | `inference.py` → `log_step()` | Required |
+| `[END]` log always emitted | `inference.py` → `finally: log_end()` | Required |
+| Reads `API_BASE_URL` with default | `inference.py` | Required |
+| Reads `MODEL_NAME` with default | `inference.py` | Required |
+| Reads `HF_TOKEN` (no default) | `inference.py` | Required |
+| Uses OpenAI client | `from openai import OpenAI` | Required |
+| `Dockerfile` builds cleanly | `Dockerfile` | Required |
+| HF Space deploys and responds | Deployed on Hugging Face | Required |
+| Inference runs in < 20 min | Max 30 steps × ~20s/step = ~10 min | Required |
+| Runs in 2 vCPU / 8 GB RAM | Pure Python in-memory sim, no real k8s | Required |
+| README with all required sections | `README.md` | Required |
+
+---
+
+## 11. Submission Checklist
+
+Before submitting, verify all of these:
+
+- [ ] `inference.py` is in the **root directory** (not inside `classes/`)
+- [ ] `inference.py` has default values for `API_BASE_URL` and `MODEL_NAME`
+- [ ] `inference.py` raises `ValueError` if `HF_TOKEN` is missing
+- [ ] `[START]`, `[STEP]`, `[END]` format matches the spec **exactly** (field names, order, lowercase booleans)
+- [ ] `openenv validate` passes locally
+- [ ] `docker build` completes without errors
+- [ ] `docker run` starts the server and responds to `GET /state`
+- [ ] HF Space is in **Running** state (not Building, not Stopped)
+- [ ] All 3 tasks can be reset and stepped without crashing
+- [ ] All 3 graders return a float between 0.0 and 1.0
+- [ ] Running `inference.py` end-to-end completes in under 20 minutes
+- [ ] `README.md` includes baseline scores table
+- [ ] `tests/test_integration.py` passes cleanly
+
+---
+
+## 12. Key Technical Decisions
+
+### Why a simulated cluster, not a real one?
+
+Running `kind` or `minikube` inside a Hugging Face Space container with 2 vCPU / 8 GB RAM is not feasible. The Kubernetes control plane alone (etcd + apiserver + scheduler + controller-manager) consumes ~1.5–2 GB RAM before any workloads run. An in-memory Python simulator is the only viable approach within the hardware constraints. It is also faster (no scheduling latency), fully deterministic (same input = same output), and easier to test.
+
+### Why a constrained action space?
+
+Free-form kubectl text strings are nearly impossible to grade deterministically. By defining ~7 typed Pydantic action models, we make the action space clear to the LLM (easier to prompt), easy to validate (Pydantic does the type checking), and easy to grade (executor calls predictable world methods). This also keeps the action space small enough that the LLM can reason about it effectively without getting lost in kubectl's hundreds of sub-commands.
+
+### Why FastAPI?
+
+OpenEnv environments are expected to be HTTP servers. FastAPI gives automatic OpenAPI documentation (at `/docs`), Pydantic integration for request/response validation, async support for when we need it, and a clean decorator syntax that makes `main.py` easy to read. It is also trivial to run with `uvicorn` inside a Docker container.
+
+### Why partial rewards matter for the hackathon
+
+The PS explicitly states: *"The reward function must provide feedback throughout the task trajectory, not just at completion."* Binary rewards (0 until success, then 1) are explicitly penalised in the environment design score. Our graders implement milestone-based partial rewards, which also makes the environment more useful for actual RL training — sparse rewards make training slow and unstable.
+
+---
+
+*COEnv — Meta × Hugging Face OpenEnv RL Hackathon*  
+*Team: Sandeep (RL environment) · You (Simulation) · Third Person (Actions & Workers)*

config.json

@@ -0,0 +1,14 @@
+{
+  "num_nodes": 5,
+  "node_cpu_capacity": 4,
+  "node_mem_capacity": 8192,
+  "pod_cpu_request": 250,
+  "pod_mem_request": 128,
+  "pod_cpu_limit": 500,
+  "pod_mem_limit": 256,
+  "crash_loop_failure_rate": 0.7,
+  "oom_kill_failure_rate": 0.6,
+  "node_failure_rate": 0.3,
+  "cascade_failure_probability": 0.5,
+  "task_timeout_values": 300
+}

inference.py

@@ -0,0 +1,86 @@
+"""
+COEnv Inference Script
+Used by validators to run episodes with LLMs
+"""
+
+import os
+import sys
+import json
+import argparse
+import requests
+from typing import Dict, Any, Optional
+
+API_BASE_URL = os.getenv("API_BASE_URL", "http://localhost:8000")
+MODEL_NAME = os.getenv("MODEL_NAME", "Qwen3-30B")
+HF_TOKEN = os.getenv("HF_TOKEN")
+
+
+def main():
+    parser = argparse.ArgumentParser(description='Run COEnv inference')
+    parser.add_argument('--api-base-url', type=str, default=API_BASE_URL, help='Base URL for the COEnv API')
+    parser.add_argument('--model-name', type=str, default=MODEL_NAME, help='Name of the model to use')
+    parser.add_argument('--hf-token', type=str, default=HF_TOKEN, help='Hugging Face token (if needed)')
+    parser.add_argument('--task-id', type=str, default='pod_recovery', help='Task ID to run')
+    parser.add_argument('--max-steps', type=int, default=15, help='Maximum steps per episode')
+    
+    args = parser.parse_args()
+    
+    api_base_url = args.api_base_url.rstrip('/')
+    model_name = args.model_name
+    hf_token = args.hf_token or HF_TOKEN
+    task_id = args.task_id
+    max_steps = args.max_steps
+    
+    print(f"[START] task={task_id} env=coenv model={model_name}")
+    
+    reset_url = f"{api_base_url}/reset"
+    try:
+        response = requests.post(reset_url, json={"task": task_id})
+        response.raise_for_status()
+        observation = response.json()
+    except Exception as e:
+        print(f"[ERROR] Failed to reset environment: {e}")
+        return 1
+    
+    total_reward = []
+    
+    for step in range(1, max_steps + 1):
+        action = {
+            "action_type": "describe",
+            "resource_type": "deployment",
+            "name": "frontend"
+        }
+        action_str = f"describe('deployment','frontend')"
+        
+        step_url = f"{api_base_url}/step"
+        try:
+            response = requests.post(step_url, json={"action": action})
+            response.raise_for_status()
+            result = response.json()
+            
+            reward = result.get('reward', 0.0)
+            done = result.get('done', False)
+            info = result.get('info', {})
+            error_str = "null"
+            
+            if 'error' in info and info['error']:
+                error_str = f"\"{info['error']}\""
+            
+            total_reward.append(reward)
+            
+            print(f"[STEP] step={step} action={action_str} reward={reward:.2f} done={done} error={error_str}")
+            
+            if done:
+                print(f"[END] success={str(done).lower()} steps={step} rewards={total_reward}")
+                return 0
+                
+        except Exception as e:
+            print(f"[ERROR] Failed to step environment: {e}")
+            print(f"[STEP] step={step} action={action_str} reward=0.00 done=false error=\"{str(e)}\"")
+    
+    print(f"[END] success=false steps={max_steps} rewards={total_reward}")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

opencode.json

@@ -0,0 +1,18 @@
+{
+  "$schema": "https://opencode.ai/config.json",
+
+  "agent": {
+    "planner": {
+      "entry": "./.opencode/agents/planner-agent.ts"
+    },
+    "memory": {
+      "entry": "./.opencode/agents/memory-agent.ts"
+    },
+    "executor": {
+      "entry": "./.opencode/agents/executor-agent.ts"
+    },
+    "web": {
+      "entry": "./.opencode/agents/web-agent.ts"
+    }
+  }
+}

openenv.yaml

@@ -1,7 +1,20 @@
-spec_version: 1
-name: COEnv
-type: space
-runtime: fastapi
-app: server.app:app
-port: 8000
-
+name: coenv
+version: 0.1.0
+description: Kubernetes Cluster Simulator for OpenEnv RL Hackathon
+entrypoint: server/app.py
+inference: inference.py
+endpoints:
+  reset: /reset
+  step: /step
+  state: /state
+tasks:
+  - id: pod_recovery
+    difficulty: easy
+  - id: autoscaling
+    difficulty: medium
+  - id: incident
+    difficulty: hard
+artifacts:
+  action_schema: server/actions/
+  observation_schema: server/models.py
+  reward_schema: server/models.py

requirements.txt

@@ -0,0 +1,4 @@
+fastapi>=0.100.0
+uvicorn>=0.23.0
+pydantic>=2.0.0
+requests>=2.31.0

server/COEnv_environment.py

@@ -1,104 +1,430 @@
-# 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.
-
 """
-Coenv Environment Implementation.
-
-A simple test environment that echoes back messages sent to it.
-Perfect for testing HTTP server infrastructure.
+COEnv Environment - Cluster Simulator
+In-memory dict that holds cluster state: nodes, pods, deployments, services.
+Has methods like get_pods(), apply_patch(), tick() to advance time.
+This is the brain of the whole project.
 """
 
-from uuid import uuid4
-
-from openenv.core.env_server.interfaces import Environment
-from openenv.core.env_server.types import State
-
-try:
-    from ..models import CoenvAction, CoenvObservation
-except ImportError:
-    from models import CoenvAction, CoenvObservation
-
-
-class CoenvEnvironment(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 = CoenvEnvironment()
-        >>> obs = env.reset()
-        >>> print(obs.echoed_message)  # "Coenv environment ready!"
-        >>>
-        >>> obs = env.step(CoenvAction(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 COEnv environment."""
-        self._state = State(episode_id=str(uuid4()), step_count=0)
-        self._reset_count = 0
-
-    def reset(self) -> CoenvObservation:
-        """
-        Reset the environment.
-
-        Returns:
-            CoenvObservation with a ready message
-        """
-        self._state = State(episode_id=str(uuid4()), step_count=0)
-        self._reset_count += 1
-
-        return CoenvObservation(
-            echoed_message="Coenv environment ready!",
-            message_length=0,
-            done=False,
-            reward=0.0,
-        )
-
-    def step(self, action: CoenvAction) -> CoenvObservation:  # type: ignore[override]
-        """
-        Execute a step in the environment by echoing the message.
-
-        Args:
-            action: CoenvAction containing the message to echo
-
-        Returns:
-            CoenvObservation 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
+from typing import Dict, List, Any, Optional, Literal
+from datetime import datetime
+import random
+import time
 
-        return CoenvObservation(
-            echoed_message=message,
-            message_length=length,
-            done=False,
-            reward=reward,
-            metadata={"original_message": message, "step": self._state.step_count},
-        )
+from .models import (
+    NodeStatus, PodStatus, DeploymentStatus, ServiceStatus, 
+    ClusterEvent, ClusterObservation, KubeAction, RewardSignal,
+    ConfigMapStatus, HPAStatus
+)
 
-    @property
-    def state(self) -> State:
-        """
-        Get the current environment state.
 
-        Returns:
-            Current State with episode_id and step_count
-        """
-        return self._state
+class World:
+    """In-memory Kubernetes cluster simulator"""
+    
+    def __init__(self, config: Dict[str, Any]):
+        self.config = config
+        self.cluster_state = self._initialize_healthy_cluster()
+        self.step_count = 0
+        self.events = []
+        self._event_counter = 0
+        
+    def _initialize_healthy_cluster(self) -> Dict[str, List[Dict]]:
+        """Initialize a healthy cluster state based on config"""
+        nodes = []
+        for i in range(self.config.get("num_nodes", 3)):
+            nodes.append({
+                "name": f"node-{i+1}",
+                "status": "Ready",
+                "cpu_capacity": self.config.get("node_cpu_capacity", 4),
+                "mem_capacity": self.config.get("node_mem_capacity", 8192),
+                "cpu_usage": 0.0,
+                "mem_usage": 0.0,
+                "last_updated": datetime.now().isoformat()
+            })
+        
+        pods = []
+        deployments = []
+        services = []
+        configmaps = []
+        hpas = []
+        
+        # Create some default deployments and their pods
+        default_deployments = [
+            {"name": "frontend", "image": "nginx:1.21", "replicas": 3},
+            {"name": "backend", "image": "python:3.9", "replicas": 2},
+            {"name": "database", "image": "postgres:13", "replicas": 1},
+            {"name": "auth-service", "image": "auth:latest", "replicas": 2},
+            {"name": "api-gateway", "image": "nginx:alpine", "replicas": 2}
+        ]
+        
+        for dep in default_deployments:
+            deployments.append({
+                "name": dep["name"],
+                "desired_replicas": dep["replicas"],
+                "available_replicas": dep["replicas"],
+                "image": dep["image"],
+                "last_updated": datetime.now().isoformat()
+            })
+            
+            # Create pods for this deployment
+            for j in range(dep["replicas"]):
+                pod_name = f"{dep['name']}-{random.randint(1000, 9999)}-{''.join([chr(random.randint(97, 122)) for _ in range(5)])}"
+                pods.append({
+                    "name": pod_name,
+                    "status": "Running",
+                    "node": nodes[j % len(nodes)]["name"] if nodes else None,
+                    "restarts": 0,
+                    "cpu_request": self.config.get("pod_cpu_request", 500),
+                    "mem_request": self.config.get("pod_mem_request", 256),
+                    "cpu_limit": self.config.get("pod_cpu_limit", 1000),
+                    "mem_limit": self.config.get("pod_mem_limit", 512),
+                    "deployment": dep["name"],
+                    "last_updated": datetime.now().isoformat()
+                })
+        
+        # Create some default services
+        default_services = [
+            {"name": "frontend-service", "type": "ClusterIP", "ports": [{"port": 80, "targetPort": 80}]},
+            {"name": "backend-service", "type": "ClusterIP", "ports": [{"port": 8080, "targetPort": 8080}]},
+            {"name": "database-service", "type": "ClusterIP", "ports": [{"port": 5432, "targetPort": 5432}]},
+            {"name": "auth-service-service", "type": "ClusterIP", "ports": [{"port": 8000, "targetPort": 8000}]}
+        ]
+        
+        for svc in default_services:
+            services.append({
+                "name": svc["name"],
+                "type": svc["type"],
+                "ports": svc["ports"],
+                "selector": {"app": svc["name"].replace("-service", "")},
+                "cluster_ip": f"10.96.{len(services)+1}.{len(services)+1}",
+                "last_updated": datetime.now().isoformat()
+            })
+        
+        # Create some default configmaps
+        default_configmaps = [
+            {"name": "frontend-config", "data": {"DB_HOST": "db.prod.internal", "DB_PORT": "5432"}},
+            {"name": "backend-config", "data": {"LOG_LEVEL": "info", "CACHE_SIZE": "100"}},
+            {"name": "database-config", "data": {"MAX_CONNECTIONS": "100", "TIMEOUT": "30"}}
+        ]
+        
+        for cm in default_configmaps:
+            configmaps.append({
+                "name": cm["name"],
+                "data": cm["data"],
+                "last_updated": datetime.now().isoformat()
+            })
+        
+        # Create some default HPAs
+        default_hpas = [
+            {"name": "frontend-hpa", "min_replicas": 2, "max_replicas": 10, "cpu_target_percent": 70},
+            {"name": "backend-hpa", "min_replicas": 1, "max_replicas": 5, "cpu_target_percent": 80}
+        ]
+        
+        for hpa in default_hpas:
+            hpas.append({
+                "name": hpa["name"],
+                "min_replicas": hpa["min_replicas"],
+                "max_replicas": hpa["max_replicas"],
+                "current_replicas": hpa["min_replicas"],
+                "cpu_target_percent": hpa["cpu_target_percent"],
+                "last_updated": datetime.now().isoformat()
+            })
+        
+        return {
+            "nodes": nodes,
+            "pods": pods,
+            "deployments": deployments,
+            "services": services,
+            "configmaps": configmaps,
+            "hpas": hpas
+        }
+    
+    def get_pods(self, namespace: Optional[str] = None, selector: Optional[Dict[str, str]] = None) -> List[PodStatus]:
+        """Returns filtered pod list (mimics kubectl get pods)"""
+        pods = [PodStatus(**pod) for pod in self.cluster_state["pods"]]
+        # Simple filtering by namespace (not fully implemented - just returns all for now)
+        return pods
+    
+    def get_nodes(self) -> List[NodeStatus]:
+        """Get all nodes as Pydantic models"""
+        return [NodeStatus(**node) for node in self.cluster_state["nodes"]]
+    
+    def get_deployments(self) -> List[DeploymentStatus]:
+        """Get all deployments as Pydantic models"""
+        return [DeploymentStatus(**dep) for dep in self.cluster_state["deployments"]]
+    
+    def get_services(self) -> List[ServiceStatus]:
+        """Get all services as Pydantic models"""
+        return [ServiceStatus(**svc) for svc in self.cluster_state["services"]]
+    
+    def get_configmaps(self) -> List[ConfigMapStatus]:
+        """Get all configmaps as Pydantic models"""
+        return [ConfigMapStatus(**cm) for cm in self.cluster_state["configmaps"]]
+    
+    def get_hpas(self) -> List[HPAStatus]:
+        """Get all HPAs as Pydantic models"""
+        return [HPAStatus(**hpa) for hpa in self.cluster_state["hpas"]]
+    
+    def get_events(self) -> List[ClusterEvent]:
+        """Get all events"""
+        return self.events.copy()
+    
+    def apply_patch(self, resource_type: str, name: str, patch: Dict[str, Any]) -> bool:
+        """Apply a patch to a resource"""
+        try:
+            if resource_type == "deployment":
+                for dep in self.cluster_state["deployments"]:
+                    if dep["name"] == name:
+                        dep.update(patch)
+                        dep["last_updated"] = datetime.now().isoformat()
+                        if "desired_replicas" in patch or "available_replicas" in patch:
+                            self._update_pods_for_deployment(name, dep.get("desired_replicas", dep["desired_replicas"]))
+                        return True
+            
+            elif resource_type == "pod":
+                for pod in self.cluster_state["pods"]:
+                    if pod["name"] == name:
+                        pod.update(patch)
+                        pod["last_updated"] = datetime.now().isoformat()
+                        return True
+                        
+            elif resource_type == "node":
+                for node in self.cluster_state["nodes"]:
+                    if node["name"] == name:
+                        node.update(patch)
+                        node["last_updated"] = datetime.now().isoformat()
+                        return True
+                        
+            elif resource_type == "service":
+                for svc in self.cluster_state["services"]:
+                    if svc["name"] == name:
+                        svc.update(patch)
+                        svc["last_updated"] = datetime.now().isoformat()
+                        return True
+            
+            elif resource_type == "configmap":
+                for cm in self.cluster_state["configmaps"]:
+                    if cm["name"] == name:
+                        cm.update(patch)
+                        cm["last_updated"] = datetime.now().isoformat()
+                        return True
+            
+            elif resource_type == "hpa":
+                for hpa in self.cluster_state["hpas"]:
+                    if hpa["name"] == name:
+                        hpa.update(patch)
+                        hpa["last_updated"] = datetime.now().isoformat()
+                        return True
+            
+            return False
+        except Exception as e:
+            print(f"Error applying patch: {e}")
+            return False
+    
+    def _update_pods_for_deployment(self, deployment_name: str, desired_replicas: int):
+        """Update pods count for a deployment"""
+        current_pods = [p for p in self.cluster_state["pods"] if p.get("deployment") == deployment_name]
+        current_count = len(current_pods)
+        
+        if desired_replicas > current_count:
+            nodes = self.cluster_state["nodes"]
+            for i in range(desired_replicas - current_count):
+                deployment = next((d for d in self.cluster_state["deployments"] if d["name"] == deployment_name), None)
+                if deployment:
+                    pod_name = f"{deployment_name}-{random.randint(1000, 9999)}-{''.join([chr(random.randint(97, 122)) for _ in range(5)])}"
+                    node = nodes[i % len(nodes)] if nodes else None
+                    self.cluster_state["pods"].append({
+                        "name": pod_name,
+                        "status": "Pending",
+                        "node": node["name"] if node else None,
+                        "restarts": 0,
+                        "cpu_request": self.config.get("pod_cpu_request", 500),
+                        "mem_request": self.config.get("pod_mem_request", 256),
+                        "cpu_limit": self.config.get("pod_cpu_limit", 1000),
+                        "mem_limit": self.config.get("pod_mem_limit", 512),
+                        "deployment": deployment_name,
+                        "last_updated": datetime.now().isoformat()
+                    })
+        elif desired_replicas < current_count:
+            pods_to_remove = current_pods[desired_replicas:]
+            for pod in pods_to_remove:
+                self.cluster_state["pods"].remove(pod)
+    
+    def scale(self, deployment_name: str, replicas: int) -> bool:
+        """Changes replica count"""
+        return self.apply_patch("deployment", deployment_name, {"desired_replicas": replicas})
+    
+    def delete_pod(self, pod_name: str) -> bool:
+        """Removes a pod (it gets recreated by the deployment controller on next tick)"""
+        pod_index = None
+        for i, pod in enumerate(self.cluster_state["pods"]):
+            if pod["name"] == pod_name:
+                pod_index = i
+                break
+        
+        if pod_index is not None:
+            del self.cluster_state["pods"][pod_index]
+            
+            event_type: Literal["Normal"] = "Normal"  # type: ignore
+            event = ClusterEvent(
+                event_id=f"event-delpod-{random.randint(1000, 9999)}",
+                timestamp=datetime.now().isoformat(),
+                type=event_type,
+                reason="UserDeleted",
+                message=f"pod/{pod_name} deleted by user",
+                involved_object=pod_name
+            )
+            self.events.append(event)
+            
+            return True
+        return False
+    
+    def rollout_restart(self, deployment: str) -> bool:
+        """Restart a deployment rollout"""
+        # Delete all pods for this deployment - they'll get recreated with new config
+        pods_to_delete = [p for p in self.cluster_state["pods"] if p.get("deployment") == deployment]
+        
+        for pod in pods_to_delete:
+            event_type: Literal["Normal"] = "Normal"  # type: ignore
+            event = ClusterEvent(
+                event_id=f"event-restart-{random.randint(1000, 9999)}",
+                timestamp=datetime.now().isoformat(),
+                type=event_type,
+                reason="RolledOut",
+                message=f"Deployment {deployment} rollout restart triggered",
+                involved_object=deployment
+            )
+            self.events.append(event)
+        
+        # Delete pods - they'll be recreated on next tick
+        self.cluster_state["pods"] = [p for p in self.cluster_state["pods"] if p.get("deployment") != deployment]
+        
+        return True
+    
+    def tick(self):
+        """Advances simulated time by one step. Pods in CrashLoopBackOff increment their restart counter. Pending pods on ready nodes eventually transition to Running. Dead nodes stay dead unless drained."""
+        self.step_count += 1
+        
+        # Simulate some natural changes in resource usage
+        for node in self.cluster_state["nodes"]:
+            node["cpu_usage"] = max(0, min(100, node["cpu_usage"] + random.uniform(-5, 5)))
+            node["mem_usage"] = max(0, min(100, node["mem_usage"] + random.uniform(-5, 5)))
+            node["last_updated"] = datetime.now().isoformat()
+        
+        # Update pod statuses based on node status
+        for pod in self.cluster_state["pods"]:
+            node_name = pod.get("node")
+            if node_name:
+                node = next((n for n in self.cluster_state["nodes"] if n["name"] == node_name), None)
+                if node and node["status"] != "Ready":
+                    if pod["status"] == "Running":
+                        pod["status"] = "Unknown"
+                    elif pod["status"] == "Pending":
+                        pod["status"] = "Unknown"
+                elif node and node["status"] == "Ready" and pod["status"] == "Pending":
+                    if random.random() > 0.7:
+                        pod["status"] = "Running"
+            pod["last_updated"] = datetime.now().isoformat()
+        
+        # Update deployment available replicas based on running pods
+        for deployment in self.cluster_state["deployments"]:
+            running_pods = [p for p in self.cluster_state["pods"] 
+                          if p.get("deployment") == deployment["name"] and p["status"] == "Running"]
+            deployment["available_replicas"] = len(running_pods)
+            deployment["last_updated"] = datetime.now().isoformat()
+        
+        # Re-create pods for deployments that need them
+        for deployment in self.cluster_state["deployments"]:
+            desired = deployment.get("desired_replicas", 0)
+            current_pods = [p for p in self.cluster_state["pods"] if p.get("deployment") == deployment["name"]]
+            current_count = len(current_pods)
+            
+            if current_count < desired:
+                nodes = self.cluster_state["nodes"]
+                for i in range(desired - current_count):
+                    pod_name = f"{deployment['name']}-{random.randint(1000, 9999)}-{''.join([chr(random.randint(97, 122)) for _ in range(5)])}"
+                    node = nodes[i % len(nodes)] if nodes else None
+                    self.cluster_state["pods"].append({
+                        "name": pod_name,
+                        "status": "Running",
+                        "node": node["name"] if node else None,
+                        "restarts": 0,
+                        "cpu_request": self.config.get("pod_cpu_request", 500),
+                        "mem_request": self.config.get("pod_mem_request", 256),
+                        "cpu_limit": self.config.get("pod_cpu_limit", 1000),
+                        "mem_limit": self.config.get("pod_mem_limit", 512),
+                        "deployment": deployment["name"],
+                        "last_updated": datetime.now().isoformat()
+                    })
+        
+        # Generate occasional events
+        if random.random() < 0.3:
+            self._generate_event()
+    
+    def _generate_event(self):
+        """Generate a realistic cluster event"""
+        event_types = [
+            {"type": "Normal", "reason": "Scheduled", "message": "Successfully assigned node"},
+            {"type": "Warning", "reason": "FailedScheduling", "message": "0/3 nodes are available: 3 Insufficient cpu."},
+            {"type": "Normal", "reason": "Pulling", "message": "Pulling image \"nginx:1.21\""},
+            {"type": "Normal", "reason": "Pulled", "message": "Successfully pulled image \"nginx:1.21\""},
+            {"type": "Normal", "reason": "Created", "message": "Created container"},
+            {"type": "Normal", "reason": "Started", "message": "Started container"},
+            {"type": "Warning", "reason": "BackOff", "message": "Back-off restarting failed container"},
+            {"type": "Normal", "reason": "Killing", "message": "Stopping container"}
+        ]
+        
+        event = random.choice(event_types)
+        involved_objects = []
+        involved_objects.extend([p["name"] for p in self.cluster_state["pods"][:3]])
+        involved_objects.extend([d["name"] for d in self.cluster_state["deployments"][:3]])
+        involved_objects.extend([n["name"] for n in self.cluster_state["nodes"][:3]])
+        
+        if not involved_objects:
+            involved_objects = ["cluster"]
+            
+        event_type: Literal["Normal", "Warning"] = event["type"]  # type: ignore
+        self.events.append(ClusterEvent(
+            event_id=f"event-{self._event_counter:04d}",
+            timestamp=datetime.now().isoformat(),
+            type=event_type,
+            reason=event["reason"],
+            message=event["message"],
+            involved_object=random.choice(involved_objects)
+        ))
+        self._event_counter += 1
+        
+        if len(self.events) > 100:
+            self.events = self.events[-50:]
+    
+    def get_full_state(self) -> Dict[str, Any]:
+        """Get the full cluster state for debugging"""
+        return {
+            "nodes": self.get_nodes(),
+            "pods": self.get_pods(),
+            "deployments": self.get_deployments(),
+            "services": self.get_services(),
+            "configmaps": self.get_configmaps(),
+            "hpas": self.get_hpas(),
+            "events": self.get_events(),
+            "step": self.step_count
+        }
+    
+    def reset_to_healthy(self):
+        """Reset cluster to healthy state"""
+        self.cluster_state = self._initialize_healthy_cluster()
+        self.step_count = 0
+        self.events = []
+        self._event_counter = 0
+    
+    def reset(self, condition=None):
+        """Reset the world state and optionally inject a failure condition"""
+        self.reset_to_healthy()
+        if condition:
+            condition.inject(self)
+        return self.get_observation()
+    
+    def get_observation(self, objective: str = "Maintain cluster health"):
+        """Serialises the current state into a ClusterObservation Pydantic model"""
+        observation_dict = self.get_full_state()
+        observation_dict["objective"] = objective
+        return ClusterObservation(**observation_dict)

server/Dockerfile

@@ -1,80 +0,0 @@
-# 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=COEnv
-
-# 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

@@ -1,11 +1,25 @@
-# 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.
+"""COEnv - Kubernetes Cluster Simulator for OpenEnv"""
 
-"""Coenv environment server components."""
+__version__ = "0.1.0"
 
-from .COEnv_environment import CoenvEnvironment
+from .COEnv_environment import World
+from .models import (
+    ClusterObservation,
+    RewardSignal,
+    KubeAction,
+    PodStatus,
+    NodeStatus,
+    DeploymentStatus,
+    ServiceStatus
+)
 
-__all__ = ["CoenvEnvironment"]
+__all__ = [
+    "World",
+    "ClusterObservation",
+    "RewardSignal",
+    "KubeAction",
+    "PodStatus",
+    "NodeStatus",
+    "DeploymentStatus",
+    "ServiceStatus"
+]

server/actions/__init__.py

@@ -0,0 +1,13 @@
+"""COEnv Actions - Action definitions"""
+
+__all__ = ["scale_action", "delete_pod_action", "patch_action", "rollout_action", "hpa_action", "drain_action", "describe_action"]
+
+from .scale_action import ScaleAction
+from .delete_pod_action import DeletePodAction
+from .patch_action import PatchAction
+from .rollout_action import RolloutRestartAction
+from .hpa_action import SetHPAAction
+from .drain_action import DrainNodeAction
+from .describe_action import DescribeAction
+
+__all__ += ["ScaleAction", "DeletePodAction", "PatchAction", "RolloutRestartAction", "SetHPAAction", "DrainNodeAction", "DescribeAction"]

server/actions/delete_pod_action.py

@@ -0,0 +1,10 @@
+"""COEnv Actions - Delete pod action"""
+
+from pydantic import BaseModel, Field
+from typing import Literal
+
+
+class DeletePodAction(BaseModel):
+    """Delete a specific pod"""
+    action_type: Literal["delete_pod"] = "delete_pod"
+    pod_name: str = Field(..., description="Pod name to delete")

server/actions/describe_action.py

@@ -0,0 +1,11 @@
+"""COEnv Actions - Describe action"""
+
+from pydantic import BaseModel, Field
+from typing import Literal
+
+
+class DescribeAction(BaseModel):
+    """Describe/get details of a resource"""
+    action_type: Literal["describe"] = "describe"
+    resource_type: Literal["deployment", "pod", "node", "service", "configmap"] = Field(..., description="Resource type")
+    name: str = Field(..., description="Resource name")

server/actions/drain_action.py

@@ -0,0 +1,10 @@
+"""COEnv Actions - Drain action"""
+
+from pydantic import BaseModel, Field
+from typing import Literal
+
+
+class DrainNodeAction(BaseModel):
+    """Drain a node (evict all pods)"""
+    action_type: Literal["drain_node"] = "drain_node"
+    node_name: str = Field(..., description="Node name to drain")

server/actions/hpa_action.py

@@ -0,0 +1,13 @@
+"""COEnv Actions - HPA action"""
+
+from pydantic import BaseModel, Field
+from typing import Literal
+
+
+class SetHPAAction(BaseModel):
+    """Set HorizontalPodAutoscaler for a deployment"""
+    action_type: Literal["set_hpa"] = "set_hpa"
+    deployment: str = Field(..., description="Deployment name")
+    min_replicas: int = Field(..., ge=1, le=50, description="Minimum replicas")
+    max_replicas: int = Field(..., ge=1, le=100, description="Maximum replicas")
+    cpu_target_percent: int = Field(..., ge=1, le=100, description="CPU target percentage")

server/actions/patch_action.py

@@ -0,0 +1,12 @@
+"""COEnv Actions - Patch action"""
+
+from pydantic import BaseModel, Field
+from typing import Literal, Dict, Any
+
+
+class PatchAction(BaseModel):
+    """Patch a resource with specific changes"""
+    action_type: Literal["patch"] = "patch"
+    resource_type: Literal["deployment", "pod", "node", "service", "configmap"] = Field(..., description="Resource type")
+    name: str = Field(..., description="Resource name")
+    patch: Dict[str, Any] = Field(..., description="Patch to apply")

server/actions/rollout_action.py

@@ -0,0 +1,10 @@
+"""COEnv Actions - Rollout restart action"""
+
+from pydantic import BaseModel, Field
+from typing import Literal
+
+
+class RolloutRestartAction(BaseModel):
+    """Restart a deployment rollout"""
+    action_type: Literal["rollout_restart"] = "rollout_restart"
+    deployment: str = Field(..., description="Deployment name to restart")

server/actions/scale_action.py

@@ -0,0 +1,11 @@
+"""COEnv Actions - Scale action"""
+
+from pydantic import BaseModel, Field
+from typing import Literal
+
+
+class ScaleAction(BaseModel):
+    """Scale a deployment to a specific replica count"""
+    action_type: Literal["scale"] = "scale"
+    deployment: str = Field(..., description="Deployment name to scale")
+    replicas: int = Field(..., ge=0, le=100, description="Number of replicas")

server/app.py

@@ -1,84 +1,283 @@
-# 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 Coenv Environment.
+COEnv FastAPI Application
+Exposes /reset /step /state endpoints
+"""
 
-This module creates an HTTP server that exposes the CoenvEnvironment
-over HTTP and WebSocket endpoints, compatible with EnvClient.
+from fastapi import FastAPI, HTTPException
+from pydantic import BaseModel, Field
+from typing import Dict, Any, Optional, List, Literal
+import uvicorn
+import json
+import os
+import sys
 
-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
+from .COEnv_environment import World
+from .models import ClusterObservation, RewardSignal, KubeAction
 
-Usage:
-    # Development (with auto-reload):
-    uvicorn server.app:app --reload --host 0.0.0.0 --port 8000
+app = FastAPI(title="COEnv", description="Kubernetes Simulator for OpenEnv")
 
-    # Production:
-    uvicorn server.app:app --host 0.0.0.0 --port 8000 --workers 4
+# Global world instance
+world_instance: Optional[World] = None
+config: Dict[str, Any] = {}
+current_task: Optional[str] = None
+current_objective: str = ""
 
-    # 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
+def load_config():
+    """Load configuration from config.json"""
+    global config
+    config_path = os.path.join(os.path.dirname(__file__), "..", "config.json")
+    try:
+        with open(config_path, 'r') as f:
+            config = json.load(f)
+    except FileNotFoundError:
+        config = {
+            "num_nodes": 3,
+            "node_cpu_capacity": 4,
+            "node_mem_capacity": 8192,
+            "pod_cpu_request": 250,
+            "pod_mem_request": 128,
+            "pod_cpu_limit": 500,
+            "pod_mem_limit": 256,
+            "crash_loop_failure_rate": 0.7,
+            "oom_kill_failure_rate": 0.6,
+            "node_failure_rate": 0.3,
+            "cascade_failure_probability": 0.5,
+            "task_timeout_values": 300,
+            "tasks": {
+                "pod_recovery": {"max_steps": 15, "success_threshold": 0.9},
+                "autoscaling": {"max_steps": 20, "success_threshold": 0.85},
+                "incident": {"max_steps": 30, "success_threshold": 0.80}
+            }
+        }
 
-try:
-    from ..models import CoenvAction, CoenvObservation
-    from .COEnv_environment import CoenvEnvironment
-except ModuleNotFoundError:
-    from models import CoenvAction, CoenvObservation
-    from server.COEnv_environment import CoenvEnvironment
 
+# Import conditions for task injection
+def get_condition_for_task(task_id: str):
+    """Get the condition injector for a task"""
+    if task_id == "pod_recovery":
+        from .conditions.crash_loop import CrashLoopCondition
+        return CrashLoopCondition(world_instance, config)
+    elif task_id == "autoscaling":
+        from .conditions.oom_kill import OOMKillCondition
+        return OOMKillCondition(world_instance, config)
+    elif task_id == "incident":
+        from .conditions.cascade_failure import CascadeFailureCondition
+        return CascadeFailureCondition(world_instance, config)
+    return None
 
-# Create the app with web interface and README integration
-app = create_app(
-    CoenvEnvironment,
-    CoenvAction,
-    CoenvObservation,
-    env_name="COEnv",
-    max_concurrent_envs=1,  # increase this number to allow more concurrent WebSocket sessions
-)
 
+def get_objective_for_task(task_id: str) -> str:
+    """Get the objective string for a task"""
+    objectives = {
+        "pod_recovery": "The frontend deployment is crash-looping. Diagnose and fix the root cause so that all pods reach Running state.",
+        "autoscaling": "Traffic has spiked 10×. The api-server deployment is overloaded. Configure autoscaling and ensure p95 latency stays below 500ms.",
+        "incident": "A cascading incident has degraded auth-service, api-gateway, and data-processor. Identify the root cause and restore all three services to healthy state without data loss."
+    }
+    return objectives.get(task_id, "Maintain cluster health")
 
-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 COEnv.server.app
+@app.on_event("startup")
+async def startup_event():
+    """Initialize the world on startup"""
+    global world_instance, current_task, current_objective
+    load_config()
+    world_instance = World(config)
+    print("COEnv initialized")
 
-    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 COEnv.server.app:app --workers 4
-    """
-    import uvicorn
+class ResetRequest(BaseModel):
+    """Request body for /reset endpoint"""
+    task: Optional[str] = Field(default="pod_recovery", description="Task ID to initialize")
 
-    uvicorn.run(app, host=host, port=port)
 
+@app.post("/reset")
+async def reset(request: ResetRequest = ResetRequest()):
+    """Reset the environment and return initial observation"""
+    global world_instance, current_task, current_objective
+    
+    if world_instance is None:
+        raise HTTPException(status_code=500, detail="World not initialized")
+    
+    current_task = request.task
+    current_objective = get_objective_for_task(request.task)
+    
+    # Get condition for the task and inject it
+    condition = get_condition_for_task(request.task)
+    
+    # Reset with the condition
+    observation = world_instance.reset(condition)
+    
+    return observation
+
+
+class StepRequest(BaseModel):
+    """Request body for /step endpoint"""
+    action: Dict[str, Any] = Field(..., description="Action to execute")
+
+
+@app.post("/step")
+async def step(request: StepRequest):
+    """Apply an action and return next observation, reward, done, info"""
+    global world_instance, current_task, current_objective
+    
+    if world_instance is None:
+        raise HTTPException(status_code=500, detail="World not initialized")
+    
+    action = request.action
+    action_type = action.get("action_type", "")
+    
+    # Execute action
+    info = {}
+    reward = 0.0
+    done = False
+    
+    try:
+        if action_type == "scale":
+            deployment = action.get("deployment", "")
+            replicas = action.get("replicas", 1)
+            world_instance.scale(deployment, replicas)
+            info["scaled"] = deployment
+            info["replicas"] = replicas
+            
+        elif action_type == "delete_pod":
+            pod_name = action.get("pod_name", "")
+            world_instance.delete_pod(pod_name)
+            info["deleted"] = pod_name
+            
+        elif action_type == "patch":
+            resource_type = action.get("resource_type", "")
+            name = action.get("name", "")
+            patch = action.get("patch", {})
+            world_instance.apply_patch(resource_type, name, patch)
+            info["patched"] = f"{resource_type}/{name}"
+            
+        elif action_type == "rollout_restart":
+            deployment = action.get("deployment", "")
+            world_instance.rollout_restart(deployment)
+            info["restarted"] = deployment
+            
+        elif action_type == "drain_node":
+            node_name = action.get("node_name", "")
+            world_instance.apply_patch("node", node_name, {"status": "SchedulingDisabled"})
+            info["drained"] = node_name
+            
+        elif action_type == "set_hpa":
+            deployment = action.get("deployment", "")
+            min_replicas = action.get("min_replicas", 1)
+            max_replicas = action.get("max_replicas", 10)
+            cpu_target = action.get("cpu_target_percent", 80)
+            hpa_name = f"{deployment}-hpa"
+            world_instance.apply_patch("hpa", hpa_name, {
+                "min_replicas": min_replicas,
+                "max_replicas": max_replicas,
+                "cpu_target_percent": cpu_target
+            })
+            info["hpa_set"] = deployment
+            
+        elif action_type == "describe":
+            # Investigation action - no state change
+            resource_type = action.get("resource_type", "")
+            name = action.get("name", "")
+            info["described"] = f"{resource_type}/{name}"
+            
+        else:
+            info["error"] = f"Unknown action type: {action_type}"
+            reward = -0.1  # Penalty for invalid action
+            
+    except Exception as e:
+        info["error"] = str(e)
+        reward = -0.1
+    
+    # Always tick after an action
+    world_instance.tick()
+    
+    # Calculate reward based on current state
+    reward = calculate_reward(world_instance, current_task)
+    
+    # Check if task is done
+    max_steps = config.get("tasks", {}).get(current_task, {}).get("max_steps", 15)
+    if world_instance.step_count >= max_steps:
+        done = True
+    
+    # Check if all pods are running (simplified done check)
+    if check_task_complete(world_instance, current_task):
+        done = True
+    
+    observation = world_instance.get_observation(current_objective)
+    
+    reward_signal = RewardSignal(reward=reward, done=done, info=info)
+    
+    return {
+        "observation": observation.model_dump(),
+        "reward": reward_signal.reward,
+        "done": reward_signal.done,
+        "info": reward_signal.info
+    }
 
-if __name__ == "__main__":
-    import argparse
 
-    parser = argparse.ArgumentParser()
-    parser.add_argument("--port", type=int, default=8000)
-    args = parser.parse_args()
-    main(port=args.port)
+def calculate_reward(world: World, task_id: str) -> float:
+    """Calculate reward based on current state"""
+    if task_id == "pod_recovery":
+        pods = world.get_pods()
+        frontend_pods = [p for p in pods if p.deployment == "frontend"]
+        running = [p for p in frontend_pods if p.status == "Running"]
+        if frontend_pods:
+            return len(running) / len(frontend_pods)
+    elif task_id == "autoscaling":
+        pods = world.get_pods()
+        backend_pods = [p for p in pods if p.deployment == "backend"]
+        running = [p for p in backend_pods if p.status == "Running"]
+        if backend_pods:
+            return min(len(running) / len(backend_pods), 1.0)
+    elif task_id == "incident":
+        pods = world.get_pods()
+        key_services = ["auth-service", "api-gateway", "frontend"]
+        healthy_count = 0
+        for svc in key_services:
+            svc_pods = [p for p in pods if p.deployment == svc]
+            running = [p for p in svc_pods if p.status == "Running"]
+            if svc_pods and len(running) >= len(svc_pods) * 0.8:
+                healthy_count += 1
+        return healthy_count / len(key_services) if key_services else 0.0
+    
+    return 0.0
+
+
+def check_task_complete(world: World, task_id: str) -> bool:
+    """Check if task is complete"""
+    if task_id == "pod_recovery":
+        pods = world.get_pods()
+        frontend_pods = [p for p in pods if p.deployment == "frontend"]
+        running = [p for p in frontend_pods if p.status == "Running"]
+        return len(frontend_pods) > 0 and len(running) == len(frontend_pods)
+    elif task_id == "autoscaling":
+        pods = world.get_pods()
+        backend_pods = [p for p in pods if p.deployment == "backend"]
+        running = [p for p in backend_pods if p.status == "Running"]
+        return len(backend_pods) >= 2 and len(running) >= 2
+    elif task_id == "incident":
+        pods = world.get_pods()
+        key_services = ["auth-service", "api-gateway", "frontend"]
+        for svc in key_services:
+            svc_pods = [p for p in pods if p.deployment == svc]
+            running = [p for p in svc_pods if p.status == "Running"]
+            if svc_pods and len(running) < len(svc_pods) * 0.8:
+                return False
+        return True
+    return False
+
+
+@app.get("/state")
+async def get_state():
+    """Return full current simulator state"""
+    global world_instance, current_objective
+    
+    if world_instance is None:
+        raise HTTPException(status_code=500, detail="World not initialized")
+    
+    return world_instance.get_observation(current_objective).model_dump()
+
+
+if __name__ == "__main__":
+    uvicorn.run(app, host="0.0.0.0", port=8000)

server/conditions/__init__.py

@@ -0,0 +1,15 @@
+"""COEnv Conditions - Failure injectors"""
+
+__all__ = ["crash_loop", "oom_kill", "node_failure", "cascade_failure"]
+
+from .crash_loop import CrashLoopCondition
+from .oom_kill import OOMKillCondition
+from .node_failure import NodeFailureCondition
+from .cascade_failure import CascadeFailureCondition
+
+__all__ += [
+    "CrashLoopCondition",
+    "OOMKillCondition", 
+    "NodeFailureCondition",
+    "CascadeFailureCondition"
+]

server/conditions/cascade_failure.py

@@ -0,0 +1,85 @@
+"""
+CascadeFailureCondition - Simulates multi-service dependency failure
+"""
+
+from typing import Dict, List, Any, Optional
+from ..COEnv_environment import World
+import random
+
+
+class CascadeFailureCondition:
+    """Injects cascading failures across multiple services"""
+    
+    def __init__(self, world: World, config: Dict[str, Any]):
+        self.world = world
+        self.config = config
+        
+    def inject(self, root_cause_service: Optional[str] = None, failure_probability: Optional[float] = None):
+        """
+        Inject cascading failures starting from a root cause service
+        
+        Args:
+            root_cause_service: Specific service to start failure (None for random)
+            failure_probability: Probability of failure propagating to dependencies (0.0-1.0)
+        """
+        if failure_probability is None:
+            failure_probability = self.config.get("cascade_failure_probability", 0.7)
+        else:
+            failure_probability = float(failure_probability)
+            
+        if root_cause_service is None:
+            critical_services = ["auth-service", "database", "api-gateway"]
+            deployments = self.world.get_deployments()
+            critical_deployments = [d for d in deployments if d.name in critical_services]
+            if critical_deployments:
+                root_cause_service = random.choice(critical_deployments).name
+            else:
+                deployments = self.world.get_deployments()
+                root_cause_service = random.choice(deployments).name if deployments else "frontend"
+        
+        root_deployment = next((d for d in self.world.get_deployments() if d.name == root_cause_service), None)
+        if root_deployment:
+            from ..oom_kill import OOMKillCondition
+            oom_condition = OOMKillCondition(self.world, self.config)
+            oom_condition.inject(target_deployment=root_cause_service, failure_rate=0.8)
+            
+            self._add_cascade_event(f"Root cause failure in {root_cause_service}", "Warning")
+        
+        deployments = self.world.get_deployments()
+        for deployment in deployments:
+            if deployment.name != root_cause_service and failure_probability is not None and random.random() < failure_probability:
+                failure_type = random.choice(["crashloop", "oom", "slow"])
+                
+                if failure_type == "crashloop":
+                    from ..crash_loop import CrashLoopCondition
+                    condition = CrashLoopCondition(self.world, self.config)
+                    condition.inject(target_deployment=deployment.name, failure_rate=0.6)
+                elif failure_type == "oom":
+                    from ..oom_kill import OOMKillCondition
+                    condition = OOMKillCondition(self.world, self.config)
+                    condition.inject(target_deployment=deployment.name, failure_rate=0.6)
+                else:
+                    pods = [p for p in self.world.get_pods() if p.deployment == deployment.name]
+                    for pod in pods[:1]:
+                        patch = {
+                            "cpu_request": int(pod.cpu_request * 1.5) if pod.cpu_request else 750,
+                            "mem_request": int(pod.mem_request * 1.5) if pod.mem_request else 384
+                        }
+                        self.world.apply_patch("pod", pod.name, patch)
+                
+                self._add_cascade_event(f"Cascading failure detected in {deployment.name}", "Warning")
+    
+    def _add_cascade_event(self, message: str, event_type: str):
+        """Add a cascade failure event"""
+        from ..models import ClusterEvent
+        from datetime import datetime
+        
+        event = ClusterEvent(
+            event_id=f"event-cascade-{random.randint(1000, 9999)}",
+            timestamp=datetime.now().isoformat(),
+            type=event_type,
+            reason="CascadeFailure",
+            message=message,
+            involved_object="cluster"
+        )
+        self.world.events.append(event)

server/conditions/crash_loop.py

@@ -0,0 +1,62 @@
+"""
+CrashLoopCondition - Simulates pods stuck in CrashLoopBackOff
+"""
+
+from typing import Dict, List, Any, Optional
+from ..COEnv_environment import World
+import random
+
+
+class CrashLoopCondition:
+    """Injects CrashLoopBackOff failures into pods"""
+    
+    def __init__(self, world: World, config: Dict[str, Any]):
+        self.world = world
+        self.config = config
+        
+    def inject(self, target_deployment: Optional[str] = None, failure_rate: Optional[float] = None):
+        """
+        Inject crash loop failures into pods
+        
+        Args:
+            target_deployment: Specific deployment to target (None for random)
+            failure_rate: Probability of each pod failing (0.0-1.0)
+        """
+        if failure_rate is None:
+            failure_rate = self.config.get("crash_loop_failure_rate", 0.8)
+        else:
+            failure_rate = float(failure_rate)
+            
+        deployments = self.world.get_deployments()
+        
+        if target_deployment is not None:
+            target_deps = [d for d in deployments if d.name == target_deployment]
+        else:
+            target_deps = [random.choice(deployments)] if deployments else []
+            
+        for deployment in target_deps:
+            pods = [p for p in self.world.get_pods() if p.deployment == deployment.name]
+            
+            for pod in pods:
+                if failure_rate is not None and random.random() < failure_rate:
+                    patch = {
+                        "status": "CrashLoopBackOff",
+                        "restarts": random.randint(5, 20)
+                    }
+                    self.world.apply_patch("pod", pod.name, patch)
+                    self._add_crashloop_event(pod.name)
+    
+    def _add_crashloop_event(self, pod_name: str):
+        """Add a crashloop event"""
+        from .models import ClusterEvent
+        from datetime import datetime
+        
+        event = ClusterEvent(
+            event_id=f"event-crashloop-{random.randint(1000, 9999)}",
+            timestamp=datetime.now().isoformat(),
+            type="Warning",
+            reason="BackOff",
+            message=f"Back-off restarting failed container pod/{pod_name}",
+            involved_object=pod_name
+        )
+        self.world.events.append(event)

server/conditions/node_failure.py

@@ -0,0 +1,68 @@
+"""
+NodeFailureCondition - Simulates node outages and scheduling disruption
+"""
+
+from typing import Dict, List, Any, Optional
+from ..COEnv_environment import World
+import random
+
+
+class NodeFailureCondition:
+    """Injects node failures into the cluster"""
+    
+    def __init__(self, world: World, config: Dict[str, Any]):
+        self.world = world
+        self.config = config
+        
+    def inject(self, target_node: Optional[str] = None, failure_rate: Optional[float] = None):
+        """
+        Inject node failures
+        
+        Args:
+            target_node: Specific node to target (None for random)
+            failure_rate: Probability of node failing (0.0-1.0)
+        """
+        if failure_rate is None:
+            failure_rate = self.config.get("node_failure_rate", 0.3)
+        else:
+            failure_rate = float(failure_rate)
+            
+        nodes = self.world.get_nodes()
+        
+        if target_node:
+            target_nodes = [n for n in nodes if n.name == target_node]
+        else:
+            target_nodes = [n for n in nodes if failure_rate is not None and random.random() < failure_rate]
+            
+        for node in target_nodes:
+            patch = {
+                "status": "NotReady",
+                "cpu_usage": 0.0,
+                "mem_usage": 0.0
+            }
+            self.world.apply_patch("node", node.name, patch)
+            
+            pods_on_node = [p for p in self.world.get_pods() if p.node == node.name]
+            for pod in pods_on_node:
+                patch = {
+                    "node": None,
+                    "status": "Pending"
+                }
+                self.world.apply_patch("pod", pod.name, patch)
+                
+            self._add_node_failure_event(node.name)
+    
+    def _add_node_failure_event(self, node_name: str):
+        """Add a node failure event"""
+        from models import ClusterEvent
+        from datetime import datetime
+        
+        event = ClusterEvent(
+            event_id=f"event-nodefail-{random.randint(1000, 9999)}",
+            timestamp=datetime.now().isoformat(),
+            type="Warning",
+            reason="NodeNotReady",
+            message=f"Node {node_name} status is now: NodeNotReady",
+            involved_object=node_name
+        )
+        self.world.events.append(event)

server/conditions/oom_kill.py

@@ -0,0 +1,74 @@
+"""
+OOMKillCondition - Simulates memory-limit failures causing repeated restarts
+"""
+
+from typing import Dict, List, Any, Optional
+from ..COEnv_environment import World
+import random
+
+
+class OOMKillCondition:
+    """Injects OOMKill failures into pods"""
+    
+    def __init__(self, world: World, config: Dict[str, Any]):
+        self.world = world
+        self.config = config
+        
+    def inject(self, target_deployment: Optional[str] = None, failure_rate: Optional[float] = None):
+        """
+        Inject OOMKill failures into pods
+        
+        Args:
+            target_deployment: Specific deployment to target (None for random)
+            failure_rate: Probability of each pod failing (0.0-1.0)
+        """
+        if failure_rate is None:
+            failure_rate = self.config.get("oom_kill_failure_rate", 0.6)
+        else:
+            # Ensure failure_rate is a float
+            failure_rate = float(failure_rate)
+            
+        deployments = self.world.get_deployments()
+        
+        if target_deployment is not None:
+            target_deps = [d for d in deployments if d.name == target_deployment]
+        else:
+            # Target a random deployment
+            target_deps = [random.choice(deployments)] if deployments else []
+            
+        for deployment in target_deps:
+            # Get pods for this deployment
+            pods = [p for p in self.world.get_pods() if p.deployment == deployment.name]
+            
+            for pod in pods:
+                if failure_rate is not None and random.random() < failure_rate:
+                    # Simulate OOMKill by setting high memory usage and restart count
+                    patch = {
+                        "status": "Running",  # OOMKill pods often show as Running but crash
+                        "restarts": random.randint(10, 30)  # High restart count from OOM
+                    }
+                    self.world.apply_patch("pod", pod.name, patch)
+                    
+                    # Also reduce the pod's memory limit to simulate the condition that caused OOM
+                    mem_patch = {
+                        "mem_limit": max(64, pod.mem_limit // 2) if pod.mem_limit else 128
+                    }
+                    self.world.apply_patch("pod", pod.name, mem_patch)
+                    
+                    # Add event
+                    self._add_oom_event(pod.name)
+    
+    def _add_oom_event(self, pod_name: str):
+        """Add an OOMKill event"""
+        from .models import ClusterEvent
+        from datetime import datetime
+        
+        event = ClusterEvent(
+            event_id=f"event-oom-{random.randint(1000, 9999)}",
+            timestamp=datetime.now().isoformat(),
+            type="Warning",
+            reason="OOMKilling",
+            message=f"Container {pod_name} exceeded memory limit",
+            involved_object=pod_name
+        )
+        self.world.events.append(event)

server/executor.py

@@ -0,0 +1,144 @@
+"""COEnv Executor - Action execution bridge"""
+
+from typing import Dict, Any, Optional
+
+
+class ExecutionResult:
+    """Result of action execution"""
+    
+    def __init__(self, observation, reward: float, done: bool, info: Dict[str, Any]):
+        self.observation = observation
+        self.reward = reward
+        self.done = done
+        self.info = info
+
+
+class Executor:
+    """Maps validated actions to world method calls"""
+    
+    def __init__(self, world, grader):
+        self.world = world
+        self.grader = grader
+    
+    def execute(self, action: Dict[str, Any], task_id: str, max_steps: int) -> ExecutionResult:
+        """
+        Execute an action and return the result
+        
+        Args:
+            action: The action to execute
+            task_id: Current task ID for grading
+            max_steps: Maximum steps for the episode
+            
+        Returns:
+            ExecutionResult with observation, reward, done, and info
+        """
+        action_type = action.get("action_type", "")
+        info = {}
+        reward = 0.0
+        done = False
+        
+        try:
+            if action_type == "scale":
+                deployment = action.get("deployment", "")
+                replicas = action.get("replicas", 1)
+                self.world.scale(deployment, replicas)
+                info["scaled"] = deployment
+                info["replicas"] = replicas
+                
+            elif action_type == "delete_pod":
+                pod_name = action.get("pod_name", "")
+                self.world.delete_pod(pod_name)
+                info["deleted"] = pod_name
+                
+            elif action_type == "patch":
+                resource_type = action.get("resource_type", "")
+                name = action.get("name", "")
+                patch = action.get("patch", {})
+                self.world.apply_patch(resource_type, name, patch)
+                info["patched"] = f"{resource_type}/{name}"
+                
+            elif action_type == "rollout_restart":
+                deployment = action.get("deployment", "")
+                self.world.rollout_restart(deployment)
+                info["restarted"] = deployment
+                
+            elif action_type == "drain_node":
+                node_name = action.get("node_name", "")
+                self.world.apply_patch("node", node_name, {"status": "SchedulingDisabled"})
+                info["drained"] = node_name
+                
+            elif action_type == "set_hpa":
+                deployment = action.get("deployment", "")
+                min_replicas = action.get("min_replicas", 1)
+                max_replicas = action.get("max_replicas", 10)
+                cpu_target = action.get("cpu_target_percent", 80)
+                hpa_name = f"{deployment}-hpa"
+                self.world.apply_patch("hpa", hpa_name, {
+                    "min_replicas": min_replicas,
+                    "max_replicas": max_replicas,
+                    "cpu_target_percent": cpu_target
+                })
+                info["hpa_set"] = deployment
+                
+            elif action_type == "describe":
+                # Investigation action - no state change
+                resource_type = action.get("resource_type", "")
+                name = action.get("name", "")
+                info["described"] = f"{resource_type}/{name}"
+                
+            else:
+                info["error"] = f"Unknown action type: {action_type}"
+                reward = -0.1
+                
+        except Exception as e:
+            info["error"] = str(e)
+            reward = -0.1
+        
+        # Always advance time after an action
+        self.world.tick()
+        
+        # Calculate reward
+        world_state = self.world.get_full_state()
+        reward = self.grader.grade(world_state, self.world.step_count, max_steps)
+        
+        # Check if done
+        if self.world.step_count >= max_steps:
+            done = True
+        
+        # Check task completion
+        if self._check_task_complete(task_id):
+            done = True
+        
+        observation = self.world.get_observation()
+        
+        return ExecutionResult(
+            observation=observation,
+            reward=reward,
+            done=done,
+            info=info
+        )
+    
+    def _check_task_complete(self, task_id: str) -> bool:
+        """Check if task is complete"""
+        pods = self.world.get_pods()
+        
+        if task_id == "pod_recovery":
+            frontend_pods = [p for p in pods if p.deployment == "frontend"]
+            running = [p for p in frontend_pods if p.status == "Running"]
+            return len(frontend_pods) > 0 and len(running) == len(frontend_pods)
+        
+        elif task_id == "autoscaling":
+            backend_pods = [p for p in pods if p.deployment == "backend"]
+            running = [p for p in backend_pods if p.status == "Running"]
+            return len(backend_pods) >= 2 and len(running) >= 2
+        
+        elif task_id == "incident":
+            key_services = ["auth-service", "api-gateway", "frontend"]
+            for svc in key_services:
+                svc_pods = [p for p in pods if p.deployment == svc]
+                running = [p for p in svc_pods if p.status == "Running"]
+                if svc_pods and len(running) < len(svc_pods) * 0.8:
+                    return False
+            return True
+        
+        return False

server/graders/__init__.py

@@ -0,0 +1,9 @@
+"""COEnv Graders"""
+
+__all__ = ["grader_pod_recovery", "grader_autoscaling", "grader_incident"]
+
+from .grader_pod_recovery import grade as pod_recovery_grade
+from .grader_autoscaling import grade as autoscaling_grade
+from .grader_incident import grade as incident_grade
+
+__all__ += ["pod_recovery_grade", "autoscaling_grade", "incident_grade"]

server/graders/grader_autoscaling.py

@@ -0,0 +1,40 @@
+"""
+Grader for Autoscaling Task
+"""
+
+from typing import Dict, Any
+
+
+def grade(world_state: Dict[str, Any], step: int, max_steps: int) -> float:
+    """Grade the autoscaling task"""
+    score = 0.0
+    
+    # Check backend deployment status
+    backend_deployment = next((d for d in world_state.get("deployments", []) if d.get("name") == "backend"), None)
+    if not backend_deployment:
+        return 0.0
+    
+    # Check if we have adequate replicas
+    desired = backend_deployment.get("desired_replicas", 0)
+    available = backend_deployment.get("available_replicas", 0)
+    
+    if desired > 0:
+        replica_ratio = min(available / desired, 1.0)
+        score += replica_ratio * 0.4  # 40% for proper scaling
+    
+    # Check backend pod health
+    backend_pods = [p for p in world_state.get("pods", []) 
+                   if p.get("deployment") == "backend" and p.get("status") == "Running"]
+    total_backend_pods = [p for p in world_state.get("pods", []) 
+                         if p.get("deployment") == "backend"]
+    
+    if total_backend_pods:
+        health_ratio = len(backend_pods) / len(total_backend_pods)
+        score += health_ratio * 0.4  # 40% for pod health
+    
+    # Efficiency bonus
+    if max_steps > 0:
+        efficiency = 1.0 - (step / max_steps)
+        score += efficiency * 0.2  # 20% for efficiency
+    
+    return min(score, 1.0)

server/graders/grader_incident.py

@@ -0,0 +1,50 @@
+"""
+Grader for Incident Task
+"""
+
+from typing import Dict, Any
+
+
+def grade(world_state: Dict[str, Any], step: int, max_steps: int) -> float:
+    """Grade the incident task"""
+    score = 0.0
+    
+    # Key services that should be healthy after incident resolution
+    key_services = ["auth-service", "api-gateway", "frontend"]
+    healthy_services = 0
+    
+    for service_name in key_services:
+        # Check if deployment exists and has running pods
+        deployment = next((d for d in world_state.get("deployments", []) if d.get("name") == service_name), None)
+        if deployment:
+            desired = deployment.get("desired_replicas", 0)
+            available = deployment.get("available_replicas", 0)
+            
+            if desired > 0:
+                # Consider service healthy if at least 80% of desired replicas are available
+                if available / desired >= 0.8:
+                    healthy_services += 1
+    
+    # Score based on proportion of healthy services
+    if key_services:
+        service_health_score = healthy_services / len(key_services)
+        score += service_health_score * 0.6  # 60% for service health
+    
+    # Check for absence of crashlooping pods in key services
+    key_service_pods = [p for p in world_state.get("pods", []) 
+                       if p.get("deployment") in key_services]
+    crashloop_pods = [p for p in key_service_pods 
+                     if p.get("status") == "CrashLoopBackOff"]
+    
+    if key_service_pods:
+        crashloop_ratio = len(crashloop_pods) / len(key_service_pods)
+        # Penalize for crashlooping pods (inverse relationship)
+        health_bonus = (1.0 - crashloop_ratio) * 0.3  # 30% for no crashloops
+        score += health_bonus
+    
+    # Efficiency bonus
+    if max_steps > 0:
+        efficiency = 1.0 - (step / max_steps)
+        score += efficiency * 0.1  # 10% for efficiency
+    
+    return min(score, 1.0)

server/graders/grader_pod_recovery.py

@@ -0,0 +1,31 @@
+"""
+Grader for Pod Recovery Task
+"""
+
+from typing import Dict, Any
+
+
+def grade(world_state: Dict[str, Any], step: int, max_steps: int) -> float:
+    """Grade the pod recovery task"""
+    score = 0.0
+    
+    # Count running frontend pods
+    frontend_pods = [p for p in world_state.get("pods", []) 
+                    if p.get("deployment") == "frontend" and p.get("status") == "Running"]
+    total_frontend_pods = [p for p in world_state.get("pods", []) 
+                          if p.get("deployment") == "frontend"]
+    
+    if total_frontend_pods:
+        running_ratio = len(frontend_pods) / len(total_frontend_pods)
+        score += running_ratio * 0.5
+    
+    # Bonus for all pods running
+    if total_frontend_pods and len(frontend_pods) == len(total_frontend_pods):
+        score += 0.4
+    
+    # Efficiency bonus (faster is better)
+    if max_steps > 0:
+        efficiency = 1.0 - (step / max_steps)
+        score += efficiency * 0.1
+    
+    return min(score, 1.0)

server/models.py

@@ -0,0 +1,151 @@
+"""
+KubeSimEnv Models - Pydantic models for OpenEnv compliance
+All typed models are mandatory for OpenEnv spec compliance.
+Every endpoint uses these.
+"""
+
+from pydantic import BaseModel, Field
+from typing import List, Dict, Any, Optional, Literal
+from datetime import datetime
+
+
+class NodeStatus(BaseModel):
+    """Status of a Kubernetes node"""
+    name: str
+    status: Literal["Ready", "NotReady", "Unknown", "SchedulingDisabled"]
+    cpu_capacity: int  # in cores
+    mem_capacity: int  # in MB
+    cpu_usage: float = Field(ge=0, le=100)  # percentage
+    mem_usage: float = Field(ge=0, le=100)  # percentage
+    last_updated: str  # ISO timestamp
+
+
+class PodStatus(BaseModel):
+    """Status of a Kubernetes pod"""
+    name: str
+    status: Literal["Pending", "Running", "Succeeded", "Failed", "Unknown", "CrashLoopBackOff"]
+    node: Optional[str] = None
+    restarts: int = 0
+    cpu_request: int = Field(default=0)  # in millicores
+    mem_request: int = Field(default=0)  # in MB
+    cpu_limit: Optional[int] = Field(default=None)  # in millicores
+    mem_limit: Optional[int] = Field(default=None)  # in MB
+    deployment: Optional[str] = None
+    last_updated: str  # ISO timestamp
+
+
+class DeploymentStatus(BaseModel):
+    """Status of a Kubernetes deployment"""
+    name: str
+    desired_replicas: int
+    available_replicas: int
+    image: str
+    last_updated: str  # ISO timestamp
+
+
+class ServiceStatus(BaseModel):
+    """Status of a Kubernetes service"""
+    name: str
+    type: Literal["ClusterIP", "NodePort", "LoadBalancer", "ExternalName"]
+    ports: List[Dict[str, Any]]
+    selector: Optional[Dict[str, str]] = None
+    cluster_ip: Optional[str] = None
+    last_updated: str  # ISO timestamp
+
+
+class ConfigMapStatus(BaseModel):
+    """Status of a Kubernetes ConfigMap"""
+    name: str
+    data: Dict[str, str]
+    last_updated: str  # ISO timestamp
+
+
+class HPAStatus(BaseModel):
+    """Status of a HorizontalPodAutoscaler"""
+    name: str
+    min_replicas: int
+    max_replicas: int
+    current_replicas: int
+    cpu_target_percent: int
+    last_updated: str  # ISO timestamp
+
+
+class ClusterEvent(BaseModel):
+    """Kubernetes-style event"""
+    event_id: str
+    timestamp: str  # ISO timestamp
+    type: Literal["Normal", "Warning"]
+    reason: str
+    message: str
+    involved_object: str
+
+
+class ClusterObservation(BaseModel):
+    """Main observation model - typed cluster snapshot"""
+    nodes: List[NodeStatus]
+    pods: List[PodStatus]
+    deployments: List[DeploymentStatus]
+    services: List[ServiceStatus]
+    configmaps: List[ConfigMapStatus]
+    hpas: List[HPAStatus]
+    events: List[ClusterEvent]
+    step: int
+    objective: str
+
+
+class RewardSignal(BaseModel):
+    """Reward signal returned by step()"""
+    reward: float
+    done: bool
+    info: Dict[str, Any] = Field(default_factory=dict)
+
+
+# Action Models - These represent the structured action space
+class KubeAction(BaseModel):
+    """Base action model"""
+    action_type: Literal[
+        "scale", "delete_pod", "patch", "rollout_restart", 
+        "set_hpa", "drain_node", "describe"
+    ]
+
+
+class ScaleAction(KubeAction):
+    """Scale a deployment to a specific replica count"""
+    deployment: str
+    replicas: int
+
+
+class DeletePodAction(KubeAction):
+    """Delete a specific pod"""
+    pod_name: str
+
+
+class PatchAction(KubeAction):
+    """Patch a resource with specific changes"""
+    resource_type: Literal["deployment", "pod", "node", "service"]
+    name: str
+    patch: Dict[str, Any]
+
+
+class RolloutRestartAction(KubeAction):
+    """Restart a deployment rollout"""
+    deployment: str
+
+
+class SetHPAAction(KubeAction):
+    """Set HorizontalPodAutoscaler for a deployment"""
+    deployment: str
+    min_replicas: int
+    max_replicas: int
+    cpu_target_percent: int
+
+
+class DrainNodeAction(KubeAction):
+    """Drain a node (evict all pods)"""
+    node_name: str
+
+
+class DescribeAction(KubeAction):
+    """Describe/get details of a resource"""
+    resource_type: Literal["deployment", "pod", "node", "service"]
+    name: str

server/requirements.txt

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

server/tasks/__init__.py

@@ -0,0 +1,9 @@
+"""COEnv Tasks"""
+
+__all__ = ["task_pod_recovery", "task_autoscaling", "task_incident"]
+
+from .task_pod_recovery import PodRecoveryTask
+from .task_autoscaling import AutoscalingTask
+from .task_incident import IncidentTask
+
+__all__ += ["PodRecoveryTask", "AutoscalingTask", "IncidentTask"]

server/tasks/task_autoscaling.py

@@ -0,0 +1,47 @@
+"""
+Autoscaling Task - Medium difficulty
+Configure HPA to handle traffic spike
+"""
+
+from typing import Dict, Any
+from ..COEnv_environment import World
+
+
+class AutoscalingTask:
+    """Autoscaling task implementation"""
+    
+    def __init__(self, world: World, config: Dict[str, Any]):
+        self.world = world
+        self.config = config
+        self.task_id = "autoscaling"
+        self.description = "Configure HPA to handle traffic spike"
+        
+    def reset(self):
+        """Reset the task to initial state"""
+        self.world.reset_to_healthy()
+        
+        backend_pods = [p for p in self.world.get_pods() if p.deployment == "backend"]
+        for pod in backend_pods:
+            patch = {
+                "cpu_request": int(pod.cpu_request * 3) if pod.cpu_request else 750,
+                "mem_request": int(pod.mem_request * 3) if pod.mem_request else 384
+            }
+            self.world.apply_patch("pod", pod.name, patch)
+        
+        self.objective = "Backend service is overloaded due to traffic spike. Configure HPA to automatically scale the backend deployment based on CPU utilization."
+        
+    def is_complete(self) -> bool:
+        """Check if the task is complete"""
+        backend_deployment = next((d for d in self.world.get_deployments() if d.name == "backend"), None)
+        if not backend_deployment:
+            return False
+            
+        backend_pods = [p for p in self.world.get_pods() 
+                       if p.deployment == "backend" and p.status == "Running"]
+        return len(backend_pods) >= 2
+    
+    def get_observation(self) -> Dict[str, Any]:
+        """Get current observation for the task"""
+        observation = self.world.get_full_state()
+        observation["objective"] = self.objective
+        return observation

server/tasks/task_incident.py

@@ -0,0 +1,48 @@
+"""
+Incident Task - Hard difficulty
+Handle multi-service cascading incident
+"""
+
+from typing import Dict, Any
+from ..COEnv_environment import World
+from ..conditions.cascade_failure import CascadeFailureCondition
+
+
+class IncidentTask:
+    """Incident task implementation"""
+    
+    def __init__(self, world: World, config: Dict[str, Any]):
+        self.world = world
+        self.config = config
+        self.task_id = "incident"
+        self.description = "Handle multi-service cascading incident"
+        
+    def reset(self):
+        """Reset the task to initial state"""
+        self.world.reset_to_healthy()
+        
+        cascade_condition = CascadeFailureCondition(self.world, self.config)
+        cascade_condition.inject(root_cause_service="auth-service", failure_probability=0.6)
+        
+        self.objective = "Auth-service OOMKill has caused cascading failures. Identify the root cause, fix memory limits, restart workloads, and verify downstream recovery."
+        
+    def is_complete(self) -> bool:
+        """Check if the task is complete"""
+        key_services = ["auth-service", "api-gateway", "frontend"]
+        healthy_services = 0
+        
+        for service_name in key_services:
+            deployment = next((d for d in self.world.get_deployments() if d.name == service_name), None)
+            if deployment:
+                running_pods = [p for p in self.world.get_pods() 
+                              if p.deployment == service_name and p.status == "Running"]
+                if len(running_pods) >= deployment.desired_replicas * 0.8:
+                    healthy_services += 1
+        
+        return healthy_services >= len(key_services) * 0.67
+    
+    def get_observation(self) -> Dict[str, Any]:
+        """Get current observation for the task"""
+        observation = self.world.get_full_state()
+        observation["objective"] = self.objective
+        return observation

server/tasks/task_pod_recovery.py

@@ -0,0 +1,45 @@
+"""
+Pod Recovery Task - Easy difficulty
+Fix crash-looping pods by identifying and patching bad configuration
+"""
+
+from typing import Dict, Any
+from ..COEnv_environment import World
+from ..conditions.crash_loop import CrashLoopCondition
+
+
+class PodRecoveryTask:
+    """Pod recovery task implementation"""
+    
+    def __init__(self, world: World, config: Dict[str, Any]):
+        self.world = world
+        self.config = config
+        self.task_id = "pod_recovery"
+        self.description = "Fix crash-looping pods by identifying and patching bad configuration"
+        
+    def reset(self):
+        """Reset the task to initial state"""
+        self.world.reset_to_healthy()
+        
+        crash_loop_condition = CrashLoopCondition(self.world, self.config)
+        crash_loop_condition.inject(target_deployment="frontend", failure_rate=0.8)
+        
+        self.objective = "All frontend pods should be running. Investigate the CrashLoopBackOff pods and fix the configuration issue."
+        
+    def is_complete(self) -> bool:
+        """Check if the task is complete"""
+        frontend_pods = [p for p in self.world.get_pods() 
+                        if p.deployment == "frontend" and p.status == "Running"]
+        total_frontend_pods = [p for p in self.world.get_pods() 
+                              if p.deployment == "frontend"]
+        
+        if not total_frontend_pods:
+            return False
+            
+        return len(frontend_pods) == len(total_frontend_pods)
+    
+    def get_observation(self) -> Dict[str, Any]:
+        """Get current observation for the task"""
+        observation = self.world.get_full_state()
+        observation["objective"] = self.objective
+        return observation

server/utils.py

@@ -0,0 +1,190 @@
+"""
+KubeSimEnv Utils - Probability helpers and simulation utilities
+Random failure rate generators, latency simulators, resource usage curves.
+Makes the simulation feel realistic and non-deterministic in the right ways.
+"""
+
+import random
+import math
+from typing import Dict, List, Any, Optional
+from datetime import datetime, timedelta
+
+
+class ProbabilityHelpers:
+    """Helpers for generating realistic probabilities and distributions"""
+    
+    @staticmethod
+    def weighted_random_choice(choices: List[Any], weights: List[float]) -> Any:
+        """Make a weighted random choice"""
+        if not choices or not weights or len(choices) != len(weights):
+            return random.choice(choices) if choices else None
+        
+        # Normalize weights
+        total_weight = sum(weights)
+        if total_weight == 0:
+            return random.choice(choices)
+        
+        normalized_weights = [w / total_weight for w in weights]
+        
+        # Make choice
+        r = random.random()
+        cumulative_weight = 0
+        for choice, weight in zip(choices, normalized_weights):
+            cumulative_weight += weight
+            if r <= cumulative_weight:
+                return choice
+        return choices[-1]  # Fallback
+    
+    @staticmethod
+    def exponential_backoff(attempt: int, base_delay: float = 1.0, max_delay: float = 60.0) -> float:
+        """Calculate exponential backoff delay"""
+        delay = base_delay * (2 ** attempt)
+        return min(delay, max_delay)
+    
+    @staticmethod
+    def poisson_arrival_rate(lambda_rate: float, time_window: float) -> int:
+        """Generate number of events in time window using Poisson distribution"""
+        # Simple approximation - in reality would use numpy.random.poisson
+        return int(lambda_rate * time_window + random.gauss(0, math.sqrt(lambda_rate * time_window)))
+    
+    @staticmethod
+    def failure_probability_over_time(base_rate: float, time_elapsed: float, 
+                                    max_rate: float = 1.0) -> float:
+        """Calculate failure probability that increases over time"""
+        probability = base_rate * (1 + math.log(1 + time_elapsed))
+        return min(probability, max_rate)
+    
+    @staticmethod
+    def random_failure_rate(min_rate: float = 0.1, max_rate: float = 0.9) -> float:
+        """Generate a random failure rate within bounds"""
+        return random.uniform(min_rate, max_rate)
+
+
+class LatencySimulator:
+    """Simulates network and service latency"""
+    
+    def __init__(self, base_latency_ms: float = 50.0):
+        self.base_latency_ms = base_latency_ms
+        self.load_factor = 1.0
+        
+    def set_load(self, load_factor: float):
+        """Set system load factor (1.0 = normal, >1.0 = overloaded)"""
+        self.load_factor = max(0.1, load_factor)
+        
+    def get_latency(self) -> float:
+        """Get simulated latency in milliseconds"""
+        # Base latency + load-dependent component + random jitter
+        load_latency = self.base_latency_ms * (self.load_factor - 1.0) * 2
+        jitter = random.gauss(0, self.base_latency_ms * 0.1)
+        latency = self.base_latency_ms + max(0, load_latency) + jitter
+        return max(1.0, latency)  # Minimum 1ms latency
+    
+    def get_latency_with_spike(self, spike_probability: float = 0.05, 
+                             spike_multiplier: float = 5.0) -> float:
+        """Get latency with occasional spikes"""
+        latency = self.get_latency()
+        if random.random() < spike_probability:
+            latency *= spike_multiplier
+        return latency
+
+
+class ResourceUsageSimulator:
+    """Simulates realistic CPU and memory usage patterns"""
+    
+    def __init__(self):
+        self.time_offset = random.uniform(0, 2 * math.pi)
+        
+    def get_cpu_usage(self, base_usage: float = 0.3, 
+                     variation: float = 0.2) -> float:
+        """Get CPU usage as percentage (0-100)"""
+        # Simulate daily patterns with some noise
+        time_factor = (datetime.now().timestamp() / 3600) % 24  # Hours in day
+        daily_pattern = 0.5 * math.sin(2 * math.pi * time_factor / 24) + 0.5
+        
+        usage = base_usage + variation * daily_pattern
+        usage += random.gauss(0, 0.05)  # Noise
+        return max(0.0, min(1.0, usage)) * 100  # Clamp to 0-100%
+    
+    def get_memory_usage(self, base_usage: float = 0.4,
+                        variation: float = 0.15) -> float:
+        """Get memory usage as percentage (0-100)"""
+        # Memory usage tends to creep up over time (simulate leak)
+        time_factor = min((datetime.now().timestamp() / 86400) % 7, 1.0)  # Weekly pattern
+        leak_factor = 0.1 * time_factor  # Slow leak over week
+        
+        usage = base_usage + leak_factor
+        usage += random.gauss(0, 0.03)  # Noise
+        return max(0.0, min(1.0, usage)) * 100  # Clamp to 0-100%
+    
+    def get_resource_curve(self, resource_type: str, 
+                          time_elapsed: float) -> float:
+        """Get resource usage following a specific curve"""
+        if resource_type == "cpu":
+            # CPU: periodic with bursts
+            return 0.3 + 0.4 * math.sin(time_elapsed / 100) + 0.2 * random.random()
+        elif resource_type == "memory":
+            # Memory: gradual increase with occasional GC drops
+            base = 0.2 + 0.6 * (1 - math.exp(-time_elapsed / 1000))
+            gc_drop = 0.3 if random.random() < 0.01 else 0  # Occasional GC
+            return max(0, base - gc_drop)
+        elif resource_type == "disk":
+            # Disk: steady growth
+            return 0.1 + 0.8 * min(time_elapsed / 10000, 1.0)
+        else:
+            return 0.5
+
+
+class NetworkSimulator:
+    """Simulates network conditions and partitions"""
+    
+    def __init__(self):
+        self.partition_probability = 0.01
+        self.latency_ms = 10.0
+        self.bandwidth_mbps = 1000.0
+        
+    def simulate_partition(self) -> bool:
+        """Return True if network partition is simulated"""
+        return random.random() < self.partition_probability
+    
+    def get_latency(self) -> float:
+        """Get network latency in milliseconds"""
+        # Base latency with occasional spikes
+        latency = self.latency_ms + random.gauss(0, self.latency_ms * 0.2)
+        if random.random() < 0.05:  # 5% chance of spike
+            latency *= random.uniform(2, 10)
+        return max(1.0, latency)
+    
+    def get_bandwidth(self) -> float:
+        """Get available bandwidth in Mbps"""
+        # Bandwidth varies with usage and conditions
+        usage_factor = random.uniform(0.3, 0.9)
+        condition_factor = random.uniform(0.8, 1.2)
+        return self.bandwidth_mbps * usage_factor * condition_factor
+
+
+def generate_failure_scenario(config: Dict[str, Any]) -> Dict[str, Any]:
+    """Generate a random failure scenario based on config"""
+    scenario = {
+        "type": random.choice(["crashloop", "oom", "node_failure", "cascade"]),
+        "severity": random.uniform(0.3, 0.9),
+        "duration": random.randint(30, 300),  # seconds
+        "affected_components": []
+    }
+    
+    # Add specific parameters based on type
+    if scenario["type"] == "crashloop":
+        scenario["failure_rate"] = config.get("crash_loop_failure_rate", 0.7)
+    elif scenario["type"] == "oom":
+        scenario["failure_rate"] = config.get("oom_kill_failure_rate", 0.6)
+    elif scenario["type"] == "node_failure":
+        scenario["failure_rate"] = config.get("node_failure_rate", 0.4)
+    elif scenario["type"] == "cascade":
+        scenario["probability"] = config.get("cascade_failure_probability", 0.5)
+    
+    return scenario
+
+
+def apply_realistic_noise(value: float, noise_percent: float = 10.0) -> float:
+    """Apply realistic noise to a value"""
+    noise = random.gauss(0, value * (noise_percent / 100.0))
+    return max(0, value + noise)

server/validator.py

@@ -0,0 +1,167 @@
+"""COEnv Validator - Action validation"""
+
+from typing import Dict, Any, Optional, Tuple
+
+
+class Validator:
+    """Validates actions before execution"""
+    
+    def __init__(self, world):
+        self.world = world
+    
+    def validate(self, action: Dict[str, Any]) -> Tuple[bool, Optional[str]]:
+        """
+        Validate an action before execution
+        
+        Returns:
+            (is_valid, error_message)
+        """
+        action_type = action.get("action_type", "")
+        
+        if action_type == "scale":
+            return self._validate_scale(action)
+        elif action_type == "delete_pod":
+            return self._validate_delete_pod(action)
+        elif action_type == "patch":
+            return self._validate_patch(action)
+        elif action_type == "rollout_restart":
+            return self._validate_rollout_restart(action)
+        elif action_type == "set_hpa":
+            return self._validate_set_hpa(action)
+        elif action_type == "drain_node":
+            return self._validate_drain_node(action)
+        elif action_type == "describe":
+            return self._validate_describe(action)
+        else:
+            return False, f"Unknown action type: {action_type}"
+    
+    def _validate_scale(self, action: Dict[str, Any]) -> Tuple[bool, Optional[str]]:
+        deployment = action.get("deployment", "")
+        replicas = action.get("replicas", 0)
+        
+        if not deployment:
+            return False, "Deployment name is required"
+        
+        if replicas < 0 or replicas > 100:
+            return False, "Replicas must be between 0 and 100"
+        
+        # Check if deployment exists
+        deployments = self.world.get_deployments()
+        if not any(d.name == deployment for d in deployments):
+            return False, f"Deployment '{deployment}' does not exist"
+        
+        return True, None
+    
+    def _validate_delete_pod(self, action: Dict[str, Any]) -> Tuple[bool, Optional[str]]:
+        pod_name = action.get("pod_name", "")
+        
+        if not pod_name:
+            return False, "Pod name is required"
+        
+        # Check if pod exists
+        pods = self.world.get_pods()
+        if not any(p.name == pod_name for p in pods):
+            return False, f"Pod '{pod_name}' does not exist"
+        
+        return True, None
+    
+    def _validate_patch(self, action: Dict[str, Any]) -> Tuple[bool, Optional[str]]:
+        resource_type = action.get("resource_type", "")
+        name = action.get("name", "")
+        
+        if not resource_type:
+            return False, "Resource type is required"
+        
+        if not name:
+            return False, "Resource name is required"
+        
+        # Check if resource exists
+        if resource_type == "deployment":
+            deployments = self.world.get_deployments()
+            if not any(d.name == name for d in deployments):
+                return False, f"Deployment '{name}' does not exist"
+        elif resource_type == "pod":
+            pods = self.world.get_pods()
+            if not any(p.name == name for p in pods):
+                return False, f"Pod '{name}' does not exist"
+        elif resource_type == "node":
+            nodes = self.world.get_nodes()
+            if not any(n.name == name for n in nodes):
+                return False, f"Node '{name}' does not exist"
+        elif resource_type == "service":
+            services = self.world.get_services()
+            if not any(s.name == name for s in services):
+                return False, f"Service '{name}' does not exist"
+        elif resource_type == "configmap":
+            configmaps = self.world.get_configmaps()
+            if not any(cm.name == name for cm in configmaps):
+                return False, f"ConfigMap '{name}' does not exist"
+        
+        return True, None
+    
+    def _validate_rollout_restart(self, action: Dict[str, Any]) -> Tuple[bool, Optional[str]]:
+        deployment = action.get("deployment", "")
+        
+        if not deployment:
+            return False, "Deployment name is required"
+        
+        # Check if deployment exists
+        deployments = self.world.get_deployments()
+        if not any(d.name == deployment for d in deployments):
+            return False, f"Deployment '{deployment}' does not exist"
+        
+        return True, None
+    
+    def _validate_set_hpa(self, action: Dict[str, Any]) -> Tuple[bool, Optional[str]]:
+        deployment = action.get("deployment", "")
+        min_replicas = action.get("min_replicas", 0)
+        max_replicas = action.get("max_replicas", 0)
+        
+        if not deployment:
+            return False, "Deployment name is required"
+        
+        if min_replicas < 1 or min_replicas > 50:
+            return False, "Min replicas must be between 1 and 50"
+        
+        if max_replicas < 1 or max_replicas > 100:
+            return False, "Max replicas must be between 1 and 100"
+        
+        if min_replicas > max_replicas:
+            return False, "Min replicas cannot be greater than max replicas"
+        
+        # Check if deployment exists
+        deployments = self.world.get_deployments()
+        if not any(d.name == deployment for d in deployments):
+            return False, f"Deployment '{deployment}' does not exist"
+        
+        return True, None
+    
+    def _validate_drain_node(self, action: Dict[str, Any]) -> Tuple[bool, Optional[str]]:
+        node_name = action.get("node_name", "")
+        
+        if not node_name:
+            return False, "Node name is required"
+        
+        # Check if node exists
+        nodes = self.world.get_nodes()
+        if not any(n.name == node_name for n in nodes):
+            return False, f"Node '{node_name}' does not exist"
+        
+        # Check if node is already drained/scheduling disabled
+        node = next((n for n in nodes if n.name == node_name), None)
+        if node and node.status == "SchedulingDisabled":
+            return False, f"Node '{node_name}' is already drained"
+        
+        return True, None
+    
+    def _validate_describe(self, action: Dict[str, Any]) -> Tuple[bool, Optional[str]]:
+        resource_type = action.get("resource_type", "")
+        name = action.get("name", "")
+        
+        if not resource_type:
+            return False, "Resource type is required"
+        
+        if not name:
+            return False, "Resource name is required"
+        
+        return True, None

server/worker.py

@@ -0,0 +1,101 @@
+"""COEnv Worker - Episode loop manager"""
+
+from typing import Dict, Any, Optional, List
+
+
+class EpisodeResult:
+    """Result of a single episode"""
+    
+    def __init__(self, rewards: List[float], steps: int, success: bool, info: Dict[str, Any]):
+        self.rewards = rewards
+        self.steps = steps
+        self.success = success
+        self.info = info
+
+
+class Worker:
+    """Manages the full lifecycle of a single episode"""
+    
+    def __init__(self, world, executor, validator):
+        self.world = world
+        self.executor = executor
+        self.validator = validator
+    
+    def run_episode(self, task_id: str, task_objective: str, max_steps: int, 
+                    get_action_fn=None) -> EpisodeResult:
+        """
+        Run a single episode
+        
+        Args:
+            task_id: Task ID to run
+            task_objective: Objective string for the task
+            max_steps: Maximum steps for the episode
+            get_action_fn: Function to get action from agent (if None, uses random)
+            
+        Returns:
+            EpisodeResult with rewards, steps, success, and info
+        """
+        import random
+        
+        # Reset world with task condition
+        observation = self.world.reset(task_objective)
+        rewards = []
+        steps = 0
+        info = {}
+        
+        for step in range(1, max_steps + 1):
+            steps = step
+            
+            # Get action from agent (or random for now)
+            if get_action_fn:
+                action = get_action_fn(observation)
+            else:
+                # Random action for testing
+                action = self._random_action()
+            
+            # Validate action
+            is_valid, error_msg = self.validator.validate(action)
+            if not is_valid:
+                info["error"] = error_msg
+                # Still execute but with penalty
+                result = self.executor.execute(action, task_id, max_steps)
+                result.reward = -0.1  # Penalty for invalid action
+            else:
+                # Execute valid action
+                result = self.executor.execute(action, task_id, max_steps)
+            
+            rewards.append(result.reward)
+            
+            if result.done:
+                info["success"] = True
+                info["final_reward"] = result.reward
+                break
+        else:
+            info["success"] = False
+            info["final_reward"] = rewards[-1] if rewards else 0.0
+        
+        return EpisodeResult(
+            rewards=rewards,
+            steps=steps,
+            success=info.get("success", False),
+            info=info
+        )
+    
+    def _random_action(self) -> Dict[str, Any]:
+        """Generate a random action for testing"""
+        import random
+        
+        action_types = [
+            {"action_type": "describe", "resource_type": "deployment", "name": "frontend"},
+            {"action_type": "describe", "resource_type": "pod", "name": "frontend-abc123"},
+            {"action_type": "scale", "deployment": "frontend", "replicas": random.randint(1, 5)},
+            {"action_type": "scale", "deployment": "backend", "replicas": random.randint(1, 5)},
+            {"action_type": "delete_pod", "pod_name": "frontend-xyz789"},
+            {"action_type": "patch", "resource_type": "deployment", "name": "frontend", 
+             "patch": {"desired_replicas": random.randint(1, 5)}},
+            {"action_type": "rollout_restart", "deployment": "frontend"},
+            {"action_type": "set_hpa", "deployment": "backend", 
+             "min_replicas": 1, "max_replicas": 10, "cpu_target_percent": 80},
+        ]
+        
+        return random.choice(action_types)

tests/test_environment.py

@@ -0,0 +1,46 @@
+"""
+Test - Environment (from test_world.py)
+"""
+
+import sys
+import os
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', 'server'))
+
+from COEnv_environment import World
+
+
+def test_world_initialization():
+    """Test that the world initializes correctly"""
+    config = {
+        "num_nodes": 2,
+        "node_cpu_capacity": 4,
+        "node_mem_capacity": 8192,
+        "pod_cpu_request": 250,
+        "pod_mem_request": 128,
+        "pod_cpu_limit": 500,
+        "pod_mem_limit": 256,
+    }
+    
+    world = World(config)
+    print("World initialized successfully")
+    
+    nodes = world.get_nodes()
+    pods = world.get_pods()
+    deployments = world.get_deployments()
+    services = world.get_services()
+    
+    print(f"Nodes: {len(nodes)}")
+    print(f"Pods: {len(pods)}")
+    print(f"Deployments: {len(deployments)}")
+    print(f"Services: {len(services)}")
+    
+    assert len(nodes) == 2
+    assert len(pods) > 0
+    assert len(deployments) > 0
+    assert len(services) > 0
+    
+    print("All tests passed!")
+
+
+if __name__ == "__main__":
+    test_world_initialization()