diff --git a/pyproject.toml b/pyproject.toml
index 3f9376e1..9cecc87c 100644
--- a/pyproject.toml
+++ b/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"
\ No newline at end of file
+cai-logs = "tools.logs:main"
+cai-asciinema = "tools.asciinema:main"
\ No newline at end of file
diff --git a/src/cai/repl/commands/base.py b/src/cai/repl/commands/base.py
index 4f540f85..e91a55df 100644
--- a/src/cai/repl/commands/base.py
+++ b/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
diff --git a/src/cai/repl/commands/help.py b/src/cai/repl/commands/help.py
index 4df6e498..8b49a4b9 100644
--- a/src/cai/repl/commands/help.py
+++ b/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(
diff --git a/src/cai/repl/commands/kill.py b/src/cai/repl/commands/kill.py
index aba42cac..fcfc1cde 100644
--- a/src/cai/repl/commands/kill.py
+++ b/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]")
diff --git a/src/cai/repl/ui/toolbar.py b/src/cai/repl/ui/toolbar.py
index 6f3edde6..3b4a6547 100644
--- a/src/cai/repl/ui/toolbar.py
+++ b/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}>ENV: {active_env_icon} {active_env_name}|"
- f"IP: {
- ip_address} | "
- f"OS: {
- os_name} {os_version} | "
- f"Ollama: {
- ollama_status} | "
- f"Model: {
- os.getenv('CAI_MODEL', 'default')} | "
- f"Max Turns: {
- os.getenv('CAI_MAX_TURNS', 'inf')} | "
- f"Price Limit: {
- os.getenv('CAI_PRICE_LIMIT', 'inf')} | "
+ f"IP: {ip_address} | "
+ f"OS: {os_name} {os_version} | "
+ f"Ollama: {ollama_status} | "
+ f"Model: {os.getenv('CAI_MODEL', 'default')} | "
+ f"Max Turns: {os.getenv('CAI_MAX_TURNS', 'inf')} | "
+ f"Price Limit: {os.getenv('CAI_PRICE_LIMIT', 'inf')} | "
f"{current_time_with_tz}"
)
toolbar_cache['last_update'] = datetime.datetime.now()
diff --git a/src/cai/tools/common.py b/src/cai/tools/common.py
index 1ba7ac28..4e4f6291 100644
--- a/src/cai/tools/common.py
+++ b/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
diff --git a/tools/asciinema.py b/tools/asciinema.py
new file mode 100644
index 00000000..1d703c48
--- /dev/null
+++ b/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()
diff --git a/tools/replay.py b/tools/replay.py
index fd34a63d..5c6987ef 100644
--- a/tools/replay.py
+++ b/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():