From 2bba7f56e3097ffd618a46b1e6c13c80cf2a5691 Mon Sep 17 00:00:00 2001 From: Ed_ Date: Sun, 5 Jul 2026 19:49:37 -0400 Subject: [PATCH] refactor(ai_client): remove gemini_cli provider from ai_client Drop the standalone Gemini CLI adapter from the AI client surface: delete the import, the PROVIDERS entry, the module state, the 3 functions (_list_gemini_cli_models, _send_cli_round_result, _send_gemini_cli), and the 8 dispatch branches. PROVIDERS now has 7 entries; _gemini_sdk remainder is unaffected. --- src/ai_client.py | 185 +------------------- src/gemini_cli_adapter.py | 193 --------------------- tests/test_ai_client_cli.py | 33 ---- tests/test_gemini_cli_adapter.py | 85 --------- tests/test_gemini_cli_adapter_parity.py | 49 ------ tests/test_gemini_cli_edge_cases.py | 41 ----- tests/test_gemini_cli_integration.py | 32 ---- tests/test_gemini_cli_parity_regression.py | 15 -- tests/test_mock_gemini_cli.py | 69 -------- 9 files changed, 6 insertions(+), 696 deletions(-) delete mode 100644 src/gemini_cli_adapter.py delete mode 100644 tests/test_ai_client_cli.py delete mode 100644 tests/test_gemini_cli_adapter.py delete mode 100644 tests/test_gemini_cli_adapter_parity.py delete mode 100644 tests/test_gemini_cli_edge_cases.py delete mode 100644 tests/test_gemini_cli_integration.py delete mode 100644 tests/test_gemini_cli_parity_regression.py delete mode 100644 tests/test_mock_gemini_cli.py diff --git a/src/ai_client.py b/src/ai_client.py index 19112cd6..fd3f36ae 100644 --- a/src/ai_client.py +++ b/src/ai_client.py @@ -46,7 +46,6 @@ from src import performance_monitor from src import project_manager from src import provider_state from src.events import EventEmitter -from src.gemini_cli_adapter import GeminiCliAdapter from src.project_files import FileItem from src.tool_presets import ToolPreset, Tool from src.tool_bias import BiasProfile @@ -59,7 +58,7 @@ from src.tool_presets import ToolPresetManager # imported from src/vendor_capabilities.py (deleted in # module_taxonomy_refactor_20260627 Phase 2.1). -PROVIDERS: List[str] = ["gemini", "anthropic", "gemini_cli", "deepseek", "minimax", "qwen", "grok", "llama"] +PROVIDERS: List[str] = ["gemini", "anthropic", "deepseek", "minimax", "qwen", "grok", "llama"] # DEFAULT_TOOL_CATEGORIES moved from src/models.py in # post_module_taxonomy_de_cruft_20260627 Phase 3. The categories are the @@ -161,8 +160,6 @@ _BIAS_ENGINE = ToolBiasEngine() _active_tool_preset: Optional[ToolPreset] = None _active_bias_profile: Optional[BiasProfile] = None -_gemini_cli_adapter: Optional[GeminiCliAdapter] = None - # Injected by gui.py - called when AI wants to run a command. confirm_and_run_callback: Optional[Callable[[str, str, Optional[Callable[[str], str]], Optional[Callable[[str, str], Result[str]]]], Optional[str]]] = None @@ -543,7 +540,7 @@ def set_provider(provider: str, model: str, validate: bool = True) -> None: """Updates the active LLM provider and model name. When validate is True (default), the model is checked against the provider's - LIVE model list, which for gemini_cli/minimax means a blocking subprocess / + LIVE model list, which for minimax means a blocking subprocess / network call (and importing the provider SDK). Pass validate=False during startup so the GUI's first frame is not blocked ΓÇö AppController._fetch_models corrects the model against the live list shortly after, off the main thread. @@ -553,13 +550,7 @@ def set_provider(provider: str, model: str, validate: bool = True) -> None: if not validate: _model = model return - if provider == "gemini_cli": - valid_models = _list_gemini_cli_models() - if model != "mock" and (model not in valid_models or model.startswith("deepseek")): - _model = "gemini-3-flash-preview" - else: - _model = model - elif provider == "minimax": + if provider == "minimax": result = _set_minimax_provider_result(model) fallback_result = _list_minimax_models_result("") valid_models = result.data if result.ok else fallback_result.data @@ -590,7 +581,6 @@ def reset_session() -> None: global _minimax_client global _qwen_client global _CACHED_ANTHROPIC_TOOLS, _CACHED_DEEPSEEK_TOOLS - global _gemini_cli_adapter if _gemini_client and _gemini_cache: _delete_gemini_cache_result() _gemini_client = None @@ -600,10 +590,6 @@ def reset_session() -> None: _gemini_cache_created_at = None _gemini_cached_file_paths = [] - # Preserve binary_path if adapter exists - old_path = _gemini_cli_adapter.binary_path if _gemini_cli_adapter else "gemini" - _gemini_cli_adapter = GeminiCliAdapter(binary_path=old_path) - _anthropic_client = None provider_state.clear_all() _deepseek_client = None @@ -626,7 +612,6 @@ def list_models(provider: str) -> list[str]: result = _list_anthropic_models_result() return result.data if result.ok else [] elif provider == "deepseek": return _list_deepseek_models(creds["deepseek"]["api_key"]) - elif provider == "gemini_cli": return _list_gemini_cli_models() elif provider == "minimax": result = _list_minimax_models_result(creds["minimax"]["api_key"]) return result.data if result.ok else [] @@ -911,7 +896,6 @@ async def _execute_tool_calls_concurrently( tasks = [] for fc in calls: if provider == "gemini": name, args, call_id = fc.name, dict(fc.args), fc.name # Gemini 1.0.0 doesn't have call IDs in types.Part - elif provider == "gemini_cli": name, args, call_id = cast(str, fc.get("name")), cast(Metadata, fc.get("args", {})), cast(str, fc.get("id")) elif provider == "anthropic": name, args, call_id = cast(str, getattr(fc, "name")), cast(Metadata, getattr(fc, "input")), cast(str, getattr(fc, "id")) elif provider == "deepseek": tool_info = fc.get("function", {}) @@ -1724,16 +1708,6 @@ def get_gemini_cache_stats() -> Metadata: "cached_files": _gemini_cached_file_paths, } -def _list_gemini_cli_models() -> list[str]: - return [ - "gemini-3-flash-preview", - "gemini-3.1-pro-preview", - "gemini-2.5-pro", - "gemini-2.5-flash", - "gemini-2.0-flash", - "gemini-2.5-flash-lite", - ] - def _list_gemini_models_result(api_key: str) -> Result[list[str]]: """List available Gemini models via google-genai SDK. @@ -1854,28 +1828,6 @@ def _create_gemini_cache_result(sys_instr: str, tools_decl: Any, file_items: lis errors=[ErrorInfo(kind=ErrorKind.INTERNAL, message=f"failed to create gemini cache: {type(e).__name__}: {e}", source="ai_client._create_gemini_cache_result", original=e)], ) -def _send_cli_round_result(r_idx: int, adapter: Any, payload: Any, safety_settings: list[Any], sys_instr: str, stream_callback: Optional[Callable[[str], None]]) -> Result[Metadata]: - """Call the Gemini CLI adapter for one round. Returns Result[resp_data]. - - On SDK failure, emits a response_received event with the error info - (preserving the original side-effect semantics) and returns - Result(errors=[ErrorInfo]). The caller (_send in _send_gemini_cli) - re-raises the original exception to preserve the outer catch flow. - """ - events.emit("request_start", payload={"provider": "gemini_cli", "model": _model, "round": r_idx}) - if r_idx > 0: - _append_comms("OUT", "request", {"message": f"[CLI] [round {r_idx}] [msg {len(payload)}]"}) - send_payload: Any = json.dumps(payload) if isinstance(payload, list) else payload - try: - resp_data = adapter.send(cast(str, send_payload), safety_settings=safety_settings, system_instruction=sys_instr, model=_model, stream_callback=stream_callback) - return Result(data=resp_data) - except Exception as e: - events.emit("response_received", payload={"provider": "gemini_cli", "model": _model, "usage": {}, "latency": 0, "round": r_idx, "error": str(e)}) - return Result( - data=None, - errors=[ErrorInfo(kind=ErrorKind.INTERNAL, message=str(e), source="ai_client._send_cli_round_result", original=e)], - ) - def _extract_gemini_thoughts_result(resp: Any) -> Result[str]: """Extracts concatenated thinking text from a Gemini response object's parts. @@ -2127,118 +2079,7 @@ def _send_gemini(md_content: str, user_message: str, base_dir: str, if monitor.enabled: monitor.end_component("ai_client._send_gemini") return Result(data="", errors=[_classify_gemini_error(e, source="ai_client.gemini")]) -def _send_gemini_cli(md_content: str, user_message: str, base_dir: str, - file_items: list[Metadata] | None = None, - discussion_history: str = "", - pre_tool_callback: Optional[Callable[[str, str, Optional[Callable[[str], str]]], Optional[str]]] = None, - qa_callback: Optional[Callable[[str], str]] = None, - stream_callback: Optional[Callable[[str], None]] = None, - patch_callback: Optional[Callable[[str, str], Result[str]]] = None) -> Result[str]: - from src.openai_compatible import OpenAICompatibleRequest, NormalizedResponse - from src.openai_schemas import UsageStats - """ - [C: src/ai_server.py:_handle_send] - Functional Purpose: Sends requests to Gemini via the headless Gemini CLI subprocess adapter. - Parameters & Inputs: md_content, user_message, base_dir, file_items, discussion_history, callbacks. - Immediate-Mode DAG / Thread Context: Called by: send; Calls: run_with_tool_loop, GeminiCliAdapter.send - SSDL: - [I:run_with_tool_loop] -> [I:GeminiCliAdapter.send] -> [T:Result] - Thread Boundaries: Runs on caller thread (typically an async worker thread). - """ - global _gemini_cli_adapter - try: - if _gemini_cli_adapter is None: - _gemini_cli_adapter = GeminiCliAdapter(binary_path="gemini") - adapter = _gemini_cli_adapter - mcp_client.configure(file_items or [], [base_dir]) - sys_instr = f"{_get_combined_system_prompt()}\n\n\n{md_content}\n" - safety_settings = [{'category': 'HARM_CATEGORY_DANGEROUS_CONTENT', 'threshold': 'BLOCK_ONLY_HIGH'}] - payload: Union[str, list[Metadata]] = user_message - if adapter.session_id is None: - if discussion_history: - payload = f"[DISCUSSION HISTORY]\n\n{discussion_history}\n\n---\n\n{user_message}" - all_text: list[str] = [] - cumulative_tool_bytes = 0 - def _send(r_idx: int) -> NormalizedResponse: - if adapter is None: - return NormalizedResponse(text="(adapter unavailable)", tool_calls=[], usage=UsageStats(input_tokens=0, output_tokens=0, cache_read_tokens=0, cache_creation_tokens=0), raw_response=None) - send_result = _send_cli_round_result(r_idx, adapter, payload, safety_settings, sys_instr, stream_callback) - if not send_result.ok: - raise cast(Exception, send_result.errors[0].original) from None - resp_data = send_result.data - cli_stderr = resp_data.get("stderr", "") - if cli_stderr: - sys.stderr.write(f"\n--- Gemini CLI stderr ---\n{cli_stderr}\n-------------------------\n") - sys.stderr.flush() - txt = cast(str, resp_data.get("text", "")) - if txt: all_text.append(txt) - calls = cast(List[dict[str, Any]], resp_data.get("tool_calls", [])) - usage = adapter.last_usage or {} - latency = adapter.last_latency - events.emit("response_received", payload={"provider": "gemini_cli", "model": _model, "usage": usage, "latency": latency, "round": r_idx}) - log_calls: list[Metadata] = [] - for c in calls: - log_calls.append({"name": c.get("name"), "args": c.get("args"), "id": c.get("id")}) - _append_comms("IN", "response", { - "round": r_idx, - "stop_reason": "TOOL_USE" if calls else "STOP", - "text": txt, - "tool_calls": log_calls, - "usage": usage - }) - if txt and calls: - cb = get_comms_log_callback_result().data - if cb: - cb({ - "ts": project_manager.now_ts(), - "direction": "IN", - "kind": "history_add", - "payload": {"role": "AI", "content": txt} - }) - return NormalizedResponse(text=txt, tool_calls=calls, usage=UsageStats(input_tokens=usage.get("prompt_tokens", 0), output_tokens=usage.get("completion_tokens", 0), cache_read_tokens=0, cache_creation_tokens=0), raw_response=resp_data) - - def _pre_dispatch(r_idx: int, calls: list[Metadata]) -> list[Metadata]: - nonlocal payload, cumulative_tool_bytes, file_items - tool_results_for_cli: list[Metadata] = [] - results_iter: list[tuple[str, str, str, str]] = [] - from src.ai_client import _execute_tool_calls_concurrently as _executor - try: - loop = asyncio.get_running_loop() - results_iter = loop.run_until_complete(_executor(calls, base_dir, pre_tool_callback, qa_callback, r_idx, "gemini_cli", patch_callback)) if False else asyncio.run_coroutine_threadsafe(_executor(calls, base_dir, pre_tool_callback, qa_callback, r_idx, "gemini_cli", patch_callback), loop).result() - except RuntimeError: - results_iter = asyncio.run(_executor(calls, base_dir, pre_tool_callback, qa_callback, r_idx, "gemini_cli", patch_callback)) - for i, (name, call_id, out, _) in enumerate(results_iter): - if i == len(results_iter) - 1: - if file_items: - _reread_result = _reread_file_items_result(file_items) - file_items, changed = _reread_result.data - ctx = _build_file_diff_text(changed) - if ctx: - out += f"\n\n{_get_context_marker()}\n\n{ctx}" - if r_idx == MAX_TOOL_ROUNDS: - out += "\n\n[SYSTEM: MAX ROUNDS. PROVIDE FINAL ANSWER.]" - out = _truncate_tool_output(out) - cumulative_tool_bytes += len(out) - tool_results_for_cli.append({"role": "tool", "tool_call_id": call_id, "name": name, "content": out}) - _append_comms("IN", "tool_result", {"name": name, "id": call_id, "output": out}) - events.emit("tool_execution", payload={"status": "completed", "tool": name, "result": out, "round": r_idx}) - payload = tool_results_for_cli - if cumulative_tool_bytes > _MAX_TOOL_OUTPUT_BYTES: - _append_comms("OUT", "request", {"message": f"[TOOL OUTPUT BUDGET EXCEEDED: {cumulative_tool_bytes} bytes]"}) - return calls - - run_with_tool_loop( - client=adapter, request=lambda _i: cast(OpenAICompatibleRequest, None), - base_dir=base_dir, vendor_name="gemini_cli", - pre_tool_callback=pre_tool_callback, qa_callback=qa_callback, - stream_callback=stream_callback, patch_callback=patch_callback, - send_func=_send, on_pre_dispatch=_pre_dispatch, - ) - final_text = all_text[-1] if all_text else "(No text returned)" - return Result(data=final_text) - except Exception as e: - return Result(data="", errors=[ErrorInfo(kind=ErrorKind.INTERNAL, message=str(e), source="ai_client.gemini_cli", original=e)]) #endregion: Gemini Provider @@ -3299,11 +3140,11 @@ def get_token_stats(md_content: str) -> Metadata: global _provider, _gemini_client, _model, _CHARS_PER_TOKEN total_tokens = 0 p = str(_provider).lower().strip() - if p in ("gemini", "gemini_cli"): + if p == "gemini": total_tokens = _count_gemini_tokens_for_stats_result(md_content).data if total_tokens == 0: total_tokens = max(1, int(len(md_content) / _CHARS_PER_TOKEN)) - limit = _GEMINI_MAX_INPUT_TOKENS if p in ["gemini", "gemini_cli"] else _ANTHROPIC_MAX_PROMPT_TOKENS + limit = _GEMINI_MAX_INPUT_TOKENS if p == "gemini" else _ANTHROPIC_MAX_PROMPT_TOKENS if p == "deepseek": limit = 64000 pct = (total_tokens / limit * 100) if limit > 0 else 0 @@ -3359,7 +3200,7 @@ def send( Immediate-Mode DAG / Thread Context: Called by: send() and direct public callers verifying error structures. Calls: performance_monitor, rag_engine.search, _append_comms, _send_gemini, - _send_gemini_cli, _send_anthropic, _send_deepseek, _send_minimax, + _send_anthropic, _send_deepseek, _send_minimax, _send_qwen, _send_llama, _send_grok, _send_llama_native SSDL: @@ -3391,11 +3232,6 @@ def send( md_content, user_message, base_dir, file_items, discussion_history, pre_tool_callback, qa_callback, enable_tools, stream_callback, patch_callback ) - elif p == "gemini_cli": - res = _send_gemini_cli( - md_content, user_message, base_dir, file_items, discussion_history, - pre_tool_callback, qa_callback, stream_callback, patch_callback - ) elif p == "anthropic": res = _send_anthropic( md_content, user_message, base_dir, file_items, discussion_history, @@ -3502,11 +3338,6 @@ def run_subagent_summarization(file_path: str, content: str, is_code: bool, outl return r.json()["choices"][0]["message"]["content"] except Exception as e: return f"ERROR: DeepSeek summarization failed: {e}" - elif _provider == "gemini_cli": - # Using the adapter for a one-off call - adapter = GeminiCliAdapter(binary_path="gemini") - resp_data = adapter.send(prompt, model=_model) - return resp_data.get("text", "") return "ERROR: Unsupported provider for sub-agent summarization" def run_discussion_compression(discussion_text: str) -> str: @@ -3553,10 +3384,6 @@ def run_discussion_compression(discussion_text: str) -> str: max_tokens=2048 ) return resp.choices[0].message.content or "" - elif p == "gemini_cli": - adapter = GeminiCliAdapter(binary_path="gemini") - resp_data = adapter.send(prompt, model=_model) - return resp_data.get("text", "") return f"ERROR: Unsupported provider for discussion compression: '{p}'" #endregion: Subagent Summarization diff --git a/src/gemini_cli_adapter.py b/src/gemini_cli_adapter.py deleted file mode 100644 index 95b7d870..00000000 --- a/src/gemini_cli_adapter.py +++ /dev/null @@ -1,193 +0,0 @@ -""" -Gemini CLI Adapter - Subprocess wrapper for the `gemini` CLI tool. - -This module provides an adapter for running the Google Gemini CLI as a subprocess, -parsing its streaming JSON output, and handling session management. - -Key Features: - - Streaming JSON output parsing (init, message, chunk, tool_use, result) - - Session persistence via --resume flag - - Non-blocking line-by-line reading with stream_callback - - Token estimation via character count heuristic (4 chars/token) - - CLI call logging via session_logger - -Integration: - - Used by ai_client.py as the 'gemini_cli' provider - - Enables synchronous HITL bridge via GEMINI_CLI_HOOK_CONTEXT env var - -Thread Safety: - - Each GeminiCliAdapter instance maintains its own session_id - - Not thread-safe. Use separate instances per thread. - -Configuration: - - binary_path: Path to the `gemini` CLI (from project config [gemini_cli].binary_path) - -Output Protocol: - The CLI emits JSON-L lines: - {"type": "init", "session_id": "..."} - {"type": "message", "content": "...", "role": "assistant"} - {"type": "tool_use", "name": "...", "parameters": {...}} - {"type": "result", "status": "success", "stats": {"total_tokens": N}} - -See Also: - - docs/guide_architecture.md for CLI adapter integration - - src/ai_client.py for provider dispatch -""" -import json -import os -import subprocess -import sys -import time - -from typing import Optional, Callable, Any - -from src import session_logger - - -class GeminiCliAdapter: - """ - Adapter for the Gemini CLI that parses streaming JSON output. - """ - def __init__(self, binary_path: str = "gemini"): - """Initializes the adapter with the path to the gemini CLI executable.""" - self.binary_path = binary_path - self.session_id: Optional[str] = None - self.last_usage: Optional[dict[str, Any]] = None - self.last_latency: float = 0.0 - - def send(self, message: str, safety_settings: list[Any] | None = None, system_instruction: str | None = None, model: str | None = None, stream_callback: Optional[Callable[[str], None]] = None) -> dict[str, Any]: - """ - Sends a message to the Gemini CLI and processes the streaming JSON output. - Uses non-blocking line-by-line reading to allow stream_callback. - """ - start_time = time.time() - command_parts = [self.binary_path] - if model: - command_parts.extend(['-m', f'"{model}"']) - command_parts.extend(['--prompt', '""']) - if self.session_id: - command_parts.extend(['--resume', self.session_id]) - command_parts.extend(['--output-format', 'stream-json']) - command = " ".join(command_parts) - - prompt_text = message - if system_instruction: - prompt_text = f"{system_instruction}\n\n{message}" - - accumulated_text = "" - tool_calls = [] - stdout_content = [] - - env = os.environ.copy() - env["GEMINI_CLI_HOOK_CONTEXT"] = "manual_slop" - - import shlex - # shlex.split handles quotes correctly even on Windows if we are careful. - # We want to split the entire binary_path into its components. - if os.name == 'nt': - # On Windows, shlex.split with default posix=True might swallow backslashes. - # Using posix=False is better for Windows paths. - cmd_list = shlex.split(self.binary_path, posix=False) - else: - cmd_list = shlex.split(self.binary_path) - - if model: - cmd_list.extend(['-m', model]) - cmd_list.extend(['--prompt', '""']) - if self.session_id: - cmd_list.extend(['--resume', self.session_id]) - cmd_list.extend(['--output-format', 'stream-json']) - - # Filter out empty strings and strip quotes (Popen doesn't want them in cmd_list elements) - cmd_list = [c.strip('"') for c in cmd_list if c] - sys.stderr.write(f"[DEBUG] GeminiCliAdapter cmd_list: {cmd_list}\n") - sys.stderr.flush() - - process = subprocess.Popen( - cmd_list, - stdin = subprocess.PIPE, - stdout = subprocess.PIPE, - stderr = subprocess.PIPE, - text = True, - encoding = "utf-8", - shell = False, - env = env - ) - - # Use communicate to avoid pipe deadlocks with large input/output. - # This blocks until the process exits, so we lose real-time streaming, - # but it's much more robust. We then simulate streaming by processing the output. - try: - stdout_final, stderr_final = process.communicate(input=prompt_text, timeout=60.0) - except subprocess.TimeoutExpired: - process.kill() - stdout_final, stderr_final = process.communicate() - stderr_final += "\n\n[ERROR] Gemini CLI subprocess timed out after 60 seconds." - # Mock a JSON error result to bubble up - stdout_final += '\n{"type": "result", "status": "error", "error": "subprocess timeout"}\n' - - last_decode_error = None - for line in stdout_final.splitlines(): - line = line.strip() - if not line: continue - stdout_content.append(line) - try: - data = json.loads(line) - msg_type = data.get("type") - if msg_type == "init": - if "session_id" in data: - self.session_id = data.get("session_id") - elif msg_type == "message" or msg_type == "chunk": - role = data.get("role", "") - if role in ["assistant", "model"] or not role: - content = data.get("content", data.get("text")) - if content: - accumulated_text += content - if stream_callback: - stream_callback(content) - elif msg_type == "result": - self.last_usage = data.get("stats") or data.get("usage") - if data.get("status") == "error": - raise Exception(data.get("error", "Unknown CLI error")) - if "session_id" in data: - self.session_id = data.get("session_id") - elif msg_type == "tool_use": - tc = { - "name": data.get("tool_name", data.get("name")), - "args": data.get("parameters", data.get("args", {})), - "id": data.get("tool_id", data.get("id")) - } - if tc["name"]: - tool_calls.append(tc) - except json.JSONDecodeError as e: - last_decode_error = e - continue - - current_latency = time.time() - start_time - if process.returncode != 0 and not accumulated_text and not tool_calls: - if last_decode_error: - raise Exception(f"Gemini CLI failed (exit {process.returncode}) with JSONDecodeError: {last_decode_error}\nOutput: {stdout_final}") - raise Exception(f"Gemini CLI failed with exit {process.returncode}\nStderr: {stderr_final}") - session_logger.open_session() - session_logger.log_cli_call( - command = command, - stdin_content = prompt_text, - stdout_content = "\n".join(stdout_content), - stderr_content = stderr_final, - latency = current_latency - ) - self.last_latency = current_latency - - return { - "text": accumulated_text, - "tool_calls": tool_calls, - "stderr": stderr_final - } - - def count_tokens(self, contents: list[str]) -> int: - """ - Provides a character-based token estimation for the Gemini CLI. - Uses 4 chars/token as a conservative average. - """ - total_chars = len("\n".join(contents)) - return total_chars // 4 \ No newline at end of file diff --git a/tests/test_ai_client_cli.py b/tests/test_ai_client_cli.py deleted file mode 100644 index fb987bb9..00000000 --- a/tests/test_ai_client_cli.py +++ /dev/null @@ -1,33 +0,0 @@ -from unittest.mock import patch, MagicMock -from src import ai_client -from src.result_types import Result - - -def test_ai_client_send_gemini_cli() -> None: - test_message = "Hello, this is a test prompt for the CLI adapter." - test_response = "This is a dummy response from the Gemini CLI." - ai_client.reset_session() - ai_client.set_provider("gemini_cli", "gemini-2.5-flash-lite") - with patch("src.ai_client.GeminiCliAdapter") as MockAdapterClass: - mock_adapter_instance = MagicMock() - mock_adapter_instance.send.return_value = { - "text": test_response, - "tool_calls": [], - } - mock_adapter_instance.last_usage = {"total_tokens": 100} - mock_adapter_instance.last_latency = 0.5 - mock_adapter_instance.session_id = "test-session" - MockAdapterClass.return_value = mock_adapter_instance - ai_client._gemini_cli_adapter = mock_adapter_instance - with patch.object(ai_client.events, "emit") as mock_emit: - result = ai_client.send( - md_content="", - user_message=test_message, - base_dir=".", - ) - mock_adapter_instance.send.assert_called() - emitted_event_names = [call.args[0] for call in mock_emit.call_args_list] - assert "request_start" in emitted_event_names - assert "response_received" in emitted_event_names - assert result.ok - assert result.data == test_response diff --git a/tests/test_gemini_cli_adapter.py b/tests/test_gemini_cli_adapter.py deleted file mode 100644 index 2effe2e7..00000000 --- a/tests/test_gemini_cli_adapter.py +++ /dev/null @@ -1,85 +0,0 @@ -import json -from unittest.mock import patch, MagicMock -from src.gemini_cli_adapter import GeminiCliAdapter - - -class TestGeminiCliAdapter: - @patch("subprocess.Popen") - def test_send_starts_subprocess_with_correct_args( - self, mock_popen: MagicMock - ) -> None: - adapter = GeminiCliAdapter(binary_path="gemini") - mock_process = MagicMock() - mock_process.communicate.return_value = ( - '{"type": "message", "content": "hello"}', - "", - ) - mock_process.returncode = 0 - mock_popen.return_value = mock_process - adapter.send("test prompt") - assert mock_popen.called - args, kwargs = mock_popen.call_args - cmd_list = args[0] - assert "gemini" in cmd_list - assert "--prompt" in cmd_list - assert "--output-format" in cmd_list - assert "stream-json" in cmd_list - - @patch("subprocess.Popen") - def test_send_parses_jsonl_output(self, mock_popen: MagicMock) -> None: - adapter = GeminiCliAdapter() - stdout_str = '{"type": "message", "content": "Hello "}\n{"type": "message", "content": "world!"}\n' - mock_process = MagicMock() - mock_process.communicate.return_value = (stdout_str, "") - mock_process.returncode = 0 - mock_popen.return_value = mock_process - result = adapter.send("msg") - assert result["text"] == "Hello world!" - - @patch("subprocess.Popen") - def test_send_handles_tool_use_events(self, mock_popen: MagicMock) -> None: - adapter = GeminiCliAdapter() - tool_json = { - "type": "tool_use", - "tool_name": "read_file", - "parameters": {"path": "test.txt"}, - "tool_id": "call_123", - } - stdout_str = json.dumps(tool_json) + "\n" - mock_process = MagicMock() - mock_process.communicate.return_value = (stdout_str, "") - mock_process.returncode = 0 - mock_popen.return_value = mock_process - result = adapter.send("msg") - assert len(result["tool_calls"]) == 1 - assert result["tool_calls"][0]["name"] == "read_file" - assert result["tool_calls"][0]["args"]["path"] == "test.txt" - - @patch("subprocess.Popen") - def test_send_captures_usage_metadata(self, mock_popen: MagicMock) -> None: - adapter = GeminiCliAdapter() - result_json = {"type": "result", "stats": {"total_tokens": 50}} - stdout_str = json.dumps(result_json) + "\n" - mock_process = MagicMock() - mock_process.communicate.return_value = (stdout_str, "") - mock_process.returncode = 0 - mock_popen.return_value = mock_process - adapter.send("msg") - assert adapter.last_usage is not None - assert adapter.last_usage.get("total_tokens") == 50 - - @patch("subprocess.Popen") - def test_full_flow_integration(self, mock_popen: MagicMock) -> None: - adapter = GeminiCliAdapter() - msg_json = {"type": "message", "content": "Final response"} - result_json = { - "type": "result", - "stats": {"total_tokens": 25, "input_tokens": 10, "output_tokens": 15}, - } - stdout_str = json.dumps(msg_json) + "\n" + json.dumps(result_json) + "\n" - mock_process = MagicMock() - mock_process.communicate.return_value = (stdout_str, "") - mock_process.returncode = 0 - mock_popen.return_value = mock_process - result = adapter.send("test") - assert "Final response" in result["text"] diff --git a/tests/test_gemini_cli_adapter_parity.py b/tests/test_gemini_cli_adapter_parity.py deleted file mode 100644 index d9feaa28..00000000 --- a/tests/test_gemini_cli_adapter_parity.py +++ /dev/null @@ -1,49 +0,0 @@ -import unittest -import json -from unittest.mock import patch, MagicMock -from src.gemini_cli_adapter import GeminiCliAdapter - -class TestGeminiCliAdapterParity(unittest.TestCase): - def setUp(self) -> None: - self.adapter = GeminiCliAdapter(binary_path="gemini") - - def tearDown(self) -> None: - pass - - def test_count_tokens_fallback(self) -> None: - contents = ["Hello", "world!"] - estimated = self.adapter.count_tokens(contents) - self.assertEqual(estimated, 3) - - @patch('src.gemini_cli_adapter.subprocess.Popen') - def test_send_starts_subprocess_with_model(self, mock_popen: MagicMock) -> None: - mock_process = MagicMock() - mock_process.communicate.return_value = ('{"type": "message", "content": "hi"}', '') - mock_process.returncode = 0 - mock_popen.return_value = mock_process - self.adapter.send("test", model="gemini-2.0-flash") - args, _ = mock_popen.call_args - cmd_list = args[0] - self.assertIn("-m", cmd_list) - self.assertIn("gemini-2.0-flash", cmd_list) - - @patch('src.gemini_cli_adapter.subprocess.Popen') - def test_send_parses_tool_calls_from_streaming_json(self, mock_popen: MagicMock) -> None: - tool_call_json = { - "type": "tool_use", - "tool_name": "list_directory", - "parameters": {"path": "."}, - "tool_id": "call_abc" - } - mock_process = MagicMock() - stdout_output = ( - json.dumps(tool_call_json) + "\n" + - '{"type": "message", "content": "I listed the files."}' - ) - mock_process.communicate.return_value = (stdout_output, '') - mock_process.returncode = 0 - mock_popen.return_value = mock_process - result = self.adapter.send("msg") - self.assertEqual(len(result["tool_calls"]), 1) - self.assertEqual(result["tool_calls"][0]["name"], "list_directory") - self.assertEqual(result["text"], "I listed the files.") diff --git a/tests/test_gemini_cli_edge_cases.py b/tests/test_gemini_cli_edge_cases.py deleted file mode 100644 index eb6f7d46..00000000 --- a/tests/test_gemini_cli_edge_cases.py +++ /dev/null @@ -1,41 +0,0 @@ -from unittest.mock import patch, MagicMock -from src.gemini_cli_adapter import GeminiCliAdapter -from src import mcp_client -from src.result_types import Result - -def test_gemini_cli_context_bleed_prevention() -> None: - import src.ai_client as ai_client - ai_client._gemini_cli_adapter = None - with patch('src.gemini_cli_adapter.subprocess.Popen') as mock_popen: - adapter = GeminiCliAdapter() - mock_process = MagicMock() - stdout_output = ( - '{"type": "message", "role": "user", "content": "Echoed user prompt"}' + "\n" + - '{"type": "message", "role": "model", "content": "Model response"}' - ) - mock_process.communicate.return_value = (stdout_output, '') - mock_process.returncode = 0 - mock_popen.return_value = mock_process - result = adapter.send("msg") - assert result["text"] == "Model response" - -def test_gemini_cli_parameter_resilience() -> None: - with patch('src.mcp_client.read_file', return_value="content") as mock_read: - mcp_client.dispatch("read_file", {"file_path": "aliased.txt"}) - mock_read.assert_called_once_with("aliased.txt") - with patch('src.mcp_client.list_directory', return_value="files") as mock_list: - mcp_client.dispatch("list_directory", {"dir_path": "aliased_dir"}) - mock_list.assert_called_once_with("aliased_dir") - -def test_gemini_cli_loop_termination() -> None: - import src.ai_client as ai_client - ai_client._gemini_cli_adapter = None - with patch('src.gemini_cli_adapter.subprocess.Popen') as mock_popen: - mock_process = MagicMock() - mock_process.communicate.return_value = ('{"type": "message", "content": "Final answer", "tool_calls": []}', "") - mock_process.returncode = 0 - mock_popen.return_value = mock_process - ai_client.set_provider("gemini_cli", "gemini-2.0-flash") - result = ai_client.send("context", "prompt") - assert result.ok - assert result.data == "Final answer" diff --git a/tests/test_gemini_cli_integration.py b/tests/test_gemini_cli_integration.py deleted file mode 100644 index 84654cf9..00000000 --- a/tests/test_gemini_cli_integration.py +++ /dev/null @@ -1,32 +0,0 @@ -from unittest.mock import MagicMock -from src import ai_client -from src.result_types import Result - - -def test_gemini_cli_full_integration() -> None: - ai_client.reset_session() - ai_client.set_provider("gemini_cli", "gemini-2.0-flash") - mock_adapter = MagicMock() - mock_adapter.send.return_value = { - "text": "Final integrated answer", - "tool_calls": [], - } - mock_adapter.last_usage = {"total_tokens": 10} - ai_client._gemini_cli_adapter = mock_adapter - result = ai_client.send("context", "integrated test") - assert result.ok - assert "Final integrated answer" in result.data - - -def test_gemini_cli_rejection_and_history() -> None: - ai_client.reset_session() - ai_client.set_provider("gemini_cli", "gemini-2.0-flash") - mock_adapter = MagicMock() - mock_adapter.send.return_value = { - "text": "", - "tool_calls": [{"name": "run_powershell", "args": {"script": "dir"}}], - } - mock_adapter.last_usage = {} - ai_client._gemini_cli_adapter = mock_adapter - result = ai_client.send("ctx", "msg", pre_tool_callback=lambda *a, **kw: None) - assert result is not None diff --git a/tests/test_gemini_cli_parity_regression.py b/tests/test_gemini_cli_parity_regression.py deleted file mode 100644 index 94695535..00000000 --- a/tests/test_gemini_cli_parity_regression.py +++ /dev/null @@ -1,15 +0,0 @@ -from unittest.mock import patch, MagicMock -from src.result_types import Result - -def test_send_invokes_adapter_send() -> None: - import src.ai_client as ai_client - ai_client._gemini_cli_adapter = None - with patch('src.gemini_cli_adapter.subprocess.Popen') as mock_popen: - mock_process = MagicMock() - mock_process.communicate.return_value = ('{"type": "message", "content": "Hello from mock adapter"}', '') - mock_process.returncode = 0 - mock_popen.return_value = mock_process - ai_client.set_provider("gemini_cli", "gemini-2.0-flash") - res = ai_client.send("context", "msg") - assert res.ok - assert res.data == "Hello from mock adapter" diff --git a/tests/test_mock_gemini_cli.py b/tests/test_mock_gemini_cli.py deleted file mode 100644 index 5cd676b7..00000000 --- a/tests/test_mock_gemini_cli.py +++ /dev/null @@ -1,69 +0,0 @@ -import subprocess -import json - - -def get_message_content(stdout): - for line in stdout.splitlines(): - line = line.strip() - if not line: - continue - try: - obj = json.loads(line) - if isinstance(obj, dict) and obj.get('type') == 'message': - return obj.get('content', '') - except json.JSONDecodeError: - continue - return '' - - -def run_mock(prompt): - return subprocess.run( - ['uv', 'run', 'python', 'tests/mock_gemini_cli.py'], - input=prompt, - capture_output=True, - text=True, - cwd='.' - ) - - -def test_epic_prompt_returns_track_json(): - result = run_mock('PATH: Epic Initialization — please produce tracks') - assert result.returncode == 0 - assert 'function_call' not in result.stdout - content = get_message_content(result.stdout) - parsed = json.loads(content) - assert isinstance(parsed, list) - assert len(parsed) > 0 - for item in parsed: - assert 'id' in item - assert 'title' in item - - -def test_sprint_prompt_returns_ticket_json(): - result = run_mock('Please generate the implementation tickets for this track.') - assert result.returncode == 0 - assert 'function_call' not in result.stdout - content = get_message_content(result.stdout) - parsed = json.loads(content) - assert isinstance(parsed, list) - assert len(parsed) > 0 - for item in parsed: - assert 'id' in item - assert 'description' in item - assert 'status' in item - assert 'assigned_to' in item - - -def test_worker_prompt_returns_plain_text(): - result = run_mock('Please read test.txt\nYou are assigned to Ticket T1.\nTask Description: do something') - assert result.returncode == 0 - assert 'function_call' not in result.stdout - content = get_message_content(result.stdout) - assert content != '' - - -def test_tool_result_prompt_returns_plain_text(): - result = run_mock('role: tool\nHere are the results: {"content": "done"}') - assert result.returncode == 0 - content = get_message_content(result.stdout) - assert content != ''