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

@@ -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":
+ if provider == "minimax":
-  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":
   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<context>\n{md_content}\n</context>"
-  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

src/gemini_cli_adapter.py

@@ -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

tests/test_ai_client_cli.py

@@ -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="<context></context>",
-    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

tests/test_gemini_cli_adapter.py

@@ -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"]

tests/test_gemini_cli_adapter_parity.py

@@ -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.")

tests/test_gemini_cli_edge_cases.py

@@ -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"

tests/test_gemini_cli_integration.py

@@ -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

tests/test_gemini_cli_parity_regression.py

@@ -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"

tests/test_mock_gemini_cli.py

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