github/OpenHands
1
0
Fork 0
mirror of https://github.com/OpenHands/OpenHands.git synced 2025-12-26 13:52:43 +08:00
Code Issues Packages Projects Releases Wiki Activity

Refactor(V1): OpenHands CLI + Agent SDK (#10905)

Browse Source 
Co-authored-by: openhands <openhands@all-hands.dev>
This commit is contained in:
Rohit Malhotra 2025-09-10 09:51:55 -04:00 committed by GitHub
parent b08238c841
commit cf982e0134
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
35 changed files with 4484 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

83
.github/workflows/cli-build-test.yml vendored Normal file
View File

@ -0,0 +1,83 @@
# Workflow that builds and tests the CLI binary executable
name: CLI - Build and Test Binary
# Run on pushes to main branch and all pull requests, but only when CLI files change
on:
push:
branches:
- main
paths:
- 'openhands-cli/**'
pull_request:
paths:
- 'openhands-cli/**'
# Cancel previous runs if a new commit is pushed
concurrency:
group: ${{ github.workflow }}-${{ (github.head_ref && github.ref) || github.run_id }}
cancel-in-progress: true
jobs:
build-and-test-binary:
name: Build and test binary executable
runs-on: ubuntu-latest
steps:
- name: Checkout repository
uses: actions/checkout@v4
with:
fetch-depth: 0
- name: Set up Python
uses: actions/setup-python@v5
with:
python-version: 3.12
- name: Install uv
uses: astral-sh/setup-uv@v3
with:
version: "latest"
- name: Install dependencies
working-directory: openhands-cli
run: |
uv sync --dev
- name: Build binary executable
working-directory: openhands-cli
run: |
./build.sh --install-pyinstaller --no-test
- name: Test binary startup and /help command
working-directory: openhands-cli
env:
LITELLM_API_KEY: dummy-ci-key
LITELLM_MODEL: dummy-ci-model
run: |
# Test that binary starts and responds to /help command
echo "Testing binary startup and /help command..."
# Send /help command and then exit
echo -e "/help\n/exit" | timeout 30s ./dist/openhands-cli 2>&1 | tee output.log || true
# Check that the application started successfully
if grep -q "OpenHands CLI Help" output.log; then
echo "✅ Application started and /help command works correctly"
else
echo "❌ /help command output not found"
echo "Full output:"
cat output.log
exit 1
fi
# Check for expected help content
if grep -q "Available commands:" output.log && grep -q "/exit - Exit the application" output.log; then
echo "✅ Help content is correct"
else
echo "❌ Expected help content not found"
echo "Full output:"
cat output.log
exit 1
fi
echo "✅ Binary test completed successfully"

20
.github/workflows/lint.yml vendored
View File

@ -37,7 +37,7 @@ jobs:
npm run make-i18n && tsc
npm run check-translation-completeness
# Run lint on the python code
# Run lint on the python code (excluding CLI and enterprise)
lint-python:
name: Lint python
runs-on: blacksmith-4vcpu-ubuntu-2204
@ -73,6 +73,24 @@ jobs:
working-directory: ./enterprise
run: pre-commit run --all-files --config ./dev_config/python/.pre-commit-config.yaml
lint-cli-python:
name: Lint CLI python
runs-on: blacksmith-4vcpu-ubuntu-2204
steps:
- uses: actions/checkout@v4
with:
fetch-depth: 0
- name: Set up python
uses: useblacksmith/setup-python@v6
with:
python-version: 3.12
cache: "pip"
- name: Install pre-commit
run: pip install pre-commit==3.7.0
- name: Run pre-commit hooks
working-directory: ./openhands-cli
run: pre-commit run --all-files --config ../dev_config/python/.pre-commit-config.yaml
# Check version consistency across documentation
check-version-consistency:
name: Check version consistency

30
.github/workflows/py-tests.yml vendored
View File

@ -104,3 +104,33 @@ jobs:
- name: Run Unit Tests
working-directory: ./enterprise
run: PYTHONPATH=".:$PYTHONPATH" poetry run pytest --forked -n auto -svv -p no:ddtrace -p no:ddtrace.pytest_bdd -p no:ddtrace.pytest_benchmark ./tests/unit
# Run CLI unit tests
test-cli-python:
name: CLI Unit Tests
runs-on: blacksmith-4vcpu-ubuntu-2204
steps:
- name: Checkout repository
uses: actions/checkout@v4
with:
fetch-depth: 0
- name: Set up Python
uses: useblacksmith/setup-python@v6
with:
python-version: 3.12
- name: Install uv
uses: astral-sh/setup-uv@v3
with:
version: "latest"
- name: Install dependencies
working-directory: ./openhands-cli
run: |
uv sync --extra dev
- name: Run CLI unit tests
working-directory: ./openhands-cli
run: |
uv run pytest -v

3
.gitignore vendored
View File

@ -31,7 +31,8 @@ requirements.txt
# Usually these files are written by a python script from a template
# before PyInstaller builds the exe, so as to inject date/other infos into it.
*.manifest
*.spec
# Note: openhands-cli.spec is intentionally tracked for CLI builds
# *.spec
# Installer logs
pip-log.txt

8
dev_config/python/.pre-commit-config.yaml
View File

@ -3,9 +3,9 @@ repos:
rev: v5.0.0
hooks:
- id: trailing-whitespace
exclude: ^(docs/|modules/|python/|openhands-ui/|third_party/|enterprise/)
exclude: ^(docs/|modules/|python/|openhands-ui/|third_party/|enterprise/|openhands-cli/)
- id: end-of-file-fixer
exclude: ^(docs/|modules/|python/|openhands-ui/|third_party/|enterprise/)
exclude: ^(docs/|modules/|python/|openhands-ui/|third_party/|enterprise/|openhands-cli/)
- id: check-yaml
args: ["--allow-multiple-documents"]
- id: debug-statements
@ -28,12 +28,12 @@ repos:
entry: ruff check --config dev_config/python/ruff.toml
types_or: [python, pyi, jupyter]
args: [--fix, --unsafe-fixes]
exclude: ^(third_party/|enterprise/)
exclude: ^(third_party/|enterprise/|openhands-cli/)
# Run the formatter.
- id: ruff-format
entry: ruff format --config dev_config/python/ruff.toml
types_or: [python, pyi, jupyter]
exclude: ^(third_party/|enterprise/)
exclude: ^(third_party/|enterprise/|openhands-cli/)
- repo: https://github.com/pre-commit/mirrors-mypy
rev: v1.15.0

52
openhands-cli/.gitignore vendored Normal file
View File

@ -0,0 +1,52 @@
# Python
__pycache__/
*.py[cod]
*$py.class
*.so
.Python
build/
develop-eggs/
dist/
downloads/
eggs/
.eggs/
lib/
lib64/
parts/
sdist/
var/
wheels/
*.egg-info/
.installed.cfg
*.egg
# Virtual Environment
.env
.venv
env/
venv/
ENV/
# IDE
.idea/
.vscode/
*.swp
*.swo
# Testing
.pytest_cache/
.coverage
htmlcov/
.tox/
.nox/
.coverage.*
coverage.xml
*.cover
.hypothesis/
# PyInstaller
# Usually these files are written by a python script from a template
# before PyInstaller builds the exe, so as to inject date/other infos into it.
*.manifest
# Note: We keep our custom spec file in version control
# *.spec

46
openhands-cli/Makefile Normal file
View File

@ -0,0 +1,46 @@
.PHONY: help install install-dev test format clean run
# Default target
help:
@echo "OpenHands CLI - Available commands:"
@echo " install - Install the package"
@echo " install-dev - Install with development dependencies"
@echo " test - Run tests"
@echo " format - Format code with ruff"
@echo " clean - Clean build artifacts"
@echo " run - Run the CLI"
# Install the package
install:
uv sync
# Install with development dependencies
install-dev:
uv sync --extra dev
# Run tests
test:
uv run pytest
# Format code
format:
uv run ruff format openhands_cli/
# Clean build artifacts
clean:
rm -rf .venv/
find . -type d -name "__pycache__" -exec rm -rf {} +
find . -type f -name "*.pyc" -delete
# Run the CLI
run:
uv run openhands-cli
# Install UV if not present
install-uv:
@if ! command -v uv &> /dev/null; then \
echo "Installing UV..."; \
curl -LsSf https://astral.sh/uv/install.sh | sh; \
else \
echo "UV is already installed"; \
fi

45
openhands-cli/README.md Normal file
View File

@ -0,0 +1,45 @@
# OpenHands CLI
A lightweight CLI/TUI to interact with the OpenHands agent (powered by agent-sdk). Build and run locally or as a single executable.
## Quickstart
- Prerequisites: Python 3.12+, curl
- Install uv (package manager):
```bash
curl -LsSf https://astral.sh/uv/install.sh | sh
# Restart your shell so "uv" is on PATH, or follow the installer hint
```
### Run the CLI locally
```bash
# Install dependencies (incl. dev tools)
make install-dev
# Optional: install pre-commit hooks
make install-pre-commit-hooks
# Start the CLI
make run
# or
uv run openhands-cli
```
Tip: Set your model key (one of) so the agent can talk to an LLM:
```bash
export OPENAI_API_KEY=...
# or
export LITELLM_API_KEY=...
```
### Build a standalone executable
```bash
# Build (installs PyInstaller if needed)
./build.sh --install-pyinstaller
# The binary will be in dist/
./dist/openhands-cli # macOS/Linux
# dist/openhands-cli.exe # Windows
```
For advanced development (adding deps, updating the spec file, debugging builds), see Development.md.

195
openhands-cli/build.py Executable file
View File

@ -0,0 +1,195 @@
#!/usr/bin/env python3
"""
Build script for OpenHands CLI using PyInstaller.
This script packages the OpenHands CLI into a standalone executable binary
using PyInstaller with the custom spec file.
"""
import argparse
import os
import shutil
import subprocess
import sys
from pathlib import Path
def clean_build_directories() -> None:
"""Clean up previous build artifacts."""
print('🧹 Cleaning up previous build artifacts...')
build_dirs = ['build', 'dist', '__pycache__']
for dir_name in build_dirs:
if os.path.exists(dir_name):
print(f' Removing {dir_name}/')
shutil.rmtree(dir_name)
# Clean up .pyc files
for root, _dirs, files in os.walk('.'):
for file in files:
if file.endswith('.pyc'):
os.remove(os.path.join(root, file))
print('✅ Cleanup complete!')
def check_pyinstaller() -> bool:
"""Check if PyInstaller is available."""
try:
subprocess.run(
['uv', 'run', 'pyinstaller', '--version'], check=True, capture_output=True
)
return True
except (subprocess.CalledProcessError, FileNotFoundError):
print(
'❌ PyInstaller is not available. Use --install-pyinstaller flag or install manually with:'
)
print(' uv add --dev pyinstaller')
return False
def build_executable(
spec_file: str = 'openhands-cli.spec',
clean: bool = True,
install_pyinstaller: bool = False,
) -> bool:
"""Build the executable using PyInstaller."""
if clean:
clean_build_directories()
# Check if PyInstaller is available (installation is handled by build.sh)
if not check_pyinstaller():
return False
print(f'🔨 Building executable using {spec_file}...')
try:
# Run PyInstaller with uv
cmd = ['uv', 'run', 'pyinstaller', spec_file, '--clean']
print(f'Running: {" ".join(cmd)}')
subprocess.run(cmd, check=True, capture_output=True, text=True)
print('✅ Build completed successfully!')
# Check if the executable was created
dist_dir = Path('dist')
if dist_dir.exists():
executables = list(dist_dir.glob('*'))
if executables:
print('📁 Executable(s) created in dist/:')
for exe in executables:
size = exe.stat().st_size / (1024 * 1024) # Size in MB
print(f' - {exe.name} ({size:.1f} MB)')
else:
print('⚠️ No executables found in dist/ directory')
return True
except subprocess.CalledProcessError as e:
print(f'❌ Build failed: {e}')
if e.stdout:
print('STDOUT:', e.stdout)
if e.stderr:
print('STDERR:', e.stderr)
return False
def test_executable() -> bool:
"""Test the built executable with simplified checks."""
print('🧪 Testing the built executable...')
exe_path = Path('dist/openhands-cli')
if not exe_path.exists():
# Try with .exe extension for Windows
exe_path = Path('dist/openhands-cli.exe')
if not exe_path.exists():
print('❌ Executable not found!')
return False
try:
# Make executable on Unix-like systems
if os.name != 'nt':
os.chmod(exe_path, 0o755)
# Simple test: Check that executable can start and respond to /help command
print(' Testing executable startup and /help command...')
result = subprocess.run(
[str(exe_path)],
capture_output=True,
text=True,
timeout=15,
input='/help\n/exit\n', # Send /help command then exit
env={
**os.environ,
'LITELLM_API_KEY': 'dummy-test-key',
'LITELLM_MODEL': 'dummy-model',
},
)
# Check for expected help output
output = result.stdout + result.stderr
if 'OpenHands CLI Help' in output and 'Available commands:' in output:
print(' ✅ Executable starts and /help command works correctly')
return True
else:
print(' ❌ Expected help output not found')
print(' Combined output:', output[:1000])
return False
except subprocess.TimeoutExpired:
print(' ❌ Executable test timed out')
return False
except Exception as e:
print(f'❌ Error testing executable: {e}')
return False
def main() -> int:
"""Main function."""
parser = argparse.ArgumentParser(description='Build OpenHands CLI executable')
parser.add_argument(
'--spec', default='openhands-cli.spec', help='PyInstaller spec file to use'
)
parser.add_argument(
'--no-clean', action='store_true', help='Skip cleaning build directories'
)
parser.add_argument(
'--no-test', action='store_true', help='Skip testing the built executable'
)
parser.add_argument(
'--install-pyinstaller',
action='store_true',
help='Install PyInstaller using uv before building',
)
args = parser.parse_args()
print('🚀 OpenHands CLI Build Script')
print('=' * 40)
# Check if spec file exists
if not os.path.exists(args.spec):
print(f"❌ Spec file '{args.spec}' not found!")
return 1
# Build the executable
if not build_executable(
args.spec, clean=not args.no_clean, install_pyinstaller=args.install_pyinstaller
):
return 1
# Test the executable
if not args.no_test:
if not test_executable():
print('❌ Executable test failed, build process failed')
return 1
print('\n🎉 Build process completed!')
print("📁 Check the 'dist/' directory for your executable")
return 0
if __name__ == '__main__':
sys.exit(main())

48
openhands-cli/build.sh Executable file
View File

@ -0,0 +1,48 @@
#!/bin/bash
#
# Shell script wrapper for building OpenHands CLI executable.
#
# This script provides a simple interface to build the OpenHands CLI
# using PyInstaller with uv package management.
#
set -e # Exit on any error
echo "🚀 OpenHands CLI Build Script"
echo "=============================="
# Check if uv is available
if ! command -v uv &> /dev/null; then
echo "❌ uv is required but not found! Please install uv first."
exit 1
fi
# Parse arguments to check for --install-pyinstaller
INSTALL_PYINSTALLER=false
PYTHON_ARGS=()
for arg in "$@"; do
case $arg in
--install-pyinstaller)
INSTALL_PYINSTALLER=true
PYTHON_ARGS+=("$arg")
;;
*)
PYTHON_ARGS+=("$arg")
;;
esac
done
# Install PyInstaller if requested
if [ "$INSTALL_PYINSTALLER" = true ]; then
echo "📦 Installing PyInstaller with uv..."
if uv add --dev pyinstaller; then
echo "✅ PyInstaller installed successfully with uv!"
else
echo "❌ Failed to install PyInstaller"
exit 1
fi
fi
# Run the Python build script using uv
uv run python build.py "${PYTHON_ARGS[@]}"

