Metadata-Version: 2.4
Name: gertlabs
Version: 2.0.0
Summary: Python SDK for the Gert Labs AI evaluation platform
Project-URL: Homepage, https://gertlabs.com
Project-URL: Repository, https://github.com/Gert-Labs-Inc/gert_labs
Project-URL: Documentation, https://gertlabs.com/docs
Author: Gert Labs
License-Expression: MIT
License-File: LICENSE
Requires-Python: >=3.10
Requires-Dist: httpx>=0.27
Requires-Dist: websocket-client>=1.6
Provides-Extra: data
Requires-Dist: pandas>=2.0; extra == 'data'
Requires-Dist: pyarrow>=15.0; extra == 'data'
Provides-Extra: test
Requires-Dist: pytest>=8.0; extra == 'test'
Provides-Extra: verifiers
Requires-Dist: verifiers<0.2,>=0.1.14; extra == 'verifiers'
Description-Content-Type: text/markdown

# Gert Labs Python SDK

A thin, synchronous Python wrapper over the Gert Labs REST API, for AI
researchers. Connect your own model or a public LLM endpoint to our environments:

- **Play locally.** Observations come to you, you run inference in your own
  environment, and you submit actions.
- **Connect a model.** Register an OpenAI-compatible endpoint once on the platform and let the
  platform drive it server-side for large-scale code generation, evaluation, and
  dataset builds.

Either way, collect training data at scale (code-submission evaluations,
counterfactual branch data, session replays) and load it into pandas.

## Install

```bash
pip install gertlabs           # core
pip install 'gertlabs[data]'   # + pandas/pyarrow for exports.load()
```

## Authentication

Create an API key in the dashboard (`sk_gert_...`) and set it in your
environment:

```bash
export GERT_API_KEY=sk_gert_...
```

```python
from gertlabs import GertClient
client = GertClient()                       # reads GERT_API_KEY
# or: GertClient(api_key="sk_gert_...", base_url="http://localhost:8080/api/v1")
```

Every method maps to an API endpoint and returns plain dicts. Errors raise a
`GertError` subclass (`AuthenticationError`, `ValidationError`,
`InsufficientCreditsError`, `RateLimitError`, ...). Async work returns a
`job_id`; block on it with `client.jobs.wait(job_id)`.

## Discover games

List the available environments and their tags. The slugs here (for example
`market_simulator`) are what you pass as `game=` in the examples below.

```python
for g in client.games.list():
    print(g["slug"], g["tags"])

client.games.list(tag="strategy")    # filter to one category
client.games.tags(active=True)       # tag taxonomy + per-tag game counts
```

You don't have to name a game. `play.create`, `dataset_builds.*`, and
`exports.create` all accept `tags` instead, and the platform matches an
environment for you:

```python
s = client.play.create(tags=["strategy"], seats=1, vs_ai=True, max_ticks=600)
```

## Play locally with your own model

Control one or more seats yourself: the platform sends observations, you run your
model in your own environment, and you submit actions. No provider registration or
org required. Fill the other seats with the platform's AI (`vs_ai=True`), or
control every seat yourself for self-play (`seats=player_count`).

```python
from gertlabs import GertClient
client = GertClient()

s = client.play.create(game="market_simulator", seats=1, vs_ai=True, max_ticks=600)
prompt = s["prompt"]            # game rules + observation/action schema

def decide(observation):
    # your local model/policy
    return []                   # see prompt for valid actions

with client.play.connect(s["session_id"], s["player_token"]) as ws:
    for msg in ws:
        if msg["type"] == "observation":
            ws.send_action(decide(msg["data"]))
        elif msg["type"] == "game_completed":
            print(msg["scores"]); break
```

To instead send REST over WebSockets, poll `client.play.get(session_id, player_token)` for
the latest observation and submit with `client.play.act(...)`, using the same loop.

## RL rollouts (pure environment, for training)

For RL training, drive the environment from your own loop: Gert is the game +
the scoreboard, and your
trainer calls your own model. Open a run once (a billing lease amortized over
many episodes), then `reset → step → done`. By default your policy controls
every seat (full self-play; a 1-player game is simply the environment), so any
game is trainable with no opponent corpus (score-actions and complete are the
exception: their playouts are played by rated code submissions, so a brand-new
game needs some before those two scorers work); reward is each policy seat's
dense per-tick score delta. Frozen league opposition is opt-in (`opponent_seats=`),
with the opponent skill band (`min_percentile`/`max_percentile`) as its
curriculum knob; for solo/coop games that declare it, `difficulty=` ([0,1],
explicit `game` + no opponent seats) is the environmental curriculum axis.

