"""
Emergency cleanup utilities for continuity plugin.
"""
import os
import signal
import subprocess
import time
from pathlib import Path
from typing import List
[docs]
def get_continuity_processes() -> List[int]:
"""Get list of continuity process PIDs.
Returns:
List of process IDs running continuity
"""
try:
# Use pgrep to find continuity processes
result = subprocess.run(
["pgrep", "-f", "continuity"], capture_output=True, text=True
)
if result.returncode == 0:
return [int(pid) for pid in result.stdout.strip().split("\n") if pid]
return []
except Exception:
return []
[docs]
def terminate_processes(pids: List[int], force: bool = False) -> bool:
"""Terminate continuity processes.
Args:
pids: List of process IDs to terminate
force: Use SIGKILL instead of SIGTERM
Returns:
True if all processes terminated successfully
"""
if not pids:
return True
signal_type = signal.SIGKILL if force else signal.SIGTERM
signal_name = "KILL" if force else "TERM"
for pid in pids:
try:
print(f"Sending {signal_name} to PID {pid}")
os.kill(pid, signal_type)
except ProcessLookupError:
# Process already gone
pass
except PermissionError:
print(f"Permission denied killing PID {pid}")
return False
except Exception as e:
print(f"Error killing PID {pid}: {e}")
return False
# Wait for processes to terminate
time.sleep(2)
# Check if any are still running
remaining = []
for pid in pids:
try:
os.kill(pid, 0) # Test if process exists
remaining.append(pid)
except ProcessLookupError:
# Process is gone
pass
return len(remaining) == 0
[docs]
def clean_lock_files() -> int:
"""Clean up lock files.
Returns:
Number of files removed
"""
continuity_dir = Path.home() / ".continuity"
if not continuity_dir.exists():
return 0
count = 0
for lock_file in continuity_dir.glob("*.lock"):
try:
lock_file.unlink()
print(f"Removed lock file: {lock_file}")
count += 1
except Exception as e:
print(f"Error removing {lock_file}: {e}")
return count
[docs]
def clean_socket_files() -> int:
"""Clean up socket files.
Returns:
Number of files removed
"""
continuity_dir = Path.home() / ".continuity"
if not continuity_dir.exists():
return 0
count = 0
for sock_file in continuity_dir.glob("*.sock"):
try:
sock_file.unlink()
print(f"Removed socket file: {sock_file}")
count += 1
except Exception as e:
print(f"Error removing {sock_file}: {e}")
return count
[docs]
def clean_temp_files() -> int:
"""Clean up temporary files.
Returns:
Number of files removed
"""
continuity_dir = Path.home() / ".continuity"
if not continuity_dir.exists():
return 0
count = 0
patterns = ["*.tmp", ".last_*", "*.pid"]
for pattern in patterns:
for temp_file in continuity_dir.glob(pattern):
try:
temp_file.unlink()
print(f"Removed temp file: {temp_file}")
count += 1
except Exception as e:
print(f"Error removing {temp_file}: {e}")
return count
[docs]
def emergency_cleanup(force: bool = False) -> bool:
"""Perform emergency cleanup of continuity resources.
Args:
force: Use aggressive cleanup (SIGKILL, etc.)
Returns:
True if cleanup was successful
"""
print("🧹 Emergency continuity cleanup starting...")
print("=" * 50)
success = True
# Find and terminate processes
print("🔍 Checking for continuity processes...")
pids = get_continuity_processes()
if not pids:
print("✅ No continuity processes found")
else:
print(f"📋 Found {len(pids)} continuity processes: {pids}")
# Try graceful termination first
if not force:
print("🛑 Attempting graceful termination...")
if not terminate_processes(pids, force=False):
print("⚠️ Graceful termination failed, trying force...")
force = True
if force:
remaining_pids = get_continuity_processes()
if remaining_pids:
print("🔨 Force terminating remaining processes...")
if not terminate_processes(remaining_pids, force=True):
print("❌ Some processes could not be terminated")
success = False
# Final check
final_pids = get_continuity_processes()
if not final_pids:
print("✅ All continuity processes terminated")
else:
print(
f"❌ Warning: {len(final_pids)} processes still running: {final_pids}"
)
success = False
# Clean up files
print("\n🗑️ Cleaning up lock and socket files...")
lock_count = clean_lock_files()
socket_count = clean_socket_files()
temp_count = clean_temp_files()
print(
f"✅ Cleaned up {lock_count} locks, {socket_count} sockets, {temp_count} temp files"
)
print("\n" + "=" * 50)
if success:
print("🎉 Cleanup completed successfully!")
else:
print("⚠️ Cleanup completed with warnings")
print("\n💡 To prevent future issues:")
print(" - Always use 'continuity server --daemon'")
print(" - Check 'continuity status' before starting servers")
print(" - Use this cleanup if issues persist")
return success
if __name__ == "__main__":
import sys
force = "--force" in sys.argv
emergency_cleanup(force=force)