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 != ''