"""

This module provides a CLI interface for testing and

interacting with CAI agents.

Environment Variables

---------------------

Required:

N/A

Optional:

CTF_NAME: Name of the CTF challenge to

run (e.g. "picoctf_static_flag")

CTF_CHALLENGE: Specific challenge name

within the CTF to test

CTF_SUBNET: Network subnet for the CTF

container (default: "192.168.3.0/24")

CTF_IP: IP address for the CTF

container (default: "192.168.3.100")

CTF_INSIDE: Whether to conquer the CTF from

within container (default: "true")

CAI_MODEL: Model to use for agents

(default: "alias1")

CAI_DEBUG: Set debug output level (default: "1")

- 0: Only tool outputs

- 1: Verbose debug output

- 2: CLI debug output

CAI_BRIEF: Enable/disable brief output mode (default: "false")

CAI_MAX_TURNS: Maximum number of turns for

agent interactions (default: "inf")

CAI_TRACING: Enable/disable OpenTelemetry tracing

(default: "true"). When enabled, traces execution

flow and agent interactions for debugging and analysis.

CAI_AGENT_TYPE: Specify the agents to use it could take

the value of (default: "one_tool_agent"). Use "/agent"

command in CLI to list all available agents.

CAI_STATE: Enable/disable stateful mode (default: "false").

When enabled, the agent will use a state agent to keep

track of the state of the network and the flags found.

CAI_MEMORY: Enable/disable memory mode (default: "false")

- episodic: use episodic memory

- semantic: use semantic memory

- all: use both episodic and semantic memorys

CAI_MEMORY_ONLINE: Enable/disable online memory mode

(default: "false")

CAI_MEMORY_OFFLINE: Enable/disable offline memory

(default: "false")

CAI_ENV_CONTEXT: Add environment context, dirs and

current env available (default: "true")

CAI_MEMORY_ONLINE_INTERVAL: Number of turns between

online memory updates (default: "5")

CAI_PRICE_LIMIT: Price limit for the conversation in dollars

(default: "1")

CAI_SUPPORT_MODEL: Model to use for the support agent

(default: "o3-mini")

CAI_SUPPORT_INTERVAL: Number of turns between support agent

executions (default: "5")

CAI_STREAM: Enable/disable streaming output in rich panel

(default: "false")

CAI_TELEMETRY: Enable/disable telemetry (default: "true")

CAI_PARALLEL: Number of parallel agent instances to run

(default: "1"). When set to values greater than 1,

executes multiple instances of the same agent in

parallel and displays all results.

CAI_GUARDRAILS: Enable/disable security guardrails for agents

(default: "true"). When enabled, applies security guardrails

to prevent potentially dangerous outputs and inputs. Set to

"false" to disable all guardrail functionality.

Extensions (only applicable if the right extension is installed):

"report"

CAI_REPORT: Enable/disable reporter mode. Possible values:

- ctf (default): do a report from a ctf resolution

- nis2: do a report for nis2

- pentesting: do a report from a pentesting

Usage Examples:

# Run against a CTF

CTF_NAME="kiddoctf" CTF_CHALLENGE="02 linux ii" \

CAI_AGENT_TYPE="one_tool_agent" CAI_MODEL="alias1" \

CAI_TRACING="false" cai

# Run a harder CTF

CTF_NAME="hackableii" CAI_AGENT_TYPE="redteam_agent" \

CTF_INSIDE="False" CAI_MODEL="alias1" \

CAI_TRACING="false" cai

# Run without a target in human-in-the-loop mode, generating a report

CAI_TRACING=False CAI_REPORT=pentesting CAI_MODEL="alias1" \

cai

# Run with online episodic memory

# registers memory every 5 turns:

# limits the cost to 5 dollars

CTF_NAME="hackableII" CAI_MEMORY="episodic" \

CAI_MODEL="alias1" CAI_MEMORY_ONLINE="True" \

CTF_INSIDE="False" CTF_HINTS="False" \

CAI_PRICE_LIMIT="5" cai

# Run with custom long_term_memory interval

# Executes memory long_term_memory every 3 turns:

CTF_NAME="hackableII" CAI_MEMORY="episodic" \

CAI_MODEL="alias1" CAI_MEMORY_ONLINE_INTERVAL="3" \

CAI_MEMORY_ONLINE="False" CTF_INSIDE="False" \

CTF_HINTS="False" cai

# Run with parallel agents (3 instances)

CTF_NAME="hackableII" CAI_AGENT_TYPE="redteam_agent" \

CAI_MODEL="alias1" CAI_PARALLEL="3" cai

"""

