mirror of
https://github.com/stablyai/orca.git
synced 2026-10-12 16:03:03 +00:00
Improve mobile terminal streaming performance
Co-authored-by: Orca <help@stably.ai>
This commit is contained in:
@@ -97,6 +97,7 @@ const STATUS_LABELS: Record<ConnectionState, string> = {
|
||||
function TerminalPaneView({
|
||||
handle,
|
||||
active,
|
||||
keyboardLift,
|
||||
onRef,
|
||||
onWebReady,
|
||||
onSelectionMode,
|
||||
@@ -107,6 +108,7 @@ function TerminalPaneView({
|
||||
}: {
|
||||
handle: string
|
||||
active: boolean
|
||||
keyboardLift: number
|
||||
onRef: (handle: string, ref: TerminalWebViewHandle | null) => void
|
||||
onWebReady: (handle: string) => void
|
||||
onSelectionMode: (handle: string, active: boolean) => void
|
||||
@@ -125,7 +127,11 @@ function TerminalPaneView({
|
||||
return (
|
||||
<View
|
||||
pointerEvents={active ? 'auto' : 'none'}
|
||||
style={[styles.terminalPane, !active && styles.terminalPaneHidden]}
|
||||
style={[
|
||||
styles.terminalPane,
|
||||
keyboardLift > 0 && { transform: [{ translateY: -keyboardLift }] },
|
||||
!active && styles.terminalPaneHidden
|
||||
]}
|
||||
>
|
||||
<TerminalWebView
|
||||
ref={setRef}
|
||||
@@ -171,8 +177,9 @@ export default function SessionScreen() {
|
||||
const [deleteKeyTarget, setDeleteKeyTarget] = useState<CustomKey | null>(null)
|
||||
// Why: in Expo SDK 55 edge-to-edge mode the OS does NOT resize the window when
|
||||
// the IME opens — the keyboard draws on top of the app. We track the keyboard
|
||||
// height ourselves and apply it as paddingBottom on the input/accessory area
|
||||
// so the input lifts above the IME and the terminal flex container shrinks.
|
||||
// height ourselves and translate the input/accessory area above the IME without
|
||||
// changing the terminal frame height, so keyboard open/close does not resize
|
||||
// the desktop PTY.
|
||||
const [keyboardHeight, setKeyboardHeight] = useState(0)
|
||||
// Why: server-authoritative display mode per terminal. The runtime is the
|
||||
// single source of truth — this state is populated from subscribe responses.
|
||||
@@ -286,7 +293,8 @@ export default function SessionScreen() {
|
||||
{
|
||||
terminal: handle,
|
||||
client: { id: deviceTokenRef.current!, type: 'mobile' as const },
|
||||
viewport: viewportRef.current ?? undefined
|
||||
viewport: viewportRef.current ?? undefined,
|
||||
capabilities: { terminalBinaryStream: 1 }
|
||||
},
|
||||
(result) => {
|
||||
if (subscribeSeqRef.current.get(handle) !== seq) return
|
||||
@@ -319,6 +327,9 @@ export default function SessionScreen() {
|
||||
} else if (eventSeq != null && data.type === 'scrollback') {
|
||||
layoutSeqRef.current.set(handle, eventSeq)
|
||||
}
|
||||
if (data.type === 'subscribed') {
|
||||
return
|
||||
}
|
||||
if (data.type === 'scrollback') {
|
||||
if (initializedHandlesRef.current.has(handle)) {
|
||||
return
|
||||
@@ -439,11 +450,12 @@ export default function SessionScreen() {
|
||||
// dims with fresh scrollback. No resubscribe needed.
|
||||
const cols = (data.cols as number) || 80
|
||||
const rows = (data.rows as number) || 24
|
||||
const serialized =
|
||||
typeof data.serialized === 'string' && data.serialized.length > 0
|
||||
? data.serialized
|
||||
: ''
|
||||
getTerminalRef(handle)?.init(cols, rows, serialized)
|
||||
const serialized = typeof data.serialized === 'string' ? data.serialized : null
|
||||
if (serialized != null) {
|
||||
getTerminalRef(handle)?.init(cols, rows, serialized)
|
||||
} else {
|
||||
getTerminalRef(handle)?.resize(cols, rows)
|
||||
}
|
||||
if (data.displayMode) {
|
||||
setTerminalModes((prev) =>
|
||||
new Map(prev).set(handle, data.displayMode as MobileDisplayMode)
|
||||
@@ -612,11 +624,7 @@ export default function SessionScreen() {
|
||||
void loadCustomKeys().then(setCustomKeys)
|
||||
}, [])
|
||||
|
||||
// Why: drive the bottom padding from the keyboard height (edge-to-edge mode
|
||||
// doesn't resize the window) and refit xterm once the layout settles so the
|
||||
// terminal grid matches the new visible area. iOS exposes 'will' events that
|
||||
// animate in sync with the IME; Android only fires 'did' events reliably.
|
||||
// Also drives re-measurement when other layout-affecting state changes
|
||||
// Why: re-measure when non-keyboard layout-affecting state changes
|
||||
// (e.g. tab strip toggling visibility when the terminal count crosses
|
||||
// 0↔1 — without this, a freshly-created 2nd tab subscribes with a
|
||||
// stale viewport that doesn't account for the now-visible tab strip,
|
||||
@@ -665,11 +673,9 @@ export default function SessionScreen() {
|
||||
useEffect(() => {
|
||||
const onShow = (e: KeyboardEvent) => {
|
||||
setKeyboardHeight(e.endCoordinates?.height ?? 0)
|
||||
scheduleViewportRefit()
|
||||
}
|
||||
const onHide = () => {
|
||||
setKeyboardHeight(0)
|
||||
scheduleViewportRefit()
|
||||
}
|
||||
const showEvent = Platform.OS === 'ios' ? 'keyboardWillShow' : 'keyboardDidShow'
|
||||
const hideEvent = Platform.OS === 'ios' ? 'keyboardWillHide' : 'keyboardDidHide'
|
||||
@@ -680,7 +686,7 @@ export default function SessionScreen() {
|
||||
showSub.remove()
|
||||
hideSub.remove()
|
||||
}
|
||||
}, [scheduleViewportRefit])
|
||||
}, [])
|
||||
|
||||
// Why: the tab strip is hidden when only one terminal exists and shown
|
||||
// once a second is created. Crossing the 1↔2 boundary changes the
|
||||
@@ -726,13 +732,12 @@ export default function SessionScreen() {
|
||||
|
||||
useEffect(() => {
|
||||
if (connState !== 'connected') return
|
||||
// Why: on reconnect the RPC client auto-resends terminal.subscribe,
|
||||
// creating new server-side handlers. Clear local subscription state
|
||||
// so subscribeToTerminal's guards don't block fresh subscriptions,
|
||||
// and clear xterm buffers so the new scrollback snapshot replaces
|
||||
// stale content (including data that arrived while disconnected).
|
||||
clearTerminalCache()
|
||||
setTerminalsLoaded(false)
|
||||
// Why: the RPC client auto-resends terminal.subscribe on reconnect.
|
||||
// Keep the current xterm visible while the binary snapshot hydrates,
|
||||
// instead of clearing to a blank "Loading terminals" surface.
|
||||
if (initializedHandlesRef.current.size === 0) {
|
||||
setTerminalsLoaded(false)
|
||||
}
|
||||
let disposed = false
|
||||
const timers: ReturnType<typeof setTimeout>[] = []
|
||||
function addTimer(fn: () => void, ms: number) {
|
||||
@@ -1200,17 +1205,15 @@ export default function SessionScreen() {
|
||||
: `${terminals.length} terminals`
|
||||
: STATUS_LABELS[connState]
|
||||
|
||||
// Why: on Android (Samsung 3-button) the keyboard event reports only the IME
|
||||
// height; the system nav bar sits below the keyboard and adds its own height
|
||||
// on top, so we must add insets.bottom too or the input stays clipped behind
|
||||
// the nav bar. On iOS the keyboard coordinates already include the home
|
||||
// indicator region, so adding insets.bottom would double-count.
|
||||
const bottomPadding =
|
||||
// Why: keep safe-area padding in layout at all times, then visually translate
|
||||
// the controls over the terminal when the keyboard appears. iOS keyboard
|
||||
// height includes the home-indicator inset; Android IME height does not.
|
||||
const keyboardLift =
|
||||
keyboardHeight > 0
|
||||
? Platform.OS === 'ios'
|
||||
? keyboardHeight
|
||||
: keyboardHeight + insets.bottom
|
||||
: insets.bottom
|
||||
? Math.max(0, keyboardHeight - insets.bottom)
|
||||
: keyboardHeight
|
||||
: 0
|
||||
|
||||
return (
|
||||
<View style={styles.container}>
|
||||
@@ -1313,6 +1316,7 @@ export default function SessionScreen() {
|
||||
key={terminal.handle}
|
||||
handle={terminal.handle}
|
||||
active={terminal.handle === activeHandle}
|
||||
keyboardLift={keyboardLift}
|
||||
onRef={setTerminalWebViewRef}
|
||||
onWebReady={handleTerminalWebReady}
|
||||
onSelectionMode={handleSelectionMode}
|
||||
@@ -1333,9 +1337,14 @@ export default function SessionScreen() {
|
||||
</View>
|
||||
)}
|
||||
|
||||
{/* Why: bottomPadding lifts the entire input region above the keyboard
|
||||
(when shown) or above the system nav bar / home indicator (when hidden). */}
|
||||
<View style={{ paddingBottom: bottomPadding }}>
|
||||
{/* Why: translate instead of resizing so keyboard open/close does not
|
||||
trigger a server-side PTY viewport change. */}
|
||||
<View
|
||||
style={[
|
||||
styles.commandDock,
|
||||
{ paddingBottom: insets.bottom, transform: [{ translateY: -keyboardLift }] }
|
||||
]}
|
||||
>
|
||||
{/* Accessory keys */}
|
||||
<View style={styles.accessoryBar}>
|
||||
<ScrollView
|
||||
@@ -1719,6 +1728,9 @@ const styles = StyleSheet.create({
|
||||
fontSize: typography.bodySize,
|
||||
fontWeight: '600'
|
||||
},
|
||||
commandDock: {
|
||||
zIndex: 20
|
||||
},
|
||||
accessoryBar: {
|
||||
borderTopWidth: 1,
|
||||
borderTopColor: colors.borderSubtle,
|
||||
|
||||
@@ -20,6 +20,7 @@ export type TerminalSelectionEvents = {
|
||||
export type TerminalWebViewHandle = {
|
||||
write: (data: string) => void
|
||||
init: (cols: number, rows: number, initialData?: string) => void
|
||||
resize: (cols: number, rows: number) => void
|
||||
clear: () => void
|
||||
measureFitDimensions: (containerHeight?: number) => Promise<{ cols: number; rows: number } | null>
|
||||
resetZoom: () => void
|
||||
@@ -40,6 +41,7 @@ type Props = {
|
||||
type TerminalMessage =
|
||||
| { type: 'write'; id?: number; data: string }
|
||||
| { type: 'init'; id?: number; cols: number; rows: number; initialData?: string }
|
||||
| { type: 'resize'; id?: number; cols: number; rows: number }
|
||||
| { type: 'clear'; id?: number }
|
||||
| { type: 'measure'; id?: number; containerHeight?: number }
|
||||
| { type: 'reset-zoom'; id?: number }
|
||||
@@ -432,7 +434,20 @@ const XTERM_HTML = `<!DOCTYPE html>
|
||||
firstDataPending = true;
|
||||
var replayData = normalizeInitialData(initialData);
|
||||
activeAltScreenSnapshot = isAltScreenActive(replayData);
|
||||
if (term) term.dispose();
|
||||
var oldTerm = term;
|
||||
var oldSurface = surface;
|
||||
var nextSurface = null;
|
||||
if (oldTerm) {
|
||||
nextSurface = document.createElement('div');
|
||||
nextSurface.id = 'terminal-surface';
|
||||
nextSurface.style.visibility = 'hidden';
|
||||
nextSurface.style.position = 'absolute';
|
||||
nextSurface.style.left = '0';
|
||||
nextSurface.style.top = '0';
|
||||
document.getElementById('terminal-container').appendChild(nextSurface);
|
||||
surface = nextSurface;
|
||||
oldSurface.removeAttribute('id');
|
||||
}
|
||||
|
||||
term = new Terminal({
|
||||
cols: cols || 80,
|
||||
@@ -485,6 +500,14 @@ const XTERM_HTML = `<!DOCTYPE html>
|
||||
ready = true;
|
||||
afterWritesDrained(function() {
|
||||
if (gen !== terminalGeneration) return;
|
||||
if (nextSurface && oldSurface) {
|
||||
nextSurface.style.visibility = 'visible';
|
||||
nextSurface.style.position = '';
|
||||
nextSurface.style.left = '';
|
||||
nextSurface.style.top = '';
|
||||
oldSurface.remove();
|
||||
if (oldTerm) oldTerm.dispose();
|
||||
}
|
||||
applyFitScale('init-replay');
|
||||
notify({ type: 'ready', cols: cols, rows: rows });
|
||||
});
|
||||
@@ -508,6 +531,14 @@ const XTERM_HTML = `<!DOCTYPE html>
|
||||
}
|
||||
}
|
||||
|
||||
function resize(cols, rows) {
|
||||
if (!term) return;
|
||||
initRows = rows || initRows;
|
||||
term.resize(cols || term.cols, rows || term.rows);
|
||||
applyFitScale('resize-msg');
|
||||
notify({ type: 'ready', cols: cols, rows: rows });
|
||||
}
|
||||
|
||||
function notify(msg) {
|
||||
if (window.ReactNativeWebView) {
|
||||
window.ReactNativeWebView.postMessage(JSON.stringify(msg));
|
||||
@@ -576,6 +607,8 @@ const XTERM_HTML = `<!DOCTYPE html>
|
||||
}
|
||||
if (msg.type === 'init') {
|
||||
init(msg.cols, msg.rows, msg.initialData);
|
||||
} else if (msg.type === 'resize') {
|
||||
resize(msg.cols, msg.rows);
|
||||
} else if (msg.type === 'write') {
|
||||
write(msg.data);
|
||||
} else if (msg.type === 'clear') {
|
||||
@@ -1393,6 +1426,9 @@ export const TerminalWebView = forwardRef<TerminalWebViewHandle, Props>(function
|
||||
})
|
||||
postMessage({ type: 'init', cols, rows, initialData })
|
||||
},
|
||||
resize(cols: number, rows: number) {
|
||||
postMessage({ type: 'resize', cols, rows })
|
||||
},
|
||||
clear() {
|
||||
postMessage({ type: 'clear' })
|
||||
},
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
// Why: E2EE primitives for the mobile side. Uses tweetnacl for Curve25519 ECDH
|
||||
// key exchange and XSalsa20-Poly1305 authenticated encryption. Wire format is
|
||||
// base64([24-byte nonce][ciphertext]) over WebSocket text frames.
|
||||
// key exchange and XSalsa20-Poly1305 authenticated encryption. JSON RPC uses
|
||||
// base64([24-byte nonce][ciphertext]) over WebSocket text frames; terminal
|
||||
// stream frames use the raw byte bundle.
|
||||
import nacl from 'tweetnacl'
|
||||
import * as ExpoCrypto from 'expo-crypto'
|
||||
|
||||
@@ -62,19 +63,28 @@ export function publicKeyToBase64(key: Uint8Array): string {
|
||||
}
|
||||
|
||||
export function encrypt(plaintext: string, sharedKey: Uint8Array): string {
|
||||
const nonce = u8(nacl.randomBytes(nacl.box.nonceLength))
|
||||
const messageBytes = u8(new TextEncoder().encode(plaintext))
|
||||
const ciphertext = nacl.box.after(messageBytes, nonce, u8(sharedKey))
|
||||
return uint8ToBase64(encryptBytes(messageBytes, sharedKey))
|
||||
}
|
||||
|
||||
export function decrypt(encrypted: string, sharedKey: Uint8Array): string | null {
|
||||
const bundle = base64ToUint8(encrypted)
|
||||
const plaintext = decryptBytes(bundle, sharedKey)
|
||||
return plaintext ? new TextDecoder().decode(plaintext) : null
|
||||
}
|
||||
|
||||
export function encryptBytes(plaintext: Uint8Array, sharedKey: Uint8Array): Uint8Array {
|
||||
const nonce = u8(nacl.randomBytes(nacl.box.nonceLength))
|
||||
const ciphertext = nacl.box.after(u8(plaintext), nonce, u8(sharedKey))
|
||||
|
||||
const bundle = new Uint8Array(nonce.length + ciphertext.length)
|
||||
bundle.set(nonce)
|
||||
bundle.set(ciphertext, nonce.length)
|
||||
|
||||
return uint8ToBase64(bundle)
|
||||
return bundle
|
||||
}
|
||||
|
||||
export function decrypt(encrypted: string, sharedKey: Uint8Array): string | null {
|
||||
const bundle = base64ToUint8(encrypted)
|
||||
export function decryptBytes(bundle: Uint8Array, sharedKey: Uint8Array): Uint8Array | null {
|
||||
if (bundle.length < nacl.box.nonceLength + nacl.box.overheadLength) {
|
||||
return null
|
||||
}
|
||||
@@ -87,5 +97,5 @@ export function decrypt(encrypted: string, sharedKey: Uint8Array): string | null
|
||||
return null
|
||||
}
|
||||
|
||||
return new TextDecoder().decode(plaintext)
|
||||
return u8(plaintext)
|
||||
}
|
||||
|
||||
@@ -16,5 +16,5 @@
|
||||
// desktop feature added at a specific desktop protocol version. This
|
||||
// triggers a hard-block screen for users paired to older desktops.
|
||||
|
||||
export const MOBILE_PROTOCOL_VERSION = 1
|
||||
export const MIN_COMPATIBLE_DESKTOP_VERSION = 0
|
||||
export const MOBILE_PROTOCOL_VERSION = 2
|
||||
export const MIN_COMPATIBLE_DESKTOP_VERSION = 2
|
||||
|
||||
@@ -11,8 +11,15 @@ import {
|
||||
publicKeyFromBase64,
|
||||
publicKeyToBase64,
|
||||
encrypt,
|
||||
decrypt
|
||||
decrypt,
|
||||
decryptBytes
|
||||
} from './e2ee'
|
||||
import {
|
||||
TerminalStreamOpcode,
|
||||
decodeTerminalStreamFrame,
|
||||
decodeTerminalStreamJson,
|
||||
decodeTerminalStreamText
|
||||
} from './terminal-stream-protocol'
|
||||
|
||||
type PendingRequest = {
|
||||
resolve: (response: RpcResponse) => void
|
||||
@@ -27,6 +34,12 @@ type StreamRequest = {
|
||||
listener: StreamingListener
|
||||
}
|
||||
|
||||
type TerminalSnapshotState = {
|
||||
streamId: number
|
||||
meta: Record<string, unknown>
|
||||
chunks: string[]
|
||||
}
|
||||
|
||||
export type RpcClient = {
|
||||
sendRequest: (method: string, params?: unknown) => Promise<RpcResponse>
|
||||
subscribe: (method: string, params: unknown, onData: StreamingListener) => () => void
|
||||
@@ -144,6 +157,9 @@ export function connect(
|
||||
|
||||
const pending = new Map<string, PendingRequest>()
|
||||
const streamListeners = new Map<string, StreamRequest>()
|
||||
const terminalStreamListeners = new Map<number, StreamingListener>()
|
||||
const terminalStreamIdsByRequest = new Map<string, Set<number>>()
|
||||
const terminalSnapshots = new Map<number, TerminalSnapshotState>()
|
||||
const stateListeners = new Set<(state: ConnectionState) => void>()
|
||||
const connectWaiters: Array<{ resolve: () => void; reject: (e: Error) => void }> = []
|
||||
|
||||
@@ -293,14 +309,21 @@ export function connect(
|
||||
}
|
||||
|
||||
ws.onmessage = (event) => {
|
||||
void handleSocketMessage(event.data)
|
||||
}
|
||||
|
||||
async function handleSocketMessage(rawData: unknown) {
|
||||
// Why: track last-inbound for the openConnection diagnostic. Server
|
||||
// pongs and stream events both bump this — anything from the wire.
|
||||
lastInboundAt = Date.now()
|
||||
const raw = typeof event.data === 'string' ? event.data : String(event.data)
|
||||
const raw = typeof rawData === 'string' ? rawData : null
|
||||
|
||||
// Why: during handshaking, e2ee_ready is plaintext because it precedes
|
||||
// encrypted auth; e2ee_authenticated/e2ee_error are encrypted.
|
||||
if (state === 'handshaking') {
|
||||
if (raw === null) {
|
||||
return
|
||||
}
|
||||
try {
|
||||
const msg = JSON.parse(raw)
|
||||
if (msg.type === 'e2ee_ready') {
|
||||
@@ -362,6 +385,19 @@ export function connect(
|
||||
return
|
||||
}
|
||||
|
||||
if (raw === null) {
|
||||
const bytes = await websocketPayloadToUint8(rawData)
|
||||
if (!bytes) {
|
||||
return
|
||||
}
|
||||
const plaintextBytes = decryptBytes(bytes, sharedKey)
|
||||
if (!plaintextBytes) {
|
||||
return
|
||||
}
|
||||
handleTerminalBinaryFrame(plaintextBytes)
|
||||
return
|
||||
}
|
||||
|
||||
const plaintext = decrypt(raw, sharedKey)
|
||||
if (plaintext === null) {
|
||||
return
|
||||
@@ -390,7 +426,17 @@ export function connect(
|
||||
if (isStreaming) {
|
||||
const stream = streamListeners.get(response.id)
|
||||
if (stream && response.ok) {
|
||||
stream.listener((response as RpcSuccess).result)
|
||||
const result = (response as RpcSuccess).result
|
||||
if (isTerminalSubscribedResult(result)) {
|
||||
let ids = terminalStreamIdsByRequest.get(response.id)
|
||||
if (!ids) {
|
||||
ids = new Set()
|
||||
terminalStreamIdsByRequest.set(response.id, ids)
|
||||
}
|
||||
ids.add(result.streamId)
|
||||
terminalStreamListeners.set(result.streamId, stream.listener)
|
||||
}
|
||||
stream.listener(result)
|
||||
}
|
||||
return
|
||||
}
|
||||
@@ -654,6 +700,75 @@ export function connect(
|
||||
}
|
||||
}
|
||||
|
||||
function handleTerminalBinaryFrame(bytes: Uint8Array) {
|
||||
const frame = decodeTerminalStreamFrame(bytes)
|
||||
if (!frame) {
|
||||
return
|
||||
}
|
||||
const listener = terminalStreamListeners.get(frame.streamId)
|
||||
if (!listener) {
|
||||
return
|
||||
}
|
||||
if (frame.opcode === TerminalStreamOpcode.Output) {
|
||||
listener({
|
||||
type: 'data',
|
||||
streamId: frame.streamId,
|
||||
chunk: decodeTerminalStreamText(frame.payload)
|
||||
})
|
||||
return
|
||||
}
|
||||
if (frame.opcode === TerminalStreamOpcode.SnapshotStart) {
|
||||
const meta = decodeTerminalStreamJson<Record<string, unknown>>(frame.payload)
|
||||
if (!meta) {
|
||||
return
|
||||
}
|
||||
terminalSnapshots.set(frame.streamId, { streamId: frame.streamId, meta, chunks: [] })
|
||||
return
|
||||
}
|
||||
if (frame.opcode === TerminalStreamOpcode.SnapshotChunk) {
|
||||
const snapshot = terminalSnapshots.get(frame.streamId)
|
||||
if (!snapshot) {
|
||||
return
|
||||
}
|
||||
snapshot.chunks.push(decodeTerminalStreamText(frame.payload))
|
||||
return
|
||||
}
|
||||
if (frame.opcode === TerminalStreamOpcode.SnapshotEnd) {
|
||||
const snapshot = terminalSnapshots.get(frame.streamId)
|
||||
if (!snapshot) {
|
||||
return
|
||||
}
|
||||
terminalSnapshots.delete(frame.streamId)
|
||||
const kind = snapshot.meta.kind === 'resized' ? 'resized' : 'scrollback'
|
||||
listener({
|
||||
...snapshot.meta,
|
||||
type: kind,
|
||||
streamId: frame.streamId,
|
||||
serialized: snapshot.chunks.join('')
|
||||
})
|
||||
return
|
||||
}
|
||||
if (frame.opcode === TerminalStreamOpcode.Resized) {
|
||||
const meta = decodeTerminalStreamJson<Record<string, unknown>>(frame.payload)
|
||||
if (!meta) {
|
||||
return
|
||||
}
|
||||
listener({
|
||||
...meta,
|
||||
type: 'resized',
|
||||
streamId: frame.streamId
|
||||
})
|
||||
return
|
||||
}
|
||||
if (frame.opcode === TerminalStreamOpcode.Error) {
|
||||
listener({
|
||||
type: 'error',
|
||||
streamId: frame.streamId,
|
||||
message: decodeTerminalStreamText(frame.payload)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
function sendEncrypted(request: unknown): boolean {
|
||||
if (ws && ws.readyState === WebSocket.OPEN && sharedKey) {
|
||||
ws.send(encrypt(JSON.stringify(request), sharedKey))
|
||||
@@ -740,6 +855,14 @@ export function connect(
|
||||
return () => {
|
||||
const stream = streamListeners.get(id)
|
||||
streamListeners.delete(id)
|
||||
const terminalStreamIds = terminalStreamIdsByRequest.get(id)
|
||||
if (terminalStreamIds) {
|
||||
for (const streamId of terminalStreamIds) {
|
||||
terminalStreamListeners.delete(streamId)
|
||||
terminalSnapshots.delete(streamId)
|
||||
}
|
||||
terminalStreamIdsByRequest.delete(id)
|
||||
}
|
||||
if (
|
||||
stream?.method === 'terminal.subscribe' &&
|
||||
stream.params &&
|
||||
@@ -813,3 +936,38 @@ export function connect(
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
function isTerminalSubscribedResult(
|
||||
value: unknown
|
||||
): value is { type: 'subscribed'; streamId: number } {
|
||||
return (
|
||||
!!value &&
|
||||
typeof value === 'object' &&
|
||||
(value as { type?: unknown }).type === 'subscribed' &&
|
||||
typeof (value as { streamId?: unknown }).streamId === 'number'
|
||||
)
|
||||
}
|
||||
|
||||
async function websocketPayloadToUint8(value: unknown): Promise<Uint8Array | null> {
|
||||
if (value instanceof Uint8Array) {
|
||||
return value
|
||||
}
|
||||
if (value instanceof ArrayBuffer) {
|
||||
return new Uint8Array(value)
|
||||
}
|
||||
if (value && typeof value === 'object' && 'arrayBuffer' in value) {
|
||||
const blob = value as { arrayBuffer: () => Promise<ArrayBuffer> }
|
||||
return new Uint8Array(await blob.arrayBuffer())
|
||||
}
|
||||
if (typeof FileReader !== 'undefined' && value instanceof Blob) {
|
||||
return new Promise((resolve) => {
|
||||
const reader = new FileReader()
|
||||
reader.onload = () => {
|
||||
resolve(reader.result instanceof ArrayBuffer ? new Uint8Array(reader.result) : null)
|
||||
}
|
||||
reader.onerror = () => resolve(null)
|
||||
reader.readAsArrayBuffer(value)
|
||||
})
|
||||
}
|
||||
return null
|
||||
}
|
||||
|
||||
@@ -0,0 +1,79 @@
|
||||
const TERMINAL_STREAM_KIND = 0x74
|
||||
const TERMINAL_STREAM_VERSION = 1
|
||||
const HEADER_BYTES = 16
|
||||
|
||||
export enum TerminalStreamOpcode {
|
||||
Output = 1,
|
||||
SnapshotStart = 2,
|
||||
SnapshotChunk = 3,
|
||||
SnapshotEnd = 4,
|
||||
Resized = 5,
|
||||
Error = 6
|
||||
}
|
||||
|
||||
export type TerminalStreamFrame = {
|
||||
opcode: TerminalStreamOpcode
|
||||
streamId: number
|
||||
seq: number
|
||||
payload: Uint8Array
|
||||
}
|
||||
|
||||
export function encodeTerminalStreamFrame(frame: TerminalStreamFrame): Uint8Array {
|
||||
const out = new Uint8Array(HEADER_BYTES + frame.payload.length)
|
||||
const view = new DataView(out.buffer, out.byteOffset, out.byteLength)
|
||||
view.setUint8(0, TERMINAL_STREAM_KIND)
|
||||
view.setUint8(1, TERMINAL_STREAM_VERSION)
|
||||
view.setUint8(2, frame.opcode)
|
||||
view.setUint8(3, 0)
|
||||
view.setUint32(4, frame.streamId, true)
|
||||
const seq = Math.max(0, Math.floor(frame.seq))
|
||||
view.setUint32(8, Math.floor(seq / 0x100000000), true)
|
||||
view.setUint32(12, seq >>> 0, true)
|
||||
out.set(frame.payload, HEADER_BYTES)
|
||||
return out
|
||||
}
|
||||
|
||||
export function decodeTerminalStreamFrame(bytes: Uint8Array): TerminalStreamFrame | null {
|
||||
if (bytes.length < HEADER_BYTES) {
|
||||
return null
|
||||
}
|
||||
const view = new DataView(bytes.buffer, bytes.byteOffset, bytes.byteLength)
|
||||
if (view.getUint8(0) !== TERMINAL_STREAM_KIND || view.getUint8(1) !== TERMINAL_STREAM_VERSION) {
|
||||
return null
|
||||
}
|
||||
const opcode = view.getUint8(2)
|
||||
if (!isTerminalStreamOpcode(opcode)) {
|
||||
return null
|
||||
}
|
||||
const high = view.getUint32(8, true)
|
||||
const low = view.getUint32(12, true)
|
||||
return {
|
||||
opcode,
|
||||
streamId: view.getUint32(4, true),
|
||||
seq: high * 0x100000000 + low,
|
||||
payload: bytes.slice(HEADER_BYTES)
|
||||
}
|
||||
}
|
||||
|
||||
export function decodeTerminalStreamJson<T>(payload: Uint8Array): T | null {
|
||||
try {
|
||||
return JSON.parse(new TextDecoder().decode(payload)) as T
|
||||
} catch {
|
||||
return null
|
||||
}
|
||||
}
|
||||
|
||||
export function decodeTerminalStreamText(payload: Uint8Array): string {
|
||||
return new TextDecoder().decode(payload)
|
||||
}
|
||||
|
||||
function isTerminalStreamOpcode(value: number): value is TerminalStreamOpcode {
|
||||
return (
|
||||
value === TerminalStreamOpcode.Output ||
|
||||
value === TerminalStreamOpcode.SnapshotStart ||
|
||||
value === TerminalStreamOpcode.SnapshotChunk ||
|
||||
value === TerminalStreamOpcode.SnapshotEnd ||
|
||||
value === TerminalStreamOpcode.Resized ||
|
||||
value === TerminalStreamOpcode.Error
|
||||
)
|
||||
}
|
||||
@@ -52,6 +52,10 @@ export type RpcContext = {
|
||||
// server reap all subscriptions for a closing socket, even when other
|
||||
// sockets for the same deviceToken stay alive (multi-screen mobile).
|
||||
connectionId?: string
|
||||
// Why: mobile terminal traffic is byte-oriented and bypasses JSON streaming
|
||||
// responses after the binary terminal cutover. Undefined on Unix/socket
|
||||
// transports and non-E2EE WebSocket paths.
|
||||
sendBinary?: (bytes: Uint8Array<ArrayBufferLike>) => void
|
||||
}
|
||||
|
||||
export type RpcHandler<TParams> = (params: TParams, ctx: RpcContext) => Promise<unknown> | unknown
|
||||
|
||||
@@ -77,7 +77,7 @@ export class RpcDispatcher {
|
||||
async dispatchStreaming(
|
||||
request: RpcRequest,
|
||||
reply: (response: string) => void,
|
||||
options?: { connectionId?: string }
|
||||
options?: { connectionId?: string; sendBinary?: (bytes: Uint8Array<ArrayBufferLike>) => void }
|
||||
): Promise<void> {
|
||||
const meta = this.meta()
|
||||
const method = this.registry.get(request.method)
|
||||
@@ -100,7 +100,8 @@ export class RpcDispatcher {
|
||||
try {
|
||||
const result = await method.handler(parsedParams.value, {
|
||||
runtime: this.runtime,
|
||||
connectionId: options?.connectionId
|
||||
connectionId: options?.connectionId,
|
||||
sendBinary: options?.sendBinary
|
||||
})
|
||||
reply(JSON.stringify(successResponse(request.id, meta, result)))
|
||||
} catch (error) {
|
||||
@@ -118,7 +119,11 @@ export class RpcDispatcher {
|
||||
try {
|
||||
await method.handler(
|
||||
parsedParams.value,
|
||||
{ runtime: this.runtime, connectionId: options?.connectionId },
|
||||
{
|
||||
runtime: this.runtime,
|
||||
connectionId: options?.connectionId,
|
||||
sendBinary: options?.sendBinary
|
||||
},
|
||||
emit
|
||||
)
|
||||
} catch (error) {
|
||||
|
||||
@@ -2,7 +2,7 @@
|
||||
// It owns the handshake state machine and transparent encrypt/decrypt so the RPC
|
||||
// handler only sees plaintext JSON, identical to the Unix socket path.
|
||||
import type { WebSocket } from 'ws'
|
||||
import { deriveSharedKey, encrypt, decrypt } from './e2ee-crypto'
|
||||
import { deriveSharedKey, encrypt, decrypt, encryptBytes, decryptBytes } from './e2ee-crypto'
|
||||
|
||||
type ChannelState = 'awaiting_hello' | 'awaiting_auth' | 'ready'
|
||||
|
||||
@@ -40,7 +40,11 @@ export class E2EEChannel {
|
||||
// can forward decrypted messages. Kept as a callback rather than constructor
|
||||
// param because the handler needs the encrypt function for replies.
|
||||
private messageHandler:
|
||||
| ((plaintext: string, encryptedReply: (response: string) => void) => void)
|
||||
| ((
|
||||
plaintext: string,
|
||||
encryptedReply: (response: string) => void,
|
||||
encryptedBinaryReply: (response: Uint8Array<ArrayBufferLike>) => void
|
||||
) => void)
|
||||
| null = null
|
||||
|
||||
deviceToken: string | null = null
|
||||
@@ -58,13 +62,21 @@ export class E2EEChannel {
|
||||
}
|
||||
|
||||
onMessage(
|
||||
handler: (plaintext: string, encryptedReply: (response: string) => void) => void
|
||||
handler: (
|
||||
plaintext: string,
|
||||
encryptedReply: (response: string) => void,
|
||||
encryptedBinaryReply: (response: Uint8Array<ArrayBufferLike>) => void
|
||||
) => void
|
||||
): void {
|
||||
this.messageHandler = handler
|
||||
}
|
||||
|
||||
handleRawMessage(raw: string): void {
|
||||
handleRawMessage(raw: string | Uint8Array<ArrayBufferLike>): void {
|
||||
if (this.state === 'awaiting_hello') {
|
||||
if (typeof raw !== 'string') {
|
||||
this.onError(4001, 'Invalid handshake message')
|
||||
return
|
||||
}
|
||||
this.handleHello(raw)
|
||||
return
|
||||
}
|
||||
@@ -73,12 +85,17 @@ export class E2EEChannel {
|
||||
return
|
||||
}
|
||||
|
||||
if (typeof raw !== 'string') {
|
||||
const plaintextBytes = decryptBytes(raw, this.sharedKey)
|
||||
if (plaintextBytes === null) {
|
||||
this.trackDecryptFailure()
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
const plaintext = decrypt(raw, this.sharedKey)
|
||||
if (plaintext === null) {
|
||||
this.consecutiveFailures++
|
||||
if (this.consecutiveFailures >= MAX_CONSECUTIVE_DECRYPT_FAILURES) {
|
||||
this.onError(4003, 'Too many decryption failures')
|
||||
}
|
||||
this.trackDecryptFailure()
|
||||
return
|
||||
}
|
||||
|
||||
@@ -100,7 +117,20 @@ export class E2EEChannel {
|
||||
}
|
||||
this.ws.send(encrypt(response, this.sharedKey))
|
||||
}
|
||||
this.messageHandler?.(plaintext, encryptedReply)
|
||||
const encryptedBinaryReply = (response: Uint8Array<ArrayBufferLike>) => {
|
||||
if (!this.sharedKey || this.ws.readyState !== this.ws.OPEN) {
|
||||
return
|
||||
}
|
||||
this.ws.send(Buffer.from(encryptBytes(response, this.sharedKey)), { binary: true })
|
||||
}
|
||||
this.messageHandler?.(plaintext, encryptedReply, encryptedBinaryReply)
|
||||
}
|
||||
|
||||
private trackDecryptFailure(): void {
|
||||
this.consecutiveFailures++
|
||||
if (this.consecutiveFailures >= MAX_CONSECUTIVE_DECRYPT_FAILURES) {
|
||||
this.onError(4003, 'Too many decryption failures')
|
||||
}
|
||||
}
|
||||
|
||||
private handleHello(raw: string): void {
|
||||
|
||||
@@ -1,5 +1,12 @@
|
||||
import { describe, expect, it } from 'vitest'
|
||||
import { generateKeyPair, deriveSharedKey, encrypt, decrypt } from './e2ee-crypto'
|
||||
import {
|
||||
generateKeyPair,
|
||||
deriveSharedKey,
|
||||
encrypt,
|
||||
decrypt,
|
||||
encryptBytes,
|
||||
decryptBytes
|
||||
} from './e2ee-crypto'
|
||||
|
||||
describe('e2ee-crypto', () => {
|
||||
it('encrypt/decrypt round-trips with shared key', () => {
|
||||
@@ -76,4 +83,17 @@ describe('e2ee-crypto', () => {
|
||||
|
||||
expect(decrypt(tampered, shared)).toBeNull()
|
||||
})
|
||||
|
||||
it('encrypt/decrypt round-trips raw bytes for binary terminal frames', () => {
|
||||
const server = generateKeyPair()
|
||||
const client = generateKeyPair()
|
||||
const serverShared = deriveSharedKey(server.secretKey, client.publicKey)
|
||||
const clientShared = deriveSharedKey(client.secretKey, server.publicKey)
|
||||
const payload = new Uint8Array([0, 1, 2, 127, 128, 255])
|
||||
|
||||
const encrypted = encryptBytes(payload, clientShared)
|
||||
const decrypted = decryptBytes(encrypted, serverShared)
|
||||
|
||||
expect(decrypted).toEqual(payload)
|
||||
})
|
||||
})
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
// Why: shared E2EE primitives for the desktop side. Wraps tweetnacl to provide
|
||||
// encrypt/decrypt with the NaCl box format: [24-byte nonce][ciphertext], encoded
|
||||
// as base64 for transmission over WebSocket text frames.
|
||||
// encrypt/decrypt with the NaCl box format: [24-byte nonce][ciphertext]. JSON
|
||||
// RPC uses base64 text frames; terminal streams use the raw byte bundle.
|
||||
import nacl from 'tweetnacl'
|
||||
|
||||
export function generateKeyPair(): nacl.BoxKeyPair {
|
||||
@@ -12,19 +12,31 @@ export function deriveSharedKey(ourSecretKey: Uint8Array, peerPublicKey: Uint8Ar
|
||||
}
|
||||
|
||||
export function encrypt(plaintext: string, sharedKey: Uint8Array): string {
|
||||
const nonce = nacl.randomBytes(nacl.box.nonceLength)
|
||||
const messageBytes = new TextEncoder().encode(plaintext)
|
||||
const ciphertext = nacl.box.after(messageBytes, nonce, sharedKey)
|
||||
return Buffer.from(encryptBytes(messageBytes, sharedKey)).toString('base64')
|
||||
}
|
||||
|
||||
export function decrypt(encrypted: string, sharedKey: Uint8Array): string | null {
|
||||
const bundle = Uint8Array.from(Buffer.from(encrypted, 'base64'))
|
||||
const plaintext = decryptBytes(bundle, sharedKey)
|
||||
return plaintext ? new TextDecoder().decode(plaintext) : null
|
||||
}
|
||||
|
||||
export function encryptBytes(
|
||||
plaintext: Uint8Array<ArrayBufferLike>,
|
||||
sharedKey: Uint8Array
|
||||
): Uint8Array {
|
||||
const nonce = nacl.randomBytes(nacl.box.nonceLength)
|
||||
const ciphertext = nacl.box.after(plaintext, nonce, sharedKey)
|
||||
|
||||
const bundle = new Uint8Array(nonce.length + ciphertext.length)
|
||||
bundle.set(nonce)
|
||||
bundle.set(ciphertext, nonce.length)
|
||||
|
||||
return Buffer.from(bundle).toString('base64')
|
||||
return bundle
|
||||
}
|
||||
|
||||
export function decrypt(encrypted: string, sharedKey: Uint8Array): string | null {
|
||||
const bundle = Uint8Array.from(Buffer.from(encrypted, 'base64'))
|
||||
export function decryptBytes(bundle: Uint8Array, sharedKey: Uint8Array): Uint8Array | null {
|
||||
if (bundle.length < nacl.box.nonceLength + nacl.box.overheadLength) {
|
||||
return null
|
||||
}
|
||||
@@ -37,5 +49,5 @@ export function decrypt(encrypted: string, sharedKey: Uint8Array): string | null
|
||||
return null
|
||||
}
|
||||
|
||||
return new TextDecoder().decode(plaintext)
|
||||
return plaintext
|
||||
}
|
||||
|
||||
@@ -2,6 +2,13 @@
|
||||
import { z } from 'zod'
|
||||
import { defineMethod, defineStreamingMethod, type RpcAnyMethod } from '../core'
|
||||
import { OptionalFiniteNumber, OptionalString, requiredString } from '../schemas'
|
||||
import type { OrcaRuntimeService } from '../../orca-runtime'
|
||||
import {
|
||||
TerminalStreamOpcode,
|
||||
encodeTerminalStreamFrame,
|
||||
encodeTerminalStreamJson,
|
||||
encodeTerminalStreamText
|
||||
} from '../../../../shared/terminal-stream-protocol'
|
||||
|
||||
// Why: when a mobile client subscribes the server resizes the PTY to phone
|
||||
// dims and serializes the buffer. Sending only the visible screen meant
|
||||
@@ -10,6 +17,87 @@ import { OptionalFiniteNumber, OptionalString, requiredString } from '../schemas
|
||||
// agent runs (Claude Code chats, command output) reachable. The mobile
|
||||
// WebView's xterm has a 5000-row buffer so this fits comfortably.
|
||||
const MOBILE_SUBSCRIBE_SCROLLBACK_ROWS = 1000
|
||||
const MOBILE_SNAPSHOT_BYTE_BUDGET = 512 * 1024
|
||||
const TERMINAL_STREAM_CHUNK_BYTES = 48 * 1024
|
||||
let nextTerminalStreamId = 1
|
||||
|
||||
type SnapshotFrameOptions = {
|
||||
kind: 'scrollback' | 'resized'
|
||||
cols: number
|
||||
rows: number
|
||||
data: string
|
||||
displayMode?: string
|
||||
reason?: string
|
||||
seq?: number
|
||||
truncated?: boolean
|
||||
truncatedByByteBudget?: boolean
|
||||
}
|
||||
|
||||
type SerializedSnapshot = {
|
||||
data: string
|
||||
cols: number
|
||||
rows: number
|
||||
scrollbackRows: number
|
||||
truncatedByByteBudget: boolean
|
||||
} | null
|
||||
|
||||
function sendSnapshotFrames(
|
||||
sendFrame: (opcode: TerminalStreamOpcode, payload?: Uint8Array<ArrayBufferLike>) => void,
|
||||
options: SnapshotFrameOptions
|
||||
): { bytes: number; chunks: number } {
|
||||
sendFrame(
|
||||
TerminalStreamOpcode.SnapshotStart,
|
||||
encodeTerminalStreamJson({
|
||||
kind: options.kind,
|
||||
cols: options.cols,
|
||||
rows: options.rows,
|
||||
displayMode: options.displayMode,
|
||||
reason: options.reason,
|
||||
seq: options.seq,
|
||||
truncated: options.truncated === true,
|
||||
truncatedByByteBudget: options.truncatedByByteBudget === true
|
||||
})
|
||||
)
|
||||
const bytes = encodeTerminalStreamText(options.data)
|
||||
let chunks = 0
|
||||
for (let offset = 0; offset < bytes.length; offset += TERMINAL_STREAM_CHUNK_BYTES) {
|
||||
chunks++
|
||||
sendFrame(
|
||||
TerminalStreamOpcode.SnapshotChunk,
|
||||
bytes.slice(offset, offset + TERMINAL_STREAM_CHUNK_BYTES)
|
||||
)
|
||||
}
|
||||
sendFrame(TerminalStreamOpcode.SnapshotEnd)
|
||||
return { bytes: bytes.byteLength, chunks }
|
||||
}
|
||||
|
||||
async function serializeBudgetedMobileSnapshot(
|
||||
runtime: OrcaRuntimeService,
|
||||
ptyId: string,
|
||||
isMobile: boolean
|
||||
): Promise<SerializedSnapshot> {
|
||||
if (!isMobile) {
|
||||
const serialized = await runtime.serializeTerminalBuffer(ptyId, { scrollbackRows: 0 })
|
||||
return serialized ? { ...serialized, scrollbackRows: 0, truncatedByByteBudget: false } : null
|
||||
}
|
||||
const candidates = [MOBILE_SUBSCRIBE_SCROLLBACK_ROWS, 500, 250, 100, 25, 0]
|
||||
for (const rows of candidates) {
|
||||
const serialized = await runtime.serializeTerminalBuffer(ptyId, { scrollbackRows: rows })
|
||||
if (!serialized) {
|
||||
return null
|
||||
}
|
||||
const bytes = new TextEncoder().encode(serialized.data).byteLength
|
||||
if (bytes <= MOBILE_SNAPSHOT_BYTE_BUDGET || rows === 0) {
|
||||
return {
|
||||
...serialized,
|
||||
scrollbackRows: rows,
|
||||
truncatedByByteBudget:
|
||||
rows < MOBILE_SUBSCRIBE_SCROLLBACK_ROWS || bytes > MOBILE_SNAPSHOT_BYTE_BUDGET
|
||||
}
|
||||
}
|
||||
}
|
||||
return null
|
||||
}
|
||||
|
||||
const TerminalHandle = z.object({
|
||||
terminal: requiredString('Missing terminal handle')
|
||||
@@ -127,6 +215,11 @@ const TerminalSubscribe = TerminalHandle.extend({
|
||||
cols: z.number().int().min(20).max(240),
|
||||
rows: z.number().int().min(8).max(120)
|
||||
})
|
||||
.optional(),
|
||||
capabilities: z
|
||||
.object({
|
||||
terminalBinaryStream: z.literal(1).optional()
|
||||
})
|
||||
.optional()
|
||||
})
|
||||
|
||||
@@ -384,9 +477,10 @@ export const TERMINAL_METHODS: RpcAnyMethod[] = [
|
||||
defineStreamingMethod({
|
||||
name: 'terminal.subscribe',
|
||||
params: TerminalSubscribe,
|
||||
handler: async (params, { runtime }, emit) => {
|
||||
handler: async (params, { runtime, connectionId, sendBinary }, emit) => {
|
||||
let leaf = runtime.resolveLeafForHandle(params.terminal)
|
||||
const isMobile = params.client?.type === 'mobile'
|
||||
const useBinaryStream = isMobile && params.capabilities?.terminalBinaryStream === 1
|
||||
|
||||
// Why: the left pane's PTY spawns asynchronously after the tab is created.
|
||||
// Mobile clients that subscribe before the PTY is ready would get a bare
|
||||
@@ -403,30 +497,99 @@ export const TERMINAL_METHODS: RpcAnyMethod[] = [
|
||||
|
||||
if (!leaf?.ptyId) {
|
||||
const read = await runtime.readTerminal(params.terminal)
|
||||
emit({
|
||||
type: 'scrollback',
|
||||
lines: read.tail,
|
||||
truncated: read.truncated,
|
||||
serialized: undefined,
|
||||
cols: undefined,
|
||||
rows: undefined
|
||||
})
|
||||
emit({ type: 'subscribed', streamId: null, lines: read.tail, truncated: read.truncated })
|
||||
emit({ type: 'end' })
|
||||
return
|
||||
}
|
||||
|
||||
if (isMobile && (!useBinaryStream || !sendBinary)) {
|
||||
throw new Error('binary_terminal_stream_required')
|
||||
}
|
||||
|
||||
const ptyId = leaf.ptyId
|
||||
const clientId = params.client?.id
|
||||
if (!isMobile) {
|
||||
const read = await runtime.readTerminal(params.terminal)
|
||||
const serialized = await serializeBudgetedMobileSnapshot(runtime, ptyId, false)
|
||||
const size = runtime.getTerminalSize(ptyId)
|
||||
const displayMode = runtime.getMobileDisplayMode(ptyId)
|
||||
const seq = runtime.getLayout(ptyId)?.seq
|
||||
emit({
|
||||
type: 'scrollback',
|
||||
lines: read.tail,
|
||||
truncated: read.truncated,
|
||||
serialized: serialized?.data,
|
||||
cols: serialized?.cols ?? size?.cols,
|
||||
rows: serialized?.rows ?? size?.rows,
|
||||
displayMode,
|
||||
seq
|
||||
})
|
||||
|
||||
await new Promise<void>((resolve) => {
|
||||
const unsubscribeData = runtime.subscribeToTerminalData(ptyId, (data) => {
|
||||
emit({ type: 'data', chunk: data })
|
||||
})
|
||||
const unsubscribeFit = runtime.subscribeToFitOverrideChanges(ptyId, (event) => {
|
||||
emit({
|
||||
type: 'fit-override-changed',
|
||||
mode: event.mode,
|
||||
cols: event.cols,
|
||||
rows: event.rows
|
||||
})
|
||||
})
|
||||
runtime.registerSubscriptionCleanup(
|
||||
params.terminal,
|
||||
() => {
|
||||
unsubscribeData()
|
||||
unsubscribeFit()
|
||||
emit({ type: 'end' })
|
||||
resolve()
|
||||
},
|
||||
connectionId
|
||||
)
|
||||
})
|
||||
return
|
||||
}
|
||||
|
||||
const streamId = nextTerminalStreamId++
|
||||
let cursor = 0
|
||||
let closed = false
|
||||
let buffering = true
|
||||
const pendingOutput: string[] = []
|
||||
const sendFrame = (
|
||||
opcode: TerminalStreamOpcode,
|
||||
payload: Uint8Array<ArrayBufferLike> = new Uint8Array()
|
||||
): void => {
|
||||
if (closed || !sendBinary) {
|
||||
return
|
||||
}
|
||||
sendBinary(encodeTerminalStreamFrame({ opcode, streamId, seq: cursor++, payload }))
|
||||
}
|
||||
// Server-side auto-fit: resize PTY to phone dims before serializing scrollback
|
||||
if (isMobile && clientId) {
|
||||
await runtime.handleMobileSubscribe(ptyId, clientId, params.viewport)
|
||||
}
|
||||
|
||||
const read = await runtime.readTerminal(params.terminal)
|
||||
const serialized = await runtime.serializeTerminalBuffer(ptyId, {
|
||||
scrollbackRows: isMobile ? MOBILE_SUBSCRIBE_SCROLLBACK_ROWS : 0
|
||||
const unsubscribeData = runtime.subscribeToTerminalData(ptyId, (data) => {
|
||||
if (closed) {
|
||||
return
|
||||
}
|
||||
if (buffering) {
|
||||
pendingOutput.push(data)
|
||||
return
|
||||
}
|
||||
sendBinary!(
|
||||
encodeTerminalStreamFrame({
|
||||
opcode: TerminalStreamOpcode.Output,
|
||||
streamId,
|
||||
seq: cursor++,
|
||||
payload: encodeTerminalStreamText(data)
|
||||
})
|
||||
)
|
||||
})
|
||||
|
||||
const read = await runtime.readTerminal(params.terminal)
|
||||
const serialized = await serializeBudgetedMobileSnapshot(runtime, ptyId, true)
|
||||
const size = runtime.getTerminalSize(ptyId)
|
||||
const displayMode = runtime.getMobileDisplayMode(ptyId)
|
||||
// Why: emit the current layout seq with the initial scrollback so
|
||||
@@ -435,40 +598,61 @@ export const TERMINAL_METHODS: RpcAnyMethod[] = [
|
||||
// See docs/mobile-terminal-layout-state-machine.md.
|
||||
const seq = runtime.getLayout(ptyId)?.seq
|
||||
emit({
|
||||
type: 'scrollback',
|
||||
type: 'subscribed',
|
||||
streamId,
|
||||
lines: read.tail,
|
||||
truncated: read.truncated,
|
||||
serialized: serialized?.data,
|
||||
cols: serialized?.cols ?? size?.cols,
|
||||
rows: serialized?.rows ?? size?.rows,
|
||||
displayMode,
|
||||
seq
|
||||
})
|
||||
const snapshotStats = sendSnapshotFrames(sendFrame, {
|
||||
kind: 'scrollback',
|
||||
cols: serialized?.cols ?? size?.cols ?? 80,
|
||||
rows: serialized?.rows ?? size?.rows ?? 24,
|
||||
displayMode,
|
||||
seq,
|
||||
truncated: read.truncated,
|
||||
truncatedByByteBudget: serialized?.truncatedByByteBudget,
|
||||
data: serialized?.data ?? ''
|
||||
})
|
||||
console.log('[mobile-terminal-stream] snapshot', {
|
||||
terminal: params.terminal,
|
||||
streamId,
|
||||
kind: 'scrollback',
|
||||
bytes: snapshotStats.bytes,
|
||||
chunks: snapshotStats.chunks,
|
||||
scrollbackRows: serialized?.scrollbackRows,
|
||||
truncatedByByteBudget: serialized?.truncatedByByteBudget === true
|
||||
})
|
||||
buffering = false
|
||||
for (const item of pendingOutput.splice(0)) {
|
||||
sendBinary!(
|
||||
encodeTerminalStreamFrame({
|
||||
opcode: TerminalStreamOpcode.Output,
|
||||
streamId,
|
||||
seq: cursor++,
|
||||
payload: encodeTerminalStreamText(item)
|
||||
})
|
||||
)
|
||||
}
|
||||
|
||||
await new Promise<void>((resolve) => {
|
||||
const unsubscribeData = runtime.subscribeToTerminalData(ptyId, (data) => {
|
||||
emit({ type: 'data', chunk: data })
|
||||
})
|
||||
|
||||
// Inline resize events replace the old fit-override-changed event for
|
||||
// mobile clients. They include fresh serialized scrollback so the client
|
||||
// can reinitialize xterm without resubscribing.
|
||||
const unsubscribeResize = runtime.subscribeToTerminalResize(ptyId, async (event) => {
|
||||
// Why: mobile subscriptions need the same scrollback on inline
|
||||
// resize as on initial subscribe — without it, toggling phone/desktop
|
||||
// mode or the keyboard-driven refit would silently wipe history.
|
||||
const fresh = await runtime.serializeTerminalBuffer(ptyId, {
|
||||
scrollbackRows: isMobile ? MOBILE_SUBSCRIBE_SCROLLBACK_ROWS : 0
|
||||
})
|
||||
emit({
|
||||
type: 'resized',
|
||||
cols: event.cols,
|
||||
rows: event.rows,
|
||||
serialized: fresh?.data,
|
||||
displayMode: event.displayMode,
|
||||
reason: event.reason,
|
||||
seq: event.seq
|
||||
})
|
||||
const unsubscribeResize = runtime.subscribeToTerminalResize(ptyId, (event) => {
|
||||
// Why: true PTY geometry changes should be followed by the TUI's
|
||||
// redraw output, not a full scrollback replay. The client resizes
|
||||
// xterm geometry and consumes subsequent live output on this stream.
|
||||
sendFrame(
|
||||
TerminalStreamOpcode.Resized,
|
||||
encodeTerminalStreamJson({
|
||||
cols: event.cols,
|
||||
rows: event.rows,
|
||||
displayMode: event.displayMode,
|
||||
reason: event.reason,
|
||||
seq: event.seq
|
||||
})
|
||||
)
|
||||
})
|
||||
|
||||
// Legacy fit-override-changed for non-mobile (desktop) subscribers
|
||||
@@ -488,16 +672,21 @@ export const TERMINAL_METHODS: RpcAnyMethod[] = [
|
||||
// evict each other via registerSubscriptionCleanup's
|
||||
// duplicate-key cleanup. See docs/mobile-presence-lock.md.
|
||||
const subscriptionId = clientId ? `${params.terminal}:${clientId}` : params.terminal
|
||||
runtime.registerSubscriptionCleanup(subscriptionId, () => {
|
||||
unsubscribeData()
|
||||
unsubscribeResize()
|
||||
unsubscribeFit()
|
||||
if (isMobile && clientId) {
|
||||
runtime.handleMobileUnsubscribe(ptyId, clientId)
|
||||
}
|
||||
emit({ type: 'end' })
|
||||
resolve()
|
||||
})
|
||||
runtime.registerSubscriptionCleanup(
|
||||
subscriptionId,
|
||||
() => {
|
||||
closed = true
|
||||
unsubscribeData()
|
||||
unsubscribeResize()
|
||||
unsubscribeFit()
|
||||
if (isMobile && clientId) {
|
||||
runtime.handleMobileUnsubscribe(ptyId, clientId)
|
||||
}
|
||||
emit({ type: 'end' })
|
||||
resolve()
|
||||
},
|
||||
connectionId
|
||||
)
|
||||
})
|
||||
}
|
||||
}),
|
||||
|
||||
@@ -10,6 +10,14 @@ import type { RpcTransport } from './transport'
|
||||
|
||||
const MAX_WS_MESSAGE_BYTES = 1024 * 1024
|
||||
const MAX_WS_CONNECTIONS = 32
|
||||
type WebSocketMessagePayload = string | Uint8Array<ArrayBufferLike>
|
||||
type WebSocketMessageHandler = {
|
||||
bivarianceHack(
|
||||
msg: WebSocketMessagePayload,
|
||||
reply: (response: string) => void,
|
||||
ws: WebSocket
|
||||
): void
|
||||
}['bivarianceHack']
|
||||
|
||||
// Why: mobile clients (iOS/Android) regularly background-suspend their
|
||||
// sockets without the OS sending a TCP FIN/RST, leaving the server with
|
||||
@@ -45,9 +53,7 @@ export class WebSocketTransport implements RpcTransport {
|
||||
// sweep. A socket missing from the set when the next sweep fires is
|
||||
// assumed dead and terminated.
|
||||
private wsAlive = new WeakSet<WebSocket>()
|
||||
private messageHandler:
|
||||
| ((msg: string, reply: (response: string) => void, ws: WebSocket) => void)
|
||||
| null = null
|
||||
private messageHandler: WebSocketMessageHandler | null = null
|
||||
private connectionCloseHandler:
|
||||
| ((clientId: string, ws: WebSocket, hasOtherConnections: boolean) => void)
|
||||
| null = null
|
||||
@@ -63,9 +69,7 @@ export class WebSocketTransport implements RpcTransport {
|
||||
this.heartbeatIntervalMs = heartbeatIntervalMs ?? HEARTBEAT_INTERVAL_MS
|
||||
}
|
||||
|
||||
onMessage(
|
||||
handler: (msg: string, reply: (response: string) => void, ws: WebSocket) => void
|
||||
): void {
|
||||
onMessage(handler: WebSocketMessageHandler): void {
|
||||
this.messageHandler = handler
|
||||
}
|
||||
|
||||
@@ -240,13 +244,18 @@ export class WebSocketTransport implements RpcTransport {
|
||||
this.wsAlive.add(ws)
|
||||
})
|
||||
|
||||
ws.on('message', (data) => {
|
||||
ws.on('message', (data, isBinary) => {
|
||||
// Why: any inbound traffic counts as proof of life, not just pongs.
|
||||
// RN's WebSocket runtime auto-pongs server pings transparently, but
|
||||
// app-level frames also count toward liveness so an actively-talking
|
||||
// client doesn't get terminated mid-request.
|
||||
this.wsAlive.add(ws)
|
||||
const msg = typeof data === 'string' ? data : data.toString('utf-8')
|
||||
const msg =
|
||||
typeof data === 'string'
|
||||
? data
|
||||
: isBinary
|
||||
? new Uint8Array(data as Buffer)
|
||||
: data.toString()
|
||||
this.messageHandler?.(
|
||||
msg,
|
||||
(response) => {
|
||||
|
||||
@@ -250,8 +250,14 @@ export class OrcaRuntimeRpcServer {
|
||||
ws.close(code, reason)
|
||||
}
|
||||
})
|
||||
channel.onMessage((plaintext, encryptedReply) => {
|
||||
void this.handleWebSocketMessage(plaintext, encryptedReply, wsTransport, ws)
|
||||
channel.onMessage((plaintext, encryptedReply, encryptedBinaryReply) => {
|
||||
void this.handleWebSocketMessage(
|
||||
plaintext,
|
||||
encryptedReply,
|
||||
encryptedBinaryReply,
|
||||
wsTransport,
|
||||
ws
|
||||
)
|
||||
})
|
||||
this.e2eeChannels.set(ws, channel)
|
||||
}
|
||||
@@ -411,6 +417,7 @@ export class OrcaRuntimeRpcServer {
|
||||
private async handleWebSocketMessage(
|
||||
rawMessage: string,
|
||||
reply: (response: string) => void,
|
||||
sendBinary: (response: Uint8Array<ArrayBufferLike>) => void,
|
||||
wsTransport?: WebSocketTransport,
|
||||
ws?: WebSocket
|
||||
): Promise<void> {
|
||||
@@ -451,7 +458,7 @@ export class OrcaRuntimeRpcServer {
|
||||
}
|
||||
|
||||
const connectionId = ws ? this.wsConnectionIds.get(ws) : undefined
|
||||
await this.dispatcher.dispatchStreaming(request, reply, { connectionId })
|
||||
await this.dispatcher.dispatchStreaming(request, reply, { connectionId, sendBinary })
|
||||
}
|
||||
|
||||
private buildError(id: string, code: string, message: string): RpcResponse {
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
import { describe, expect, it } from 'vitest'
|
||||
import { evaluateCompat } from './protocol-compat'
|
||||
import { DESKTOP_PROTOCOL_VERSION, MIN_COMPATIBLE_MOBILE_VERSION } from './protocol-version'
|
||||
|
||||
const MOBILE_V = 1
|
||||
|
||||
@@ -104,4 +105,20 @@ describe('evaluateCompat', () => {
|
||||
).toEqual({ kind: 'ok' })
|
||||
}
|
||||
})
|
||||
|
||||
it('hard-blocks protocol-1 mobile for the binary terminal stream cutover', () => {
|
||||
const verdict = evaluateCompat({
|
||||
mobileProtocolVersion: 1,
|
||||
minCompatibleDesktopVersion: DESKTOP_PROTOCOL_VERSION,
|
||||
desktopProtocolVersion: DESKTOP_PROTOCOL_VERSION,
|
||||
desktopMinCompatibleMobileVersion: MIN_COMPATIBLE_MOBILE_VERSION
|
||||
})
|
||||
|
||||
expect(verdict).toEqual({
|
||||
kind: 'blocked',
|
||||
reason: 'mobile-too-old',
|
||||
desktopVersion: DESKTOP_PROTOCOL_VERSION,
|
||||
requiredMobileVersion: MIN_COMPATIBLE_MOBILE_VERSION
|
||||
})
|
||||
})
|
||||
})
|
||||
|
||||
@@ -19,5 +19,5 @@
|
||||
// "kill switch": desktop can refuse old mobile builds without needing
|
||||
// a desktop release of mobile.
|
||||
|
||||
export const DESKTOP_PROTOCOL_VERSION = 1
|
||||
export const MIN_COMPATIBLE_MOBILE_VERSION = 0
|
||||
export const DESKTOP_PROTOCOL_VERSION = 2
|
||||
export const MIN_COMPATIBLE_MOBILE_VERSION = 2
|
||||
|
||||
@@ -0,0 +1,63 @@
|
||||
import { describe, expect, it } from 'vitest'
|
||||
import {
|
||||
TerminalStreamOpcode,
|
||||
decodeTerminalStreamFrame,
|
||||
decodeTerminalStreamJson,
|
||||
decodeTerminalStreamText,
|
||||
encodeTerminalStreamFrame,
|
||||
encodeTerminalStreamJson,
|
||||
encodeTerminalStreamText
|
||||
} from './terminal-stream-protocol'
|
||||
|
||||
describe('terminal-stream-protocol', () => {
|
||||
it('round-trips fixed-width binary frame headers and payloads', () => {
|
||||
const payload = encodeTerminalStreamText('hello terminal')
|
||||
const encoded = encodeTerminalStreamFrame({
|
||||
opcode: TerminalStreamOpcode.Output,
|
||||
streamId: 42,
|
||||
seq: 9,
|
||||
payload
|
||||
})
|
||||
|
||||
const decoded = decodeTerminalStreamFrame(encoded)
|
||||
|
||||
expect(decoded?.opcode).toBe(TerminalStreamOpcode.Output)
|
||||
expect(decoded?.streamId).toBe(42)
|
||||
expect(decoded?.seq).toBe(9)
|
||||
expect(decoded ? decodeTerminalStreamText(decoded.payload) : '').toBe('hello terminal')
|
||||
})
|
||||
|
||||
it('round-trips snapshot metadata JSON payloads', () => {
|
||||
const encoded = encodeTerminalStreamFrame({
|
||||
opcode: TerminalStreamOpcode.SnapshotStart,
|
||||
streamId: 7,
|
||||
seq: 1,
|
||||
payload: encodeTerminalStreamJson({ kind: 'scrollback', cols: 49, rows: 28 })
|
||||
})
|
||||
|
||||
const decoded = decodeTerminalStreamFrame(encoded)
|
||||
|
||||
expect(decoded && decodeTerminalStreamJson(decoded.payload)).toEqual({
|
||||
kind: 'scrollback',
|
||||
cols: 49,
|
||||
rows: 28
|
||||
})
|
||||
})
|
||||
|
||||
it('rejects unknown frame versions and opcodes', () => {
|
||||
const encoded = encodeTerminalStreamFrame({
|
||||
opcode: TerminalStreamOpcode.Output,
|
||||
streamId: 1,
|
||||
seq: 1,
|
||||
payload: new Uint8Array()
|
||||
})
|
||||
|
||||
const badVersion = encoded.slice()
|
||||
badVersion[1] = 99
|
||||
expect(decodeTerminalStreamFrame(badVersion)).toBeNull()
|
||||
|
||||
const badOpcode = encoded.slice()
|
||||
badOpcode[2] = 99
|
||||
expect(decodeTerminalStreamFrame(badOpcode)).toBeNull()
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,87 @@
|
||||
const TERMINAL_STREAM_KIND = 0x74
|
||||
const TERMINAL_STREAM_VERSION = 1
|
||||
const HEADER_BYTES = 16
|
||||
|
||||
export enum TerminalStreamOpcode {
|
||||
Output = 1,
|
||||
SnapshotStart = 2,
|
||||
SnapshotChunk = 3,
|
||||
SnapshotEnd = 4,
|
||||
Resized = 5,
|
||||
Error = 6
|
||||
}
|
||||
|
||||
export type TerminalStreamFrame = {
|
||||
opcode: TerminalStreamOpcode
|
||||
streamId: number
|
||||
seq: number
|
||||
payload: Uint8Array
|
||||
}
|
||||
|
||||
export function encodeTerminalStreamFrame(frame: TerminalStreamFrame): Uint8Array {
|
||||
const out = new Uint8Array(HEADER_BYTES + frame.payload.length)
|
||||
const view = new DataView(out.buffer, out.byteOffset, out.byteLength)
|
||||
view.setUint8(0, TERMINAL_STREAM_KIND)
|
||||
view.setUint8(1, TERMINAL_STREAM_VERSION)
|
||||
view.setUint8(2, frame.opcode)
|
||||
view.setUint8(3, 0)
|
||||
view.setUint32(4, frame.streamId, true)
|
||||
const seq = Math.max(0, Math.floor(frame.seq))
|
||||
view.setUint32(8, Math.floor(seq / 0x100000000), true)
|
||||
view.setUint32(12, seq >>> 0, true)
|
||||
out.set(frame.payload, HEADER_BYTES)
|
||||
return out
|
||||
}
|
||||
|
||||
export function decodeTerminalStreamFrame(bytes: Uint8Array): TerminalStreamFrame | null {
|
||||
if (bytes.length < HEADER_BYTES) {
|
||||
return null
|
||||
}
|
||||
const view = new DataView(bytes.buffer, bytes.byteOffset, bytes.byteLength)
|
||||
if (view.getUint8(0) !== TERMINAL_STREAM_KIND || view.getUint8(1) !== TERMINAL_STREAM_VERSION) {
|
||||
return null
|
||||
}
|
||||
const opcode = view.getUint8(2)
|
||||
if (!isTerminalStreamOpcode(opcode)) {
|
||||
return null
|
||||
}
|
||||
const high = view.getUint32(8, true)
|
||||
const low = view.getUint32(12, true)
|
||||
return {
|
||||
opcode,
|
||||
streamId: view.getUint32(4, true),
|
||||
seq: high * 0x100000000 + low,
|
||||
payload: bytes.slice(HEADER_BYTES)
|
||||
}
|
||||
}
|
||||
|
||||
export function encodeTerminalStreamJson(value: unknown): Uint8Array {
|
||||
return new TextEncoder().encode(JSON.stringify(value))
|
||||
}
|
||||
|
||||
export function decodeTerminalStreamJson<T>(payload: Uint8Array): T | null {
|
||||
try {
|
||||
return JSON.parse(new TextDecoder().decode(payload)) as T
|
||||
} catch {
|
||||
return null
|
||||
}
|
||||
}
|
||||
|
||||
export function encodeTerminalStreamText(value: string): Uint8Array {
|
||||
return new TextEncoder().encode(value)
|
||||
}
|
||||
|
||||
export function decodeTerminalStreamText(payload: Uint8Array): string {
|
||||
return new TextDecoder().decode(payload)
|
||||
}
|
||||
|
||||
function isTerminalStreamOpcode(value: number): value is TerminalStreamOpcode {
|
||||
return (
|
||||
value === TerminalStreamOpcode.Output ||
|
||||
value === TerminalStreamOpcode.SnapshotStart ||
|
||||
value === TerminalStreamOpcode.SnapshotChunk ||
|
||||
value === TerminalStreamOpcode.SnapshotEnd ||
|
||||
value === TerminalStreamOpcode.Resized ||
|
||||
value === TerminalStreamOpcode.Error
|
||||
)
|
||||
}
|
||||
Reference in New Issue
Block a user