"""froog API client (Python 3.8+, standard library only).

Generated from https://froogcorp.com/openapi.json by deploy/build-sdks.py; don't edit.
API version 0.12.0. Docs: https://froogcorp.com/docs/

    from froogdev import Froog           # save this file as froogdev.py (curl -fsSL https://froogcorp.com/sdk/froog.py -o froogdev.py)
    froog = Froog()                      # FROOG_API_KEY from the environment
    with froog.sandboxes.create(size="small") as sb:     # deleted when the block ends
        print(sb.exec("uname -a").stdout)

    # or one method per operation:
    froog.create_sandboxes({"name": "box"})
    froog.wait_until("box", "running")
    print(froog.exec("box", {"command": "uname -a"})["stdout"])

Async: AsyncFroog and AsyncSandbox have the same methods as coroutines.
"""
import asyncio
import base64
import functools
import json
import os
import time
import urllib.error
import urllib.parse
import urllib.request

__all__ = ["Froog", "FroogError", "Sandbox", "Files", "Process", "CommandResult", "CommandError",
           "AsyncFroog", "AsyncSandbox", "AsyncProcess"]


class FroogError(Exception):
    def __init__(self, status, code, message, request_id=None):
        super().__init__(f"{code}: {message}")
        self.status, self.code, self.message, self.request_id = status, code, message, request_id