# Load environment variables from .env file FIRST, before any imports

import os

from dotenv import load_dotenv

# Load .env file from current working directory with override

# This ensures env vars from .env take precedence over system env vars

load_dotenv(override=True)

# Configure Python warnings BEFORE any other imports

import warnings

import sys

# Custom warning handler to suppress specific warnings

def custom_warning_handler(message, category, filename, lineno, file=None, line=None):

# Only show warnings in debug mode

if os.getenv("CAI_DEBUG", "1") == "2":

# Format and print the warning

warnings.showwarning(message, category, filename, lineno, file, line)

# Otherwise, silently ignore

# Set custom warning handler

warnings.showwarning = custom_warning_handler

# Suppress ALL warnings in production mode (unless CAI_DEBUG=2)

if os.getenv("CAI_DEBUG", "1") != "2":

warnings.filterwarnings("ignore")

# Also set environment variable to prevent warnings from subprocesses

os.environ["PYTHONWARNINGS"] = "ignore"

import asyncio

import logging

import shlex

import time

# Configure comprehensive error filtering

class ComprehensiveErrorFilter(logging.Filter):

"""Filter to suppress various expected errors and warnings."""

def filter(self, record):

msg = record.getMessage().lower()

# List of patterns to suppress completely

suppress_patterns = [

"asynchronous generator",

"asyncgen",

"closedresourceerror",

"didn't stop after athrow",

"didnt stop after athrow",

"didn't stop after athrow",

"generator didn't stop",

"generator didn't stop",

"cancel scope",

"unhandled errors in a taskgroup",

"error in post_writer",

"was never awaited",

"connection error while setting up",

"error closing",

"anyio._backends",

"httpx_sse",

"connection reset by peer",

"broken pipe",

"connection aborted",

"runtime warning",

"runtimewarning",

"coroutine",

"task was destroyed",

"event loop is closed",

"session is closed",

# Add specific aiohttp session warnings

"unclosed client session",

"unclosed connector",

"client_session:",

"connector:",

"connections:",

]

# Check if any suppress pattern matches

for pattern in suppress_patterns:

if pattern in msg:

return False

# SSE connection errors during cleanup

if "sse" in msg and any(word in msg for word in ["cleanup", "closing", "shutdown", "closed"]):

return False

# MCP connection errors that we handle

if "error invoking mcp tool" in msg and "closedresourceerror" in msg:

return False

# MCP reconnection messages - change to DEBUG level

if "mcp server session not found" in msg or "successfully reconnected to mcp server" in msg:

record.levelno = logging.DEBUG

record.levelname = "DEBUG"

return True

# Apply comprehensive filter to all relevant loggers

comprehensive_filter = ComprehensiveErrorFilter()

# List of loggers to configure

loggers_to_configure = [

"openai.agents",

"mcp.client.sse",

"httpx",

"httpx_sse",

"mcp",

"asyncio",

"anyio",

"anyio._backends._asyncio",

"cai.sdk.agents",

"aiohttp", # Add aiohttp logger to suppress session warnings

]

for logger_name in loggers_to_configure:

logger = logging.getLogger(logger_name)

logger.addFilter(comprehensive_filter)

# Set appropriate level - ERROR for most, WARNING for critical ones

if logger_name in ["asyncio", "anyio", "anyio._backends._asyncio"]:

logger.setLevel(logging.ERROR) # Only show critical errors

else:

logger.setLevel(logging.WARNING)

# Suppress various warnings globally with more comprehensive patterns

warnings.filterwarnings("ignore", category=RuntimeWarning) # Ignore ALL RuntimeWarnings

warnings.filterwarnings("ignore", category=ResourceWarning) # Ignore ResourceWarnings (aiohttp sessions)

warnings.filterwarnings("ignore", message=".*asynchronous generator.*")

warnings.filterwarnings("ignore", message=".*was never awaited.*")

warnings.filterwarnings("ignore", message=".*didn't stop after athrow.*")

warnings.filterwarnings("ignore", message=".*didn't stop after athrow.*")

warnings.filterwarnings("ignore", message=".*cancel scope.*")

warnings.filterwarnings("ignore", message=".*coroutine.*was never awaited.*")