102
openhands-cli/openhands-cli.spec Normal file
View File

@ -0,0 +1,102 @@
# -*- mode: python ; coding: utf-8 -*-
"""
PyInstaller spec file for OpenHands CLI.
This spec file configures PyInstaller to create a standalone executable
for the OpenHands CLI application.
"""
from pathlib import Path
import os
import sys
from PyInstaller.utils.hooks import (
collect_submodules,
collect_data_files,
copy_metadata
)
# Get the project root directory (current working directory when running PyInstaller)
project_root = Path.cwd()
a = Analysis(
['openhands_cli/simple_main.py'],
pathex=[str(project_root)],
binaries=[],
datas=[
# Include any data files that might be needed
# Add more data files here if needed in the future
*collect_data_files('tiktoken'),
*collect_data_files('tiktoken_ext'),
*collect_data_files('litellm'),
*collect_data_files('fastmcp'),
*collect_data_files('mcp'),
# Include Jinja prompt templates required by the agent SDK
*collect_data_files('openhands.sdk.agent.agent', includes=['prompts/*.j2']),
# Include package metadata for importlib.metadata
*copy_metadata('fastmcp'),
],
hiddenimports=[
# Explicitly include modules that might not be detected automatically
*collect_submodules('openhands_cli'),
*collect_submodules('prompt_toolkit'),
# Include OpenHands SDK submodules explicitly to avoid resolution issues
*collect_submodules('openhands.sdk'),
*collect_submodules('openhands.tools'),
*collect_submodules('tiktoken'),
*collect_submodules('tiktoken_ext'),
*collect_submodules('litellm'),
*collect_submodules('fastmcp'),
# Include mcp but exclude CLI parts that require typer
'mcp.types',
'mcp.client',
'mcp.server',
'mcp.shared',
],
hookspath=[],
hooksconfig={},
runtime_hooks=[],
excludes=[
# Exclude unnecessary modules to reduce binary size
'tkinter',
'matplotlib',
'numpy',
'scipy',
'pandas',
'PIL',
'IPython',
'jupyter',
'notebook',
# Exclude mcp CLI parts that cause issues
'mcp.cli',
'mcp.cli.cli',
],
noarchive=False,
# IMPORTANT: do not use optimize=2 (-OO) because it strips docstrings used by PLY/bashlex grammar
optimize=0,
)
pyz = PYZ(a.pure)
exe = EXE(
pyz,
a.scripts,
a.binaries,
a.datas,
[],
name='openhands-cli',
debug=False,
bootloader_ignore_signals=False,
strip=True, # Strip debug symbols to reduce size
upx=True, # Use UPX compression if available
upx_exclude=[],
runtime_tmpdir=None,
console=True, # CLI application needs console
disable_windowed_traceback=False,
argv_emulation=False,
target_arch=None,
codesign_identity=None,
entitlements_file=None,
icon=None, # Add icon path here if you have one
)