class Froog:
    def __init__(self, api_key=None, base_url=None, timeout=3700, org=None):
        self.api_key = api_key or os.environ.get("FROOG_API_KEY")
        # The organization requests act in (id or name, or "personal"): X-Froog-Org.
        self.org = org or os.environ.get("FROOG_ORG")
        self.base_url = (base_url or os.environ.get("FROOG_API_URL") or "https://api.froogcorp.com").rstrip("/")
        self.timeout = timeout
        self.sandboxes = Sandboxes(self)

    def _call(self, method, path, query=None, json_body=None, raw=None, raw_response=False, auth=True, idempotency_key=None):
        url = self.base_url + path
        if query:
            q = {k: (str(v).lower() if isinstance(v, bool) else v) for k, v in query.items() if v is not None}
            if q:
                url += "?" + urllib.parse.urlencode(q, doseq=True)
        headers = {"user-agent": "froog-python"}
        if auth and self.api_key:
            headers["authorization"] = "Bearer " + self.api_key
        if self.org:
            headers["x-froog-org"] = self.org
        if idempotency_key:
            headers["idempotency-key"] = idempotency_key
        data = None
        if json_body is not None:
            headers["content-type"] = "application/json"
            data = json.dumps(json_body).encode()
        elif raw is not None:
            headers["content-type"] = "application/octet-stream"
            data = raw
        req = urllib.request.Request(url, data=data, method=method, headers=headers)
        try:
            with urllib.request.urlopen(req, timeout=self.timeout) as r:
                body = r.read()
        except urllib.error.HTTPError as e:
            code, message = f"http_{e.code}", e.reason
            try:
                err = json.loads(e.read())["error"]
                code, message = err.get("code", code), err.get("message", message)
            except Exception:
                pass
            raise FroogError(e.code, code, message, e.headers.get("x-request-id")) from None
        if raw_response:
            return body
        return json.loads(body) if body else None

    def wait_until(self, name, state, timeout=600):
        """Poll until the sandbox reaches `state` (e.g. "running" after create or start)."""
        until = time.time() + timeout
        while True:
            sb = self.get_sandbox(name)
            if sb["state"] == state:
                return sb
            if sb["state"] == "error":
                raise FroogError(409, "sandbox_error", f"{name}: {sb.get('error')}")
            if time.time() > until:
                raise FroogError(408, "timeout", f"{name} didn't become {state}")
            time.sleep(2)

    def _stream(self, path, query=None):
        """A GET answered with NDJSON: yield each line as a dict as it arrives, until the stream ends."""
        url = self.base_url + path
        q = {k: (str(v).lower() if isinstance(v, bool) else v) for k, v in (query or {}).items() if v is not None}
        if q:
            url += "?" + urllib.parse.urlencode(q, doseq=True)
        headers = {"user-agent": "froog-python", "accept": "application/x-ndjson"}
        if self.api_key:
            headers["authorization"] = "Bearer " + self.api_key
        if self.org:
            headers["x-froog-org"] = self.org
        try:
            r = urllib.request.urlopen(urllib.request.Request(url, headers=headers), timeout=self.timeout)
        except urllib.error.HTTPError as e:
            code, message = f"http_{e.code}", e.reason
            try:
                err = json.loads(e.read())["error"]
                code, message = err.get("code", code), err.get("message", message)
            except Exception:
                pass
            raise FroogError(e.code, code, message, e.headers.get("x-request-id")) from None
        with r:
            for line in r:
                if line.strip():
                    yield json.loads(line)

    def exec_stream(self, name, body):
        """Run a command and yield its output as it arrives: dicts of type started, stdout/stderr (with
        `data`), then exit (`exit_code`, `timed_out`) or error. `POST /v1/sandboxes/{name}/exec` with stream."""
        url = self.base_url + "/v1/sandboxes/" + urllib.parse.quote(str(name), safe="") + "/exec"
        headers = {"user-agent": "froog-python", "content-type": "application/json", "accept": "application/x-ndjson"}
        if self.api_key:
            headers["authorization"] = "Bearer " + self.api_key
        if self.org:
            headers["x-froog-org"] = self.org
        req = urllib.request.Request(url, data=json.dumps(dict(body, stream=True)).encode(), method="POST", headers=headers)
        try:
            r = urllib.request.urlopen(req, timeout=self.timeout)
        except urllib.error.HTTPError as e:
            code, message = f"http_{e.code}", e.reason
            try:
                err = json.loads(e.read())["error"]
                code, message = err.get("code", code), err.get("message", message)
            except Exception:
                pass
            raise FroogError(e.code, code, message, e.headers.get("x-request-id")) from None
        with r:
            for line in r:
                if line.strip():
                    yield json.loads(line)

    def mcp(self, body):
        """MCP server (Streamable HTTP, stateless): JSON-RPC 2.0 initialize, tools/list and tools/call, with tools to list, create, run commands in, fork, snapshot, stop and delete sandboxes, run background commands and read their output, read and write files and host ports. A scoped key may call only the tools its actions allow. An exec call with _meta.progressToken from a client that accepts text/event-stream is answered as server-sent events (output as notifications/progress, then the result). With FROOG_SERVER_MCP_OAUTH on, a 401 points at OAuth metadata (MCP authorization) and the access tokens are scoped API keys. `POST /mcp`"""
        return self._call("POST", f"/mcp", json_body=body)

    def get_account(self):
        """The signed-in account. `GET /v1/account`"""
        return self._call("GET", f"/v1/account")

    def get_alert_settings(self):
        """Alerts settings: which emails the account gets, and automatic rebuilds. `GET /v1/account/alerts`"""
        return self._call("GET", f"/v1/account/alerts")

    def update_alert_settings(self, body):
        """Turn kinds of email, or automatic rebuilds, on or off. `PATCH /v1/account/alerts`"""
        return self._call("PATCH", f"/v1/account/alerts", json_body=body)

    def close_account(self):
        """Close the account: cancel the plan now, stop every sandbox, revoke every key; reopen within 30 days. Needs the header X-Froog-Confirm-Close set to the account's email, and a full-access key or a dashboard session. `POST /v1/account/close`"""
        return self._call("POST", f"/v1/account/close")

    def list_connections(self):
        """Ways to sign in to the account (Google, GitHub, email, AgentID); the primary one's address gets the account's emails. `GET /v1/account/connections`"""
        return self._call("GET", f"/v1/account/connections")

    def send_connection_email_code(self, body):
        """Send a 6-digit code to an address to add it as a sign-in connection. `POST /v1/account/connections/email`"""
        return self._call("POST", f"/v1/account/connections/email", json_body=body)

    def verify_connection_email_code(self, body):
        """Add the address with the code from the email. `POST /v1/account/connections/email/verify`"""
        return self._call("POST", f"/v1/account/connections/email/verify", json_body=body)

    def remove_connection(self, id):
        """Remove a sign-in connection (not the last one: 409 last_connection). `DELETE /v1/account/connections/{id}`"""
        return self._call("DELETE", f"/v1/account/connections/{urllib.parse.quote(str(id), safe='')}")

    def get_data_retention(self):
        """Data controls: zero data retention, snapshots off, monitoring off. `GET /v1/account/data-retention`"""
        return self._call("GET", f"/v1/account/data-retention")

    def update_data_retention(self, body):
        """Change data controls (dashboard session only). `PATCH /v1/account/data-retention`"""
        return self._call("PATCH", f"/v1/account/data-retention", json_body=body)

    def set_primary_connection(self, body):
        """Make a connection primary: its address gets the account's emails. `PUT /v1/account/primary`"""
        return self._call("PUT", f"/v1/account/primary", json_body=body)

    def reopen_account(self):
        """Reopen a closed account within its 30-day window (the plan isn't restored). `POST /v1/account/reopen`"""
        return self._call("POST", f"/v1/account/reopen")

    def set_ai_environment_profile(self, env, body):
        """The credential profile an environment's sandboxes get. `PUT /v1/ai/environments/{env}`"""
        return self._call("PUT", f"/v1/ai/environments/{urllib.parse.quote(str(env), safe='')}", json_body=body)

    def list_ai_profiles(self):
        """Credential profiles and their accounts (never the keys). `GET /v1/ai/profiles`"""
        return self._call("GET", f"/v1/ai/profiles")

    def create_ai_profile(self, body):
        """Create a credential profile, empty or as a copy. `POST /v1/ai/profiles`"""
        return self._call("POST", f"/v1/ai/profiles", json_body=body)

    def delete_ai_profile(self, profile):
        """Delete a profile (not the default, not one an environment uses). `DELETE /v1/ai/profiles/{profile}`"""
        return self._call("DELETE", f"/v1/ai/profiles/{urllib.parse.quote(str(profile), safe='')}")

    def get_ai_profile(self, profile):
        """One credential profile (name or id). `GET /v1/ai/profiles/{profile}`"""
        return self._call("GET", f"/v1/ai/profiles/{urllib.parse.quote(str(profile), safe='')}")

    def update_ai_profile(self, profile, body):
        """Rename a profile, or make it the default. `PATCH /v1/ai/profiles/{profile}`"""
        return self._call("PATCH", f"/v1/ai/profiles/{urllib.parse.quote(str(profile), safe='')}", json_body=body)

    def add_ai_credential(self, profile, body):
        """Add a provider key or custom API to a profile (another account when the provider has one). `POST /v1/ai/profiles/{profile}/credentials`"""
        return self._call("POST", f"/v1/ai/profiles/{urllib.parse.quote(str(profile), safe='')}/credentials", json_body=body)

    def remove_ai_credential(self, profile, id):
        """Remove a key from a profile. `DELETE /v1/ai/profiles/{profile}/credentials/{id}`"""
        return self._call("DELETE", f"/v1/ai/profiles/{urllib.parse.quote(str(profile), safe='')}/credentials/{urllib.parse.quote(str(id), safe='')}")

    def switch_ai_credential(self, profile, id):
        """Use this account first for its provider. `PUT /v1/ai/profiles/{profile}/credentials/{id}/active`"""
        return self._call("PUT", f"/v1/ai/profiles/{urllib.parse.quote(str(profile), safe='')}/credentials/{urllib.parse.quote(str(id), safe='')}/active")

    def get_ai_proxy(self):
        """The AI proxy: on or off, blocked, subscriptions withheld, its URL. `GET /v1/ai/proxy`"""
        return self._call("GET", f"/v1/ai/proxy")

    def update_ai_proxy(self, body):
        """Turn the AI proxy on or off (sandboxes switch at their next start), block every proxied sandbox. `PUT /v1/ai/proxy`"""
        return self._call("PUT", f"/v1/ai/proxy", json_body=body)

    def get_ai_usage(self, days=None):
        """Proxied AI usage per sandbox: requests, tokens, estimated cost. `GET /v1/ai/usage`"""
        return self._call("GET", f"/v1/ai/usage", query={"days": days})

    def list_api_keys(self):
        """API keys (secrets are never returned). `GET /v1/api-keys`"""
        return self._call("GET", f"/v1/api-keys")

    def create_api_key(self, body):
        """Create an API key: a preset or actions, sandboxes or labels, expiry. `POST /v1/api-keys`"""
        return self._call("POST", f"/v1/api-keys", json_body=body)

    def revoke_api_key(self, id):
        """Revoke an API key. `DELETE /v1/api-keys/{id}`"""
        return self._call("DELETE", f"/v1/api-keys/{urllib.parse.quote(str(id), safe='')}")

    def rotate_api_key(self, id):
        """A new secret for a key (same id, scope and expiry). `POST /v1/api-keys/{id}/rotate`"""
        return self._call("POST", f"/v1/api-keys/{urllib.parse.quote(str(id), safe='')}/rotate")

    def list_audit_events(self, action=None, sandbox=None, since=None, until=None, sort=None, limit=None, cursor=None):
        """The account's audit log, newest first (90 days): who created, deleted or changed what, from which IP. `GET /v1/audit`"""
        return self._call("GET", f"/v1/audit", query={"action": action, "sandbox": sandbox, "since": since, "until": until, "sort": sort, "limit": limit, "cursor": cursor})

    def export_audit(self, format=None, action=None, sandbox=None, since=None, until=None, sort=None):
        """Every matching audit entry of the last 90 days as one CSV or NDJSON file. `GET /v1/audit/export`"""
        return self._call("GET", f"/v1/audit/export", query={"format": format, "action": action, "sandbox": sandbox, "since": since, "until": until, "sort": sort}, raw_response=True)

    def get_billing(self):
        """Plan, rates, this period's usage, wallet and auto top-up. `GET /v1/billing`"""
        return self._call("GET", f"/v1/billing")

    def set_extra_starts(self, body):
        """Turn extra starts on or off: starts past the start limits paid from the wallet. `PUT /v1/billing/extra-starts`"""
        return self._call("PUT", f"/v1/billing/extra-starts", json_body=body)

    def cancel_plan(self, keep_disks=None):
        """Cancel the plan at the end of the period. `DELETE /v1/billing/plan`"""
        return self._call("DELETE", f"/v1/billing/plan", query={"keep_disks": keep_disks})

    def choose_plan(self, body):
        """Choose or change the plan (the first one returns a Stripe Checkout page). `POST /v1/billing/plan`"""
        return self._call("POST", f"/v1/billing/plan", json_body=body)

    def billing_portal(self):
        """A Stripe page to change the card and download invoices. `POST /v1/billing/portal`"""
        return self._call("POST", f"/v1/billing/portal")

    def end_trial(self):
        """End the free trial now and start the plan. `POST /v1/billing/trial/end`"""
        return self._call("POST", f"/v1/billing/trial/end")

    def set_auto_topup(self, body):
        """Set auto top-up (0 = off). `PUT /v1/billing/wallet/auto-topup`"""
        return self._call("PUT", f"/v1/billing/wallet/auto-topup", json_body=body)

    def wallet_pay(self, body):
        """Charge the saved card and add the money to the wallet. `POST /v1/billing/wallet/pay`"""
        return self._call("POST", f"/v1/billing/wallet/pay", json_body=body)

    def wallet_quote(self, amount_usd):
        """What a wallet payment adds, with its bonus. `GET /v1/billing/wallet/quote`"""
        return self._call("GET", f"/v1/billing/wallet/quote", query={"amount_usd": amount_usd})

    def get_ci_overview(self, repo=None, state=None, limit=None):
        """CI on froog: setup, 7-day stats and the last GitHub Actions jobs (runs-on: froog). `GET /v1/ci`"""
        return self._call("GET", f"/v1/ci", query={"repo": repo, "state": state, "limit": limit})

    def get_ci_analytics(self, days=None, repo=None):
        """CI analytics: per repository and job, runs, failure rate, p50/p90 duration and wait, cost, CPU and memory, a size advice, and flaky tests. `GET /v1/ci/analytics`"""
        return self._call("GET", f"/v1/ci/analytics", query={"days": days, "repo": repo})

    def get_ci_job(self, id):
        """One CI job: state, reason, wait, run time and sandbox. `GET /v1/ci/jobs/{id}`"""
        return self._call("GET", f"/v1/ci/jobs/{urllib.parse.quote(str(id), safe='')}")

    def get_ci_job_log(self, id):
        """One CI job's log from GitHub (text, the last 1 MiB). `GET /v1/ci/jobs/{id}/logs`"""
        return self._call("GET", f"/v1/ci/jobs/{urllib.parse.quote(str(id), safe='')}/logs", raw_response=True)

    def get_ci_job_metrics(self, id):
        """One CI job's CPU and memory over time, and its steps with their times. `GET /v1/ci/jobs/{id}/metrics`"""
        return self._call("GET", f"/v1/ci/jobs/{urllib.parse.quote(str(id), safe='')}/metrics")

    def get_ci_job_tests(self, id):
        """One CI job's test results, from the JUnit XML files it wrote (failed first). `GET /v1/ci/jobs/{id}/tests`"""
        return self._call("GET", f"/v1/ci/jobs/{urllib.parse.quote(str(id), safe='')}/tests")

    def search_ci_logs(self, q=None, days=None, limit=None, order=None):
        """Search the kept logs of finished CI jobs, line by line. `GET /v1/ci/logs/search`"""
        return self._call("GET", f"/v1/ci/logs/search", query={"q": q, "days": days, "limit": limit, "order": order})

    def list_ci_monitors(self):
        """CI monitors: alerts when matching jobs fail or run long, with their last alerts. `GET /v1/ci/monitors`"""
        return self._call("GET", f"/v1/ci/monitors")

    def create_ci_monitor(self, body):
        """Add a CI monitor (email and the ci.monitor.triggered webhook event). `POST /v1/ci/monitors`"""
        return self._call("POST", f"/v1/ci/monitors", json_body=body)

    def delete_ci_monitor(self, id):
        """Remove a CI monitor. `DELETE /v1/ci/monitors/{id}`"""
        return self._call("DELETE", f"/v1/ci/monitors/{urllib.parse.quote(str(id), safe='')}")

    def list_ci_runs(self, repo=None, state=None, workflow=None, branch=None, q=None, limit=None, offset=None):
        """CI workflow runs, newest first, a page at a time: each run's GitHub title, event, number, actor and its jobs on froog, plus the total and the workflows, repositories and branches to filter by. `GET /v1/ci/runs`"""
        return self._call("GET", f"/v1/ci/runs", query={"repo": repo, "state": state, "workflow": workflow, "branch": branch, "q": q, "limit": limit, "offset": offset})

    def get_ci_run(self, run_id):
        """One CI workflow run (GitHub's run id) and its jobs on froog. `GET /v1/ci/runs/{run_id}`"""
        return self._call("GET", f"/v1/ci/runs/{urllib.parse.quote(str(run_id), safe='')}")

    def set_ci_settings(self, body):
        """Let jobs from public repositories run, optionally only for listed GitHub users. `PUT /v1/ci/settings`"""
        return self._call("PUT", f"/v1/ci/settings", json_body=body)

    def start_cli_login(self, body):
        """Start a browser login for the CLI. `POST /v1/cli-login`"""
        return self._call("POST", f"/v1/cli-login", json_body=body, auth=False)

    def poll_cli_login(self, body):
        """Poll a CLI login (202 until approved). `POST /v1/cli-login/poll`"""
        return self._call("POST", f"/v1/cli-login/poll", json_body=body, auth=False)

    def deny_cli_login(self, code):
        """Deny a CLI login. `DELETE /v1/cli-login/{code}`"""
        return self._call("DELETE", f"/v1/cli-login/{urllib.parse.quote(str(code), safe='')}")

    def get_cli_login(self, code):
        """Describe a pending CLI login. `GET /v1/cli-login/{code}`"""
        return self._call("GET", f"/v1/cli-login/{urllib.parse.quote(str(code), safe='')}")

    def approve_cli_login(self, code):
        """Approve a CLI login (creates a key). `POST /v1/cli-login/{code}/approve`"""
        return self._call("POST", f"/v1/cli-login/{urllib.parse.quote(str(code), safe='')}/approve")

    def create_dashboard_link(self):
        """A link that opens the dashboard signed in: single use, 15 minutes (full-access keys and sessions only). `POST /v1/dashboard-link`"""
        return self._call("POST", f"/v1/dashboard-link")

    def get_deletion_operation(self, id):
        """A deletion operation (dop_…): an account close or a zero data retention erasure. `GET /v1/deletion-operations/{id}`"""
        return self._call("GET", f"/v1/deletion-operations/{urllib.parse.quote(str(id), safe='')}")

    def list_env(self):
        """Account environment variables (names and previews). `GET /v1/env`"""
        return self._call("GET", f"/v1/env")

    def remove_env(self, name):
        """Remove an account environment variable. `DELETE /v1/env/{name}`"""
        return self._call("DELETE", f"/v1/env/{urllib.parse.quote(str(name), safe='')}")

    def set_env(self, name, body):
        """Set an account environment variable. `PUT /v1/env/{name}`"""
        return self._call("PUT", f"/v1/env/{urllib.parse.quote(str(name), safe='')}", json_body=body)

    def get_environment(self):
        """Repos, secret files and setup script every new sandbox gets. `GET /v1/environment`"""
        return self._call("GET", f"/v1/environment")

    def update_environment(self, body):
        """Change repos or the setup script. `PATCH /v1/environment`"""
        return self._call("PATCH", f"/v1/environment", json_body=body)

    def remove_environment_file(self, path):
        """Remove a secret file. `DELETE /v1/environment/files`"""
        return self._call("DELETE", f"/v1/environment/files", query={"path": path})

    def set_environment_file(self, body):
        """Write a secret file. `PUT /v1/environment/files`"""
        return self._call("PUT", f"/v1/environment/files", json_body=body)

    def list_environments(self):
        """Named environments: default marker, latest version, sandboxes per version. `GET /v1/environments`"""
        return self._call("GET", f"/v1/environments")

    def create_environment(self, body):
        """Create an environment (empty, or a copy of another's latest version). `POST /v1/environments`"""
        return self._call("POST", f"/v1/environments", json_body=body)

    def delete_environment(self, env):
        """Delete an environment (its sandboxes keep their version; not the default). `DELETE /v1/environments/{env}`"""
        return self._call("DELETE", f"/v1/environments/{urllib.parse.quote(str(env), safe='')}")

    def get_named_environment(self, env):
        """One environment: repos, setup scripts, secret files (names and sizes), latest version. `GET /v1/environments/{env}`"""
        return self._call("GET", f"/v1/environments/{urllib.parse.quote(str(env), safe='')}")

    def update_named_environment(self, env, body):
        """Change repos or setup scripts (a new version), rename, or make it the default. `PATCH /v1/environments/{env}`"""
        return self._call("PATCH", f"/v1/environments/{urllib.parse.quote(str(env), safe='')}", json_body=body)

    def remove_named_environment_file(self, env, path):
        """Remove a secret file (a new version). `DELETE /v1/environments/{env}/files`"""
        return self._call("DELETE", f"/v1/environments/{urllib.parse.quote(str(env), safe='')}/files", query={"path": path})

    def set_named_environment_file(self, env, body):
        """Write a secret file (a new version). `PUT /v1/environments/{env}/files`"""
        return self._call("PUT", f"/v1/environments/{urllib.parse.quote(str(env), safe='')}/files", json_body=body)

    def upgrade_environment(self, env, body):
        """Move its sandboxes to the latest version (running ones now, stopped ones at their next start). `POST /v1/environments/{env}/upgrade`"""
        return self._call("POST", f"/v1/environments/{urllib.parse.quote(str(env), safe='')}/upgrade", json_body=body)

    def list_environment_versions(self, env):
        """An environment's versions, newest first, with sandboxes on each. `GET /v1/environments/{env}/versions`"""
        return self._call("GET", f"/v1/environments/{urllib.parse.quote(str(env), safe='')}/versions")

    def send_feedback(self, body):
        """Send feedback to the froog team. `POST /v1/feedback`"""
        return self._call("POST", f"/v1/feedback", json_body=body)

    def download_signed(self, token):
        """Download through a signed link. `GET /v1/file-urls/{token}`"""
        return self._call("GET", f"/v1/file-urls/{urllib.parse.quote(str(token), safe='')}", raw_response=True, auth=False)

    def upload_signed(self, token, data):
        """Upload through a signed link (raw bytes). `PUT /v1/file-urls/{token}`"""
        return self._call("PUT", f"/v1/file-urls/{urllib.parse.quote(str(token), safe='')}", raw=data, auth=False)

    def get_github(self):
        """The GitHub connection: linked installations and the repositories the sandbox token is limited to. `GET /v1/github`"""
        return self._call("GET", f"/v1/github")

    def connect_github(self):
        """A link that installs the froog GitHub App on a GitHub account or organization. `POST /v1/github/connect`"""
        return self._call("POST", f"/v1/github/connect")

    def disconnect_github_installation(self, id):
        """Unlink a GitHub installation from the account. `DELETE /v1/github/installations/{id}`"""
        return self._call("DELETE", f"/v1/github/installations/{urllib.parse.quote(str(id), safe='')}")

    def list_github_repos(self):
        """Repositories the linked installations can see. `GET /v1/github/repos`"""
        return self._call("GET", f"/v1/github/repos")

    def set_github_token_repos(self, body):
        """Limit the GitHub token in each sandbox to these repositories (empty: all the app can see). `PUT /v1/github/token-repos`"""
        return self._call("PUT", f"/v1/github/token-repos", json_body=body)

    def list_images(self):
        """Operating systems a sandbox can run. `GET /v1/images`"""
        return self._call("GET", f"/v1/images")

    def list_inboxes(self):
        """Your agent inboxes. `GET /v1/inboxes`"""
        return self._call("GET", f"/v1/inboxes")

    def create_inbox(self, body):
        """Make an inbox ($2 a month): an email address that receives mail for an agent. Uses a paid seat, or adds one; 402 inbox_payment_required without an inbox subscription. `POST /v1/inboxes`"""
        return self._call("POST", f"/v1/inboxes", json_body=body)

    def get_inbox_billing(self):
        """What you pay for inboxes: seats paid for, inboxes used, the subscription's status. `GET /v1/inboxes/billing`"""
        return self._call("GET", f"/v1/inboxes/billing")

    def buy_inboxes(self, body):
        """A Stripe Checkout page that buys inboxes (default: yours plus one). `POST /v1/inboxes/billing/checkout`"""
        return self._call("POST", f"/v1/inboxes/billing/checkout", json_body=body)

    def delete_inbox(self, inbox):
        """Delete an inbox and its messages (the address is never given out again; its seat stays paid until the period ends). `DELETE /v1/inboxes/{inbox}`"""
        return self._call("DELETE", f"/v1/inboxes/{urllib.parse.quote(str(inbox), safe='')}")

    def get_inbox(self, inbox):
        """One inbox (by id, address, or the part before @). `GET /v1/inboxes/{inbox}`"""
        return self._call("GET", f"/v1/inboxes/{urllib.parse.quote(str(inbox), safe='')}")

    def update_inbox(self, inbox, body):
        """Rename an inbox, or attach it to a sandbox ("" detaches it). `PATCH /v1/inboxes/{inbox}`"""
        return self._call("PATCH", f"/v1/inboxes/{urllib.parse.quote(str(inbox), safe='')}", json_body=body)

    def list_inbox_messages(self, inbox, since=None, limit=None):
        """Received messages, newest first, without their bodies. `GET /v1/inboxes/{inbox}/messages`"""
        return self._call("GET", f"/v1/inboxes/{urllib.parse.quote(str(inbox), safe='')}/messages", query={"since": since, "limit": limit})

    def delete_inbox_message(self, inbox, id):
        """Delete a message. `DELETE /v1/inboxes/{inbox}/messages/{id}`"""
        return self._call("DELETE", f"/v1/inboxes/{urllib.parse.quote(str(inbox), safe='')}/messages/{urllib.parse.quote(str(id), safe='')}")

    def get_inbox_message(self, inbox, id):
        """One message with its text, HTML, codes and links. `GET /v1/inboxes/{inbox}/messages/{id}`"""
        return self._call("GET", f"/v1/inboxes/{urllib.parse.quote(str(inbox), safe='')}/messages/{urllib.parse.quote(str(id), safe='')}")

    def get_inbox_message_raw(self, inbox, id):
        """The message as received (message/rfc822). `GET /v1/inboxes/{inbox}/messages/{id}/raw`"""
        return self._call("GET", f"/v1/inboxes/{urllib.parse.quote(str(inbox), safe='')}/messages/{urllib.parse.quote(str(id), safe='')}/raw", raw_response=True)

    def wait_for_inbox_message(self, inbox, since=None, after=None, sender=None, subject=None, timeout=None):
        """Wait for the next message (its one-time codes and verification links already pulled out); message is null on timeout. `GET /v1/inboxes/{inbox}/wait`"""
        return self._call("GET", f"/v1/inboxes/{urllib.parse.quote(str(inbox), safe='')}/wait", query={"since": since, "after": after, "sender": sender, "subject": subject, "timeout": timeout})

    def get_limits(self):
        """What the account can run now: sandboxes at once, sizes, trial, start limits. `GET /v1/limits`"""
        return self._call("GET", f"/v1/limits")

    def provider_models(self):
        """The harnesses, their models, effort levels and fast mode. `GET /v1/provider-models`"""
        return self._call("GET", f"/v1/provider-models", auth=False)

    def list_sandboxes(self, label=None, name=None, name_prefix=None, state=None, sort=None, limit=None, cursor=None):
        """Sandboxes, oldest first, filtered and paged. `GET /v1/sandboxes`"""
        return self._call("GET", f"/v1/sandboxes", query={"label": label, "name": name, "name_prefix": name_prefix, "state": state, "sort": sort, "limit": limit, "cursor": cursor})

    def create_sandboxes(self, body, idempotency_key=None):
        """Create sandboxes (202; poll until running). `POST /v1/sandboxes`"""
        return self._call("POST", f"/v1/sandboxes", json_body=body, idempotency_key=idempotency_key)

    def delete_sandbox(self, name):
        """Delete a sandbox (saved snapshots stay). `DELETE /v1/sandboxes/{name}`"""
        return self._call("DELETE", f"/v1/sandboxes/{urllib.parse.quote(str(name), safe='')}")

    def get_sandbox(self, name):
        """One sandbox. `GET /v1/sandboxes/{name}`"""
        return self._call("GET", f"/v1/sandboxes/{urllib.parse.quote(str(name), safe='')}")

    def update_sandbox(self, name, body):
        """Rename; change the auto-stop timer (counting from now while running), lifetime, auto-delete, auto-start or labels. `PATCH /v1/sandboxes/{name}`"""
        return self._call("PATCH", f"/v1/sandboxes/{urllib.parse.quote(str(name), safe='')}", json_body=body)

    def get_sandbox_ai(self, name, days=None):
        """A sandbox's AI profile, proxy state, block, and usage per day and model. `GET /v1/sandboxes/{name}/ai`"""
        return self._call("GET", f"/v1/sandboxes/{urllib.parse.quote(str(name), safe='')}/ai", query={"days": days})

    def update_sandbox_ai(self, name, body):
        """Block or unblock a sandbox's AI requests, or give it another profile. `PUT /v1/sandboxes/{name}/ai`"""
        return self._call("PUT", f"/v1/sandboxes/{urllib.parse.quote(str(name), safe='')}/ai", json_body=body)

    def get_browser_link(self, name, body, theme=None):
        """A link to the browser view: only Chrome, 1920x1080. `POST /v1/sandboxes/{name}/browser`"""
        return self._call("POST", f"/v1/sandboxes/{urllib.parse.quote(str(name), safe='')}/browser", json_body=body, query={"theme": theme})

    def list_commands(self, name, mode=None, sort=None, limit=None, cursor=None):
        """Commands run in the sandbox, newest first (90 days, also while stopped). `GET /v1/sandboxes/{name}/command-history`"""
        return self._call("GET", f"/v1/sandboxes/{urllib.parse.quote(str(name), safe='')}/command-history", query={"mode": mode, "sort": sort, "limit": limit, "cursor": cursor})

    def get_command(self, name, id):
        """One command's record. `GET /v1/sandboxes/{name}/command-history/{id}`"""
        return self._call("GET", f"/v1/sandboxes/{urllib.parse.quote(str(name), safe='')}/command-history/{urllib.parse.quote(str(id), safe='')}")

    def read_command_logs(self, name, id, stream=None, offset=None, encoding=None):
        """A command's stored output from an offset (at most 1 MiB). `GET /v1/sandboxes/{name}/command-history/{id}/logs`"""
        return self._call("GET", f"/v1/sandboxes/{urllib.parse.quote(str(name), safe='')}/command-history/{urllib.parse.quote(str(id), safe='')}/logs", query={"stream": stream, "offset": offset, "encoding": encoding})

    def list_conversations(self, name):
        """Conversations, most recent first. `GET /v1/sandboxes/{name}/conversations`"""
        return self._call("GET", f"/v1/sandboxes/{urllib.parse.quote(str(name), safe='')}/conversations")

    def delete_conversation(self, name, id):
        """Delete a conversation and its history. `DELETE /v1/sandboxes/{name}/conversations/{id}`"""
        return self._call("DELETE", f"/v1/sandboxes/{urllib.parse.quote(str(name), safe='')}/conversations/{urllib.parse.quote(str(id), safe='')}")

    def get_conversation(self, name, id):
        """A conversation and its turns (id or current). `GET /v1/sandboxes/{name}/conversations/{id}`"""
        return self._call("GET", f"/v1/sandboxes/{urllib.parse.quote(str(name), safe='')}/conversations/{urllib.parse.quote(str(id), safe='')}")

    def get_desktop_link(self, name, theme=None):
        """A link to the desktop (1920x1080) in froog's viewer. `POST /v1/sandboxes/{name}/desktop`"""
        return self._call("POST", f"/v1/sandboxes/{urllib.parse.quote(str(name), safe='')}/desktop", query={"theme": theme})

    def list_events(self, name, conversation=None, cursor=None, limit=None):
        """Every conversation's events: a JSON page, or a stream (Accept: text/event-stream or application/x-ndjson). `GET /v1/sandboxes/{name}/events`"""
        return self._call("GET", f"/v1/sandboxes/{urllib.parse.quote(str(name), safe='')}/events", query={"conversation": conversation, "cursor": cursor, "limit": limit})

    def exec(self, name, body):
        """Run a command and wait for it; with stream: true, NDJSON events as output arrives. `POST /v1/sandboxes/{name}/exec`"""
        return self._call("POST", f"/v1/sandboxes/{urllib.parse.quote(str(name), safe='')}/exec", json_body=body)

    def delete_file(self, name, path, recursive=None):
        """Delete a file or directory. `DELETE /v1/sandboxes/{name}/files`"""
        return self._call("DELETE", f"/v1/sandboxes/{urllib.parse.quote(str(name), safe='')}/files", query={"path": path, "recursive": recursive})

    def file_info(self, name, path):
        """A file's details or a directory listing. `GET /v1/sandboxes/{name}/files`"""
        return self._call("GET", f"/v1/sandboxes/{urllib.parse.quote(str(name), safe='')}/files", query={"path": path})

    def upload_archive(self, name, data, path):
        """Unpack a .tar.gz into a directory (raw bytes, up to 100 MiB). `PUT /v1/sandboxes/{name}/files/archive`"""
        return self._call("PUT", f"/v1/sandboxes/{urllib.parse.quote(str(name), safe='')}/files/archive", raw=data, query={"path": path})

    def read_file(self, name, path):
        """Download a file (raw bytes, up to 100 MiB). `GET /v1/sandboxes/{name}/files/content`"""
        return self._call("GET", f"/v1/sandboxes/{urllib.parse.quote(str(name), safe='')}/files/content", query={"path": path}, raw_response=True)

    def write_file(self, name, data, path, mode=None):
        """Upload a file (raw bytes, up to 100 MiB, atomic). `PUT /v1/sandboxes/{name}/files/content`"""
        return self._call("PUT", f"/v1/sandboxes/{urllib.parse.quote(str(name), safe='')}/files/content", raw=data, query={"path": path, "mode": mode})

    def create_download_url(self, name, body):
        """A link that downloads a file without an API key. `POST /v1/sandboxes/{name}/files/download-url`"""
        return self._call("POST", f"/v1/sandboxes/{urllib.parse.quote(str(name), safe='')}/files/download-url", json_body=body)

    def find_in_files(self, name, pattern, path, regex=None):
        """Lines that contain a text. `GET /v1/sandboxes/{name}/files/find`"""
        return self._call("GET", f"/v1/sandboxes/{urllib.parse.quote(str(name), safe='')}/files/find", query={"pattern": pattern, "path": path, "regex": regex})

    def make_directory(self, name, body):
        """Create a directory (and parents). `POST /v1/sandboxes/{name}/files/mkdir`"""
        return self._call("POST", f"/v1/sandboxes/{urllib.parse.quote(str(name), safe='')}/files/mkdir", json_body=body)

    def move_file(self, name, body):
        """Move or rename. `POST /v1/sandboxes/{name}/files/move`"""
        return self._call("POST", f"/v1/sandboxes/{urllib.parse.quote(str(name), safe='')}/files/move", json_body=body)

    def set_file_permissions(self, name, body):
        """Change a file's or directory's permissions (chmod). `POST /v1/sandboxes/{name}/files/permissions`"""
        return self._call("POST", f"/v1/sandboxes/{urllib.parse.quote(str(name), safe='')}/files/permissions", json_body=body)

    def replace_in_files(self, name, body):
        """Replace a text in files, byte for byte. `POST /v1/sandboxes/{name}/files/replace`"""
        return self._call("POST", f"/v1/sandboxes/{urllib.parse.quote(str(name), safe='')}/files/replace", json_body=body)

    def search_files(self, name, pattern, path):
        """Files whose name matches a glob. `GET /v1/sandboxes/{name}/files/search`"""
        return self._call("GET", f"/v1/sandboxes/{urllib.parse.quote(str(name), safe='')}/files/search", query={"pattern": pattern, "path": path})

    def create_upload_url(self, name, body):
        """A link that takes a PUT of a file without an API key. `POST /v1/sandboxes/{name}/files/upload-url`"""
        return self._call("POST", f"/v1/sandboxes/{urllib.parse.quote(str(name), safe='')}/files/upload-url", json_body=body)

    def watch_files(self, name, path, recursive=None, timeout_secs=None):
        """Stream file changes under a path (NDJSON: ready, then create / modify / delete, then end). `GET /v1/sandboxes/{name}/files/watch`: a generator of event dicts."""
        return self._stream(f"/v1/sandboxes/{urllib.parse.quote(str(name), safe='')}/files/watch", query={"path": path, "recursive": recursive, "timeout_secs": timeout_secs})

    def fork_sandbox(self, name, body, idempotency_key=None):
        """New sandboxes from a snapshot of this one. `POST /v1/sandboxes/{name}/fork`"""
        return self._call("POST", f"/v1/sandboxes/{urllib.parse.quote(str(name), safe='')}/fork", json_body=body, idempotency_key=idempotency_key)

    def get_sandbox_github_token(self, name, body):
        """A one-hour GitHub token for the sandbox (what its git credential helper uses; needs the environment). `POST /v1/sandboxes/{name}/github-token`"""
        return self._call("POST", f"/v1/sandboxes/{urllib.parse.quote(str(name), safe='')}/github-token", json_body=body)

    def interrupt(self, name, body):
        """Stop running turns and cancel queued prompts (one conversation, or all). `POST /v1/sandboxes/{name}/interrupt`"""
        return self._call("POST", f"/v1/sandboxes/{urllib.parse.quote(str(name), safe='')}/interrupt", json_body=body)

    def list_jobs(self, name):
        """Background commands. `GET /v1/sandboxes/{name}/jobs`"""
        return self._call("GET", f"/v1/sandboxes/{urllib.parse.quote(str(name), safe='')}/jobs")

    def start_job(self, name, body):
        """Start a background command. `POST /v1/sandboxes/{name}/jobs`"""
        return self._call("POST", f"/v1/sandboxes/{urllib.parse.quote(str(name), safe='')}/jobs", json_body=body)

    def read_job(self, name, id, offset=None, limit=None):
        """A background command's state and output from an offset. `GET /v1/sandboxes/{name}/jobs/{id}`"""
        return self._call("GET", f"/v1/sandboxes/{urllib.parse.quote(str(name), safe='')}/jobs/{urllib.parse.quote(str(id), safe='')}", query={"offset": offset, "limit": limit})

    def write_job_stdin(self, name, id, body):
        """Write to a background command's open standard input (started with stdin_open), or close it. `POST /v1/sandboxes/{name}/jobs/{id}/stdin`"""
        return self._call("POST", f"/v1/sandboxes/{urllib.parse.quote(str(name), safe='')}/jobs/{urllib.parse.quote(str(id), safe='')}/stdin", json_body=body)

    def stop_job(self, name, id):
        """Stop a background command. `POST /v1/sandboxes/{name}/jobs/{id}/stop`"""
        return self._call("POST", f"/v1/sandboxes/{urllib.parse.quote(str(name), safe='')}/jobs/{urllib.parse.quote(str(id), safe='')}/stop")

    def get_sandbox_metrics(self, name):
        """CPU, memory, disk and network of a running sandbox, read now. `GET /v1/sandboxes/{name}/metrics`"""
        return self._call("GET", f"/v1/sandboxes/{urllib.parse.quote(str(name), safe='')}/metrics")

    def get_sandbox_metrics_history(self, name, since=None, until=None, step=None):
        """CPU, memory and disk every 30 seconds, kept 7 days, averaged per step. `GET /v1/sandboxes/{name}/metrics/history`"""
        return self._call("GET", f"/v1/sandboxes/{urllib.parse.quote(str(name), safe='')}/metrics/history", query={"since": since, "until": until, "step": step})

    def list_ports(self, name):
        """Inbound ports open on the sandbox (besides SSH), including its public URLs' ports, and the host IPv4 address and port each opened one is reached at. `GET /v1/sandboxes/{name}/ports`"""
        return self._call("GET", f"/v1/sandboxes/{urllib.parse.quote(str(name), safe='')}/ports")

    def open_port(self, name, body):
        """Open an inbound TCP or UDP port: on the sandbox's IPv6 address, and on a public port of its host's IPv4 address. `POST /v1/sandboxes/{name}/ports`"""
        return self._call("POST", f"/v1/sandboxes/{urllib.parse.quote(str(name), safe='')}/ports", json_body=body)

    def set_ports(self, name, body):
        """Set the opened ports, replacing the ones opened before (kept across stops and starts). `PUT /v1/sandboxes/{name}/ports`"""
        return self._call("PUT", f"/v1/sandboxes/{urllib.parse.quote(str(name), safe='')}/ports", json_body=body)

    def watch_ports(self, name, timeout_secs=None):
        """Stream TCP ports starting or stopping to listen (NDJSON: ready, open / close, end). `GET /v1/sandboxes/{name}/ports/watch`: a generator of event dicts."""
        return self._stream(f"/v1/sandboxes/{urllib.parse.quote(str(name), safe='')}/ports/watch", query={"timeout_secs": timeout_secs})

    def close_port(self, name, port, protocol):
        """Close an opened port (IPv6 and IPv4). `DELETE /v1/sandboxes/{name}/ports/{port}/{protocol}`"""
        return self._call("DELETE", f"/v1/sandboxes/{urllib.parse.quote(str(name), safe='')}/ports/{urllib.parse.quote(str(port), safe='')}/{urllib.parse.quote(str(protocol), safe='')}")

    def list_processes(self, name):
        """Processes, busiest first (a stopped sandbox: the sample taken at stop). `GET /v1/sandboxes/{name}/processes`"""
        return self._call("GET", f"/v1/sandboxes/{urllib.parse.quote(str(name), safe='')}/processes")

    def list_process_records(self, name, since=None, until=None, running=None, limit=None):
        """Processes seen in the 30-second samples, with start and end times, kept 72 hours. `GET /v1/sandboxes/{name}/processes/records`"""
        return self._call("GET", f"/v1/sandboxes/{urllib.parse.quote(str(name), safe='')}/processes/records", query={"since": since, "until": until, "running": running, "limit": limit})

    def list_process_samples(self, name, since=None, until=None, limit=None):
        """The 12 busiest processes every 30 seconds, kept 7 days, newest first. `GET /v1/sandboxes/{name}/processes/samples`"""
        return self._call("GET", f"/v1/sandboxes/{urllib.parse.quote(str(name), safe='')}/processes/samples", query={"since": since, "until": until, "limit": limit})

    def prompt(self, name, body):
        """Prompt a coding agent and wait for its answer (continues the current conversation). `POST /v1/sandboxes/{name}/prompt`"""
        return self._call("POST", f"/v1/sandboxes/{urllib.parse.quote(str(name), safe='')}/prompt", json_body=body)

    def list_prompts(self, name, conversation=None, limit=None):
        """Prompts, newest first. `GET /v1/sandboxes/{name}/prompts`"""
        return self._call("GET", f"/v1/sandboxes/{urllib.parse.quote(str(name), safe='')}/prompts", query={"conversation": conversation, "limit": limit})

    def queue_prompt(self, name, body):
        """Queue a prompt for a coding agent (returns at once; follow its events). `POST /v1/sandboxes/{name}/prompts`"""
        return self._call("POST", f"/v1/sandboxes/{urllib.parse.quote(str(name), safe='')}/prompts", json_body=body)

    def get_prompt(self, name, id):
        """One prompt: status (queued, running, done, failed, interrupted), result, error. `GET /v1/sandboxes/{name}/prompts/{id}`"""
        return self._call("GET", f"/v1/sandboxes/{urllib.parse.quote(str(name), safe='')}/prompts/{urllib.parse.quote(str(id), safe='')}")

    def prompt_events(self, name, id, cursor=None, limit=None):
        """A prompt's events: a JSON page, or a stream until it ends (Accept: text/event-stream or application/x-ndjson). `GET /v1/sandboxes/{name}/prompts/{id}/events`"""
        return self._call("GET", f"/v1/sandboxes/{urllib.parse.quote(str(name), safe='')}/prompts/{urllib.parse.quote(str(id), safe='')}/events", query={"cursor": cursor, "limit": limit})

    def sandbox_provider_models(self, name):
        """The model catalog plus the models the harnesses list in this sandbox. `GET /v1/sandboxes/{name}/provider-models`"""
        return self._call("GET", f"/v1/sandboxes/{urllib.parse.quote(str(name), safe='')}/provider-models")

    def restore_sandbox(self, name, body):
        """Roll the sandbox back to a snapshot. `POST /v1/sandboxes/{name}/restore`"""
        return self._call("POST", f"/v1/sandboxes/{urllib.parse.quote(str(name), safe='')}/restore", json_body=body)

    def delete_auto_snapshots(self, name):
        """Delete every automatic snapshot (saved ones and the one a stopped sandbox starts from stay). `DELETE /v1/sandboxes/{name}/snapshots`"""
        return self._call("DELETE", f"/v1/sandboxes/{urllib.parse.quote(str(name), safe='')}/snapshots")

    def list_snapshots(self, name, sort=None, limit=None, cursor=None):
        """A sandbox's snapshots, oldest first, paged. `GET /v1/sandboxes/{name}/snapshots`"""
        return self._call("GET", f"/v1/sandboxes/{urllib.parse.quote(str(name), safe='')}/snapshots", query={"sort": sort, "limit": limit, "cursor": cursor})

    def save_snapshot(self, name, body):
        """Save a snapshot (with a label: an account template). `POST /v1/sandboxes/{name}/snapshots`"""
        return self._call("POST", f"/v1/sandboxes/{urllib.parse.quote(str(name), safe='')}/snapshots", json_body=body)

    def delete_snapshot(self, name, snap):
        """Delete a snapshot. `DELETE /v1/sandboxes/{name}/snapshots/{snap}`"""
        return self._call("DELETE", f"/v1/sandboxes/{urllib.parse.quote(str(name), safe='')}/snapshots/{urllib.parse.quote(str(snap), safe='')}")

    def get_snapshot(self, name, snap):
        """One snapshot by id or label; latest is the newest available one. `GET /v1/sandboxes/{name}/snapshots/{snap}`"""
        return self._call("GET", f"/v1/sandboxes/{urllib.parse.quote(str(name), safe='')}/snapshots/{urllib.parse.quote(str(snap), safe='')}")

    def snapshot_file_info(self, name, snap, path):
        """A file's details, or a folder's entries, inside a snapshot (the sandbox may be stopped). `GET /v1/sandboxes/{name}/snapshots/{snap}/files`"""
        return self._call("GET", f"/v1/sandboxes/{urllib.parse.quote(str(name), safe='')}/snapshots/{urllib.parse.quote(str(snap), safe='')}/files", query={"path": path})

    def download_from_snapshot(self, name, snap, path):
        """Download a file (raw bytes) or a folder (tar) out of a snapshot. `GET /v1/sandboxes/{name}/snapshots/{snap}/files/download`"""
        return self._call("GET", f"/v1/sandboxes/{urllib.parse.quote(str(name), safe='')}/snapshots/{urllib.parse.quote(str(snap), safe='')}/files/download", query={"path": path}, raw_response=True)

    def get_ssh_info(self, name):
        """Where and how to SSH in (host, port, pinned host key). `GET /v1/sandboxes/{name}/ssh`"""
        return self._call("GET", f"/v1/sandboxes/{urllib.parse.quote(str(name), safe='')}/ssh")

    def start_sandbox(self, name, body=None):
        """Start a stopped sandbox (its auto-stop timer starts again), optionally with a new size, auto-stop timer, environment or env vars. `POST /v1/sandboxes/{name}/start`"""
        return self._call("POST", f"/v1/sandboxes/{urllib.parse.quote(str(name), safe='')}/start", json_body=body)

    def steer(self, name, body):
        """Send a message to a running turn (Claude Code takes it mid-turn; others get it as the next turn). `POST /v1/sandboxes/{name}/steer`"""
        return self._call("POST", f"/v1/sandboxes/{urllib.parse.quote(str(name), safe='')}/steer", json_body=body)

    def stop_sandbox(self, name, body=None):
        """Stop: run ~/.froog/pre-stop, snapshot the disk, stop billing (refused and retried if the snapshot fails; force keeps the last good snapshot; the answer's stop_id tracks it). `POST /v1/sandboxes/{name}/stop`"""
        return self._call("POST", f"/v1/sandboxes/{urllib.parse.quote(str(name), safe='')}/stop", json_body=body)

    def touch_sandbox(self, name):
        """Record activity (last_active_at); the auto-stop deadline doesn't move (PATCH ttl_secs does). `POST /v1/sandboxes/{name}/touch`"""
        return self._call("POST", f"/v1/sandboxes/{urllib.parse.quote(str(name), safe='')}/touch")

    def list_urls(self, name):
        """Public URLs of the sandbox's ports. `GET /v1/sandboxes/{name}/urls`"""
        return self._call("GET", f"/v1/sandboxes/{urllib.parse.quote(str(name), safe='')}/urls")

    def host_port(self, name, body):
        """Host a port (private with a token unless public), or rename or title it. `POST /v1/sandboxes/{name}/urls`"""
        return self._call("POST", f"/v1/sandboxes/{urllib.parse.quote(str(name), safe='')}/urls", json_body=body)

    def remove_url(self, name, port):
        """Remove a public URL. `DELETE /v1/sandboxes/{name}/urls/{port}`"""
        return self._call("DELETE", f"/v1/sandboxes/{urllib.parse.quote(str(name), safe='')}/urls/{urllib.parse.quote(str(port), safe='')}")

    def list_url_tokens(self, name, port):
        """Live viewer tokens of a private URL (secrets are never returned). `GET /v1/sandboxes/{name}/urls/{port}/tokens`"""
        return self._call("GET", f"/v1/sandboxes/{urllib.parse.quote(str(name), safe='')}/urls/{urllib.parse.quote(str(port), safe='')}/tokens")

    def create_url_token(self, name, port, body):
        """Another token for a private URL that expires and can be revoked on its own (the secret is returned once). `POST /v1/sandboxes/{name}/urls/{port}/tokens`"""
        return self._call("POST", f"/v1/sandboxes/{urllib.parse.quote(str(name), safe='')}/urls/{urllib.parse.quote(str(port), safe='')}/tokens", json_body=body)

    def revoke_url_token(self, name, port, id):
        """Revoke a viewer token. `DELETE /v1/sandboxes/{name}/urls/{port}/tokens/{id}`"""
        return self._call("DELETE", f"/v1/sandboxes/{urllib.parse.quote(str(name), safe='')}/urls/{urllib.parse.quote(str(port), safe='')}/tokens/{urllib.parse.quote(str(id), safe='')}")

    def get_sandbox_usage(self, name, since=None, until=None):
        """One sandbox's running time and cost in a window, at list prices. `GET /v1/sandboxes/{name}/usage`"""
        return self._call("GET", f"/v1/sandboxes/{urllib.parse.quote(str(name), safe='')}/usage", query={"since": since, "until": until})

    def wait_for_port(self, name, body):
        """Wait until something listens on a TCP port inside the sandbox (ready: false after the timeout). `POST /v1/sandboxes/{name}/wait`"""
        return self._call("POST", f"/v1/sandboxes/{urllib.parse.quote(str(name), safe='')}/wait", json_body=body)

    def list_saved_snapshots(self):
        """Saved snapshots of the account (templates for from). `GET /v1/saved-snapshots`"""
        return self._call("GET", f"/v1/saved-snapshots")

    def delete_saved_snapshot(self, name):
        """Remove a saved snapshot. `DELETE /v1/saved-snapshots/{name}`"""
        return self._call("DELETE", f"/v1/saved-snapshots/{urllib.parse.quote(str(name), safe='')}")

    def get_saved_snapshot(self, name):
        """One saved snapshot and its status. `GET /v1/saved-snapshots/{name}`"""
        return self._call("GET", f"/v1/saved-snapshots/{urllib.parse.quote(str(name), safe='')}")

    def saved_snapshot_file_info(self, name, path):
        """A file's details, or a folder's entries, inside a saved snapshot. `GET /v1/saved-snapshots/{name}/files`"""
        return self._call("GET", f"/v1/saved-snapshots/{urllib.parse.quote(str(name), safe='')}/files", query={"path": path})

    def download_from_saved_snapshot(self, name, path):
        """Download a file (raw bytes) or a folder (tar) out of a saved snapshot. `GET /v1/saved-snapshots/{name}/files/download`"""
        return self._call("GET", f"/v1/saved-snapshots/{urllib.parse.quote(str(name), safe='')}/files/download", query={"path": path}, raw_response=True)

    def list_snapshot_shares(self, name):
        """A saved snapshot's share codes (not the codes themselves). `GET /v1/saved-snapshots/{name}/shares`"""
        return self._call("GET", f"/v1/saved-snapshots/{urllib.parse.quote(str(name), safe='')}/shares")

    def share_saved_snapshot(self, name):
        """Make a share code another account can create sandboxes from (shown once). `POST /v1/saved-snapshots/{name}/shares`"""
        return self._call("POST", f"/v1/saved-snapshots/{urllib.parse.quote(str(name), safe='')}/shares")

    def revoke_snapshot_share(self, name, id):
        """Revoke a share code. `DELETE /v1/saved-snapshots/{name}/shares/{id}`"""
        return self._call("DELETE", f"/v1/saved-snapshots/{urllib.parse.quote(str(name), safe='')}/shares/{urllib.parse.quote(str(id), safe='')}")

    def list_account_snapshots(self, sort=None, limit=None, cursor=None):
        """Every snapshot of the account, newest first, with its sandbox, paged. `GET /v1/snapshots`"""
        return self._call("GET", f"/v1/snapshots", query={"sort": sort, "limit": limit, "cursor": cursor})

    def list_ssh_keys(self):
        """SSH public keys that can log in to your sandboxes. `GET /v1/ssh-keys`"""
        return self._call("GET", f"/v1/ssh-keys")

    def add_ssh_key(self, body):
        """Add an SSH public key. `POST /v1/ssh-keys`"""
        return self._call("POST", f"/v1/ssh-keys", json_body=body)

    def remove_ssh_key(self, id):
        """Remove an SSH key. `DELETE /v1/ssh-keys/{id}`"""
        return self._call("DELETE", f"/v1/ssh-keys/{urllib.parse.quote(str(id), safe='')}")

    def get_status(self):
        """Platform status and 90-day uptime. `GET /v1/status`"""
        return self._call("GET", f"/v1/status", auth=False)

    def get_stop(self, id):
        """A stop's progress: pending, failing (retried), completed, superseded or failed. `GET /v1/stops/{id}`"""
        return self._call("GET", f"/v1/stops/{urllib.parse.quote(str(id), safe='')}")

    def list_subscriptions(self):
        """Subscription sign-ins (Claude, ChatGPT): connected or not. `GET /v1/subscriptions`"""
        return self._call("GET", f"/v1/subscriptions")

    def sign_out_subscription(self, provider):
        """Sign out of a subscription (every sandbox loses it at its next start). `DELETE /v1/subscriptions/{provider}`"""
        return self._call("DELETE", f"/v1/subscriptions/{urllib.parse.quote(str(provider), safe='')}")

    def cancel_subscription_login(self, provider):
        """Cancel the sign-in in progress. `DELETE /v1/subscriptions/{provider}/login`"""
        return self._call("DELETE", f"/v1/subscriptions/{urllib.parse.quote(str(provider), safe='')}/login")

    def get_subscription_login(self, provider):
        """The sign-in in progress (stores the credential once it's done). `GET /v1/subscriptions/{provider}/login`"""
        return self._call("GET", f"/v1/subscriptions/{urllib.parse.quote(str(provider), safe='')}/login")

    def start_subscription_login(self, provider, body):
        """Start a subscription sign-in in one of your running sandboxes. `POST /v1/subscriptions/{provider}/login`"""
        return self._call("POST", f"/v1/subscriptions/{urllib.parse.quote(str(provider), safe='')}/login", json_body=body)

    def send_subscription_code(self, provider, body):
        """Claude: paste the code the sign-in page showed. `POST /v1/subscriptions/{provider}/login/code`"""
        return self._call("POST", f"/v1/subscriptions/{urllib.parse.quote(str(provider), safe='')}/login/code", json_body=body)

    def remove_telemetry(self):
        """Stop the OpenTelemetry export. `DELETE /v1/telemetry`"""
        return self._call("DELETE", f"/v1/telemetry")

    def get_telemetry(self):
        """The OpenTelemetry endpoint (null when none). `GET /v1/telemetry`"""
        return self._call("GET", f"/v1/telemetry")

    def set_telemetry(self, body):
        """Set the OpenTelemetry endpoint; sends a test export. `PUT /v1/telemetry`"""
        return self._call("PUT", f"/v1/telemetry", json_body=body)

    def get_usage(self, month=None):
        """Usage this billing period: per-second running time, snapshots, what the plan covers. `GET /v1/usage`"""
        return self._call("GET", f"/v1/usage", query={"month": month})

    def join_waitlist(self, body):
        """Join the waitlist. `POST /v1/waitlist`"""
        return self._call("POST", f"/v1/waitlist", json_body=body, auth=False)

    def list_webhooks(self):
        """Webhooks. `GET /v1/webhooks`"""
        return self._call("GET", f"/v1/webhooks")

    def add_webhook(self, body):
        """Add a webhook (the signing secret is returned once). `POST /v1/webhooks`"""
        return self._call("POST", f"/v1/webhooks", json_body=body)

    def remove_webhook(self, id):
        """Remove a webhook. `DELETE /v1/webhooks/{id}`"""
        return self._call("DELETE", f"/v1/webhooks/{urllib.parse.quote(str(id), safe='')}")

    def get_webhook(self, id):
        """One webhook. `GET /v1/webhooks/{id}`"""
        return self._call("GET", f"/v1/webhooks/{urllib.parse.quote(str(id), safe='')}")

    def update_webhook(self, id, body):
        """Change a webhook's URL, events, name, or pause it (the secret stays). `PATCH /v1/webhooks/{id}`"""
        return self._call("PATCH", f"/v1/webhooks/{urllib.parse.quote(str(id), safe='')}", json_body=body)

    def list_webhook_deliveries(self, id, limit=None):
        """Delivery attempts of the last 30 days, newest first. `GET /v1/webhooks/{id}/deliveries`"""
        return self._call("GET", f"/v1/webhooks/{urllib.parse.quote(str(id), safe='')}/deliveries", query={"limit": limit})

    def rotate_webhook(self, id):
        """A new signing secret. `POST /v1/webhooks/{id}/rotate`"""
        return self._call("POST", f"/v1/webhooks/{urllib.parse.quote(str(id), safe='')}/rotate")

    def test_webhook(self, id):
        """Send a test event now. `POST /v1/webhooks/{id}/test`"""
        return self._call("POST", f"/v1/webhooks/{urllib.parse.quote(str(id), safe='')}/test")


# ---------------------------------------------------------------- objects
# Hand-written in deploy/sdk/objects.py.in, appended by deploy/build-sdks.py.


def _client(client):
    if client is None:
        return Froog()
    return getattr(client, "sync", client)  # an AsyncFroog wraps a Froog


def _exec_body(command, cwd=None, env=None, stdin=None, timeout=None):
    body = {"command": command}
    if cwd is not None:
        body["cwd"] = cwd
    if env:
        body["env"] = dict(env)
    if isinstance(stdin, (bytes, bytearray)):
        body["stdin_b64"] = base64.b64encode(bytes(stdin)).decode()
    elif stdin is not None:
        body["stdin"] = stdin
    if timeout is not None:
        body["timeout_secs"] = int(timeout)
    return body


class CommandResult:
    """What `Sandbox.exec` returns: `stdout`, `stderr`, `exit_code`, `timed_out`, `command_id` (also `r["stdout"]`)."""

    def __init__(self, data, command=None):
        self.data = data
        self.command = command
        self.stdout = data.get("stdout", "")
        self.stderr = data.get("stderr", "")
        self.exit_code = data.get("exit_code")
        self.timed_out = bool(data.get("timed_out", False))
        self.command_id = data.get("command_id")

    @property
    def ok(self):
        return self.exit_code == 0

    def __getitem__(self, key):
        return self.data[key]

    def get(self, key, default=None):
        return self.data.get(key, default)

    def __repr__(self):
        return f"CommandResult(exit_code={self.exit_code!r}, stdout={self.stdout[:60]!r}, stderr={self.stderr[:60]!r})"