warnings.filterwarnings("ignore", message=".*generator.*didn't stop.*")

warnings.filterwarnings("ignore", message=".*Task was destroyed.*")

warnings.filterwarnings("ignore", message=".*Event loop is closed.*")

# Add specific aiohttp session warnings

warnings.filterwarnings("ignore", message=".*Unclosed client session.*")

warnings.filterwarnings("ignore", message=".*Unclosed connector.*")

warnings.filterwarnings("ignore", message=".*client_session:.*")

warnings.filterwarnings("ignore", message=".*connector:.*")

warnings.filterwarnings("ignore", message=".*connections:.*")

# Also configure Python's warning system to be less verbose

import sys

if not sys.warnoptions:

warnings.simplefilter("ignore", RuntimeWarning)

warnings.simplefilter("ignore", ResourceWarning) # Also ignore ResourceWarnings

# Additional aiohttp warning suppression

def suppress_aiohttp_warnings():

"""Suppress aiohttp specific warnings about unclosed sessions."""

try:

import aiohttp

# Suppress aiohttp warnings about unclosed sessions

aiohttp_logger = logging.getLogger("aiohttp")

aiohttp_logger.setLevel(logging.ERROR) # Only show errors, not warnings

# Also suppress aiohttp.client warnings

aiohttp_client_logger = logging.getLogger("aiohttp.client")

aiohttp_client_logger.setLevel(logging.ERROR)

# Suppress aiohttp.connector warnings

aiohttp_connector_logger = logging.getLogger("aiohttp.connector")

aiohttp_connector_logger.setLevel(logging.ERROR)

except ImportError:

# aiohttp not installed, skip

pass

# Call the function to suppress aiohttp warnings

suppress_aiohttp_warnings()

# OpenAI imports

from openai import AsyncOpenAI

from rich.console import Console

from cai import is_pentestperf_available

# CAI agents and metrics imports

from cai.agents import get_agent_by_name

from cai.internal.components.metrics import process_metrics

# CAI REPL imports

from cai.repl.commands import FuzzyCommandCompleter, handle_command as commands_handle_command

# Add import for parallel configs at the top of the file

from cai.repl.commands.parallel import PARALLEL_CONFIGS, ParallelConfig, PARALLEL_AGENT_INSTANCES

# Global storage for shared message histories (keyed by a unique identifier)

UNIFIED_MESSAGE_HISTORIES = {}

from cai.repl.ui.banner import display_banner, display_quick_guide

from cai.repl.ui.keybindings import create_key_bindings

from cai.repl.ui.logging import setup_session_logging

from cai.repl.ui.prompt import get_user_input

from cai.repl.ui.toolbar import get_toolbar_with_refresh

# CAI SDK imports

from cai.sdk.agents import Agent, OpenAIChatCompletionsModel, Runner, set_tracing_disabled

from cai.sdk.agents.items import ToolCallOutputItem

from cai.sdk.agents.exceptions import OutputGuardrailTripwireTriggered, InputGuardrailTripwireTriggered

from cai.sdk.agents.models.openai_chatcompletions import (

get_agent_message_history,

get_all_agent_histories,

)

# Import handled where needed to avoid circular imports

from cai.sdk.agents.run_to_jsonl import get_session_recorder

from cai.sdk.agents.global_usage_tracker import GLOBAL_USAGE_TRACKER

from cai.sdk.agents.stream_events import RunItemStreamEvent

# CAI utility imports

from cai.util import (

color,

fix_litellm_transcription_annotations,

setup_ctf,

start_active_timer,

start_idle_timer,

stop_active_timer,

stop_idle_timer,

)

ctf_global = None

messages_ctf = ""

ctf_init = 1

previous_ctf_name = os.getenv("CTF_NAME", None)

if is_pentestperf_available() and os.getenv("CTF_NAME", None):

ctf, messages_ctf = setup_ctf()

ctf_global = ctf

ctf_init = 0

# NOTE: This is needed when using LiteLLM Proxy Server

#

# external_client = AsyncOpenAI(

# base_url = os.getenv('LITELLM_BASE_URL', 'http://localhost:4000'),

# api_key=os.getenv('LITELLM_API_KEY', 'key'))

#

# set_default_openai_client(external_client)

# Global variables for timing tracking

global START_TIME

START_TIME = time.time()

set_tracing_disabled(True)