3
openhands-cli/openhands_cli/__init__.py Normal file
View File

@ -0,0 +1,3 @@
"""OpenHands CLI package."""
__version__ = '0.1.0'

128
openhands-cli/openhands_cli/agent_chat.py Normal file
View File

@ -0,0 +1,128 @@
#!/usr/bin/env python3
"""
Agent chat functionality for OpenHands CLI.
Provides a conversation interface with an AI agent using OpenHands patterns.
"""
import logging
from prompt_toolkit import PromptSession, print_formatted_text
from prompt_toolkit.formatted_text import HTML
from openhands.sdk import (
Message,
TextContent,
)
from openhands_cli.runner import ConversationRunner
from openhands_cli.setup import setup_agent
from openhands_cli.tui import (
CommandCompleter,
display_help,
display_welcome,
)
from openhands_cli.user_actions import UserConfirmation, exit_session_confirmation
logger = logging.getLogger(__name__)
def run_cli_entry() -> None:
"""Run the agent chat session using the agent SDK.
Raises:
AgentSetupError: If agent setup fails
KeyboardInterrupt: If user interrupts the session
EOFError: If EOF is encountered
"""
# Setup agent - let exceptions bubble up
conversation = setup_agent()
# Generate session ID
import uuid
session_id = str(uuid.uuid4())[:8]
display_welcome(session_id)
# Create prompt session with command completer
session = PromptSession(completer=CommandCompleter())
# Create conversation runner to handle state machine logic
runner = ConversationRunner(conversation)
# Main chat loop
while True:
try:
# Get user input
user_input = session.prompt(
HTML('<gold>> </gold>'),
multiline=False,
)
if not user_input.strip():
continue
# Handle commands
command = user_input.strip().lower()
message = Message(
role='user',
content=[TextContent(text=user_input)],
)
if command == '/exit':
exit_confirmation = exit_session_confirmation()
if exit_confirmation == UserConfirmation.ACCEPT:
print_formatted_text(HTML('\n<yellow>Goodbye! 👋</yellow>'))
break
elif command == '/clear':
display_welcome(session_id)
continue
elif command == '/help':
display_help()
continue
elif command == '/status':
print_formatted_text(HTML(f'<grey>Session ID: {session_id}</grey>'))
print_formatted_text(HTML('<grey>Status: Active</grey>'))
confirmation_status = (
'enabled' if conversation.state.confirmation_mode else 'disabled'
)
print_formatted_text(
HTML(f'<grey>Confirmation mode: {confirmation_status}</grey>')
)
continue
elif command == '/confirm':
current_mode = runner.confirmation_mode
runner.set_confirmation_mode(not current_mode)
new_status = 'enabled' if not current_mode else 'disabled'
print_formatted_text(
HTML(f'<yellow>Confirmation mode {new_status}</yellow>')
)
continue
elif command == '/new':
print_formatted_text(
HTML('<yellow>Starting new conversation...</yellow>')
)
session_id = str(uuid.uuid4())[:8]
display_welcome(session_id)
continue
elif command == '/resume':
if not conversation.state.agent_paused:
print_formatted_text(
HTML('<red>No paused conversation to resume...</red>')
)
continue
# Resume without new message
message = None
runner.process_message(message)
print() # Add spacing
except KeyboardInterrupt:
exit_confirmation = exit_session_confirmation()
if exit_confirmation == UserConfirmation.ACCEPT:
print_formatted_text(HTML('\n<yellow>Goodbye! 👋</yellow>'))
break
continue

5
openhands-cli/openhands_cli/listeners/__init__.py Normal file
View File

@ -0,0 +1,5 @@
from openhands_cli.listeners.pause_listener import PauseListener
__all__ = [
'PauseListener',
]

83
openhands-cli/openhands_cli/listeners/pause_listener.py Normal file
View File

@ -0,0 +1,83 @@
import threading
from collections.abc import Callable, Iterator
from contextlib import contextmanager
from prompt_toolkit import HTML, print_formatted_text
from prompt_toolkit.input import Input, create_input
from prompt_toolkit.keys import Keys
from openhands.sdk import Conversation
class PauseListener(threading.Thread):
"""Background key listener that triggers pause on Ctrl-P.
Starts and stops around agent run() loops to avoid interfering with user prompts.
"""
def __init__(
self,
on_pause: Callable,
input_source: Input | None = None, # used to pipe inputs for unit tests
):
super().__init__(daemon=True)
self.on_pause = on_pause
self._stop_event = threading.Event()
self._pause_event = threading.Event()
self._input = input_source or create_input()
def _detect_pause_key_presses(self) -> bool:
pause_detected = False
for key_press in self._input.read_keys():
pause_detected = pause_detected or key_press.key == Keys.ControlP
pause_detected = pause_detected or key_press.key == Keys.ControlC
pause_detected = pause_detected or key_press.key == Keys.ControlD
return pause_detected
def _execute_pause(self) -> None:
self._pause_event.set() # Mark pause event occurred
print_formatted_text(HTML(''))
print_formatted_text(
HTML('<gold>Pausing agent once step is completed...</gold>')
)
try:
self.on_pause()
except Exception:
pass
def run(self) -> None:
try:
with self._input.raw_mode():
# User hasn't paused and pause listener hasn't been shut down
while not (self.is_paused() or self.is_stopped()):
if self._detect_pause_key_presses():
self._execute_pause()
finally:
try:
self._input.close()
except Exception:
pass
def stop(self) -> None:
self._stop_event.set()
def is_stopped(self) -> bool:
return self._stop_event.is_set()
def is_paused(self) -> bool:
return self._pause_event.is_set()
@contextmanager
def pause_listener(
conversation: Conversation, input_source: Input | None = None
) -> Iterator[PauseListener]:
"""Ensure PauseListener always starts/stops cleanly."""
listener = PauseListener(on_pause=conversation.pause, input_source=input_source)
listener.start()
try:
yield listener
finally:
listener.stop()

29
openhands-cli/openhands_cli/pt_style.py Normal file
View File

@ -0,0 +1,29 @@
from prompt_toolkit.styles import Style, merge_styles
from prompt_toolkit.styles.base import BaseStyle
from prompt_toolkit.styles.defaults import default_ui_style
# Centralized helper for CLI styles so we can safely merge our custom colors
# with prompt_toolkit's default UI style. This preserves completion menu and
# fuzzy-match visibility across different terminal themes (e.g., Ubuntu).
COLOR_GOLD = '#FFD700'
COLOR_GREY = '#808080'
COLOR_AGENT_BLUE = '#4682B4' # Steel blue - readable on light/dark backgrounds
def get_cli_style() -> BaseStyle:
base = default_ui_style()
custom = Style.from_dict(
{
'gold': COLOR_GOLD,
'grey': COLOR_GREY,
'prompt': f'{COLOR_GOLD} bold',
# Ensure good contrast for fuzzy matches on the selected completion row
# across terminals/themes (e.g., Ubuntu GNOME, Alacritty, Kitty).
# See https://github.com/All-Hands-AI/OpenHands/issues/10330
'completion-menu.completion.current fuzzymatch.outside': 'fg:#ffffff bg:#888888',
'selected': COLOR_GOLD,
'risk-high': '#FF0000 bold', # Red bold for HIGH risk
}
)
return merge_styles([base, custom])

117
openhands-cli/openhands_cli/runner.py Normal file
View File

