manual_slop/scripts/cli_tool_bridge.py

import sys
import json
import logging
import os

# Add project root to sys.path so we can import api_hook_client
# This helps in cases where the script is run from different directories
project_root = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
if project_root not in sys.path:
    sys.path.append(project_root)

try:
    from api_hook_client import ApiHookClient
except ImportError:
    # Fallback if the script is run from the project root directly,
    # or if the above path append didn't work for some reason.
    try:
        from api_hook_client import ApiHookClient
    except ImportError:
        # Use basic print for fatal errors if logging isn't set up yet
        print("FATAL: Failed to import ApiHookClient. Ensure it's in the Python path.", file=sys.stderr)
        sys.exit(1) # Exit if the core dependency cannot be imported


def main():
    # Setup basic logging to stderr.
    # Set level to DEBUG to capture all messages, including debug info.
    logging.basicConfig(level=logging.DEBUG, format='%(asctime)s - %(levelname)s - %(message)s', stream=sys.stderr)

    logging.debug("CLI Tool Bridge script started.")

    try:
        # 1. Read JSON from sys.stdin
        input_data = sys.stdin.read()
        
        if not input_data:
            logging.debug("No input received from stdin. Exiting gracefully.")
            return

        logging.debug(f"Received raw input data: {input_data}")

        try:
            hook_input = json.loads(input_data)
        except json.JSONDecodeError:
            logging.error("Failed to decode JSON from stdin.")
            print(json.dumps({
                "decision": "deny",
                "reason": "Invalid JSON received from stdin."
            }))
            return

        # Initialize variables for tool name and arguments
        tool_name = None
        tool_args = {}

        # 2. Try to parse input in Gemini API format ('name', 'input')
        logging.debug("Attempting to parse input in Gemini API format ('name', 'input').")
        if 'name' in hook_input and hook_input['name'] is not None:
            tool_name = hook_input['name']
            logging.debug(f"Found Gemini API format tool name: {tool_name}")
        
        if 'input' in hook_input and hook_input['input'] is not None:
            if isinstance(hook_input['input'], dict):
                tool_args = hook_input['input']
                logging.debug(f"Found Gemini API format tool input: {tool_args}")
            else:
                logging.warning("Gemini API format 'input' is not a dictionary. Ignoring.")

        # 3. If Gemini format wasn't fully present, try the legacy format ('tool_name', 'tool_input')
        if tool_name is None:
            logging.debug("Gemini API format not fully detected. Falling back to legacy format ('tool_name', 'tool_input').")
            tool_name = hook_input.get('tool_name')
            if tool_name:
                logging.debug(f"Found legacy format tool name: {tool_name}")

            tool_input_legacy = hook_input.get('tool_input')
            if tool_input_legacy is not None:
                if isinstance(tool_input_legacy, dict):
                    tool_args = tool_input_legacy
                    logging.debug(f"Found legacy format tool input: {tool_args}")
                else:
                    logging.warning("Legacy format 'tool_input' is not a dictionary. Ignoring.")

        # Final checks on resolved tool_name and tool_args
        if tool_name is None:
            logging.error("Could not determine tool name from input.")
            print(json.dumps({
                "decision": "deny",
                "reason": "Could not determine tool name from input. Expected 'name' or 'tool_name'."
            }))
            return

        if not isinstance(tool_args, dict):
            logging.error(f"Resolved tool_args is not a dictionary: {tool_args}")
            print(json.dumps({
                "decision": "deny",
                "reason": "Resolved tool arguments are not in a valid dictionary format."
            }))
            return

        logging.debug(f"Resolved tool_name: '{tool_name}', tool_args: {tool_args}")

        # 4. Check context — if not running via Manual Slop, we pass through (allow)
        # This prevents the hook from affecting normal CLI usage.
        hook_context = os.environ.get("GEMINI_CLI_HOOK_CONTEXT")
        logging.debug(f"Checking GEMINI_CLI_HOOK_CONTEXT: '{hook_context}'")
        if hook_context != "manual_slop":
             logging.debug(f"GEMINI_CLI_HOOK_CONTEXT is '{hook_context}', NOT 'manual_slop'. Allowing execution without confirmation.")
             print(json.dumps({
                "decision": "allow",
                "reason": f"Non-programmatic usage (GEMINI_CLI_HOOK_CONTEXT={hook_context})."
            }))
             return

        # 5. Use 'ApiHookClient' (assuming GUI is on http://127.0.0.1:8999)
        logging.debug("GEMINI_CLI_HOOK_CONTEXT is 'manual_slop'. Proceeding with API Hook Client.")
        client = ApiHookClient(base_url="http://127.0.0.1:8999")

        try:
            # 6. Request confirmation
            # This is a blocking call that waits for the user in the GUI
            logging.debug(f"Requesting confirmation for tool '{tool_name}' with args: {tool_args}")
            response = client.request_confirmation(tool_name, tool_args)

            if response and response.get('approved') is True:
                # 7. Print 'allow' decision
                logging.debug("User approved tool execution.")
                print(json.dumps({"decision": "allow"}))
            else:
                # 8. Print 'deny' decision
                reason = response.get('reason', 'User rejected tool execution in GUI.') if response else 'No response from GUI.'
                logging.debug(f"User denied tool execution. Reason: {reason}")
                print(json.dumps({
                    "decision": "deny",
                    "reason": reason
                }))

        except Exception as e:
            # 9. Handle cases where hook server is not reachable or other API errors
            # If we ARE in manual_slop context but can't reach the server, we should DENY
            # because the user expects to be in control.
            logging.error(f"API Hook Client error: {str(e)}", exc_info=True)
            print(json.dumps({
                "decision": "deny",
                "reason": f"Manual Slop hook server unreachable or API error: {str(e)}"
            }))

    except Exception as e:
        # Fallback for unexpected errors during initial processing (e.g., stdin read)
        logging.error(f"An unexpected error occurred in the main bridge logic: {str(e)}", exc_info=True)
        print(json.dumps({
            "decision": "deny",
            "reason": f"Internal bridge error: {str(e)}"
        }))

if __name__ == "__main__":
    main()