Add asciinema tool, fix outstanding issues with f-strings

Signed-off-by: Víctor Mayoral Vilches <v.mayoralv@gmail.com>

pyproject.toml

@@ -162,4 +162,5 @@ format-command = "ruff format --stdin-filename {filename}"
 [project.scripts]
 cai = "cai.cli:main"
 cai-replay = "tools.replay:main"
-cai-logs = "tools.logs:main"
+cai-logs = "tools.logs:main"
+cai-asciinema = "tools.asciinema:main"

src/cai/repl/commands/base.py

@@ -87,11 +87,9 @@ class Command:
         Returns:
             True if the command was handled successfully, False otherwise
         """
+        subcommands = ', '.join(self.get_subcommands())
         console.print(
-            f"[yellow]{
-                self.name} command requires a subcommand: {
-                ', '.join(
-                    self.get_subcommands())}[/yellow]")
+            f"[yellow]{self.name} command requires a subcommand: {subcommands}[/yellow]")
         return False
 
     def handle_unknown_subcommand(self, subcommand: str) -> bool:
@@ -104,8 +102,7 @@ class Command:
             True if the command was handled successfully, False otherwise
         """
         console.print(
-            f"[red]Unknown {
-                self.name} subcommand: {subcommand}[/red]")
+            f"[red]Unknown {self.name} subcommand: {subcommand}[/red]")
         return False
 
 

src/cai/repl/commands/help.py

@@ -819,9 +819,8 @@ class HelpCommand(Command):
                     platform = platform_manager.get_platform(platform_name)
                     commands = platform.get_commands()
                     if commands:
-                        examples.append(
-                            f"[green]/platform {platform_name} {
-                                commands[0]}[/green] - Example {platform_name} command")
+                        command_example = f"[green]/platform {platform_name} {commands[0]}[/green] - Example {platform_name} command"
+                        examples.append(command_example)
 
                 if examples:
                     console.print(Panel(

src/cai/repl/commands/kill.py

@@ -69,8 +69,7 @@ class KillCommand(Command):
             return False
         except ProcessLookupError:
             console.print(
-                f"[yellow]No process with PID {
-                    args[0]} found[/yellow]")
+                f"[yellow]No process with PID {args[0]} found[/yellow]")
             return False
         except Exception as e:  # pylint: disable=broad-exception-caught
             console.print(f"[red]Error killing process: {str(e)}[/red]")

src/cai/repl/ui/toolbar.py

@@ -113,18 +113,12 @@ def update_toolbar_in_background():
         # Update the cache
         toolbar_cache['html'] = HTML(
             f"<{active_env_color}><b>ENV:</b> {active_env_icon} {active_env_name}</{active_env_color}>|"
-            f"<ansired><b>IP:</b></ansired> <ansigreen>{
-                ip_address}</ansigreen> | "
-            f"<ansiyellow><b>OS:</b></ansiyellow> <ansiblue>{
-                os_name} {os_version}</ansiblue> | "
-            f"<ansicyan><b>Ollama:</b></ansicyan> <ansimagenta>{
-                ollama_status}</ansimagenta> | "
-            f"<ansiyellow><b>Model:</b></ansiyellow> <ansigreen>{
-                os.getenv('CAI_MODEL', 'default')}</ansigreen> | "
-            f"<ansicyan><b>Max Turns:</b></ansicyan> <ansiblue>{
-                os.getenv('CAI_MAX_TURNS', 'inf')}</ansiblue> | "
-            f"<ansiyellow><b>Price Limit:</b></ansiyellow> <ansiblue>{
-                os.getenv('CAI_PRICE_LIMIT', 'inf')}</ansiblue> | "
+            f"<ansired><b>IP:</b></ansired> <ansigreen>{ip_address}</ansigreen> | "
+            f"<ansiyellow><b>OS:</b></ansiyellow> <ansiblue>{os_name} {os_version}</ansiblue> | "
+            f"<ansicyan><b>Ollama:</b></ansicyan> <ansimagenta>{ollama_status}</ansimagenta> | "
+            f"<ansiyellow><b>Model:</b></ansiyellow> <ansigreen>{os.getenv('CAI_MODEL', 'default')}</ansigreen> | "
+            f"<ansicyan><b>Max Turns:</b></ansicyan> <ansiblue>{os.getenv('CAI_MAX_TURNS', 'inf')}</ansiblue> | "
+            f"<ansiyellow><b>Price Limit:</b></ansiyellow> <ansiblue>{os.getenv('CAI_PRICE_LIMIT', 'inf')}</ansiblue> | "
             f"<ansigray>{current_time_with_tz}</ansigray>"
         )
         toolbar_cache['last_update'] = datetime.datetime.now()