@ -0,0 +1,117 @@
from prompt_toolkit import HTML, print_formatted_text
from openhands.sdk import Conversation, Message
from openhands.sdk.event.utils import get_unmatched_actions
from openhands_cli.listeners.pause_listener import PauseListener, pause_listener
from openhands_cli.user_actions import ask_user_confirmation
from openhands_cli.user_actions.types import UserConfirmation
class ConversationRunner:
"""Handles the conversation state machine logic cleanly."""
def __init__(self, conversation: Conversation):
self.conversation = conversation
self.confirmation_mode = False
def set_confirmation_mode(self, confirmation_mode: bool) -> None:
self.confirmation_mode = confirmation_mode
self.conversation.set_confirmation_mode(confirmation_mode)
def _start_listener(self) -> None:
self.listener = PauseListener(on_pause=self.conversation.pause)
self.listener.start()
def _print_run_status(self) -> None:
print_formatted_text('')
if self.conversation.state.agent_paused:
print_formatted_text(
HTML(
'<yellow>Resuming paused conversation...</yellow><grey> (Press Ctrl-P to pause)</grey>'
)
)
else:
print_formatted_text(
HTML(
'<yellow>Agent running...</yellow><grey> (Press Ctrl-P to pause)</grey>'
)
)
print_formatted_text('')
def process_message(self, message: Message | None) -> None:
"""Process a user message through the conversation.
Args:
message: The user message to process
"""
self._print_run_status()
# Send message to conversation
if message:
self.conversation.send_message(message)
if self.confirmation_mode:
self._run_with_confirmation()
else:
self._run_without_confirmation()
def _run_without_confirmation(self) -> None:
with pause_listener(self.conversation):
self.conversation.run()
def _run_with_confirmation(self) -> None:
# If agent was paused, resume with confirmation request
if self.conversation.state.agent_waiting_for_confirmation:
user_confirmation = self._handle_confirmation_request()
if user_confirmation == UserConfirmation.DEFER:
return
while True:
with pause_listener(self.conversation) as listener:
self.conversation.run()
if listener.is_paused():
break
# In confirmation mode, agent either finishes or waits for user confirmation
if self.conversation.state.agent_finished:
break
elif self.conversation.state.agent_waiting_for_confirmation:
user_confirmation = self._handle_confirmation_request()
if user_confirmation == UserConfirmation.DEFER:
return
else:
raise Exception('Infinite loop')
def _handle_confirmation_request(self) -> UserConfirmation:
"""Handle confirmation request from user.
Returns:
UserConfirmation indicating the user's choice
"""
pending_actions = get_unmatched_actions(self.conversation.state.events)
if pending_actions:
user_confirmation, reason = ask_user_confirmation(pending_actions)
if user_confirmation == UserConfirmation.REJECT:
self.conversation.reject_pending_actions(
reason or 'User rejected the actions'
)
elif user_confirmation == UserConfirmation.DEFER:
self.conversation.pause()
elif user_confirmation == UserConfirmation.ALWAYS_ACCEPT:
# Disable confirmation mode when user selects "Always proceed"
print_formatted_text(
HTML(
'<yellow>Confirmation mode disabled. Agent will proceed without asking.</yellow>'
)
)
self.set_confirmation_mode(False)
return user_confirmation
return UserConfirmation.ACCEPT

62
openhands-cli/openhands_cli/setup.py Normal file
View File

@ -0,0 +1,62 @@
import os
from prompt_toolkit import HTML, print_formatted_text
from pydantic import SecretStr
from openhands.sdk import (
LLM,
Agent,
Conversation,
Tool,
)
from openhands.tools import (
BashExecutor,
FileEditorExecutor,
execute_bash_tool,
str_replace_editor_tool,
)
def setup_agent() -> Conversation:
"""
Setup the agent with environment variables.
"""
# Get API configuration from environment
api_key = os.getenv('LITELLM_API_KEY') or os.getenv('OPENAI_API_KEY')
model = os.getenv('LITELLM_MODEL', 'gpt-4o-mini')
base_url = os.getenv('LITELLM_BASE_URL')
if not api_key:
print_formatted_text(
HTML(
'<red>Error: No API key found. Please set LITELLM_API_KEY or OPENAI_API_KEY environment variable.</red>'
)
)
raise Exception(
'No API key found. Please set LITELLM_API_KEY or OPENAI_API_KEY environment variable.'
)
llm = LLM(
model=model,
api_key=SecretStr(api_key) if api_key else None,
base_url=base_url,
)
# Setup tools
cwd = os.getcwd()
bash = BashExecutor(working_dir=cwd)
file_editor = FileEditorExecutor()
tools: list[Tool] = [
execute_bash_tool.set_executor(executor=bash),
str_replace_editor_tool.set_executor(executor=file_editor),
]
# Create agent
agent = Agent(llm=llm, tools=tools)
conversation = Conversation(agent=agent)
print_formatted_text(
HTML(f'<green>✓ Agent initialized with model: {model}</green>')
)
return conversation

45
openhands-cli/openhands_cli/simple_main.py Normal file
View File

@ -0,0 +1,45 @@
#!/usr/bin/env python3
"""
Simple main entry point for OpenHands CLI.
This is a simplified version that demonstrates the TUI functionality.
"""
import traceback
from prompt_toolkit import print_formatted_text
from prompt_toolkit.formatted_text import HTML
def main() -> None:
"""Main entry point for the OpenHands CLI.
Raises:
ImportError: If agent chat dependencies are missing
Exception: On other error conditions
"""
try:
# Start agent chat directly by default
from openhands_cli.agent_chat import run_cli_entry
run_cli_entry()
except ImportError as e:
print_formatted_text(
HTML(f'<red>Error: Agent chat requires additional dependencies: {e}</red>')
)
print_formatted_text(
HTML('<yellow>Please ensure the agent SDK is properly installed.</yellow>')
)
raise
except KeyboardInterrupt:
print_formatted_text(HTML('\n<yellow>Goodbye! 👋</yellow>'))
except EOFError:
print_formatted_text(HTML('\n<yellow>Goodbye! 👋</yellow>'))
except Exception as e:
print_formatted_text(HTML(f'<red>Error starting agent chat: {e}</red>'))
traceback.print_exc()
raise
if __name__ == '__main__':
main()

92
openhands-cli/openhands_cli/tui.py Normal file
View File

@ -0,0 +1,92 @@
from collections.abc import Generator
from prompt_toolkit import print_formatted_text
from prompt_toolkit.completion import CompleteEvent, Completer, Completion
from prompt_toolkit.document import Document
from prompt_toolkit.formatted_text import HTML
from prompt_toolkit.shortcuts import clear
from openhands_cli import __version__
from openhands_cli.pt_style import get_cli_style
DEFAULT_STYLE = get_cli_style()
# Available commands with descriptions
COMMANDS = {
'/exit': 'Exit the application',
'/help': 'Display available commands',
'/clear': 'Clear the screen',
'/status': 'Display conversation details',
'/confirm': 'Toggle confirmation mode on/off',
'/new': 'Create a new conversation',
'/resume': 'Resume a paused conversation',
}
class CommandCompleter(Completer):
"""Custom completer for commands with interactive dropdown."""
def get_completions(
self, document: Document, complete_event: CompleteEvent
) -> Generator[Completion, None, None]:
text = document.text_before_cursor.lstrip()
if text.startswith('/'):
for command, description in COMMANDS.items():
if command.startswith(text):
yield Completion(
command,
start_position=-len(text),
display_meta=description,
style='bg:ansidarkgray fg:gold',
)
def display_banner(session_id: str) -> None:
print_formatted_text(
HTML(r"""<gold>
___ _ _ _
/ _ \ _ __ ___ _ __ | | | | __ _ _ __ __| |___
| | | | '_ \ / _ \ '_ \| |_| |/ _` | '_ \ / _` / __|
| |_| | |_) | __/ | | | _ | (_| | | | | (_| \__ \
\___ /| .__/ \___|_| |_|_| |_|\__,_|_| |_|\__,_|___/
|_|
</gold>"""),
style=DEFAULT_STYLE,
)
print_formatted_text(HTML(f'<grey>OpenHands CLI v{__version__}</grey>'))
print_formatted_text('')
print_formatted_text(HTML(f'<grey>Initialized conversation {session_id}</grey>'))
print_formatted_text('')
def display_help() -> None:
"""Display help information about available commands."""
print_formatted_text('')
print_formatted_text(HTML('<gold>🤖 OpenHands CLI Help</gold>'))
print_formatted_text(HTML('<grey>Available commands:</grey>'))
print_formatted_text('')
for command, description in COMMANDS.items():
print_formatted_text(HTML(f' <white>{command}</white> - {description}'))
print_formatted_text('')
print_formatted_text(HTML('<grey>Tips:</grey>'))
print_formatted_text(' • Type / and press Tab to see command suggestions')
print_formatted_text(' • Use arrow keys to navigate through suggestions')
print_formatted_text(' • Press Enter to select a command')
print_formatted_text('')
def display_welcome(session_id: str = 'chat') -> None:
"""Display welcome message."""
clear()
display_banner(session_id)
print_formatted_text(HTML("<gold>Let's start building!</gold>"))
print_formatted_text(
HTML(
'<green>What do you want to build? <grey>Type /help for help</grey></green>'
)
)
print()

7
openhands-cli/openhands_cli/user_actions/__init__.py Normal file
View File

@ -0,0 +1,7 @@
from openhands_cli.user_actions.agent_action import ask_user_confirmation
from openhands_cli.user_actions.exit_session import (
exit_session_confirmation,
)
from openhands_cli.user_actions.types import UserConfirmation
__all__ = ['ask_user_confirmation', 'exit_session_confirmation', 'UserConfirmation']

71
openhands-cli/openhands_cli/user_actions/agent_action.py Normal file
View File

