Optimize terminal input: coalesce keystrokes and hot-path RPC

Batch typing into short frames, send UTF-8 instead of base64 for text,
exempt stream methods from the general rate limit, and skip heavy
middleware on terminalInput/resize for lower latency.

server/handlers/containers.js

@@ -575,17 +575,23 @@ export function registerContainerHandlers(session) {
     return { success: true, message: 'Exec session started', execId: exec.id }
   })
 
-  session.respond('execInput', async (args) => {
-    const key = `exec:${args.execId}`
-    const entry = session.state.get(key)
-    if (!entry) throw new Error('Exec session not found')
-    const inputData =
-      args.encoding === 'base64'
-        ? Buffer.from(args.data, 'base64')
-        : Buffer.from(args.data || '', 'utf8')
-    entry.stream.write(inputData)
-    return { success: true }
-  })
+  session.respond(
+    'execInput',
+    (args) => {
+      const key = `exec:${args.execId}`
+      const entry = session.state.get(key)
+      if (!entry) return null
+      const inputData =
+        args.encoding === 'base64'
+          ? Buffer.from(args.data || '', 'base64')
+          : Buffer.from(args.data || '', 'utf8')
+      if (inputData.length && entry.stream && !entry.stream.writableEnded) {
+        entry.stream.write(inputData)
+      }
+      return null
+    },
+    { hot: true }
+  )
 }
 
 async function duplicateContainer(args, session) {

server/handlers/terminal.js

@@ -151,43 +151,52 @@ export function registerTerminalHandlers(session) {
     }
   })
 