def update_agent_models_recursively(agent, new_model, visited=None):

"""

Recursively update the model for an agent and all agents in its handoffs.

Args:

agent: The agent to update

new_model: The new model string to set

visited: Set of agent names already visited to prevent infinite loops

"""

if visited is None:

visited = set()

# Avoid infinite loops by tracking visited agents

if agent.name in visited:

return

visited.add(agent.name)

# Update the main agent's model

if hasattr(agent, "model") and hasattr(agent.model, "model"):

agent.model.model = new_model

# Also ensure the agent name is set correctly in the model

if hasattr(agent.model, "agent_name"):

agent.model.agent_name = agent.name

# IMPORTANT: Clear any cached state in the model that might be model-specific

# This ensures the model doesn't have stale state from the previous model

if hasattr(agent.model, "_client"):

# Force recreation of the client on next use

agent.model._client = None

if hasattr(agent.model, "_converter"):

# Reset the converter's state

if hasattr(agent.model._converter, "recent_tool_calls"):

agent.model._converter.recent_tool_calls.clear()

if hasattr(agent.model._converter, "tool_outputs"):

agent.model._converter.tool_outputs.clear()

# Update models for all handoff agents

if hasattr(agent, "handoffs"):

for handoff_item in agent.handoffs:

# Handle both direct Agent references and Handoff objects

if hasattr(handoff_item, "on_invoke_handoff"):

# This is a Handoff object

# For handoffs created with the handoff() function, the agent is stored

# in the closure of the on_invoke_handoff function

# We can try to extract it from the function's closure

try:

# Get the closure variables of the handoff function

if (

hasattr(handoff_item.on_invoke_handoff, "__closure__")

and handoff_item.on_invoke_handoff.__closure__

):

for cell in handoff_item.on_invoke_handoff.__closure__:

if hasattr(cell.cell_contents, "model") and hasattr(

cell.cell_contents, "name"

):

# This looks like an agent

handoff_agent = cell.cell_contents

update_agent_models_recursively(handoff_agent, new_model, visited)

break

except Exception:

# If we can't extract the agent from closure, skip it

pass

elif hasattr(handoff_item, "model"):

# This is a direct Agent reference

update_agent_models_recursively(handoff_item, new_model, visited)

def run_cai_cli(

starting_agent, context_variables=None, max_turns=float("inf"), force_until_flag=False, initial_prompt=None

):

"""

Run a simple interactive CLI loop for CAI.

Args:

starting_agent: The initial agent to use for the conversation

context_variables: Optional dictionary of context variables to initialize the session

max_turns: Maximum number of interaction turns before terminating (default: infinity)

force_until_flag: Whether to force execution until a flag is found

initial_prompt: Optional initial prompt to execute immediately before entering interactive mode

Returns:

None

"""

ACTIVE_TIME = 0 # TODO: review this variable

agent = starting_agent

turn_count = 0

idle_time = 0

console = Console()

last_model = os.getenv("CAI_MODEL", "alias1")

last_agent_type = os.getenv("CAI_AGENT_TYPE", "one_tool_agent")

parallel_count = int(os.getenv("CAI_PARALLEL", "1"))

use_initial_prompt = initial_prompt is not None

# Reset cost tracking at the start

from cai.util import COST_TRACKER

COST_TRACKER.reset_agent_costs()

# Reset simple agent manager for clean start

from cai.sdk.agents.simple_agent_manager import AGENT_MANAGER

AGENT_MANAGER.reset_registry()

# Register the starting agent with AGENT_MANAGER

starting_agent_name = getattr(starting_agent, "name", last_agent_type)

AGENT_MANAGER.switch_to_single_agent(starting_agent, starting_agent_name)

# Initialize command completer and key bindings

command_completer = FuzzyCommandCompleter()

current_text = [""]

kb = create_key_bindings(current_text)

# Setup session logging

history_file = setup_session_logging()

# Initialize session logger and display the filename

session_logger = get_session_recorder()

# Start global usage tracking session

GLOBAL_USAGE_TRACKER.start_session(

session_id=session_logger.session_id,

agent_name=None # Will be updated when agent is selected

)

# Display banner

display_banner(console)

print("\n")

display_quick_guide(console)

# Function to get the short name of the agent for display

def get_agent_short_name(agent):

if hasattr(agent, "name"):

# Return the full agent name instead of just the first word

