Function train_step

metaclaw/cli.py:449–501

Trigger a single RL training step on the running MetaClaw instance. Sends a request to the running proxy's admin endpoint and waits for the training step to complete. Requires MetaClaw to be running in RL/auto mode. Useful for bench-driven experiments or manual control.

(port: int | None, timeout: int)

    help="Timeout in seconds (default: 600).",
)
def train_step(port: int | None, timeout: int):
    """Trigger a single RL training step on the running MetaClaw instance.

    Sends a request to the running proxy's admin endpoint and waits for the
    training step to complete.  Requires MetaClaw to be running in RL/auto
    mode.  Useful for bench-driven experiments or manual control.
    """
    import json
    import urllib.request
    import urllib.error

    if port is None:
        cs = ConfigStore()
        port = cs.get("proxy.port") or 30000

    url = f"http://localhost:{port}/v1/admin/train_step"
    req = urllib.request.Request(
        url,
        data=b"{}",
        headers={"Content-Type": "application/json"},
        method="POST",
    )
    try:
        with urllib.request.urlopen(req, timeout=timeout) as resp:
            result = json.loads(resp.read().decode())
    except urllib.error.HTTPError as e:
        body = e.read().decode() if e.fp else ""
        click.echo(f"Error ({e.code}): {body}", err=True)
        sys.exit(1)
    except urllib.error.URLError as e:
        click.echo(
            f"Cannot connect to MetaClaw at port {port}: {e.reason}\n"
            "Is MetaClaw running? Try: metaclaw status",
            err=True,
        )
        sys.exit(1)
    except Exception as e:
        click.echo(f"Error: {e}", err=True)
        sys.exit(1)

    status = result.get("status", "unknown")
    if status == "ok":
        click.echo(
            f"Train step #{result.get('step', '?')} complete: "
            f"samples={result.get('samples', '?')} "
            f"mean_reward={result.get('mean_reward', '?')} "
            f"success_rate={result.get('success_rate', '?')}"
        )
    elif status == "skipped":
        click.echo(f"Skipped: {result.get('message', 'no samples')}")
    else:
        click.echo(f"Error: {result.get('message', 'unknown error')}", err=True)
        sys.exit(1)


@metaclaw.command(name="config")