"""Direkter Text-API-Pfad (MiniMax) + RAM-Gate für opencode-Spawns.""" import asyncio import httpx import pytest import agents TOPIC = "t" # ── Routing: wann läuft ein Call über die API statt über opencode? ─────────────────── def test_use_text_api_routing(monkeypatch): monkeypatch.setenv("MINIMAX_API_KEY", "k") monkeypatch.delenv("CREATOR_TEXT_API", raising=False) api = agents._use_text_api assert api("minimax", "minimax/MiniMax-M3", "none", None) is True assert api("minimax", "minimax-kalt/MiniMax-M2.7-highspeed", "none", None) is True # Tools, Streaming, andere Provider/Modelle → Prozess-Pfad assert api("minimax", "minimax/MiniMax-M3", "files", None) is False assert api("minimax", "minimax/MiniMax-M3", "none", lambda s: None) is False assert api("claude", "claude-sonnet-4-6", "none", None) is False assert api("lokal", "ollama/qwen3.5:9b", "none", None) is False # Kill-Switch und fehlender Key → Fallback monkeypatch.setenv("CREATOR_TEXT_API", "0") assert api("minimax", "minimax/MiniMax-M3", "none", None) is False monkeypatch.delenv("CREATOR_TEXT_API") monkeypatch.delenv("MINIMAX_API_KEY") assert api("minimax", "minimax/MiniMax-M3", "none", None) is False async def test_run_agent_dispatches_to_api(monkeypatch): """none+minimax → API-Runner; opencode wird nicht angefasst (auch kein which-Check).""" called = {} async def fake_api(agent_key, prompt, timeout, model, label=""): called["api"] = (agent_key, model) return 0, "out", "", {"input": 1, "output": 1, "reasoning": 0, "cache_read": 0, "cache_write": 0} async def fail_oc(*a, **kw): raise AssertionError("opencode-Pfad darf nicht laufen") monkeypatch.delenv("CREATOR_FAKE_AGENTS", raising=False) monkeypatch.setenv("MINIMAX_API_KEY", "k") monkeypatch.setattr(agents, "_run_text_api", fake_api) monkeypatch.setattr(agents, "_run_opencode", fail_oc) monkeypatch.setattr(agents.shutil, "which", lambda c: None) # API-Pfad braucht kein Binary monkeypatch.setattr(agents, "resolve_role", lambda p, r: ("minimax", "minimax/MiniMax-M3")) rc, out, err = await agents.run_agent("blocks-t-a", "p", 5, provider="minimax", role="judge") assert (rc, out) == (0, "out") and called["api"][1] == "minimax/MiniMax-M3" async def test_run_agent_kill_switch_uses_opencode(monkeypatch): called = {} async def fake_oc(agent_key, prompt, timeout, provider, model, capabilities, on_line=None, label=""): called["oc"] = agent_key return 0, "out", "" monkeypatch.delenv("CREATOR_FAKE_AGENTS", raising=False) monkeypatch.setenv("MINIMAX_API_KEY", "k") monkeypatch.setenv("CREATOR_TEXT_API", "0") monkeypatch.setattr(agents, "_run_opencode", fake_oc) monkeypatch.setattr(agents.shutil, "which", lambda c: "/bin/true") monkeypatch.setattr(agents, "resolve_role", lambda p, r: ("minimax", "minimax/MiniMax-M3")) rc, *_ = await agents.run_agent("blocks-t-b", "p", 5, provider="minimax") assert rc == 0 and called["oc"] == "blocks-t-b" async def test_run_agent_api_tokens_in_event_meta(monkeypatch): """API-Tokens landen im Event-Meta; die OpenCode-Session-DB wird NICHT konsultiert.""" recorded = [] async def sink(**kw): recorded.append(kw) async def fake_api(agent_key, prompt, timeout, model, label=""): return 0, "out", "", {"input": 5, "output": 2, "reasoning": 0, "cache_read": 100, "cache_write": 0} monkeypatch.delenv("CREATOR_FAKE_AGENTS", raising=False) monkeypatch.setenv("MINIMAX_API_KEY", "k") monkeypatch.setattr(agents, "on_event", sink) monkeypatch.setattr(agents, "_run_text_api", fake_api) monkeypatch.setattr(agents, "_session_tokens", lambda k: pytest.fail("Session-DB-Lookup im API-Pfad")) monkeypatch.setattr(agents.shutil, "which", lambda c: "/bin/true") monkeypatch.setattr(agents, "resolve_role", lambda p, r: ("minimax", "minimax/MiniMax-M3")) rc, *_ = await agents.run_agent("blocks-t-tok", "p", 5, provider="minimax", scope=TOPIC) assert rc == 0 assert recorded and recorded[0]["meta"]["tokens"] == { "input": 5, "output": 2, "reasoning": 0, "cache_read": 100, "cache_write": 0} # ── API-Runner: Request-Bau, Antwort-Extraktion, Fehlerfälle ───────────────────────── class _FakeResp: def __init__(self, status_code=200, data=None, text=""): self.status_code = status_code self._data = data or {} self.text = text def json(self): return self._data def _fake_client(monkeypatch, seen, resp=None, exc=None): class _Client: def __init__(self, **kw): seen["client_kw"] = kw async def __aenter__(self): return self async def __aexit__(self, *a): return False async def post(self, url, json=None, headers=None): seen.update(url=url, body=json, headers=headers) if exc is not None: raise exc return resp monkeypatch.setattr(agents.httpx, "AsyncClient", _Client) async def test_api_request_and_response(monkeypatch): monkeypatch.setenv("MINIMAX_API_KEY", "geheim") seen = {} resp = _FakeResp(data={ "content": [{"type": "thinking", "thinking": "hm"}, {"type": "text", "text": "A"}, {"type": "text", "text": "B"}], "usage": {"input_tokens": 5, "output_tokens": 2, "cache_read_input_tokens": 100, "cache_creation_input_tokens": 1}, "stop_reason": "end_turn"}) _fake_client(monkeypatch, seen, resp=resp) rc, out, err, tokens = await agents._run_text_api("k", "PROMPT", 5, "minimax-kalt/MiniMax-M3") assert (rc, out, err) == (0, "AB", "") # Thinking-Block übersprungen assert tokens == {"input": 5, "output": 2, "reasoning": 0, "cache_read": 100, "cache_write": 1} assert seen["url"] == agents._API_URL assert seen["headers"]["x-api-key"] == "geheim" assert seen["headers"]["anthropic-version"] == agents._API_VERSION b = seen["body"] assert b["model"] == "MiniMax-M3" and b["max_tokens"] == agents._API_MAX_TOKENS assert b["messages"] == [{"role": "user", "content": "PROMPT"}] assert b["temperature"] == 0.2 and b["thinking"] == {"type": "disabled"} # kalt-Route async def test_api_model_options_per_route(monkeypatch): monkeypatch.setenv("MINIMAX_API_KEY", "k") seen = {} resp = _FakeResp(data={"content": [{"type": "text", "text": "x"}], "usage": {}}) _fake_client(monkeypatch, seen, resp=resp) await agents._run_text_api("k", "p", 5, "minimax-kalt/MiniMax-M2.7-highspeed") assert seen["body"]["temperature"] == 0.3 and "thinking" not in seen["body"] await agents._run_text_api("k", "p", 5, "minimax/MiniMax-M3") # nativ: Endpunkt-Defaults assert "temperature" not in seen["body"] and "thinking" not in seen["body"] async def test_api_errors(monkeypatch): monkeypatch.setenv("MINIMAX_API_KEY", "k") seen = {} _fake_client(monkeypatch, seen, resp=_FakeResp(status_code=500, text="kaputt")) rc, out, err, tokens = await agents._run_text_api("k", "p", 5, "minimax/MiniMax-M3") assert rc == 1 and out == "" and "HTTP 500" in err and tokens is None _fake_client(monkeypatch, seen, exc=httpx.ConnectError("down")) rc, _, err, _ = await agents._run_text_api("k", "p", 5, "minimax/MiniMax-M3") assert rc == 1 and "ConnectError" in err _fake_client(monkeypatch, seen, exc=httpx.ReadTimeout("langsam")) with pytest.raises(asyncio.TimeoutError): await agents._run_text_api("k", "p", 5, "minimax/MiniMax-M3") # leere Antwort (nur Thinking) → rc 1, Tokens bleiben sichtbar resp = _FakeResp(data={"content": [{"type": "thinking", "thinking": "…"}], "usage": {"input_tokens": 3}, "stop_reason": "end_turn"}) _fake_client(monkeypatch, seen, resp=resp) rc, _, err, tokens = await agents._run_text_api("k", "p", 5, "minimax/MiniMax-M3") assert rc == 1 and "empty response" in err and tokens["input"] == 3 async def test_api_truncation_flagged(monkeypatch): monkeypatch.setenv("MINIMAX_API_KEY", "k") seen = {} resp = _FakeResp(data={"content": [{"type": "text", "text": "halb"}], "usage": {}, "stop_reason": "max_tokens"}) _fake_client(monkeypatch, seen, resp=resp) rc, out, err, _ = await agents._run_text_api("k", "p", 5, "minimax/MiniMax-M3") assert rc == 0 and out == "halb" and "max_tokens" in err # ── RAM-Gate ───────────────────────────────────────────────────────────────────────── @pytest.fixture def ram_gate(monkeypatch): monkeypatch.setattr(agents, "RAM_MIN_FREE_PCT", 20) monkeypatch.setattr(agents, "_RAM_POLL_S", 0.01) monkeypatch.setattr(agents, "_opencode_recent_starts", []) return monkeypatch async def test_ram_gate_admits_with_free_ram(ram_gate): ram_gate.setattr(agents, "_opencode_running", 5) ram_gate.setattr(agents, "_meminfo", lambda: (4_000_000, 8_000_000)) # 50 % frei assert await agents._ram_gate("k") is True assert len(agents._opencode_recent_starts) == 1 # Commit registriert async def test_ram_gate_waits_when_low(ram_gate): ram_gate.setattr(agents, "_opencode_running", 5) ram_gate.setattr(agents, "_meminfo", lambda: (800_000, 8_000_000)) # 10 % frei with pytest.raises(asyncio.TimeoutError): await asyncio.wait_for(agents._ram_gate("k"), 0.1) # RAM wird frei → Gate lässt nach ≥1 Poll durch vals = iter([(800_000, 8_000_000)]) ram_gate.setattr(agents, "_meminfo", lambda: next(vals, (4_000_000, 8_000_000))) assert await agents._ram_gate("k") is True async def test_ram_gate_floor_and_fail_open(ram_gate): ram_gate.setattr(agents, "_meminfo", lambda: (100_000, 8_000_000)) # fast nichts frei ram_gate.setattr(agents, "_opencode_running", 1) # unter Floor assert await agents._ram_gate("k") is True ram_gate.setattr(agents, "_opencode_running", 5) ram_gate.setattr(agents, "_meminfo", lambda: None) # kein /proc/meminfo assert await agents._ram_gate("k") is True ram_gate.setattr(agents, "RAM_MIN_FREE_PCT", 0) # Gate aus ram_gate.setattr(agents, "_meminfo", lambda: pytest.fail("Gate aus liest kein meminfo")) assert await agents._ram_gate("k") is True async def test_ram_gate_commit_accounting(ram_gate): """Knapp über der Schwelle, aber 2 frische Zulassungen → deren geschätzter RSS zählt.""" import time as _time ram_gate.setattr(agents, "_opencode_running", 5) ram_gate.setattr(agents, "_meminfo", lambda: (1_700_000, 8_000_000)) # 21 % frei agents._opencode_recent_starts.extend([_time.monotonic(), _time.monotonic()]) with pytest.raises(asyncio.TimeoutError): await asyncio.wait_for(agents._ram_gate("k"), 0.1) async def test_ram_gate_cancelled_scope(ram_gate): ram_gate.setattr(agents, "_opencode_running", 5) ram_gate.setattr(agents, "_meminfo", lambda: (800_000, 8_000_000)) agents.cancel_scope("blocks-cxl-") try: assert await agents._ram_gate("blocks-cxl-x") is False finally: agents.clear_scope("blocks-cxl-") def test_meminfo_reads_proc(): mem = agents._meminfo() assert mem is not None and 0 < mem[0] <= mem[1] # Linux-Testumgebung