```python
from gertlabs import GertClient, rollouts
client = GertClient()

run = client.rollout_runs.open(max_episodes=1000)

def act(observation):
    # observation includes `_action_space` (legal actions) when the game enumerates them.
    # Call YOUR model here, parse its output into an action:
    return rollouts.parse_action(my_model(rollouts.format_observation(observation)))

result = rollouts.run_episode(
    client, run["run_id"], game="pot_limit_omaha_hilo",
    act_fn=act, opponent_seats=1,                         # 1 league seat; omit for full self-play
    min_percentile=0.4, max_percentile=0.6,               # hard-but-winnable band
)
print(result["total_reward"], result["outcome"])         # scores/rewards are higher-is-better platform-wide -- train on result["steps"] (1-player games: outcome is None; score is the signal)

client.rollout_runs.close(run["run_id"])                  # settles billing at actual ticks
```

Low-level access is `client.rollout_runs.reset/step/close` if you want to manage
the loop yourself; `client.rollout_runs.list(status="active")` recovers a lost
`run_id` (e.g. after a crashed training driver) so you can close the run. For [Prime Intellect `verifiers`](https://github.com/PrimeIntellect-ai/verifiers)
users, `pip install 'gertlabs[verifiers]'` and:

```python
from gertlabs.verifiers_env import load_environment
env = load_environment(game="pot_limit_omaha_hilo", max_episodes=5000)
# hand `env` to prime-rl / your verifiers-compatible trainer
env.close()   # settle the run's billing when done (or use `with load_environment(...) as env:`)
```

**Self-play & curriculum.** Self-play is the default: every seat is yours (read
per-seat `observations`/`rewards` from the response and pass
`actions={seat: action}` to `step`). Carve league seats out with
`opponent_seats=N` — 1-2 as frozen anchors for absolute-skill measurement, or
`player_count - 1` for the classic single-agent-vs-league mode. Open with
`auto_curriculum=True` to let the opponent band track your win-rate on
league episodes, and grow a long run with
`client.rollout_runs.extend(run_id, additional_episodes=...)`.

**Vectorized loops.** `client.rollout_runs.step_batch(run_id, steps)` steps up to
128 episodes in one call — gather N observations, run one batched forward pass,
submit the N actions together, paying the per-request overhead once and waiting on
one response instead of N request tails (entries execute concurrently; the batch
counts as one request for rate limiting). Each entry is
`{"episode_id", "action"|"actions", "expected_tick"?}`; results come back in order,
with per-episode failures as `ok: False` entries that never fail the batch.

**Train across a category & bring your own opponents.** Instead of a single
`game=`, pass `tags=[...]` to `reset` / `run_episode` / the adapters and Gert picks a
random game matching all the tags each episode — one run can span a whole
category (the resolved game comes back as `result["game"]`; with `tags`, `seed` does
not fix which game is chosen). Set `opponent_source="own"` (or the default `"both"`) to
add your org's rated submissions to the opponent league. Own opponents run user-submitted
code, so they require a user-code-capable rollout backend; where unavailable, `"own"` is
rejected and `"both"` uses only system opponents. Narrow the league further with
`opponent_models=[...]` / `opponent_providers=[...]` (the LLMs that generated the opponents)
and `opponent_created_after` / `opponent_created_before` (RFC3339) — these pass through
`reset` / `run_episode` / the OpenEnv and verifiers adapters alike. An episode requesting
opponent seats needs enough distinct rated opponents in the band; if a reset fails because a
game is too thin, widen the band, use `opponent_source="both"`, pick a richer tag, or drop
`opponent_seats` (self-play needs no league). The verifiers and
OpenEnv adapters fetch each resolved game's rules automatically (cached per game).

**OpenEnv.** For Gymnasium/OpenEnv-style trainers (TorchForge, verl, TRL, SkyRL),
`gertlabs.openenv_env.GertEnv` wraps the same engine as `reset()`/`step()`/`close()`.

**Two-phase rollouts (cheap tails).** Running your own inference to terminal is the
expensive part of long episodes. Instead, drive your model through the tactically
relevant near horizon with `step`, then call
`client.rollout_runs.complete(run_id, episode_id, samples=30)` to play the episode
out from its current state with skill-band code players -- no action injected, the
parent episode untouched -- and get back score stats (a V(s) playout estimate:
`{score: {mean, ...}, win_rate, failure_rate, ...}`). At a decision point,
`score_actions` is the Q(s,a) sibling: it completes each candidate action the same
cheap way. Combine with `fork` (self-play) for branch search. `complete` is
read-only, so a fork you finish that way stays live -- release it with
`client.rollout_runs.close_episode(run_id, episode_id)` (episodes driven to
terminal auto-close on their own).

**On-policy counterfactual scoring.** `score_actions` is cheap but its branch tails
are played by code players -- an off-policy Q(s,a). When you want *your model's*
Q(s,a) -- "what happens if I take this move and then I keep playing" --
`rollouts.score_actions_on_policy` runs the same fan-out on-policy: it forks the
live self-play episode N times per candidate, injects each candidate as the fork's
first action, drives every fork in lockstep with your model (one `step_batch`
round trip per tick; pass `act_batch_fn` for one batched forward pass per tick),
and returns the **same ranked response shape** as `score_actions` -- so cheap
shortlisting and on-policy tie-breaking are a one-call swap:

```python
result = rollouts.score_actions_on_policy(
    client, run_id, episode_id,
    candidate_actions=[a1, a2, a3],   # e.g. 3 samples from your model at this state
    samples_per_action=8,             # 8 forks per candidate
    act_fn=act,                       # your inference plays the continuations
    phase_ticks=20,                   # optional: on-policy for 20 ticks, then a cheap
)                                     # code-player tail (complete) finishes each fork
best = result["actions"][0]           # action_rank == 1
client.rollout_runs.step(run_id, episode_id, best["action"])  # advance the main line
```

The fan-out is one atomic, idempotent `fork_batch` call (K x N <= 128): every
sample restores from the same snapshot, the slot claim is all-or-nothing
(`RUN_EXHAUSTED` consumes nothing -- `extend` the run and retry), and a lost
response replays the same batch via its `op_key` (best-effort, while cached)
instead of stranding one. In `phase_ticks` mode each tailed fork RETAINS
~`tail_samples` episode-equivalents after its successful `complete` (31 slots
per fork with the defaults), so keep the run's remaining budget at
`max_episodes - episodes_used >= K x N x (1 + ceil(S x T / max_ticks_per_episode))`,
with S/T the effective tail samples/ticks (30 and the full per-episode cap when
omitted). The
fan-out is validated against a free read-only probe (`get_episode`) before any
slot is claimed, so a seat typo or missing act fn costs nothing. Turn handling is
automatic: the target defaults to the on-turn seat (turn-based) or the primary
seat, an off-turn seat is rejected rather than silently skipped, and your model is
called for exactly the seats in each result's `on_turn_seats`. Failures follow the
server's any-seat rule -- a sample whose branch loses any policy seat is excluded
and counted in `failure_rate` -- and cleanup is best-effort-with-retry: forks are
closed as they finish, stragglers are retried on the way out, and anything left
idle-reaps in ~10 minutes billing only its actual ticks. With a deterministic
`act_fn` on a deterministic game the N to-terminal samples are identical --
variance comes from your policy's sampling (or, in `phase_ticks` mode, the tail's
opponent draws).

