How to Use OpenEnv Transform to Modify Observations: A Complete Guide

OpenEnv Transform provides a composable, callable interface to modify observations before they reach the client, enabling reward shaping, safety checks, and metadata enrichment without altering core environment logic.

The OpenEnv Transform system in the huggingface/OpenEnv repository offers a clean protocol for intercepting and modifying environment observations. By implementing the Transform protocol, developers can create reusable, composable modifiers that adjust rewards, append metadata, or enforce safety constraints. This architecture keeps transformation logic separate from core environment execution, promoting maintainable and testable code.

Understanding the OpenEnv Transform Architecture

The Transform Protocol

In src/openenv/core/env_server/interfaces.py, the Transform protocol defines the core contract: any callable that accepts an Observation instance and returns an (potentially modified) Observation. This simple abstraction allows stateless or lightly stateful modifications to the data flow without requiring inheritance from a specific base class.

CompositeTransform for Pipeline Composition

The CompositeTransform class in src/openenv/core/env_server/base_transforms.py chains multiple transforms sequentially. When instantiated with a list of transforms, it applies each in order, passing the modified observation to the next step.

class CompositeTransform(Transform):
    """Combines multiple transforms into a single transform."""
    def __init__(self, transforms: list[Transform]):
        self.transforms = transforms

    def __call__(self, observation: Observation) -> Observation:
        for transform in self.transforms:
            observation = transform(observation)
        return observation

Order-sensitivity matters because subsequent transforms may depend on metadata added by earlier ones. The pipeline receives a copy-by-reference of the Pydantic observation model, so changes persist automatically through the chain.

Built-In Transform Implementations

The coding environment ships with ready-made transforms located in envs/coding_env/server/transforms.py.

CodeSafetyTransform for Security Enforcement

