`oo` Orchestrator

oo Orchestrator

This skill is the front door for everything oo. Its job is to detect what the user wants and either route to the right sub-skill or handle the connect flow inline.

Step 1 — Detect intent

Read the user's message and classify into ONE of:

Intent	Signals	Action
connect	message contains 0x[0-9a-fA-F]{64}, or words like "connect", "delegate", "ask agent", "send to agent"	Continue with the Connect flow below
init	"init", "set up oo", "scaffold", "create agent.json", "make me publishable", "start a new bundle"	Invoke the oo-init skill via the Skill tool
publish	"publish", "ship", "share my agent", "release my skills", "announce my agent"	Invoke oo-publish
subscribe	"subscribe", "follow", "install 's skills", "add 0x... as a subscription"	Invoke oo-subscribe
accept	"accept subscribers", "review subscription requests", "who's following me", "approve subscriptions"	Invoke oo-accept
ambiguous	bare /oo, "oo", "help me with oo", or anything that fits >1 bucket	Go to Step 2

If the intent is clear, route immediately. Don't ask.

Step 2 — Ask when ambiguous

If you cannot confidently classify, ask the user once with AskUserQuestion. Tailor the options based on whether they already have an identity:

• If ~/.co/agent.json does not exist → likely init is needed first; offer init / subscribe / connect.
• If it does exist → offer publish / subscribe / accept / connect.

Then invoke the chosen sub-skill. Never proceed past Step 2 without a clear intent.

---

Connect flow (inline)

The rest of this file handles the connect intent only. Everything else is delegated to its own skill.

Connect to remote ConnectOnion agents, delegate tasks, and handle multi-turn collaboration.

Environment Setup

Before any interaction, verify the environment is ready. Run these checks sequentially — stop on first failure:

1. Check connectonion is installed:

python -c "import connectonion; print(connectonion.__version__)"

If ImportError: run pip install connectonion, then re-check.

2. Check agent identity exists:

ls .co/keys/agent.key 2>/dev/null || ls ~/.co/keys/agent.key 2>/dev/null

If neither exists: run co init to generate identity.

3. Verify identity is usable:

python -c "
from connectonion import address
from pathlib import Path
a = address.load(Path('.co')) or address.load(Path.home() / '.co')
print(a['address'])
"

If fails: report the error and stop. The user needs to fix their .co/ directory.

Note: import connectonion prints [env] ... lines to stdout. For all environment checks, parse only the last line of stdout output. Ignore everything else.

Skip environment checks after the first successful run in a session.

Connecting to a Remote Agent

Parsing User Intent

Extract from the user's message:

• Target address: match regex 0x[0-9a-fA-F]{64} (66 chars total)
• Task description: everything else

For /oo slash command: /oo <address> <task description>

Connection Strategy: Direct-First with Relay Fallback

The default connect() library has a known issue: the relay API may not return an online field, causing direct endpoint resolution to always fail and falling back to relay — which itself may be unreliable. To work around this, the skill uses a smart connection script that:

1. Queries the relay API for the agent's registered endpoints
2. Tries each endpoint directly (verifying via /info)
3. Falls back to relay only if all direct endpoints fail

One-shot Task

Generate and execute this Python script (fill in {address} and {task}):

import sys, json, time, uuid, asyncio
import httpx, websockets
from connectonion import address
from pathlib import Path

TARGET = "{address}"
TASK = "{task}"
TIMEOUT = 60
RELAY_URL = "wss://oo.openonion.ai"

keys = address.load(Path(".co")) or address.load(Path.home() / ".co")

def _sort_endpoints(endpoints):
    def priority(url):
        if "localhost" in url or "127.0.0.1" in url:
            return 0
        if any(x in url for x in ("192.168.", "10.", "172.16.", "172.17.", "172.18.")):
            return 1
        return 2
    return sorted(endpoints, key=priority)

def discover_direct_ws(target, relay_url):
    """Query relay API for endpoints and find a working direct WebSocket."""
    https_relay = relay_url.replace("wss://", "https://").replace("ws://", "http://").rstrip("/")
    try:
        resp = httpx.get(f"{https_relay}/api/relay/agents/{target}", timeout=5)
        if resp.status_code != 200:
            return None
        info = resp.json()
    except Exception:
        return None

    endpoints = info.get("endpoints", [])
    if not endpoints:
        return None

    http_endpoints = [ep for ep in _sort_endpoints(endpoints)
                      if ep.startswith("http://") or ep.startswith("https://")]

    for http_url in http_endpoints:
        try:
            r = httpx.get(f"{http_url}/info", timeout=3, proxy=None)
            if r.status_code == 200 and r.json().get("address") == target:
                ws_url = http_url.replace("https://", "wss://").replace("http://", "ws://")
                if not ws_url.endswith("/ws"):
                    ws_url = ws_url.rstrip("/") + "/ws"
                return ws_url
        except Exception:
            continue
    return None

