diff --git a/Autotests/run_mandatory b/Autotests/run_mandatory index 8c76355a..db90654a 100644 --- a/Autotests/run_mandatory +++ b/Autotests/run_mandatory @@ -40,4 +40,5 @@ test_wschat.py unit/test_fileio_verified_writes.py unit/test_fileio_verified_deletes.py unit/test_helper_parsing.py +unit/test_llm_budget.py unit/test_openclaw_unit.py diff --git a/Autotests/unit/test_llm_budget.py b/Autotests/unit/test_llm_budget.py new file mode 100644 index 00000000..f8178c4e --- /dev/null +++ b/Autotests/unit/test_llm_budget.py @@ -0,0 +1,202 @@ +"""Unit tests for LLM token-budget handling in providers/ (no container, network or token).""" +import importlib.util +import json +import logging +import os +import sys +import types +from types import SimpleNamespace as NS + +import pytest + +_REPO_ROOT = os.path.normpath(os.path.join(os.path.dirname(__file__), "..", "..")) +_PROVIDERS_DIR = os.path.join(_REPO_ROOT, "providers") + +if _REPO_ROOT not in sys.path: + sys.path.insert(0, _REPO_ROOT) + + +def _load_modules(): + openai_stub = types.ModuleType("openai") + openai_stub.OpenAI = object + + config_stub = types.ModuleType("config") + config_stub.config_get_by_key = lambda key, default=None: default + + providers_stub = types.ModuleType("providers") + providers_stub.LLMProvider = object + providers_stub.registerLLMProvider = lambda name, provider: None + + stubs = {"openai": openai_stub, "config": config_stub, "providers": providers_stub} + saved = {name: sys.modules.get(name) for name in list(stubs) + ["lib_llm_ext"]} + sys.modules.update(stubs) + try: + loaded = {} + for name in ("lib_llm_ext", "openrouter", "openai_provider", "asione"): + file_name = "openai.py" if name == "openai_provider" else f"{name}.py" + spec = importlib.util.spec_from_file_location(name, os.path.join(_PROVIDERS_DIR, file_name)) + module = importlib.util.module_from_spec(spec) + if name == "lib_llm_ext": + sys.modules["lib_llm_ext"] = module + spec.loader.exec_module(module) + loaded[name] = module + return loaded + finally: + for name, module in saved.items(): + if module is None: + sys.modules.pop(name, None) + else: + sys.modules[name] = module + + +_MODULES = _load_modules() +llm = _MODULES["lib_llm_ext"] +openrouter = _MODULES["openrouter"] +openai_provider = _MODULES["openai_provider"] +asione = _MODULES["asione"] + +PROMPT = "You are an agent. :-:-:-: Write an empty line to /tmp/paths.txt" + + +def chat_response(content, finish_reason, completion_tokens=6000, reasoning_tokens=6000): + return NS( + choices=[NS(index=0, finish_reason=finish_reason, + message=NS(role="assistant", content=content))], + usage=NS(prompt_tokens=2900, completion_tokens=completion_tokens, total_tokens=2900 + completion_tokens, + prompt_tokens_details=NS(cached_tokens=2600), + completion_tokens_details=NS(reasoning_tokens=reasoning_tokens)), + ) + + +def responses_response(output_text, status, reason=None, output_tokens=120, reasoning_tokens=120): + return NS( + output_text=output_text, + status=status, + incomplete_details=NS(reason=reason) if reason else None, + usage=NS(input_tokens=1200, output_tokens=output_tokens, total_tokens=1200 + output_tokens, + input_tokens_details=NS(cached_tokens=0), + output_tokens_details=NS(reasoning_tokens=reasoning_tokens)), + ) + + +class FakeCreate: + """Returns the queued responses in order and records every call's kwargs.""" + + def __init__(self, *responses): + self._responses = list(responses) + self.calls = [] + + def __call__(self, **kwargs): + self.calls.append(kwargs) + item = self._responses.pop(0) + if isinstance(item, Exception): + raise item + return item + + +def make_openrouter(create, model="z-ai/glm-5.2"): + provider = openrouter.OpenRouterProviderImpl("OpenRouter", "OPENROUTER_API_KEY", model, "https://openrouter.ai/api/v1") + provider._client = NS(chat=NS(completions=NS(create=create))) + return provider + + +def make_base(create, name="ASICloud"): + provider = llm.AIProvider(name, "ASI_API_KEY", "minimax/minimax-m3", "https://example.invalid/v1") + provider._client = NS(chat=NS(completions=NS(create=create))) + return provider + + +def make_openai(create): + provider = openai_provider.OpenAIProviderImpl("OpenAI", "OPENAI_API_KEY", "gpt-5.5", "https://api.openai.com/v1") + provider._client = NS(responses=NS(create=create)) + return provider + + +def make_asione(create): + provider = asione.ASIOneProviderImpl("ASIOne", "ASIONE_API_KEY", "asi1-ultra", "https://api.asi1.ai/v1") + provider._client = NS(chat=NS(completions=NS(create=create))) + return provider + + +def sent_text(reply): + """Text of a single `(send "...")` command, or None if reply is not one.""" + if not (reply.startswith("(send ") and reply.endswith(")")): + return None + return json.loads(reply[len("(send "):-1]) + + +def test_normal_reply_is_returned_after_a_single_call(): + create = FakeCreate(chat_response('(send "hi")', "stop", completion_tokens=40, reasoning_tokens=30)) + assert make_openrouter(create).chat(PROMPT) == '(send "hi")' + assert len(create.calls) == 1 + + +def test_empty_reply_out_of_budget_is_explained(): + create = FakeCreate(chat_response("", "length")) + assert sent_text(make_openrouter(create).chat(PROMPT)) == llm.LLM_EMPTY_RESPONSE_MESSAGE + + +def test_empty_reply_with_stop_is_not_blamed_on_the_budget(): + create = FakeCreate(chat_response("", "stop", completion_tokens=0, reasoning_tokens=0)) + assert make_openrouter(create).chat(PROMPT) == "" + + +def test_openai_empty_reply_out_of_budget_is_explained(): + create = FakeCreate(responses_response("", "incomplete", "max_output_tokens")) + assert sent_text(make_openai(create).chat(PROMPT, max_tokens=120)) == llm.LLM_EMPTY_RESPONSE_MESSAGE + + +def test_openai_empty_reply_without_incomplete_reason_returns_empty(): + create = FakeCreate(responses_response("", "completed", output_tokens=0, reasoning_tokens=0)) + assert make_openai(create).chat(PROMPT) == "" + + +def test_asione_empty_reply_out_of_budget_is_explained(): + create = FakeCreate(chat_response("", "length")) + assert sent_text(make_asione(create).chat(PROMPT)) == llm.LLM_EMPTY_RESPONSE_MESSAGE + + +@pytest.mark.parametrize("effort, max_tokens, want_budget, want_enabled", [ + ("medium", 6000, 3000, True), + ("high", 6000, 4800, True), + ("medium", 120, 60, True), + ("none", 6000, 0, False), +]) +def test_asione_reasoning_budget(effort, max_tokens, want_budget, want_enabled): + create = FakeCreate(chat_response('(send "hi")', "stop", completion_tokens=40, reasoning_tokens=30)) + make_asione(create).chat(PROMPT, max_tokens=max_tokens, reasoning=effort) + body = create.calls[0]["extra_body"] + assert body["thinking_budget"] == want_budget + assert body["enable_thinking"] is want_enabled + + +@pytest.mark.parametrize("effort, want", [ + ("medium", {"enabled": True, "effort": "medium", "exclude": True}), + ("none", {"enabled": False, "effort": "none", "exclude": True}), +]) +def test_openrouter_reasoning_body(effort, want): + create = FakeCreate(chat_response('(send "hi")', "stop", completion_tokens=40, reasoning_tokens=30)) + make_openrouter(create).chat(PROMPT, reasoning=effort) + assert create.calls[0]["extra_body"]["reasoning"] == want + + +def test_usage_is_logged_at_info(caplog): + caplog.set_level(logging.INFO) + create = FakeCreate(chat_response('(send "hi")', "stop", completion_tokens=40, reasoning_tokens=30)) + make_openrouter(create).chat(PROMPT) + usage = [r for r in caplog.records if "[LLM_USAGE]" in r.getMessage()] + assert usage and all(r.levelno == logging.INFO for r in usage) + assert "finish_reason=stop" in usage[0].getMessage() + + +def test_openai_usage_is_logged_at_info(caplog): + caplog.set_level(logging.INFO) + create = FakeCreate(responses_response('(send "hi")', "completed", output_tokens=40, reasoning_tokens=30)) + make_openai(create).chat(PROMPT) + usage = [r for r in caplog.records if "[LLM_USAGE]" in r.getMessage()] + assert usage and all(r.levelno == logging.INFO for r in usage) + + +def test_api_error_returns_empty_string(): + create = FakeCreate(RuntimeError("401 invalid api key")) + assert make_openrouter(create).chat(PROMPT) == ""