Episode runtime

harness_policy.runtime.driver

High level of the runtime: the model writes programs and a kernel runs them.

One episode: reset the worker, claim the evaluator token, start a kernel bound to the episode, then loop: model reply -> <python> runs in the kernel, <verify> asks a separate verifier call, <finish> ends the episode. Program output, escalated events and fresh camera views go back to the model. The evaluator outcome is read with the token after the loop and is never shown to the model.

TaskConfig

class TaskConfig

The driver’s configuration for one task: the worker endpoint, the main model and the verifier, runtime settings, the prompt, the seeds and the per-episode budgets. Experiment.task_config builds one from an experiment.

snapshot

def snapshot() -> dict

The configuration as episode manifests and experiment.json record it (all but path and base_prompt; the lock as agent_lock; the embodiment kind when it is not the default robot, so robot records are as before).

layer_settings

def layer_settings(runtime: dict) -> dict

The interface layers (R0-R4) a harness switches: learned, reactive, handlers.

model_services

def model_services(runtime: dict, main: dict, verifier: dict) -> dict

The models and servers the kernel’s skills call: VLM, pointer, segmenter, policies, world models, and the success verifier (Tab. 5: it spans layers; direct control has none).

resolve_skills

def resolve_skills(runtime: dict, main: dict, verifier: dict, embodiment: str='robot') -> Resolution

The skills of this configuration that can run on the embodiment kind with its services: runtime.skills when it names them, else those of its interface layers.

attach_library

def attach_library(config: 'TaskConfig', directory, layers=('notes', 'skills', 'reactive'))

Return a copy of config that uses the library at directory (frozen at its current version) on top of the harness prompt without any library.

load_task_config

def load_task_config(path, *, task=None, seeds=None, model=None, deploy=None, sets=()) -> TaskConfig

