fix: Use PTY for proper terminal handling

- Replace subprocess.Popen with pty.fork() for proper TTY handling
- Fixes terminal 'bugging out' when spawning claude
- Properly handles terminal control sequences and colors
- Uses os.fork() and pty.openpty() for interactive shell support
- Maintains proper terminal restoration on exit

This fixes the issue where claude-vision would corrupt the terminal
when trying to spawn the claude subprocess.

Generated with Claude Code

Co-Authored-By: Claude <noreply@anthropic.com>
Co-Authored-By: Jean-Philippe Brule <jp@svrnty.io>
This commit is contained in:
Svrnty 2025-10-30 02:14:33 -04:00
parent 52b0813b64
commit 29ac2f0929

View File

@ -3,8 +3,13 @@ Main entry point for Claude Vision Auto
""" """
import sys import sys
import os
import time import time
import select import select
import pty
import tty
import termios
import signal
import subprocess import subprocess
from pathlib import Path from pathlib import Path
@ -40,48 +45,92 @@ def run_claude_with_vision(args: list = None):
# Build command # Build command
cmd = ['claude'] + args cmd = ['claude'] + args
# Start Claude Code process # Check if claude exists
try: if not subprocess.run(['which', 'claude'], capture_output=True).returncode == 0:
process = subprocess.Popen(
cmd,
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
bufsize=0
)
except FileNotFoundError:
print("[ERROR] 'claude' command not found") print("[ERROR] 'claude' command not found")
print("Make sure Claude Code CLI is installed") print("Make sure Claude Code CLI is installed")
sys.exit(1) sys.exit(1)
last_output_time = time.time()
output_buffer = bytearray()
# Cleanup old screenshots # Cleanup old screenshots
cleanup_old_screenshots() cleanup_old_screenshots()
# Save original terminal settings
old_tty = termios.tcgetattr(sys.stdin)
try:
# Create pseudo-terminal
master_fd, slave_fd = pty.openpty()
# Fork process
pid = os.fork()
if pid == 0:
# Child process - run claude
os.close(master_fd)
# Set up slave as stdin/stdout/stderr
os.dup2(slave_fd, 0)
os.dup2(slave_fd, 1)
os.dup2(slave_fd, 2)
if slave_fd > 2:
os.close(slave_fd)
# Execute claude
os.execvp('claude', cmd)
else:
# Parent process - handle I/O and vision analysis
os.close(slave_fd)
# Set terminal to raw mode
tty.setraw(sys.stdin.fileno())
last_output_time = time.time()
output_buffer = bytearray()
try: try:
while True: while True:
# Check if there's data to read # Check for data from claude or user
readable, _, _ = select.select([process.stdout], [], [], 0.1) readable, _, _ = select.select(
[master_fd, sys.stdin.fileno()],
[],
[],
0.1
)
if readable: for fd in readable:
char = process.stdout.read(1) if fd == master_fd:
if not char: # Read from claude process
try:
data = os.read(master_fd, 1024)
if not data:
# Process ended # Process ended
break os.waitpid(pid, 0)
return
# Print to terminal # Write to stdout
sys.stdout.buffer.write(char) os.write(sys.stdout.fileno(), data)
sys.stdout.buffer.flush()
output_buffer.extend(char) # Add to buffer for pattern matching
output_buffer.extend(data)
last_output_time = time.time() last_output_time = time.time()
# Keep buffer reasonable size # Keep buffer reasonable size
if len(output_buffer) > config.OUTPUT_BUFFER_SIZE: if len(output_buffer) > config.OUTPUT_BUFFER_SIZE:
output_buffer = output_buffer[-config.OUTPUT_BUFFER_SIZE:] output_buffer = output_buffer[-config.OUTPUT_BUFFER_SIZE:]
except OSError:
# Process ended
os.waitpid(pid, 0)
return
elif fd == sys.stdin.fileno():
# Read from user input
data = os.read(sys.stdin.fileno(), 1024)
if data:
# Forward to claude
os.write(master_fd, data)
# Check if idle (no output for threshold seconds) # Check if idle (no output for threshold seconds)
idle_time = time.time() - last_output_time idle_time = time.time() - last_output_time
@ -97,9 +146,11 @@ def run_claude_with_vision(args: list = None):
if has_keywords: if has_keywords:
if config.DEBUG: if config.DEBUG:
print("\n[DEBUG] Approval keywords detected in buffer") sys.stderr.write("\n[DEBUG] Approval keywords detected in buffer\n")
sys.stderr.flush()
print("\n[Vision] Analyzing prompt...", file=sys.stderr) sys.stderr.write("\n[Vision] Analyzing prompt...\n")
sys.stderr.flush()
# Take screenshot # Take screenshot
screenshot_path = take_screenshot() screenshot_path = take_screenshot()
@ -109,23 +160,26 @@ def run_claude_with_vision(args: list = None):
response = analyzer.analyze_screenshot(screenshot_path) response = analyzer.analyze_screenshot(screenshot_path)
if response: if response:
print(f"[Vision] Response: {response}", file=sys.stderr) sys.stderr.write(f"[Vision] Response: {response}\n")
sys.stderr.flush()
if response and response.upper() != "WAIT": if response and response.upper() != "WAIT":
# Send response # Send response
time.sleep(config.RESPONSE_DELAY) time.sleep(config.RESPONSE_DELAY)
process.stdin.write(f"{response}\n".encode('utf-8')) os.write(master_fd, f"{response}\n".encode('utf-8'))
process.stdin.flush()
# Clear buffer # Clear buffer
output_buffer.clear() output_buffer.clear()
last_output_time = time.time() last_output_time = time.time()
print("[Vision] Response sent", file=sys.stderr) sys.stderr.write("[Vision] Response sent\n")
sys.stderr.flush()
else: else:
print("[Vision] No action needed (WAIT)", file=sys.stderr) sys.stderr.write("[Vision] No action needed (WAIT)\n")
sys.stderr.flush()
else: else:
print("[Vision] Analysis failed, waiting for manual input", file=sys.stderr) sys.stderr.write("[Vision] Analysis failed, waiting for manual input\n")
sys.stderr.flush()
# Clean up screenshot # Clean up screenshot
try: try:
@ -133,25 +187,26 @@ def run_claude_with_vision(args: list = None):
except Exception: except Exception:
pass pass
else: else:
print("[Vision] Screenshot failed, waiting for manual input", file=sys.stderr) sys.stderr.write("[Vision] Screenshot failed, waiting for manual input\n")
sys.stderr.flush()
# Reset idle detection # Reset idle detection
last_output_time = time.time() last_output_time = time.time()
# Check if process is still running
if process.poll() is not None:
break
except KeyboardInterrupt: except KeyboardInterrupt:
print("\n[Claude Vision Auto] Interrupted by user") # Kill child process
process.terminate() os.kill(pid, signal.SIGTERM)
process.wait() os.waitpid(pid, 0)
sys.exit(130)
finally: finally:
# Wait for process to finish # Restore terminal settings
exit_code = process.wait() termios.tcsetattr(sys.stdin, termios.TCSADRAIN, old_tty)
sys.exit(exit_code)
# Close master fd if still open
try:
os.close(master_fd)
except:
pass
def main(): def main():