class CommandError(FroogError):
    """`exec(..., check=True)` and `Process.wait(check=True)` raise this when the command exits non-zero."""

    def __init__(self, command, exit_code, stdout="", stderr="", result=None):
        super().__init__(None, "command_failed", f"{command!r} exited with {exit_code}: {(stderr or stdout).strip()[-500:]}")
        self.command, self.exit_code, self.stdout, self.stderr, self.result = command, exit_code, stdout, stderr, result


class Sandboxes:
    """`froog.sandboxes`: create, get and list sandboxes as `Sandbox` objects."""

    def __init__(self, client):
        self.client = client

    def create(self, name=None, **kwargs):
        """Create a sandbox and wait until it runs. Keyword arguments are `CreateSandbox` fields (size, os, env, labels,
        from_=..., ...), plus wait=True, timeout=600, keep=False, idempotency_key=None."""
        return Sandbox.create(name, client=self.client, **kwargs)

    def get(self, name, keep=True):
        return Sandbox.get(name, client=self.client, keep=keep)

    def list(self, **filters):
        """`label` (a string, or a list: every one must match), `name`, `name_prefix`, `state`, `sort`, `limit`, `cursor` as in `GET /v1/sandboxes`."""
        return Sandbox.list(client=self.client, **filters)


class Sandbox:
    """A sandbox you act on by method: `sb.exec(...)`, `sb.files.read(...)`, `sb.spawn(...)`, `sb.stop()`.

        with Sandbox.create(size="small") as sb:    # deleted at the end of the block unless keep=True
            print(sb.exec("uname -a").stdout)

    `Sandbox(name)` makes a handle without a request; `Sandbox.get(name)` fetches it first.
    """

    def __init__(self, name, client=None, info=None, keep=True):
        self.name = name
        self.client = _client(client)
        self.info = info or {}
        self.keep = keep
        self.files = Files(self)

    def __repr__(self):
        return f"Sandbox({self.name!r}, state={self.info.get('state')!r})"

    def __enter__(self):
        return self

    def __exit__(self, *exc):
        self.close()
        return False

    def close(self):
        """Delete the sandbox unless `keep` is set (what leaving a `with` block does)."""
        if not self.keep:
            try:
                self.delete()
            except FroogError as e:
                if e.status != 404:
                    raise

    @classmethod
    def create(cls, name=None, client=None, wait=True, timeout=600, keep=False, idempotency_key=None, **fields):
        """Create one sandbox and (with wait=True) wait until it's running. `fields` are `CreateSandbox` fields; write
        `from_` for `from`. If waiting fails and keep is False, the sandbox is deleted again."""
        client = _client(client)
        body = {("from" if k == "from_" else k): v for k, v in fields.items() if v is not None}
        if name is not None:
            body["name"] = name
        if body.get("count") not in (None, 1):
            raise ValueError("Sandbox.create makes one sandbox; use froog.create_sandboxes for count")
        info = client.create_sandboxes(body, idempotency_key=idempotency_key)[0]
        sb = cls(info["name"], client, info, keep=keep)
        if wait:
            try:
                sb.wait_until("running", timeout)
            except BaseException:
                sb.close()
                raise
        return sb

    @classmethod
    def get(cls, name, client=None, keep=True):
        client = _client(client)
        return cls(name, client, client.get_sandbox(name), keep=keep)

    @classmethod
    def list(cls, client=None, **filters):
        client = _client(client)
        return [cls(i["name"], client, i) for i in client.list_sandboxes(**filters)]

    def refresh(self):
        """Fetch the sandbox again (`GET /v1/sandboxes/{name}`); also stored as `sb.info`."""
        self.info = self.client.get_sandbox(self.name)
        return self.info

    @property
    def state(self):
        """The current state, fetched now."""
        return self.refresh()["state"]

    def wait_until(self, state, timeout=600):
        self.info = self.client.wait_until(self.name, state, timeout)
        return self.info

    # -- commands

    def exec(self, command, cwd=None, env=None, stdin=None, timeout=None, check=False):
        """Run a command and wait for it. `stdin` is text or bytes; `timeout` in seconds. check=True raises
        CommandError when it exits non-zero."""
        r = CommandResult(self.client.exec(self.name, _exec_body(command, cwd, env, stdin, timeout)), command)
        if check and r.exit_code != 0:
            raise CommandError(command, r.exit_code, r.stdout, r.stderr, r)
        return r

    def exec_stream(self, command, cwd=None, env=None, stdin=None, timeout=None):
        """Run a command and yield events as output arrives: started, stdout/stderr (`data`), then exit or error."""
        return self.client.exec_stream(self.name, _exec_body(command, cwd, env, stdin, timeout))

    def spawn(self, command, cwd=None, env=None, stdin=None, stdin_open=False):
        """Start a background command (a job) and return its `Process`: `p.wait()`, `p.logs()`, `p.kill()`.
        With `stdin_open=True` its standard input stays open (after `stdin`, if given): `p.write(...)`, then
        `p.close_stdin()` for end of file."""
        body = _exec_body(command, cwd, env, stdin)
        if stdin_open:
            body["stdin_open"] = True
        return Process(self, self.client.start_job(self.name, body))

    def process(self, id):
        """Reconnect to a background command by its id (`job_...`)."""
        return Process(self, {"id": id})

    def processes(self):
        """Background commands, newest state."""
        return [Process(self, j) for j in self.client.list_jobs(self.name)]

    def wait_for_port(self, port, timeout=60):
        """Wait until something in the sandbox listens on TCP `port` (`POST /wait`: checked inside it, 1-600 s; an
        auto-start sandbox is started for a key that may start it). Raises a 408 `timeout` FroogError if nothing does."""
        port = int(port)
        r = self.client.wait_for_port(self.name, {"port": port, "timeout_secs": max(1, min(600, int(timeout)))})
        if not r.get("ready"):
            raise FroogError(408, "timeout", f"nothing listens on port {port} in {self.name} after {timeout} s")

    # -- lifecycle

    def start(self, wait=True, timeout=600, **fields):
        """Start a stopped sandbox (`StartSandbox` fields: size, ttl_secs, env, environment)."""
        self.info = self.client.start_sandbox(self.name, fields)
        if wait:
            self.wait_until("running", timeout)
        return self

    def stop(self, wait=True, timeout=600, force=False):
        """Stop it (disk snapshotted, billing stops). wait=True waits for the stop to finish. A stop
        whose snapshot fails is refused (it keeps running); force=True keeps the last good snapshot."""
        self.info = self.client.stop_sandbox(self.name, {"force": True} if force else None)
        stop_id = self.info.get("stop_id")
        if not wait:
            return self
        until = time.time() + timeout
        while stop_id:
            s = self.client.get_stop(stop_id)
            if s["status"] == "completed":
                break
            if s["status"] == "failed":
                raise FroogError(409, "stop_failed", f"{self.name}: {s.get('error')}")
            if s["status"] == "failing":
                raise FroogError(409, "stop_refused", f"{self.name} keeps running (stop refused, retried by froog): {s.get('error')}")
            if time.time() > until:
                raise FroogError(408, "timeout", f"{self.name} didn't stop")
            time.sleep(2)
        else:
            self.wait_until("stopped", timeout)
        try:
            self.refresh()
        except FroogError:
            pass
        return self

    def delete(self):
        self.client.delete_sandbox(self.name)

    def fork(self, name, wait=True, timeout=600, keep=False, idempotency_key=None, **fields):
        """A new sandbox from a snapshot of this one (`ForkSandbox` fields: snapshot, size, env, labels, ttl_secs, ...)."""
        body = {k: v for k, v in fields.items() if v is not None}
        body["name"] = name
        info = self.client.fork_sandbox(self.name, body, idempotency_key=idempotency_key)[0]
        sb = Sandbox(info["name"], self.client, info, keep=keep)
        if wait:
            try:
                sb.wait_until("running", timeout)
            except BaseException:
                sb.close()
                raise
        return sb

    def snapshot(self, label=None):
        """Save a snapshot (with a label it's an account template you can create from)."""
        return self.client.save_snapshot(self.name, {"label": label} if label else {})

    def host(self, port, public=False, subdomain=None, title=None):
        """Give a port an HTTPS URL and return it (private ones carry `?_token=`)."""
        body = {"port": int(port), "public": bool(public)}
        if subdomain is not None:
            body["subdomain"] = subdomain
        if title is not None:
            body["title"] = title
        return self.client.host_port(self.name, body)["url"]

    def open_port(self, port, protocol="tcp"):
        """Open an inbound port (kept across stops and starts): on the sandbox's IPv6 address, and on a public port of its host's IPv4 address (see ipv4_ports); returns the open ports."""
        return self.client.open_port(self.name, {"port": int(port), "protocol": protocol})["ports"]

    def close_port(self, port, protocol="tcp"):
        """Close a port opened with open_port; returns the open ports."""
        return self.client.close_port(self.name, int(port), protocol)["ports"]

    def ports(self):
        """Inbound ports open besides SSH, e.g. ["8080/tcp", "27015/udp"]."""
        return self.client.list_ports(self.name)["ports"]

    def ipv4_ports(self):
        """Where the opened ports are reached over IPv4, e.g. [{"port": "27015/udp", "public_port": 40000, "address": "203.0.113.7:40000"}]. Empty while stopped."""
        return self.client.list_ports(self.name).get("ipv4", [])

    def metrics(self):
        return self.client.get_sandbox_metrics(self.name)

    def touch(self):
        return self.client.touch_sandbox(self.name)

    # -- agents

    @staticmethod
    def _prompt_body(prompt, agent, conversation, new, fields):
        body = {"prompt": prompt, **{k: v for k, v in fields.items() if v is not None}}
        if agent:
            body["agent"] = agent
        if conversation:
            body["conversation_id"] = conversation
        if new:
            body["new"] = True
        return body

    def prompt(self, prompt, agent=None, conversation=None, new=False, **fields):
        """Give a coding agent in the sandbox a task and wait for its answer (`stdout`). It continues the sandbox's
        current conversation (the agent remembers it) unless new=True or conversation=ID. Other fields: model,
        effort, fast, attachments, timeout_secs."""
        return self.client.prompt(self.name, self._prompt_body(prompt, agent, conversation, new, fields))

    def queue_prompt(self, prompt, agent=None, conversation=None, new=False, **fields):
        """Queue a prompt and return it at once (`id`, `conversation_id`, `status`); follow it with `watch_prompt`."""
        return self.client.queue_prompt(self.name, self._prompt_body(prompt, agent, conversation, new, fields))

    def watch_prompt(self, prompt_id, interval=1.0):
        """Yield a prompt's events (status, text, tool_call, tool_result, steer, ...) until it ends."""
        cursor = 0
        while True:
            page = self.client.prompt_events(self.name, prompt_id, cursor=cursor)
            for e in page["events"]:
                yield e
                if e["type"] == "status" and e["data"].get("status") in ("done", "failed", "interrupted"):
                    return
            cursor = page["next_cursor"]
            if not page["events"]:
                time.sleep(interval)

    def conversations(self):
        """Agent conversations, most recent first."""
        return self.client.list_conversations(self.name)["conversations"]

    def steer(self, message, conversation=None):
        """Send a message to the running turn (Claude Code takes it mid-turn; other agents get it as the next turn)."""
        return self.client.steer(self.name, {"message": message, "conversation_id": conversation})

    def interrupt(self, conversation=None):
        """Stop running turns and cancel queued prompts: all, or one conversation's."""
        return self.client.interrupt(self.name, {"conversation_id": conversation})["interrupted"]


