Gymnasium Bridge (Client)

This file contains the low-level subprocess client for talking to the Python helper script scripts/rl/gymnasium_server.py.

The client is compact and reusable across examples:

• spawn a Python subprocess hosting an external Gymnasium environment,
• send one JSON object per line to stdin,
• read one JSON object response per line from stdout,
• validate observations/rewards/flags against a Lean side trust-boundary contract (Runtime.RL.Boundary.Contract).

Startup performs a describe handshake, checking that the external environment's declared spaces match the Lean side expectations (obsShape, nActions).

References:

• Gymnasium API reference (reset/step, terminated vs truncated): https://gymnasium.farama.org/
• The original Gym API paper (background on the env interface): https://arxiv.org/abs/1606.01540
• Gymnasium source repository (implementation reference): https://github.com/Farama-Foundation/Gymnasium
• Trust-boundary rationale and contract definition: NN.Runtime.RL.Boundary.

Low-level client

Client owns an IO.Process.Child and:

• sends a JSON object (one line) to stdin,
• reads one JSON object response from stdout,
• checks "ok": true and reports "error" strings as IO.userError.

def Runtime.RL.Gymnasium.stdio : IO.Process.StdioConfig

Standard stdio configuration for the Gymnasium subprocess protocol.

structure Runtime.RL.Gymnasium.Client (obsShape : Spec.Shape) (nActions : ℕ) : Type

Typed handle to a running Gymnasium subprocess environment.

Client provides request/response JSON helpers and enforces a Lean side Boundary.Contract on all values received from the external process.

• child : IO.Process.Child stdio

  The Python subprocess.

• contract : Boundary.Contract obsShape nActions

  Lean side contract enforced on every transition.

def Runtime.RL.Gymnasium.Client.Internal.requestObj {obsShape : Spec.Shape} {nActions : ℕ} (g : Client obsShape nActions) (j : Lean.Json) : IO (Std.TreeMap.Raw String Lean.Json compare)

Send a request object and return the response object map.

The expected response shape is: {"ok": true, ...} or {"ok": false, "error": "..."}

This stays in Client.Internal. Public callers should use spawn, reset, close, or the higher-level Session API, so JSON protocol details stay behind the Gymnasium trust boundary and out of the curated API.

def Runtime.RL.Gymnasium.Client.Internal.requireField (o : Std.TreeMap.Raw String Lean.Json compare) (field : String) : IO Lean.Json

Require that a JSON response object contains a given field.

def Runtime.RL.Gymnasium.Client.spawn {obsShape : Spec.Shape} {nActions : ℕ} (serverScript envId : String) (contract : Boundary.Contract obsShape nActions) (makeKwargs : Array (String × Lean.Json) := #[]) : IO (Client obsShape nActions)

Spawn the Gymnasium helper subprocess and perform a small handshake (describe).

The handshake checks that:

• the external environment reports the expected n_actions, and
• the external environment reports an obs_shape matching obsShape.toList.

If the handshake fails, the subprocess is terminated and an IO.userError is thrown.

makeKwargs (optional) is a JSON object encoded as an array of fields and passed through to the Python bridge as --make-kwargs <json>, which the server forwards to gym.make(envId, **kwargs). This is useful for environments that require constructor options, e.g. Atari RAM observations: makeKwargs := #[("obs_type", .str "ram")].

def Runtime.RL.Gymnasium.Client.reset {obsShape : Spec.Shape} {nActions : ℕ} (g : Client obsShape nActions) (seed? : Option ℕ := none) : IO (TorchLean.Tensor Float obsShape)

Reset the environment and return the initial observation.

def Runtime.RL.Gymnasium.Client.Internal.step {obsShape : Spec.Shape} {nActions : ℕ} (g : Client obsShape nActions) (action : ℕ) : IO (TorchLean.Tensor Float obsShape × Float × Bool × Bool)

Send one action through the JSON protocol and parse the resulting transition fields.

def Runtime.RL.Gymnasium.Client.close {obsShape : Spec.Shape} {nActions : ℕ} (g : Client obsShape nActions) : IO Unit

Close the subprocess (best-effort).

def Runtime.RL.Gymnasium.Client.withClient {α : Type} {obsShape : Spec.Shape} {nActions : ℕ} (serverScript envId : String) (contract : Boundary.Contract obsShape nActions) (k : Client obsShape nActions → IO α) : IO α

Spawn a client, run k, and ensure the subprocess is closed even if k throws.