textual v0 + CLI cleanups (#1195)

Author: jayhack

pyproject.toml

@@ -9,6 +9,7 @@ dependencies = [
   "codegen-api-client",
   "typer>=0.12.5",
   "rich>=13.7.1",
+  "textual>=0.91.0",
   "hatch-vcs>=0.4.0",
   "hatchling>=1.25.0",
   # CLI and git functionality dependencies

src/codegen/cli/cli.py

@@ -4,6 +4,7 @@
 from codegen import __version__
 
 # Import config command (still a Typer app)
+from codegen.cli.commands.agent.main import agent
 from codegen.cli.commands.agents.main import agents_app
 
 # Import the actual command functions
@@ -16,6 +17,7 @@
 from codegen.cli.commands.profile.main import profile
 from codegen.cli.commands.style_debug.main import style_debug
 from codegen.cli.commands.tools.main import tools
+from codegen.cli.commands.tui.main import tui
 from codegen.cli.commands.update.main import update
 
 install(show_locals=True)
@@ -32,13 +34,15 @@ def version_callback(value: bool):
 main = typer.Typer(name="codegen", help="Codegen - the Operating System for Code Agents.", rich_markup_mode="rich")
 
 # Add individual commands to the main app
+main.command("agent", help="Create a new agent run with a prompt.")(agent)
 main.command("claude", help="Run Claude Code with OpenTelemetry monitoring and logging.")(claude)
 main.command("init", help="Initialize or update the Codegen folder.")(init)
 main.command("login", help="Store authentication token.")(login)
 main.command("logout", help="Clear stored authentication token.")(logout)
 main.command("profile", help="Display information about the currently authenticated user.")(profile)
 main.command("style-debug", help="Debug command to visualize CLI styling (spinners, etc).")(style_debug)
 main.command("tools", help="List available tools from the Codegen API.")(tools)
+main.command("tui", help="Launch the interactive TUI interface.")(tui)
 main.command("update", help="Update Codegen to the latest or specified version")(update)
 
 # Add Typer apps as sub-applications
@@ -47,10 +51,14 @@ def version_callback(value: bool):
 main.add_typer(integrations_app, name="integrations")
 
 
-@main.callback()
-def main_callback(version: bool = typer.Option(False, "--version", callback=version_callback, is_eager=True, help="Show version and exit")):
+@main.callback(invoke_without_command=True)
+def main_callback(ctx: typer.Context, version: bool = typer.Option(False, "--version", callback=version_callback, is_eager=True, help="Show version and exit")):
     """Codegen - the Operating System for Code Agents"""
-    pass
+    if ctx.invoked_subcommand is None:
+        # No subcommand provided, launch TUI
+        from codegen.cli.tui.app import run_tui
+
+        run_tui()
 
 
 if __name__ == "__main__":

src/codegen/cli/commands/agent/__init__.py

@@ -0,0 +1 @@
+"""Agent command module."""

src/codegen/cli/commands/agent/main.py

@@ -0,0 +1,159 @@
+"""Agent command for creating remote agent runs."""
+
+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.rich.spinners import create_spinner
+from codegen.cli.utils.org import resolve_org_id
+
+console = Console()
+
+# Create the agent app
+agent_app = typer.Typer(help="Create and manage individual agent runs")
+
+
+@agent_app.command()
+def create(
+    prompt: str = typer.Option(..., "--prompt", "-p", help="The prompt to send to the agent"),
+    org_id: int | None = typer.Option(None, help="Organization ID (defaults to CODEGEN_ORG_ID/REPOSITORY_ORG_ID or auto-detect)"),
+    model: str | None = typer.Option(None, help="Model to use for this agent run (optional)"),
+    repo_id: int | None = typer.Option(None, help="Repository ID to use for this agent run (optional)"),
+):
+    """Create a new agent run with the given prompt."""
+    # Get the current token
+    token = get_current_token()
+    if not token:
+        console.print("[red]Error:[/red] Not authenticated. Please run 'codegen login' first.")
+        raise typer.Exit(1)
+
+    try:
+        # Resolve org id
+        resolved_org_id = resolve_org_id(org_id)
+        if resolved_org_id is None:
+            console.print("[red]Error:[/red] Organization ID not provided. Pass --org-id, set CODEGEN_ORG_ID, or REPOSITORY_ORG_ID.")
+            raise typer.Exit(1)
+
+        # Prepare the request payload
+        payload = {
+            "prompt": prompt,
+        }
+
+        if model:
+            payload["model"] = model
+        if repo_id:
+            payload["repo_id"] = repo_id
+
+        # Make API request to create agent run with spinner
+        spinner = create_spinner("Creating agent run...")
+        spinner.start()
+
+        try:
+            headers = {"Authorization": f"Bearer {token}", "Content-Type": "application/json"}
+            url = f"{API_ENDPOINT.rstrip('/')}/v1/organizations/{resolved_org_id}/agent/run"
+            response = requests.post(url, headers=headers, json=payload)
+            response.raise_for_status()
+            agent_run_data = response.json()
+        finally:
+            spinner.stop()
+
+        # Extract agent run information
+        run_id = agent_run_data.get("id", "Unknown")
+        status = agent_run_data.get("status", "Unknown")
+        web_url = agent_run_data.get("web_url", "")
+        created_at = agent_run_data.get("created_at", "")
+
+        # Format created date
+        if created_at:
+            try:
+                from datetime import datetime
+
+                dt = datetime.fromisoformat(created_at.replace("Z", "+00:00"))
+                created_display = dt.strftime("%B %d, %Y at %H:%M")
+            except (ValueError, TypeError):
+                created_display = created_at
+        else:
+            created_display = "Unknown"
+
+        # Status with emoji
+        status_display = status
+        if status == "COMPLETE":
+            status_display = "✅ Complete"
+        elif status == "RUNNING":
+            status_display = "🏃 Running"
+        elif status == "FAILED":
+            status_display = "❌ Failed"
+        elif status == "STOPPED":
+            status_display = "⏹️ Stopped"
+        elif status == "PENDING":
+            status_display = "⏳ Pending"
+
+        # Create result display
+        result_info = []
+        result_info.append(f"[cyan]Agent Run ID:[/cyan] {run_id}")
+        result_info.append(f"[cyan]Status:[/cyan]       {status_display}")
+        result_info.append(f"[cyan]Created:[/cyan]      {created_display}")
+        if web_url:
+            result_info.append(f"[cyan]Web URL:[/cyan]      {web_url}")
+
+        result_text = "\n".join(result_info)
+
+        console.print(
+            Panel(
+                result_text,
+                title="🤖 [bold]Agent Run Created[/bold]",
+                border_style="green",
+                box=box.ROUNDED,
+                padding=(1, 2),
+            )
+        )
+
+        # Show next steps
+        console.print("\n[dim]💡 Track progress with:[/dim] [cyan]codegen agents[/cyan]")
+        if web_url:
+            console.print(f"[dim]🌐 View in browser:[/dim]  [link]{web_url}[/link]")
+
+    except requests.RequestException as e:
+        console.print(f"[red]Error creating agent run:[/red] {e}", style="bold red")
+        if hasattr(e, "response") and e.response is not None:
+            try:
+                error_detail = e.response.json().get("detail", "Unknown error")
+                console.print(f"[red]Details:[/red] {error_detail}")
+            except (ValueError, KeyError):
+                pass
+        raise typer.Exit(1)
+    except Exception as e:
+        console.print(f"[red]Unexpected error:[/red] {e}", style="bold red")
+        raise typer.Exit(1)
+
+
+# Default callback for the agent app
+@agent_app.callback(invoke_without_command=True)
+def agent_callback(ctx: typer.Context):
+    """Create and manage individual agent runs."""
+    if ctx.invoked_subcommand is None:
+        # If no subcommand is provided, show help
+        print(ctx.get_help())
+        raise typer.Exit()
+
+
+# For backward compatibility, also allow `codegen agent --prompt "..."`
+def agent(
+    prompt: str = typer.Option(None, "--prompt", "-p", help="The prompt to send to the agent"),
+    org_id: int | None = typer.Option(None, help="Organization ID (defaults to CODEGEN_ORG_ID/REPOSITORY_ORG_ID or auto-detect)"),
+    model: str | None = typer.Option(None, help="Model to use for this agent run (optional)"),
+    repo_id: int | None = typer.Option(None, help="Repository ID to use for this agent run (optional)"),
+):
+    """Create a new agent run with the given prompt."""
+    if prompt:
+        # If prompt is provided, create the agent run
+        create(prompt=prompt, org_id=org_id, model=model, repo_id=repo_id)
+    else:
+        # If no prompt, show help
+        console.print("[red]Error:[/red] --prompt is required")
+        console.print("Usage: [cyan]codegen agent --prompt 'Your prompt here'[/cyan]")
+        raise typer.Exit(1)

src/codegen/cli/commands/agents/main.py

@@ -33,13 +33,28 @@ def list_agents(org_id: int | None = typer.Option(None, help="Organization ID (d
             raise typer.Exit(1)
 
         # Make API request to list agent runs with spinner
-        spinner = create_spinner("Fetching agent runs...")
+        spinner = create_spinner("Fetching your recent API agent runs...")
         spinner.start()
 
         try:
             headers = {"Authorization": f"Bearer {token}"}
+            # Filter to only API source type and current user's agent runs
+            params = {
+                "source_type": "API",
+                # We'll get the user_id from the /users/me endpoint
+            }
+
+            # First get the 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")
+
+            if user_id:
+                params["user_id"] = user_id
+
             url = f"{API_ENDPOINT.rstrip('/')}/v1/organizations/{resolved_org_id}/agent/runs"
-            response = requests.get(url, headers=headers)
+            response = requests.get(url, headers=headers, params=params)
             response.raise_for_status()
             response_data = response.json()
         finally:
@@ -52,42 +67,58 @@ def list_agents(org_id: int | None = typer.Option(None, help="Organization ID (d
         page_size = response_data.get("page_size", 10)
 
         if not agent_runs:
-            console.print("[yellow]No agent runs found.[/yellow]")
+            console.print("[yellow]No API agent runs found for your user.[/yellow]")
             return
 
         # Create a table to display agent runs
         table = Table(
-            title=f"Agent Runs (Page {page}, Total: {total})",
+            title=f"Your Recent API Agent Runs (Page {page}, Total: {total})",
             border_style="blue",
             show_header=True,
             title_justify="center",
         )
-        table.add_column("ID", style="cyan", no_wrap=True)
-        table.add_column("Status", style="white", justify="center")
-        table.add_column("Source", style="magenta")
         table.add_column("Created", style="dim")
-        table.add_column("Result", style="green")
+        table.add_column("Status", style="white", justify="center")
+        table.add_column("Summary", style="green")
+        table.add_column("Link", style="blue")
 
         # Add agent runs to table
         for agent_run in agent_runs:
             run_id = str(agent_run.get("id", "Unknown"))
             status = agent_run.get("status", "Unknown")
             source_type = agent_run.get("source_type", "Unknown")
             created_at = agent_run.get("created_at", "Unknown")
-            result = agent_run.get("result", "")
 
-            # Status with emoji
-            status_display = status
+            # Extract summary from task_timeline_json, similar to frontend
+            timeline = agent_run.get("task_timeline_json")
+            summary = None
+            if timeline and isinstance(timeline, dict) and "summary" in timeline:
+                if isinstance(timeline["summary"], str):
+                    summary = timeline["summary"]
+
+            # Fall back to goal_prompt if no summary
+            if not summary:
+                summary = agent_run.get("goal_prompt", "")
+
+            # Status with colored circles
             if status == "COMPLETE":
-                status_display = "✅ Complete"
+                status_display = "[green]●[/green] Complete"
+            elif status == "ACTIVE":
+                status_display = "[dim]●[/dim] Active"
             elif status == "RUNNING":
-                status_display = "🏃 Running"
+                status_display = "[dim]●[/dim] Running"
+            elif status == "CANCELLED":
+                status_display = "[yellow]●[/yellow] Cancelled"
+            elif status == "ERROR":
+                status_display = "[red]●[/red] Error"
             elif status == "FAILED":
-                status_display = "❌ Failed"
+                status_display = "[red]●[/red] Failed"
             elif status == "STOPPED":
-                status_display = "⏹️ Stopped"
+                status_display = "[yellow]●[/yellow] Stopped"
             elif status == "PENDING":
-                status_display = "⏳ Pending"
+                status_display = "[dim]●[/dim] Pending"
+            else:
+                status_display = "[dim]●[/dim] " + status
 
             # Format created date (just show date and time, not full timestamp)
             if created_at and created_at != "Unknown":
@@ -102,13 +133,20 @@ def list_agents(org_id: int | None = typer.Option(None, help="Organization ID (d
             else:
                 created_display = created_at
 
-            # Truncate result if too long
-            result_display = result[:50] + "..." if result and len(result) > 50 else result or "No result"
+            # Truncate summary if too long
+            summary_display = summary[:50] + "..." if summary and len(summary) > 50 else summary or "No summary"
+
+            # Create web link for the agent run
+            web_url = agent_run.get("web_url")
+            if not web_url:
+                # Construct URL if not provided
+                web_url = f"https://codegen.com/traces/{run_id}"
+            link_display = web_url
 
-            table.add_row(run_id, status_display, source_type, created_display, result_display)
+            table.add_row(created_display, status_display, summary_display, link_display)
 
         console.print(table)
-        console.print(f"\n[green]Showing {len(agent_runs)} of {total} agent runs[/green]")
+        console.print(f"\n[green]Showing {len(agent_runs)} of {total} API agent runs[/green]")
 
     except requests.RequestException as e:
         console.print(f"[red]Error fetching agent runs:[/red] {e}", style="bold red")