Reactor

Reactor is one session, and the tracks and commands on it. It speaks raw JSON over the wire: open a session, send commands by name, receive generic message events. It works against any model — for the commands and events a specific one accepts, see the Model API Reference, or ask the running model itself with requestSchema().

import Reactor

let reactor = try Reactor(model: "reactor/helios", jwt: token)

let statusSubscription = reactor.onStatus { status in print("status:", status) }

try await reactor.connect()
try await reactor.sendCommand("set_prompt", ["prompt": "a mountain landscape"])
try await reactor.disconnect()

Calls are async throws, and failures arrive as a thrown ReactorError — the same type an onError event delivers. final class, not a struct: a session has one owner, and a Reactor released without calling disconnect() orphans the session — the next run cannot start until it clears.

Control-event handlers — onStatus, onError, onMessage, onRuntimeMessage — run on a serial queue this SDK owns, never on the thread the library called on. Pass eventQueue: .main to the initializer to run them on the main queue instead. async calls resume on the library’s own completion thread, deliberately not through that queue — await reactor.connect() from the main actor would otherwise wait on a queue that is waiting for it.

Creating a client

Signature

public convenience init(
    model: String,
    jwt: String? = nil,
    apiURL: String = Reactor.defaultAPIURL,
    local: Bool = false,
    eventQueue: DispatchQueue? = nil
) throws

String

required

The model to connect to, as owner/name. A bare name resolves under reactor/.

String?

A token minted elsewhere, used as it is. For a server that already holds one, or a client handed one by a backend that owns the key.

String

default:"https://api.reactor.inc"

The Reactor API. Reactor.localAPIURL is http://localhost:8080, for a local runtime.

Bool

default:"false"

Accept a local Reactor API’s self-signed certificate and speak its local-development protocol. Pair it with apiURL: Reactor.localAPIURL, or leave apiURL at its default — local: true alone resolves to the local Reactor API.

DispatchQueue?

Where control-event handlers run. nil — the default — means the SDK’s own serial dispatcher queue. Pass .main to have them run on the main queue instead.

init throws — a client is fully constructed or not created at all. The one failure worth naming: ReactorError.versionMismatch when the loaded native library speaks a different ABI than this SDK was built against, which has to be caught before any other call, since past it the stack is corrupted rather than an error reported. For a trusted server holding the raw API key, exchange it first — see fetchJWT(apiKey:apiURL:options:local:) and Authentication — or use the convenience initializer that does both in one step:

Signature

public convenience init(
    model: String,
    apiKey: String,
    apiURL: String = Reactor.defaultAPIURL,
    options: TokenOptions? = nil,
    local: Bool = false,
    eventQueue: DispatchQueue? = nil
) async throws

Example

let reactor = try await Reactor(model: "reactor/helios", apiKey: apiKey)

Never ship an API key to an end-user’s device. Mint a short-lived JWT on your server and construct with jwt: instead. See Authentication.

Connecting

`connect(sessionID:connectionID:)`

Creates — or adopts — a session and brings up the transport. Resolves once the session is .ready.

Signature

public func connect(sessionID: String? = nil, connectionID: UInt32? = nil) async throws

String?

Join a session that already exists rather than creating one. This is how a second client attaches to the same session. A session adopted this way is not ended by disconnect() — it keeps running for its owner.

UInt32?

Adopt a connection slot a backend already registered for this session. The connection-level analogue of sessionID; most callers building one connection per session leave it nil. See Multiple connections per session.

Throws ReactorError.unauthorized for a token problem, ReactorError.conflict for a session a previous run left orphaned.

Example

try await secondClient.connect(sessionID: existingSession)

`reconnect()`

Cycles the connection without ending the session — after a transient failure, or deliberately from .ready.

Signature

public func reconnect() async throws

It tears the live connection down itself, so there is no need to disconnect() first — and doing so would end the very session this is about to reuse. Throws when there is no session to reconnect to.

Recvonly tracks resume automatically. Sendonly tracks do not: a track published before the reconnect is not published after it — publish again. Track/published says which side of that you are on, and pushFrame throws rather than pushing into a slot with nothing behind it.

try await reactor.reconnect()
try await camera.publish()

`disconnect()`

Ends the session server-side and tears down the transport.

Signature

public func disconnect() async throws

Not recoverable — there is no parameter that keeps the session alive instead. To disconnect and later resume the same session, call reconnect(). Only ends sessions this client created; one adopted via sessionID is left running for its owner.

`close()`

Releases the native handle immediately, without ending the session server-side.

Signature

public func close()

Idempotent, and called for you when the last reference to a Reactor goes away — deinit calls it. Call it yourself to release resources at a known point. This does not end the session; use disconnect() for that. A creator that goes away without disconnecting orphans the session, and the next run cannot start until it clears.

`status`

Signature

public var status: ReactorStatus { get }

Where the session is now, as a ReactorStatus. Readable before connect() — a client that never connected reports .disconnected rather than nothing.

`sessionID`