The driver’s configuration of an experiment file (configs/experiments/*.toml) for one task: task (default: the experiment’s first), seeds, model (a model file’s name or path, configs/models/), deploy and –set items override the file.

run_program_episode

def run_program_episode(config: TaskConfig, seed: int, directory: Path, *, chat=None, verifier=None, environment=None) -> dict

Run one episode of config.task with seed and return its summary (status, evaluator outcome, model calls, tokens, cost, times), also written with the trace and manifest under directory. chat, verifier and environment replace the configured models and worker. An error ends the episode with status “error” instead of raising.

run_task

def run_task(config: TaskConfig, output: Path, *, script: list | dict | None=None) -> dict

Run every seed. With script (a list of replies, or {“replies”, “verifier”}), replies are replayed and no model is called.

harness_policy.runtime.kernel

A persistent Python kernel for model-written programs, in its own process.

The host sends cells over stdin and reads one JSON reply per cell from a private descriptor, so anything a program prints cannot corrupt the protocol. The kernel namespace comes from a bootstrap callable module:function that receives JSON options and returns a session object with:

namespace: dict       names visible to programs
drain() -> dict       attachments and events gathered during the cell
close() -> None

PlainSession

class PlainSession
    __init__(options=None)

A bootstrap with an empty namespace, for tests and debugging.

drain

def drain()

Nothing: a plain session gathers no attachments or events.

close

def close()

Nothing to release.

KernelDied

class KernelDied(RuntimeError)

The kernel process failed to start, is not running, or did not answer in time (it is then killed).

sandbox_available

def sandbox_available() -> bool

Whether this machine can run the kernel’s sandbox (util-linux unshare with user, mount and PID namespaces).

Kernel

class Kernel
    __init__(bootstrap: str, options: dict, *, python: str=sys.executable, cwd: str | None=None, env: dict | None=None, log_path: Path | None=None, start_timeout: float=300.0)

Host-side handle. One kernel lives for one episode.

run

def run(code: str, timeout: float) -> dict

Run code as the next cell and return the kernel’s reply: status (ok, error, escalated, episode_ended or finished), clipped stdout, error, seconds, and what the session drained (attachments, events, state, usage). Raises KernelDied when the kernel is not running, exits, or does not reply within timeout seconds.

kill

def kill()

Kill the kernel process if it still runs.

close

def close()

Ask the kernel to close its session and exit (killing it when it does not within 30 s), then close the pipes and the log.

format_error

def format_error(error: BaseException) -> str

Traceback restricted to program frames, plus the frame that raised.

restore_api

def restore_api(namespace: dict, api: dict) -> list[str]

Put back the session’s API names that a program rebound or deleted (a model’s own def pick(m): ... would otherwise break every later call of the skill) and return them, sorted.

main

def main(argv=None)

The kernel process: build the session from --bootstrap with the JSON --options, then run cells read from stdin until a close request or end of input.

harness_policy.runtime.session

The robot’s session: the episode core (runtime/episode.py) for robot arms.

A robot session connects to a robot worker (the Robot API: profile, state, tick, observe, …). A step is a control tick; its command is the vector of joint targets (each arm’s joints and gripper, embodiment groups); guards observe joints, grippers and link6 poses; an execution’s scope is a region of the workspace one arm’s link6 must stay in; a guard can also open the grippers. The robot skills’ constants (ARMS, CAMERAS) live here.

Bootstrap: harness_policy.runtime.kernel –bootstrap harness_policy.runtime.session:create_session (the episode core’s, which builds the session of options[“embodiment”], default robot).

Observation

class Observation
    __init__(data)

Camera images (RGB uint8 arrays), camera parameters, and robot state.

RobotSession

class RobotSession(EpisodeCore)

The episode core for robot arms: the embodiment’s groups and constants, the gripper commands, and the joint-vector commands of every tick.

harness_policy.runtime.events

Events end every bounded execution and flow up the hierarchy.

An event is offered to program handlers first. Only an unhandled event escalates to the model, as an Escalation raised out of the running program.

Event

class Event

Why an execution ended: its kind (a built-in kind above or a guard’s own action), the source that raised it (a skill, a guard, “shield”, “environment”), details, and the robot step it happened at.

to_dict

def to_dict() -> dict

The event as a plain dict.

ExecutionTrace

class ExecutionTrace

What one bounded execution did, and why it ended.

ok

def ok() -> bool

Whether the execution ended nominally (COMPLETED or PREDICATE_SATISFIED).

ticks

def ticks() -> int

Robot steps the execution took.

to_dict

def to_dict() -> dict

The trace as a plain dict, with ok.

Escalation

class Escalation(Exception)
    __init__(trace: ExecutionTrace)

An event no handler resolved. It ends the current program.

ProgramFinished

class ProgramFinished(Exception)
    __init__(summary: str='')

A program called finish(): its own checks say the task is done. The episode ends without another model call.

EpisodeEnded

class EpisodeEnded(Exception)
    __init__(trace: ExecutionTrace)

The environment terminated or truncated the episode during a program.

harness_policy.runtime.reactive

Model-written guards and controllers evaluated at every control tick.

Reactive code must be a pure function of the observation and its own state. While it runs, an audit hook refuses file, network and process operations, and a per-tick time budget bounds it. Any violation is a REACTIVE_FAULT.

ReactiveFault

class ReactiveFault(Exception)
    __init__(source: str, reason: str)

Reactive code broke its rules: a forbidden operation, an exception, an overrun of the tick budget, or a signal that is missing or not a finite number. source names the guard or controller, reason says what happened.

call_reactive

def call_reactive(source: str, function: Callable, *args, budget_seconds: float) -> Any

Run one reactive function under the audit restriction and time budget.

read_signal

def read_signal(observation: dict, signal_name: str | Callable, source: str, budget_seconds: float) -> float

A watched signal’s value: a dotted key into observation (list items by index, as in left_ee.2) or a callable of it, run like other reactive code. A missing key or a value that is not a finite number is a ReactiveFault from source.

Guard

class Guard

A registered guard: a when predicate or a watch signal with bounds, the action it takes when it fires, how often it fired, and the watched signal’s last value.

describe

def describe() -> dict

The guard as the event it raises reports it: name, kind, action and trigger count, and for a watch its signal, bounds and last value.

ReactiveLayer

class ReactiveLayer
    __init__(budget_seconds: float, actions=BUILTIN_ACTIONS)

An episode’s guards, evaluated in registration order at every control tick of a bounded execution, each within budget_seconds.

when

def when(predicate: Callable, action: str, name: str | None=None) -> Guard

Register and return a guard that fires when predicate(observation) is true. action is stop, hold, open_gripper or an event name; a guard of the same name is replaced, and the name defaults to when_<n>.

watch

def watch(signal_name, low=None, high=None, event=None, name=None) -> Guard

Register and return a guard that fires when a signal (a dotted observation key or a callable, see read_signal) falls below low or rises above high. The event defaults to <SIGNAL>_OUT_OF_RANGE and the name to watch_<n>.

clear

def clear(names=None) -> None

Remove the named guards (a name or several; unknown names are ignored), or all.

evaluate

def evaluate(observation: dict) -> Guard | None

Return the first guard that fires on this observation, in registration order.

harness_policy.runtime.shield

A fixed, non-learned check on every command before it takes effect.

The shield runs in the environment worker, outside the process that executes model-written code, so programs cannot modify it. Every embodiment’s shield has the same interface (CommandShield): it judges one command against the measured state, and the command passes unchanged or is rejected with the reasons; it never shortens, clamps or otherwise alters a command, since a program whose command was quietly changed does not learn it was wrong. A tick’s shield field is its ShieldResult.

Shield (below) is the robot’s. It sees absolute joint position commands and, when the worker supplies forward kinematics, the end-effector positions those commands imply. It catches commands that are bugs, and nothing else. Rejected are:

• a malformed command (wrong length, a value that is not finite);
• a joint outside its limits, or an end effector outside the workspace box, further than the robot already is (contact can push a measured joint or hand slightly out; holding there or moving back in is not a bug);
• a jump: an arm’s joints more than max_joint_jump_rad, or its end effector more than max_cartesian_jump_m, from both the previous accepted command and the measured state. One control tick of a real motion moves far less; a jump is an IK solution on another branch, a target in the wrong frame, a controller that diverged. A smooth sequence is never rejected for running ahead of an arm that lags behind it, and a command from where the arm is (a hold, the next skill’s first waypoint, a retreat) is never rejected for where the last one went.

Speed and acceleration are not the shield’s: the skills time their own motions, and a physical robot’s controller enforces its own rate limits.

ArmGroup

class ArmGroup

One arm’s indices in the command vector: its joints, and its gripper if it has one.

ShieldProfile

class ShieldProfile

The limits the shield checks for one embodiment: the arm groups, joint bounds, jump limits and, optionally, the end-effector workspace box.

from_dict

def from_dict(data: dict, joint_lower, joint_upper) -> 'ShieldProfile'

A validated profile from a worker’s shield settings (groups by name with their joints and gripper, max_joint_jump_rad, and optionally workspace, max_cartesian_jump_m, joint_limit_tolerance_rad) and the joint bounds.

ShieldResult

class ShieldResult

The verdict on one command: “pass” with the command unchanged, or “rejected” with no command and the reason (the failed check and the values it compared). The command is the embodiment’s: a robot’s joint vector.

to_dict

def to_dict() -> dict

The result as a plain dict, as a tick’s shield field.

CommandShield

class CommandShield(Protocol)

The shield of one worker, whatever the embodiment: a fixed check of every command against the measured state (current) and what else the worker knows (context, such as a robot’s forward kinematics), passing it unchanged or rejecting it with reasons.

reset

def reset() -> None

Forget what earlier commands of the episode established.

check

def check(command, current, **context) -> ShieldResult

The verdict on command.

Shield

class Shield
    __init__(profile: ShieldProfile)

The robot’s command check of one worker (a CommandShield). It keeps the last accepted command, against which (and the measured state) jumps are judged until reset.

reset

def reset() -> None

Forget the last accepted command: the next is judged from the measured state.

check

def check(command, current, fk: Kinematics | None=None) -> ShieldResult

Judge an absolute joint command against the measured joints current: shape, finiteness, joint limits and joint jumps, and with fk (joints to each group’s end-effector position) the workspace box and Cartesian jumps. The first failed check rejects it; a passed command becomes the previous command.

harness_policy.runtime.chat

Multi-turn text completion for the high level.

Chat Completions covers OpenRouter and self-hosted OpenAI-compatible servers (vLLM, SGLang) through openai_chat with a base_url. Messages use the harness content parts (input_text, input_image).

ChatReply

class ChatReply

One completion: its text, usage (input and output tokens, cost in USD when the provider reports it, finish_reason), the provider’s response id, and the seconds the request took with its retries.

ChatModel

class ChatModel
    __init__(config: dict, transport=None)

A Chat Completions client for one model role (MODEL_KEYS): OpenRouter or an OpenAI-compatible server. The openrouter backend needs its key in the environment; transport replaces the HTTP call (tests).

payload

def payload(system: str, messages: list[dict]) -> dict

The request body: system and the harness messages as chat messages, the output limit and extra_body fields, and on OpenRouter usage accounting and reasoning.

complete

def complete(system: str, messages: list[dict]) -> ChatReply

One completion of messages under system, starting an on-demand service through the broker first. Failed requests are retried (retries times, or until retry_seconds have passed), except client errors other than HTTP 429; the last failure is raised as a RuntimeError.

sse_transport

def sse_transport(url, headers, payload, timeout)

POST a streamed chat completion and assemble the chunks into the non-streamed response shape. timeout bounds each read, so keep-alive comments reset it.

scripted_models

def scripted_models(script) -> dict

The models of a scripted episode: script is a list of replies, or {“replies”: […], “verifier”: […]} as scripts/replay_script.py writes it; without recorded verifier answers the verifier says YES.

ScriptedChat

class ScriptedChat
    __init__(replies: list[str])

Replays fixed replies in order. For pipeline checks without model calls.

complete

def complete(system: str, messages: list[dict]) -> ChatReply

The next scripted reply (a <finish> block once they run out), with zero usage.