Based on claude-code-tools TmuxCLIController, this refactor: - Added DockerTmuxController class for robust tmux session management - Implements send_keys() with configurable delay_enter - Implements capture_pane() for output retrieval - Implements wait_for_prompt() for pattern-based completion detection - Implements wait_for_idle() for content-hash-based idle detection - Implements wait_for_shell_prompt() for shell prompt detection Also includes workflow improvements: - Pre-task git snapshot before agent execution - Post-task commit protocol in agent guidelines Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
166 lines
5.3 KiB
Python
166 lines
5.3 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Status command handler for Luzia
|
|
Handles: luzia status [options]
|
|
|
|
Commands:
|
|
luzia status - Show dashboard
|
|
luzia status <task-id> - Show specific task
|
|
luzia status --project <name> - Show project summary
|
|
luzia status --alerts - Show only alerts
|
|
luzia status --recent N - Show last N updates
|
|
luzia status --export json - Export to JSON
|
|
luzia status --export markdown - Export to markdown
|
|
"""
|
|
|
|
import asyncio
|
|
import sys
|
|
from typing import Optional, List
|
|
import logging
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class LuziaStatusHandler:
|
|
"""Handles all luzia status subcommands"""
|
|
|
|
def __init__(self):
|
|
self.bridge = None
|
|
self._initialize()
|
|
|
|
def _initialize(self):
|
|
"""Initialize the status handler with bridge"""
|
|
try:
|
|
from luzia_status_integration import get_status_system
|
|
status_system = get_status_system()
|
|
if status_system.is_enabled() and status_system.bridge:
|
|
self.bridge = status_system.bridge
|
|
logger.info("Status handler initialized")
|
|
else:
|
|
logger.warning("Status system not fully initialized")
|
|
except Exception as e:
|
|
logger.error(f"Failed to initialize status handler: {e}")
|
|
|
|
def is_available(self) -> bool:
|
|
"""Check if status handler is available"""
|
|
return self.bridge is not None
|
|
|
|
def handle_command(self, args: List[str]) -> str:
|
|
"""
|
|
Handle status command
|
|
|
|
Args:
|
|
args: Command arguments after 'status'
|
|
|
|
Returns:
|
|
Command output as string
|
|
"""
|
|
if not self.is_available():
|
|
return "Status system not available"
|
|
|
|
if not args:
|
|
return self.bridge.get_dashboard()
|
|
|
|
# Check for flags
|
|
if args[0].startswith('--'):
|
|
flag = args[0]
|
|
|
|
if flag == '--alerts':
|
|
return self.bridge.get_alerts_only()
|
|
|
|
elif flag == '--recent':
|
|
limit = int(args[1]) if len(args) > 1 else 10
|
|
return self.bridge.get_recent_updates(limit)
|
|
|
|
elif flag == '--project':
|
|
project = args[1] if len(args) > 1 else None
|
|
if not project:
|
|
return "Usage: luzia status --project <project-name>"
|
|
return self.bridge.get_project_summary(project)
|
|
|
|
elif flag == '--export':
|
|
format_type = args[1] if len(args) > 1 else 'json'
|
|
from datetime import datetime
|
|
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
|
|
|
|
if format_type == 'json':
|
|
filepath = f"/tmp/luzia_status_{timestamp}.json"
|
|
try:
|
|
self.bridge.export_to_json(filepath)
|
|
return f"Exported to {filepath}"
|
|
except Exception as e:
|
|
return f"Export failed: {e}"
|
|
|
|
elif format_type == 'markdown':
|
|
filepath = f"/tmp/luzia_status_{timestamp}.md"
|
|
try:
|
|
self.bridge.export_to_markdown(filepath)
|
|
return f"Exported to {filepath}"
|
|
except Exception as e:
|
|
return f"Export failed: {e}"
|
|
|
|
elif flag == '--verbose':
|
|
from luzia_status_integration import get_status_system
|
|
status_system = get_status_system()
|
|
if status_system.publisher:
|
|
status_system.publisher.set_verbosity("verbose")
|
|
return "Verbosity set to verbose"
|
|
return "Cannot set verbosity"
|
|
|
|
elif flag == '--quiet':
|
|
from luzia_status_integration import get_status_system
|
|
status_system = get_status_system()
|
|
if status_system.publisher:
|
|
status_system.publisher.set_verbosity("quiet")
|
|
return "Verbosity set to quiet"
|
|
return "Cannot set verbosity"
|
|
|
|
elif flag == '--normal':
|
|
from luzia_status_integration import get_status_system
|
|
status_system = get_status_system()
|
|
if status_system.publisher:
|
|
status_system.publisher.set_verbosity("normal")
|
|
return "Verbosity set to normal"
|
|
return "Cannot set verbosity"
|
|
|
|
# Treat as task ID
|
|
task_id = args[0]
|
|
result = self.bridge.get_task_summary(task_id)
|
|
if result:
|
|
return result
|
|
return f"Task not found: {task_id}"
|
|
|
|
|
|
# Global handler instance
|
|
_handler = None
|
|
|
|
|
|
def get_status_handler() -> LuziaStatusHandler:
|
|
"""Get or create the global status handler"""
|
|
global _handler
|
|
if _handler is None:
|
|
_handler = LuziaStatusHandler()
|
|
return _handler
|
|
|
|
|
|
def handle_status_command(args: List[str]) -> int:
|
|
"""
|
|
Entry point for status command handling
|
|
This should be called from route_status() in the main luzia binary
|
|
|
|
Args:
|
|
args: Arguments after 'status'
|
|
|
|
Returns:
|
|
Exit code (0 for success)
|
|
"""
|
|
handler = get_status_handler()
|
|
|
|
if not handler.is_available():
|
|
print("Status system is not available")
|
|
return 1
|
|
|
|
result = handler.handle_command(args)
|
|
print(result)
|
|
return 0
|