This document outlines the plan for integrating the standalone tmux-based CLI agents (Claude Code and Gemini) into the existing SimulateDev backend. The goal is to enable the Tauri macOS app to execute these agents via the backend API with real-time output streaming.
- Architecture Diagrams
- Current State Analysis
- Migration Strategy
- Implementation Details
- Configuration Management
- Security & Authentication
- Key Design Decisions
- Error Handling & Recovery
- Monitoring & Observability
- Risk Mitigation
- Deployment Requirements
- Test Data & Examples
- Rollback Strategy
- Migration Checklist
- Success Criteria
- Timeline
- Implementation Tips
- Next Steps
┌─────────────────────┐ ┌─────────────────────┐
│ Tauri Frontend │ │ REST API │
│ - Task Creation │────▶│ - /api/tasks/* │
│ - Progress Display │ │ - Authentication │
│ - WebSocket Client │ └─────────┬───────────┘
└─────────────────────┘ │
▲ ▼
│ ┌─────────────────────┐
│ │ Task Service │
│ │ - Task Management │
│ │ - Agent Routing │
│ └─────────┬───────────┘
│ │
WebSocket Updates ▼
│ ┌─────────────────────┐
│ │ Tmux Service │
│ │ - Session Mgmt │
│ │ - Command Queue │
│ │ - Output Capture │
│ └─────────┬───────────┘
│ │
│ ▼
│ ┌─────────────────────────┐
│ │ CLI Agents │
│ │ ┌─────────┬─────────┐ │
│ │ │ Claude │ Gemini │ │
│ │ │ CLI │ CLI │ │
│ │ └─────────┴─────────┘ │
│ └─────────────────────────┘
│ │
│ ▼
│ ┌─────────────────────┐
┌──────────┴──────────┐ │ Output Stream │
│ WebSocket Manager │◀──│ Adapter │
│ - Real-time Updates│ │ - Format Output │
│ - Connection Mgmt │ │ - Buffer Updates │
└─────────────────────┘ └─────────┬───────────┘
│
▼
┌─────────────────────┐
│ SQLite Database │
│ - Task History │
│ - Output Logs │
└─────────────────────┘
1. Tauri App ──────▶ API: Create Task
│
2. ├──▶ Task Service: Initialize
│
3. ├──▶ Tmux Service: Create Session
│
4. ├──▶ CLI Agent: Start in Tmux Pane
│
5. └──▶ WebSocket: Connect for Updates
│
6. CLI Agent ─────────────┼──▶ Generate Output
│
7. Tmux Service ──────────┼──▶ Capture Output
│
8. Output Adapter ────────┼──▶ Format & Stream
│
9. WebSocket Manager ─────┼──▶ Send to Frontend
│
10. Tauri App ◀───────────┘ Display Progress
-
Core Files:
tmux_operations_manager.py- Main tmux session management with per-pane command queueingtmux_gemini_standalone.py- FastAPI wrapper with WebSocket streaming- Test files validating the implementation
-
Key Features:
- Per-pane command isolation preventing cross-session interference
- YOLO mode support for automated execution
- Real-time output streaming via WebSocket
- Session state management (SPAWNING → RUNNING → DONE)
- Adaptive monitoring for performance optimization
-
Agent System:
- Base agent classes for different IDE types (desktop, web, CLI)
- Agent factory pattern for instantiation
- Task service orchestrating execution
-
Communication Layer:
- WebSocket manager for real-time updates
- SQLite database for execution history
- REST API endpoints for task management
- Communicates via REST API for task execution
- WebSocket client for real-time progress updates
- Expects structured task status and output
⚠️ IMPORTANT: Incremental Migration ApproachBefore implementing any WebSocket streaming, API integration, or complex features, first focus on getting the core tmux functionality working within the backend environment. This means:
- Start Simple: Extract and adapt
tmux_operations_manager.pyto work as a backend service- Test Isolation: Ensure tmux sessions can be created, managed, and cleaned up properly
- Verify Core Logic: Confirm the per-pane command queueing system works in the new environment
- Basic CLI Agent Integration: Get one CLI agent (e.g., Gemini) working through the tmux service
Only after confirming the tmux core works reliably in its new "home" should we proceed with WebSocket streaming, API endpoints, and full integration. This incremental approach reduces risk and makes debugging much easier.
api/app/services/tmux_service.py
- Extract core functionality from
tmux_operations_manager.py - Remove duplicate enums/models (use existing backend structures)
- Adapt to use existing logging and configuration
- Integrate with backend's threading model
agents/base.py
- Add new agent type:
CLI_AGENTalongside existing types - Create
CLIAgentbase class extendingCodingAgent - Define interface for CLI-specific operations
agents/claude_cli_agent.py (rename existing to claude_code_agent.py)
agents/gemini_cli_agent.py
- Extend new
CLIAgentbase class - Implement agent-specific configurations
- Handle YOLO mode settings
- Map to existing agent response structures
api/app/services/output_stream_adapter.py
- Bridge between tmux output buffers and WebSocket manager
- Convert tmux output format to frontend-expected format
- Handle incremental updates efficiently
- Integrate with SQLite for output persistence
- Add support for CLI agent execution path
- Route CLI agents through tmux service
- Maintain compatibility with existing desktop/web agents
- Handle agent-specific execution parameters
- Adapt existing WebSocket manager for CLI agent output
- Ensure message format compatibility with Tauri frontend
- Add CLI-specific progress indicators
-- Add to execution_history table
ALTER TABLE execution_history ADD COLUMN output_buffer TEXT;
ALTER TABLE execution_history ADD COLUMN agent_session_id VARCHAR(100);
-- Add CLI agent configuration table
CREATE TABLE cli_agent_configs (
id VARCHAR(36) PRIMARY KEY,
agent_type VARCHAR(50),
yolo_mode BOOLEAN DEFAULT FALSE,
pre_commands JSON,
ready_indicators JSON,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);- Extend task execution endpoint to accept CLI agent parameters
- Add YOLO mode configuration option
- Ensure backward compatibility for existing agent types
- Test tmux service isolation
- Test CLI agent initialization
- Test output streaming adapter
- Test database operations
- End-to-end task execution via API
- WebSocket streaming validation
- Concurrent session handling
- Error recovery scenarios
- Verify Tauri app can execute CLI agents
- Validate real-time output display
- Test progress indicators
- Ensure UI responsiveness
# agents/factory.py
from .claude_cli_agent import ClaudeCliAgent
from .gemini_cli_agent import GeminiCliAgent
class AgentFactory:
@staticmethod
def create_agent(agent_type: CodingAgentIdeType, computer_use_client) -> CodingAgent:
# ... existing code ...
elif agent_type == CodingAgentIdeType.CLAUDE_CLI:
return ClaudeCliAgent(computer_use_client)
elif agent_type == CodingAgentIdeType.GEMINI_CLI:
return GeminiCliAgent(computer_use_client)# api/app/services/task_service.py
async def _execute_task_internal(self, task_id: str, github_token: str):
# ... existing code ...
# Check if CLI agent
if agent_config['coding_ide'] in ['claude_cli', 'gemini_cli']:
# Route through tmux service
result = await self._execute_cli_agent(task_id, task_request, github_token)
else:
# Existing orchestrator path
result = await self._execute_with_orchestrator(task_id, task_request, github_token)// Ensure compatibility with existing frontend
interface TaskProgressUpdate {
type: "progress" | "output" | "error" | "completion";
task_id: string;
progress?: number;
current_phase?: string;
output?: string; // For CLI agent output chunks
timestamp: string;
}# .env additions
TMUX_MAX_SESSIONS=50
TMUX_MONITOR_INTERVAL=5.0
TMUX_SESSION_TIMEOUT=1800
GEMINI_API_KEY=xxx
CLAUDE_API_KEY=xxxEach CLI agent implements its own get_config() method to provide agent-specific configuration using a structured CLIAgentConfig class:
# agents/base.py
from dataclasses import dataclass
from typing import List
@dataclass
class CLIAgentConfig:
command: List[str]
supports_yolo: bool
pre_commands: List[str]
ready_indicators: List[str]
# agents/gemini_cli_agent.py
class GeminiCliAgent(CLIAgent):
@classmethod
def get_config(cls) -> CLIAgentConfig:
return CLIAgentConfig(
command=["gemini", "2>&1"],
supports_yolo=True,
pre_commands=["export GEMINI_API_KEY=${GEMINI_API_KEY}"],
ready_indicators=["Type your message"]
)
# agents/claude_cli_agent.py
class ClaudeCliAgent(CLIAgent):
@classmethod
def get_config(cls) -> CLIAgentConfig:
return CLIAgentConfig(
command=["claude", "--permission-mode", "acceptEdits"],
supports_yolo=True,
pre_commands=[],
ready_indicators=["esc to interrupt"]
)This approach provides type safety, better IDE support, and ensures all required configuration fields are present while allowing each agent to define its own specific configuration.
- Keep existing agent architecture intact
- Add CLI agents as a new agent type alongside desktop/web agents
- Route CLI agents through tmux service while others use existing orchestrator
- Use existing WebSocket manager for streaming
- Leverage existing SQLite schema with minimal additions
- Maintain current API contract with Tauri frontend
- Extract core functionality from
tmux_operations_manager.py - Keep per-pane command queueing system intact
- Maintain session state machine (SPAWNING → RUNNING → DONE)
- Tmux Service: Session and pane management
- CLI Agents: Agent-specific configurations and behaviors
- Output Adapter: Format conversion and streaming
- Task Service: High-level orchestration
- No breaking changes to existing API
- Frontend continues to work without modifications unless modifications lead to a cleaner and simpler code
- Existing agents unaffected by CLI agent addition
class TmuxErrorHandler:
@staticmethod
async def handle_session_error(session_id: str, error: Exception):
"""Handle tmux session errors with appropriate recovery"""
if isinstance(error, TmuxSessionNotFound):
# Session died unexpectedly
await cleanup_dead_session(session_id)
await notify_user_session_failed(session_id)
elif isinstance(error, TmuxCommandTimeout):
# Command queue blocked
await force_kill_pane(session_id)
await restart_session_if_needed(session_id)- Scenario: CLI agent fails to start
- Detection: No ready indicators within timeout
- Recovery: Notify user
- Scenario: Agent produces excessive output
- Detection: Buffer size exceeds threshold
- Recovery: Implement circular buffer with size limits
- Scenario: Frontend loses connection during execution
- Detection: WebSocket disconnect event
- Recovery: Buffer output for reconnection within time window
- Automatic Retry: For transient failures (network, resource contention)
- Graceful Degradation: Fallback to non-YOLO mode if YOLO fails
- Session Recovery: Ability to reconnect to existing sessions
- Clean Shutdown: Ensure all resources cleaned up on failure
# Structured logging for tmux operations
logger.info("tmux_session_created", extra={
"session_id": session_id,
"agent_type": agent_type,
"user_id": user_id,
"yolo_mode": yolo_mode,
"duration_ms": elapsed_time
})# Install script for CLI agents
#!/bin/bash
# install_cli_agents.sh
# Install Claude CLI
curl -fsSL https://claude.ai/cli/install.sh | sh
# Install Gemini CLI
pip install google-generativeai-cli
# Verify installations
claude --version || exit 1
gemini --version || exit 1{
"small_repo": "https://github.com/saharmor/cursor-chat-view",
"medium_repo": "https://github.com/facebook/react",
"large_repo": "https://github.com/microsoft/vscode"
}# Simple test prompts
SIMPLE_PROMPTS = [
"Print the current time in Berlin",
"Fix the typo in the main function",
"Echo 'hello world'"
]
# Medium complexity prompts
MEDIUM_PROMPTS = [
"Add a README.md file with project description",
"Write a basic HTTP server that returns 'Hello World'",
"Create a JSON configuration file with database settings"
]
# Complex prompts requiring YOLO mode
COMPLEX_PROMPTS = [
"Refactor the authentication system to use JWT tokens",
"Add unit tests for the user service",
"Implement pagination for the API endpoints"
]# tests/integration/test_cli_agents.py
class TestCLIAgentIntegration:
async def test_simple_task_execution(self):
"""Test basic CLI agent execution"""
task_id = await create_task(
agent_type="gemini_cli",
prompt="Create a hello world Python script",
yolo_mode=False
)
result = await wait_for_completion(task_id, timeout=300)
assert result.success
assert "hello_world.py" in result.files_created
async def test_concurrent_sessions(self):
"""Test multiple concurrent CLI sessions"""
tasks = []
for i in range(5):
task_id = await create_task(
agent_type="claude_cli" if i % 2 else "gemini_cli",
prompt=f"Create test_file_{i}.txt",
yolo_mode=True
)
tasks.append(task_id)
results = await asyncio.gather(*[
wait_for_completion(tid) for tid in tasks
])
assert all(r.success for r in results)
async def test_session_isolation(self):
"""Verify sessions don't interfere with each other"""
# Implementation similar to test_tmux_cross_pane_input.py
passsimulatedev/
├── agents/
│ ├── base.py (updated with CLIAgent base class)
│ ├── claude_code_agent.py (existing, renamed)
│ ├── claude_cli_agent.py (new)
│ ├── gemini_cli_agent.py (new)
│ └── factory.py (updated)
├── api/
│ └── app/
│ ├── services/
│ │ ├── tmux_service.py (new)
│ │ ├── output_stream_adapter.py (new)
│ │ └── task_service.py (updated)
│ └── api/
│ └── health.py (updated)
├── tests/
│ ├── unit/
│ │ ├── test_tmux_service.py (new)
│ │ └── test_cli_agents.py (new)
│ └── integration/
│ ├── test_cli_agent_integration.py (new)
│ └── test_websocket_streaming.py (new)
└── config/
└── cli_agents.py (new)
- Back up existing database
- Document current API endpoints
- Implement Phase 1 components
- Write unit tests for new modules
- Implement Phase 2 streaming
- Update API documentation (OpenAPI/Swagger)
- Implement Phase 3 database changes
- Create integration tests
- Test with Tauri frontend
- Load testing with 20+ concurrent sessions
-
Functional Requirements
- CLI agents executable via Tauri app
- Real-time output streaming working
- Session isolation maintained
- YOLO mode functional
-
Quality Requirements
- 95%+ test coverage for new code
- No regression in existing functionality
- Clean error handling and recovery
- Zero data loss during migration
- Start Small: Test with a single CLI agent first (Gemini)
- Use Existing Patterns: Follow existing agent implementation patterns
- Test Early: Set up tmux in development environment immediately
- Monitor Resources: Watch for file descriptor and memory leaks
- Document Everything: Especially non-obvious tmux behaviors
- Don't Modify Core tmux Logic: Keep the per-pane queueing system intact
- Don't Store Secrets in Logs: Ensure API keys are never logged
- Don't Skip Cleanup: Always clean up tmux sessions on errors
- Don't Break Existing Agents: Test desktop/web agents still work
# WRONG: Tmux may not find the socket
subprocess.run(["tmux", "new-session"])
# CORRECT: Set proper environment
env = os.environ.copy()
env["TMUX_TMPDIR"] = "/tmp/tmux-sessions"
subprocess.run(["tmux", "new-session"], env=env)# WRONG: Checking state immediately after command
send_command_to_pane(pane_id, prompt)
if is_agent_ready(pane_id): # Too fast!
# CORRECT: Allow time for state changes
send_command_to_pane(pane_id, prompt)
await asyncio.sleep(0.5) # Give tmux time to process
if is_agent_ready(pane_id):# WRONG: Unbounded buffer growth
output_buffer += new_output
# CORRECT: Implement circular buffer
if len(output_buffer) > MAX_BUFFER_SIZE:
output_buffer = output_buffer[-MAX_BUFFER_SIZE:]
output_buffer += new_output# WRONG: Send every output change
for line in new_output.split('\n'):
await websocket.send(line)
# CORRECT: Batch updates
if time.time() - last_update > 0.1: # 100ms throttle
await websocket.send(batched_output)
last_update = time.time()- Enable Tmux Logging:
tmux -L debug -f /dev/null - Capture Pane History:
tmux capture-pane -p -S -1000 - Use Tmux Status Line: Display session info for debugging
- Review and approve this migration plan
- Begin Phase 1 implementation and track progress in a new markdown file
- Continue with other steps after getting confirmation from your manager, which is me!
Document Version: 1.0
Last Updated: Current Date
Author: Development Team