return agent.name

return "Agent"

# Prevent the model from using its own rich streaming to avoid conflicts

# but allow final output message to ensure all agent responses are shown

if hasattr(agent, "model"):

if hasattr(agent.model, "disable_rich_streaming"):

agent.model.disable_rich_streaming = False # Now True as the model handles streaming

if hasattr(agent.model, "suppress_final_output"):

agent.model.suppress_final_output = False # Changed to False to show all agent messages

# Set the agent name in the model for proper display in streaming panel

if hasattr(agent.model, "set_agent_name"):

agent.model.set_agent_name(get_agent_short_name(agent))

prev_max_turns = max_turns

turn_limit_reached = False

while True:

# Check if the ctf name has changed and instanciate the ctf

global previous_ctf_name

global ctf_global

global messages_ctf

global ctf_init

if previous_ctf_name != os.getenv("CTF_NAME", None):

if is_pentestperf_available():

if ctf_global:

ctf_global.stop_ctf()

ctf, messages_ctf = setup_ctf()

ctf_global = ctf

previous_ctf_name = os.getenv("CTF_NAME", None)

ctf_init = 0

# Check if CAI_MAX_TURNS has been updated via /config

current_max_turns = os.getenv("CAI_MAX_TURNS", "inf")

if current_max_turns != str(prev_max_turns):

max_turns = float(current_max_turns)

prev_max_turns = max_turns

if turn_limit_reached and turn_count < max_turns:

turn_limit_reached = False

console.print(

"[green]Turn limit increased. You can now continue using CAI.[/green]"

)

# Check if max turns is reached

if turn_count >= max_turns and max_turns != float("inf"):

if not turn_limit_reached:

turn_limit_reached = True

console.print(

f"[bold red]Error: Maximum turn limit ({int(max_turns)}) reached.[/bold red]"

)

console.print(

"[yellow]You must increase the limit using the /config command: /config CAI_MAX_TURNS=<new_value>[/yellow]"

)

console.print(

"[yellow]Only CLI commands (starting with '/') will be processed until the limit is increased.[/yellow]"

)

try:

# Start measuring user idle time

start_idle_timer()

import time

idle_start_time = time.time()

# Check if model has changed and update if needed

current_model = os.getenv("CAI_MODEL", "alias1")

# Check for agent-specific model override

agent_specific_model = os.getenv(f"CAI_{last_agent_type.upper()}_MODEL")

if agent_specific_model:

current_model = agent_specific_model

if current_model != last_model and hasattr(agent, "model"):

# Update the model recursively for the agent and all handoff agents

update_agent_models_recursively(agent, current_model)

last_model = current_model

# Check if agent type has changed and recreate agent if needed

current_agent_type = os.getenv("CAI_AGENT_TYPE", "one_tool_agent")

# Update parallel_count to reflect changes from /parallel command

parallel_count = int(os.getenv("CAI_PARALLEL", "1"))

if current_agent_type != last_agent_type:

# Check if the /agent command already handled the switch

if os.environ.get("CAI_AGENT_SWITCH_HANDLED") == "1":

os.environ["CAI_AGENT_SWITCH_HANDLED"] = "0" # Reset flag

# Just get the existing agent that was already switched

from cai.sdk.agents.simple_agent_manager import AGENT_MANAGER

# First try to get the strong reference if available

if hasattr(AGENT_MANAGER, '_current_agent_strong_ref'):

agent = AGENT_MANAGER._current_agent_strong_ref

# Clear the strong reference after using it

delattr(AGENT_MANAGER, '_current_agent_strong_ref')

else:

agent = AGENT_MANAGER.get_active_agent()

if agent:

last_agent_type = current_agent_type

else:

# If the agent is None (weak reference expired), recreate it

agent = get_agent_by_name(current_agent_type, agent_id="P1")

last_agent_type = current_agent_type

# Re-register with AGENT_MANAGER

agent_name = agent.name if hasattr(agent, "name") else current_agent_type

AGENT_MANAGER.set_active_agent(agent, agent_name, "P1")

continue

try:

# CRITICAL: Set up history transfer BEFORE creating the new agent

from cai.sdk.agents.simple_agent_manager import AGENT_MANAGER

# Get the current agent's history before switching

if hasattr(agent, "name"):

current_agent_name = agent.name

current_history = AGENT_MANAGER.get_message_history(current_agent_name)