@ -0,0 +1,71 @@
from prompt_toolkit import HTML, print_formatted_text
from openhands_cli.user_actions.types import UserConfirmation
from openhands_cli.user_actions.utils import cli_confirm, prompt_user
def ask_user_confirmation(pending_actions: list) -> tuple[UserConfirmation, str]:
"""Ask user to confirm pending actions.
Args:
pending_actions: List of pending actions from the agent
Returns:
Tuple of (UserConfirmation, reason) where reason is provided when rejecting with reason
"""
reason = ''
if not pending_actions:
return UserConfirmation.ACCEPT, reason
print_formatted_text(
HTML(
f'<yellow>🔍 Agent created {len(pending_actions)} action(s) and is waiting for confirmation:</yellow>'
)
)
for i, action in enumerate(pending_actions, 1):
tool_name = getattr(action, 'tool_name', '[unknown tool]')
print('tool name', tool_name)
action_content = (
str(getattr(action, 'action', ''))[:100].replace('\n', ' ')
or '[unknown action]'
)
print('action_content', action_content)
print_formatted_text(
HTML(f'<grey> {i}. {tool_name}: {action_content}...</grey>')
)
question = 'Choose an option:'
options = [
'Yes, proceed',
'No, reject (w/o reason)',
'No, reject with reason',
"Always proceed (don't ask again)",
]
try:
index = cli_confirm(question, options, escapable=True)
except (EOFError, KeyboardInterrupt):
print_formatted_text(HTML('\n<red>No input received; pausing agent.</red>'))
return UserConfirmation.DEFER, reason
if index == 0:
return UserConfirmation.ACCEPT, reason
elif index == 1:
return UserConfirmation.REJECT, reason
elif index == 2:
reason, should_defer = prompt_user(
'Please enter your reason for rejecting these actions: '
)
# If user pressed Ctrl+C or Ctrl+P during reason input, defer the action
if should_defer:
return UserConfirmation.DEFER, ''
return UserConfirmation.REJECT, reason
elif index == 3:
return UserConfirmation.ALWAYS_ACCEPT, reason
return UserConfirmation.REJECT, reason

18
openhands-cli/openhands_cli/user_actions/exit_session.py Normal file
View File

@ -0,0 +1,18 @@
from openhands_cli.user_actions.types import UserConfirmation
from openhands_cli.user_actions.utils import cli_confirm
def exit_session_confirmation() -> UserConfirmation:
"""
Ask user to confirm exiting session.
"""
question = 'Terminate session?'
options = ['Yes, proceed', 'No, dismiss']
index = cli_confirm(question, options) # Blocking UI, not escapable
options_mapping = {
0: UserConfirmation.ACCEPT, # User accepts termination session
1: UserConfirmation.REJECT, # User does not terminate session
}
return options_mapping.get(index, UserConfirmation.REJECT)

8
openhands-cli/openhands_cli/user_actions/types.py Normal file
View File

@ -0,0 +1,8 @@
from enum import Enum
class UserConfirmation(Enum):
ACCEPT = 'accept'
REJECT = 'reject'
DEFER = 'defer'
ALWAYS_ACCEPT = 'always_accept'

131
openhands-cli/openhands_cli/user_actions/utils.py Normal file
View File

@ -0,0 +1,131 @@
from prompt_toolkit.application import Application
from prompt_toolkit.input.base import Input
from prompt_toolkit.key_binding import KeyBindings
from prompt_toolkit.key_binding.key_processor import KeyPressEvent
from prompt_toolkit.layout.containers import HSplit, Window
from prompt_toolkit.layout.controls import FormattedTextControl
from prompt_toolkit.layout.dimension import Dimension
from prompt_toolkit.layout.layout import Layout
from prompt_toolkit.output.base import Output
from prompt_toolkit.shortcuts import prompt
from openhands_cli.tui import DEFAULT_STYLE
def build_keybindings(
choices: list[str], selected: list[int], escapable: bool
) -> KeyBindings:
"""Create keybindings for the confirm UI. Split for testability."""
kb = KeyBindings()
@kb.add('up')
def _handle_up(event: KeyPressEvent) -> None:
selected[0] = (selected[0] - 1) % len(choices)
@kb.add('down')
def _handle_down(event: KeyPressEvent) -> None:
selected[0] = (selected[0] + 1) % len(choices)
@kb.add('enter')
def _handle_enter(event: KeyPressEvent) -> None:
event.app.exit(result=selected[0])
if escapable:
@kb.add('c-c') # Ctrl+C
def _handle_hard_interrupt(event: KeyPressEvent) -> None:
event.app.exit(exception=KeyboardInterrupt())
@kb.add('c-p') # Ctrl+P
def _handle_pause_interrupt(event: KeyPressEvent) -> None:
event.app.exit(exception=KeyboardInterrupt())
@kb.add('escape') # Escape key
def _handle_escape(event: KeyPressEvent) -> None:
event.app.exit(exception=KeyboardInterrupt())
return kb
def build_layout(question: str, choices: list[str], selected_ref: list[int]) -> Layout:
"""Create the layout for the confirm UI. Split for testability."""
def get_choice_text() -> list[tuple[str, str]]:
lines: list[tuple[str, str]] = []
lines.append(('class:question', f'{question}\n\n'))
for i, choice in enumerate(choices):
is_selected = i == selected_ref[0]
prefix = '> ' if is_selected else ' '
style = 'class:selected' if is_selected else 'class:unselected'
lines.append((style, f'{prefix}{choice}\n'))
return lines
content_window = Window(
FormattedTextControl(get_choice_text),
always_hide_cursor=True,
height=Dimension(max=8),
)
return Layout(HSplit([content_window]))
def cli_confirm(
question: str = 'Are you sure?',
choices: list[str] | None = None,
initial_selection: int = 0,
escapable: bool = False,
input: Input | None = None, # strictly for unit testing
output: Output | None = None, # strictly for unit testing
) -> int:
"""Display a confirmation prompt with the given question and choices.
Returns the index of the selected choice.
"""
if choices is None:
choices = ['Yes', 'No']
selected = [initial_selection] # Using list to allow modification in closure
kb = build_keybindings(choices, selected, escapable)
layout = build_layout(question, choices, selected)
app = Application(
layout=layout,
key_bindings=kb,
style=DEFAULT_STYLE,
full_screen=False,
input=input,
output=output,
)
return int(app.run(in_thread=True))
def prompt_user(question: str) -> tuple[str, bool]:
"""Prompt user to enter a reason for rejecting actions.
Returns:
Tuple of (reason, should_defer) where:
- reason: The reason entered by the user
- should_defer: True if user pressed Ctrl+C or Ctrl+P, False otherwise
"""
kb = KeyBindings()
@kb.add('c-c')
def _(event: KeyPressEvent) -> None:
raise KeyboardInterrupt()
@kb.add('c-p')
def _(event: KeyPressEvent) -> None:
raise KeyboardInterrupt()
try:
reason = str(
prompt(
question,
style=DEFAULT_STYLE,
key_bindings=kb,
)
)
return reason.strip(), False
except KeyboardInterrupt:
return '', True

118
openhands-cli/pyproject.toml Normal file
View File

@ -0,0 +1,118 @@
[build-system]
build-backend = "hatchling.build"
requires = [ "hatchling" ]
[project]
name = "openhands-cli"
version = "0.1.0"
description = "OpenHands CLI - Terminal User Interface for OpenHands AI Agent"
readme = "README.md"
license = { text = "MIT" }
authors = [ { name = "OpenHands Team", email = "contact@all-hands.dev" } ]
requires-python = ">=3.12"
classifiers = [
"Programming Language :: Python :: 3 :: Only",
"Programming Language :: Python :: 3.12",
"Programming Language :: Python :: 3.13",
]
dependencies = [
"openhands-sdk @ git+https://github.com/All-Hands-AI/agent-sdk.git@585d4779b188694e99127431db51190db19e4352#subdirectory=openhands/sdk",
"openhands-tools @ git+https://github.com/All-Hands-AI/agent-sdk.git@585d4779b188694e99127431db51190db19e4352#subdirectory=openhands/tools",
"prompt-toolkit>=3",
"typer>=0.17.4",
]
optional-dependencies.dev = [
"black>=23",
"flake8>=6",
"isort>=5",
"mypy>=1",
"pre-commit>=3.7",
"pytest>=7",
"ruff>=0.11.8",
]
scripts.openhands-cli = "openhands_cli.simple_main:main"
[dependency-groups]
dev = [
"pre-commit>=4.3",
"pyinstaller>=6.15",
"pytest>=8.4.1",
]
[tool.poetry]
name = "openhands-cli"
version = "0.1.0"
description = "OpenHands CLI - Terminal User Interface for OpenHands AI Agent"
authors = [ "OpenHands Team <contact@all-hands.dev>" ]
license = "MIT"
readme = "README.md"
packages = [ { include = "openhands_cli" } ]
[tool.poetry.dependencies]
python = "^3.12"
prompt-toolkit = "^3.0.0"
openhands-sdk = { git = "https://github.com/All-Hands-AI/agent-sdk.git", rev = "585d4779b188694e99127431db51190db19e4352", subdirectory = "openhands/sdk" }
openhands-tools = { git = "https://github.com/All-Hands-AI/agent-sdk.git", rev = "585d4779b188694e99127431db51190db19e4352", subdirectory = "openhands/tools" }
[tool.poetry.group.dev.dependencies]
pytest = "^7.0.0"
black = "^23.0.0"
isort = "^5.0.0"
flake8 = "^6.0.0"
mypy = "^1.0.0"
pre-commit = "^3.7.0"
ruff = "^0.11.8"
[tool.poetry.scripts]
openhands-cli = "openhands_cli.simple_main:main"
[tool.hatch.metadata]
allow-direct-references = true
[tool.hatch.build.targets.wheel]
packages = [ "openhands_cli" ]
# Keep Poetry configuration for compatibility
[tool.black]
line-length = 88
target-version = [ 'py312' ]
[tool.ruff]
target-version = "py312"
line-length = 88
format.indent-style = "space"
format.quote-style = "double"
format.line-ending = "auto"
format.skip-magic-trailing-comma = false
lint.select = [
"B", # flake8-bugbear
"C4", # flake8-comprehensions
"E", # pycodestyle errors
"F", # pyflakes
"I", # isort
"UP", # pyupgrade
"W", # pycodestyle warnings
]
lint.ignore = [
"B008", # do not perform function calls in argument defaults
"C901", # too complex
"E501", # line too long, handled by black
]
[tool.isort]
profile = "black"
line_length = 88
[tool.mypy]
python_version = "3.12"
warn_return_any = true
warn_unused_configs = true
disallow_untyped_defs = true
ignore_missing_imports = true
[tool.uv.sources]
openhands-sdk = { git = "https://github.com/All-Hands-AI/agent-sdk.git", subdirectory = "openhands/sdk", rev = "585d4779b188694e99127431db51190db19e4352" }