## Quickstart: build a training dataset across many environments

The platform's batch engine runs your model across a whole category of game environments in
one job: generate code, evaluate it, and export the results.

```python
from gertlabs import GertClient
client = GertClient()

# Register your model endpoint ONCE in the dashboard (Org > Providers): it stores
# your upstream key and sets where your prompts are routed, so it's an interactive
# setup step. Then resolve its id here to use it from automation:
pid = next(p["provider_id"] for p in client.providers.list() if p["name"] == "my-model-v3")

# Estimate cost before spending
est = client.dataset_builds.evaluate_code(
    game_tags=["strategy"], custom_provider_id=pid,
    submission_count=20, match_count=200,
    export_type="both", export_format="parquet", dry_run=True,
)
print(est["estimated_credits"])

# Run across every strategy game: generate -> evaluate -> export
build = client.dataset_builds.evaluate_code(
    game_tags=["strategy"], custom_provider_id=pid,
    submission_count=20, match_count=200,
    export_type="both", export_format="parquet",
)
result = client.jobs.wait(build["job_id"], timeout=7200)["result"]

# Load each export (submissions + replays) into pandas, keyed by its declared type
frames_by_type = {}
for export_id in result["export_job_ids"]:
    client.jobs.wait(export_id)
    export_type = client.exports.get(export_id)["export_type"]
    frames_by_type[export_type] = client.exports.load(export_id)
    print(export_type, len(frames_by_type[export_type]), "rows")
```

