Terminal Session API

Overview

The Terminal Session API provides tmux-based terminal management with ZMQ messaging, process monitoring, and cross-platform support for macOS and Linux.

TermSesh Class

Manages terminal sessions using tmux and publishes output via ZMQ.

Constructor

class TermSesh:
    monitor: Optional[ProcessMonitor]
    
    def __init__(self, port=5555, terminal_app=None, session_name='zapper_session'):
        self.context = zmq.Context()
        self.publisher = self.context.socket(zmq.PUB)
        self.publisher.bind(f"tcp://*:{port}")
        self.system = platform.system()
        self.session_name = session_name
        self.terminal_process = None
        self.monitor = None
        self.tmux_stack = [""]

Parameters:

port (int) - ZMQ publisher port. Default: 5555
terminal_app (str, optional) - Custom terminal application
session_name (str) - Tmux session identifier. Default: ‘zapper_session’

Attributes:

context - ZMQ context for messaging
publisher - ZMQ PUB socket for broadcasting terminal output
system - Platform identifier (Darwin/Linux)
session_name - Tmux session name
terminal_process - Subprocess handle
monitor - ProcessMonitor instance
tmux_stack - History of terminal output states

Session Management

open_new_terminal()

Create and attach to a new tmux session with platform-specific terminal window.

def open_new_terminal(self):
    """Create an interactive terminal session with proper error handling"""
    try:
        self.kill_tmux_session()
        
        subprocess.run(['tmux', 'new-session', '-d', '-s', self.session_name])
        
        # Open a new terminal window and attach to the tmux session
        if platform.system() == "Darwin":  # macOS
            apple_script = f'''
                tell application "Terminal"
                    do script "tmux attach-session -t {self.session_name}"
                end tell
            '''
            subprocess.Popen(['osascript', '-e', apple_script])
        elif platform.system() == "Linux":
            subprocess.Popen([
                'gnome-terminal', '--', 'tmux', 'attach-session', '-t', self.session_name
            ])
        else:
            raise NotImplementedError("Windows not yet supported")
        
        time.sleep(1)  # Allow time for tmux session to start
        return True
    except Exception as e:
        print(f"Error creating tmux session: {e}")
        traceback.print_exc()
        return False

Returns:

bool - True if session created successfully, False otherwise

Platform Support:

macOS: Uses AppleScript to open Terminal.app
Linux: Uses gnome-terminal
Windows: Not yet supported (raises NotImplementedError)

Example:

term_sesh = TermSesh(session_name='my_session')
if term_sesh.open_new_terminal():
    print("Session started successfully")

kill_tmux_session()

Terminate the tmux session if it exists.

def kill_tmux_session(self):
    """Kill the tmux session if it exists"""
    try:
        # Check if session exists
        result = subprocess.run(
            ['tmux', 'has-session', '-t', self.session_name],
            capture_output=True
        )
        if result.returncode == 0:  # Session exists
            subprocess.run(['tmux', 'kill-session', '-t', self.session_name])
            print(f"Killed tmux session: {self.session_name}")
    except Exception as e:
        print(f"Error killing tmux session: {e}")

Example:

term_sesh.kill_tmux_session()

is_terminal_active()

Check if terminal session is active and valid.

def is_terminal_active(self) -> bool:
    """Check if terminal session is active and valid"""
    return (self.terminal_process is not None and
            self.terminal_process.poll() is None)

Returns:

bool - True if terminal is running, False otherwise

Command Execution

send_to_terminal()

Send a command to the tmux session for execution.

def send_to_terminal(self, command: str) -> Optional[str]:
    """Send a command to the tmux session"""
    try:
        subprocess.run(['tmux', 'send-keys', '-t', self.session_name, command, 'C-m'])
        return "Command sent"
    except Exception as e:
        print(f"Error sending to tmux session: {e}")
        return None

Parameters:

command (str) - Shell command to execute

Returns:

str | None - “Command sent” on success, None on error

Example:

term_sesh.send_to_terminal("ls -la")
term_sesh.send_to_terminal("python script.py")

Output Management

read_tmux_output()

Read and process output from the tmux session.

def read_tmux_output(self):
    """Read output from the tmux session"""
    try:
        result = subprocess.run(
            ['tmux', 'capture-pane', '-t', self.session_name, '-p'], 
            capture_output=True, 
            text=True
        )
        if result.stdout:
            cleaned_output = self.clean_tmux_output(result.stdout)
            self.tmux_stack.append(cleaned_output)
            if len(self.tmux_stack[-2]) < len(cleaned_output):
                print(cleaned_output[len(self.tmux_stack[-2]):])
                return cleaned_output[len(self.tmux_stack[-2]):]
        return ""
    except Exception as e:
        print(f"Error reading tmux session: {e}")
        return ""

Returns:

str - New output since last read, or empty string

Behavior:

Captures current tmux pane content
Cleans and formats output
Tracks changes in tmux_stack
Returns only incremental output

clean_tmux_output()

Clean and format tmux output by removing ANSI codes and extra whitespace.

def clean_tmux_output(self, raw_output: str) -> str:
    """Clean and format tmux output by removing ANSI escape sequences and extra whitespace"""
    import re

    # Remove ANSI escape sequences
    ansi_escape = re.compile(r'\x1B(?:[@-Z\\-_]|\[[0-?]*[ -/]*[@-~])')
    # Remove unicode characters commonly used in prompts
    unicode_chars = re.compile(r'[\ue0b0-\ue0b3\ue5ff\ue615\ue606\uf48a\uf489\ue0a0]')

    # Clean the output
    cleaned = ansi_escape.sub('', raw_output)
    cleaned = unicode_chars.sub('', cleaned)

    # Split into lines and remove empty lines
    lines = [line.strip() for line in cleaned.split('\n')]
    lines = [line for line in lines if line]

    # Remove duplicate consecutive lines
    unique_lines = []
    prev_line = None
    for line in lines:
        if line != prev_line:
            unique_lines.append(line)
            prev_line = line

    return '\n'.join(unique_lines)

