feat: ttyd process pool + terminal instance cache for instant session switching

Two independent optimizations that eliminate the ~700ms-1.2s lag when switching sessions:

### Backend: ttyd process pool (ttyd.py + main.py)
Instead of killing and respawning a single ttyd process on every session switch, maintain a pool of ttyd processes - one per connected session, each on its own port (7682-7701). Revisiting a session reuses the existing process instantly.

- New pool API: get_or_spawn(), pool_port(), kill_session(), kill_all(), kill_orphans()
- Backward-compat aliases preserved: kill_ttyd(), spawn_ttyd(), kill_orphan_ttyd(), TTYD_PORT
- /connect endpoint now returns in ~0ms for already-connected sessions vs 700ms+ before
- WS proxy reads ?session= query param to route to correct ttyd port
- Per-port PID files for orphan recovery across restarts

### Frontend: terminal instance cache (terminal.js + app.js)
Keep up to 5 xterm.js Terminal instances alive in hidden container divs. Switching to an already-visited session is a CSS display swap + fit() call - no xterm.js teardown/recreation, no WebSocket reconnection, no addon reloading.

- New switchTerminal() entry point with LRU cache (max 5)
- Each cached entry has its own container div, term, WebSocket, addons
- Cache-aware /connect skip: if terminal is cached with live WS, bypass backend entirely
- Animation gate fixed: 0ms for session-to-session switches (was always 260ms)
- Cache cleanup on session delete

### Test changes (minimal)
- test_ttyd.py: Added pool cleanup in fixture for isolation
- test_api.py: Updated 4 tests to mock new pool API
- test_ws_proxy.py: Updated auto-spawn test for pool API

All 1,306 tests pass.

Generated with Amplifier