Submission exports use one provider-attribution contract in both JSON and
Parquet (schema v3; v4 adds the pinned-difficulty label on replay and branch-exploration rows, which never pair with score_context anchors): `provider` is the readable generation-provider label,
`provider_source` is `gertlabs` for the system corpus or
`org:<immutable-org-slug>` for organization-owned data, and `model` is
independent. Nested branch/request/transcript metadata and replay seats use the
same readable attribution; internal `custom:<uuid>` routing keys are not
exported. `org:gertlabs` is distinct from the built-in source `gertlabs`.
Rows without recoverable attribution omit it rather than using placeholders.

```python
subs = frames_by_type["submissions"]
print(subs[["provider", "provider_source", "model"]].drop_duplicates())
```

## Counterfactual data (branch exploration)

```python
build = client.dataset_builds.explore(
    game_tags=["strategy"], custom_provider_id=pid,
    parent_count=5, samples_per_action=3, export_format="parquet",
)
result = client.jobs.wait(build["job_id"], timeout=7200)["result"]
client.jobs.wait(result["export_job_id"])        # singular in this mode
df = client.exports.load(result["export_job_id"])
```

## Agentic branching (process-reward data from a generation run)

Turn an agentic generation into a Q(s,a) / process-reward data source: at each
decision point (every `branch_interval` turns) it samples `max_actions` candidate
actions and scores each with `samples_per_action` rollouts that hold the action
fixed. Each record is `(state, action, value)` -- the action (reasoning + tool
calls) is the asset, the value is its label.

```python
sub = client.submissions.create(
    game="market_simulator", pipeline="agentic", custom_provider_id=pid,
    max_turns=60, branch=True, max_actions=3, samples_per_action=5,
    branch_interval=5, branch_mode="best_of_n",   # or "measure" (don't adopt)
)
client.jobs.wait(sub["job_id"])

# Inspect the recorded actions (action + value per decision point) ...
scores = client.jobs.branch_scores(sub["job_id"])
# ... or export the whole dataset (your own data, so it's free):
job = client.exports.create(export_type="agentic_branches", format="parquet")
client.jobs.wait(job["job_id"])
df = client.exports.load(job["job_id"])
```

The export is self-contained. JSON envelopes carry a `transcripts` map (each
submission's full generation transcript), a `request_contexts` map (the request
frame: system prompt, exact tool specs, verbatim base per-step prompts, ideation,
and the effective inference settings incl. `transcript_window` -- message lists
are a deterministic window over the transcript), and `value_contexts` (the
durable `branch_config` with effective opponent/eval/model provenance, plus the
pinned game version and `created_at` -- the submission's finalization time, an
upper bound on when its decision points were recorded, vs the envelope's
`exported_at`). The state a point's actions branched from is the request
frame plus `transcripts[submission_id][:state_entry_count]`, and each point
carries `pruned` -- the projection's prune verdicts over that state, computed
by the generation code itself and shipped as data (`{}` = nothing pruned;
`null` = not computed, e.g. the frame predates the current projection rules).
A `score_context` block keyed by game slug anchors raw scores against the
game's current-regime canonical-horizon league at export time (on every
export type). It carries the game's declared `score_version` (the
scoring-regime identity; raw scores from different versions are not
comparable) and `canonical_ticks` (the league's match horizon). Submissions
rows carry `score_version` and `score_ticks` labels written with their
aggregates and cleared with them on a regime bump -- a row is comparable
with the anchors when its `score_version` matches AND its `score_ticks`
equals `canonical_ticks` (a custom-horizon evaluation self-describes via its
own `score_ticks` instead of over-claiming). Agentic values are frozen at
their generation pin: pair them with the anchors when
`value_contexts[...].branch_config.score_version` matches the anchors'
`score_version` and `branch_config.ticks` equals `canonical_ticks`
(`game_version` is generation provenance, not the pairing key).
`completeness` counts present/missing pieces. In parquet the per-submission payloads are
nullable `transcript`/`request_context`/`value_context` columns on the
submission's first row IN EACH CHUNK (a submission whose decision points span
chunks repeats them), so a plain `exports.load()` DataFrame is fully
self-contained -- deduplicate before indexing:

```python
import json

from gertlabs import transcript

df = client.exports.load(job["job_id"])
frames = (
    df.dropna(subset=["transcript", "request_context"])
      .drop_duplicates("submission_id")
      .set_index("submission_id")
)

row = df.iloc[0]
state = transcript.reconstruct_messages(
    frames.loc[row.submission_id, "transcript"],           # raw JSON string column
    json.loads(frames.loc[row.submission_id, "request_context"]),
    int(row.decision_index),
    row.pruned,                                            # exported verdict map
    state_entry_count=int(row.state_entry_count),
)
# {"system", "tools", "messages"}: the model's exact input at this decision
# point, windowed under the frame's static transcript_window (pass
# windowed=False for the full history). The structural projection is
# conformance-tested against the generation code via a golden fixture it
# generates; the prune verdicts come precomputed from that code in the export.
```

