feat(ai): support stdin for Gemini CLI and verify with integration test

tests/mock_gemini_cli.py

@@ -0,0 +1,83 @@
+import sys
+import json
+import subprocess
+import os
+
+def main():
+    # The GUI calls: <binary> run --output-format stream-json
+    # The prompt is now passed via stdin.
+    
+    # Debug log to stderr
+    sys.stderr.write(f"DEBUG: mock_gemini_cli called with args: {sys.argv}\n")
+    
+    # Read prompt from stdin for debug
+    prompt = sys.stdin.read()
+    sys.stderr.write(f"DEBUG: Received prompt via stdin ({len(prompt)} chars)\n")
+    sys.stderr.flush()
+
+    if "run" not in sys.argv:
+        return
+
+    # Simulate the 'BeforeTool' hook by calling the bridge directly.
+    bridge_path = os.path.abspath("scripts/cli_tool_bridge.py")
+    
+    tool_call = {
+        "tool_name": "read_file",
+        "tool_input": {"path": "test.txt"}
+    }
+    
+    sys.stderr.write(f"DEBUG: Calling bridge at {bridge_path}\n")
+    sys.stderr.flush()
+
+    # Bridge reads from stdin
+    process = subprocess.Popen(
+        [sys.executable, bridge_path],
+        stdin=subprocess.PIPE,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+        text=True
+    )
+    stdout, stderr = process.communicate(input=json.dumps(tool_call))
+    
+    sys.stderr.write(f"DEBUG: Bridge stdout: {stdout}\n")
+    sys.stderr.write(f"DEBUG: Bridge stderr: {stderr}\n")
+    sys.stderr.flush()
+
+    try:
+        decision_data = json.loads(stdout.strip())
+        decision = decision_data.get("decision")
+    except Exception as e:
+        sys.stderr.write(f"DEBUG: Failed to parse bridge output: {e}\n")
+        decision = "deny"
+
+    # Output JSONL to stdout
+    if decision == "allow":
+        print(json.dumps({
+            "type": "tool_use", 
+            "name": "read_file", 
+            "args": {"path": "test.txt"}
+        }), flush=True)
+        
+        print(json.dumps({
+            "type": "message", 
+            "text": "I read the file. It contains: 'Hello from mock!'"
+        }), flush=True)
+        
+        print(json.dumps({
+            "type": "result", 
+            "usage": {"total_tokens": 50},
+            "session_id": "mock-session-123"
+        }), flush=True)
+    else:
+        print(json.dumps({
+            "type": "message", 
+            "text": f"Tool execution was denied. Decision: {decision}"
+        }), flush=True)
+        print(json.dumps({
+            "type": "result", 
+            "usage": {"total_tokens": 10},
+            "session_id": "mock-session-denied"
+        }), flush=True)
+
+if __name__ == "__main__":
+    main()

tests/test_gemini_cli_integration.py

@@ -0,0 +1,86 @@
+import pytest
+import time
+import os
+import sys
+import requests
+from api_hook_client import ApiHookClient
+
+def test_gemini_cli_full_integration(live_gui):
+    """
+    Integration test for the Gemini CLI provider and tool bridge.
+    """
+    client = ApiHookClient("http://127.0.0.1:8999")
+    
+    # 1. Setup paths and configure the GUI
+    mock_script = os.path.abspath("tests/mock_gemini_cli.py")
+    # Wrap in quotes for shell execution if path has spaces
+    cli_cmd = f'"{sys.executable}" "{mock_script}"'
+    
+    # Set provider and binary path via GUI hooks
+    # Note: Using set_value which now triggers the property setter in gui_2.py
+    print(f"[TEST] Setting current_provider to gemini_cli")
+    client.set_value("current_provider", "gemini_cli")
+    print(f"[TEST] Setting gcli_path to {cli_cmd}")
+    client.set_value("gcli_path", cli_cmd)
+    
+    # Verify settings were applied
+    assert client.get_value("current_provider") == "gemini_cli"
+    assert client.get_value("gcli_path") == cli_cmd
+
+    # Clear events
+    client.get_events()
+    
+    # 2. Trigger a message in the GUI
+    print("[TEST] Sending user message...")
+    client.set_value("ai_input", "Please read test.txt")
+    client.click("btn_gen_send")
+    
+    # 3. Monitor for the 'ask_received' event
+    print("[TEST] Waiting for ask_received event...")
+    request_id = None
+    timeout = 30
+    start_time = time.time()
+    while time.time() - start_time < timeout:
+        events = client.get_events()
+        if events:
+            print(f"[TEST] Received {len(events)} events: {[e.get('type') for e in events]}")
+        for ev in events:
+            if ev.get("type") == "ask_received":
+                request_id = ev.get("request_id")
+                print(f"[TEST] Found request_id: {request_id}")
+                break
+        if request_id:
+            break
+        time.sleep(0.5)
+
+    assert request_id is not None, "Timed out waiting for 'ask_received' event from the bridge"
+
+    # 4. Respond to the permission request
+    print("[TEST] Responding to ask with approval")
+    resp = requests.post(
+        "http://127.0.0.1:8999/api/ask/respond",
+        json={
+            "request_id": request_id,
+            "response": {"approved": True}
+        }
+    )
+    assert resp.status_code == 200
+
+    # 5. Verify that the final response is displayed in the GUI
+    print("[TEST] Waiting for final message in history...")
+    final_message_received = False
+    start_time = time.time()
+    while time.time() - start_time < timeout:
+        session = client.get_session()
+        entries = session.get("session", {}).get("entries", [])
+        for entry in entries:
+            content = entry.get("content", "")
+            if "Hello from mock!" in content:
+                print(f"[TEST] Success! Found message: {content[:50]}...")
+                final_message_received = True
+                break
+        if final_message_received:
+            break
+        time.sleep(1.0)
+
+    assert final_message_received, "Final message from mock CLI was not found in the GUI history"