if current_history:

AGENT_MANAGER._pending_history_transfer = list(current_history) # Make a copy

# Now create the new agent

agent = get_agent_by_name(current_agent_type, agent_id="P1")

last_agent_type = current_agent_type

# Reset cost tracking for the new agent

from cai.util import COST_TRACKER

COST_TRACKER.reset_agent_costs()

# Use the new switch_to_single_agent method for proper cleanup

# IMPORTANT: Always use the agent's proper name, not the agent key

agent_name = agent.name if hasattr(agent, "name") else current_agent_type

# Check if this agent has already been switched to by /agent command

# If so, don't switch again to avoid clearing the history

current_active_agent = AGENT_MANAGER.get_active_agent()

current_active_name = AGENT_MANAGER._active_agent_name

if current_active_name == agent_name:

# Just update the agent reference

agent = current_active_agent

else:

# Switch to the new agent

AGENT_MANAGER.switch_to_single_agent(agent, agent_name)

# Sync the model's history with AGENT_MANAGER's history

# This ensures the model has its own history from AGENT_MANAGER

if hasattr(agent, "model") and hasattr(agent.model, "message_history"):

agent_history = AGENT_MANAGER.get_message_history(agent_name)

# Clear model's history and sync with AGENT_MANAGER

agent.model.message_history.clear()

if agent_history:

# Use extend() to avoid circular addition

agent.model.message_history.extend(agent_history)

# Configure the new agent's model flags

if hasattr(agent, "model"):

if hasattr(agent.model, "disable_rich_streaming"):

agent.model.disable_rich_streaming = (

False # Now False to let model handle streaming

)

if hasattr(agent.model, "suppress_final_output"):

agent.model.suppress_final_output = (

False # Changed to False to show all agent messages

)

# Apply current model to the new agent and all its handoff agents

# Check for agent-specific model override

agent_specific_model = os.getenv(f"CAI_{current_agent_type.upper()}_MODEL")

model_to_apply = (

agent_specific_model if agent_specific_model else current_model

)

update_agent_models_recursively(agent, model_to_apply)

last_model = model_to_apply

# Set agent name in the model for streaming display

if hasattr(agent.model, "set_agent_name"):

agent.model.set_agent_name(get_agent_short_name(agent))

# Clear any asyncio tasks that might be lingering from the previous agent

# This helps prevent event loop issues after agent switching

try:

# Get all running tasks

all_tasks = asyncio.all_tasks() if hasattr(asyncio, 'all_tasks') else asyncio.Task.all_tasks()

# Cancel tasks that aren't the current task

current_task = asyncio.current_task() if hasattr(asyncio, 'current_task') else asyncio.Task.current_task()

for task in all_tasks:

if task != current_task and not task.done():

task.cancel()

except RuntimeError:

# Not in an async context, which is fine

pass

except Exception as e:

# Log the error but don't display it unless in debug mode

logger = logging.getLogger(__name__)

logger.debug(f"Error switching agent: {str(e)}")

if os.getenv("CAI_DEBUG", "1") == "2":

console.print(f"[red]Error switching agent: {str(e)}[/red]")

if not force_until_flag and ctf_init != 0:

# Use initial prompt on first iteration if provided

if use_initial_prompt:

user_input = initial_prompt

use_initial_prompt = False # Only use it once

else:

# Get user input with command completion and history

user_input = get_user_input(

command_completer, kb, history_file, get_toolbar_with_refresh, current_text

)

else:

user_input = messages_ctf

ctf_init = 1

idle_time += time.time() - idle_start_time

# Stop measuring user idle time and start measuring active time

stop_idle_timer()

start_active_timer()

if not user_input.strip():

user_input = "User input is empty, maybe wants to continue" # Set a default message to continue the conversation

# In parallel mode, all configured agents will run automatically

# No agent selection menu - just run all agents

except KeyboardInterrupt:

# Print newline to ensure clean prompt display after interrupt

print()

def format_time(seconds):

mins, secs = divmod(int(seconds), 60)

hours, mins = divmod(mins, 60)

return f"{hours:02d}:{mins:02d}:{secs:02d}"

Total = time.time() - START_TIME

idle_time += time.time() - idle_start_time

# Save parallel agents' histories if we were in parallel mode

try:

if PARALLEL_CONFIGS and PARALLEL_ISOLATION.is_parallel_mode():

# Save each parallel agent's history