async def direct_connect(ws_url, target, keys, task, timeout):
    """Connect directly to agent WebSocket, send task, return result."""
    async with websockets.connect(ws_url, proxy=None) as ws:
        # Signed CONNECT
        ts = int(time.time())
        payload = {"to": target, "timestamp": ts}
        canonical = json.dumps(payload, sort_keys=True, separators=(",", ":"))
        signature = address.sign(keys, canonical.encode())
        connect_msg = {
            "type": "CONNECT", "timestamp": ts, "to": target,
            "payload": payload, "from": keys["address"], "signature": signature.hex()
        }
        await ws.send(json.dumps(connect_msg))

        # Wait for CONNECTED
        raw = await asyncio.wait_for(ws.recv(), timeout=10)
        event = json.loads(raw)
        if event.get("type") == "ERROR":
            raise ConnectionError(f"Auth error: {event.get('message', event.get('error'))}")
        if event.get("type") != "CONNECTED":
            raise ConnectionError(f"Unexpected: {event.get('type')}")

        # Signed INPUT
        ts2 = int(time.time())
        input_id = str(uuid.uuid4())
        input_payload = {"prompt": task, "timestamp": ts2}
        input_canonical = json.dumps(input_payload, sort_keys=True, separators=(",", ":"))
        input_sig = address.sign(keys, input_canonical.encode())
        input_msg = {
            "type": "INPUT", "input_id": input_id, "prompt": task, "timestamp": ts2,
            "payload": input_payload, "from": keys["address"], "signature": input_sig.hex()
        }
        await ws.send(json.dumps(input_msg))

        # Stream until OUTPUT
        while True:
            msg = await asyncio.wait_for(ws.recv(), timeout=timeout)
            ev = json.loads(msg)
            t = ev.get("type")
            if t == "OUTPUT":
                return ev.get("result", ""), True
            elif t == "ask_user":
                return ev.get("text", ""), False
            elif t == "ERROR":
                raise ConnectionError(f"Agent error: {ev.get('message', ev.get('error'))}")

# --- Main ---
result_text, done = None, None

# Step 1: Try direct connection
ws_url = discover_direct_ws(TARGET, RELAY_URL)
if ws_url:
    try:
        result_text, done = asyncio.run(direct_connect(ws_url, TARGET, keys, TASK, TIMEOUT))
        print(f"CO_METHOD: direct", flush=True)
    except Exception as e:
        print(f"CO_DIRECT_FAIL: {e}", flush=True)

# Step 2: Fallback to relay
if result_text is None:
    try:
        from connectonion import connect
        agent = connect(TARGET, keys=keys)
        response = agent.input(TASK, timeout=TIMEOUT)
        result_text, done = response.text, response.done
        print(f"CO_METHOD: relay", flush=True)
    except Exception as e:
        print(f"CO_RELAY_FAIL: {e}", flush=True)
        sys.exit(1)

print(f"CO_RESPONSE: {json.dumps(result_text)}", flush=True)
print(f"CO_DONE: {done}", flush=True)

Execute via your shell tool. Parse stdout — only lines starting with CO_ matter, ignore all others. The CO_RESPONSE value is JSON-encoded (to handle multi-line responses). Decode it before presenting to the user.

• CO_DONE: True → return CO_RESPONSE content to the user. Done.
• CO_DONE: False → the remote agent is asking a follow-up question. See Multi-turn Task below.
• CO_METHOD: direct → connected directly (fastest path).
• CO_METHOD: relay → connected via relay fallback.
• CO_DIRECT_FAIL: ... → direct failed, trying relay next.
• CO_RELAY_FAIL: ... → both methods failed. Report error to user.

For long-running tasks, increase timeout to 300.

Multi-turn Task

Multi-turn requires maintaining session state. Use separate one-shot calls per turn since stdin interaction is unreliable in agent environments. Each turn is a new connection but passes context through the conversation.

For the first turn, use the one-shot script above. For follow-up turns, include conversation context in the prompt (e.g., prepend prior exchanges).

Response Handling

After each round, parse stdout — only CO_ prefixed lines matter, ignore all others:

• CO_DONE: True → return CO_RESPONSE content to the user. Done.
• CO_DONE: False → the remote agent asked a follow-up question:
  • If you can answer from context (file contents, prior conversation, your own knowledge) → answer automatically. Do not bother the user.
  • If you need the user's input → show CO_RESPONSE to the user, wait for their reply, then send another round.
  • Loop until CO_DONE: True or 10 rounds.

Error Handling

If the script fails, check stderr for these patterns:

Error in stderr	Cause	Action
ImportError: No module named 'connectonion'	Not installed	Run pip install connectonion
address.load() returns None	No identity	Run co init
TimeoutError	Remote agent unreachable or slow	Verify address, check network/proxy, increase timeout
ConnectionRefused or relay lookup fail	Agent offline	Confirm remote agent is running with host()
CO_DIRECT_FAIL + CO_RELAY_FAIL	Both paths failed	Agent likely offline — check with operator
Trust/permission error	Not authorized	Tell user to contact the remote agent admin for access
InsufficientCreditsError	No credits for co/ models	Run co status to check balance
Script hangs (no output for >90s)	Remote agent requesting onboard (invite code/payment)	Kill script, tell user the remote agent requires onboarding