Teleop guide (beyond-line-of-sight teleoperation)

⚠️ Not yet released: in-development v1.5.0-SNAPSHOT capability; interface and protocol may change. Do not use in production before the official release.

New in v1.5.0. Teleop = beyond-line-of-sight remote takeover: a remote operator drives the robot's voice interaction over SDK, independent of local visibility. Third interaction mode alongside passive / active, using its own event family and the AgentPolicy.TELEOP policy. The Python API mirrors Java 1:1 with PEP 8 snake_case naming.

When to use

• Robot is unattended or deployed in poor visibility / network conditions and a human operator needs to take over specific interactions.
• The application needs to bypass the LLM and let a human voice-drive a short window (soothing complaints, verifying orders, etc.).
• The operator's audio is uploaded via SDK and routed by the gateway to the robot addressed by agent_id.

Flow

Strict order: enter → any number of start_audio / append_audio_delta / done_audio → exit. keepalive may be called at any frequency during the session (recommended every 10 seconds, see Robot online detection (keepalive)).

              enter                        exit
   ┌───────────────────────┐       ┌────────────────────┐
   │  TeleopRequest.enter  │       │  TeleopRequest.exit │
   │     (on_enter_ack)    │       │     (on_exit_ack)    │
   └───────────┬───────────┘       └────────────┬────────┘
               │  any number of times            │
               ▼    (fire-and-forget)            │
   ┌───────────────────────┐                     │
   │  start_audio(param)   │                     │
   │  append_audio_delta(*)│─────────────────────┘
   │  done_audio           │
   └───────────────────────┘

   ┌─────────────────────────────┐
   │ TeleopRequest.keepalive     │ ← every 10 s recommended; code != 0 means offline
   │   (on_keepalive_ack)        │
   └─────────────────────────────┘

• enter / keepalive / exit have ack callbacks. code == 0 = success, code == -1 = server rejected (msg carries errorMsg), code == 1000 = local connection closed (callback fires synchronously on caller thread).
• start_audio / append_audio_delta / done_audio are fire-and-forget (no ACK).

Event family & wire format

Python event-type enum lives at linksoul_agentsdk.enums.AgentEventType:

Every teleop envelope carries:

{
  "type": "<event type>",
  "agentId": "<target robot agentId>",
  "teleopId": "teleop_<21 random chars>",
  "eventId": "event_<21 random chars>",
  "agentMode": "teleop"
}

teleopId is generated by IdGenerator.generate_teleop_id(); agentMode is always "teleop" (matches AgentPolicy.TELEOP).

start_audio parameter

start_audio(param=None) accepts an optional AgentParam per audio segment for voice / VAD tuning:

Python construction:

start_param = (
    AgentParam.create()
    .set_string("role", "male")
    .set_integer("threshold", 0)
)
event_id = teleop.start_audio(start_param)

On the wire these travel as param.extParam.role / param.extParam.threshold (matching Java).

Robot online detection (keepalive)

A teleop session usually lasts a while, during which the robot may go offline due to network loss, low battery, or on-site power failure. The SDK exposes TeleopRequest.keepalive(...) as a one-shot probe. Applications should call it periodically (recommended every 10 seconds) to check whether the robot is still online.

Signature & callback

class TeleopKeepaliveCallback:
    """Robot online probe callback"""
    def on_keepalive_ack(self, teleop_id: str, event_id: str, code: int, msg: str) -> None:
        raise NotImplementedError

# TeleopRequest method
def keepalive(self, keepalive_callback: TeleopKeepaliveCallback, timeout: int) -> str: ...

• timeout is a local placeholder (< 50 is clamped to 50, same as enter/exit).
• Returns the outgoing message's event_id.
• Each call sends one agentsdk.teleop.keepalive message; the gateway routes it to the robot identified by teleop_id; the robot responds with success == true when online.

code semantics (same as enter / exit)

Periodic probe + auto exit on offline

Any code != 0 should be treated as "robot went offline" — stop the timer, exit the session, and unregister_teleop to release resources, so no further append_audio_delta is wasted on a dead session.

import threading
from linksoul_agentsdk import TeleopKeepaliveCallback, TeleopExitCallback

def start_keepalive(teleop_request, agent_sdk, interval_sec: int = 10):
    """Call after enter succeeds; probes every interval_sec seconds"""
    state = {"online": True}

    def tick():
        if not state["online"]:
            return

        class _Cb(TeleopKeepaliveCallback):
            def on_keepalive_ack(self, teleop_id, event_id, code, msg):
                if code != 0 and state["online"]:
                    state["online"] = False
                    print(f"robot offline => code={code}, msg={msg}")

                    class _ExitCb(TeleopExitCallback):
                        def on_exit_ack(self, teleop_id, event_id, code, msg):
                            agent_sdk.unregister_teleop(teleop_id)
                    teleop_request.exit(_ExitCb(), 2000)

        teleop_request.keepalive(_Cb(), 2000)
        if state["online"]:
            threading.Timer(interval_sec, tick).start()

    threading.Timer(interval_sec, tick).start()
    return state  # caller may set state["online"] = False to stop probing

Also set state["online"] = False when your business logic invokes exit, to stop further probes.

API cheat-sheet

• Request class: linksoul_agentsdk.TeleopRequest
• Callback bases: linksoul_agentsdk.TeleopEnterCallback / TeleopKeepaliveCallback / TeleopExitCallback
• Register: AgentSdk.register_teleop(TeleopRequest) / AgentSdk.unregister_teleop(teleop_id)
• ID: IdGenerator.generate_teleop_id()

Full signatures: Teleop API. Runnable example: Teleop example.

Lifecycle & cleanup

• AgentSdkTeleopMgr maintains active requests keyed by teleop_id. Always register_teleop(...) before calling enter.
• Every request method call refreshes an internal update_ts; requests idle for 7200 s are GC'd by a background sweeper.
• After exit (or when abandoning the session), call agent_sdk.unregister_teleop(teleop_id) to release eagerly.
• AgentSdk.release() clears all teleop requests owned by this instance (bulk-removed by _linksky_client reference).

Java ↔ Python mapping