diff --git a/.claude/settings.local.json b/.claude/settings.local.json new file mode 100644 index 000000000..d6965e74c --- /dev/null +++ b/.claude/settings.local.json @@ -0,0 +1,11 @@ +{ + "permissions": { + "allow": [ + "mcp__codegen-tools__linear_get_teams", + "mcp__codegen-tools__linear_search_issues", + "mcp__codegen-tools__linear_create_issue" + ], + "deny": [], + "ask": [] + } +} diff --git a/src/codegen/cli/commands/claude/main.py b/src/codegen/cli/commands/claude/main.py index 0d32d2a46..a8f3a33c5 100644 --- a/src/codegen/cli/commands/claude/main.py +++ b/src/codegen/cli/commands/claude/main.py @@ -4,15 +4,13 @@ import signal import subprocess import sys -import threading -import time import requests import typer from rich import box +from rich.console import Console from rich.panel import Panel - from codegen.cli.api.endpoints import API_ENDPOINT from codegen.cli.auth.token_manager import get_current_token from codegen.cli.commands.claude.claude_log_watcher import ClaudeLogWatcherManager @@ -20,13 +18,11 @@ from codegen.cli.commands.claude.config.mcp_setup import add_codegen_mcp_server, cleanup_codegen_mcp_server from codegen.cli.commands.claude.hooks import cleanup_claude_hook, ensure_claude_hook, get_codegen_url from codegen.cli.commands.claude.quiet_console import console -from rich.console import Console - -t_console = Console() - from codegen.cli.rich.spinners import create_spinner from codegen.cli.utils.org import resolve_org_id +t_console = Console() + def _run_claude_background(resolved_org_id: int, prompt: str | None) -> None: """Create a background agent run with Claude context and exit.""" @@ -208,9 +204,9 @@ def claude( raise typer.Exit(1) if background is not None: - # Use the value from --background as the prompt, with --prompt as fallback - final_prompt = background or prompt + # Use the value from --background as the prompt + final_prompt = background _run_claude_background(resolved_org_id, final_prompt) return - _run_claude_interactive(resolved_org_id, no_mcp) \ No newline at end of file + _run_claude_interactive(resolved_org_id, no_mcp) diff --git a/src/codegen/cli/tui/app.py b/src/codegen/cli/tui/app.py index 8f9d4a000..2898f4481 100644 --- a/src/codegen/cli/tui/app.py +++ b/src/codegen/cli/tui/app.py @@ -1,77 +1,77 @@ -"""Main TUI application for Codegen CLI.""" +"""Minimal TUI interface for Codegen CLI.""" -import asyncio -import webbrowser +import signal +import sys +import termios +import tty +from datetime import datetime +from typing import Any -from textual.app import App, ComposeResult -from textual.binding import Binding -from textual.containers import Container, Vertical -from textual.widgets import DataTable, Footer, Header, Static +import requests +import typer -from codegen.cli.auth.token_manager import get_current_token, get_cached_organizations, get_current_org_name, get_org_name_from_cache +from codegen.cli.api.endpoints import API_ENDPOINT +from codegen.cli.auth.token_manager import get_current_token +from codegen.cli.commands.agent.main import pull from codegen.cli.utils.org import resolve_org_id -class CodegenTUI(App): - """Simple Codegen TUI for browsing agent runs.""" - - CSS_PATH = "codegen_theme.tcss" - TITLE = "Codegen CLI" - BINDINGS = [ - Binding("escape,ctrl+c", "quit", "Quit", priority=True), - Binding("enter", "open_url", "Details", show=True), - Binding("r", "refresh", "Refresh", show=True), - Binding("o", "select_org", "Org", show=True), - Binding("p", "select_repo", "Repo", show=True), - ] +class MinimalTUI: + """Minimal non-full-screen TUI for browsing agent runs.""" def __init__(self): - super().__init__() self.token = get_current_token() self.is_authenticated = bool(self.token) if self.is_authenticated: self.org_id = resolve_org_id() - self.agent_runs = [] - - def compose(self) -> ComposeResult: - """Create child widgets for the app.""" - yield Header() - if not self.is_authenticated: - yield Container(Static("⚠️ Not authenticated. Please run 'codegen login' first.", classes="warning-message"), id="auth-warning") - else: - with Vertical(): - # Show current organization info - using id for updating - org_name = get_current_org_name() - org_display = f" ({org_name})" if org_name else f" (ID: {self.org_id})" if self.org_id else "" - yield Static(f"🤖 Your Recent API Agent Runs{org_display}", classes="title", id="title-text") - yield Static("Use ↑↓ to navigate, Enter for details, R to refresh, O for org, P for repo, Esc to quit", classes="help") - table = DataTable(id="agents-table", cursor_type="row") - table.add_columns("Created", "Status", "Summary") - yield table - yield Footer() - - def on_mount(self) -> None: - """Called when app starts.""" - if self.is_authenticated and self.org_id: - task = asyncio.create_task(self._load_agents_data()) - # Store reference to prevent garbage collection - self._load_task = task - - # Ensure the table has focus for key events - try: - table = self.query_one("#agents-table", DataTable) - table.focus() - except Exception: - # Table might not be ready yet, will focus after data loads - pass - - async def _load_agents_data(self) -> None: - """Load agents data into the table.""" - table = self.query_one("#agents-table", DataTable) - table.clear() - + self.agent_runs: list[dict[str, Any]] = [] + self.selected_index = 0 + self.running = True + self.show_action_menu = False + self.action_menu_selection = 0 + + # Tab management + self.tabs = ["recents", "new", "web"] + self.current_tab = 0 + + # New tab state + self.prompt_input = "" + self.cursor_position = 0 + self.input_mode = False # When true, we're typing in the input box + + # Set up signal handler for Ctrl+C + signal.signal(signal.SIGINT, self._signal_handler) + + def _get_webapp_domain(self) -> str: + """Get the webapp domain based on environment.""" + # Simple environment detection - can be expanded later + import os + + env = os.getenv("ENV", "staging").lower() + + if env == "production": + return "codegen.com" + elif env == "local": + return "localhost:3000" + else: # staging or default + return "chadcode.sh" + + def _generate_agent_url(self, agent_id: str) -> str: + """Generate the complete agent URL.""" + domain = self._get_webapp_domain() + protocol = "http" if "localhost" in domain else "https" + return f"{protocol}://{domain}/x/{agent_id}" + + def _signal_handler(self, signum, frame): + """Handle Ctrl+C gracefully without clearing screen.""" + self.running = False + print("\n") # Just add a newline and exit + sys.exit(0) + + def _load_agent_runs(self) -> bool: + """Load the last 10 agent runs.""" if not self.token or not self.org_id: - return + return False try: import requests @@ -80,229 +80,538 @@ async def _load_agents_data(self) -> None: headers = {"Authorization": f"Bearer {self.token}"} - # First get the current user ID + # Get current user ID user_response = requests.get(f"{API_ENDPOINT.rstrip('/')}/v1/users/me", headers=headers) user_response.raise_for_status() user_data = user_response.json() user_id = user_data.get("id") - # Filter to only API source type and current user's agent runs + # Fetch agent runs - limit to 10 params = { "source_type": "API", - "limit": 20, # Show recent 20 + "limit": 10, } if user_id: params["user_id"] = user_id - # Fetch agent runs url = f"{API_ENDPOINT.rstrip('/')}/v1/organizations/{self.org_id}/agent/runs" response = requests.get(url, headers=headers, params=params) response.raise_for_status() response_data = response.json() - agent_runs = response_data.get("items", []) - self.agent_runs = agent_runs # Store for URL opening - - for agent_run in agent_runs: - run_id = str(agent_run.get("id", "Unknown")) - status = agent_run.get("status", "Unknown") - created_at = agent_run.get("created_at", "Unknown") - - # Use summary from API response (backend now handles extraction) - summary = agent_run.get("summary", "") or "No summary" - - # Status with colored circles - if status == "COMPLETE": - status_display = "● Complete" - elif status == "ACTIVE": - status_display = "● Active" - elif status == "RUNNING": - status_display = "● Running" - elif status == "CANCELLED": - status_display = "● Cancelled" - elif status == "ERROR": - status_display = "● Error" - elif status == "FAILED": - status_display = "● Failed" - elif status == "STOPPED": - status_display = "● Stopped" - elif status == "PENDING": - status_display = "● Pending" - else: - status_display = "● " + status - - # Format created date - if created_at and created_at != "Unknown": - try: - from datetime import datetime - - dt = datetime.fromisoformat(created_at.replace("Z", "+00:00")) - created_display = dt.strftime("%m/%d %H:%M") - except (ValueError, TypeError): - created_display = created_at[:16] if len(created_at) > 16 else created_at - else: - created_display = created_at - - # Truncate summary if too long - summary_display = summary[:60] + "..." if summary and len(summary) > 60 else summary or "No summary" - - table.add_row(created_display, status_display, summary_display, key=run_id) + self.agent_runs = response_data.get("items", []) + return True except Exception as e: - # If API call fails, show error in table - table.add_row("Error", f"Failed to load: {e}", "") - - # Ensure table has focus after data is loaded - table.focus() - - def on_data_table_row_selected(self, event: DataTable.RowSelected) -> None: - """Handle DataTable row selection (Enter key).""" - if event.data_table.id == "agents-table": - self.notify("🔍 Enter key pressed - opening agent details...", timeout=1) - self.action_open_url() - - def action_open_url(self) -> None: - """Open the selected agent run detail screen.""" - table = self.query_one("#agents-table", DataTable) - if table.cursor_row is not None and table.cursor_row < len(self.agent_runs): - agent_run = self.agent_runs[table.cursor_row] - run_id = agent_run.get("id", "Unknown") - - self.notify(f"📱 Opening details for agent run {run_id}...", timeout=2) - - # Import here to avoid circular imports - from codegen.cli.tui.agent_detail import AgentDetailTUI - - # Create and push the agent detail screen - detail_screen = AgentDetailTUI(agent_run, self.org_id) - self.push_screen(detail_screen) - elif table.cursor_row is None: - self.notify("❌ No row selected", severity="error", timeout=2) - elif len(self.agent_runs) == 0: - self.notify("❌ No agent runs available", severity="error", timeout=2) - else: - self.notify(f"❌ Invalid row selection: {table.cursor_row}/{len(self.agent_runs)}", severity="error", timeout=2) + print(f"Error loading agent runs: {e}") + return False + + def _format_status(self, status: str, agent_run: dict | None = None) -> str: + """Format status with colored indicators matching kanban style.""" + # Check if this agent has a merged PR (done status) + is_done = False + if agent_run: + github_prs = agent_run.get("github_pull_requests", []) + for pr in github_prs: + if pr.get("state") == "closed" and pr.get("merged", False): + is_done = True + break + + if is_done: + return "\033[34m✓\033[0m Done" # Blue checkmark for merged PR + + status_map = { + "COMPLETE": "\033[38;2;52;211;153m○\033[0m Complete", # emerald-400 + "ACTIVE": "\033[38;2;162;119;255m●\033[0m Active", # #a277ff (purple from badge) + "RUNNING": "\033[38;2;162;119;255m●\033[0m Running", # #a277ff (purple from badge) + "ERROR": "\033[38;2;248;113;113m●\033[0m Error", # red-400 + "FAILED": "\033[38;2;248;113;113m●\033[0m Failed", # red-400 + "CANCELLED": "\033[38;2;156;163;175m○\033[0m Cancelled", # gray-400 + "STOPPED": "\033[38;2;156;163;175m○\033[0m Stopped", # gray-400 + "PENDING": "\033[38;2;156;163;175m○\033[0m Pending", # gray-400 + "TIMEOUT": "\033[38;2;251;146;60m●\033[0m Timeout", # orange-400 + "MAX_ITERATIONS_REACHED": "\033[38;2;251;191;36m●\033[0m Max Iterations", # amber-400 + "OUT_OF_TOKENS": "\033[38;2;251;191;36m●\033[0m Out of Tokens", # amber-400 + "EVALUATION": "\033[38;2;196;181;253m●\033[0m Evaluation", # purple-400 + } + return status_map.get(status, f"\033[37m○\033[0m {status}") + + def _strip_ansi_codes(self, text: str) -> str: + """Strip ANSI color codes from text.""" + import re + + ansi_escape = re.compile(r"\x1B(?:[@-Z\\-_]|\[[0-?]*[ -/]*[@-~])") + return ansi_escape.sub("", text) + + def _format_date(self, created_at: str) -> str: + """Format creation date.""" + if not created_at or created_at == "Unknown": + return "Unknown" - def action_refresh(self) -> None: - """Refresh agent runs data.""" - if self.is_authenticated: - # Refresh org ID and title - old_org_id = self.org_id - self.org_id = resolve_org_id() - self._refresh_title() - - if self.org_id: - self.notify("🔄 Refreshing...", timeout=1) - task = asyncio.create_task(self._load_agents_data()) - # Store reference to prevent garbage collection - self._refresh_task = task + try: + dt = datetime.fromisoformat(created_at.replace("Z", "+00:00")) + return dt.strftime("%m/%d %H:%M") + except (ValueError, TypeError): + return created_at[:16] if len(created_at) > 16 else created_at + + def _display_header(self): + """Display the header with tabs.""" + # Simple header with indigo slashes and Codegen text + print("\033[38;2;82;19;217m" + "/" * 20 + " Codegen\033[0m") + print() # Add blank line between header and tabs + + # Display tabs + tab_line = "" + for i, tab in enumerate(self.tabs): + if i == self.current_tab: + tab_line += f"\033[34m[{tab}]\033[0m " # Blue for active tab + else: + tab_line += f"\033[90m{tab}\033[0m " # Gray for inactive tabs + + print(tab_line) + print() + + def _display_agent_list(self): + """Display the list of agent runs.""" + if not self.agent_runs: + print("No agent runs found.") + return + + for i, agent_run in enumerate(self.agent_runs): + # Highlight selected item + prefix = "→ " if i == self.selected_index and not self.show_action_menu else " " + + status = self._format_status(agent_run.get("status", "Unknown"), agent_run) + created = self._format_date(agent_run.get("created_at", "Unknown")) + summary = agent_run.get("summary", "No summary") or "No summary" + + # No need to truncate summary as much since we removed the URL column + if len(summary) > 60: + summary = summary[:57] + "..." + + # Color coding: indigo blue for selected, darker gray for others (but keep status colors) + if i == self.selected_index and not self.show_action_menu: + # Blue timestamp and summary for selected row, but preserve status colors + line = f"\033[34m{prefix}{created:<10}\033[0m {status} \033[34m{summary}\033[0m" else: - self.notify("❌ No organization configured", severity="error") + # Gray text for non-selected rows, but preserve status colors + line = f"\033[90m{prefix}{created:<10}\033[0m {status} \033[90m{summary}\033[0m" + + print(line) + + # Show action menu right below the selected row if it's expanded + if i == self.selected_index and self.show_action_menu: + self._display_inline_action_menu(agent_run) + + def _display_new_tab(self): + """Display the new agent creation interface.""" + print("Create a new agent run:") + print() + + # Input prompt label + print("Prompt:") + + # Get terminal width, default to 80 if can't determine + try: + import os + + terminal_width = os.get_terminal_size().columns + except (OSError, AttributeError): + terminal_width = 80 + + # Calculate input box width (leave some margin) + box_width = max(60, terminal_width - 4) + + # Input box with cursor + input_display = self.prompt_input + if self.input_mode: + # Add cursor indicator when in input mode + if self.cursor_position <= len(input_display): + input_display = input_display[: self.cursor_position] + "█" + input_display[self.cursor_position :] + + # Handle long input that exceeds box width + if len(input_display) > box_width - 4: + # Show portion around cursor + start_pos = max(0, self.cursor_position - (box_width // 2)) + input_display = input_display[start_pos : start_pos + box_width - 4] + + # Display full-width input box with simple border like Claude Code + border_style = "\033[34m" if self.input_mode else "\033[90m" # Blue when active, gray when inactive + reset = "\033[0m" + + print(border_style + "┌" + "─" * (box_width - 2) + "┐" + reset) + padding = box_width - 4 - len(input_display.replace("█", "")) + print(border_style + "│" + reset + f" {input_display}{' ' * max(0, padding)} " + border_style + "│" + reset) + print(border_style + "└" + "─" * (box_width - 2) + "┘" + reset) + print() + + if self.input_mode: + print("\033[90mType your prompt • [Enter] create agent • [Esc] cancel\033[0m") else: - self.notify("❌ Not authenticated", severity="error") + print("\033[90m[Enter] start typing • [Tab] switch tabs • [Q] quit\033[0m") - def action_select_org(self) -> None: - """Launch the organization selector TUI.""" - if not self.is_authenticated: - self.notify("❌ Not authenticated. Please login first.", severity="error") + def _create_background_agent(self, prompt: str): + """Create a background agent run.""" + if not self.token or not self.org_id: + print("\n❌ Not authenticated or no organization configured.") + input("Press Enter to continue...") return - # Check if organizations are available in cache - cached_orgs = get_cached_organizations() - if not cached_orgs: - self.notify("❌ No organizations found. Please run 'codegen login' to refresh.", severity="error") + if not prompt.strip(): + print("\n❌ Please enter a prompt.") + input("Press Enter to continue...") return + print(f"\n🔄 Creating agent run with prompt: '{prompt[:50]}{'...' if len(prompt) > 50 else ''}'") + try: - # Import here to avoid circular imports - from codegen.cli.commands.org.tui import OrgSelectorTUI - - # Launch the organization selector as a sub-screen - def on_org_selected(): - # Debug callback trigger - self.notify("🔧 Org selector callback triggered!", timeout=0.5) - - # Refresh the org_id and reload data when org selector closes - old_org_id = self.org_id - self.org_id = resolve_org_id() - - # Debug org ID change - self.notify(f"🔧 Org ID: {old_org_id} → {self.org_id}", timeout=1) - - # Refresh the title to show new organization - self._refresh_title() - - if self.org_id and self.org_id != old_org_id: - self.notify("🔄 Refreshing data for new organization...", timeout=1) - task = asyncio.create_task(self._load_agents_data()) - self._refresh_task = task - elif not self.org_id: - self.notify("❌ No organization configured", severity="error") - else: - self.notify("ℹ Organization unchanged", timeout=1) - - org_selector = OrgSelectorTUI() - # Set up a callback when the screen is dismissed - self.push_screen(org_selector, callback=lambda _: on_org_selected()) + payload = {"prompt": prompt.strip()} + headers = { + "Authorization": f"Bearer {self.token}", + "Content-Type": "application/json", + "x-codegen-client": "codegen__claude_code", + } + url = f"{API_ENDPOINT.rstrip('/')}/v1/organizations/{self.org_id}/agent/run" + response = requests.post(url, headers=headers, json=payload, timeout=30) + response.raise_for_status() + agent_run_data = response.json() + + run_id = agent_run_data.get("id", "Unknown") + status = agent_run_data.get("status", "Unknown") + web_url = self._generate_agent_url(run_id) + + print("\n✅ Agent run created successfully!") + print(f" Run ID: {run_id}") + print(f" Status: {status}") + print(f" Web URL: {web_url}") + + # Clear the input + self.prompt_input = "" + self.cursor_position = 0 + self.input_mode = False + + # Optionally refresh the recents tab if we're going back to it + if hasattr(self, "_load_agent_runs"): + print("\n🔄 Refreshing recents...") + self._load_agent_runs() + except Exception as e: - self.notify(f"❌ Failed to launch org selector: {e}", severity="error") + print(f"\n❌ Failed to create agent run: {e}") - def action_select_repo(self) -> None: - """Launch the repository selector TUI.""" - if not self.is_authenticated: - self.notify("❌ Not authenticated. Please login first.", severity="error") - return + input("\nPress Enter to continue...") + + def _display_web_tab(self): + """Display the web interface access tab.""" + print("Open Web Interface:") + print() + print(" \033[34m→ Open Web (localhost:3000/me)\033[0m") + print() + print("Press Enter to open the web interface in your browser.") + + def _pull_agent_branch(self, agent_id: str): + """Pull the PR branch for an agent run locally.""" + print(f"\n🔄 Pulling PR branch for agent {agent_id}...") + print("─" * 50) try: - # Import here to avoid circular imports - from codegen.cli.commands.repo.tui import RepoSelectorTUI - - # Launch the repository selector as a sub-screen - def on_repo_selected(): - # Debug callback trigger - self.notify("🔧 Repo selector callback triggered!", timeout=0.5) - - # Notify user about repo change (repos don't affect agent runs directly in this TUI) - self.notify("✅ Repository configuration updated!", timeout=2) - - repo_selector = RepoSelectorTUI() - # Set up a callback when the screen is dismissed - self.push_screen(repo_selector, callback=lambda _: on_repo_selected()) + # Call the existing pull command with the agent_id + pull(agent_id=int(agent_id), org_id=self.org_id) + + except typer.Exit as e: + # typer.Exit is expected for both success and failure cases + if e.exit_code == 0: + print("\n✅ Pull completed successfully!") + else: + print(f"\n❌ Pull failed (exit code: {e.exit_code})") + except ValueError: + print(f"\n❌ Invalid agent ID: {agent_id}") except Exception as e: - self.notify(f"❌ Failed to launch repo selector: {e}", severity="error") + print(f"\n❌ Unexpected error during pull: {e}") + + print("─" * 50) + input("Press Enter to continue...") + + def _display_content(self): + """Display content based on current tab.""" + if self.current_tab == 0: # recents + self._display_agent_list() + elif self.current_tab == 1: # new + self._display_new_tab() + elif self.current_tab == 2: # web + self._display_web_tab() + + def _display_inline_action_menu(self, agent_run: dict): + """Display action menu inline below the selected row.""" + agent_id = agent_run.get("id", "unknown") + web_url = self._generate_agent_url(agent_id) + # Extract just the domain/path part without protocol for display + display_url = web_url.replace("https://", "").replace("http://", "") + + # Check if there are GitHub PRs associated with this agent run + github_prs = agent_run.get("github_pull_requests", []) + + # Start with basic web option + options = [f"open in web ({display_url})"] + + # Only add pull locally if there are PRs + if github_prs: + options.insert(0, "pull locally") # Add as first option + + # Add PR option if available + if github_prs: + pr_url = github_prs[0].get("url", "") + if pr_url: + # Extract just the GitHub part for display + pr_display = pr_url.replace("https://github.com/", "github.com/") + options.append(f"open PR ({pr_display})") + + for i, option in enumerate(options): + if i == 0: + # Always highlight first (top) option in blue + print(f" \033[34m→ {option}\033[0m") + else: + # All other options in gray + print(f" \033[90m {option}\033[0m") + + print("\033[90m [Enter] select, [C] close\033[0m") - def _refresh_title(self) -> None: - """Refresh the title to show current organization.""" + def _get_char(self): + """Get a single character from stdin, handling arrow keys.""" try: - if self.is_authenticated: - title_widget = self.query_one("#title-text", Static) - - # Try to get org name from cache first (more reliable after org change) - org_name = None - if self.org_id: - org_name = get_org_name_from_cache(self.org_id) - - # Fallback to stored org name - if not org_name: - org_name = get_current_org_name() - - org_display = f" ({org_name})" if org_name else f" (ID: {self.org_id})" if self.org_id else "" - new_title = f"🤖 Your Recent API Agent Runs{org_display}" - title_widget.update(new_title) - - # Debug notification - self.notify(f"🔧 Title updated: {new_title}", timeout=0.5) - except Exception as e: - # Debug any errors - self.notify(f"❌ Error updating title: {e}", severity="error", timeout=1) + fd = sys.stdin.fileno() + old_settings = termios.tcgetattr(fd) + try: + tty.setcbreak(fd) + ch = sys.stdin.read(1) + + # Handle escape sequences (arrow keys) + if ch == "\x1b": # ESC + ch2 = sys.stdin.read(1) + if ch2 == "[": + ch3 = sys.stdin.read(1) + return f"\x1b[{ch3}" + else: + return ch + ch2 + return ch + finally: + termios.tcsetattr(fd, termios.TCSADRAIN, old_settings) + except (ImportError, OSError, termios.error): + # Fallback for systems where tty manipulation doesn't work + print("\nUse: ↑(w)/↓(s) navigate, Enter details, R refresh, Q quit") + try: + return input("> ").strip()[:1].lower() or "\n" + except KeyboardInterrupt: + return "q" + + def _handle_keypress(self, key: str): + """Handle key presses for navigation.""" + # Global quit + if key.lower() == "q" or key == "\x03": # q or Ctrl+C + self.running = False + return + + # Tab switching (unless in input mode) + if not self.input_mode and key == "\t": # Tab key + self.current_tab = (self.current_tab + 1) % len(self.tabs) + # Reset state when switching tabs + self.show_action_menu = False + self.action_menu_selection = 0 + self.selected_index = 0 + return + + # Handle based on current context + if self.input_mode: + self._handle_input_mode_keypress(key) + elif self.show_action_menu: + self._handle_action_menu_keypress(key) + elif self.current_tab == 0: # recents tab + self._handle_recents_keypress(key) + elif self.current_tab == 1: # new tab + self._handle_new_tab_keypress(key) + elif self.current_tab == 2: # web tab + self._handle_web_tab_keypress(key) + + def _handle_input_mode_keypress(self, key: str): + """Handle keypresses when in text input mode.""" + if key.lower() == "c": # 'C' key - exit input mode + self.input_mode = False + elif key == "\r" or key == "\n": # Enter - create agent run + if self.prompt_input.strip(): # Only create if there's actual content + self._create_background_agent(self.prompt_input) + else: + self.input_mode = False # Exit input mode if empty + elif key == "\x7f" or key == "\b": # Backspace + if self.cursor_position > 0: + self.prompt_input = self.prompt_input[: self.cursor_position - 1] + self.prompt_input[self.cursor_position :] + self.cursor_position -= 1 + elif key == "\x1b[C": # Right arrow + self.cursor_position = min(len(self.prompt_input), self.cursor_position + 1) + elif key == "\x1b[D": # Left arrow + self.cursor_position = max(0, self.cursor_position - 1) + elif len(key) == 1 and key.isprintable(): # Regular character + self.prompt_input = self.prompt_input[: self.cursor_position] + key + self.prompt_input[self.cursor_position :] + self.cursor_position += 1 + + def _handle_action_menu_keypress(self, key: str): + """Handle action menu keypresses.""" + if key == "\r" or key == "\n": # Enter + self._execute_inline_action() + self.show_action_menu = False # Close menu after action + elif key.lower() == "c" or key == "\x1b[D": # 'C' key or Left arrow to close + self.show_action_menu = False # Close menu + self.action_menu_selection = 0 # Reset selection + + def _handle_recents_keypress(self, key: str): + """Handle keypresses in the recents tab.""" + if key == "\x1b[A" or key.lower() == "w": # Up arrow or W + self.selected_index = max(0, self.selected_index - 1) + self.show_action_menu = False # Close any open menu + self.action_menu_selection = 0 + elif key == "\x1b[B" or key.lower() == "s": # Down arrow or S + self.selected_index = min(len(self.agent_runs) - 1, self.selected_index + 1) + self.show_action_menu = False # Close any open menu + self.action_menu_selection = 0 + elif key == "\x1b[C": # Right arrow - open action menu + self.show_action_menu = True # Open action menu + self.action_menu_selection = 0 # Reset to first option + elif key == "\x1b[D": # Left arrow - close action menu + self.show_action_menu = False # Close action menu + self.action_menu_selection = 0 + elif key == "\r" or key == "\n" or key.lower() == "e": # Enter or E + self.show_action_menu = True # Open action menu + self.action_menu_selection = 0 # Reset to first option + elif key.lower() == "r": + self._refresh() + self.show_action_menu = False # Close menu on refresh + self.action_menu_selection = 0 + + def _handle_new_tab_keypress(self, key: str): + """Handle keypresses in the new tab.""" + if key == "\r" or key == "\n": # Enter - start input mode + if not self.input_mode: + self.input_mode = True + self.cursor_position = len(self.prompt_input) + else: + # If already in input mode, Enter should create the agent + self._create_background_agent(self.prompt_input) + + def _handle_web_tab_keypress(self, key: str): + """Handle keypresses in the web tab.""" + if key == "\r" or key == "\n": # Enter - open web interface + try: + import webbrowser + + webbrowser.open("http://localhost:3000/me") + print("\n✅ Opening web interface in browser...") + except Exception as e: + print(f"\n❌ Failed to open browser: {e}") + input("Press Enter to continue...") + + def _execute_inline_action(self): + """Execute the selected action from the inline menu.""" + if not (0 <= self.selected_index < len(self.agent_runs)): + return + + agent_run = self.agent_runs[self.selected_index] + agent_id = agent_run.get("id", "unknown") + web_url = self._generate_agent_url(agent_id) + + # Get the available options to map selection to action + github_prs = agent_run.get("github_pull_requests", []) + options = ["open in web"] + + if github_prs: + options.insert(0, "pull locally") # Add as first option + + if github_prs and github_prs[0].get("url"): + options.append("open PR") + + # Always execute the first (top) option + if len(options) > 0: + selected_option = options[0] + + if selected_option == "pull locally": + self._pull_agent_branch(agent_id) + elif selected_option.startswith("open in web"): + try: + import webbrowser + + webbrowser.open(web_url) + # No pause - let it flow back naturally to collapsed state + except Exception as e: + print(f"\n❌ Failed to open browser: {e}") + input("Press Enter to continue...") # Only pause on errors + elif selected_option == "open PR": + pr_url = github_prs[0]["url"] + try: + import webbrowser + + webbrowser.open(pr_url) + # No pause - seamless flow back to collapsed state + except Exception as e: + print(f"\n❌ Failed to open PR: {e}") + input("Press Enter to continue...") # Only pause on errors + + def _open_agent_details(self): + """Toggle the inline action menu.""" + self.show_action_menu = not self.show_action_menu + if not self.show_action_menu: + self.action_menu_selection = 0 # Reset selection when closing + + def _refresh(self): + """Refresh the agent runs list.""" + if self._load_agent_runs(): + self.selected_index = 0 # Reset selection + + def _clear_and_redraw(self): + """Clear screen and redraw everything.""" + # Move cursor to top and clear screen from cursor down + print("\033[H\033[J", end="") + self._display_header() + self._display_content() + + # Show appropriate instructions based on context + if self.input_mode: + print("\n\033[90mType your prompt • [Enter] create • [C] cancel • [Q] quit\033[0m") + elif self.show_action_menu: + print("\n\033[90m[Enter] select • [C] close • [Q] quit\033[0m") + elif self.current_tab == 0: # recents + print("\n\033[90m[Tab] switch tabs • (↑↓) navigate • (←→) open/close • [Enter] actions • [R] refresh • [Q] quit\033[0m") + elif self.current_tab == 1: # new + print("\n\033[90m[Tab] switch tabs • [Enter] start typing • [Q] quit\033[0m") + elif self.current_tab == 2: # web + print("\n\033[90m[Tab] switch tabs • [Enter] open web • [Q] quit\033[0m") + + def run(self): + """Run the minimal TUI.""" + if not self.is_authenticated: + print("⚠️ Not authenticated. Please run 'codegen login' first.") + return + + print("Loading...") + if not self._load_agent_runs(): + print("Failed to load agent runs. Please check your authentication and try again.") + return + + # Initial display + self._clear_and_redraw() + + # Main event loop + while self.running: + try: + key = self._get_char() + self._handle_keypress(key) + if self.running: # Only redraw if we're still running + self._clear_and_redraw() + except KeyboardInterrupt: + # This should be handled by the signal handler, but just in case + break - def action_quit(self) -> None: - """Quit the application.""" - self.exit() + print() # Add newline before exiting def run_tui(): - """Run the Codegen TUI.""" - app = CodegenTUI() - app.run() + """Run the minimal Codegen TUI.""" + tui = MinimalTUI() + tui.run() diff --git a/src/codegen/cli/utils/url.py b/src/codegen/cli/utils/url.py index d12b3de83..ba9a9648f 100644 --- a/src/codegen/cli/utils/url.py +++ b/src/codegen/cli/utils/url.py @@ -12,13 +12,12 @@ class DomainRegistry(Enum): def get_domain() -> str: """Get the appropriate domain based on the current environment.""" - match global_env.ENV: - case Environment.PRODUCTION: - return DomainRegistry.PRODUCTION.value - case Environment.STAGING: - return DomainRegistry.STAGING.value - case _: - return DomainRegistry.LOCAL.value + if global_env.ENV == Environment.PRODUCTION: + return DomainRegistry.PRODUCTION.value + elif global_env.ENV == Environment.STAGING: + return DomainRegistry.STAGING.value + else: + return DomainRegistry.LOCAL.value def generate_webapp_url(path: str = "", params: dict | None = None, protocol: str = "https") -> str: