Something went wrong. Try again.
personal memory agent
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448# SPDX-License-Identifier: AGPL-3.0-only# Copyright (c) 2026 sol pbc
from __future__ import annotations
import asyncioimport jsonimport tracebackfrom types import SimpleNamespace
import httpximport openaiimport pytest
from solstone.think.models import LOCAL_MODELfrom solstone.think.providers import openhandsfrom solstone.think.providers.cli import ProviderKeyMissingError, QuotaExhaustedErrorfrom solstone.think.providers.local_endpoint import ( LOCAL_ENDPOINT_CONTRACT_COPY, LOCAL_ENDPOINT_UNREACHABLE_COPY, LocalEndpoint,)from solstone.think.providers.shared import PROVIDER_ERROR_TEXT_CAP_CHARSfrom solstone.think.talents import TalentHookErrorfrom tests.openhands_fakes import install_fake_openhands
@pytest.fixturedef fake_openhands(monkeypatch): return install_fake_openhands(monkeypatch)
@pytest.fixturedef run_env(monkeypatch, tmp_path): monkeypatch.setattr(openhands, "get_journal", lambda: tmp_path) monkeypatch.setattr(openhands, "get_project_root", lambda: tmp_path) monkeypatch.setattr(openhands, "now_ms", lambda: 123456) monkeypatch.setenv("OPENAI_API_KEY", "test-key") return { "provider": "openai", "model": "gpt-5", "prompt": "Do the work.", "session_id": "11111111-1111-1111-1111-111111111111", "day": "20260522", }
def _openai_response(status_code: int, headers: dict[str, str] | None = None): request = httpx.Request("POST", "https://api.openai.com/v1/responses") return httpx.Response(status_code, request=request, headers=headers or {})
def test_retry_delay_ms_reads_retry_after_seconds_header(): exc = SimpleNamespace( response=SimpleNamespace(headers={"retry-after": "30"}), )
assert openhands._retry_delay_ms(exc) == 30000
def test_retry_delay_ms_returns_none_without_header(): exc = SimpleNamespace(response=SimpleNamespace(headers={}))
assert openhands._retry_delay_ms(exc) is None
def test_unwrap_provider_exception_prefers_cause(): provider_exc = RuntimeError("provider") wrapper = RuntimeError("wrapper") wrapper.__cause__ = provider_exc
assert openhands._unwrap_provider_exception(wrapper) is provider_exc
def test_unwrap_provider_exception_uses_context_without_cause(): provider_exc = RuntimeError("provider") wrapper = RuntimeError("wrapper") wrapper.__context__ = provider_exc
assert openhands._unwrap_provider_exception(wrapper) is provider_exc
def test_run_cogitate_quota_path_raises_quota_without_error_event( fake_openhands, run_env,): quota_exc = openai.RateLimitError( "rate limited", response=_openai_response(429, {"retry-after": "30"}), body={}, )
async def fail(_conversation): raise quota_exc
fake_openhands.Conversation.arun_impl = fail events: list[dict] = []
with pytest.raises(QuotaExhaustedError) as raised: asyncio.run(openhands.run_cogitate(run_env, events.append))
assert raised.value.retry_delay_ms == 30000 assert events == []
def test_run_cogitate_generic_error_emits_event_and_marks_evented( fake_openhands, run_env,): generic_exc = RuntimeError("boom")
async def fail(conversation): usage = conversation.agent.llm.metrics.accumulated_token_usage usage.prompt_tokens = 12 usage.completion_tokens = 4 conversation.agent.llm.metrics.token_usages = [object()] raise generic_exc
fake_openhands.Conversation.arun_impl = fail events: list[dict] = []
with pytest.raises(RuntimeError) as raised: asyncio.run(openhands.run_cogitate(run_env, events.append))
assert raised.value is generic_exc assert getattr(generic_exc, "_evented") is True assert len(events) == 1 assert events[0]["event"] == "error" assert events[0]["error"] == "boom" assert events[0]["reason_code"] == "unknown" assert events[0]["provider"] == "openai" assert "RuntimeError: boom" in events[0]["trace"] assert events[0]["usage"]["total_tokens"] > 0 assert events[0]["ts"] == 123456
def test_run_cogitate_cloud_error_event_caps_error_text( fake_openhands, run_env,): generic_exc = RuntimeError("x" * (PROVIDER_ERROR_TEXT_CAP_CHARS + 1))
async def fail(_conversation): raise generic_exc
fake_openhands.Conversation.arun_impl = fail events: list[dict] = []
with pytest.raises(RuntimeError) as raised: asyncio.run(openhands.run_cogitate(run_env, events.append))
assert raised.value is generic_exc assert len(events) == 1 assert len(events[0]["error"]) <= PROVIDER_ERROR_TEXT_CAP_CHARS assert events[0]["error"] == str(generic_exc)[:PROVIDER_ERROR_TEXT_CAP_CHARS]
def test_run_cogitate_talent_hook_error_propagates_without_provider_event( fake_openhands, run_env,): hook_exc = TalentHookError( "post", "broken_hook", "chat", RuntimeError("hook exploded"), )
async def fail(_conversation): raise hook_exc
fake_openhands.Conversation.arun_impl = fail events: list[dict] = []
with pytest.raises(TalentHookError) as raised: asyncio.run(openhands.run_cogitate(run_env, events.append))
assert raised.value is hook_exc assert events == [] assert not getattr(hook_exc, "_evented", False)
def test_run_cogitate_error_before_usage_baseline_omits_usage( fake_openhands, run_env, monkeypatch,): build_exc = RuntimeError("llm exploded")
def fail_build(_provider, _model, *, num_retries=None, **_kwargs): raise build_exc
monkeypatch.setattr(openhands, "_build_llm", fail_build) events: list[dict] = []
with pytest.raises(RuntimeError) as raised: asyncio.run(openhands.run_cogitate(run_env, events.append))
assert raised.value is build_exc assert getattr(build_exc, "_evented") is True assert len(events) == 1 assert events[0]["event"] == "error" assert events[0]["error"] == "llm exploded" assert events[0]["reason_code"] == "unknown" assert events[0]["provider"] == "openai" assert "RuntimeError: llm exploded" in events[0]["trace"] assert "usage" not in events[0] assert events[0]["ts"] == 123456
def test_run_cogitate_missing_key_fails_pre_network( fake_openhands, run_env, monkeypatch, tmp_path,): monkeypatch.delenv(openhands._API_KEY_ENV["openai"], raising=False) events: list[dict] = []
with pytest.raises(ProviderKeyMissingError) as raised: asyncio.run(openhands.run_cogitate(run_env, events.append))
assert raised.value.provider == "openai" assert raised.value.env_key == openhands._API_KEY_ENV["openai"] assert raised.value.reason_code == "provider_key_missing" assert len(events) == 1 assert events[0]["event"] == "error" assert events[0]["reason_code"] == "provider_key_missing" assert events[0]["provider"] == "openai" assert fake_openhands.LLM.instances == [] assert fake_openhands.Conversation.instances == [] assert not ( tmp_path / ".cache" / "cogitate-history" / run_env["session_id"] ).exists()
def test_run_cogitate_local_byo_error_event_uses_fixed_copy_and_redacts( fake_openhands, run_env, monkeypatch,): token = "test-token-PLACEHOLDER" endpoint = LocalEndpoint( base_url="http://byo.example/openai", served_model_id="served-model", credential=token, is_bundled=False, )
class BadRequestError(RuntimeError): status_code = 400
async def fail(_conversation): raise BadRequestError(f"bad request with {token}")
monkeypatch.setattr( "solstone.think.providers.local_endpoint.resolve_local_endpoint", lambda: endpoint, ) fake_openhands.Conversation.arun_impl = fail events: list[dict] = [] local_env = {**run_env, "provider": "local", "model": LOCAL_MODEL}
with pytest.raises(BadRequestError): asyncio.run(openhands.run_cogitate(local_env, events.append))
assert len(events) == 1 assert events[0]["event"] == "error" assert events[0]["error"] == LOCAL_ENDPOINT_CONTRACT_COPY assert events[0]["reason_code"] == "local_endpoint_contract_failed" assert token not in events[0]["trace"]
def test_run_cogitate_local_byo_context_body_event_redacts( fake_openhands, run_env, monkeypatch,): from tests.test_local import _FAKE_F2_PROMPT_OVERFLOW_BODY
token = "SENTINEL-BYO-CONTEXT-CRED-219a" endpoint = LocalEndpoint( base_url="http://byo.example/openai", served_model_id="served-model", credential=token, is_bundled=False, )
class BadRequestError(RuntimeError): status_code = 400
def __init__(self) -> None: body = _FAKE_F2_PROMPT_OVERFLOW_BODY + f" {token}" + ("x" * 6000) super().__init__(body) self.message = body self.body = body
async def fail(_conversation): raise BadRequestError()
monkeypatch.setattr( "solstone.think.providers.local_endpoint.resolve_local_endpoint", lambda: endpoint, ) fake_openhands.Conversation.arun_impl = fail events: list[dict] = [] local_env = {**run_env, "provider": "local", "model": LOCAL_MODEL}
with pytest.raises(BadRequestError) as raised: asyncio.run(openhands.run_cogitate(local_env, events.append))
assert len(events) == 1 assert events[0]["reason_code"] == "context_window_exceeded" assert len(events[0]["error"]) <= PROVIDER_ERROR_TEXT_CAP_CHARS assert token not in json.dumps(events) assert token not in str(raised.value)
def test_run_cogitate_local_window_threaded_once( fake_openhands, run_env, monkeypatch,): endpoint = LocalEndpoint( base_url="https://spp.example.test", served_model_id="confidential-model", credential="confidential-token", is_bundled=False, is_confidential=True, ) window_calls = [] llm_calls = [] agent_calls = [] original_build_llm = openhands._build_llm
def resolve_window(resolved_endpoint): window_calls.append(resolved_endpoint) return 16384
def spy_build_llm(*args, **kwargs): llm_calls.append(kwargs) return original_build_llm(*args, **kwargs)
def spy_build_agent(**kwargs): agent_calls.append(kwargs) return SimpleNamespace( llm=kwargs["llm"], tools=kwargs["tool_specs"], include_default_tools=kwargs["include_default_tools"], system_prompt=kwargs["system_prompt"], condenser=SimpleNamespace(max_tokens=kwargs["condenser_max_tokens"]), )
monkeypatch.setattr( "solstone.think.providers.local_endpoint.resolve_local_endpoint", lambda: endpoint, ) monkeypatch.setattr( "solstone.think.providers.local_endpoint.resolve_endpoint_served_window", resolve_window, ) monkeypatch.setattr( "solstone.think.services.spp_transport.confidential_egress_base_url", lambda _base_url: "http://127.0.0.1:4567", ) monkeypatch.setattr(openhands, "_build_llm", spy_build_llm) monkeypatch.setattr(openhands, "_build_cogitate_agent", spy_build_agent) local_env = {**run_env, "provider": "local", "model": LOCAL_MODEL}
asyncio.run(openhands.run_cogitate(local_env, lambda _event: None))
assert window_calls == [endpoint] assert llm_calls[0]["endpoint"] is endpoint assert llm_calls[0]["served_window"] == 16384 assert agent_calls[0][ "condenser_max_tokens" ] == openhands._local_condenser_max_tokens(16384) llm = fake_openhands.LLM.instances[0] assert llm.max_input_tokens == 16384 assert llm.max_output_tokens == 4096 assert llm.litellm_extra_body == { "chat_template_kwargs": {"enable_thinking": False} }
@pytest.mark.parametrize("exc_type", [httpx.ConnectError, httpx.ConnectTimeout])def test_run_cogitate_byo_connection_error_classifies_unreachable_no_wall_clock( fake_openhands, run_env, monkeypatch, exc_type,): sentinel = "SENTINEL-BYO-CRED-9f3a2b" endpoint = LocalEndpoint( base_url="http://byo.example/openai", served_model_id="served-model", credential=sentinel, is_bundled=False, )
async def fail(_conversation): raise exc_type(f"connection failed {sentinel}")
monkeypatch.setattr( "solstone.think.providers.local_endpoint.resolve_local_endpoint", lambda: endpoint, ) fake_openhands.Conversation.arun_impl = fail events: list[dict] = [] local_env = {**run_env, "provider": "local", "model": LOCAL_MODEL}
with pytest.raises(exc_type) as raised: asyncio.run(openhands.run_cogitate(local_env, events.append))
assert len(events) == 1 assert events[0]["event"] == "error" assert events[0]["reason_code"] == "local_endpoint_unreachable" assert events[0]["error"] == LOCAL_ENDPOINT_UNREACHABLE_COPY assert "wall_clock_exceeded" not in {event.get("reason_code") for event in events} assert sentinel not in json.dumps(events) assert all(sentinel not in event.get("trace", "") for event in events) assert sentinel not in str(raised.value) serialized = "".join( traceback.format_exception( type(raised.value), raised.value, raised.value.__traceback__, ) ) assert sentinel not in serialized
def test_run_cogitate_propagates_quota_unwrapped(fake_openhands, run_env): quota = QuotaExhaustedError("quota", retry_delay_ms=111)
async def fail(_conversation): raise quota
fake_openhands.Conversation.arun_impl = fail events: list[dict] = []
with pytest.raises(QuotaExhaustedError) as raised: asyncio.run(openhands.run_cogitate(run_env, events.append))
assert raised.value is quota assert raised.value.retry_delay_ms == 111 assert events == []