Co-Authored-By: Amplifier <240397093+microsoft-amplifier@users.noreply.github.com>
This commit is contained in:
Ken
2026-05-28 00:48:21 +00:00
parent 967b30f8e0
commit e032d540b0
7 changed files with 661 additions and 143 deletions
+19 -3
View File
@@ -2295,6 +2295,9 @@ function _executeKill(name, remoteId) {
.then(function() { .then(function() {
closeFlyoutMenu(); closeFlyoutMenu();
showToast('Session \'' + name + '\' killed'); showToast('Session \'' + name + '\' killed');
// Clean up terminal cache entry for the deleted session
var sessionKey = remoteId ? (remoteId + ':' + name) : name;
if (window._destroyCachedTerminal) window._destroyCachedTerminal(sessionKey);
if (_viewingSession === name && (_viewingRemoteId ?? '') === (remoteId || '')) { if (_viewingSession === name && (_viewingRemoteId ?? '') === (remoteId || '')) {
closeSession(); closeSession();
} }
@@ -2893,7 +2896,7 @@ async function openSession(name, opts = {}) {
initSidebar(); initSidebar();
renderSidebar(_currentSessions, name, _viewingRemoteId); renderSidebar(_currentSessions, name, _viewingRemoteId);
resolve(); resolve();
}, opts.skipAnimation ? 0 : 260); }, opts.skipAnimation ? 0 : (tile ? 260 : 0));
// If setTimeout is stubbed (e.g. in test env), resolve immediately so we don't hang // If setTimeout is stubbed (e.g. in test env), resolve immediately so we don't hang
if (timerId == null) resolve(); if (timerId == null) resolve();
}); });
@@ -2913,9 +2916,16 @@ async function openSession(name, opts = {}) {
const fab = $('new-session-fab'); const fab = $('new-session-fab');
if (fab) fab.classList.add('hidden'); if (fab) fab.classList.add('hidden');
// Always spawn ttyd for this session — ensures correct session after service restart or page restore
// _deviceId holds the device_id string (was integer remoteId index in old protocol) // _deviceId holds the device_id string (was integer remoteId index in old protocol)
var _deviceId = opts.remoteId != null ? opts.remoteId : ''; var _deviceId = opts.remoteId != null ? opts.remoteId : '';
// Skip /connect POST if the session is already cached with a live WebSocket —
// this is the biggest win, bypassing the entire backend round-trip.
var sessionKey = _deviceId ? (_deviceId + ':' + name) : name;
var isCached = window._isTerminalCached && window._isTerminalCached(sessionKey);
if (!isCached) {
// Spawn ttyd for this session — ensures correct session after service restart or page restore
try { try {
if (_deviceId !== '') { if (_deviceId !== '') {
// Remote session: route connect POST through same-origin federation proxy // Remote session: route connect POST through same-origin federation proxy
@@ -2927,6 +2937,7 @@ async function openSession(name, opts = {}) {
showToast(err.message || 'Connection failed'); showToast(err.message || 'Connection failed');
return closeSession(); return closeSession();
} }
}
// Persist active_remote_id so restoreState() can reopen remote sessions after page refresh // Persist active_remote_id so restoreState() can reopen remote sessions after page refresh
api('PATCH', '/api/state', { active_session: name, active_remote_id: _deviceId || null }).catch(function() {}); api('PATCH', '/api/state', { active_session: name, active_remote_id: _deviceId || null }).catch(function() {});
@@ -2940,7 +2951,9 @@ async function openSession(name, opts = {}) {
await animDone; await animDone;
// Mount terminal NOW — /connect has completed, new ttyd is serving the correct session // Mount terminal NOW — /connect has completed, new ttyd is serving the correct session
if (window._openTerminal) window._openTerminal(name, _deviceId, getDisplaySettings().fontSize); // Prefer _switchTerminal (cache-aware) over _openTerminal (always creates fresh)
if (window._switchTerminal) window._switchTerminal(name, _deviceId, getDisplaySettings().fontSize);
else if (window._openTerminal) window._openTerminal(name, _deviceId, getDisplaySettings().fontSize);
} }
/** /**
@@ -3912,6 +3925,9 @@ function killSession(name, remoteId) {
api('DELETE', endpoint) api('DELETE', endpoint)
.then(function() { .then(function() {
showToast('Session \'' + name + '\' killed'); showToast('Session \'' + name + '\' killed');
// Clean up terminal cache entry for the deleted session
var sessionKey = remoteId ? (remoteId + ':' + name) : name;
if (window._destroyCachedTerminal) window._destroyCachedTerminal(sessionKey);
// If we deleted the session we're currently viewing, return to dashboard // If we deleted the session we're currently viewing, return to dashboard
if (_viewingSession === name && (_viewingRemoteId ?? '') === (remoteId || '')) { if (_viewingSession === name && (_viewingRemoteId ?? '') === (remoteId || '')) {
closeSession(); closeSession();
+349 -13
View File
@@ -1,7 +1,7 @@
// Phase 2b implementation — terminal.js // Phase 2b implementation — terminal.js
// xterm.js Terminal + FitAddon initialization (task-12) // xterm.js Terminal + FitAddon initialization (task-12)
// ─── Module-level state ─────────────────────────────────────────────────────── // ——— Module-level state ———————————————————————————————————————————
let _term = null; let _term = null;
let _fitAddon = null; let _fitAddon = null;
let _ws = null; let _ws = null;
@@ -15,7 +15,20 @@ let _resizeObserver = null;
let _ctrlActive = false; // mobile toolbar: Ctrl modifier toggle (one-shot) let _ctrlActive = false; // mobile toolbar: Ctrl modifier toggle (one-shot)
let _altActive = false; // mobile toolbar: Alt modifier toggle (one-shot) let _altActive = false; // mobile toolbar: Alt modifier toggle (one-shot)
// ─── Module-level encoding helpers ────────────────────────────────────────── // ——— Terminal instance cache ——————————————————————————————————————
// Maps sessionKey → {container, term, fitAddon, ws, searchAddon, resizeObserver,
// reconnectTimer, overlayTimer, reconnectAttempts, currentSession, remoteId}
var _terminalCache = new Map();
var _MAX_CACHED = 5; // LRU eviction when exceeded
var _activeSessionKey = null; // currently visible session key
var _cacheOrder = []; // LRU tracking (most recent at end)
// Session key format: remoteId ? (remoteId + ':' + name) : name
function _buildSessionKey(name, remoteId) {
return remoteId ? (remoteId + ':' + name) : name;
}
// ——— Module-level encoding helpers ————————————————————————————————
// Hoisted here so the clipboard key handler (in openTerminal) can also use them. // Hoisted here so the clipboard key handler (in openTerminal) can also use them.
const _encoder = typeof TextEncoder !== 'undefined' ? new TextEncoder() : null; const _encoder = typeof TextEncoder !== 'undefined' ? new TextEncoder() : null;
// TextDecoder: used to decode UTF-8 bytes received from ttyd before writing to xterm.js. // TextDecoder: used to decode UTF-8 bytes received from ttyd before writing to xterm.js.
@@ -33,7 +46,7 @@ function _encodePayload(typeChar, str) {
return payload; return payload;
} }
// ─── Clipboard helpers ─────────────────────────────────────────────────────── // ——— Clipboard helpers ———————————————————————————————————————————
// Ctrl+Shift+C: copy terminal selection to system clipboard // Ctrl+Shift+C: copy terminal selection to system clipboard
// Ctrl+Shift+V: handled natively by xterm.js (browser paste event → xterm → WebSocket) // Ctrl+Shift+V: handled natively by xterm.js (browser paste event → xterm → WebSocket)
@@ -53,7 +66,64 @@ function _copyToClipboard(text) {
} }
} }
// ─── Forward declarations ───────────────────────────────────────────────────── // ——— Cache helpers ———————————————————————————————————————————————
/**
* Sync module-level variables to point at the active cached entry's values.
* This keeps backward compat for search, font size changes, etc.
*/
function _syncModuleState(entry) {
if (entry) {
_term = entry.term;
_fitAddon = entry.fitAddon;
_ws = entry.ws;
_searchAddon = entry.searchAddon;
_resizeObserver = entry.resizeObserver;
_reconnectTimer = entry.reconnectTimer;
_overlayTimer = entry.overlayTimer;
_reconnectAttempts = entry.reconnectAttempts;
_currentSession = entry.currentSession;
} else {
_term = null;
_fitAddon = null;
_ws = null;
_searchAddon = null;
_resizeObserver = null;
_reconnectTimer = null;
_overlayTimer = null;
_reconnectAttempts = 0;
_currentSession = null;
}
}
/**
* Update the cached entry from current module-level variables.
* Call this after any module-level state changes to keep the entry in sync.
*/
function _syncEntryFromModule(key) {
var entry = _terminalCache.get(key);
if (!entry) return;
entry.term = _term;
entry.fitAddon = _fitAddon;
entry.ws = _ws;
entry.searchAddon = _searchAddon;
entry.resizeObserver = _resizeObserver;
entry.reconnectTimer = entry.reconnectTimer; // managed per-entry
entry.overlayTimer = entry.overlayTimer; // managed per-entry
entry.reconnectAttempts = _reconnectAttempts;
entry.currentSession = _currentSession;
}
/**
* Move a key to the end of _cacheOrder (most recently used).
*/
function _touchCacheOrder(key) {
var idx = _cacheOrder.indexOf(key);
if (idx !== -1) _cacheOrder.splice(idx, 1);
_cacheOrder.push(key);
}
// ——— Forward declarations ——————————————————————————————————————————
function connectWebSocket(name, remoteId) { function connectWebSocket(name, remoteId) {
// Always connect to the same origin — remote sessions route through the // Always connect to the same origin — remote sessions route through the
@@ -65,8 +135,8 @@ function connectWebSocket(name, remoteId) {
// Remote session via federation proxy — same origin, different path // Remote session via federation proxy — same origin, different path
url = proto + '//' + location.host + '/federation/' + remoteId + '/terminal/ws'; url = proto + '//' + location.host + '/federation/' + remoteId + '/terminal/ws';
} else { } else {
// Local session: same origin // Local session: add ?session= query param so backend proxy can route to correct ttyd port
url = proto + '//' + location.host + '/terminal/ws'; url = proto + '//' + location.host + '/terminal/ws?session=' + encodeURIComponent(name);
} }
const reconnectOverlay = document.getElementById('reconnect-overlay'); const reconnectOverlay = document.getElementById('reconnect-overlay');
// Use module-level _encodePayload (hoisted above connectWebSocket) // Use module-level _encodePayload (hoisted above connectWebSocket)
@@ -254,7 +324,7 @@ function initVisualViewport() {
window.visualViewport.addEventListener('resize', _vpHandler); window.visualViewport.addEventListener('resize', _vpHandler);
} }
// ─── Terminal creation ──────────────────────────────────────────────────────── // ——— Terminal creation ————————————————————————————————————————————
/** /**
* Create (or recreate) the xterm.js Terminal and FitAddon instances. * Create (or recreate) the xterm.js Terminal and FitAddon instances.
@@ -336,7 +406,7 @@ function createTerminal(fontSize) {
} }
} }
// ─── Search helpers ────────────────────────────────────────────────────────────────────────────────────────────────── // ——— Search helpers ——————————————————————————————————————————————————————————————————————
function _openSearch() { function _openSearch() {
var bar = document.getElementById('terminal-search-bar'); var bar = document.getElementById('terminal-search-bar');
@@ -371,7 +441,7 @@ function _searchPrev() {
} }
} }
// ─── Open / close ───────────────────────────────────────────────────────────── // ——— Open / close —————————————————————————————————————————————————
/** /**
* Open a terminal session inside #terminal-container. * Open a terminal session inside #terminal-container.
@@ -559,10 +629,248 @@ function openTerminal(sessionName, remoteId, fontSize) {
_initMobileToolbar(); _initMobileToolbar();
} }
// ——— switchTerminal — cache-aware session switching ————————————————
/**
* Switch to a terminal session, using cached instance if available.
* This is the preferred entry point for session switching — eliminates
* xterm.js teardown/recreation when switching between already-visited sessions.
* @param {string} sessionName
* @param {string} [remoteId] Optional federation remote ID.
* @param {number} [fontSize=14]
*/
function switchTerminal(sessionName, remoteId, fontSize) {
var key = _buildSessionKey(sessionName, remoteId);
// 1. Already showing this session — just focus and return
if (_activeSessionKey === key) {
if (_term) _term.focus();
return;
}
// 2. Hide the currently active entry (don't dispose — keep it cached)
if (_activeSessionKey && _terminalCache.has(_activeSessionKey)) {
var activeEntry = _terminalCache.get(_activeSessionKey);
// Save current module state back to the active entry before switching
activeEntry.ws = _ws;
activeEntry.reconnectTimer = _reconnectTimer;
activeEntry.overlayTimer = _overlayTimer;
activeEntry.reconnectAttempts = _reconnectAttempts;
if (activeEntry.container) {
activeEntry.container.style.display = 'none';
}
}
// 3. If target is in cache — instant switch, no network
if (_terminalCache.has(key)) {
var entry = _terminalCache.get(key);
if (entry.container) {
entry.container.style.display = '';
}
_syncModuleState(entry);
_activeSessionKey = key;
_touchCacheOrder(key);
if (_fitAddon) {
try { _fitAddon.fit(); } catch (_) {}
}
if (_term) _term.focus();
return;
}
// 4. Not in cache — create new entry
// Evict LRU if at capacity
if (_terminalCache.size >= _MAX_CACHED && _cacheOrder.length > 0) {
var evictKey = _cacheOrder[0];
destroyCachedEntry(evictKey);
}
// Create a new sub-container div inside #terminal-container
var parentContainer = document.getElementById('terminal-container');
if (!parentContainer) {
console.warn('[switchTerminal] #terminal-container not found');
return;
}
var subContainer = document.createElement('div');
subContainer.className = 'terminal-instance';
subContainer.setAttribute('data-session', key);
subContainer.style.width = '100%';
subContainer.style.height = '100%';
parentContainer.appendChild(subContainer);
// Reset module-level state for fresh creation
_currentSession = null;
_reconnectAttempts = 0;
_reconnectTimer = null;
_overlayTimer = null;
_ws = null;
_currentSession = sessionName;
createTerminal(fontSize);
_term.open(subContainer);
// Auto-refit on container resize
_resizeObserver = null;
if (typeof ResizeObserver !== 'undefined') {
var _roTimer = null;
_resizeObserver = new ResizeObserver(function() {
clearTimeout(_roTimer);
_roTimer = setTimeout(function() {
if (_fitAddon) try { _fitAddon.fit(); } catch (_) {}
}, 50);
});
_resizeObserver.observe(subContainer);
}
// Clipboard integration
_term.attachCustomKeyEventHandler(function(e) {
if (e.type !== 'keydown') return true;
if (e.ctrlKey && e.shiftKey && (e.key === 'C' || e.code === 'KeyC')) {
var sel = _term.getSelection();
if (sel) _copyToClipboard(sel);
return false;
}
if (e.ctrlKey && !e.shiftKey && (e.key === 'f' || e.key === 'F' || e.code === 'KeyF')) {
_openSearch();
return false;
}
return true;
});
_term.onSelectionChange(function() {
var sel = _term.getSelection();
if (sel) _copyToClipboard(sel);
});
_term.parser.registerOscHandler(52, function(data) {
var parts = data.split(';');
if (parts.length >= 2) {
try {
var text = atob(parts[1]);
_copyToClipboard(text);
} catch (e) {}
}
return true;
});
if (_fitAddon) {
var fitAddonRef = _fitAddon;
var raf = typeof requestAnimationFrame !== 'undefined' ? requestAnimationFrame : function(fn) { fn(); };
raf(function() {
try { fitAddonRef.fit(); } catch (_) {}
setTimeout(function() {
if (_fitAddon) try { _fitAddon.fit(); } catch (_) {}
}, 500);
});
}
// Right-click context menu suppression
subContainer.addEventListener('contextmenu', function(e) {
if (e.shiftKey || e.ctrlKey || e.metaKey) return;
e.preventDefault();
});
connectWebSocket(sessionName, remoteId);
initVisualViewport();
_initAndroidIMEFix(subContainer);
_initMobileToolbar();
// Wire search bar (idempotent)
var searchInput = document.getElementById('terminal-search-input');
var searchClose = document.getElementById('terminal-search-close');
var searchNextBtn = document.getElementById('terminal-search-next');
var searchPrevBtn = document.getElementById('terminal-search-prev');
if (searchInput) {
var newInput = searchInput.cloneNode(true);
searchInput.parentNode.replaceChild(newInput, searchInput);
searchInput = newInput;
searchInput.addEventListener('input', function() {
if (_searchAddon && searchInput.value) _searchAddon.findNext(searchInput.value);
else if (_searchAddon) _searchAddon.clearDecorations();
});
searchInput.addEventListener('keydown', function(e) {
if (e.key === 'Enter') { e.preventDefault(); if (e.shiftKey) _searchPrev(); else _searchNext(); }
if (e.key === 'Escape') { e.preventDefault(); _closeSearch(); }
});
}
if (searchClose) { var nc = searchClose.cloneNode(true); searchClose.parentNode.replaceChild(nc, searchClose); nc.addEventListener('click', _closeSearch); }
if (searchNextBtn) { var nn = searchNextBtn.cloneNode(true); searchNextBtn.parentNode.replaceChild(nn, searchNextBtn); nn.addEventListener('click', _searchNext); }
if (searchPrevBtn) { var np = searchPrevBtn.cloneNode(true); searchPrevBtn.parentNode.replaceChild(np, searchPrevBtn); np.addEventListener('click', _searchPrev); }
// Store in cache
var newEntry = {
container: subContainer,
term: _term,
fitAddon: _fitAddon,
ws: _ws,
searchAddon: _searchAddon,
resizeObserver: _resizeObserver,
reconnectTimer: _reconnectTimer,
overlayTimer: _overlayTimer,
reconnectAttempts: _reconnectAttempts,
currentSession: _currentSession,
remoteId: remoteId || '',
};
_terminalCache.set(key, newEntry);
_activeSessionKey = key;
_touchCacheOrder(key);
}
// ——— Cache entry teardown ————————————————————————————————————————
/**
* Full teardown of one cached entry: close WS, dispose term, remove container div, delete from cache.
* @param {string} key - session key to destroy
*/
function destroyCachedEntry(key) {
var entry = _terminalCache.get(key);
if (!entry) return;
// Clean up timers
if (entry.reconnectTimer) clearTimeout(entry.reconnectTimer);
if (entry.overlayTimer) clearTimeout(entry.overlayTimer);
// Close WebSocket
if (entry.ws) {
// Null out currentSession on entry so close handler won't reconnect
entry.currentSession = null;
entry.ws.close();
}
// Disconnect ResizeObserver
if (entry.resizeObserver) entry.resizeObserver.disconnect();
// Dispose xterm.js Terminal
if (entry.term) entry.term.dispose();
// Remove container div from DOM
if (entry.container && entry.container.parentNode) {
entry.container.parentNode.removeChild(entry.container);
}
// Remove from cache and LRU order
_terminalCache.delete(key);
var idx = _cacheOrder.indexOf(key);
if (idx !== -1) _cacheOrder.splice(idx, 1);
// If this was the active session, clear module state
if (_activeSessionKey === key) {
_activeSessionKey = null;
_syncModuleState(null);
}
}
/** /**
* Close the current terminal session and clean up all resources. * Close the current terminal session and clean up all resources.
* If the active session has a cache entry, destroys it.
*/ */
function closeTerminal() { function closeTerminal() {
// If we have a cached active session, destroy its cache entry
if (_activeSessionKey && _terminalCache.has(_activeSessionKey)) {
destroyCachedEntry(_activeSessionKey);
}
if (_vpHandler) { if (_vpHandler) {
if (window.visualViewport) window.visualViewport.removeEventListener('resize', _vpHandler); if (window.visualViewport) window.visualViewport.removeEventListener('resize', _vpHandler);
_vpHandler = null; _vpHandler = null;
@@ -601,11 +909,41 @@ function closeTerminal() {
_closeSearch(); _closeSearch();
_currentSession = null; _currentSession = null;
_reconnectAttempts = 0; // reset backoff on intentional close _reconnectAttempts = 0; // reset backoff on intentional close
_activeSessionKey = null;
} }
// ─── Expose to app.js ───────────────────────────────────────────────────────── /**
window._openTerminal = openTerminal; * Destroy all cached terminal entries.
*/
function closeAllTerminals() {
var keys = Array.from(_terminalCache.keys());
for (var i = 0; i < keys.length; i++) {
destroyCachedEntry(keys[i]);
}
_activeSessionKey = null;
_cacheOrder = [];
// Also clean up any non-cached module state
closeTerminal();
}
/**
* Check if a session is cached with a live (open) WebSocket.
* @param {string} sessionKey
* @returns {boolean}
*/
function isTerminalCached(sessionKey) {
var entry = _terminalCache.get(sessionKey);
if (!entry) return false;
return entry.ws && entry.ws.readyState === WebSocket.OPEN;
}
// ——— Expose to app.js ——————————————————————————————————————————————
window._switchTerminal = switchTerminal; // NEW — preferred entry point (cache-aware)
window._openTerminal = openTerminal; // KEEP for backward compat (creates fresh, no cache check)
window._closeTerminal = closeTerminal; window._closeTerminal = closeTerminal;
window._closeAllTerminals = closeAllTerminals; // NEW — destroy all cached entries
window._destroyCachedTerminal = destroyCachedEntry; // NEW — for deleting specific sessions
window._isTerminalCached = isTerminalCached; // NEW — check if session is cached with live WS
window._openSearch = _openSearch; window._openSearch = _openSearch;
window._closeSearch = _closeSearch; window._closeSearch = _closeSearch;
@@ -734,5 +1072,3 @@ function _initMobileToolbar() {
// Do NOT focus terminal — keeps soft keyboard hidden // Do NOT focus terminal — keeps soft keyboard hidden
}); });
} }
+52 -46
View File
@@ -78,7 +78,16 @@ from muxplex.settings import (
from muxplex.pruning import load_pruning_state, save_pruning_state from muxplex.pruning import load_pruning_state, save_pruning_state
from muxplex.views import normalize_session_keys, prune_stale_keys from muxplex.views import normalize_session_keys, prune_stale_keys
from muxplex.identity import load_device_id from muxplex.identity import load_device_id
from muxplex.ttyd import kill_orphan_ttyd, kill_ttyd, spawn_ttyd, TTYD_PORT from muxplex.ttyd import (
TTYD_PORT,
_ttyd_is_listening,
get_or_spawn,
kill_orphan_ttyd,
kill_session,
kill_ttyd, # noqa: F401 — backward compat re-export used by tests
pool_port,
spawn_ttyd, # noqa: F401 — backward compat re-export used by tests
)
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
# Configuration # Configuration
@@ -715,10 +724,11 @@ async def create_session(payload: CreateSessionPayload) -> dict:
async def connect_session(name: str) -> dict: async def connect_session(name: str) -> dict:
"""Connect to a tmux session via ttyd. """Connect to a tmux session via ttyd.
Kills any existing ttyd process, spawns a new one attached to *name*, Uses the process pool: if the session already has a live ttyd, returns
and updates the active_session in persistent state. immediately. Otherwise spawns a new ttyd on a free port, polls until
listening, and returns the port.
Returns {active_session: name, ttyd_port: 7682}. Returns {active_session: name, ttyd_port: <port>}.
Raises HTTP 404 if *name* is not in the known session list (when non-empty). Raises HTTP 404 if *name* is not in the known session list (when non-empty).
""" """
known = get_session_list() known = get_session_list()
@@ -726,40 +736,34 @@ async def connect_session(name: str) -> dict:
raise HTTPException(status_code=404, detail=f"Session '{name}' not found") raise HTTPException(status_code=404, detail=f"Session '{name}' not found")
_log.info("Connecting to session '%s'", name) _log.info("Connecting to session '%s'", name)
await kill_ttyd() port = await get_or_spawn(name)
await spawn_ttyd(name)
# Wait for ttyd to actually bind its port before returning.
# This eliminates the 0.8s blind sleep in the WebSocket proxy path —
# the client can connect immediately when this endpoint responds.
for _attempt in range(20): # up to ~1s (20 × 50ms)
if _ttyd_is_listening():
break
await asyncio.sleep(0.05)
async with state_lock: async with state_lock:
state = load_state() state = load_state()
state["active_session"] = name state["active_session"] = name
save_state(state) save_state(state)
return {"active_session": name, "ttyd_port": TTYD_PORT} return {"active_session": name, "ttyd_port": port}
@app.delete("/api/sessions/current") @app.delete("/api/sessions/current")
async def delete_current_session() -> dict: async def delete_current_session() -> dict:
"""Disconnect the current ttyd session. """Disconnect the current ttyd session.
Kills the running ttyd process and clears active_session in persistent state. Kills the pooled ttyd for the active session and clears active_session
in persistent state.
Returns {active_session: None}. Returns {active_session: None}.
""" """
await kill_ttyd()
async with state_lock: async with state_lock:
state = load_state() state = load_state()
active = state.get("active_session")
state["active_session"] = None state["active_session"] = None
save_state(state) save_state(state)
if active:
await kill_session(active)
return {"active_session": None} return {"active_session": None}
@@ -807,6 +811,10 @@ async def delete_session(name: str) -> dict:
except Exception: except Exception:
_log.warning("Delete command failed: %r", command) _log.warning("Delete command failed: %r", command)
# Clean up the pool entry (ttyd will die when tmux session dies,
# but clean up immediately so the port is freed).
await kill_session(name)
return {"ok": True, "name": name} return {"ok": True, "name": name}
@@ -988,19 +996,7 @@ async def instance_info() -> dict:
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
def _ttyd_is_listening() -> bool: # _ttyd_is_listening is imported from muxplex.ttyd (pool-aware, accepts port param)
"""Return True if something is accepting TCP connections on TTYD_PORT.
Uses a raw socket connect (no WebSocket handshake, no PTY spawned).
Takes < 1 ms on localhost when ttyd is running; fails immediately with
ConnectionRefusedError when it's not. OSError/TimeoutError are also
caught so the caller always gets a bool.
"""
try:
with socket.create_connection(("127.0.0.1", TTYD_PORT), timeout=0.5):
return True
except (ConnectionRefusedError, OSError, TimeoutError):
return False
async def _ws_auth_check(websocket: WebSocket) -> bool: async def _ws_auth_check(websocket: WebSocket) -> bool:
@@ -1033,9 +1029,10 @@ async def _ws_auth_check(websocket: WebSocket) -> bool:
async def terminal_ws_proxy(websocket: WebSocket) -> None: async def terminal_ws_proxy(websocket: WebSocket) -> None:
"""Proxy WebSocket frames between the browser and ttyd. """Proxy WebSocket frames between the browser and ttyd.
Checks that ttyd is alive BEFORE accepting the browser WebSocket. If ttyd Resolves the target ttyd port from the process pool BEFORE accepting the
is not listening (e.g. after a service restart), auto-spawns it using the browser WebSocket. If ttyd is not in the pool (e.g. after a service
active_session from state, then waits briefly for it to bind its port. restart), auto-spawns it using the session from query params or
active_session from state, then waits for it to bind its port.
Only after ttyd is confirmed reachable does the function call Only after ttyd is confirmed reachable does the function call
websocket.accept() — so the browser's 'open' event only fires once a real websocket.accept() — so the browser's 'open' event only fires once a real
@@ -1047,29 +1044,38 @@ async def terminal_ws_proxy(websocket: WebSocket) -> None:
if not await _ws_auth_check(websocket): if not await _ws_auth_check(websocket):
return return
# Ensure ttyd is reachable BEFORE accepting the browser WS. # Determine which session (and port) to connect to.
# After a service restart ttyd is dead but clients reconnect immediately. # Priority: ?session= query param > active_session from state.
# Auto-spawn from active_session so the browser's 'open' event only fires session = websocket.query_params.get("session", "")
# when a real relay is possible — eliminates the 0→1→0→1 counter bounce. if not session:
if not _ttyd_is_listening():
try: try:
async with state_lock: async with state_lock:
state = load_state() state = load_state()
session_name = state.get("active_session") session = state.get("active_session") or ""
if session_name: except Exception:
session = ""
# Look up port from pool, or auto-spawn if not listening
port = pool_port(session) if session else None
if port is None and session:
if not _ttyd_is_listening(pool_port(session) or TTYD_PORT):
_log.info( _log.info(
"WS proxy: ttyd not listening, auto-spawning for '%s'", "WS proxy: ttyd not listening, auto-spawning for '%s'",
session_name, session,
) )
await kill_ttyd() try:
await spawn_ttyd(session_name) port = await get_or_spawn(session)
await asyncio.sleep(0.8) # wait for ttyd to bind its port
except Exception as exc: except Exception as exc:
_log.warning("WS proxy: failed to auto-spawn ttyd: %s", exc) _log.warning("WS proxy: failed to auto-spawn ttyd: %s", exc)
if port is None:
# Fallback to TTYD_PORT (legacy single-process or externally managed ttyd)
port = TTYD_PORT
await websocket.accept(subprotocol="tty") await websocket.accept(subprotocol="tty")
ttyd_url = f"ws://localhost:{TTYD_PORT}/ws" ttyd_url = f"ws://localhost:{port}/ws"
try: try:
async with websockets.connect( async with websockets.connect(
ttyd_url, subprotocols=[Subprotocol("tty")] ttyd_url, subprotocols=[Subprotocol("tty")]
+20 -34
View File
@@ -371,15 +371,10 @@ def test_connect_session_returns_200(client, monkeypatch):
"""POST /api/sessions/{name}/connect returns 200 and correct body when session exists.""" """POST /api/sessions/{name}/connect returns 200 and correct body when session exists."""
monkeypatch.setattr("muxplex.main.get_session_list", lambda: ["alpha"]) monkeypatch.setattr("muxplex.main.get_session_list", lambda: ["alpha"])
async def mock_kill(): async def mock_get_or_spawn(name):
return True return 7682
monkeypatch.setattr("muxplex.main.kill_ttyd", mock_kill) monkeypatch.setattr("muxplex.main.get_or_spawn", mock_get_or_spawn)
async def mock_spawn(name):
pass
monkeypatch.setattr("muxplex.main.spawn_ttyd", mock_spawn)
response = client.post("/api/sessions/alpha/connect") response = client.post("/api/sessions/alpha/connect")
assert response.status_code == 200 assert response.status_code == 200
@@ -394,15 +389,10 @@ def test_connect_session_sets_active_session(client, monkeypatch):
monkeypatch.setattr("muxplex.main.get_session_list", lambda: ["alpha"]) monkeypatch.setattr("muxplex.main.get_session_list", lambda: ["alpha"])
async def mock_kill(): async def mock_get_or_spawn(name):
return True return 7682
monkeypatch.setattr("muxplex.main.kill_ttyd", mock_kill) monkeypatch.setattr("muxplex.main.get_or_spawn", mock_get_or_spawn)
async def mock_spawn(name):
pass
monkeypatch.setattr("muxplex.main.spawn_ttyd", mock_spawn)
client.post("/api/sessions/alpha/connect") client.post("/api/sessions/alpha/connect")
@@ -410,25 +400,21 @@ def test_connect_session_sets_active_session(client, monkeypatch):
assert state["active_session"] == "alpha" assert state["active_session"] == "alpha"
def test_connect_session_kills_existing_ttyd(client, monkeypatch): def test_connect_session_calls_get_or_spawn(client, monkeypatch):
"""POST /api/sessions/{name}/connect calls kill_ttyd then spawn_ttyd.""" """POST /api/sessions/{name}/connect calls get_or_spawn with the session name."""
monkeypatch.setattr("muxplex.main.get_session_list", lambda: ["alpha"]) monkeypatch.setattr("muxplex.main.get_session_list", lambda: ["alpha"])
call_order = [] spawn_calls = []
async def mock_kill(): async def mock_get_or_spawn(name):
call_order.append("kill") spawn_calls.append(name)
return True return 7682
async def mock_spawn(name): monkeypatch.setattr("muxplex.main.get_or_spawn", mock_get_or_spawn)
call_order.append(("spawn", name))
monkeypatch.setattr("muxplex.main.kill_ttyd", mock_kill)
monkeypatch.setattr("muxplex.main.spawn_ttyd", mock_spawn)
response = client.post("/api/sessions/alpha/connect") response = client.post("/api/sessions/alpha/connect")
assert response.status_code == 200 assert response.status_code == 200
assert call_order == ["kill", ("spawn", "alpha")] assert spawn_calls == ["alpha"]
def test_connect_nonexistent_session_returns_404(client, monkeypatch): def test_connect_nonexistent_session_returns_404(client, monkeypatch):
@@ -445,7 +431,7 @@ def test_connect_nonexistent_session_returns_404(client, monkeypatch):
def test_delete_current_kills_ttyd_and_clears_active(client, monkeypatch): def test_delete_current_kills_ttyd_and_clears_active(client, monkeypatch):
"""DELETE /api/sessions/current kills ttyd and clears active_session.""" """DELETE /api/sessions/current kills the session's ttyd and clears active_session."""
from muxplex.state import load_state, save_state from muxplex.state import load_state, save_state
# Set up initial state with active session # Set up initial state with active session
@@ -458,19 +444,19 @@ def test_delete_current_kills_ttyd_and_clears_active(client, monkeypatch):
} }
) )
kill_called = [] kill_session_calls = []
async def mock_kill(): async def mock_kill_session(name):
kill_called.append(True) kill_session_calls.append(name)
return True return True
monkeypatch.setattr("muxplex.main.kill_ttyd", mock_kill) monkeypatch.setattr("muxplex.main.kill_session", mock_kill_session)
response = client.delete("/api/sessions/current") response = client.delete("/api/sessions/current")
assert response.status_code == 200 assert response.status_code == 200
data = response.json() data = response.json()
assert data["active_session"] is None assert data["active_session"] is None
assert len(kill_called) == 1 assert kill_session_calls == ["alpha"]
# Verify state was persisted # Verify state was persisted
state = load_state() state = load_state()
+2
View File
@@ -32,6 +32,8 @@ def use_tmp_pid_dir(tmp_path, monkeypatch):
tmp_pid_path = tmp_pid_dir / "ttyd.pid" tmp_pid_path = tmp_pid_dir / "ttyd.pid"
monkeypatch.setattr("muxplex.ttyd.TTYD_PID_DIR", tmp_pid_dir) monkeypatch.setattr("muxplex.ttyd.TTYD_PID_DIR", tmp_pid_dir)
monkeypatch.setattr("muxplex.ttyd.TTYD_PID_PATH", tmp_pid_path) monkeypatch.setattr("muxplex.ttyd.TTYD_PID_PATH", tmp_pid_path)
# Clear the pool between tests
ttyd_mod._pool.clear()
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
+10 -18
View File
@@ -160,27 +160,19 @@ def test_ws_proxy_checks_ttyd_before_accepting():
def test_ws_proxy_auto_spawns_ttyd_when_dead(monkeypatch): def test_ws_proxy_auto_spawns_ttyd_when_dead(monkeypatch):
"""WS proxy must call spawn_ttyd when _ttyd_is_listening returns False.""" """WS proxy must call get_or_spawn when pool has no entry for the session."""
import asyncio
spawn_calls = [] spawn_calls = []
async def mock_spawn_ttyd(name: str): async def mock_get_or_spawn(name: str):
spawn_calls.append(name) spawn_calls.append(name)
return 7682
async def mock_kill_ttyd(): # pool_port returns None → no existing ttyd for this session
pass monkeypatch.setattr("muxplex.main.pool_port", lambda name: None)
# _ttyd_is_listening returns False → nothing listening on fallback port either
async def mock_sleep(_delay: float): monkeypatch.setattr("muxplex.main._ttyd_is_listening", lambda port=7682: False)
pass # no-op so tests don't actually wait # get_or_spawn is what the proxy should call to auto-spawn
monkeypatch.setattr("muxplex.main.get_or_spawn", mock_get_or_spawn)
# Patch _ttyd_is_listening to report ttyd as dead
monkeypatch.setattr("muxplex.main._ttyd_is_listening", lambda: False)
# Patch spawn_ttyd / kill_ttyd so tests don't touch real processes
monkeypatch.setattr("muxplex.main.spawn_ttyd", mock_spawn_ttyd)
monkeypatch.setattr("muxplex.main.kill_ttyd", mock_kill_ttyd)
# asyncio.sleep is called after spawn — patch to be a no-op
monkeypatch.setattr(asyncio, "sleep", mock_sleep)
# Provide a fake websockets.connect that immediately closes (no real ttyd) # Provide a fake websockets.connect that immediately closes (no real ttyd)
fake_ws = FakeTtydWs(responses=[]) fake_ws = FakeTtydWs(responses=[])
@@ -197,7 +189,7 @@ def test_ws_proxy_auto_spawns_ttyd_when_dead(monkeypatch):
pass pass
assert spawn_calls == ["test-session"], ( assert spawn_calls == ["test-session"], (
"spawn_ttyd must be called with active_session when ttyd is not listening" "get_or_spawn must be called with active_session when ttyd is not in pool"
) )
+192 -12
View File
@@ -2,14 +2,22 @@
ttyd process lifecycle management for the tmux-web muxplex. ttyd process lifecycle management for the tmux-web muxplex.
Constants: Constants:
TTYD_PID_DIR — directory for the PID file (default: ~/.local/share/tmux-web/) TTYD_PID_DIR — directory for PID files (default: ~/.local/share/tmux-web/)
TTYD_PID_PATH — full path to the PID file (TTYD_PID_DIR / 'ttyd.pid') TTYD_PID_PATH — full path to the legacy PID file (TTYD_PID_DIR / 'ttyd.pid')
TTYD_PORT — port ttyd listens on (7682) TTYD_PORT — base port for the ttyd pool (7682)
Module state: Module state:
_active_process — the currently running ttyd subprocess (or None) _active_process — the currently running ttyd subprocess (or None) [backward compat]
_pool — session→PoolEntry mapping for the process pool
Public API: Pool API:
get_or_spawn(session_name) — Return port; spawn if needed, poll until listening
pool_port(session_name) — Look up port for session without spawning
kill_session(session_name) — Kill ttyd for one session, remove from pool
kill_all() — Kill all pooled ttyd processes
kill_orphans() — Startup cleanup: PID files, port range scan
Backward-compat API:
spawn_ttyd(session_name) — spawn ttyd attached to a tmux session, write PID file spawn_ttyd(session_name) — spawn ttyd attached to a tmux session, write PID file
kill_ttyd() — kill the running ttyd process, clean up PID file kill_ttyd() — kill the running ttyd process, clean up PID file
kill_orphan_ttyd() — kill any orphaned ttyd from a previous coordinator run kill_orphan_ttyd() — kill any orphaned ttyd from a previous coordinator run
@@ -19,8 +27,10 @@ import asyncio
import os import os
import re as _re import re as _re
import signal import signal
import socket as _socket
import subprocess as _subprocess import subprocess as _subprocess
import time import time
from dataclasses import dataclass
from pathlib import Path from pathlib import Path
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
@@ -33,17 +43,65 @@ TTYD_PID_PATH: Path = TTYD_PID_DIR / "ttyd.pid"
TTYD_PORT: int = 7682 TTYD_PORT: int = 7682
_PORT_MIN: int = 7682
_PORT_MAX: int = 7701
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
# Module state # Module state
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
_active_process: asyncio.subprocess.Process | None = None _active_process: asyncio.subprocess.Process | None = None
@dataclass
class PoolEntry:
"""A pooled ttyd process and its assigned port."""
process: asyncio.subprocess.Process
port: int
_pool: dict[str, PoolEntry] = {}
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
# Internal helpers # Internal helpers
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
def _pid_path_for_port(port: int) -> Path:
"""Return the per-port PID file path: ``ttyd-{port}.pid`` in TTYD_PID_DIR."""
return TTYD_PID_DIR / f"ttyd-{port}.pid"
def _is_alive(entry: PoolEntry) -> bool:
"""Return True if the pool entry's process has not exited."""
return entry.process.returncode is None
def _ttyd_is_listening(port: int = TTYD_PORT) -> bool:
"""Return True if something is accepting TCP connections on *port*.
Uses a raw socket connect (no WebSocket handshake, no PTY spawned).
Takes < 1 ms on localhost when ttyd is running; fails immediately with
ConnectionRefusedError when it's not. OSError/TimeoutError are also
caught so the caller always gets a bool.
"""
try:
with _socket.create_connection(("127.0.0.1", port), timeout=0.5):
return True
except (ConnectionRefusedError, OSError, TimeoutError):
return False
def _allocate_port() -> int:
"""Return the next free port in the pool range [_PORT_MIN, _PORT_MAX]."""
used_ports = {e.port for e in _pool.values()}
for port in range(_PORT_MIN, _PORT_MAX + 1):
if port not in used_ports:
return port
raise RuntimeError("ttyd pool exhausted: all ports in range are in use")
def _pids_on_port(port: int) -> list[int]: def _pids_on_port(port: int) -> list[int]:
"""Return PIDs of all processes listening on *port*. """Return PIDs of all processes listening on *port*.
@@ -142,7 +200,118 @@ def _kill_pids_on_port(port: int, sig: int) -> bool:
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
# Public API # Pool API
# ---------------------------------------------------------------------------
async def get_or_spawn(session_name: str) -> int:
"""Return the port for *session_name*, spawning ttyd if needed.
If the session already has a live ttyd in the pool (process alive AND
port is listening), returns immediately. Otherwise allocates the next
free port from [_PORT_MIN, _PORT_MAX], spawns ttyd, polls until
listening (up to ~1 s), and returns the port.
"""
# Check existing pool entry — require both alive process AND listening port
entry = _pool.get(session_name)
if entry is not None:
if _is_alive(entry) and _ttyd_is_listening(entry.port):
return entry.port
# Stale entry — clean up
try:
entry.process.terminate()
except ProcessLookupError:
pass
_pid_path_for_port(entry.port).unlink(missing_ok=True)
del _pool[session_name]
# Allocate next free port
port = _allocate_port()
# Force-free the port (catches races)
if _kill_pids_on_port(port, signal.SIGKILL):
await asyncio.sleep(0.3)
# Spawn ttyd
proc = await asyncio.create_subprocess_exec(
"ttyd",
"-W",
"-m",
"3",
"-p",
str(port),
"tmux",
"attach",
"-t",
session_name,
stdout=asyncio.subprocess.DEVNULL,
stderr=asyncio.subprocess.DEVNULL,
start_new_session=True,
)
# Write per-port PID file
TTYD_PID_DIR.mkdir(parents=True, exist_ok=True)
_pid_path_for_port(port).write_text(str(proc.pid))
# Add to pool
_pool[session_name] = PoolEntry(process=proc, port=port)
# Poll until listening (up to ~1 s)
for _ in range(20):
if _ttyd_is_listening(port):
break
await asyncio.sleep(0.05)
return port
def pool_port(session_name: str) -> int | None:
"""Look up the port for *session_name* without spawning.
Returns the port if the session has a live pool entry, else None.
"""
entry = _pool.get(session_name)
if entry is not None and _is_alive(entry):
return entry.port
return None
async def kill_session(session_name: str) -> bool:
"""Kill the ttyd process for *session_name* and remove from pool.
Returns True if a pool entry was found and cleaned up, False otherwise.
"""
entry = _pool.pop(session_name, None)
if entry is None:
return False
try:
entry.process.terminate()
try:
await asyncio.wait_for(entry.process.wait(), timeout=2.0)
except asyncio.TimeoutError:
entry.process.kill()
except ProcessLookupError:
pass
_pid_path_for_port(entry.port).unlink(missing_ok=True)
return True
async def kill_all() -> None:
"""Kill all pooled ttyd processes and clear the pool."""
for name in list(_pool):
await kill_session(name)
async def kill_orphans() -> None:
"""Startup cleanup: read PID files, SIGTERM, scan port range for stragglers.
Handles both legacy ``ttyd.pid`` and per-port ``ttyd-{port}.pid`` files.
"""
await kill_orphan_ttyd()
# ---------------------------------------------------------------------------
# Backward-compat public API
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
@@ -226,17 +395,28 @@ async def kill_ttyd() -> bool:
async def kill_orphan_ttyd() -> bool: async def kill_orphan_ttyd() -> bool:
"""Kill any orphaned ttyd process left over from a previous coordinator run. """Kill any orphaned ttyd process left over from a previous coordinator run.
On coordinator startup, checks for a stale PID file from a previous run. Handles both legacy ``ttyd.pid`` and pool-era ``ttyd-{port}.pid`` files,
If found, kills the process (if still running) and removes the PID file. plus scans the full port range for stragglers.
Prevents two ttyd instances running simultaneously after a coordinator
restart or crash.
Delegates to kill_ttyd() for all process management and PID file cleanup.
Returns: Returns:
True — an orphan was found (process was dead or alive). True — an orphan was found (process was dead or alive).
False — no PID file found, or PID file contained invalid content. False — no PID file found, or PID file contained invalid content.
""" """
# Pool-era cleanup: per-port PID files
if TTYD_PID_DIR.exists():
for pid_file in TTYD_PID_DIR.glob("ttyd-*.pid"):
try:
pid = int(pid_file.read_text().strip())
os.kill(pid, signal.SIGTERM)
except (ValueError, ProcessLookupError, PermissionError):
pass
pid_file.unlink(missing_ok=True)
# Scan additional ports beyond TTYD_PORT
for port in range(_PORT_MIN + 1, _PORT_MAX + 1):
_kill_pids_on_port(port, signal.SIGTERM)
# Original behavior: legacy PID file + TTYD_PORT cleanup
return await kill_ttyd() return await kill_ttyd()