1
openhands-cli/tests/__init__.py Normal file
View File

@ -0,0 +1 @@
"""Tests for OpenHands CLI."""

333
openhands-cli/tests/test_confirmation_mode.py Normal file
View File

@ -0,0 +1,333 @@
#!/usr/bin/env python3
"""
Tests for confirmation mode functionality in OpenHands CLI.
"""
import os
from concurrent.futures import ThreadPoolExecutor
from typing import Any
from unittest.mock import MagicMock, patch
import pytest
from openhands_cli.runner import ConversationRunner
from openhands_cli.setup import setup_agent
from openhands_cli.user_actions import agent_action, ask_user_confirmation, utils
from openhands_cli.user_actions.types import UserConfirmation
from prompt_toolkit.input.defaults import create_pipe_input
from prompt_toolkit.output.defaults import DummyOutput
from openhands.sdk import ActionBase
from tests.utils import _send_keys
class MockAction(ActionBase):
"""Mock action schema for testing."""
command: str
class TestConfirmationMode:
"""Test suite for confirmation mode functionality."""
def test_setup_agent_creates_conversation(self) -> None:
"""Test that setup_agent creates a conversation successfully."""
with patch.dict(os.environ, {'LITELLM_API_KEY': 'test-key'}):
with (
patch('openhands_cli.setup.LLM'),
patch('openhands_cli.setup.Agent'),
patch('openhands_cli.setup.Conversation') as mock_conversation,
patch('openhands_cli.setup.BashExecutor'),
patch('openhands_cli.setup.FileEditorExecutor'),
):
mock_conv_instance = MagicMock()
mock_conversation.return_value = mock_conv_instance
result = setup_agent()
# Verify conversation was created and returned
assert result == mock_conv_instance
mock_conversation.assert_called_once()
def test_conversation_runner_set_confirmation_mode(self) -> None:
"""Test that ConversationRunner can set confirmation mode."""
mock_conversation = MagicMock()
runner = ConversationRunner(mock_conversation)
# Test enabling confirmation mode
runner.set_confirmation_mode(True)
assert runner.confirmation_mode is True
mock_conversation.set_confirmation_mode.assert_called_with(True)
# Test disabling confirmation mode
runner.set_confirmation_mode(False)
assert runner.confirmation_mode is False
mock_conversation.set_confirmation_mode.assert_called_with(False)
def test_conversation_runner_initial_state(self) -> None:
"""Test that ConversationRunner starts with confirmation mode disabled."""
mock_conversation = MagicMock()
runner = ConversationRunner(mock_conversation)
# Verify initial state
assert runner.confirmation_mode is False
def test_setup_agent_without_api_key(self) -> None:
"""Test that setup_agent raises exception when API key is missing."""
with patch.dict(os.environ, {}, clear=True):
with (
patch('openhands_cli.setup.print_formatted_text'),
pytest.raises(Exception, match='No API key found'),
):
setup_agent()
def test_ask_user_confirmation_empty_actions(self) -> None:
"""Test that ask_user_confirmation returns ACCEPT for empty actions list."""
result, reason = ask_user_confirmation([])
assert result == UserConfirmation.ACCEPT
assert reason == ''
@patch('openhands_cli.user_actions.agent_action.cli_confirm')
def test_ask_user_confirmation_yes(self, mock_cli_confirm: Any) -> None:
"""Test that ask_user_confirmation returns ACCEPT when user selects yes."""
mock_cli_confirm.return_value = 0 # First option (Yes, proceed)
mock_action = MagicMock()
mock_action.tool_name = 'bash'
mock_action.action = 'ls -la'
result, reason = ask_user_confirmation([mock_action])
assert result == UserConfirmation.ACCEPT
assert reason == ''
@patch('openhands_cli.user_actions.agent_action.cli_confirm')
def test_ask_user_confirmation_no(self, mock_cli_confirm: Any) -> None:
"""Test that ask_user_confirmation returns REJECT when user selects no."""
mock_cli_confirm.return_value = 1 # Second option (No, reject)
mock_action = MagicMock()
mock_action.tool_name = 'bash'
mock_action.action = 'rm -rf /'
result, reason = ask_user_confirmation([mock_action])
assert result == UserConfirmation.REJECT
assert reason == ''
@patch('openhands_cli.user_actions.agent_action.cli_confirm')
def test_ask_user_confirmation_y_shorthand(self, mock_cli_confirm: Any) -> None:
"""Test that ask_user_confirmation accepts first option as yes."""
mock_cli_confirm.return_value = 0 # First option (Yes, proceed)
mock_action = MagicMock()
mock_action.tool_name = 'bash'
mock_action.action = 'echo hello'
result, reason = ask_user_confirmation([mock_action])
assert result == UserConfirmation.ACCEPT
assert reason == ''
@patch('openhands_cli.user_actions.agent_action.cli_confirm')
def test_ask_user_confirmation_n_shorthand(self, mock_cli_confirm: Any) -> None:
"""Test that ask_user_confirmation accepts second option as no."""
mock_cli_confirm.return_value = 1 # Second option (No, reject)
mock_action = MagicMock()
mock_action.tool_name = 'bash'
mock_action.action = 'dangerous command'
result, reason = ask_user_confirmation([mock_action])
assert result == UserConfirmation.REJECT
assert reason == ''
@patch('openhands_cli.user_actions.agent_action.cli_confirm')
def test_ask_user_confirmation_invalid_then_yes(
self, mock_cli_confirm: Any
) -> None:
"""Test that ask_user_confirmation handles selection and accepts yes."""
mock_cli_confirm.return_value = 0 # First option (Yes, proceed)
mock_action = MagicMock()
mock_action.tool_name = 'bash'
mock_action.action = 'echo test'
result, reason = ask_user_confirmation([mock_action])
assert result == UserConfirmation.ACCEPT
assert reason == ''
assert mock_cli_confirm.call_count == 1
@patch('openhands_cli.user_actions.agent_action.cli_confirm')
def test_ask_user_confirmation_keyboard_interrupt(
self, mock_cli_confirm: Any
) -> None:
"""Test that ask_user_confirmation handles KeyboardInterrupt gracefully."""
mock_cli_confirm.side_effect = KeyboardInterrupt()
mock_action = MagicMock()
mock_action.tool_name = 'bash'
mock_action.action = 'echo test'
result, reason = ask_user_confirmation([mock_action])
assert result == UserConfirmation.DEFER
assert reason == ''
@patch('openhands_cli.user_actions.agent_action.cli_confirm')
def test_ask_user_confirmation_eof_error(self, mock_cli_confirm: Any) -> None:
"""Test that ask_user_confirmation handles EOFError gracefully."""
mock_cli_confirm.side_effect = EOFError()
mock_action = MagicMock()
mock_action.tool_name = 'bash'
mock_action.action = 'echo test'
result, reason = ask_user_confirmation([mock_action])
assert result == UserConfirmation.DEFER
assert reason == ''
def test_ask_user_confirmation_multiple_actions(self) -> None:
"""Test that ask_user_confirmation displays multiple actions correctly."""
with (
patch(
'openhands_cli.user_actions.agent_action.cli_confirm'
) as mock_cli_confirm,
patch(
'openhands_cli.user_actions.agent_action.print_formatted_text'
) as mock_print,
):
mock_cli_confirm.return_value = 0 # First option (Yes, proceed)
mock_action1 = MagicMock()
mock_action1.tool_name = 'bash'
mock_action1.action = 'ls -la'
mock_action2 = MagicMock()
mock_action2.tool_name = 'str_replace_editor'
mock_action2.action = 'create file.txt'
result, reason = ask_user_confirmation([mock_action1, mock_action2])
assert result == UserConfirmation.ACCEPT
assert reason == ''
# Verify that both actions were displayed
assert mock_print.call_count >= 3 # Header + 2 actions
@patch('openhands_cli.user_actions.agent_action.prompt_user')
@patch('openhands_cli.user_actions.agent_action.cli_confirm')
def test_ask_user_confirmation_no_with_reason(
self, mock_cli_confirm: Any, mock_prompt_user: Any
) -> None:
"""Test that ask_user_confirmation returns REJECT when user selects 'No (with reason)'."""
mock_cli_confirm.return_value = 2 # Third option (No, with reason)
mock_prompt_user.return_value = ('This action is too risky', False)
mock_action = MagicMock()
mock_action.tool_name = 'bash'
mock_action.action = 'rm -rf /'
result, reason = ask_user_confirmation([mock_action])
assert result == UserConfirmation.REJECT
assert reason == 'This action is too risky'
mock_prompt_user.assert_called_once()
@patch('openhands_cli.user_actions.agent_action.prompt_user')
@patch('openhands_cli.user_actions.agent_action.cli_confirm')
def test_ask_user_confirmation_no_with_reason_cancelled(
self, mock_cli_confirm: Any, mock_prompt_user: Any
) -> None:
"""Test that ask_user_confirmation falls back to DEFER when reason input is cancelled."""
mock_cli_confirm.return_value = 2 # Third option (No, with reason)
mock_prompt_user.return_value = ('', True) # User cancelled reason input
mock_action = MagicMock()
mock_action.tool_name = 'bash'
mock_action.action = 'dangerous command'
result, reason = ask_user_confirmation([mock_action])
assert result == UserConfirmation.DEFER
assert reason == ''
mock_prompt_user.assert_called_once()
def test_user_confirmation_is_escapable_e2e(
self, monkeypatch: pytest.MonkeyPatch
) -> None:
"""E2E: non-escapable should ignore Ctrl-C/Ctrl-P/Esc; only Enter returns."""
real_cli_confirm = utils.cli_confirm
with create_pipe_input() as pipe:
output = DummyOutput()
def wrapper(
question: str,
choices: list[str] | None = None,
initial_selection: int = 0,
escapable: bool = False,
**extra: object,
) -> int:
# keep original params; inject test IO
return real_cli_confirm(
question=question,
choices=choices,
initial_selection=initial_selection,
escapable=escapable,
input=pipe,
output=output,
)
# Patch the symbol the caller uses
monkeypatch.setattr(agent_action, 'cli_confirm', wrapper, raising=True)
with ThreadPoolExecutor(max_workers=1) as ex:
fut = ex.submit(
ask_user_confirmation, [MockAction(command='echo hello world')]
)
_send_keys(pipe, '\x03') # Ctrl-C (ignored)
result, reason = fut.result(timeout=2.0)
assert result == UserConfirmation.DEFER # escaped confirmation view
assert reason == ''
@patch('openhands_cli.user_actions.agent_action.cli_confirm')
def test_ask_user_confirmation_always_accept(self, mock_cli_confirm: Any) -> None:
"""Test that ask_user_confirmation returns ALWAYS_ACCEPT when user selects fourth option."""
mock_cli_confirm.return_value = 3 # Fourth option (Always proceed)
mock_action = MagicMock()
mock_action.tool_name = 'bash'
mock_action.action = 'echo test'
result, reason = ask_user_confirmation([mock_action])
assert result == UserConfirmation.ALWAYS_ACCEPT
assert reason == ''
def test_conversation_runner_handles_always_accept(self) -> None:
"""Test that ConversationRunner disables confirmation mode when ALWAYS_ACCEPT is returned."""
mock_conversation = MagicMock()
runner = ConversationRunner(mock_conversation)
# Enable confirmation mode first
runner.set_confirmation_mode(True)
assert runner.confirmation_mode is True
# Mock the conversation state to simulate waiting for confirmation
mock_conversation.state.agent_waiting_for_confirmation = True
mock_conversation.state.agent_finished = False
# Mock get_unmatched_actions to return some actions
with patch('openhands_cli.runner.get_unmatched_actions') as mock_get_actions:
mock_action = MagicMock()
mock_action.tool_name = 'bash'
mock_action.action = 'echo test'
mock_get_actions.return_value = [mock_action]
# Mock ask_user_confirmation to return ALWAYS_ACCEPT
with patch('openhands_cli.runner.ask_user_confirmation') as mock_ask:
mock_ask.return_value = (UserConfirmation.ALWAYS_ACCEPT, '')
# Mock print_formatted_text to avoid output during test
with patch('openhands_cli.runner.print_formatted_text'):
result = runner._handle_confirmation_request()
# Verify that confirmation mode was disabled
assert result == UserConfirmation.ALWAYS_ACCEPT
assert runner.confirmation_mode is False
mock_conversation.set_confirmation_mode.assert_called_with(False)