The CodeSafetyTransform scans submitted code for dangerous patterns such as import os or eval(. Upon detection, it penalizes the observation with a default reward of -1.0 and records the violation in metadata["safety_violation"].

CodeQualityTransform for Reward Shaping

The CodeQualityTransform evaluates code conciseness and syntax validity. It adjusts observation.reward based on code length and AST parsing results, encouraging efficient solutions while penalizing syntax errors.

These transforms are composed via the factory function create_safe_coding_transform(), which returns a CompositeTransform containing both safety and quality checks.

Wiring Transforms into the Environment

The PythonCodeActEnv class in envs/coding_env/server/python_codeact_env.py demonstrates practical integration. During initialization, it loads the composite transform:

from .transforms import create_safe_coding_transform

class PythonCodeActEnv(Environment):
    def __init__(self):
        self.transform = create_safe_coding_transform()
        self._executor = PyExecutor()
        self._state = CodeState()

When a step executes, the raw CodeObservation flows through the internal _apply_transform method before returning to the client:

def _apply_transform(self, observation: Observation) -> Observation:
    return self.transform(observation)

This wiring ensures every observation—containing stdout, stderr, exit_code, and submitted code metadata—passes through the transformation pipeline automatically.

Creating Custom OpenEnv Transforms

To extend the system, implement the Transform protocol by defining a callable class with a __call__ method. The following example implements a token budget enforcer in my_transform.py:

from openenv.core.env_server.interfaces import Transform
from openenv.core.env_server.types import Observation

class TokenBudgetTransform(Transform):
    """Penalise observations that exceed a token budget."""
    def __init__(self, max_tokens: int = 500, penalty: float = -0.5):
        self.max_tokens = max_tokens
        self.penalty = penalty

    def __call__(self, observation: Observation) -> Observation:
        token_count = observation.metadata.get("token_count", 0)
        if token_count > self.max_tokens:
            observation.reward = (observation.reward or 0) + self.penalty
            observation.metadata["budget_violation"] = token_count
        return observation

Compose this custom transform with existing ones by creating a new factory function:

from .transforms import create_safe_coding_transform
from .my_transform import TokenBudgetTransform

def create_extended_coding_transform() -> CompositeTransform:
    return CompositeTransform([
        CodeSafetyTransform(),
        CodeQualityTransform(),
        TokenBudgetTransform(max_tokens=400),
    ])

Replace the environment's transform attribute at runtime with env.transform = create_extended_coding_transform().

Practical Implementation Examples

Running the Coding Environment with Default Transforms

from envs.coding_env.server.python_codeact_env import PythonCodeActEnv
from envs.coding_env.models import CodeAction

env = PythonCodeActEnv()
obs = env.reset()

action = CodeAction(code="print('Hello world!')")
obs = env.step(action)

print("Stdout:", obs.stdout)          # -> Hello world!

print("Reward:", obs.reward)          # -> 0.1 (concise bonus) if < 100 chars

print("Metadata:", obs.metadata)      # -> {'last_code': "print('Hello world!')"}

Adding a Custom Safety Rule

Extend CodeSafetyTransform to block additional patterns like subprocess.Popen:

from envs.coding_env.server.transforms import CodeSafetyTransform, CodeQualityTransform
from openenv.core.env_server.base_transforms import CompositeTransform

class StrictSafetyTransform(CodeSafetyTransform):
    def __init__(self):
        super().__init__(penalty=-2.0)
        self.dangerous_patterns.append(r"subprocess\.Popen")

def create_strict_transform() -> CompositeTransform:
    return CompositeTransform([StrictSafetyTransform(), CodeQualityTransform()])

env = PythonCodeActEnv()
env.transform = create_strict_transform()

Now any step containing subprocess.Popen incurs a -2.0 penalty and sets metadata["safety_violation"].

Testing Transform Effects

def test_safety_transform():
    env = PythonCodeActEnv()
    env.reset()
    unsafe_action = CodeAction(code="import os\nos.system('ls')")
    obs = env.step(unsafe_action)
    assert obs.reward == -1.0
    assert obs.metadata["safety_violation"] == r"import\s+os"

Summary

• OpenEnv Transform is a protocol-based system defined in src/openenv/core/env_server/interfaces.py that enables observation modification through simple callable classes.
• CompositeTransform in src/openenv/core/env_server/base_transforms.py chains multiple transforms sequentially, with order determining the final observation state.
• Built-in transforms like CodeSafetyTransform and CodeQualityTransform provide immediate utility for security scanning and reward shaping in coding environments.
• Custom transforms implement the __call__(self, observation: Observation) -> Observation signature and can modify rewards, metadata, or observation contents.
• The PythonCodeActEnv demonstrates integration via the _apply_transform method, which automatically routes observations through the configured pipeline.

Frequently Asked Questions

What is the OpenEnv Transform protocol?

The OpenEnv Transform protocol is a callable interface defined in src/openenv/core/env_server/interfaces.py that requires a __call__ method accepting an Observation and returning an Observation. Any class or function implementing this signature can serve as a transform, making the system flexible and language-agnostic within Python.

How does CompositeTransform handle multiple transforms?

CompositeTransform iterates through its transforms list in order, calling each transform with the observation returned by the previous one. This sequential execution means transforms can build upon metadata or reward modifications added earlier in the chain, but also requires careful ordering to avoid unintended overwrites.

Can I modify observations in place or must I return new instances?

Transforms receive a copy-by-reference of the Pydantic observation model, so you can modify the observation in place by mutating attributes like observation.reward or observation.metadata. However, you must return the observation object (whether modified or replaced) for the changes to propagate through the CompositeTransform pipeline.

Where should I place custom transform implementations?

Place custom transforms in your project's transform module or alongside environment-specific code. Import the Transform protocol from openenv.core.env_server.interfaces and the observation types from openenv.core.env_server.types. For coding environments, you can extend classes in envs/coding_env/server/transforms.py or create new modules following the same pattern.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}