opencode_microservice.py

"""
OpenCode Microservice - Tmux Screen Capture Approach
====================================================
Uses tmux to capture terminal output
"""

import asyncio
import logging
import os
import json
import random
import string
import subprocess
from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
from typing import Optional
import shutil
import time

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("opencode_microservice")

app = FastAPI(title="OpenCode Microservice")

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

class OpenCodeSession:
    def __init__(self):
        self.tmux_session = "opencode"
        self.message_count = 0
        self.max_messages = 20
        self.session_id = None
        self.is_running = False
        
    def generate_identity(self):
        return {
            "device_id": ''.join(random.choices(string.hexdigits.lower(), k=32)),
            "session_id": ''.join(random.choices(string.hexdigits.lower(), k=16)),
        }
    
    def run_tmux_cmd(self, cmd):
        """Run tmux command"""
        try:
            result = subprocess.run(
                ['tmux', '-L', self.tmux_session] + cmd,
                capture_output=True,
                text=True,
                timeout=5
            )
            return result.stdout, result.stderr, result.returncode
        except Exception as e:
            logger.error(f"Tmux error: {e}")
            return "", str(e), 1
    
    async def start(self):
        """Start OpenCode in tmux session"""
        if self.is_running:
            await self.stop()
        
        identity = self.generate_identity()
        self.session_id = identity["session_id"][:8]
        
        # Cleanup
        auth_paths = [
            os.path.expanduser("~/.local/share/opencode"),
            os.path.expanduser("~/.config/opencode"),
        ]
        for path in auth_paths:
            if os.path.exists(path):
                shutil.rmtree(path, ignore_errors=True)
        
        # Kill existing tmux session
        subprocess.run(['tmux', 'kill-session', '-t', self.tmux_session], capture_output=True)
        
        # Create config
        session_dir = f"/tmp/opencode_micro_{self.session_id}"
        os.makedirs(session_dir, exist_ok=True)
        
        config = {
            "$schema": "https://opencode.ai/config.json",
            "theme": "opencode",
            "provider": {
                "opencode-zen": {
                    "npm": "@ai-sdk/openai-compatible",
                    "options": {"baseURL": "https://opencode.ai/zen/v1"},
                    "models": {
                        "kimi-k2.5-free": {"name": "Kimi K2.5 Free"},
                        "minimax-m2.5-free": {"name": "MiniMax M2.5 Free"},
                        "big-pickle": {"name": "Big Pickle"},
                        "glm-4.7": {"name": "GLM 4.7"}
                    }
                }
            },
            "model": "opencode-zen/kimi-k2.5-free",
            "autoshare": False,
            "autoupdate": False,
            "anonymous": True,
        }
        
        with open(f"{session_dir}/config.json", 'w') as f:
            json.dump(config, f, indent=2)
        
        try:
            # Create tmux session with OpenCode
            env_vars = f"export OPENCODE_CONFIG={session_dir}/config.json; export OPENCODE_NO_AUTH=1;"
            cmd = f'{env_vars} npx -y opencode-ai'
            
            subprocess.Popen([
                'tmux', 'new-session', '-d', '-s', self.tmux_session,
                '-c', session_dir,
                'bash', '-c', cmd
            ])
            
            self.is_running = True
            self.message_count = 0
            
            # Wait for startup
            await asyncio.sleep(5)
            
            # Select model
            self.run_tmux_cmd(['send-keys', 'C-x'])
            await asyncio.sleep(0.5)
            self.run_tmux_cmd(['send-keys', 'm'])
            await asyncio.sleep(1)
            self.run_tmux_cmd(['send-keys', 'Enter'])
            await asyncio.sleep(2)
            
            logger.info(f"✅ OpenCode started in tmux - Session: {self.session_id}")
            return True
            
        except Exception as e:
            logger.error(f"Failed to start OpenCode: {e}")
            return False
    
    def capture_screen(self):
        """Capture tmux pane content"""
        stdout, _, code = self.run_tmux_cmd(['capture-pane', '-p', '-S', '-100'])
        if code == 0:
            return stdout
        return ""
    
    async def chat(self, message: str) -> str:
        """Send message and capture response"""
        if not self.is_running:
            success = await self.start()
            if not success:
                return "Failed to start OpenCode"
        
        try:
            # Get current screen before message
            before_screen = self.capture_screen()
            before_lines = set(before_screen.split('\n'))
            
            # Send message
            self.run_tmux_cmd(['send-keys', message])
            self.run_tmux_cmd(['send-keys', 'Enter'])
            
            # Wait for AI response
            await asyncio.sleep(15)
            
            # Get screen after response
            after_screen = self.capture_screen()
            after_lines = after_screen.split('\n')
            
            # Find new lines (the response)
            response_lines = []
            for line in after_lines:
                line = line.strip()
                # Skip empty lines, prompts, and old lines
                if line and line not in before_lines and not line.startswith(('>', '$', '┌', '│', '└', '╭', '╰')):
                    # Skip the user's question
                    if message not in line:
                        response_lines.append(line)
            
            self.message_count += 1
            
            # Auto-reset
            if self.message_count >= self.max_messages:
                logger.info("🔄 Auto-reset triggered")
                await self.stop()
                await self.start()
            
            if response_lines:
                # Return last few lines (most recent response)
                return '\n'.join(response_lines[-15:])
            else:
                # Return full screen if we can't parse
                return after_screen[-2000:]  # Last 2000 chars
                
        except Exception as e:
            logger.error(f"Chat error: {e}")
            return f"Error: {str(e)}"
    
    async def stop(self):
        """Stop tmux session"""
        subprocess.run(['tmux', 'kill-session', '-t', self.tmux_session], capture_output=True)
        self.is_running = False
        self.message_count = 0
        logger.info("🛑 OpenCode stopped")

# Global session
session = OpenCodeSession()

class ChatRequest(BaseModel):
    message: str
    model: str = "kimi-k2.5-free"

@app.get("/")
def root():
    return {"status": "OpenCode Microservice Running"}

@app.get("/health")
def health():
    return {"status": "healthy", "session_active": session.is_running}

@app.post("/chat")
async def chat_endpoint(req: ChatRequest):
    try:
        response = await session.chat(req.message)
        return {
            "response": response,
            "session_id": session.session_id or "",
            "message_count": session.message_count,
            "max_messages": session.max_messages
        }
    except Exception as e:
        logger.error(f"Chat error: {e}")
        raise HTTPException(status_code=500, detail=str(e))

@app.get("/status")
async def status_endpoint():
    return {
        "is_running": session.is_running,
        "message_count": session.message_count,
        "max_messages": session.max_messages,
        "session_id": session.session_id
    }

@app.post("/reset")
async def reset_endpoint():
    await session.stop()
    success = await session.start()
    return {"status": "reset_complete" if success else "reset_failed"}

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)