diff --git a/solstone/think/providers/openhands.py b/solstone/think/providers/openhands.py index fb9596c12..96ac78d24 100644 --- a/solstone/think/providers/openhands.py +++ b/solstone/think/providers/openhands.py @@ -22,8 +22,9 @@ import tempfile import threading import traceback import uuid -from collections.abc import Callable -from contextlib import contextmanager +from collections.abc import Callable, Iterator +from contextlib import ExitStack, contextmanager +from dataclasses import dataclass from datetime import datetime, timezone from email.utils import parsedate_to_datetime from pathlib import Path @@ -100,6 +101,8 @@ _SHELL_STDOUT_CAP = 6000 _SHELL_STDERR_CAP = 6000 _SHELL_TIMEOUT_SECONDS = 30 _COST_WARNING_TEXT = "Cost calculation failed" +_OPENHANDS_CONVERSATION_LOGGER = "openhands.sdk.conversation.impl.local_conversation" +_OPENHANDS_MAX_ITERATIONS_PREFIX = "Agent reached maximum iterations limit " _LOCAL_CONDENSER_KEEP_FIRST = 4 _GENERATE_NUM_RETRIES = 2 _GEMINI_MAX_OUTPUT_TOKENS = 65_535 @@ -1616,6 +1619,36 @@ def _suppress_litellm_cost_warnings() -> Any: logger.removeFilter(warning_filter) +@dataclass +class _DiagnosticMaxIterationsDemotionState: + demoted: bool = False + + +@contextmanager +def _demote_diagnostic_max_iterations_error() -> Iterator[ + _DiagnosticMaxIterationsDemotionState +]: + state = _DiagnosticMaxIterationsDemotionState() + + class _MaxIterationsFilter(logging.Filter): + def filter(self, record: logging.LogRecord) -> bool: + if record.levelno == logging.ERROR and record.getMessage().startswith( + _OPENHANDS_MAX_ITERATIONS_PREFIX + ): + record.levelno = logging.INFO + record.levelname = "INFO" + state.demoted = True + return True + + logger = logging.getLogger(_OPENHANDS_CONVERSATION_LOGGER) + max_iterations_filter = _MaxIterationsFilter() + try: + logger.addFilter(max_iterations_filter) + yield state + finally: + logger.removeFilter(max_iterations_filter) + + def _conversation_execution_status(conversation: Any) -> str | None: try: state = conversation.state @@ -1822,7 +1855,13 @@ async def run_cogitate( conversation.send_message(prompt_body) wall_clock_s = _wall_clock_deadline_s(timeout_seconds) wall_clock_exceeded = False - with _suppress_litellm_cost_warnings(): + demotion_state = _DiagnosticMaxIterationsDemotionState() + with ExitStack() as stack: + stack.enter_context(_suppress_litellm_cost_warnings()) + if diagnostic: + demotion_state = stack.enter_context( + _demote_diagnostic_max_iterations_error() + ) run_task = asyncio.ensure_future(conversation.arun()) _done, pending = await asyncio.wait({run_task}, timeout=wall_clock_s) if run_task in pending: @@ -1844,6 +1883,15 @@ async def run_cogitate( # generic except-Exception classification path unchanged. run_task.result() + if demotion_state.demoted: + # Verbose-only at the default CLI log level; the demoted SDK line + # still renders. + LOG.info( + "Readiness probe reached the iteration limit derived from its " + "deliberate turn budget; the preceding SDK max-iterations line is " + "expected here and was recorded at INFO rather than ERROR." + ) + if sol_executor is not None: terminal_error = sol_executor.take_terminal_error() if terminal_error is not None: diff --git a/tests/fixtures/health_logs_classified.jsonl b/tests/fixtures/health_logs_classified.jsonl index 3b38b3eef..6179f82dd 100644 --- a/tests/fixtures/health_logs_classified.jsonl +++ b/tests/fixtures/health_logs_classified.jsonl @@ -45,3 +45,5 @@ {"service":"segment_dream","stream":"stderr","line":"sol: solstone isn't running. Start it with 'journal up' and retry.","level":"error"} {"service":"heartbeat","stream":"log","line":"duration=549s outcome=success","level":"info"} {"service":"observe-transcribe","stream":"stderr","line":"DEBUG:speechbrain.utils.checkpoints:Registered checkpoint save hook for _speechbrain_save","level":"debug"} +{"service":"cortex","stream":"stderr","line":"ERROR:openhands.sdk.conversation.impl.local_conversation:Agent reached maximum iterations limit (4).","level":"error"} +{"service":"cortex","stream":"stderr","line":"INFO:openhands.sdk.conversation.impl.local_conversation:Agent reached maximum iterations limit (4).","level":"info"} diff --git a/tests/test_app_support.py b/tests/test_app_support.py index b36c0fae8..05bbcdba7 100644 --- a/tests/test_app_support.py +++ b/tests/test_app_support.py @@ -864,6 +864,36 @@ def test_recent_beats_stale_under_limit(tmp_path, monkeypatch): )[::-1] +def test_recent_errors_ignore_demoted_openhands_max_iterations_line( + tmp_path, + monkeypatch, +): + health_dir = _health_dir(tmp_path, monkeypatch) + recent = (datetime.now() - timedelta(hours=1)).isoformat(timespec="seconds") + demoted = ( + "INFO:openhands.sdk.conversation.impl.local_conversation:" + "Agent reached maximum iterations limit (4)." + ) + untouched = ( + "ERROR:openhands.sdk.conversation.impl.local_conversation:" + "Agent reached maximum iterations limit (4)." + ) + _write_log( + health_dir, + "cortex.log", + [ + f"{recent} [cortex:stderr] {demoted}", + f"{recent} [cortex:stderr] {untouched}", + ], + ) + + result = collect_recent_errors() + messages = [entry["message"] for entry in result] + + assert any(untouched in message for message in messages) + assert not any(demoted in message for message in messages) + + def test_unparseable_line_inherits_preceding_timestamp(tmp_path, monkeypatch): health_dir = _health_dir(tmp_path, monkeypatch) line_dt = datetime.now() - timedelta(hours=2) diff --git a/tests/test_cogitate_diagnostic.py b/tests/test_cogitate_diagnostic.py index f1baa58c4..cc29a903b 100644 --- a/tests/test_cogitate_diagnostic.py +++ b/tests/test_cogitate_diagnostic.py @@ -4,9 +4,12 @@ from __future__ import annotations import asyncio +import logging from pathlib import Path from types import SimpleNamespace +import pytest + from solstone.think.cogitate_contract import ( COGITATE_DIAGNOSTIC_PREAMBLE, capabilities_for_access_tier, @@ -17,8 +20,27 @@ from solstone.think.cogitate_policy import ( ) from solstone.think.providers import emit_final_tool, openhands from solstone.think.providers.shared import CANNED_GENERATE_NUM_RETRIES +from tests._logging_isolation import preserve_global_logging from tests.openhands_fakes import _REGISTERED_TOOLS, install_fake_openhands +_MAX_ITERATIONS_4 = f"{openhands._OPENHANDS_MAX_ITERATIONS_PREFIX}(4)." +_MAX_ITERATIONS_62 = f"{openhands._OPENHANDS_MAX_ITERATIONS_PREFIX}(62)." +_OTHER_SDK_ERROR = "Conversation failed to start." +_DIAGNOSTIC_MAX_ITERATIONS_INFO = ( + "Readiness probe reached the iteration limit derived from its deliberate turn " + "budget; the preceding SDK max-iterations line is expected here and was " + "recorded at INFO rather than ERROR." +) + + +class _ListHandler(logging.Handler): + def __init__(self, level: int = logging.NOTSET) -> None: + super().__init__(level) + self.records: list[logging.LogRecord] = [] + + def emit(self, record: logging.LogRecord) -> None: + self.records.append(record) + def _run_config(monkeypatch, tmp_path, **overrides): monkeypatch.setattr(openhands, "get_journal", lambda: tmp_path) @@ -59,6 +81,77 @@ def _install_emit_final_arun(fake_openhands, content: str) -> None: fake_openhands.Conversation.arun_impl = emit_final +def _agent_message(fake_openhands, content: str): + return fake_openhands.MessageEvent( + source="agent", + llm_message=SimpleNamespace(content=[SimpleNamespace(text=content)]), + ) + + +def _install_log_and_final_arun( + fake_openhands, + message: str, + content: str, + *, + emit_final: bool = True, +) -> None: + async def log_and_finish(conversation): + logging.getLogger(openhands._OPENHANDS_CONVERSATION_LOGGER).error(message) + event = ( + _emit_final_action(fake_openhands, content) + if emit_final + else _agent_message(fake_openhands, content) + ) + for callback in conversation.callbacks: + callback(event) + + fake_openhands.Conversation.arun_impl = log_and_finish + + +def _fresh_fake_openhands(monkeypatch): + fake_openhands = install_fake_openhands(monkeypatch) + emit_final_tool._EMIT_FINAL_TYPES.clear() + _REGISTERED_TOOLS.clear() + return fake_openhands + + +def _sdk_logger() -> logging.Logger: + return logging.getLogger(openhands._OPENHANDS_CONVERSATION_LOGGER) + + +def _run_with_sdk_log_capture( + config: dict, + events: list[dict], +) -> tuple[str | None, list[logging.LogRecord]]: + sdk_logger = _sdk_logger() + before_filters = tuple(sdk_logger.filters) + handler = _ListHandler(logging.INFO) + sdk_logger.addHandler(handler) + try: + result = asyncio.run(openhands.run_cogitate(config, events.append)) + finally: + sdk_logger.removeHandler(handler) + assert tuple(sdk_logger.filters) == before_filters + return result, handler.records + + +def _run_with_provider_log_capture( + config: dict, + events: list[dict], +) -> tuple[str | None, list[logging.LogRecord], list[logging.LogRecord]]: + provider_logger = logging.getLogger("solstone.think.providers.openhands") + old_provider_level = provider_logger.level + provider_handler = _ListHandler(logging.INFO) + provider_logger.addHandler(provider_handler) + provider_logger.setLevel(logging.INFO) + try: + result, sdk_records = _run_with_sdk_log_capture(config, events) + finally: + provider_logger.setLevel(old_provider_level) + provider_logger.removeHandler(provider_handler) + return result, sdk_records, provider_handler.records + + def test_diagnostic_access_tier_caps_disable_all_journal_tools(): caps = capabilities_for_access_tier("diagnostic") @@ -135,3 +228,123 @@ def test_diagnostic_run_registers_only_emit_final_and_writes_no_files( assert not any(path.exists() for path in created_temp_dirs) assert [event["event"] for event in events] == ["finish"] assert events[0]["result"] == "diagnostic ok" + + +@pytest.mark.parametrize("message", [_MAX_ITERATIONS_4, _MAX_ITERATIONS_62]) +def test_diagnostic_sdk_max_iterations_error_demotes_to_info( + monkeypatch, + tmp_path, + message: str, +): + with preserve_global_logging(): + fake_openhands = _fresh_fake_openhands(monkeypatch) + _install_log_and_final_arun(fake_openhands, message, "diagnostic ok") + config = _run_config(monkeypatch, tmp_path) + events: list[dict] = [] + + result, records = _run_with_sdk_log_capture(config, events) + + matching = [record for record in records if record.getMessage() == message] + assert result == "diagnostic ok" + assert len(matching) == 1 + assert matching[0].levelno == logging.INFO + assert matching[0].levelname == "INFO" + + +def test_non_diagnostic_sdk_max_iterations_error_stays_error(monkeypatch, tmp_path): + with preserve_global_logging(): + fake_openhands = _fresh_fake_openhands(monkeypatch) + _install_log_and_final_arun( + fake_openhands, + _MAX_ITERATIONS_4, + "non diagnostic ok", + emit_final=False, + ) + config = _run_config(monkeypatch, tmp_path) + del config["diagnostic"] + events: list[dict] = [] + + result, records = _run_with_sdk_log_capture(config, events) + + matching = [ + record for record in records if record.getMessage() == _MAX_ITERATIONS_4 + ] + assert result == "non diagnostic ok" + assert len(matching) == 1 + assert matching[0].levelno == logging.ERROR + assert matching[0].levelname == "ERROR" + + +def test_diagnostic_keeps_other_sdk_errors_at_error(monkeypatch, tmp_path): + with preserve_global_logging(): + fake_openhands = _fresh_fake_openhands(monkeypatch) + _install_log_and_final_arun(fake_openhands, _OTHER_SDK_ERROR, "diagnostic ok") + config = _run_config(monkeypatch, tmp_path) + events: list[dict] = [] + + result, records = _run_with_sdk_log_capture(config, events) + + matching = [record for record in records if record.getMessage() == _OTHER_SDK_ERROR] + assert result == "diagnostic ok" + assert len(matching) == 1 + assert matching[0].levelno == logging.ERROR + assert matching[0].levelname == "ERROR" + + +def test_diagnostic_max_iterations_filter_removed_on_normal_and_raised_exit(): + with preserve_global_logging(): + sdk_logger = _sdk_logger() + before_filters = tuple(sdk_logger.filters) + + with openhands._demote_diagnostic_max_iterations_error() as state: + assert state.demoted is False + assert tuple(sdk_logger.filters) == before_filters + + with pytest.raises(RuntimeError, match="boom"): + with openhands._demote_diagnostic_max_iterations_error(): + raise RuntimeError("boom") + assert tuple(sdk_logger.filters) == before_filters + + +def test_diagnostic_companion_info_emits_once_when_demoted(monkeypatch, tmp_path): + with preserve_global_logging(): + fake_openhands = _fresh_fake_openhands(monkeypatch) + _install_log_and_final_arun(fake_openhands, _MAX_ITERATIONS_4, "diagnostic ok") + config = _run_config(monkeypatch, tmp_path) + events: list[dict] = [] + + result, _sdk_records, provider_records = _run_with_provider_log_capture( + config, + events, + ) + + companion = [ + record + for record in provider_records + if record.getMessage() == _DIAGNOSTIC_MAX_ITERATIONS_INFO + ] + assert result == "diagnostic ok" + assert len(companion) == 1 + assert companion[0].levelno == logging.INFO + assert companion[0].levelname == "INFO" + + +def test_diagnostic_companion_info_absent_without_demotion(monkeypatch, tmp_path): + with preserve_global_logging(): + fake_openhands = _fresh_fake_openhands(monkeypatch) + _install_emit_final_arun(fake_openhands, "diagnostic ok") + config = _run_config(monkeypatch, tmp_path) + events: list[dict] = [] + + result, _sdk_records, provider_records = _run_with_provider_log_capture( + config, + events, + ) + + companion = [ + record + for record in provider_records + if record.getMessage() == _DIAGNOSTIC_MAX_ITERATIONS_INFO + ] + assert result == "diagnostic ok" + assert companion == [] diff --git a/tests/test_openhands_sdk_shape.py b/tests/test_openhands_sdk_shape.py index 322a45b6e..f1503531a 100644 --- a/tests/test_openhands_sdk_shape.py +++ b/tests/test_openhands_sdk_shape.py @@ -7,9 +7,14 @@ import asyncio import inspect import os import threading +from pathlib import Path from pydantic import Field +from solstone.think.providers.openhands import ( + _OPENHANDS_CONVERSATION_LOGGER, + _OPENHANDS_MAX_ITERATIONS_PREFIX, +) from tests._logging_isolation import preserve_global_logging os.environ.setdefault("OPENHANDS_SUPPRESS_BANNER", "1") @@ -74,6 +79,18 @@ def test_local_conversation_methods_match_provider_await_sites(monkeypatch): assert inspect.iscoroutinefunction(LocalConversation.send_message) is False +def test_local_conversation_logger_and_max_iterations_prefix_shape(monkeypatch): + monkeypatch.setenv("OPENHANDS_SUPPRESS_BANNER", "1") + with preserve_global_logging(): + from openhands.sdk.conversation.impl import local_conversation + + assert local_conversation.logger.name == _OPENHANDS_CONVERSATION_LOGGER + source_path = Path(local_conversation.__file__) + source = source_path.read_text(encoding="utf-8") + + assert _OPENHANDS_MAX_ITERATIONS_PREFIX in source + + def _shape_tool(executor): return _ShapeTool( description="SDK shape test tool",