Parameters:

raw_output (str) - Raw tmux output with ANSI codes

Returns:

str - Cleaned and formatted output

Cleaning Steps:

Remove ANSI escape sequences
Remove Unicode prompt characters
Strip whitespace from lines
Remove empty lines
Deduplicate consecutive identical lines

Code Transmission

send_code_segment()

Send compressed code segments with metadata via ZMQ.

def send_code_segment(self, code_data):
    """
    Send code segments with metadata
    code_data = {
        'file_path': 'path/to/file',
        'code': 'actual code content',
        'metadata': {...},
        'action': 'analyze/edit/debug'
    }
    """
    # Compress the code content
    compressed_code = base64.b64encode(
        zlib.compress(code_data['code'].encode())
    ).decode()
    code_data['code'] = compressed_code

    # Send the data
    self.publisher.send_json(code_data)

Parameters:

code_data (dict) - Dictionary containing:
- file_path (str) - Path to source file
- code (str) - Source code content
- metadata (dict, optional) - Additional metadata
- action (str) - Action type: ‘analyze’, ‘edit’, or ‘debug’

Behavior:

Compresses code using zlib
Base64 encodes compressed data
Publishes via ZMQ publisher socket

Example:

code_data = {
    'file_path': 'src/main.py',
    'code': 'def hello():\n    print("Hello")\n',
    'action': 'analyze',
    'metadata': {'language': 'python'}
}
term_sesh.send_code_segment(code_data)

ProcessMonitor Class

Monitors tmux output in a background thread and publishes updates via ZMQ.

Constructor

class ProcessMonitor:
    def __init__(self, term_sesh):
        self.term_sesh = term_sesh
        self.monitor_thread = None
        self.is_running = False

Parameters:

term_sesh (TermSesh) - TermSesh instance to monitor

Attributes:

term_sesh - Associated terminal session
monitor_thread - Background monitoring thread
is_running - Thread execution flag

Methods

start_monitoring()

Start background thread to monitor tmux output.

def start_monitoring(self):
    self.is_running = True
    self.monitor_thread = threading.Thread(target=self.monitor_tmux)
    self.monitor_thread.daemon = True
    self.monitor_thread.start()
    print('monitor thread starting')

Behavior:

Creates daemon thread
Starts monitoring loop
Non-blocking operation

monitor_tmux()

Continuously read tmux output and publish via ZMQ.

def monitor_tmux(self):
    while self.is_running:
        output = self.term_sesh.read_tmux_output()
        if output:
            self.term_sesh.publisher.send_json({
                'type': 'tmux_output',
                'data': output,
                'timestamp': time.time()
            })
        time.sleep(1)

Message Format:

{
  "type": "tmux_output",
  "data": "<terminal output>",
  "timestamp": 1234567890.123
}

Polling Interval: 1 second

Complete Example

from term_sesh import TermSesh
from process_monitor import ProcessMonitor

# Create terminal session
term_sesh = TermSesh(port=5555, session_name='dev_session')

# Start tmux session with monitoring
if term_sesh.open_new_terminal():
    monitor = ProcessMonitor(term_sesh)
    monitor.start_monitoring()
    
    # Execute commands
    term_sesh.send_to_terminal("cd /project")
    term_sesh.send_to_terminal("python test.py")
    
    # Send code for analysis
    code_data = {
        'file_path': 'src/app.py',
        'code': open('src/app.py').read(),
        'action': 'debug'
    }
    term_sesh.send_code_segment(code_data)
    
    # Cleanup
    monitor.is_running = False
    term_sesh.kill_tmux_session()

Source References

/home/daytona/workspace/source/cli/term_sesh.py - TermSesh implementation
/home/daytona/workspace/source/cli/process_monitor.py - ProcessMonitor implementation

Core Modules

Agents

CLI

Terminal Session API

Overview

TermSesh Class

Constructor

Session Management

open_new_terminal()

kill_tmux_session()

is_terminal_active()

Command Execution

send_to_terminal()

Output Management

read_tmux_output()

clean_tmux_output()

Code Transmission

send_code_segment()

ProcessMonitor Class

Constructor

Methods

start_monitoring()

monitor_tmux()

Complete Example

Source References

Build docs developers (and LLMs) love

Core Modules

Agents

CLI

​Overview

​TermSesh Class

​Constructor

​Session Management

​open_new_terminal()

​kill_tmux_session()

​is_terminal_active()

​Command Execution

​send_to_terminal()

​Output Management

​read_tmux_output()

​clean_tmux_output()

​Code Transmission

​send_code_segment()

​ProcessMonitor Class

​Constructor

​Methods

​start_monitoring()

​monitor_tmux()

​Complete Example

​Source References

Build docs developers (and LLMs) love

Overview

TermSesh Class

Constructor

Session Management

open_new_terminal()

kill_tmux_session()

is_terminal_active()

Command Execution

send_to_terminal()

Output Management

read_tmux_output()

clean_tmux_output()

Code Transmission

send_code_segment()

ProcessMonitor Class

Constructor

Methods

start_monitoring()

monitor_tmux()

Complete Example

Source References