src/cai/tools/common.py

@@ -136,9 +136,7 @@ class ShellSession:  # pylint: disable=too-many-instance-attributes
         if self.ctf:
             self.is_running = True
             self.output_buffer.append(
-                f"[Session {
-                    self.session_id}] Started CTF command: {
-                    self.command}")
+                f"[Session {self.session_id}] Started CTF command: {self.command}")
             try:
                 output = self.ctf.get_shell(self.command)
                 self.output_buffer.append(output)
@@ -162,9 +160,7 @@ class ShellSession:  # pylint: disable=too-many-instance-attributes
             )
             self.is_running = True
             self.output_buffer.append(
-                f"[Session {
-                    self.session_id}] Started: {
-                    self.command}")
+                f"[Session {self.session_id}] Started: {self.command}")
             # Start a thread to read output
             threading.Thread(target=self._read_output, daemon=True).start()
         except Exception as e:  # pylint: disable=broad-except

tools/asciinema.py

@@ -0,0 +1,87 @@
+#!/usr/bin/env python3
+"""
+Tool to record asciinema sessions of JSONL replay files.
+
+Usage:
+    cai-asciinema path/to/file.jsonl 0.5
+
+This tool wraps asciinema recording to capture replay sessions.
+"""
+import argparse
+import os
+import subprocess
+import sys
+
+
+def parse_arguments():
+    """Parse command line arguments."""
+    parser = argparse.ArgumentParser(
+        description="Record asciinema sessions of JSONL replay files.",
+        formatter_class=argparse.RawDescriptionHelpFormatter,
+        epilog="""
+Examples:
+  cai-asciinema path/to/file.jsonl 0.5
+  cai-asciinema conversation.jsonl 1.0
+"""
+    )
+
+    parser.add_argument(
+        "jsonl_file",
+        help="Path to the JSONL file containing conversation history"
+    )
+
+    parser.add_argument(
+        "replay_delay",
+        type=float,
+        help="Time in seconds to wait between actions"
+    )
+
+    parser.add_argument(
+        "--output",
+        "-o",
+        type=str,
+        help="Output file path for the recording (optional)"
+    )
+
+    return parser.parse_args()
+
+
+def main():
+    """Main function to record asciinema session."""
+    args = parse_arguments()
+
+    # Validate that the JSONL file exists
+    if not os.path.exists(args.jsonl_file):
+        print(f"Error: File {args.jsonl_file} not found", file=sys.stderr)
+        sys.exit(1)
+
+    # Build the command to record using the same Python interpreter
+    replay_command = f"{sys.executable} tools/replay.py {args.jsonl_file} {args.replay_delay}"
+
+    # Build asciinema command
+    asciinema_cmd = ["asciinema", "rec", f"--command={replay_command}", "--overwrite"]
+
+    # Add output file if specified
+    if args.output:
+        asciinema_cmd.append(args.output)
+
+    print(f"Recording asciinema session for {args.jsonl_file} with delay {args.replay_delay}s...")
+    print(f"Running: {' '.join(asciinema_cmd)}")
+
+    try:
+        # Execute the asciinema command
+        result = subprocess.run(asciinema_cmd, check=True)
+        # result = subprocess.run(replay_command, check=True)
+        print("Recording completed successfully!")
+        return result.returncode
+    except subprocess.CalledProcessError as e:
+        print(f"Error: asciinema recording failed with exit code {e.returncode}", file=sys.stderr)
+        sys.exit(e.returncode)
+    except FileNotFoundError:
+        print("Error: asciinema not found. Please install asciinema first.", file=sys.stderr)
+        print("Install with: pip install asciinema", file=sys.stderr)
+        sys.exit(1)
+
+
+if __name__ == "__main__":
+    main()