`exports.download_files()` remains the exact-artifact path, and the only way to
read the parquet file metadata (`score_context:<slug>`,
`completeness`).

The final submission records its branching provenance in `branch_config` (`mode`
is `best_of_n` for a curated best-of-N selection, `measure` for the representative
main line, absent if not branched). Filter by it anywhere submissions are
read -- e.g. exclude curated output from a capability benchmark:

```python
reps = client.submissions.list(game="market_simulator", mine=True, branch_mode="none,measure")
job = client.exports.create(export_type="submissions", branch_modes=["none", "measure"], format="parquet")
```

## Export only top performers (filter by evaluation score)

```python
job = client.exports.create(
    export_type="submissions", min_percentile=0.9,
    tags=["strategy"], format="parquet",
)
client.jobs.wait(job["job_id"])
df = client.exports.load(job["job_id"])
```

## Connected model: run it server-side, then branch

Register a provider (see below) and the platform runs it as an AI seat, so you can
spectate or branch without running inference yourself.

```python
session = client.play.create(
    game="market_simulator", autostart=True,
    ai_mode="agentic_player", custom_provider_id=pid, spectate_mode="private",
)
with client.play.spectate(session["session_id"]) as ws:
    for msg in ws:
        if msg["type"] == "game_completed":
            print(msg["scores"]); break

# fan out 8 counterfactual branches from a checkpoint
client.play.branch(session["session_id"], count=8)
```

## Fine-grained control

For a single hand-written submission instead of a batch build:

```python
sub = client.submissions.create(game="market_simulator", language="python", code=SOURCE)
client.jobs.wait(sub["job_id"]) if "job_id" in sub else None
ev = client.submissions.evaluate(sub["submission_id"], match_count=500)
client.jobs.wait(ev["job_id"])
print(client.submissions.get(sub["submission_id"])["elo_rating"])
```

## Resource reference

| Resource | Methods |
|----------|---------|
| `client.games` | `list`, `get`, `tags` |
| `client.dataset_builds` | `evaluate_code`, `explore` |
| `client.exports` | `list`, `create`, `get`, `download`, `download_files`, `load`, `reset_tracking` |
| `client.jobs` | `get`, `wait`, `branch_scores` |
| `client.providers` | `list` (create/update/delete are dashboard-only -- see below) |
| `client.submissions` | `list`, `get`, `create`, `delete`, `bulk_delete`, `evaluate`, `batch_evaluate`, `evaluations` |
| `client.sessions` | `list`, `get`, `logs`, `branches`, `branch_scores`, `delete` |
| `client.billing` | `balance`, `usage` |
| `client.play` | `create`, `join`, `branch`, `get`, `act`, `leave`, `connect`, `spectate` |
| `client.rollout_runs` | `open`, `list`, `reset`, `get_episode`, `step`, `step_batch`, `fork`, `fork_batch`, `score_actions`, `complete`, `close_episode`, `extend`, `close` |

List methods follow cursor pagination automatically and return a full list.
Cap results with `max_items=N` and tune the wire page size with `page_size=`
(server max 100); other keyword arguments are forwarded as filters. The SDK owns
the `limit` query param, so pass `max_items=`/`page_size=` rather than `limit=`.

## Providers are configured in the dashboard

Provider registration isn't available through the SDK: it stores a secret and is
gated to logged-in dashboard sessions (API keys are rejected). Create a provider
in the dashboard under Org > Providers; the SDK only lists them
(`client.providers.list()`), so you can resolve a `provider_id`. Model
identifiers in `allowed_models` are recorded verbatim as provenance on
everything they generate (submissions, sessions, branch data, exports) --
register each checkpoint as a distinct model rather than re-pointing an
existing identifier at new weights.

## Errors

API errors raise a `GertError` subclass (`AuthenticationError`,
`PermissionError`, `NotFoundError`, `ValidationError`, `ConflictError`,
`InsufficientCreditsError`, `RateLimitError`, `ServerError`). Each carries
`.code`, `.status`, `.request_id`, and `.body` (the full parsed error response).
For example, starting a dataset build while one is already running raises
`ConflictError`, and `e.body["existing_job_id"]` is the id of the in-flight job:

```python
from gertlabs import ConflictError
try:
    build = client.dataset_builds.evaluate_code(game_tags=["strategy"], custom_provider_id=pid)
except ConflictError as e:
    build = {"job_id": e.body["existing_job_id"]}   # wait on the existing build
client.jobs.wait(build["job_id"])
```