class Files:
    """`sb.files`: read, write, list and remove files in a sandbox (paths relative to /home/user, or absolute)."""

    def __init__(self, sandbox):
        self.sandbox = sandbox

    @property
    def _c(self):
        return self.sandbox.client

    def read(self, path):
        """The file's bytes."""
        return self._c.read_file(self.sandbox.name, path)

    def read_text(self, path, encoding="utf-8"):
        return self.read(path).decode(encoding)

    def write(self, path, data, mode=None):
        """Write bytes or text (atomic; parent directories are created). `mode` like "755"."""
        if isinstance(data, str):
            data = data.encode()
        self._c.write_file(self.sandbox.name, bytes(data), path, None if mode is None else str(mode))

    def info(self, path):
        return self._c.file_info(self.sandbox.name, path)

    def list(self, path="."):
        """A directory's entries: dicts with name, kind (file | dir | link | other), size, modified."""
        return self.info(path).get("entries") or []

    def exists(self, path):
        try:
            self.info(path)
            return True
        except FroogError as e:
            if e.status == 404:
                return False
            raise

    def remove(self, path, recursive=False):
        self._c.delete_file(self.sandbox.name, path, recursive=True if recursive else None)

    def mkdir(self, path, mode=None):
        body = {"path": path}
        if mode is not None:
            body["mode"] = str(mode)
        self._c.make_directory(self.sandbox.name, body)

    def move(self, src, dst):
        self._c.move_file(self.sandbox.name, {"from": src, "to": dst})

    def watch(self, path=".", recursive=True, timeout=None):
        """Yield changes under `path` as they happen: dicts with type "change", event (create | modify | delete), path
        and dir. Stops after `timeout` seconds (default and at most 3600), or when you stop iterating."""
        for e in self._c.watch_files(self.sandbox.name, path, recursive=recursive, timeout_secs=None if timeout is None else int(timeout)):
            if e.get("type") == "change":
                yield e
            elif e.get("type") == "error" or (e.get("type") == "end" and e.get("reason") == "error"):
                raise FroogError(409, "watch_failed", e.get("message") or "the watch failed")


