import importlib.util import tempfile import time import unittest from pathlib import Path from unittest.mock import patch ADAPTER_PATH = Path(__file__).with_name("glm-kimi-adapter.py") SPEC = importlib.util.spec_from_file_location("glm_kimi_adapter", ADAPTER_PATH) ADAPTER = importlib.util.module_from_spec(SPEC) SPEC.loader.exec_module(ADAPTER) def response(message, finish_reason="tool_calls", tokens=10): return { "model": "test/model", "usage": { "prompt_tokens": tokens, "completion_tokens": 1, "total_tokens": tokens + 1, "prompt_tokens_details": {"cached_tokens": 0}, "completion_tokens_details": {"reasoning_tokens": 0}, }, "choices": [{"finish_reason": finish_reason, "message": message}], } def subagent_result(agent_id): return { "result": f"Ergebnis {agent_id}", "usage": { "prompt_tokens": 2, "completion_tokens": 1, "total_tokens": 3, "cached_tokens": 0, "reasoning_tokens": 0, }, "turns": 1, "tool_calls": 0, "errors": [], "status": "completed", } class AdapterTests(unittest.TestCase): def setUp(self): self.provider = { "base_url": "https://example.invalid/v1", "__id": "test", } self.liveness = ADAPTER.LivenessMonitor(interval_seconds=0) def test_timeout_zero_is_forwarded_as_no_requests_timeout(self): captured = {} class FakeResponse: status_code = 200 text = "" @staticmethod def json(): return {"ok": True} def fake_post(*args, **kwargs): captured["timeout"] = kwargs["timeout"] return FakeResponse() with patch.object(ADAPTER.requests, "post", side_effect=fake_post): result = ADAPTER.post_chat_completion( "https://example.invalid", {}, {}, 0, self.liveness, "test", "Testaufruf", ) self.assertEqual({"ok": True}, result) self.assertIsNone(captured["timeout"]) def test_requested_tensorx_models_have_explicit_effort_mapping(self): self.assertEqual("thinking", ADAPTER.MODEL_EFFORT_TYPE["qwen"]) self.assertEqual("thinking", ADAPTER.MODEL_EFFORT_TYPE["z-ai"]) def test_custom_mode_exposes_spawn_subagent(self): captured = {} def fake_post(url, headers, body, timeout, liveness, activity_id, activity_label): captured["tools"] = body["tools"] return response( {"role": "assistant", "content": "Fertig."}, finish_reason="stop", ) with tempfile.TemporaryDirectory() as temp_dir, \ patch.object(ADAPTER, "post_chat_completion", side_effect=fake_post): result = ADAPTER.run_agent_loop( provider=self.provider, model="qwen/qwen3.8-flash-next", system_prompt="System", user_prompt="Aufgabe", api_key="key", effort="high", root=temp_dir, output_dir=temp_dir, max_turns=0, temperature=1.0, timeout=0, liveness=self.liveness, mode="custom", ) tool_names = { tool["function"]["name"] for tool in captured["tools"] } self.assertIn("spawn_subagent", tool_names) self.assertFalse(result["is_error"]) def test_liveness_monitor_reports_active_main_and_subagent_operations(self): lines = [] monitor = ADAPTER.LivenessMonitor(interval_seconds=0.02) with patch.object(ADAPTER, "log_stderr", side_effect=lines.append): monitor.set("main", "Hauptagent wartet auf Subagenten") monitor.set("subagent-1", "Subagent 1 wartet auf API-Antwort") monitor.start() time.sleep(0.06) monitor.stop() output = "\n".join(lines) self.assertIn("LIFESIGN: Prozess lebt", output) self.assertIn("Hauptagent wartet auf Subagenten", output) self.assertIn("Subagent 1 wartet auf API-Antwort", output) self.assertIn("keinen serverseitigen Fortschritt", output) def test_subagents_from_one_turn_run_in_parallel_without_count_limit(self): first_message = { "role": "assistant", "content": "Ich delegiere.", "tool_calls": [ { "id": f"spawn-{index}", "type": "function", "function": { "name": "spawn_subagent", "arguments": ( '{"description":"Aufgabe %d",' '"subagent_type":"explore"}' % index ), }, } for index in range(12) ], } responses = [ response(first_message), response( {"role": "assistant", "content": "Fertig."}, finish_reason="stop", ), ] starts = [] def fake_post(*args, **kwargs): return responses.pop(0) def fake_subagent(*args, **kwargs): agent_id = args[10] starts.append(time.monotonic()) time.sleep(0.15) return subagent_result(agent_id) with tempfile.TemporaryDirectory() as temp_dir: started = time.monotonic() with patch.object(ADAPTER, "post_chat_completion", side_effect=fake_post), \ patch.object(ADAPTER, "run_subagent", side_effect=fake_subagent): result = ADAPTER.run_agent_loop( provider=self.provider, model="test/model", system_prompt="System", user_prompt="Aufgabe", api_key="key", effort="high", root=temp_dir, output_dir=temp_dir, max_turns=0, temperature=1.0, timeout=0, liveness=self.liveness, mode="builtin", subagent_max_turns=0, ) elapsed = time.monotonic() - started self.assertEqual(12, result["subagent_stats"]["spawned"]) self.assertEqual(12, result["subagent_stats"]["completed"]) self.assertEqual(0, result["subagent_stats"]["failed"]) self.assertLess(max(starts) - min(starts), 0.12) self.assertLess(elapsed, 0.6) def test_api_error_is_not_masked_by_earlier_content(self): first_message = { "role": "assistant", "content": "Zwischenstand", "tool_calls": [{ "id": "list-1", "type": "function", "function": { "name": "list_directory", "arguments": '{"path":""}', }, }], } calls = [response(first_message), RuntimeError("Transportfehler")] def fake_post(*args, **kwargs): item = calls.pop(0) if isinstance(item, Exception): raise item return item with tempfile.TemporaryDirectory() as temp_dir, \ patch.object(ADAPTER, "post_chat_completion", side_effect=fake_post): result = ADAPTER.run_agent_loop( provider=self.provider, model="test/model", system_prompt="System", user_prompt="Aufgabe", api_key="key", effort="high", root=temp_dir, output_dir=temp_dir, max_turns=0, temperature=1.0, timeout=0, liveness=self.liveness, mode="solo", ) self.assertEqual("Zwischenstand", result["result"]) self.assertTrue(result["is_error"]) self.assertEqual("error", result["subtype"]) self.assertIn("Transportfehler", result["errors"][0]) if __name__ == "__main__": unittest.main()