saved_count = 0

for idx, config in enumerate(PARALLEL_CONFIGS, 1):

instance_key = (config.agent_name, idx)

if instance_key in PARALLEL_AGENT_INSTANCES:

instance_agent = PARALLEL_AGENT_INSTANCES[instance_key]

if hasattr(instance_agent, 'model') and hasattr(instance_agent.model, 'message_history'):

# The agent's message history should already be updated in PARALLEL_ISOLATION

# via the add_to_message_history method, but let's make sure

agent_id = config.id or f"P{idx}"

if instance_agent.model.message_history:

PARALLEL_ISOLATION.replace_isolated_history(agent_id, instance_agent.model.message_history)

saved_count += 1

if saved_count > 0:

# Sync isolated histories with AGENT_MANAGER for display

for idx, config in enumerate(PARALLEL_CONFIGS, 1):

agent_id = config.id or f"P{idx}"

isolated_history = PARALLEL_ISOLATION.get_isolated_history(agent_id)

if isolated_history:

# Get the agent display name

from cai.agents import get_available_agents

available_agents = get_available_agents()

if config.agent_name in available_agents:

agent = available_agents[config.agent_name]

agent_display_name = getattr(agent, "name", config.agent_name)

# Add instance number if needed

total_count = sum(1 for c in PARALLEL_CONFIGS if c.agent_name == config.agent_name)

if total_count > 1:

instance_num = 0

for c in PARALLEL_CONFIGS:

if c.agent_name == config.agent_name:

instance_num += 1

if c.id == config.id:

break

agent_display_name = f"{agent_display_name} #{instance_num}"

# Clear and replace the history in AGENT_MANAGER

AGENT_MANAGER.clear_history(agent_display_name)

for msg in isolated_history:

AGENT_MANAGER.add_to_history(agent_display_name, msg)

except Exception as e:

# Only log this error in debug mode

logger = logging.getLogger(__name__)

logger.debug(f"Error saving parallel agents' histories: {str(e)}")

# Clean up any pending tool calls before exiting

try:

# Access the converter directly to clean up any pending tool calls

# converter is instance-based, access via agent.model._converter

# Check if any tool calls are pending (have been issued but don't have responses)

pending_calls = []

if hasattr(agent.model, "_converter") and hasattr(

agent.model._converter, "recent_tool_calls"

):

for call_id, call_info in list(

agent.model._converter.recent_tool_calls.items()

):

# Check if this tool call has a corresponding response in message_history

tool_response_exists = False

for msg in agent.model.message_history:

if msg.get("role") == "tool" and msg.get("tool_call_id") == call_id:

tool_response_exists = True

break

# If no tool response exists, create a synthetic one

if not tool_response_exists:

# First ensure there's a matching assistant message with this tool call

assistant_exists = False

for msg in agent.model.message_history:

if (

msg.get("role") == "assistant"

and msg.get("tool_calls")

and any(

tc.get("id") == call_id for tc in msg.get("tool_calls", [])

)

):

assistant_exists = True

break

# Add assistant message if needed

if not assistant_exists:

tool_call_msg = {

"role": "assistant",

"content": None,

"tool_calls": [

{

"id": call_id,

"type": "function",

"function": {

"name": call_info.get("name", "unknown_function"),

"arguments": call_info.get("arguments", "{}"),

},

}

],

}

agent.model.add_to_message_history(tool_call_msg)

# Add a synthetic tool response

tool_msg = {

"role": "tool",

"tool_call_id": call_id,

"content": "Operation interrupted by user (Keyboard Interrupt during shutdown)",

}

agent.model.add_to_message_history(tool_msg)

pending_calls.append(call_info.get("name", "unknown"))

# Apply message list fixes to ensure consistency

if pending_calls:

from cai.util import fix_message_list

agent.model.message_history[:] = fix_message_list(agent.model.message_history)

except Exception:

pass

try:

# Get more accurate active and idle time measurements from the timer functions

from cai.util import COST_TRACKER, get_active_time_seconds, get_idle_time_seconds

# Use the precise measurements from our timers

active_time_seconds = get_active_time_seconds()

idle_time_seconds = get_idle_time_seconds()

# Format for display

active_time_formatted = format_time(active_time_seconds)

idle_time_formatted = format_time(idle_time_seconds)

# Get session cost from the global cost tracker

session_cost = COST_TRACKER.session_total_cost