_DONE = ("exited", "killed", "lost")


class Process:
    """A background command (a job). `p.wait()` returns its exit code, `p.logs()` its output so far (stdout and stderr
    together), `p.follow()` yields new output until it ends, `p.kill()` stops it. `sb.process(id)` reconnects."""

    def __init__(self, sandbox, info):
        self.sandbox = sandbox
        self.id = info["id"]
        self.info = info
        self.offset = 0  # where follow() reads next

    def __repr__(self):
        return f"Process({self.id!r}, state={self.info.get('state')!r})"

    def _read(self, offset, limit=None):
        out = self.sandbox.client.read_job(self.sandbox.name, self.id, offset=offset, limit=limit)
        self.info = dict(self.info, **{k: out[k] for k in ("state", "exit_code", "started_at") if k in out})
        return out

    def refresh(self):
        self._read(0, 0)
        return self.info

    @property
    def state(self):
        return self.refresh()["state"]

    @property
    def exit_code(self):
        return self.info.get("exit_code")

    def logs(self):
        """All its output so far."""
        chunks, offset = [], 0
        while True:
            out = self._read(offset)
            chunks.append(out["output"])
            if out["next_offset"] <= offset or out["next_offset"] >= out["output_size"]:
                return "".join(chunks)
            offset = out["next_offset"]

    def follow(self, poll=1.0):
        """Yield output as it arrives, from where the last follow() stopped, until the command ends."""
        while True:
            out = self._read(self.offset)
            if out["output"]:
                yield out["output"]
            moved = out["next_offset"] > self.offset
            self.offset = out["next_offset"]
            if out["state"] in _DONE and self.offset >= out["output_size"]:
                return
            if not moved:
                time.sleep(poll)

    def wait(self, timeout=None, poll=1.0, check=False):
        """Wait until it ends; returns the exit code (None when it was killed or lost)."""
        until = None if timeout is None else time.time() + timeout
        while self.refresh()["state"] not in _DONE:
            if until is not None and time.time() > until:
                raise FroogError(408, "timeout", f"{self.id} still running after {timeout} s")
            time.sleep(poll)
        code = self.exit_code
        if check and code != 0:
            raise CommandError(self.info.get("command", self.id), code, self.logs())
        return code

    def kill(self):
        """Stop it (TERM to its process group, KILL after 3 s)."""
        self.sandbox.client.stop_job(self.sandbox.name, self.id)

    def write(self, data, close=False):
        """Write text or bytes to its standard input (it was spawned with `stdin_open=True`); `close=True` then
        closes it."""
        body = {"data_b64": base64.b64encode(bytes(data)).decode()} if isinstance(data, (bytes, bytearray)) else {"data": data}
        if close:
            body["close"] = True
        self.sandbox.client.write_job_stdin(self.sandbox.name, self.id, body)

    def close_stdin(self):
        """Close its standard input: the command reads end of file."""
        self.sandbox.client.write_job_stdin(self.sandbox.name, self.id, {"close": True})