Signature

public var sessionID: String? { get }

The session’s id, once there is one.

Commands and uploads

`sendCommand(_:_:uploads:)`

Sends a command to the model and waits for its correlated reply.

Signature

@discardableResult
public func sendCommand(
    _ name: String,
    _ args: JSONValue? = nil,
    uploads: [String: FileRef] = [:]
) async throws -> CommandReply?

String

required

The command name. Must match a command the model defines.

JSONValue?

The payload. JSONValue is ExpressibleByDictionaryLiteral, so a Swift dictionary literal works directly.

[String: FileRef]

Files to pass as named parameters — see uploadFile(at:).

The reply is a CommandReply, or nil when the handler ran and acknowledged the command without returning a message, as an auto-generated set_<field> setter does. nil is not a failure and is not folded into one:

Example

if let reply = try await reactor.sendCommand("set_prompt", ["prompt": "a mountain landscape"]) {
    print(reply.data ?? "no data")
}

A typed twin takes an Encodable argument instead of a JSONValue dictionary — the arguments: label is required, since JSONValue is itself Encodable and an unlabelled overload would be ambiguous with the one above:

Signature

@discardableResult
public func sendCommand(
    _ name: String,
    arguments: some Encodable & Sendable,
    uploads: [String: FileRef] = [:]
) async throws -> CommandReply?

`uploadFile(at:)`

Uploads a local file and returns a FileRef to pass into a command.

Signature

public func uploadFile(at url: URL) async throws -> FileRef

Needs a .ready session — the upload is created against it.

Example

let ref = try await reactor.uploadFile(at: photoURL)
try await reactor.sendCommand("set_image", uploads: ["image": ref])

`uploadData(_:name:mimeType:)`

The same result as uploadFile(at:), for a caller who has the bytes rather than a path — a frame just rendered, a buffer just decoded.

Signature

public func uploadData(_ data: Data, name: String, mimeType: String) async throws -> FileRef

Example

let ref = try await reactor.uploadData(pngData, name: "frame.png", mimeType: "image/png")

`requestSchema()`

Signature

public func requestSchema() async throws -> JSONValue

The model’s command schema, as an OpenAPI document — the same schema published on the Model API Reference pages, fetched from the running model. What to read when a command is rejected: it is the model’s own account of what it accepts, which is more current than any documentation.

Recordings

`requestClip(_:)`

Asks for a clip covering the last duration of the session.

Signature

public func requestClip(_ duration: Duration) async throws -> Clip

Resolves when the platform has accepted the request, which is not the same as the clip being ready — download(_:to:readyTimeout:progress:) is what waits for that.

Example

let clip = try await reactor.requestClip(.seconds(10))
try await reactor.download(clip, to: destinationURL)

See Recordings for the full flow.

`requestRecording()`

Signature

public func requestRecording() async throws -> Clip

The same, covering the whole session up to now.

`download(_:to:readyTimeout:progress:)`

Downloads a clip’s segments into one playable file.

Signature

@discardableResult
public func download(
    _ clip: Clip,
    to url: URL,
    readyTimeout: Duration? = nil,
    progress: (@Sendable (DownloadProgress) -> Void)? = nil
) async throws -> DownloadResult

Clip

required

What requestClip(_:) or requestRecording() answered.

URL

required

The file to create. Opened before the first segment is fetched, so an unwritable path fails early.

Duration?

How long to keep waiting past the runtime’s own prediction. nil — the default — waits as long as the session can still produce the clip, the only sane answer for a model generating slower than real time: a clip becomes ready because the model keeps generating, so once the session is gone a “not ready” is a “not ready” forever.

((DownloadProgress) -> Void)?

Called after each segment is written, on the download’s own thread.

Example

let result = try await reactor.download(clip, to: destinationURL) { progress in
    print("\(progress.done)/\(progress.total)")
}

This download outlives the client. If the client is closed mid-download the call fails with a message saying the file may still arrive — because it may.

Tracks

`tracks`

Every track the session declared, as a TrackList — for discovery, and for a caller who would rather not hardcode a name.

Signature

public var tracks: TrackList { get }

Example

let output = try reactor.tracks.withDirection(.recvonly).withKind(.video).one()

Empty until the model’s capabilities arrive, shortly after connect().

`track(_:)`

The track called name, as a Track — the only way to push frames into one, receive its frames, or pause it.

Signature

public func track(_ name: String) throws -> Track

Example

let camera = try reactor.track("camera")
try await camera.publish()

Throws ReactorError.notFound, naming what the session does declare, for a name that is not among them. Before the session has declared anything, any name is accepted — that is what lets handlers be registered ahead of connecting.

`pausedTracks`

Signature

public var pausedTracks: Set<String> { get }

The names of the currently paused tracks. Recvonly tracks resume automatically once connected, so this is empty on a healthy session until the caller pauses something.

Events

Every on* method returns a Subscription: a token that cancels the handler when it is released.