106
openhands-cli/tests/test_conversation_runner.py Normal file
View File

@ -0,0 +1,106 @@
from types import SimpleNamespace
from unittest.mock import MagicMock, patch
import pytest
from openhands_cli.runner import ConversationRunner
from openhands_cli.user_actions.types import UserConfirmation
class TestConversationRunner:
def _setup_conversation_mock(
self,
agent_paused: bool = False,
agent_waiting_for_confirmation: bool = False,
agent_finished: bool = False,
) -> MagicMock:
convo = MagicMock()
convo.state = SimpleNamespace(
agent_paused=agent_paused,
agent_waiting_for_confirmation=agent_waiting_for_confirmation,
agent_finished=agent_finished,
events=[],
)
return convo
@pytest.mark.parametrize('paused', [False, True])
def test_non_confirmation_mode_runs_once(self, paused: bool) -> None:
"""
1. Confirmation mode is not on
2. Process message resumes paused conversation or continues running conversation
"""
convo = self._setup_conversation_mock(
agent_paused=paused,
agent_waiting_for_confirmation=False,
agent_finished=False,
)
cr = ConversationRunner(convo)
cr.set_confirmation_mode(False)
with patch.object(convo, 'run') as run_mock:
cr.process_message(message=None)
run_mock.assert_called_once()
@pytest.mark.parametrize(
'confirmation, agent_paused, agent_finished, expected_run_calls',
[
# Case 1: Agent paused & waiting; user DEFERS -> early return, no run()
(UserConfirmation.DEFER, True, False, 0),
# Case 2: Agent waiting; user ACCEPTS -> run() once, break (finished=True)
(UserConfirmation.ACCEPT, False, True, 1),
],
)
def test_confirmation_mode_waiting_and_user_decision_controls_run(
self,
confirmation: UserConfirmation,
agent_paused: bool,
agent_finished: bool,
expected_run_calls: int,
) -> None:
"""
1. Agent may be paused but is waiting for consent on actions
2. If paused, we should have asked for confirmation on action
3. If not paused, we should still ask for confirmation on actions
4. If deferred no run call to agent should be made
5. If accepted, run call to agent should be made
"""
convo = self._setup_conversation_mock(
agent_paused=agent_paused,
agent_waiting_for_confirmation=True,
agent_finished=agent_finished,
)
cr = ConversationRunner(convo)
cr.set_confirmation_mode(True)
with (
patch.object(cr, '_handle_confirmation_request', return_value=confirmation),
patch.object(convo, 'run') as run_mock,
):
cr.process_message(message=None)
assert run_mock.call_count == expected_run_calls
def test_confirmation_mode_not_waiting__runs_once_when_finished_true(self) -> None:
"""
1. Agent was not waiting
2. Agent finished without any actions
3. Conversation should finished without asking user for instructions
"""
convo = self._setup_conversation_mock(
agent_paused=True,
agent_waiting_for_confirmation=False,
agent_finished=True,
)
cr = ConversationRunner(convo)
cr.set_confirmation_mode(True)
with (
patch.object(cr, '_handle_confirmation_request') as _mock_h,
patch.object(convo, 'run') as run_mock,
):
cr.process_message(message=None)
# No confirmation was needed up front; we still expect exactly one run.
run_mock.assert_called_once()
_mock_h.assert_not_called()

107
openhands-cli/tests/test_exit_session_confirmation.py Normal file
View File