# ---------------------------------------------------------------- asyncio
# The same API as coroutines. Each call runs the blocking client on the default thread pool, which keeps this file
# standard-library only.


async def _to_thread(fn, *args, **kwargs):
    loop = asyncio.get_running_loop()
    return await loop.run_in_executor(None, functools.partial(fn, *args, **kwargs))


_END = object()


async def _aiter(gen):
    try:
        while True:
            item = await _to_thread(next, gen, _END)
            if item is _END:
                return
            yield item
    finally:
        gen.close()


class AsyncFroog:
    """Every `Froog` method as a coroutine: `await AsyncFroog().list_sandboxes()`, plus `exec_stream` as an async
    iterator and `froog.sandboxes.create(...)` returning `AsyncSandbox`."""

    def __init__(self, api_key=None, base_url=None, timeout=3700, client=None, org=None):
        self.sync = client or Froog(api_key, base_url, timeout, org)
        self.sandboxes = AsyncSandboxes(self.sync)

    def __getattr__(self, attr):
        fn = getattr(self.sync, attr)
        if not callable(fn):
            return fn

        @functools.wraps(fn)
        async def call(*args, **kwargs):
            return await _to_thread(fn, *args, **kwargs)

        return call

    def exec_stream(self, name, body):
        return _aiter(self.sync.exec_stream(name, body))