-  session.respond('terminalInput', async (args) => {
-    const sessions = getSessions(session)
-    const sessionId = resolveSessionId(args)
-    const entry = sessions.get(sessionId) || session.state.get('terminal')
-    if (!entry) throw new Error('No active terminal session')
-    if (args.containerId && args.containerId !== entry.containerId) {
-      throw new Error('Terminal session container mismatch')
-    }
-    const inputData =
-      args.encoding === 'base64'
-        ? Buffer.from(args.data, 'base64')
-        : Buffer.from(args.data || '', 'utf8')
-    if (!entry.stream.writableEnded) {
-      entry.stream.write(inputData)
-    }
-    return { success: true }
-  })
+  // Hot path: events (id=0) — keep handler sync/cheap; no audit/metrics in session layer
+  session.respond(
+    'terminalInput',
+    (args) => {
+      const sessions = getSessions(session)
+      const sessionId = resolveSessionId(args)
+      const entry = sessions.get(sessionId) || session.state.get('terminal')
+      if (!entry) return null
+      if (args.containerId && args.containerId !== entry.containerId) return null
 
-  session.respond('terminalResize', async (args) => {
-    const sessions = getSessions(session)
-    const sessionId = resolveSessionId(args)
-    const entry = sessions.get(sessionId) || session.state.get('terminal')
-    if (!entry) return { success: false, message: 'No terminal session' }
-    if (args.containerId && args.containerId !== entry.containerId) {
-      return { success: false, message: 'Container mismatch' }
-    }
-    const cols = Number(args.cols)
-    const rows = Number(args.rows)
-    if (cols > 1 && rows > 0) {
-      try {
-        await entry.exec.resize({ h: rows, w: cols })
-      } catch (err) {
-        return { success: false, message: err.message }
+      let inputData
+      if (args.encoding === 'base64') {
+        inputData = Buffer.from(args.data || '', 'base64')
+      } else {
+        // Default utf8 string in JSON (optimal for keystrokes)
+        inputData = Buffer.from(args.data || '', 'utf8')
       }
-    }
-    return { success: true, cols, rows }
-  })
+      if (inputData.length && entry.stream && !entry.stream.writableEnded) {
+        entry.stream.write(inputData)
+      }
+      return null
+    },
+    { hot: true }
+  )
+
+  session.respond(
+    'terminalResize',
+    async (args) => {
+      const sessions = getSessions(session)
+      const sessionId = resolveSessionId(args)
+      const entry = sessions.get(sessionId) || session.state.get('terminal')
+      if (!entry) return null
+      if (args.containerId && args.containerId !== entry.containerId) return null
+      const cols = Number(args.cols)
+      const rows = Number(args.rows)
+      if (cols > 1 && rows > 0) {
+        try {
+          await entry.exec.resize({ h: rows, w: cols })
+        } catch {
+          // ignore transient resize errors
+        }
+      }
+      return null
+    },
+    { hot: true }
+  )
 
   session.respond('killTerminal', async (args) => {
     const sessions = getSessions(session)

server/rpc/session.js

@@ -52,14 +52,44 @@ export class PeerSession {
    * @param {string} method
    * @param {(args: any, session: PeerSession) => Promise<any>|any} handler
    */
-  respond(method, handler) {
+  /**
+   * @param {string} method
+   * @param {(args: any, session: PeerSession) => Promise<any>|any} handler
+   * @param {{ hot?: boolean }} [opts] - hot path: skip schema/audit/metrics noise (streams)
+   */
+  respond(method, handler, opts = {}) {
+    const hot = opts.hot === true || rateLimiter.isStreamMethod?.(method)
     this.rpc.respond(method, encodings, async (args) => {
       if (!rateLimiter.isAllowed(this, method)) {
         const err = new Error('Rate limit exceeded. Please wait before making more requests.')
         err.code = 'RATE_LIMIT_EXCEEDED'
-        recordRpc(method, { ok: false, denied: true })
+        if (!hot) recordRpc(method, { ok: false, denied: true })
         throw err
       }
+
+      // Hot path: terminal/stream traffic — minimal middleware
+      if (hot) {
+        try {
+          assertAllowed(this.role, method)
+          const result = await handler(args ?? {}, this)
+          return result
+        } catch (err) {
+          if (err?.code === 'PERMISSION_DENIED') {
+            audit({
+              method,
+              peerId: this.id,
+              role: this.role,
+              ok: false,
+              error: err.message,
+              force: true,
+            })
+          }
+          const safe = new Error(sanitizeError(err))
+          safe.code = err.code || 'UNKNOWN_ERROR'
+          throw safe
+        }
+      }
+
       const t0 = Date.now()
       try {
         assertAllowed(this.role, method)

server/utils/rateLimiter.js

@@ -9,15 +9,30 @@ class RateLimiter {
     
     // Configuration
     this.config = {
-      maxRequests: 100,        // Max requests per window
+      maxRequests: 100,        // Max requests per window (non-stream)
       windowMs: 60000,         // 1 minute window
       commandLimits: {
         deployContainer: { max: 5, windowMs: 60000 },    // 5 deployments per minute
         dockerCommand: { max: 30, windowMs: 10000 },     // 30 commands per 10 seconds
         startContainer: { max: 20, windowMs: 60000 },   // 20 starts per minute
         stopContainer: { max: 20, windowMs: 60000 },     // 20 stops per minute
-      }
+      },
     };
+
+    /**
+     * High-frequency stream methods (keystrokes, resizes, chunks).
+     * Exempt from the general 100/min cap; optional soft ceiling.
+     */
+    this.streamMethods = new Set([
+      'terminalInput',
+      'terminalResize',
+      'execInput',
+      'attachInput',
+      'loadImageChunk',
+      'binaryStreamChunk',
+      'dockerTerminalResize',
+    ]);
+    this.streamLimit = { max: 6000, windowMs: 60000 }; // ~100/s sustained
   }
 
   /**
@@ -52,11 +67,24 @@ class RateLimiter {
     if (!this.requests.has(peerId)) {
       this.requests.set(peerId, {
         general: [],
+        stream: [],
         commands: {}
       });
     }
     
     const peerData = this.requests.get(peerId);
+    if (!peerData.stream) peerData.stream = [];
+
+    // Stream methods: separate high ceiling, never consume general budget
+    if (this.streamMethods.has(command)) {
+      const streamWindow = now - this.streamLimit.windowMs;
+      peerData.stream = peerData.stream.filter((t) => t > streamWindow);
+      if (peerData.stream.length >= this.streamLimit.max) {
+        return false;
+      }
+      peerData.stream.push(now);
+      return true;
+    }
     
     // Check general rate limit
     const generalWindow = now - this.config.windowMs;
@@ -93,6 +121,11 @@ class RateLimiter {
     return true;
   }
 
+  /** @param {string} method */
+  isStreamMethod(method) {
+    return this.streamMethods.has(method);
+  }
+
   /**
    * Clean up old entries
    * @param {number} now - Current timestamp