"""An AI trading agent on Orblivion. You write its strategy in plain words, in strategy.txt. On every tick a model on Orbio's gateway reads it with the agent's market view and book and decides; Orblivion checks every answer before anything trades. Orblivion's own Orbio credits pay for the model (funding "orblivion"), so you need no LLM key. python agent.py keys make the wallet's key in key_dir (never printed) and show its address python agent.py options the server's strategy templates and models, and who pays for this wallet python agent.py paper [MINUTES] create the agent on paper (no money), then watch it python agent.py watch [MINUTES] its decisions, trades and model spend as they come python agent.py live [MINUTES] after you type SEND: a live agent, its session, the setup and the executor's gas python agent.py status the agents, today's model spend, and the executor's gas python agent.py off after you type SEND: stop them, end the session, get the executor's gas back python agent.py off --chain-only after you type SEND: end the live session on chain from agent.json alone Watching lasts 30 minutes unless you give another number; Ctrl+C stops it sooner. The agent runs on Orblivion's servers whether you watch or not: `off` stops it. The key stays in key_dir (default ~/.orblivion-ai-agent), outside this folder, and is never printed. Nothing that moves money happens unless you type SEND. agent.json keeps the live session's revocation record (no secret: the delegation the wallet signed, its signature and what the setup approved). With it, `off` builds the session's on-chain revocation here and sends it straight to Robinhood Chain, so the session ends on chain even when Orblivion doesn't answer. That needs an SDK with Grant.revocation_record; with an older one, `off` still works, with the on-chain steps from Orblivion's answer. """ import json import os import re import sys import textwrap import time try: from orblivion import OrblivionClient, OrblivionError, ai_params except ImportError: # pragma: no cover sys.exit('The Orblivion SDK is missing: pip install "orblivion>=0.1.1"') HERE = os.path.dirname(os.path.abspath(__file__)) EXPLORER = "https://robin.etherscan.io" CHAIN_ID = 4663 # Robinhood Chain UPDATE_SDK = "update the SDK for a revoke that needs nothing from Orblivion: pip install -U orblivion" # --- settings: your strategy is in strategy.txt; these say how the agent runs it --------------------------------------- SETTINGS = { "name": "ai-agent", "pairs": ["ETH-USDG"], # what it may trade, against USDG (up to 6 pairs) "model": "anthropic/claude-haiku-5.5", # Orblivion's credits pay for the cheap models: see `options` "template": "trend", # the server's template strategy.txt started from "interval_s": 900, # it thinks every 15 minutes, the fastest Orblivion's credits pay for "budget_usd_day": 0.25, # the most its model calls may cost in a UTC day "max_orders": 2, # orders in one answer (a live agent makes at most one trade a tick) "limits": {"max_trade_usd": 10, "max_day_usd": 40, "max_position_usd": 100}, # checked on every order "paper_book": {"USDG": 50, "ETH": 0.02}, # what the paper agent starts with "session": {"max_trades": 100, "hours": 168}, # live: what the wallet signs (per trade and per day: the limits) "max_gas_usd": 3, # live: the most this script sends the executor for gas "key_dir": "~/.orblivion-ai-agent", # the wallet's key and agent.json, outside this folder "api_url": None, # None: ORBLIVION_API_URL if set, else https://api.orblivion.com } # --- the key, the client and what the script remembers (all in key_dir) ---------------------------------------------- def key_dir(): return os.path.abspath(os.path.expanduser(SETTINGS["key_dir"])) def load_wallet(): """The wallet's key from /wallet.json, as an eth_account account. Never printed.""" from eth_account import Account path = os.path.join(key_dir(), "wallet.json") if not os.path.exists(path): sys.exit(f"No wallet key in {key_dir()}: run python agent.py keys first.") with open(path, encoding="utf-8") as f: return Account.from_key(json.load(f)["private_key"]) def client_for(account): """An Orblivion client for the wallet. It signs in by itself on its first call.""" return OrblivionClient(base_url=SETTINGS["api_url"], account=account) def saved(): """agent.json: the agents this script made, {"paper": {...}, "live": {...}}.""" try: with open(os.path.join(key_dir(), "agent.json"), encoding="utf-8") as f: return json.load(f) except (OSError, ValueError): return {} def save(state): with open(os.path.join(key_dir(), "agent.json"), "w", encoding="utf-8") as f: json.dump(state, f, indent=1) def read_strategy(): """strategy.txt without its # lines: your words, as the model gets them.""" with open(os.path.join(HERE, "strategy.txt"), encoding="utf-8") as f: lines = [x for x in f.read().splitlines() if not x.lstrip().startswith("#")] text = "\n".join(lines).strip() if not 1 <= len(text) <= 1500 or text.count("\n") >= 40: sys.exit(f"strategy.txt has {len(text)} characters on {text.count(chr(10)) + 1} lines without its # lines: " "it must be 1 to 1,500 characters, at most 40 lines.") return text def params(strategy): """The AI brain's settings. funding "orblivion": Orblivion's credits pay; fallback_own=False: when today's share is used up, the agent waits for tomorrow's instead of asking for your own Orbio balance.""" s = SETTINGS return ai_params(strategy, s["model"], funding="orblivion", budget_usd_day=s["budget_usd_day"], max_orders=s["max_orders"], template=s["template"], fallback_own=False) def confirm(*lines): """Nothing that moves money happens unless SEND is typed.""" print() for line in lines: print(line) if input("Type SEND to go ahead: ").strip() != "SEND": sys.exit("Nothing was sent.") def tx(h): return f"{EXPLORER}/tx/{h}" def when(ts, fmt="%Y-%m-%d %H:%M"): return time.strftime(fmt, time.localtime(ts)) def usd(x): """Dollars as people read them: $0.10, and small amounts to the tenth of a cent ($0.0009).""" x = float(x or 0) return f"${x:.2f}" if x == 0 or x >= 0.01 else f"${x:.4f}" # --- keys and options ------------------------------------------------------------------------------------------------- def keys(): """A new key in /wallet.json, readable only by you where the system allows it. Never replaces one.""" from eth_account import Account os.makedirs(key_dir(), exist_ok=True) path = os.path.join(key_dir(), "wallet.json") made = not os.path.exists(path) if made: key = Account.create().key.hex() fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) with os.fdopen(fd, "w", encoding="utf-8") as f: json.dump({"private_key": key if key.startswith("0x") else "0x" + key}, f) print(f"Wallet {load_wallet().address} ({'new' if made else 'already there'})") print(f"\nThe key is in {path}. It never leaves this computer and is never printed: don't share or commit it.") print("Paper needs no money. To go live, send this wallet (Robinhood Chain, 4663) what the agent may trade, in ETH " f"and USDG, plus about ${SETTINGS['max_gas_usd'] + 1:g} of ETH for the setup and the executor's gas.") def who_pays(o): """Who pays for this wallet's model calls, from Orblivion's answer (GET /api/v1/hosted/ai).""" if not o.get("available"): return [" Orblivion's credits aren't on on this server: its AI agents can't think today."] out = [f" Today's pool {usd(o.get('pool_usd'))}, {usd(o.get('pool_left_usd'))} left; it pays for " f"{', '.join(o.get('models') or [])}, thinking every {(o.get('min_interval_s') or 900) // 60} minutes " "or slower"] if o.get("trial_open"): until = f"until {when(o['trial_until'], '%d %b')}" if o.get("trial_until") else \ f"for {o.get('trial_days')} days from your first AI agent" out.append(f" Trial open {until}: {usd(o.get('trial_share_usd'))} a day for this wallet's agents, " "paper or live") elif o.get("trial_agents_only") and not o.get("orbio_agent"): out.append(" Trial not for this wallet: the paper trial is for wallets that own an Orbio agent, or " "are one's agent wallet") else: out.append(f" Trial over: it lasts {o.get('trial_days')} days from a wallet's first AI agent") share = float(o.get("live_share_usd") or 0) out.append(f" Live share {usd(share)} a day" + (", backed by the trading fees this wallet paid" if share else f": it takes {usd(o.get('live_min_usd') or 5)} of trades through Orblivion in " f"{o.get('live_days') or 14} days and fees paid on them (this wallet: {usd(o.get('live_traded_usd'))} " f"of trades, {usd(o.get('live_fees_usd'))} of fees)")) out.append(" With no share, or once it is used up, the agent keeps ticking but doesn't think, or trade, until " "00:00 UTC.") return out def options(): opts = client_for(load_wallet()).hosted.ai_options() print("Templates (start strategy.txt from one, and set SETTINGS['template']):") for t in opts.get("templates") or []: print(f" {t['id']:<20} {t['title']}") pool = ((opts.get("funding") or {}).get("orblivion") or {}) print(f"\nModels (a tick every {SETTINGS['interval_s'] // 60} minutes):") for m in opts.get("models") or []: est = (m.get("estimate") or {}).get(str(SETTINGS["interval_s"])) or {} cost = f"about ${est['usd_per_day']:.4f} a day" if est else "no estimate" who = "Orblivion's credits" if m["id"] in (pool.get("models") or []) else "your own Orbio balance (the Lab)" print(f" {m['id']:<30} {cost:<22} paid by {who}") print("\nWho pays for this wallet's model calls (funding \"orblivion\": Orblivion's own Orbio credits):") print("\n".join(who_pays(pool))) # --- paper ------------------------------------------------------------------------------------------------------------ def paper(minutes="30"): state = saved() if state.get("paper") and not state["paper"].get("off"): sys.exit(f"Paper agent {state['paper']['id']} already runs: python agent.py watch (or off, then paper again).") owner = client_for(load_wallet()) s = SETTINGS agent = owner.hosted.create(brain="orbio_llm", params=params(read_strategy()), interval_s=s["interval_s"], pairs=s["pairs"], limits=s["limits"], paper_book=s["paper_book"], name=s["name"], mode="paper")["agent"] state["paper"] = {"id": agent["id"]} save(state) book = ", ".join(f"{v:g} {k}" for k, v in s["paper_book"].items()) print(f"Agent {agent['id']} (paper: {book} to start; no money moves)") print(f"Thinks every {s['interval_s'] // 60} minutes with {s['model']}, at most ${s['budget_usd_day']:g} a day") lim = s["limits"] print(f"Limits ${lim['max_trade_usd']:g} a trade, ${lim['max_day_usd']:g} a day, " f"${lim['max_position_usd']:g} in any token but USDG") pays(agent) watch(minutes, agent["id"]) def pays(agent): """Whether Orblivion's credits pay for this agent's model calls now, and why not.""" pool = (agent.get("ai") or {}).get("pool") or {} if pool.get("eligible"): print(f"Paid by Orblivion's credits: a {pool.get('kind')} share of {usd(pool.get('share_usd'))} today, " f"{usd(pool.get('share_left_usd'))} left") elif pool: print(f"Paid by nobody yet ({pool.get('code')}): {pool.get('why_not')}") # --- watching --------------------------------------------------------------------------------------------------------- def newest(state): for kind in ("live", "paper"): if state.get(kind) and not state[kind].get("off"): return state[kind]["id"] sys.exit("No agent running: python agent.py paper") def watch(minutes="30", agent_id=None): """Prints the agent's log as it grows: each decision with its reasoning, cost and payer, each order, trade and hold. For `minutes`, or until Ctrl+C; the agent keeps running either way.""" agent_id = agent_id or newest(saved()) c = client_for(load_wallet()) end, seen = time.time() + 60 * float(minutes), 0 print(f"\nWatching {agent_id} for {float(minutes):g} minutes (Ctrl+C stops watching; the agent runs on)") while True: rows = [r for r in reversed(c.hosted.logs(agent_id, limit=40)["rows"]) if r["id"] > seen] for r in rows: show(r) seen = r["id"] if any(r["kind"] == "decision" for r in rows): spent(c, agent_id) if time.time() >= end: return time.sleep(10) def show(r): d, at, pad = r["data"], when(r["ts"], "%H:%M:%S"), " " * 10 if r["kind"] == "decision": ai = d.get("ai") or {} if ai.get("status") == "hold": return # the model wasn't asked: the hold row says why if ai.get("status") not in (None, "ok"): print(f"{at} refused {ai.get('code')}: the model's answer broke a rule, so nothing trades this tick") return print(textwrap.fill(d.get("reasoning") or "", 116, initial_indent=f"{at} thinks ", subsequent_indent=pad + " " * 9)) payer = {"orblivion": "Orblivion's credits", "own": "your Orbio balance"}.get(ai.get("source"), "?") print(f"{pad}model {ai.get('model')}: ${float(ai.get('cost_usd') or 0):.6f}, paid by {payer}") for o in d.get("orders") or []: print(f"{pad}order sell ${o['amount_usd']:g} of {o['sell']} for {o['buy']}" + (f": {o['reason']}" if o.get("reason") else "")) elif r["kind"] == "trade": where = f" {tx(d['tx_hash'])}" if d.get("tx_hash") else " (paper)" print(f"{at} trade sold {float(d['amount_in']):.6g} {d['sell']} for {float(d['amount_out'] or 0):.6g} " f"{d['buy']}{where}") else: why = d.get("detail") or d.get("reason") or "" print(f"{at} {r['kind']:<8} {d.get('code') or ''}{': ' + why if why else ''}"[:240].rstrip()) def spent(c, agent_id): ai = c.hosted.spend(agent_id, limit=1)["ai"] pool = ai.get("pool") or {} line = f"{' ' * 10}spend ${float((ai.get('today') or {}).get('usd') or 0):.6f} today of its " \ f"{usd(ai.get('budget_usd_day'))} budget" if pool.get("share_usd") is not None: line += f"; this wallet's share of Orblivion's credits: ${float(pool.get('share_left_usd') or 0):.6f} of " \ f"{usd(pool['share_usd'])} left" print(line) # --- live ------------------------------------------------------------------------------------------------------------- def live(minutes="30"): state = saved() if state.get("live") and not state["live"].get("off"): sys.exit(f"Live agent {state['live']['id']} already runs: python agent.py watch (or off).") strategy, s, lim = read_strategy(), SETTINGS, SETTINGS["limits"] owner = client_for(load_wallet()) eth_usd = owner.marks(["ETH"])["ETH"].price8 / 1e8 tokens = sorted({t for p in s["pairs"] for t in p.split("-")}) print(f"Wallet {owner.address} {owner.rpc.balance(owner.address) / 1e18:.6f} ETH") confirm("Going live lets the agent trade this wallet's money, inside a session the wallet signs now:", f" - the session: {', '.join(tokens)} only, ${lim['max_trade_usd']:g} a trade, ${lim['max_day_usd']:g} a " f"day, {s['session']['max_trades']} trades, for {s['session']['hours']:g} hours. Orblivion's co-signer and " "the chain enforce it on every trade;", " - its setup, sent from the wallet: on a plain wallet, its code set to MetaMask's " "EIP7702StatelessDeleGator (EIP-7702), which sessions run on, and an approval of at most " f"${lim['max_day_usd']:g} of USDG through Permit2;", f" - up to ${s['max_gas_usd']:g} of ETH to the agent's executor for its gas (off sends what is left back).") if state.get("paper") and not state["paper"].get("off"): owner.hosted.pause(state["paper"]["id"]) # one share a wallet: the paper agent stops drawing on it print(f"Paused paper agent {state['paper']['id']}") made = owner.hosted.create(brain="orbio_llm", params=params(strategy), interval_s=s["interval_s"], pairs=s["pairs"], limits=lim, name=s["name"], mode="live", session=dict(s["session"]), allow_7702=True) agent, grant = made["agent"], made["grant"] record = revocation_record(grant) state["live"] = {"id": agent["id"], "session_id": grant.session_id, "executor": agent["executor"], "revocation_record": record} save(state) print(f"Agent {agent['id']} (live; it waits until it is switched on)") print(f"Executor {agent['executor']}: Orblivion's signing service holds its key (attestation checked)") print(f"Session {grant.session_id}, until {when(grant.expires)}; " f"{sum(c.ok for c in grant.checks)} of {len(grant.checks)} checks passed before the wallet signed") if record is None: print(f"Note this SDK keeps no revocation record, so off will take its on-chain steps from Orblivion: " f"{UPDATE_SDK}") for step in owner.sessions.setup(grant).steps: print(f"Setup {step.kind}: {tx(step.tx_hash) if step.sent else 'not sent'}") fund_gas(owner, agent["id"], agent["executor"], eth_usd) owner.hosted.update(agent["id"], live=True) a = owner.hosted.resume(agent["id"]) print(f"On {a.get('status')}, {a.get('trading')}: it thinks every {s['interval_s'] // 60} minutes") pays(owner.hosted.get(agent["id"])) watch(minutes, agent["id"]) def revocation_record(grant): """The session's revocation record (Grant.revocation_record()), or None on an SDK without it.""" fn = getattr(grant, "revocation_record", None) if not callable(fn): return None try: return fn() except (TypeError, ValueError): return None def fund_gas(owner, agent_id, executor, eth_usd): """Sends the executor twice what Orblivion says it needs for gas (about ten trades' worth at the gas cap), never more than max_gas_usd: one plain ETH transfer from the wallet, to the attested executor only.""" from eth_utils import to_checksum_address g = owner.hosted.executor(agent_id) if g["executor"].lower() != executor.lower(): sys.exit(f"Orblivion named executor {g['executor']}, not the attested {executor}: nothing sent.") amount = min(2 * int(g["min_balance_wei"]) - int(g.get("balance_wei") or 0), int(SETTINGS["max_gas_usd"] / eth_usd * 1e18)) if amount <= 0: print("Gas the executor has enough") return acct = load_wallet() to = to_checksum_address(executor) gas = int(owner.rpc.call("eth_estimateGas", [{"from": acct.address, "to": to, "value": hex(amount)}]), 16) signed = acct.sign_transaction({ "type": 2, "chainId": CHAIN_ID, "nonce": owner.rpc.nonce(acct.address, "pending"), "to": to, "value": amount, "data": "0x", "gas": gas * 5 // 4, "maxFeePerGas": owner.rpc.gas_price() * 2, "maxPriorityFeePerGas": 0}) h = owner.rpc.send_raw("0x" + bytes(signed.raw_transaction).hex(), owner.sequencer_url) rc = owner.rpc.wait_receipt(h).receipt ok = rc is not None and rc.get("status") == "0x1" print(f"Gas {amount / 1e18:.6f} ETH (${amount / 1e18 * eth_usd:.2f}) to the executor: {tx(h)}" + ("" if ok else " (not mined yet: check it before sending again)")) if ok and not owner.hosted.executor(agent_id).get("enough"): print(f" still short of what it needs: max_gas_usd (${SETTINGS['max_gas_usd']:g}) is too low today") # --- status and the off switch ---------------------------------------------------------------------------------------- def status(): state, owner = saved(), client_for(load_wallet()) for kind in ("paper", "live"): if not state.get(kind): continue a = owner.hosted.get(state[kind]["id"]) why = a.get("pause_reason") or {} print(f"{kind.title():<9} {a['id']} {a['status']}, {a.get('trading')}, {a.get('tick_no')} ticks" + (f"; paused: {why.get('code')}" if why else "")) ai = a.get("ai") or {} today = ai.get("today") or {} print(f" {ai.get('model')}: {today.get('calls', 0)} calls, ${float(today.get('usd') or 0):.6f} today; " f"${float(ai.get('budget_left_usd') or 0):.4f} of its {usd(ai.get('budget_usd_day'))} budget left") pays(a) if kind == "live": g = owner.hosted.executor(a["id"]) print(f"Executor {g['executor']} {usd(g.get('balance_usd'))} of ETH for gas " f"({'enough' if g.get('enough') else 'top it up'})") def off(chain_only=False): state, owner = saved(), client_for(load_wallet()) paper_, live_ = state.get("paper") or {}, state.get("live") or {} record = live_.get("revocation_record") local = bool(record) and hasattr(owner.sessions, "disable_onchain") if chain_only: return off_on_chain(state, owner, live_, record, local) todo = [] if paper_ and not paper_.get("off"): todo.append(f" - stop the paper agent {paper_['id']} (final);") if live_ and not live_.get("off"): todo += [f" - stop the live agent {live_['id']} (final) and end session {live_['session_id']}: Orblivion " "stops co-signing at once, then transactions from the wallet end it on chain (the delegation " "disabled, the setup's approvals back to 0). " + ( "They are built here from agent.json and sent straight to Robinhood Chain, even if Orblivion " "doesn't answer;" if local else "They come from Orblivion's answer" + ( f" ({UPDATE_SDK});" if record else " (agent.json holds no revocation record);")), " - send what is left of the executor's gas back to this wallet."] elif live_.get("off") and not live_.get("refunded"): if live_.get("stopped") is False: # its session ended on chain while Orblivion couldn't stop it todo.append(f" - stop the live agent {live_['id']} (final);") todo.append(" - send what is left of the executor's gas back to this wallet.") if not todo: sys.exit("Nothing to switch off.") confirm("The off switch:", *todo) if paper_ and not paper_.get("off"): try: owner.hosted.stop(paper_["id"]) paper_["off"] = True save(state) print(f"Stopped paper agent {paper_['id']}") except OrblivionError as e: if not (local and live_ and not live_.get("off")): raise # Orblivion down: the live session still ends on chain below; the paper agent (no money) waits for a rerun print(f"Not stopped paper agent {paper_['id']} ({e.code}: {e.message}): run off again later") if live_ and not live_.get("off"): if not live_.get("stopped"): try: owner.hosted.stop(live_["id"]) live_["stopped"] = True save(state) print(f"Stopped live agent {live_['id']}: Orblivion co-signs nothing more for it") except OrblivionError as e: if not local: raise live_["stopped"] = False # the next off tries again save(state) print(f"Not stopped at Orblivion ({e.code}: {e.message}): ending its session on chain anyway") if local: r = owner.sessions.revoke(live_["session_id"], record=record) else: r = owner.sessions.revoke(live_["session_id"]) api_error = getattr(r, "api_error", None) if api_error: print(f"Not revoked at Orblivion: session {live_['session_id']} ({api_error})") else: print(f"Revoked session {live_['session_id']} at Orblivion: {r.effective or r.status}") show_steps(r) on = getattr(r, "on_chain", None) status = (on or {}).get("status") if status == "confirmed": print("On chain confirmed: the delegation reads as disabled and the approvals as 0, on every RPC") elif status == "server_reported": print("On chain sent, and read back as Orblivion named it: which roots were disabled is Orblivion's word. " "Running off again can't check it better: " + ( UPDATE_SDK if record else "agent.json holds no revocation record (the session was granted with " "an older SDK). Sessions granted now keep one.")) elif status == "unconfirmed" and on.get("signed_live") and local: # a root the wallet signed, named by Orblivion and still live: agent.json is most likely an older session's print(f"On chain {status}: Orblivion's answer carries a root your wallet signed that still reads live (" + ", ".join(h[:10] + "…" for h in on["signed_live"]) + "): agent.json is most likely an older " "session's record, and that root this session's own.") confirm("Disable that root too: one more transaction from the wallet, straight to Robinhood Chain.", "Your wallet signed it, so it is yours. If it is another session of yours that you still use, " "don't: this would end that one too.") r = owner.sessions.revoke(live_["session_id"], record=record, disable_signed=True) show_steps(r) on = getattr(r, "on_chain", None) or {} status = on.get("status") if not on.get("signed_live") and on.get("from_record") == "confirmed": print("On chain every root of agent.json and every root your wallet signed that Orblivion named reads " "disabled, and the approvals 0, on every RPC") status = "confirmed" else: print(f"On chain {status}: " + "; ".join(on.get("detail") or []) + ". Run python agent.py off again: it sends what is still needed") elif status == "unconfirmed" and on.get("from_record") == "confirmed": # everything the record holds reads done; only Orblivion's answer (another root it names) stands in the way print(f"On chain {status}: " + "; ".join(on.get("detail") or []) + ". Everything agent.json holds reads " "done on every RPC; only Orblivion's answer stands in the way (it names a root agent.json doesn't " "hold, and none your wallet signed), and running off again won't change that. If agent.json is this " "session's own record, take the chain's word from it alone: python agent.py off --chain-only") elif status: print(f"On chain {status}: " + "; ".join(on.get("detail") or []) + ". Run python agent.py off again: it sends what is still needed") # off only once the chain confirms it (or, with an SDK that doesn't read it back, Orblivion's word as before) if status in ("confirmed", "server_reported") or (on is None and not api_error): live_["off"] = True save(state) elif live_ and live_.get("stopped") is False: try: owner.hosted.stop(live_["id"]) live_["stopped"] = True save(state) print(f"Stopped live agent {live_['id']}") except OrblivionError as e: print(f"Not stopped at Orblivion ({e.code}: {e.message}): run off again later") if live_ and not live_.get("refunded"): refund(owner, live_["id"]) live_["refunded"] = True save(state) def show_steps(r): for step in r.steps: how = getattr(step, "status", None) print(f" {step.kind}: {tx(step.tx_hash) if step.sent else 'not sent'}" + (f" ({how})" if how and how != "ok" else "")) def off_on_chain(state, owner, live_, record, local): """off --chain-only: the live session's on-chain steps from agent.json's revocation record alone (disable_onchain), and the chain's answer read from it alone. Orblivion isn't asked anything; a later off stops the agent there and gets the executor's gas back.""" if not live_ or live_.get("off"): sys.exit("No live session to end on chain. python agent.py off does the rest.") if not local: sys.exit("off --chain-only ends the live session from agent.json's revocation record alone, so it needs " + ( f"an SDK that can: {UPDATE_SDK}" if record else "a record, and agent.json holds none (the session was " "granted with an older SDK): run python agent.py off")) confirm(f"End session {live_['session_id']} on chain from agent.json alone: transactions from the wallet disable its " "delegation and set the setup's approvals back to 0, sent straight to Robinhood Chain.", "Orblivion isn't asked anything: python agent.py off afterwards stops the agent there and gets the " "executor's gas back.") r = owner.sessions.disable_onchain(record) show_steps(r) on = getattr(r, "on_chain", None) or {} if on.get("status") == "confirmed": print("On chain confirmed from agent.json alone: the delegation reads as disabled and the approvals as 0, on " "every RPC") live_["off"] = True if not live_.get("stopped"): live_["stopped"] = False save(state) print(f"Session {live_['session_id']} ended on chain. Next: python agent.py off stops the agent at Orblivion " "and gets the executor's gas back.") else: print(f"On chain {on.get('status')}: " + "; ".join(on.get("detail") or [])) print("The chain doesn't confirm it ended yet: run python agent.py off --chain-only again.") def refund(owner, agent_id): """The executor's leftover gas, back to this wallet. Orblivion refuses it while a trade co-signed just before the stop could still land (at most about two minutes), and says how long to wait.""" for _ in range(5): try: r = owner.hosted.refund_gas(agent_id) print(f"Refunded {int(r['value_wei']) / 1e18:.6f} ETH from the executor to this wallet: " f"{tx(r['tx_hash'])} ({r.get('status')})") return except OrblivionError as e: wait = re.search(r"try again in (\d+) s", e.message or "") if not wait: raise print(f"Waiting {wait.group(1)} s: Orblivion holds the refund until no trade signed before the stop " "can land") time.sleep(int(wait.group(1)) + 2) sys.exit("The refund is still held: run python agent.py off again in a few minutes.") def main(argv): acts = {"keys": keys, "options": options, "paper": paper, "watch": watch, "live": live, "status": status, "off": off} if not argv or argv[0] not in acts: sys.exit(__doc__) if hasattr(sys.stdout, "reconfigure"): sys.stdout.reconfigure(encoding="utf-8", errors="replace") try: if argv[0] == "off": off(chain_only="--chain-only" in argv[1:]) else: acts[argv[0]](*argv[1:2] if argv[0] in ("paper", "watch", "live") else ()) except OrblivionError as e: sys.exit(f"\nRefused: {e.code}: {e.message}") except KeyboardInterrupt: sys.exit("\nStopped. An agent you made keeps running at Orblivion: python agent.py off stops it.") if __name__ == "__main__": main(sys.argv[1:])