#!/usr/bin/env python3 """ Real Gemini Thinking Mode Workflow Navigates to gemini.google.com, logs in, selects Thinking mode, sends prompt, extracts response. """ import os import time import json from datetime import datetime from playwright.sync_api import sync_playwright, TimeoutError as PlaywrightTimeoutError SCREENSHOTS_DIR = "/mnt/e/genesis-system/deep_think_results/screenshots" OUTPUT_FILE = "/mnt/e/genesis-system/deep_think_results/REAL_DT_GENESIS_STRATEGY.md" STATUS_FILE = "/mnt/e/genesis-system/hive/progress/DT_GENESIS_STRATEGY_STATUS.md" EMAIL = "kinan@agileadapt.com" PASSWORD = "Systema55" PROMPT_TEXT = """Genesis is an AI operating system built for autonomous revenue generation. It uses multi-agent orchestration, voice AI widgets, browser automation, and a living knowledge graph. What are the 3 highest-leverage technical investments Genesis should make in the next 60 days to maximise revenue velocity and competitive moat? Be specific, prioritised, and consider: Telnyx voice widget ($97-497/mo), agency GTM, Australian market compliance, and memory/KG architecture.""" os.makedirs(SCREENSHOTS_DIR, exist_ok=True) def take_screenshot(page, name): path = f"{SCREENSHOTS_DIR}/gemini_workflow_{name}.png" try: page.screenshot(path=path) print(f" [SCREENSHOT] {path}") except Exception as e: print(f" [SCREENSHOT FAILED] {name}: {e}") return path def log(msg): print(f"[{datetime.now().strftime('%H:%M:%S')}] {msg}") def main(): log("Starting Gemini Thinking Mode workflow") with sync_playwright() as p: # Launch browser with persistent context to reuse any saved auth browser = p.chromium.launch( headless=False, # visible so we can see what's happening slow_mo=500, args=[ '--no-sandbox', '--disable-dev-shm-usage', '--disable-blink-features=AutomationControlled', ] ) context = browser.new_context( viewport={"width": 1280, "height": 900}, user_agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/121.0.0.0 Safari/537.36", ) page = context.new_page() mode_used = "Unknown" try: # Step 1: Navigate to Gemini log("Step 1: Navigating to https://gemini.google.com") page.goto("https://gemini.google.com", wait_until="domcontentloaded", timeout=30000) time.sleep(3) take_screenshot(page, "01_initial_load") log(f" URL: {page.url}") # Check if already logged in current_url = page.url log(f" Current URL after load: {current_url}") # Step 2: Handle login if needed if "accounts.google.com" in current_url or page.locator("text=Sign in").count() > 0: log("Step 2: Login required - proceeding with credentials") # Check if we're on the Google sign-in page or need to click sign in if "gemini.google.com" in current_url: # Try clicking sign in button try: sign_in_btn = page.locator("text=Sign in").first if sign_in_btn.is_visible(timeout=3000): sign_in_btn.click() time.sleep(2) except: pass # Navigate to accounts.google.com sign-in if "accounts.google.com" not in page.url: page.goto("https://accounts.google.com/signin", wait_until="domcontentloaded", timeout=30000) time.sleep(2) take_screenshot(page, "02_login_page") # Enter email log(" Entering email...") email_input = page.locator('input[type="email"]') email_input.wait_for(state="visible", timeout=15000) email_input.fill(EMAIL) page.keyboard.press("Enter") time.sleep(3) take_screenshot(page, "03_email_entered") # Enter password log(" Entering password...") try: pwd_input = page.locator('input[type="password"]') pwd_input.wait_for(state="visible", timeout=15000) pwd_input.fill(PASSWORD) page.keyboard.press("Enter") time.sleep(5) take_screenshot(page, "04_password_entered") except Exception as e: log(f" Password entry issue: {e}") take_screenshot(page, "04_password_error") log(f" URL after login attempt: {page.url}") else: log("Step 2: Already logged in - skipping login") take_screenshot(page, "02_already_logged_in") # Wait for Gemini to fully load log("Step 3: Waiting for Gemini to load fully...") time.sleep(5) # Navigate to Gemini main page if not there if "gemini.google.com" not in page.url: log(" Navigating to Gemini main page...") page.goto("https://gemini.google.com", wait_until="domcontentloaded", timeout=30000) time.sleep(5) take_screenshot(page, "05_gemini_loaded") log(f" URL: {page.url}") # Step 4: Find and click mode picker log("Step 4: Looking for mode picker button...") time.sleep(3) # Try to find mode picker mode_picker_selectors = [ '[aria-label="Open mode picker"]', '[aria-label="Mode picker"]', 'button[aria-label*="mode" i]', 'button[aria-label*="Mode" i]', '[data-test-id="mode-picker"]', ] mode_picker_clicked = False for selector in mode_picker_selectors: try: btn = page.locator(selector).first if btn.is_visible(timeout=3000): log(f" Found mode picker: {selector}") btn.click() time.sleep(2) mode_picker_clicked = True break except: continue if not mode_picker_clicked: log(" Mode picker not found with standard selectors, trying alternatives...") # Try to find any dropdown or model selector take_screenshot(page, "06_no_mode_picker") # Look for any model/mode related buttons all_buttons = page.locator("button").all() log(f" Found {len(all_buttons)} buttons on page") for i, btn in enumerate(all_buttons[:20]): try: label = btn.get_attribute("aria-label") or "" text = btn.inner_text() or "" if any(word in (label + text).lower() for word in ["model", "mode", "gemini", "ultra", "pro", "think"]): log(f" Potential mode button [{i}]: aria-label='{label}' text='{text[:50]}'") except: pass # Try clicking on a Gemini model selector if visible try: # Look for "Gemini 2.0 Flash" or similar model labels model_label = page.locator("text=Gemini").first if model_label.is_visible(timeout=2000): model_label.click() time.sleep(2) mode_picker_clicked = True log(" Clicked on Gemini text element") except: pass take_screenshot(page, "06_after_mode_picker") # Step 5: Select Thinking mode log("Step 5: Looking for Thinking mode option...") thinking_selectors = [ '[data-test-id="bard-mode-option-thinking"]', '[aria-label*="Thinking" i]', 'text=Thinking', '[data-mode="thinking"]', ] thinking_selected = False for selector in thinking_selectors: try: opt = page.locator(selector).first if opt.is_visible(timeout=3000): log(f" Found Thinking option: {selector}") opt.click() time.sleep(2) thinking_selected = True mode_used = "Thinking" break except: continue if not thinking_selected: log(" Thinking mode not found, trying Pro or 2.0 Flash Thinking...") pro_selectors = [ '[data-test-id="bard-mode-option-pro"]', 'text=2.0 Flash Thinking', 'text=Gemini 2.0 Flash Thinking', '[aria-label*="Pro" i]', 'text=Pro', 'text=1.5 Pro', 'text=2.0 Pro', ] for selector in pro_selectors: try: opt = page.locator(selector).first if opt.is_visible(timeout=2000): log(f" Found alternative mode: {selector}") opt.click() time.sleep(2) thinking_selected = True mode_used = f"Alternative ({selector})" break except: continue if not thinking_selected: log(" No special mode found - using default Gemini mode") mode_used = "Default Gemini" # Press Escape to close any open dropdown page.keyboard.press("Escape") time.sleep(1) take_screenshot(page, "07_mode_selected") log(f" Mode selected: {mode_used}") # Step 6: Enter prompt log("Step 6: Entering prompt...") time.sleep(2) prompt_selectors = [ '[aria-label="Enter a prompt for Gemini"]', '[aria-label*="prompt" i]', '[placeholder*="prompt" i]', 'div[contenteditable="true"]', 'textarea', 'rich-textarea', ] prompt_entered = False for selector in prompt_selectors: try: inp = page.locator(selector).first if inp.is_visible(timeout=3000): log(f" Found input: {selector}") inp.click() time.sleep(1) # For contenteditable, use keyboard if "contenteditable" in selector or inp.get_attribute("contenteditable") == "true": page.keyboard.type(PROMPT_TEXT, delay=10) else: inp.fill(PROMPT_TEXT) prompt_entered = True break except Exception as e: log(f" Selector {selector} failed: {e}") continue if not prompt_entered: log(" Trying direct keyboard approach...") page.keyboard.press("Tab") time.sleep(1) page.keyboard.type(PROMPT_TEXT, delay=10) prompt_entered = True time.sleep(2) take_screenshot(page, "08_prompt_entered") # Step 7: Send the message log("Step 7: Sending message...") send_selectors = [ '[aria-label="Send message"]', 'button[aria-label*="Send" i]', '[data-test-id="send-button"]', 'button[type="submit"]', ] sent = False for selector in send_selectors: try: btn = page.locator(selector).first if btn.is_visible(timeout=3000): log(f" Found send button: {selector}") btn.click() sent = True break except: continue if not sent: log(" Send button not found, using Enter key...") page.keyboard.press("Enter") sent = True log(" Message sent!") take_screenshot(page, "09_message_sent") # Step 8: Wait for complete response log("Step 8: Waiting for complete Gemini response (up to 3 minutes)...") log(" This may take 60-180 seconds for Thinking mode...") response_text = "" max_wait = 180 # 3 minutes check_interval = 10 # check every 10 seconds elapsed = 0 last_length = 0 stable_count = 0 while elapsed < max_wait: time.sleep(check_interval) elapsed += check_interval # Try to extract response text try: response_text = page.evaluate(""" () => { // Try model-response elements let responses = Array.from(document.querySelectorAll('model-response')).map(r => r.innerText || '').join('\\n'); if (responses.length > 50) return responses; // Try message-content let msgs = Array.from(document.querySelectorAll('.message-content, .response-content, [data-message-id]')).map(r => r.innerText || '').join('\\n'); if (msgs.length > 50) return msgs; // Try any large text blocks let all = Array.from(document.querySelectorAll('p, .markdown')).map(r => r.innerText || '').filter(t => t.length > 100).join('\\n'); return all; } """) except Exception as e: log(f" Error extracting response: {e}") current_length = len(response_text) log(f" [{elapsed}s] Response length: {current_length} chars") # Check if response is growing or stable if current_length > 100: if current_length == last_length: stable_count += 1 if stable_count >= 3: # stable for 30 seconds log(" Response appears complete (stable for 30s)") break else: stable_count = 0 last_length = current_length # Take progress screenshots at key intervals if elapsed in [30, 60, 120]: take_screenshot(page, f"10_progress_{elapsed}s") # Final screenshot take_screenshot(page, "11_final_response") log(f" Final response length: {len(response_text)} chars") # Step 9: Extract full response with more comprehensive JS log("Step 9: Extracting full response text...") try: full_response = page.evaluate(""" () => { const methods = [ () => Array.from(document.querySelectorAll('model-response')).map(r => r.innerText || '').join('\\n\\n'), () => Array.from(document.querySelectorAll('.response-container, .model-response')).map(r => r.innerText || '').join('\\n\\n'), () => { const msgs = document.querySelectorAll('[data-test-id*="response"], [class*="response"]'); return Array.from(msgs).map(r => r.innerText || '').filter(t => t.length > 50).join('\\n\\n'); }, () => { // Get all paragraph text from the conversation const paras = document.querySelectorAll('p, li, h1, h2, h3'); return Array.from(paras).map(p => p.innerText || '').filter(t => t.trim().length > 0).join('\\n'); } ]; for (const method of methods) { try { const result = method(); if (result && result.length > 100) return result; } catch(e) {} } return document.body.innerText || ''; } """) if full_response and len(full_response) > len(response_text): response_text = full_response except Exception as e: log(f" Full extraction error: {e}") log(f" Extracted response: {len(response_text)} chars") except Exception as e: log(f"CRITICAL ERROR: {e}") take_screenshot(page, "ERROR_state") response_text = f"ERROR: {e}" finally: take_screenshot(page, "99_final_state") browser.close() # Step 10: Save response to file log("Step 10: Saving response to file...") now = datetime.now().strftime("%Y-%m-%d %H:%M:%S") output_content = f"""# Real Gemini Thinking Mode — Genesis Strategy **Date**: 2026-02-19 **Timestamp**: {now} **Mode**: {mode_used} **Account**: {EMAIL} ## Prompt {PROMPT_TEXT} ## Response {response_text if response_text else "[No response captured - check screenshots]"} --- *Generated by Genesis Playwright MCP workflow* *Screenshots saved to: {SCREENSHOTS_DIR}* """ with open(OUTPUT_FILE, "w", encoding="utf-8") as f: f.write(output_content) log(f" Response saved to: {OUTPUT_FILE}") # Step 11: Write status file log("Step 11: Writing status file...") status = "SUCCESS" if response_text and len(response_text) > 200 else "PARTIAL" status_content = f"""# DT Genesis Strategy — Workflow Status **Status**: {status} **Date**: 2026-02-19 **Timestamp**: {now} **Mode Used**: {mode_used} ## Results - Response length: {len(response_text)} characters - Output file: {OUTPUT_FILE} - Screenshots: {SCREENSHOTS_DIR}/gemini_workflow_*.png ## Completion {"Full response captured successfully." if status == "SUCCESS" else "Partial response - check screenshots for details."} --- *Workflow executed by Genesis Playwright automation* """ with open(STATUS_FILE, "w", encoding="utf-8") as f: f.write(status_content) log(f" Status saved to: {STATUS_FILE}") log(f"WORKFLOW COMPLETE - Status: {status}") return { "status": status, "mode_used": mode_used, "response_length": len(response_text), "output_file": OUTPUT_FILE, "status_file": STATUS_FILE, } if __name__ == "__main__": result = main() print(f"\n=== FINAL RESULT ===") print(json.dumps(result, indent=2))