tools/replay.py

@@ -154,83 +154,133 @@ def replay_conversation(messages: List[Dict], replay_delay: float = 0.5, usage:
                     message["tool_outputs"][call_id] = tool_outputs[call_id]
 
     for i, message in enumerate(messages):
-        # Add delay between actions
-        if i > 0:
-            time.sleep(replay_delay)
+        try:
+            # Add delay between actions
+            if i > 0:
+                time.sleep(replay_delay)
 
-        role = message.get("role", "")
-        content = message.get("content")
-        content = str(content).strip() if content is not None else ""
-        sender = message.get("sender", role)
-        model = message.get("model", file_model)
+            role = message.get("role", "")
+            content = message.get("content")
+            content = str(content).strip() if content is not None else ""
+            sender = message.get("sender", role)
+            model = message.get("model", file_model)
 
-        # Skip system messages
-        if role == "system":
-            continue
+            # Skip system messages
+            if role == "system":
+                continue
 
-        # Handle user messages
-        if role == "user":
-            # Use cli_print_agent_messages for user messages
-            print(color(f"CAI> ", fg="cyan") + f"{content}")
+            # Handle user messages
+            if role == "user":
+                # Use cli_print_agent_messages for user messages
+                print(color(f"CAI> ", fg="cyan") + f"{content}")
 
-            turn_counter += 1
-            interaction_counter = 0
+                turn_counter += 1
+                interaction_counter = 0
 
-        # Handle assistant messages
-        elif role == "assistant":
-            # Check if there are tool calls
-            tool_calls = message.get("tool_calls", [])
-            tool_outputs = message.get("tool_outputs", {})
+            # Handle assistant messages
+            elif role == "assistant":
+                # Check if there are tool calls
+                tool_calls = message.get("tool_calls", [])
+                tool_outputs = message.get("tool_outputs", {})
 
-            if tool_calls:
-                # Print the assistant message with tool calls
-                cli_print_agent_messages(
-                    sender,
-                    content or "",
-                    interaction_counter,
-                    model,
-                    debug,
-                    interaction_input_tokens=message.get("input_tokens", 0),
-                    interaction_output_tokens=message.get("output_tokens", 0),
-                    interaction_reasoning_tokens=message.get("reasoning_tokens", 0),
-                    total_input_tokens=total_input_tokens,
-                    total_output_tokens=total_output_tokens,
-                    total_reasoning_tokens=message.get("total_reasoning_tokens", 0),
-                    interaction_cost=message.get("interaction_cost", 0.0),
-                    total_cost=total_cost
-                )
+                if tool_calls:
+                    # Print the assistant message with tool calls
+                    cli_print_agent_messages(
+                        sender,
+                        content or "",
+                        interaction_counter,
+                        model,
+                        debug,
+                        interaction_input_tokens=message.get("input_tokens", 0),
+                        interaction_output_tokens=message.get("output_tokens", 0),
+                        interaction_reasoning_tokens=message.get("reasoning_tokens", 0),
+                        total_input_tokens=total_input_tokens,
+                        total_output_tokens=total_output_tokens,
+                        total_reasoning_tokens=message.get("total_reasoning_tokens", 0),
+                        interaction_cost=message.get("interaction_cost", 0.0),
+                        total_cost=total_cost
+                    )
 
-                # Print each tool call with its output
-                for tool_call in tool_calls:
-                    function = tool_call.get("function", {})
-                    name = function.get("name", "")
-                    arguments = function.get("arguments", "{}")
-                    call_id = tool_call.get("id", "")
+                    # Print each tool call with its output
+                    for tool_call in tool_calls:
+                        function = tool_call.get("function", {})
+                        name = function.get("name", "")
+                        arguments = function.get("arguments", "{}")
+                        call_id = tool_call.get("id", "")
 
-                    # Get the tool output if available
-                    tool_output = ""
-                    if call_id and call_id in tool_outputs:
-                        tool_output = tool_outputs[call_id]
+                        # Get the tool output if available
+                        tool_output = ""
+                        if call_id and call_id in tool_outputs:
+                            tool_output = tool_outputs[call_id]
 
-                    # Skip empty tool calls
-                    if not name:
-                        continue
+                        # Skip empty tool calls
+                        if not name:
+                            continue
 
-                    try:
-                        # Try to parse arguments as JSON
-                        if arguments and isinstance(arguments, str) and arguments.strip().startswith("{"):
-                            args_obj = json.loads(arguments)
-                        else:
+                        try:
+                            # Try to parse arguments as JSON
+                            if arguments and isinstance(arguments, str) and arguments.strip().startswith("{"):
+                                args_obj = json.loads(arguments)
+                            else:
+                                args_obj = arguments
+                        except json.JSONDecodeError:
                             args_obj = arguments
-                    except json.JSONDecodeError:
-                        args_obj = arguments
 
-                    # Print the tool call and output
+                        # Print the tool call and output
+                        cli_print_tool_output(
+                            tool_name=name,
+                            args=args_obj,
+                            output=tool_output,  # Use the matched tool output
+                            call_id=call_id,
+                            token_info={
+                                "interaction_input_tokens": message.get("input_tokens", 0),
+                                "interaction_output_tokens": message.get("output_tokens", 0),
+                                "interaction_reasoning_tokens": message.get("reasoning_tokens", 0),
+                                "total_input_tokens": total_input_tokens,
+                                "total_output_tokens": total_output_tokens,
+                                "total_reasoning_tokens": message.get("total_reasoning_tokens", 0),
+                                "model": model,
+                                "interaction_cost": message.get("interaction_cost", 0.0),
+                                "total_cost": total_cost
+                            }
+                        )
+                else:
+                    # Print regular assistant message
+                    cli_print_agent_messages(
+                        sender,
+                        content or "",
+                        interaction_counter,
+                        model,
+                        debug,
+                        interaction_input_tokens=message.get("input_tokens", 0),
+                        interaction_output_tokens=message.get("output_tokens", 0),
+                        interaction_reasoning_tokens=message.get("reasoning_tokens", 0),
+                        total_input_tokens=total_input_tokens,
+                        total_output_tokens=total_output_tokens,
+                        total_reasoning_tokens=message.get("total_reasoning_tokens", 0),
+                        interaction_cost=message.get("interaction_cost", 0.0),
+                        total_cost=total_cost
+                    )
+                interaction_counter += 1  # iterate the interaction counter
+
+            # Handle tool messages - only those not already displayed with assistant messages
+            elif role == "tool":
+                # Check if we've already displayed this tool output with an assistant message
+                tool_call_id = message.get("tool_call_id", "")
+
+                # Skip tool messages that have been displayed with an assistant message
+                is_already_displayed = False
+                for prev_msg in messages[:i]:
+                    if prev_msg.get("role") == "assistant" and tool_call_id in prev_msg.get("tool_outputs", {}):
+                        is_already_displayed = True
+                        break
+
+                if not is_already_displayed and content:  # Only show if there's actual content
+                    tool_name = message.get("name", message.get("tool_call_id", "unknown"))
                     cli_print_tool_output(
-                        tool_name=name,
-                        args=args_obj,
-                        output=tool_output,  # Use the matched tool output
-                        call_id=call_id,
+                        tool_name=tool_name,
+                        args="",
+                        output=content,
                         token_info={
                             "interaction_input_tokens": message.get("input_tokens", 0),
                             "interaction_output_tokens": message.get("output_tokens", 0),
@@ -243,77 +293,34 @@ def replay_conversation(messages: List[Dict], replay_delay: float = 0.5, usage:
                             "total_cost": total_cost
                         }
                     )
+
+            # Handle any other message types
             else:
-                # Print regular assistant message
-                cli_print_agent_messages(
-                    sender,
-                    content or "",
-                    interaction_counter,
-                    model,
-                    debug,
-                    interaction_input_tokens=message.get("input_tokens", 0),
-                    interaction_output_tokens=message.get("output_tokens", 0),
-                    interaction_reasoning_tokens=message.get("reasoning_tokens", 0),
-                    total_input_tokens=total_input_tokens,
-                    total_output_tokens=total_output_tokens,
-                    total_reasoning_tokens=message.get("total_reasoning_tokens", 0),
-                    interaction_cost=message.get("interaction_cost", 0.0),
-                    total_cost=total_cost
-                )
-            interaction_counter += 1  # iterate the interaction counter
+                if content:  # Only display if there's actual content
+                    cli_print_agent_messages(
+                        sender or role,
+                        content,
+                        interaction_counter,
+                        model,
+                        debug,
+                        interaction_input_tokens=message.get("input_tokens", 0),
+                        interaction_output_tokens=message.get("output_tokens", 0),
+                        interaction_reasoning_tokens=message.get("reasoning_tokens", 0),
+                        total_input_tokens=total_input_tokens,
+                        total_output_tokens=total_output_tokens,
+                        total_reasoning_tokens=message.get("total_reasoning_tokens", 0),
+                        interaction_cost=message.get("interaction_cost", 0.0),
+                        total_cost=total_cost
+                    )
 
-        # Handle tool messages - only those not already displayed with assistant messages
-        elif role == "tool":
-            # Check if we've already displayed this tool output with an assistant message
-            tool_call_id = message.get("tool_call_id", "")
-
-            # Skip tool messages that have been displayed with an assistant message
-            is_already_displayed = False
-            for prev_msg in messages[:i]:
-                if prev_msg.get("role") == "assistant" and tool_call_id in prev_msg.get("tool_outputs", {}):
-                    is_already_displayed = True
-                    break
-
-            if not is_already_displayed and content:  # Only show if there's actual content
-                tool_name = message.get("name", message.get("tool_call_id", "unknown"))
-                cli_print_tool_output(
-                    tool_name=tool_name,
-                    args="",
-                    output=content,
-                    token_info={
-                        "interaction_input_tokens": message.get("input_tokens", 0),
-                        "interaction_output_tokens": message.get("output_tokens", 0),
-                        "interaction_reasoning_tokens": message.get("reasoning_tokens", 0),
-                        "total_input_tokens": total_input_tokens,
-                        "total_output_tokens": total_output_tokens,
-                        "total_reasoning_tokens": message.get("total_reasoning_tokens", 0),
-                        "model": model,
-                        "interaction_cost": message.get("interaction_cost", 0.0),
-                        "total_cost": total_cost
-                    }
-                )
-
-        # Handle any other message types
-        else:
-            if content:  # Only display if there's actual content
-                cli_print_agent_messages(
-                    sender or role,
-                    content,
-                    interaction_counter,
-                    model,
-                    debug,
-                    interaction_input_tokens=message.get("input_tokens", 0),
-                    interaction_output_tokens=message.get("output_tokens", 0),
-                    interaction_reasoning_tokens=message.get("reasoning_tokens", 0),
-                    total_input_tokens=total_input_tokens,
-                    total_output_tokens=total_output_tokens,
-                    total_reasoning_tokens=message.get("total_reasoning_tokens", 0),
-                    interaction_cost=message.get("interaction_cost", 0.0),
-                    total_cost=total_cost
-                )
-
-        # Force flush stdout to ensure immediate printing
-        sys.stdout.flush()
+            # Force flush stdout to ensure immediate printing
+            sys.stdout.flush()
+            
+        except Exception as e:
+            # Handle any errors during message processing
+            print(color(f"Warning: Error processing message {i+1}: {str(e)}", fg="yellow"))
+            print(color("Continuing with next message...", fg="yellow"))
+            continue
 
 
 def parse_arguments():