@ -0,0 +1,107 @@
#!/usr/bin/env python3
"""
Tests for exit_session_confirmation functionality in OpenHands CLI.
"""
from collections.abc import Iterator
from concurrent.futures import ThreadPoolExecutor
from unittest.mock import MagicMock, patch
import pytest
from openhands_cli.user_actions import (
exit_session,
exit_session_confirmation,
utils,
)
from openhands_cli.user_actions.types import UserConfirmation
from prompt_toolkit.input.defaults import create_pipe_input
from prompt_toolkit.output.defaults import DummyOutput
from tests.utils import _send_keys
QUESTION = 'Terminate session?'
OPTIONS = ['Yes, proceed', 'No, dismiss']
@pytest.fixture()
def confirm_patch() -> Iterator[MagicMock]:
"""Patch cli_confirm once per test and yield the mock."""
with patch('openhands_cli.user_actions.exit_session.cli_confirm') as m:
yield m
def _assert_called_once_with_defaults(mock_cli_confirm: MagicMock) -> None:
"""Ensure the question/options are correct and 'escapable' is not enabled."""
mock_cli_confirm.assert_called_once()
args, kwargs = mock_cli_confirm.call_args
# Positional args
assert args == (QUESTION, OPTIONS)
# Should not opt into escapable mode
assert 'escapable' not in kwargs or kwargs['escapable'] is False
class TestExitSessionConfirmation:
"""Test suite for exit_session_confirmation functionality."""
@pytest.mark.parametrize(
'index,expected',
[
(0, UserConfirmation.ACCEPT), # Yes
(1, UserConfirmation.REJECT), # No
(999, UserConfirmation.REJECT), # Invalid => default reject
(-1, UserConfirmation.REJECT), # Negative => default reject
],
)
def test_index_mapping(
self, confirm_patch: MagicMock, index: int, expected: UserConfirmation
) -> None:
"""All index-to-result mappings, including invalid/negative, in one place."""
confirm_patch.return_value = index
result = exit_session_confirmation()
assert isinstance(result, UserConfirmation)
assert result == expected
_assert_called_once_with_defaults(confirm_patch)
def test_exit_session_confirmation_non_escapable_e2e(
self, monkeypatch: pytest.MonkeyPatch
) -> None:
"""E2E: non-escapable should ignore Ctrl-C/Ctrl-P/Esc; only Enter returns."""
real_cli_confirm = utils.cli_confirm
with create_pipe_input() as pipe:
output = DummyOutput()
def wrapper(
question: str,
choices: list[str] | None = None,
initial_selection: int = 0,
escapable: bool = False,
**extra: object,
) -> int:
# keep original params; inject test IO
return real_cli_confirm(
question=question,
choices=choices,
initial_selection=initial_selection,
escapable=escapable,
input=pipe,
output=output,
)
# Patch the symbol the caller uses
monkeypatch.setattr(exit_session, 'cli_confirm', wrapper, raising=True)
with ThreadPoolExecutor(max_workers=1) as ex:
fut = ex.submit(exit_session_confirmation)
_send_keys(pipe, '\x03') # Ctrl-C (ignored)
_send_keys(pipe, '\x10') # Ctrl-P (ignored)
_send_keys(pipe, '\x1b') # Esc (ignored)
_send_keys(pipe, '\x1b[B') # Arrow Down to "No, dismiss"
_send_keys(pipe, '\r') # Enter
result = fut.result(timeout=2.0)
assert result == UserConfirmation.REJECT

65
openhands-cli/tests/test_main.py Normal file
View File

@ -0,0 +1,65 @@
"""Tests for main entry point functionality."""
from unittest.mock import MagicMock, patch
import pytest
from openhands_cli import simple_main
class TestMainEntryPoint:
"""Test the main entry point behavior."""
@patch('openhands_cli.agent_chat.run_cli_entry')
def test_main_starts_agent_chat_directly(
self, mock_run_agent_chat: MagicMock
) -> None:
"""Test that main() starts agent chat directly without menu."""
mock_run_agent_chat.return_value = None
# Should complete without raising an exception
simple_main.main()
# Should call run_agent_chat directly
mock_run_agent_chat.assert_called_once()
@patch('openhands_cli.agent_chat.run_cli_entry')
def test_main_handles_import_error(self, mock_run_agent_chat: MagicMock) -> None:
"""Test that main() handles ImportError gracefully."""
mock_run_agent_chat.side_effect = ImportError('Missing dependency')
# Should raise ImportError (no longer using sys.exit)
with pytest.raises(ImportError) as exc_info:
simple_main.main()
assert str(exc_info.value) == 'Missing dependency'
@patch('openhands_cli.agent_chat.run_cli_entry')
def test_main_handles_keyboard_interrupt(
self, mock_run_agent_chat: MagicMock
) -> None:
"""Test that main() handles KeyboardInterrupt gracefully."""
mock_run_agent_chat.side_effect = KeyboardInterrupt()
# Should complete without raising an exception (graceful exit)
simple_main.main()
@patch('openhands_cli.agent_chat.run_cli_entry')
def test_main_handles_eof_error(self, mock_run_agent_chat: MagicMock) -> None:
"""Test that main() handles EOFError gracefully."""
mock_run_agent_chat.side_effect = EOFError()
# Should complete without raising an exception (graceful exit)
simple_main.main()
@patch('openhands_cli.agent_chat.run_cli_entry')
def test_main_handles_general_exception(
self, mock_run_agent_chat: MagicMock
) -> None:
"""Test that main() handles general exceptions."""
mock_run_agent_chat.side_effect = Exception('Unexpected error')
# Should raise Exception (no longer using sys.exit)
with pytest.raises(Exception) as exc_info:
simple_main.main()
assert str(exc_info.value) == 'Unexpected error'

52
openhands-cli/tests/test_pause_listener.py Normal file
View File

@ -0,0 +1,52 @@
#!/usr/bin/env python3
"""
Tests for pause listener in OpenHands CLI.
"""
import time
from unittest.mock import MagicMock
from openhands_cli.listeners.pause_listener import PauseListener, pause_listener
from prompt_toolkit.input.defaults import create_pipe_input
from openhands.sdk import Conversation
class TestPauseListener:
"""Test suite for PauseListener class."""
def test_pause_listener_stop(self) -> None:
"""Test PauseListener stop functionality."""
mock_callback = MagicMock()
listener = PauseListener(on_pause=mock_callback)
listener.start()
# Initially not paused
assert not listener.is_paused()
assert listener.is_alive()
# Stop the listener
listener.stop()
# Listner was shutdown not paused
assert not listener.is_paused()
assert listener.is_stopped()
def test_pause_listener_context_manager(self) -> None:
"""Test pause_listener context manager."""
mock_conversation = MagicMock(spec=Conversation)
with create_pipe_input() as pipe:
with pause_listener(mock_conversation, pipe) as listener:
assert isinstance(listener, PauseListener)
assert listener.on_pause == mock_conversation.pause
# Listener should be started (daemon thread)
assert listener.is_alive()
assert not listener.is_paused()
pipe.send_text('\x10') # Ctrl-P
time.sleep(0.1)
assert listener.is_paused()
assert listener.is_stopped()
assert not listener.is_alive()

92
openhands-cli/tests/test_tui.py Normal file
View File

@ -0,0 +1,92 @@
"""Tests for TUI functionality."""
from openhands_cli.tui import COMMANDS, CommandCompleter
from prompt_toolkit.completion import CompleteEvent
from prompt_toolkit.document import Document
class TestCommandCompleter:
"""Test the CommandCompleter class."""
def test_command_completion_with_slash(self) -> None:
"""Test that commands are completed when starting with /."""
completer = CommandCompleter()
document = Document('/')
completions = list(completer.get_completions(document, CompleteEvent()))
# Should return all available commands
assert len(completions) == len(COMMANDS)
# Check that all commands are included
completion_texts = [c.text for c in completions]
for command in COMMANDS.keys():
assert command in completion_texts
def test_command_completion_partial_match(self) -> None:
"""Test that partial command matches work correctly."""
completer = CommandCompleter()
document = Document('/ex')
completions = list(completer.get_completions(document, CompleteEvent()))
# Should return only /exit
assert len(completions) == 1
assert completions[0].text == '/exit'
# display_meta is a FormattedText object, so we need to check its content
# Extract the text from FormattedText
meta_text = completions[0].display_meta
if hasattr(meta_text, '_formatted_text'):
# Extract text from FormattedText
text_content = ''.join([item[1] for item in meta_text._formatted_text])
else:
text_content = str(meta_text)
assert COMMANDS['/exit'] in text_content
def test_command_completion_no_slash(self) -> None:
"""Test that no completions are returned without /."""
completer = CommandCompleter()
document = Document('help')
completions = list(completer.get_completions(document, CompleteEvent()))
# Should return no completions
assert len(completions) == 0
def test_command_completion_no_match(self) -> None:
"""Test that no completions are returned for non-matching commands."""
completer = CommandCompleter()
document = Document('/nonexistent')
completions = list(completer.get_completions(document, CompleteEvent()))
# Should return no completions
assert len(completions) == 0
def test_command_completion_styling(self) -> None:
"""Test that completions have proper styling."""
completer = CommandCompleter()
document = Document('/help')
completions = list(completer.get_completions(document, CompleteEvent()))
assert len(completions) == 1
completion = completions[0]
assert completion.style == 'bg:ansidarkgray fg:gold'
assert completion.start_position == -5 # Length of "/help"
def test_commands_dict() -> None:
"""Test that COMMANDS dictionary contains expected commands."""
expected_commands = {
'/exit',
'/help',
'/clear',
'/status',
'/confirm',
'/new',
'/resume',
}
assert set(COMMANDS.keys()) == expected_commands
# Check that all commands have descriptions
for command, description in COMMANDS.items():
assert isinstance(command, str)
assert command.startswith('/')
assert isinstance(description, str)
assert len(description) > 0

9
openhands-cli/tests/utils.py Normal file
View File

@ -0,0 +1,9 @@
import time
from prompt_toolkit.input import PipeInput
def _send_keys(pipe: PipeInput, text: str, delay: float = 0.05) -> None:
"""Helper: small delay then send keys to avoid race with app.run()."""
time.sleep(delay)
pipe.send_text(text)

2176
openhands-cli/uv.lock generated Normal file
View File

File diff suppressed because it is too large Load Diff