Terminal Integration Patterns
Terminal Integration Patterns
When to Use This Skill
Use this skill when implementing terminal integration patterns patterns in your codebase.
How to Use This Skill
- Review the patterns and examples below
- Apply the relevant patterns to your implementation
- Follow the best practices outlined in this skill
Level 1: Quick Reference (Under 500 tokens)
Core Terminal Integration Patterns
1. PTY Process Spawning (Node.js)
const pty = require('node-pty');
// Spawn shell with PTY
const shell = pty.spawn('bash', [], {
name: 'xterm-256color',
cols: 80,
rows: 24,
cwd: process.env.HOME,
env: process.env
});
shell.onData((data) => {
console.log('Output:', data);
});
shell.write('ls -la\r');
2. WebSocket Terminal Bridge
const WebSocket = require('ws');
const wss = new WebSocket.Server({ port: 8080 });
wss.on('connection', (ws) => {
const terminal = pty.spawn('bash', [], {
name: 'xterm-256color',
cols: 80,
rows: 24
});
terminal.onData((data) => ws.send(data));
ws.on('message', (msg) => terminal.write(msg));
ws.on('close', () => terminal.kill());
});
3. Xterm.js Frontend
<!DOCTYPE html>
<html>
<head>
<link rel="stylesheet" href="node_modules/xterm/css/xterm.css" />
<script src="node_modules/xterm/lib/xterm.js"></script>
</head>
<body>
<div id="terminal"></div>
<script>
const term = new Terminal();
term.open(document.getElementById('terminal'));
const ws = new WebSocket('ws://localhost:8080');
ws.onmessage = (e) => term.write(e.data);
term.onData((data) => ws.send(data));
</script>
</body>
</html>
4. Terminal Session Persistence
import json
from datetime import datetime
class TerminalSession:
def __init__(self, session_id):
self.session_id = session_id
self.buffer = []
self.started_at = datetime.utcnow()
def record(self, data):
self.buffer.append({
'timestamp': datetime.utcnow().isoformat(),
'data': data
})
def save(self):
with open(f'sessions/{self.session_id}.json', 'w') as f:
json.dump({
'session_id': self.session_id,
'started_at': self.started_at.isoformat(),
'buffer': self.buffer
}, f)
5. Container Terminal Exec
# Docker exec with PTY
docker exec -it container_name /bin/bash
# Kubernetes exec
kubectl exec -it pod-name -- /bin/bash
# Programmatic access
docker exec -i container_name /bin/bash << 'EOF'
ls -la
pwd
EOF
Level 2: Implementation Details (Under 2000 tokens)
Production Terminal Server
1. Complete WebSocket Terminal Server
// terminal-server.js
const express = require('express');
const http = require('http');
const WebSocket = require('ws');
const pty = require('node-pty');
const os = require('os');
class TerminalServer {
constructor(port = 8080) {
this.port = port;
this.app = express();
this.server = http.createServer(this.app);
this.wss = new WebSocket.Server({ server: this.server });
this.terminals = new Map();
this.logs = new Map();
}
start() {
this.setupRoutes();
this.setupWebSocket();
this.server.listen(this.port, () => {
console.log(`Terminal server listening on port ${this.port}`);
});
}
setupRoutes() {
// Serve static files
this.app.use(express.static('public'));
// Terminal creation endpoint
this.app.post('/terminals', (req, res) => {
const terminalId = this.createTerminal();
res.json({ id: terminalId });
});
// Terminal list
this.app.get('/terminals', (req, res) => {
res.json({
terminals: Array.from(this.terminals.keys())
});
});
// Session history
this.app.get('/terminals/:id/history', (req, res) => {
const logs = this.logs.get(req.params.id) || [];
res.json({ history: logs });
});
}
setupWebSocket() {
this.wss.on('connection', (ws, req) => {
const terminalId = new URL(req.url, 'http://localhost').searchParams.get('id');
if (!terminalId || !this.terminals.has(terminalId)) {
ws.close(1008, 'Terminal not found');
return;
}
const terminal = this.terminals.get(terminalId);
const logs = this.logs.get(terminalId);
// Send existing output
terminal.onData((data) => {
if (ws.readyState === WebSocket.OPEN) {
ws.send(data);
logs.push({ timestamp: Date.now(), type: 'output', data });
}
});
// Handle input
ws.on('message', (msg) => {
const message = JSON.parse(msg);
if (message.type === 'input') {
terminal.write(message.data);
logs.push({ timestamp: Date.now(), type: 'input', data: message.data });
} else if (message.type === 'resize') {
terminal.resize(message.cols, message.rows);
}
});
// Cleanup
ws.on('close', () => {
// Keep terminal alive but disconnect client
console.log(`Client disconnected from terminal ${terminalId}`);
});
});
}
createTerminal() {
const terminalId = this.generateId();
const shell = os.platform() === 'win32' ? 'powershell.exe' : 'bash';
const terminal = pty.spawn(shell, [], {
name: 'xterm-256color',
cols: 80,
rows: 24,
cwd: process.env.HOME,
env: process.env
});
terminal.on('exit', () => {
this.terminals.delete(terminalId);
console.log(`Terminal ${terminalId} exited`);
});
this.terminals.set(terminalId, terminal);
this.logs.set(terminalId, []);
return terminalId;
}
generateId() {
return Math.random().toString(36).substring(2, 15);
}
}
// Start server
const server = new TerminalServer(8080);
server.start();
2. Advanced Xterm.js Client
// terminal-client.js
import { Terminal } from 'xterm';
import { FitAddon } from 'xterm-addon-fit';
import { WebLinksAddon } from 'xterm-addon-web-links';
import { SearchAddon } from 'xterm-addon-search';
class TerminalClient {
constructor(containerElement) {
this.container = containerElement;
this.terminal = new Terminal({
cursorBlink: true,
fontSize: 14,
fontFamily: 'Menlo, Monaco, "Courier New", monospace',
theme: {
background: '#1e1e1e',
foreground: '#d4d4d4',
cursor: '#ffffff'
}
});
// Addons
this.fitAddon = new FitAddon();
this.webLinksAddon = new WebLinksAddon();
this.searchAddon = new SearchAddon();
this.terminal.loadAddon(this.fitAddon);
this.terminal.loadAddon(this.webLinksAddon);
this.terminal.loadAddon(this.searchAddon);
this.ws = null;
this.terminalId = null;
}
async connect() {
// Create terminal session
const response = await fetch('/terminals', { method: 'POST' });
const data = await response.json();
this.terminalId = data.id;
// Open terminal
this.terminal.open(this.container);
this.fitAddon.fit();
// Connect WebSocket
this.ws = new WebSocket(`ws://${location.host}/?id=${this.terminalId}`);
this.ws.onopen = () => {
console.log('Terminal connected');
this.terminal.focus();
};
this.ws.onmessage = (event) => {
this.terminal.write(event.data);
};
this.ws.onerror = (error) => {
console.error('WebSocket error:', error);
this.terminal.write('\r\n\x1b[31mConnection error\x1b[0m\r\n');
};
this.ws.onclose = () => {
this.terminal.write('\r\n\x1b[33mConnection closed\x1b[0m\r\n');
};
// Handle input
this.terminal.onData((data) => {
if (this.ws && this.ws.readyState === WebSocket.OPEN) {
this.ws.send(JSON.stringify({ type: 'input', data }));
}
});
// Handle resize
this.terminal.onResize(({ cols, rows }) => {
if (this.ws && this.ws.readyState === WebSocket.OPEN) {
this.ws.send(JSON.stringify({ type: 'resize', cols, rows }));
}
});
// Fit on window resize
window.addEventListener('resize', () => {
this.fitAddon.fit();
});
}
disconnect() {
if (this.ws) {
this.ws.close();
this.ws = null;
}
}
search(term) {
this.searchAddon.findNext(term);
}
}
// Usage
const terminalClient = new TerminalClient(document.getElementById('terminal'));
terminalClient.connect();
3. Session Recording & Playback
# terminal-recorder.py
import json
import time
from datetime import datetime
from pathlib import Path
class TerminalRecorder:
def __init__(self, session_id):
self.session_id = session_id
self.events = []
self.start_time = time.time()
self.output_dir = Path('recordings')
self.output_dir.mkdir(exist_ok=True)
def record_output(self, data):
"""Record terminal output"""
self.events.append({
'type': 'output',
'time': time.time() - self.start_time,
'data': data.decode('utf-8', errors='replace') if isinstance(data, bytes) else data
})
def record_input(self, data):
"""Record terminal input"""
self.events.append({
'type': 'input',
'time': time.time() - self.start_time,
'data': data
})
def record_resize(self, cols, rows):
"""Record terminal resize"""
self.events.append({
'type': 'resize',
'time': time.time() - self.start_time,
'cols': cols,
'rows': rows
})
def save(self):
"""Save recording to file"""
recording = {
'version': 2,
'width': 80,
'height': 24,
'timestamp': datetime.utcnow().isoformat(),
'duration': time.time() - self.start_time,
'events': self.events
}
output_file = self.output_dir / f'{self.session_id}.json'
with open(output_file, 'w') as f:
json.dump(recording, f, indent=2)
return str(output_file)
class TerminalPlayer:
def __init__(self, recording_file):
with open(recording_file) as f:
self.recording = json.load(f)
self.events = self.recording['events']
def play(self, speed=1.0):
"""Playback recorded session"""
for i, event in enumerate(self.events):
if i > 0:
# Calculate delay
delay = (event['time'] - self.events[i-1]['time']) / speed
time.sleep(delay)
if event['type'] == 'output':
print(event['data'], end='', flush=True)
def export_text(self):
"""Export session as plain text"""
return ''.join(
event['data']
for event in self.events
if event['type'] == 'output'
)
4. Docker Container Terminal Integration
# docker-terminal.py
import docker
import asyncio
import websockets
import json
class DockerTerminal:
def __init__(self):
self.client = docker.from_env()
async def connect_to_container(self, websocket, container_id):
"""Connect WebSocket to container terminal"""
try:
container = self.client.containers.get(container_id)
# Create exec instance
exec_id = container.exec_run(
'/bin/bash',
stdin=True,
tty=True,
stream=True,
demux=True,
socket=True
)
socket = exec_id.output._sock
async def read_from_container():
loop = asyncio.get_event_loop()
while True:
data = await loop.run_in_executor(None, socket.recv, 4096)
if not data:
break
await websocket.send(data.decode('utf-8', errors='replace'))
async def write_to_container():
async for message in websocket:
data = json.loads(message)
if data['type'] == 'input':
socket.send(data['data'].encode())
await asyncio.gather(
read_from_container(),
write_to_container()
)
except Exception as e:
await websocket.send(json.dumps({
'type': 'error',
'message': str(e)
}))
async def websocket_handler(websocket, path):
"""WebSocket handler for Docker terminal"""
params = dict(param.split('=') for param in path[1:].split('&'))
container_id = params.get('container')
if not container_id:
await websocket.close(1008, 'Container ID required')
return
terminal = DockerTerminal()
await terminal.connect_to_container(websocket, container_id)
# Start WebSocket server
start_server = websockets.serve(websocket_handler, 'localhost', 8765)
asyncio.get_event_loop().run_until_complete(start_server)
asyncio.get_event_loop().run_forever()
Level 3: Complete Reference (Full tokens)
Production-Grade Terminal Infrastructure
1. Multi-User Terminal Server with Authentication
// secure-terminal-server.js
const express = require('express');
const session = require('express-session');
const passport = require('passport');
const LocalStrategy = require('passport-local').Strategy;
const WebSocket = require('ws');
const pty = require('node-pty');
const Redis = require('ioredis');
class SecureTerminalServer {
constructor(config) {
this.config = config;
this.app = express();
this.redis = new Redis(config.redis);
this.terminals = new Map();
this.userSessions = new Map();
this.setupMiddleware();
this.setupAuthentication();
this.setupRoutes();
this.setupWebSocket();
}
setupMiddleware() {
this.app.use(express.json());
this.app.use(session({
secret: this.config.sessionSecret,
resave: false,
saveUninitialized: false,
cookie: { secure: true, httpOnly: true, maxAge: 3600000 }
}));
this.app.use(passport.initialize());
this.app.use(passport.session());
}
setupAuthentication() {
passport.use(new LocalStrategy(
async (username, password, done) => {
// Validate user (implement your auth logic)
const user = await this.validateUser(username, password);
return done(null, user || false);
}
));
passport.serializeUser((user, done) => {
done(null, user.id);
});
passport.deserializeUser(async (id, done) => {
const user = await this.getUser(id);
done(null, user);
});
}
setupRoutes() {
// Login
this.app.post('/login', passport.authenticate('local'), (req, res) => {
res.json({ success: true, user: req.user });
});
// Create terminal (authenticated)
this.app.post('/terminals', this.requireAuth.bind(this), async (req, res) => {
const terminalId = await this.createTerminalForUser(req.user);
res.json({ id: terminalId });
});
// List user's terminals
this.app.get('/terminals', this.requireAuth.bind(this), (req, res) => {
const userTerminals = this.getUserTerminals(req.user.id);
res.json({ terminals: userTerminals });
});
}
requireAuth(req, res, next) {
if (req.isAuthenticated()) {
return next();
}
res.status(401).json({ error: 'Unauthorized' });
}
async createTerminalForUser(user) {
const terminalId = this.generateId();
// Spawn terminal with user's context
const terminal = pty.spawn('bash', [], {
name: 'xterm-256color',
cols: 80,
rows: 24,
cwd: `/home/${user.username}`,
env: {
...process.env,
USER: user.username,
HOME: `/home/${user.username}`
},
uid: user.uid, // Run as user
gid: user.gid
});
// Store terminal metadata in Redis
await this.redis.hset(`terminal:${terminalId}`, {
user_id: user.id,
username: user.username,
created_at: Date.now(),
last_active: Date.now()
});
this.terminals.set(terminalId, terminal);
if (!this.userSessions.has(user.id)) {
this.userSessions.set(user.id, new Set());
}
this.userSessions.get(user.id).add(terminalId);
// Auto-cleanup on exit
terminal.on('exit', () => {
this.cleanupTerminal(terminalId, user.id);
});
return terminalId;
}
setupWebSocket() {
this.wss = new WebSocket.Server({ noServer: true });
this.wss.on('connection', async (ws, req, user) => {
const url = new URL(req.url, 'http://localhost');
const terminalId = url.searchParams.get('id');
// Verify terminal ownership
const terminalData = await this.redis.hgetall(`terminal:${terminalId}`);
if (terminalData.user_id !== user.id.toString()) {
ws.close(1008, 'Access denied');
return;
}
const terminal = this.terminals.get(terminalId);
if (!terminal) {
ws.close(1008, 'Terminal not found');
return;
}
// Pipe terminal output to WebSocket
terminal.onData((data) => {
if (ws.readyState === WebSocket.OPEN) {
ws.send(data);
}
});
// Handle WebSocket messages
ws.on('message', async (msg) => {
const message = JSON.parse(msg);
if (message.type === 'input') {
terminal.write(message.data);
// Log command for audit
await this.logCommand(user.id, terminalId, message.data);
} else if (message.type === 'resize') {
terminal.resize(message.cols, message.rows);
}
// Update last active
await this.redis.hset(`terminal:${terminalId}`, 'last_active', Date.now());
});
});
}
async logCommand(userId, terminalId, command) {
// Audit logging
await this.redis.rpush(`audit:${userId}`, JSON.stringify({
terminal_id: terminalId,
command,
timestamp: Date.now()
}));
}
async cleanupTerminal(terminalId, userId) {
this.terminals.delete(terminalId);
await this.redis.del(`terminal:${terminalId}`);
const userTerminals = this.userSessions.get(userId);
if (userTerminals) {
userTerminals.delete(terminalId);
}
}
getUserTerminals(userId) {
const terminals = this.userSessions.get(userId);
return terminals ? Array.from(terminals) : [];
}
generateId() {
return Math.random().toString(36).substring(2, 15);
}
async validateUser(username, password) {
// Implement user validation
return null;
}
async getUser(id) {
// Implement user retrieval
return null;
}
}
module.exports = SecureTerminalServer;
2. Kubernetes Pod Terminal Integration
# k8s-terminal.py
from kubernetes import client, config, stream
import asyncio
import websockets
import json
class KubernetesTerminal:
def __init__(self):
config.load_kube_config()
self.v1 = client.CoreV1Api()
async def exec_pod_command(self, websocket, namespace, pod_name, container=None):
"""Execute interactive shell in Kubernetes pod"""
exec_command = ['/bin/bash']
try:
# Create WebSocket stream to pod
resp = stream.stream(
self.v1.connect_get_namespaced_pod_exec,
pod_name,
namespace,
command=exec_command,
container=container,
stderr=True,
stdin=True,
stdout=True,
tty=True,
_preload_content=False
)
async def read_from_pod():
"""Read output from pod and send to WebSocket"""
while resp.is_open():
await asyncio.sleep(0.01)
if resp.peek_stdout():
output = resp.read_stdout()
await websocket.send(json.dumps({
'type': 'output',
'data': output
}))
if resp.peek_stderr():
error = resp.read_stderr()
await websocket.send(json.dumps({
'type': 'error',
'data': error
}))
async def write_to_pod():
"""Write input from WebSocket to pod"""
async for message in websocket:
data = json.loads(message)
if data['type'] == 'input':
resp.write_stdin(data['data'])
await asyncio.gather(
read_from_pod(),
write_to_pod()
)
except Exception as e:
await websocket.send(json.dumps({
'type': 'error',
'message': str(e)
}))
finally:
resp.close()
Best Practices:
- Use PTY for proper terminal emulation
- Implement session persistence for reconnection
- Add authentication and authorization
- Record sessions for audit compliance
- Handle terminal resize events properly
- Implement connection timeouts and cleanup
- Use secure WebSocket (wss://) in production
- Sanitize and validate all input commands
Success Output
When successful, this skill MUST output:
✅ SKILL COMPLETE: terminal-integration-patterns
Completed:
- [x] PTY process spawning configured (node-pty)
- [x] WebSocket bridge established
- [x] xterm.js frontend integrated
- [x] Session recording/playback implemented
- [x] Container terminal exec working (Docker/K8s)
Outputs:
- terminal-server.js (WebSocket server)
- terminal-client.js (xterm.js frontend)
- terminal-recorder.py (session recording)
- public/index.html (terminal UI)
Verification:
- Terminal connection established: ws://localhost:8080
- Input/output bidirectional: ✓
- Resize events handled: ✓
- Session persistence: ✓
Completion Checklist
Before marking this skill as complete, verify:
- node-pty installed and spawning shell processes
- WebSocket server listening on configured port
- xterm.js loaded and terminal rendered in browser
- Bidirectional data flow working (input → shell → output → browser)
- Terminal resize events propagated correctly
- Session recording captures input/output with timestamps
- Playback reproduces session accurately
- Authentication middleware protects WebSocket endpoint
- Container exec integration tested (docker/kubectl)
- Graceful cleanup on connection close
Failure Indicators
This skill has FAILED if:
- ❌ PTY process spawn errors (shell not found, permission denied)
- ❌ WebSocket connection refused or times out
- ❌ Terminal displays garbled output (encoding issues)
- ❌ Input not reaching shell (write to closed PTY)
- ❌ Output not displayed in terminal (WebSocket send failures)
- ❌ Resize events not applied (terminal size mismatch)
- ❌ Session recording incomplete or corrupted
- ❌ Unauthenticated access to terminal sessions
- ❌ Memory leaks from unclosed terminals
- ❌ Container exec hangs or fails to connect
When NOT to Use
Do NOT use this skill when:
- Simple command execution sufficient (use child_process.exec instead)
- No interactive shell needed (batch scripts, cron jobs)
- Security model prohibits shell access (use API endpoints)
- Browser not available (server-side only application)
- Real-time collaboration not required (use SSH directly)
- Container access restricted (use kubectl logs for output only)
Use alternatives:
- child_process.exec - For non-interactive command execution
- ssh-exec-patterns - For remote server command execution
- log-streaming-patterns - For read-only output monitoring
- api-command-patterns - For structured command interfaces
Anti-Patterns (Avoid)
| Anti-Pattern | Problem | Solution |
|---|---|---|
| No authentication on WebSocket | Anyone can access terminals | Implement session-based auth with passport |
| Running as root user | Security vulnerability | Spawn terminals with user's UID/GID |
| No session timeouts | Terminals accumulate indefinitely | Implement idle timeout and auto-cleanup |
| Exposing shell over HTTP | Credentials/data transmitted in clear | Use WSS (WebSocket Secure) in production |
| No input sanitization | Command injection vulnerabilities | Validate/escape input (though PTY inherently safer) |
| Single terminal per page | Poor resource utilization | Multiplex terminals via terminal IDs |
| No terminal size detection | Output wraps incorrectly | Always send resize events on window change |
| Storing passwords in recordings | Security/compliance violation | Redact sensitive input in session logs |
Principles
This skill embodies:
- #1 Search Before Create - Uses node-pty and xterm.js (industry standard libraries)
- #5 Eliminate Ambiguity - Clear WebSocket message protocol (type: input/output/resize)
- #6 Clear, Understandable, Explainable - Session recordings with timestamps for auditability
- #8 No Assumptions - Validates terminal exists before WebSocket attachment
- Security First - Authentication, authorization, session isolation
- Separation of Concerns - Server (PTY), transport (WebSocket), client (xterm.js) layers
Standard: CODITECT-STANDARD-AUTOMATION.md
Version: 1.1.0 | Updated: 2026-01-04 | Quality Standard Applied