diff --git a/connectonion/useful_plugins/__init__.py b/connectonion/useful_plugins/__init__.py index d80c34b..f416bf8 100644 --- a/connectonion/useful_plugins/__init__.py +++ b/connectonion/useful_plugins/__init__.py @@ -21,6 +21,6 @@ from .tool_approval import tool_approval, handle_mode_change from .auto_compact import auto_compact from .prefer_write_tool import prefer_write_tool -from .ulw import ulw, handle_ulw_mode_change +from .ulw import ulw, UltraWork -__all__ = ['re_act', 'eval', 'image_result_formatter', 'shell_approval', 'gmail_plugin', 'calendar_plugin', 'ui_stream', 'system_reminder', 'tool_approval', 'handle_mode_change', 'auto_compact', 'prefer_write_tool', 'ulw', 'handle_ulw_mode_change'] \ No newline at end of file +__all__ = ['re_act', 'eval', 'image_result_formatter', 'shell_approval', 'gmail_plugin', 'calendar_plugin', 'ui_stream', 'system_reminder', 'tool_approval', 'handle_mode_change', 'auto_compact', 'prefer_write_tool', 'ulw', 'UltraWork'] \ No newline at end of file diff --git a/connectonion/useful_plugins/ulw.py b/connectonion/useful_plugins/ulw.py index 4e86008..7f3657a 100644 --- a/connectonion/useful_plugins/ulw.py +++ b/connectonion/useful_plugins/ulw.py @@ -1,24 +1,28 @@ """ -Purpose: ULW (Ultra Light Work) plugin - autonomous agent mode with turn-based checkpoints +Purpose: ULW (Ultra Light Work) plugin - keep improving until truly good enough LLM-Note: Dependencies: imports from [core/events.py] | imported by [useful_plugins/__init__.py] - Data flow: mode_change to 'ulw' → set skip_tool_approval=True → on_complete fires → continue until max turns - State/Effects: sets mode, ulw_turns, ulw_turns_used, skip_tool_approval in session - Integration: communicates with tool_approval via skip_tool_approval flag in session + Data flow: on_complete fires → check if result says "genuinely complete" → if not and rounds < max, call agent.input() again + State/Effects: tracks _ulw_rounds in current_session (private, underscore prefix) + Integration: pure plugin, no core changes needed, composable with other plugins Errors: no explicit error handling, agent.input() failures propagate ULW Mode - Ultra Light Work. When in ULW mode: -1. All tool approvals are skipped (via skip_tool_approval flag) -2. Agent keeps working until max turns reached -3. At checkpoint, user can continue, switch mode, or stop +1. Agent completes its initial task +2. Plugin intercepts and prompts agent to self-critique +3. Agent continues improving until it says "genuinely complete" or max_rounds reached Usage: from connectonion import Agent - from connectonion.useful_plugins import tool_approval, ulw + from connectonion.useful_plugins import ulw - agent = Agent("worker", plugins=[tool_approval, ulw]) + agent = Agent("worker", plugins=[ulw]) + + # With custom max rounds: + from connectonion.useful_plugins.ulw import UltraWork + agent = Agent("worker", plugins=[UltraWork(max_rounds=3)]) """ from typing import TYPE_CHECKING @@ -29,7 +33,7 @@ from ..core.agent import Agent -ULW_DEFAULT_TURNS = 100 +ULW_DEFAULT_MAX_ROUNDS = 5 ULW_CONTINUE_PROMPT = """Review what you've done so far. Consider: - Are there edge cases not handled? @@ -40,106 +44,42 @@ Continue improving, or say "genuinely complete" if nothing meaningful left to do.""" -def _log(agent: 'Agent', message: str) -> None: - """Log message via agent's logger if available.""" - if hasattr(agent, 'logger') and agent.logger: - agent.logger.print(message) +def _make_ulw_handler(max_rounds: int): + @on_complete + def ulw_keep_working(agent: 'Agent') -> None: + """If rounds remaining and not genuinely complete, start another improvement round.""" + rounds_used = agent.current_session.get('_ulw_rounds', 0) + if rounds_used >= max_rounds: + return -def handle_ulw_mode_change(agent: 'Agent', turns: int = None) -> None: - """Handle mode change to ULW. + result = agent.current_session.get('result', '') or '' + if 'genuinely complete' in result.lower(): + return - Called when frontend sends { type: 'mode_change', mode: 'ulw', turns: N } + agent.current_session['_ulw_rounds'] = rounds_used + 1 + agent.input(ULW_CONTINUE_PROMPT) - Sets up ULW state: - - mode = 'ulw' - - ulw_turns = max turns before checkpoint - - ulw_turns_used = 0 - - skip_tool_approval = True (tells tool_approval to skip all checks) + return ulw_keep_working - Args: - agent: Agent instance - turns: Max turns before checkpoint (default: 100) - """ - old_mode = agent.current_session.get('mode', 'safe') - # Set ULW state - agent.current_session['mode'] = 'ulw' - agent.current_session['ulw_turns'] = turns or ULW_DEFAULT_TURNS - agent.current_session['ulw_turns_used'] = 0 - agent.current_session['skip_tool_approval'] = True +# Default plugin: max 5 improvement rounds +ulw = [_make_ulw_handler(ULW_DEFAULT_MAX_ROUNDS)] - # Notify frontend - if agent.io: - agent.io.send({'type': 'mode_changed', 'mode': 'ulw', 'triggered_by': 'user'}) - _log(agent, f"[cyan]Mode changed: {old_mode} → ulw ({agent.current_session['ulw_turns']} turns)[/cyan]") +def UltraWork(max_rounds: int = ULW_DEFAULT_MAX_ROUNDS) -> list: + """Create a ULW plugin with custom max rounds. + Args: + max_rounds: Maximum improvement iterations before stopping (default: 5) -def _exit_ulw_mode(agent: 'Agent', new_mode: str = 'safe') -> None: - """Exit ULW mode and switch to another mode. + Returns: + Plugin list ready to pass to Agent(plugins=[...]) - Cleans up ULW state and clears skip_tool_approval flag. + Example: + agent = Agent("worker", plugins=[UltraWork(max_rounds=3)]) """ - agent.current_session.pop('skip_tool_approval', None) - agent.current_session.pop('ulw_turns', None) - agent.current_session.pop('ulw_turns_used', None) - agent.current_session['mode'] = new_mode - - if agent.io: - agent.io.send({'type': 'mode_changed', 'mode': new_mode, 'triggered_by': 'ulw_checkpoint'}) - - _log(agent, f"[cyan]Exited ULW mode → {new_mode}[/cyan]") - - -@on_complete -def ulw_keep_working(agent: 'Agent') -> None: - """If ULW mode and turns remaining, start another turn.""" - mode = agent.current_session.get('mode') - if mode != 'ulw': - return - - # Track turns - turns_used = agent.current_session.get('ulw_turns_used', 0) + 1 - agent.current_session['ulw_turns_used'] = turns_used - max_turns = agent.current_session.get('ulw_turns', ULW_DEFAULT_TURNS) - - if turns_used >= max_turns: - # Max turns reached - pause for user (if IO available) - if agent.io: - agent.io.send({ - 'type': 'ulw_turns_reached', - 'turns_used': turns_used, - 'max_turns': max_turns - }) - response = agent.io.receive() - - action = response.get('action') - if action == 'continue': - # Extend turns and continue - extend = response.get('turns', ULW_DEFAULT_TURNS) - agent.current_session['ulw_turns'] += extend - _log(agent, f"[cyan]ULW extended: +{extend} turns[/cyan]") - # Fall through to continue working - elif action == 'switch_mode': - # Switch to another mode - new_mode = response.get('mode', 'safe') - _exit_ulw_mode(agent, new_mode) - return # Stop working - else: - # Unknown action or stop - exit to safe mode - _exit_ulw_mode(agent, 'safe') - return - else: - # No IO, truly complete - return - - # Continue working - start another turn - agent.input(ULW_CONTINUE_PROMPT) - + return [_make_ulw_handler(max_rounds)] -# Export as plugin -ulw = [ulw_keep_working] -# Export mode handler for external use -__all__ = ['ulw', 'handle_ulw_mode_change', 'ULW_DEFAULT_TURNS'] +__all__ = ['ulw', 'UltraWork', 'ULW_DEFAULT_MAX_ROUNDS', 'ULW_CONTINUE_PROMPT'] diff --git a/tests/unit/test_useful_plugins_ulw.py b/tests/unit/test_useful_plugins_ulw.py new file mode 100644 index 0000000..5ba711b --- /dev/null +++ b/tests/unit/test_useful_plugins_ulw.py @@ -0,0 +1,150 @@ +"""Tests for ULW (Ultra Light Work) plugin.""" +""" +LLM-Note: Tests for useful plugins ulw + +What it tests: +- ULW plugin auto-activates without mode setup +- Stops when agent says "genuinely complete" +- Stops when max_rounds reached +- UltraWork factory creates plugin with custom max_rounds + +Components under test: +- Module: useful_plugins.ulw +""" + +from types import SimpleNamespace +from unittest.mock import MagicMock + +import importlib + +ulw_mod = importlib.import_module("connectonion.useful_plugins.ulw") + + +def make_agent(result='some work done', rounds=0): + agent = SimpleNamespace( + current_session={'result': result, '_ulw_rounds': rounds}, + input=MagicMock(), + ) + return agent + + +def test_ulw_starts_improvement_round(): + handler = ulw_mod._make_ulw_handler(max_rounds=5) + agent = make_agent(result='I built the REST API.') + + handler(agent) + + agent.input.assert_called_once_with(ulw_mod.ULW_CONTINUE_PROMPT) + assert agent.current_session['_ulw_rounds'] == 1 + + +def test_ulw_stops_when_genuinely_complete(): + handler = ulw_mod._make_ulw_handler(max_rounds=5) + agent = make_agent(result='The solution is genuinely complete, nothing left to improve.') + + handler(agent) + + agent.input.assert_not_called() + + +def test_ulw_stops_when_genuinely_complete_case_insensitive(): + handler = ulw_mod._make_ulw_handler(max_rounds=5) + agent = make_agent(result='This is GENUINELY COMPLETE.') + + handler(agent) + + agent.input.assert_not_called() + + +def test_ulw_stops_at_max_rounds(): + handler = ulw_mod._make_ulw_handler(max_rounds=3) + agent = make_agent(result='More work to do.', rounds=3) + + handler(agent) + + agent.input.assert_not_called() + + +def test_ulw_increments_rounds_each_call(): + handler = ulw_mod._make_ulw_handler(max_rounds=5) + agent = make_agent(result='Needs improvement.', rounds=2) + + handler(agent) + + assert agent.current_session['_ulw_rounds'] == 3 + agent.input.assert_called_once() + + +def test_ulw_default_max_rounds_is_5(): + assert ulw_mod.ULW_DEFAULT_MAX_ROUNDS == 5 + + +def test_ulw_plugin_is_list(): + assert isinstance(ulw_mod.ulw, list) + assert len(ulw_mod.ulw) == 1 + + +def test_ulw_handler_has_on_complete_event_type(): + handler = ulw_mod._make_ulw_handler(max_rounds=5) + assert handler._event_type == 'on_complete' + + +def test_ultra_work_factory_returns_list(): + plugin = ulw_mod.UltraWork(max_rounds=3) + assert isinstance(plugin, list) + assert len(plugin) == 1 + + +def test_ultra_work_factory_respects_max_rounds(): + plugin = ulw_mod.UltraWork(max_rounds=2) + handler = plugin[0] + + agent = make_agent(result='Needs more work.', rounds=2) + handler(agent) + + agent.input.assert_not_called() + + +def test_ultra_work_factory_continues_before_max_rounds(): + plugin = ulw_mod.UltraWork(max_rounds=2) + handler = plugin[0] + + agent = make_agent(result='Needs more work.', rounds=1) + handler(agent) + + agent.input.assert_called_once() + + +def test_ulw_handles_missing_rounds_key(): + handler = ulw_mod._make_ulw_handler(max_rounds=5) + # No _ulw_rounds key in session + agent = SimpleNamespace( + current_session={'result': 'some result'}, + input=MagicMock(), + ) + + handler(agent) + + agent.input.assert_called_once() + assert agent.current_session['_ulw_rounds'] == 1 + + +def test_ulw_handles_empty_result(): + handler = ulw_mod._make_ulw_handler(max_rounds=5) + agent = make_agent(result='') + + handler(agent) + + agent.input.assert_called_once() + + +def test_ulw_handles_none_result(): + handler = ulw_mod._make_ulw_handler(max_rounds=5) + agent = SimpleNamespace( + current_session={'result': None}, + input=MagicMock(), + ) + + handler(agent) + + agent.input.assert_called_once()