github/OpenHands
1
1
Fork 0
mirror of https://github.com/All-Hands-AI/OpenHands.git synced 2026-01-08 22:38:05 -05:00
Code Issues Packages Projects Releases Wiki Activity

CLI: Rm loading context (#11603)

Browse Source 
Co-authored-by: openhands <openhands@all-hands.dev>
This commit is contained in:
Rohit Malhotra
2025-11-03 11:15:47 -05:00
committed by GitHub
parent e51685dab4
commit 2e49f07451
5 changed files with 20 additions and 153 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
openhands-cli/openhands_cli/listeners/__init__.py
View File

@@ -1,4 +1,3 @@
from openhands_cli.listeners.loading_listener import LoadingContext
from openhands_cli.listeners.pause_listener import PauseListener
__all__ = ['PauseListener', 'LoadingContext']
__all__ = ['PauseListener']

63
openhands-cli/openhands_cli/listeners/loading_listener.py
View File

@@ -1,63 +0,0 @@
"""
Loading animation utilities for OpenHands CLI.
Provides animated loading screens during agent initialization.
"""
import sys
import threading
import time
def display_initialization_animation(text: str, is_loaded: threading.Event) -> None:
"""Display a spinning animation while agent is being initialized.
Args:
text: The text to display alongside the animation
is_loaded: Threading event that signals when loading is complete
"""
ANIMATION_FRAMES = ['⠋', '⠙', '⠹', '⠸', '⠼', '⠴', '⠦', '⠧', '⠇', '⠏']
i = 0
while not is_loaded.is_set():
sys.stdout.write('\n')
sys.stdout.write(
f'\033[s\033[J\033[38;2;255;215;0m[{ANIMATION_FRAMES[i % len(ANIMATION_FRAMES)]}] {text}\033[0m\033[u\033[1A'
)
sys.stdout.flush()
time.sleep(0.1)
i += 1
sys.stdout.write('\r' + ' ' * (len(text) + 10) + '\r')
sys.stdout.flush()
class LoadingContext:
"""Context manager for displaying loading animations in a separate thread."""
def __init__(self, text: str):
"""Initialize the loading context.
Args:
text: The text to display during loading
"""
self.text = text
self.is_loaded = threading.Event()
self.loading_thread: threading.Thread | None = None
def __enter__(self) -> 'LoadingContext':
"""Start the loading animation in a separate thread."""
self.loading_thread = threading.Thread(
target=display_initialization_animation,
args=(self.text, self.is_loaded),
daemon=True,
)
self.loading_thread.start()
return self
def __exit__(self, exc_type, exc_val, exc_tb) -> None:
"""Stop the loading animation and clean up the thread."""
self.is_loaded.set()
if self.loading_thread:
self.loading_thread.join(
timeout=1.0
) # Wait up to 1 second for thread to finish

36
openhands-cli/openhands_cli/setup.py
View File

@@ -6,7 +6,6 @@ from openhands.sdk import Agent, BaseConversation, Conversation, Workspace, regi
from openhands.tools.execute_bash import BashTool
from openhands.tools.file_editor import FileEditorTool
from openhands.tools.task_tracker import TaskTrackerTool
from openhands_cli.listeners import LoadingContext
from openhands_cli.locations import CONVERSATIONS_DIR, WORK_DIR
from openhands_cli.tui.settings.store import AgentStore
from openhands.sdk.security.confirmation_policy import (
@@ -70,26 +69,29 @@ def setup_conversation(
MissingAgentSpec: If agent specification is not found or invalid.
"""
with LoadingContext('Initializing OpenHands agent...'):
agent = load_agent_specs(str(conversation_id))
print_formatted_text(
HTML(f'<white>Initializing agent...</white>')
)
if not include_security_analyzer:
# Remove security analyzer from agent spec
agent = agent.model_copy(
update={"security_analyzer": None}
)
agent = load_agent_specs(str(conversation_id))
# Create conversation - agent context is now set in AgentStore.load()
conversation: BaseConversation = Conversation(
agent=agent,
workspace=Workspace(working_dir=WORK_DIR),
# Conversation will add /<conversation_id> to this path
persistence_dir=CONVERSATIONS_DIR,
conversation_id=conversation_id,
if not include_security_analyzer:
# Remove security analyzer from agent spec
agent = agent.model_copy(
update={"security_analyzer": None}
)
if include_security_analyzer:
conversation.set_confirmation_policy(AlwaysConfirm())
# Create conversation - agent context is now set in AgentStore.load()
conversation: BaseConversation = Conversation(
agent=agent,
workspace=Workspace(working_dir=WORK_DIR),
# Conversation will add /<conversation_id> to this path
persistence_dir=CONVERSATIONS_DIR,
conversation_id=conversation_id,
)
if include_security_analyzer:
conversation.set_confirmation_policy(AlwaysConfirm())
print_formatted_text(
HTML(f'<green>✓ Agent initialized with model: {agent.llm.model}</green>')

2
openhands-cli/tests/test_confirmation_mode.py
View File

@@ -73,8 +73,6 @@ class TestConfirmationMode:
persistence_dir=ANY,
conversation_id=mock_conversation_id,
)
# Verify print_formatted_text was called
mock_print.assert_called_once()
def test_setup_conversation_raises_missing_agent_spec(self) -> None:
"""Test that setup_conversation raises MissingAgentSpec when agent is not found."""

69
openhands-cli/tests/test_loading.py
View File

@@ -1,69 +0,0 @@
#!/usr/bin/env python3
"""
Unit tests for the loading animation functionality.
"""
import threading
import time
import unittest
from unittest.mock import patch
from openhands_cli.listeners.loading_listener import (
LoadingContext,
display_initialization_animation,
)
class TestLoadingAnimation(unittest.TestCase):
"""Test cases for loading animation functionality."""
def test_loading_context_manager(self):
"""Test that LoadingContext works as a context manager."""
with LoadingContext('Test loading...') as ctx:
self.assertIsInstance(ctx, LoadingContext)
self.assertEqual(ctx.text, 'Test loading...')
self.assertIsInstance(ctx.is_loaded, threading.Event)
self.assertIsNotNone(ctx.loading_thread)
# Give the thread a moment to start
time.sleep(0.1)
self.assertTrue(ctx.loading_thread.is_alive())
# After exiting context, thread should be stopped
time.sleep(0.1)
self.assertFalse(ctx.loading_thread.is_alive())
@patch('sys.stdout')
def test_animation_writes_while_running_and_stops_after(self, mock_stdout):
"""Ensure stdout is written while animation runs and stops after it ends."""
is_loaded = threading.Event()
animation_thread = threading.Thread(
target=display_initialization_animation,
args=('Test output', is_loaded),
daemon=True,
)
animation_thread.start()
# Let it run a bit and check calls
time.sleep(0.2)
calls_while_running = mock_stdout.write.call_count
self.assertGreater(calls_while_running, 0, 'Expected writes while spinner runs')
# Stop animation
is_loaded.set()
time.sleep(0.2)
animation_thread.join(timeout=1.0)
calls_after_stop = mock_stdout.write.call_count
# Wait a moment to detect any stray writes after thread finished
time.sleep(0.2)
self.assertEqual(
calls_after_stop,
mock_stdout.write.call_count,
'No extra writes should occur after animation stops',
)
if __name__ == '__main__':
unittest.main()