Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions connectonion/useful_plugins/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -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']
__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']
138 changes: 39 additions & 99 deletions connectonion/useful_plugins/ulw.py
Original file line number Diff line number Diff line change
@@ -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
Expand All @@ -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?
Expand All @@ -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']
150 changes: 150 additions & 0 deletions tests/unit/test_useful_plugins_ulw.py
Original file line number Diff line number Diff line change
@@ -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()
Loading