metrics = {

"session_time": format_time(Total),

"active_time": active_time_formatted,

"idle_time": idle_time_formatted,

"llm_time": format_time(active_time_seconds), # Using active time as LLM time

"llm_percentage": round((active_time_seconds / Total) * 100, 1)

if Total > 0

else 0.0,

"session_cost": f"${session_cost:.6f}", # Add formatted session cost

}

logging_path = (

session_logger.filename if hasattr(session_logger, "filename") else None

)

content = []

content.append(f"Session Time: {metrics['session_time']}")

content.append(

f"Active Time: {metrics['active_time']} ({metrics['llm_percentage']}%)"

)

content.append(f"Idle Time: {metrics['idle_time']}")

content.append(

f"Total Session Cost: {metrics['session_cost']}"

) # Add cost to display

if logging_path:

content.append(f"Log available at: {logging_path}")

def print_session_summary(console, metrics, logging_path=None):

"""

Print a session summary panel using Rich.

"""

from rich.box import ROUNDED

from rich.console import Group

from rich.panel import Panel

from rich.text import Text

# Create Rich Text objects for each line

text_content = []

for i, line in enumerate(content):

if "Total Session Cost" in line:

# Format cost line with special styling

cost_text = Text()

parts = line.split(":")

cost_text.append(parts[0] + ":", style="bold")

cost_text.append(parts[1], style="bold green")

text_content.append(cost_text)

else:

text_content.append(Text(line))

time_panel = Panel(

Group(*text_content),

border_style="blue",

box=ROUNDED,

padding=(0, 1),

title="[bold]Session Summary[/bold]",

title_align="left",

)

console.print(time_panel, end="")

print_session_summary(console, metrics, logging_path)

# Upload logs if telemetry is enabled by checking the

# env. variable CAI_TELEMETRY and there's internet connectivity

telemetry_enabled = os.getenv("CAI_TELEMETRY", "true").lower() != "false"

if (

telemetry_enabled

and hasattr(session_logger, "session_id")

and hasattr(session_logger, "filename")

):

process_metrics(

session_logger.filename, # should match logging_path

sid=session_logger.session_id,

)

# Log session end

if session_logger:

session_logger.log_session_end()

# End global usage tracking session

GLOBAL_USAGE_TRACKER.end_session(final_cost=COST_TRACKER.session_total_cost)

# Create symlink to the last log file

if session_logger and hasattr(session_logger, "filename"):

create_last_log_symlink(session_logger.filename)

# Prevent duplicate cost display from the COST_TRACKER exit handler

os.environ["CAI_COST_DISPLAYED"] = "true"

if is_pentestperf_available() and os.getenv("CTF_NAME", None):

ctf.stop_ctf()

except Exception:

pass

break

try:

# Check if turn limit is reached and allow only CLI commands

if (

turn_limit_reached

and not user_input.startswith("/")

and not user_input.startswith("$")

):

console.print(

"[bold red]Error: Turn limit reached. Only CLI commands are allowed.[/bold red]"

)

console.print(

"[yellow]Please use /config to increase CAI_MAX_TURNS limit.[/yellow]"

)

# Skip processing this input but continue the main loop

stop_active_timer()

start_idle_timer()

continue

# Check if we have parallel configurations to run

if (

PARALLEL_CONFIGS

and not user_input.startswith("/")

and not user_input.startswith("$")

):

# Use parallel configurations instead of normal processing

# Show which agents have custom prompts

agents_with_prompts = [(idx, config) for idx, config in enumerate(PARALLEL_CONFIGS, 1) if config.prompt]

# First ensure ALL parallel configs have agent instances (not just selected ones)

# This prevents agents from disappearing from history when not selected

from cai.agents import get_available_agents

# Setup parallel isolation for these agents

from cai.sdk.agents.parallel_isolation import PARALLEL_ISOLATION

# Get agent IDs

agent_ids = [config.id or f"P{idx}" for idx, config in enumerate(PARALLEL_CONFIGS, 1)]

# Check if we already have isolated histories (e.g., from /load parallel)

# If not, transfer the current agent's history to all parallel agents

already_has_histories = False

if PARALLEL_ISOLATION.is_parallel_mode():

# Check if at least one agent has a non-empty isolated history

for agent_id in agent_ids:

isolated_history = PARALLEL_ISOLATION.get_isolated_history(agent_id)