class AsyncSandboxes:
    def __init__(self, client):
        self.client = client

    async def create(self, name=None, **kwargs):
        return await AsyncSandbox.create(name, client=self.client, **kwargs)

    async def get(self, name, keep=True):
        return await AsyncSandbox.get(name, client=self.client, keep=keep)

    async def list(self, **filters):
        return await AsyncSandbox.list(client=self.client, **filters)


class AsyncSandbox:
    """`Sandbox` with coroutines: `async with await AsyncSandbox.create() as sb: r = await sb.exec("ls")`."""

    def __init__(self, sandbox):
        self.sync = sandbox
        self.files = AsyncFiles(sandbox.files)

    name = property(lambda self: self.sync.name)
    info = property(lambda self: self.sync.info)
    client = property(lambda self: self.sync.client)
    keep = property(lambda self: self.sync.keep, lambda self, v: setattr(self.sync, "keep", v))

    def __repr__(self):
        return "Async" + repr(self.sync)

    async def __aenter__(self):
        return self

    async def __aexit__(self, *exc):
        await self.close()
        return False

    @classmethod
    async def create(cls, name=None, client=None, **kwargs):
        return cls(await _to_thread(Sandbox.create, name, client=_client(client), **kwargs))

    @classmethod
    async def get(cls, name, client=None, keep=True):
        return cls(await _to_thread(Sandbox.get, name, client=_client(client), keep=keep))

    @classmethod
    async def list(cls, client=None, **filters):
        return [cls(s) for s in await _to_thread(Sandbox.list, client=_client(client), **filters)]

    async def close(self):
        await _to_thread(self.sync.close)

    async def refresh(self):
        return await _to_thread(self.sync.refresh)

    async def state(self):
        return (await self.refresh())["state"]

    async def wait_until(self, state, timeout=600):
        return await _to_thread(self.sync.wait_until, state, timeout)

    async def exec(self, command, **kwargs):
        return await _to_thread(self.sync.exec, command, **kwargs)

    def exec_stream(self, command, **kwargs):
        return _aiter(self.sync.exec_stream(command, **kwargs))

    async def spawn(self, command, **kwargs):
        return AsyncProcess(await _to_thread(self.sync.spawn, command, **kwargs))

    def process(self, id):
        return AsyncProcess(self.sync.process(id))

    async def processes(self):
        return [AsyncProcess(p) for p in await _to_thread(self.sync.processes)]

    async def wait_for_port(self, port, timeout=60):
        await _to_thread(self.sync.wait_for_port, port, timeout)

    async def start(self, **kwargs):
        await _to_thread(self.sync.start, **kwargs)
        return self

    async def stop(self, **kwargs):
        await _to_thread(self.sync.stop, **kwargs)
        return self

    async def delete(self):
        await _to_thread(self.sync.delete)

    async def fork(self, name, **kwargs):
        return AsyncSandbox(await _to_thread(self.sync.fork, name, **kwargs))

    async def snapshot(self, label=None):
        return await _to_thread(self.sync.snapshot, label)

    async def host(self, port, **kwargs):
        return await _to_thread(self.sync.host, port, **kwargs)

    async def open_port(self, port, protocol="tcp"):
        return await _to_thread(self.sync.open_port, port, protocol)

    async def close_port(self, port, protocol="tcp"):
        return await _to_thread(self.sync.close_port, port, protocol)

    async def ports(self):
        return await _to_thread(self.sync.ports)

    async def ipv4_ports(self):
        return await _to_thread(self.sync.ipv4_ports)

    async def metrics(self):
        return await _to_thread(self.sync.metrics)

    async def touch(self):
        return await _to_thread(self.sync.touch)


class AsyncFiles:
    def watch(self, path=".", recursive=True, timeout=None):
        return _aiter(self.sync.watch(path, recursive, timeout))

    def __init__(self, files):
        self.sync = files

    def __getattr__(self, attr):
        fn = getattr(self.sync, attr)

        @functools.wraps(fn)
        async def call(*args, **kwargs):
            return await _to_thread(fn, *args, **kwargs)

        return call


class AsyncProcess:
    def __init__(self, process):
        self.sync = process

    id = property(lambda self: self.sync.id)
    info = property(lambda self: self.sync.info)
    exit_code = property(lambda self: self.sync.exit_code)

    def __repr__(self):
        return "Async" + repr(self.sync)

    async def refresh(self):
        return await _to_thread(self.sync.refresh)

    async def logs(self):
        return await _to_thread(self.sync.logs)

    async def wait(self, timeout=None, poll=1.0, check=False):
        until = None if timeout is None else time.time() + timeout
        while (await self.refresh())["state"] not in _DONE:
            if until is not None and time.time() > until:
                raise FroogError(408, "timeout", f"{self.id} still running after {timeout} s")
            await asyncio.sleep(poll)
        code = self.sync.exit_code
        if check and code != 0:
            raise CommandError(self.info.get("command", self.id), code, await self.logs())
        return code

    async def follow(self, poll=1.0):
        p = self.sync
        while True:
            out = await _to_thread(p._read, p.offset)
            if out["output"]:
                yield out["output"]
            moved = out["next_offset"] > p.offset
            p.offset = out["next_offset"]
            if out["state"] in _DONE and p.offset >= out["output_size"]:
                return
            if not moved:
                await asyncio.sleep(poll)

    async def kill(self):
        await _to_thread(self.sync.kill)

    async def write(self, data, close=False):
        await _to_thread(self.sync.write, data, close)

    async def close_stdin(self):
        await _to_thread(self.sync.close_stdin)