A subscription cancels when it is released, not when the block returns. _ = reactor.onStatus { print($0) } registers a handler and cancels it on the same line, because nothing holds the token. Store the result somewhere that outlives the period you want the handler to fire — a property, an array. The registration methods are deliberately not @discardableResult, so the compiler warns rather than letting this pass silently.

Handler	Payload	Fires when
`onStatus`	`ReactorStatus`	The connection status changed
`onMessage`	`JSONValue`	The model sent an application message
`onRuntimeMessage`	`JSONValue`	The platform sent one — session lifecycle notices, clip readiness
`onError`	`ReactorError`	A failure arrived that no call was waiting on

Signature

public func onStatus(_ handler: @escaping @Sendable (ReactorStatus) -> Void) -> Subscription
public func onMessage(_ handler: @escaping @Sendable (JSONValue) -> Void) -> Subscription
public func onRuntimeMessage(_ handler: @escaping @Sendable (JSONValue) -> Void) -> Subscription
public func onError(_ handler: @escaping @Sendable (ReactorError) -> Void) -> Subscription

Example

let messages = reactor.onMessage { message in
    print(message["type"]?.stringValue ?? "")
}

let errors = reactor.onError { error in
    if error.recoverable { /* reconnect */ }
}

Model messages and platform messages are separate events because they are separate things: a caller reading only onMessage never has to filter the platform’s out of it.

There is no client-wide frame event. Media is delivered per track, through Track/onFrame(_:) — a single handler fed every incoming track at once could not tell them apart.

onError hands you a ReactorError, the same type a failed call throws. Match on code, or branch on recoverable when the specific code does not matter — or pattern-match directly: catch ReactorError.unauthorized. See ReactorError.

For callers who prefer for await over a closure, statusUpdates, errors, and messages expose the same events as AsyncStreams:

Signature

public var statusUpdates: AsyncStream<ReactorStatus> { get }
public var errors: AsyncStream<ReactorError> { get }
public var messages: AsyncStream<JSONValue> { get }

`Reactor.timeMicros()`

The engine’s monotonic clock, in microseconds — the epoch a frame’s capture time is read in.

Signature

public static func timeMicros() -> Int64

Read it once per unit of produced media and stamp every track with that one value: tracks are synchronised by sharing a capture time, not by reaching the encoder at the same moment.

Example

let capturedAt = Reactor.timeMicros()
try camera.pushFrame(bgra, width: 1280, height: 720, captureTimeUs: capturedAt)
try micTrack.pushFrame(samples)

Unrelated to the system clock — a UNIX timestamp is not a substitute.

`Reactor.fetchJWT(apiKey:apiURL:options:local:)`

Exchange an API key for a JWT, without creating a client.

Signature

public static func fetchJWT(
    apiKey: String,
    apiURL: String = Reactor.defaultAPIURL,
    options: TokenOptions? = nil,
    local: Bool = false
) async throws -> String

Example

let token = try await Reactor.fetchJWT(
    apiKey: apiKey, options: .init(models: ["reactor/helios"]))

For a server minting tokens for browser or native clients it does not control directly — the same role the server-side proxy plays for the browser SDK. See TokenOptions and Authentication.

Get Started

Concepts

SDK Reference

Resources

Creating a client

Connecting

`connect(sessionID:connectionID:)`

`reconnect()`

`disconnect()`

`close()`

`status`

`sessionID`

Commands and uploads

`sendCommand(_:_:uploads:)`

`uploadFile(at:)`

`uploadData(_:name:mimeType:)`

`requestSchema()`

Recordings

`requestClip(_:)`

`requestRecording()`

`download(_:to:readyTimeout:progress:)`

Tracks

`tracks`

`track(_:)`

`pausedTracks`

Events

`Reactor.timeMicros()`

`Reactor.fetchJWT(apiKey:apiURL:options:local:)`

​Creating a client

​Connecting

​connect(sessionID:connectionID:)

​reconnect()

​disconnect()

​close()

​status

​sessionID

​Commands and uploads

​sendCommand(_:_:uploads:)

​uploadFile(at:)

​uploadData(_:name:mimeType:)

​requestSchema()

​Recordings

​requestClip(_:)

​requestRecording()

​download(_:to:readyTimeout:progress:)

​Tracks

​tracks

​track(_:)

​pausedTracks

​Events

​Reactor.timeMicros()

​Reactor.fetchJWT(apiKey:apiURL:options:local:)

Creating a client

Connecting

`connect(sessionID:connectionID:)`

`reconnect()`

`disconnect()`

`close()`

`status`

`sessionID`

Commands and uploads

`sendCommand(_:_:uploads:)`

`uploadFile(at:)`

`uploadData(_:name:mimeType:)`

`requestSchema()`

Recordings

`requestClip(_:)`

`requestRecording()`

`download(_:to:readyTimeout:progress:)`

Tracks

`tracks`

`track(_:)`

`pausedTracks`

Events

`Reactor.timeMicros()`

`Reactor.fetchJWT(apiKey:apiURL:options:local:)`