Upload 89 files

app/__init__.py

@@ -0,0 +1,3 @@
+"""Lancer - Advanced AI Search API"""
+
+__version__ = "0.1.0"

app/agents/__init__.py

@@ -0,0 +1 @@
+"""Agents module."""

app/agents/browser_decision.py

@@ -0,0 +1,216 @@
+"""Shared decision helper for browser agents.
+
+This keeps browser action selection consistent across visual and stealth modes,
+while allowing a gradual move from plain JSON prompting to tool calling.
+"""
+
+from __future__ import annotations
+from typing import Any
+
+from app.agents.browser_tools import (
+    execute_browser_tool_call,
+    get_browser_tools,
+    parse_browser_json_response,
+)
+from app.agents.llm_client import generate_completion, generate_completion_response
+
+
+async def decide_browser_action(
+    *,
+    task: str,
+    current_url: str,
+    state,
+    content_preview: str,
+    blocked: bool,
+    allow_scroll: bool,
+    mode_label: str,
+    step_label: str,
+    links: list[str] | None = None,
+    max_tokens: int = 700,
+) -> dict[str, Any]:
+    """Ask the LLM for the next browser action using tools with JSON fallback."""
+    memory_context = state.get_context_for_llm()
+    history_str = "\n".join([f"- {url}" for url in state.visited_urls[-8:]]) or "(none)"
+    known_str = "\n".join([f"- {fact}" for fact in state.known_facts[-6:]]) or "(none yet)"
+    missing_str = "\n".join([f"- {point}" for point in state.missing_points[-6:]]) or "(none)"
+    recent_queries_str = "\n".join([f"- {query}" for query in state.last_queries[-6:]]) or "(none)"
+    links_str = "\n".join([f"- {url}" for url in (links or [])[:10]]) or "(none)"
+
+    scroll_rule = (
+        "- `scroll_page`: only if the current page likely contains the missing answer but needs more content\n"
+        if allow_scroll
+        else ""
+    )
+
+    prompt = f"""You are a {mode_label} browser agent.
+
+Choose the single best next action for this task. Use tool calls when available.
+If tool calling is unavailable, return valid JSON matching the same intent.
+Never reveal chain-of-thought, hidden reasoning, deliberation, or internal analysis.
+Never answer with prose paragraphs unless you are placing the final user-facing answer inside `finish_task` or JSON `answer`.
+
+TASK: {task}
+CURRENT URL: {current_url}
+STEP: {step_label}
+BLOCKED: {blocked}
+
+MEMORY:
+{memory_context}
+
+KNOWN FACTS:
+{known_str}
+
+MISSING INFO:
+{missing_str}
+
+RECENT QUERIES:
+{recent_queries_str}
+
+VISITED URLS:
+{history_str}
+
+CURRENT PAGE CONTENT:
+{content_preview or "(empty page)"}
+
+DISCOVERED LINKS:
+{links_str}
+
+Prefer these tools:
+- `search_web`: when current evidence is insufficient and you need a new query
+- `navigate_to_url`: when a discovered URL is the best next source and has not been visited
+{scroll_rule}- `finish_task`: when you have enough evidence to answer now
+
+Rules:
+1. Do not revisit URLs already listed under VISITED URLS
+2. If the page is blocked, prefer search or a different navigation target immediately
+3. Keep `known_facts` and `missing_points` concise and session-specific
+4. If you finish, provide the answer in the tool arguments or JSON output
+"""
+
+    response = await generate_completion_response(
+        messages=[{"role": "user", "content": prompt}],
+        max_tokens=max_tokens,
+        tools=get_browser_tools(allow_scroll=allow_scroll),
+        tool_choice="auto",
+        reasoning_effort="medium",
+        prefer_responses_api=True,
+    )
+
+    if response.tool_calls:
+        last_error: Exception | None = None
+        for tool_call in response.tool_calls:
+            try:
+                return execute_browser_tool_call(tool_call, allow_scroll=allow_scroll)
+            except Exception as exc:
+                last_error = exc
+        if last_error is not None:
+            raise last_error
+
+    if response.content:
+        try:
+            return parse_browser_json_response(response.content, allow_scroll=allow_scroll)
+        except Exception:
+            repaired = await _repair_browser_decision_text(
+                raw_text=response.content,
+                allow_scroll=allow_scroll,
+            )
+            if repaired is not None:
+                return repaired
+
+            return _fallback_browser_decision(
+                task=task,
+                current_url=current_url,
+                blocked=blocked,
+                allow_scroll=allow_scroll,
+                links=links or [],
+                raw_text=response.content,
+            )
+
+    return _fallback_browser_decision(
+        task=task,
+        current_url=current_url,
+        blocked=blocked,
+        allow_scroll=allow_scroll,
+        links=links or [],
+        raw_text="",
+    )
+
+
+async def _repair_browser_decision_text(
+    *,
+    raw_text: str,
+    allow_scroll: bool,
+) -> dict[str, Any] | None:
+    """Ask the model to convert its previous free-form text into strict JSON."""
+    prompt = f"""Convert the text below into exactly one valid JSON object for a browser action.
+
+Allowed actions:
+- SEARCH with field `query`
+- NAVIGATE with field `url`
+{"- SCROLL with no extra field" if allow_scroll else ""}
+- DONE with field `answer`
+
+Rules:
+1. Output JSON only
+2. Do not include reasoning
+3. Keep `reason`, `known_facts`, and `missing_points` concise
+4. If the text is only internal reasoning and not a final answer, prefer SEARCH, NAVIGATE, or {"SCROLL" if allow_scroll else "SEARCH"} over DONE
+
+TEXT:
+{raw_text}
+"""
+    try:
+        repaired_text = await generate_completion(
+            messages=[{"role": "user", "content": prompt}],
+            max_tokens=300,
+        )
+        return parse_browser_json_response(repaired_text, allow_scroll=allow_scroll)
+    except Exception:
+        return None
+
+
+def _fallback_browser_decision(
+    *,
+    task: str,
+    current_url: str,
+    blocked: bool,
+    allow_scroll: bool,
+    links: list[str],
+    raw_text: str,
+) -> dict[str, Any]:
+    """Choose a safe next action when the model fails to return structured output."""
+    unseen_links = [url for url in links if url.startswith("http")]
+    if unseen_links:
+        return {
+            "action": "NAVIGATE",
+            "value": unseen_links[0],
+            "answer": "",
+            "reason": "Fallback navigation because the model returned unstructured output",
+            "known_facts": [],
+            "missing_points": [],
+        }
+
+    if allow_scroll and not blocked and current_url.startswith("http"):
+        return {
+            "action": "SCROLL",
+            "value": "",
+            "answer": "",
+            "reason": "Fallback scroll because the model returned unstructured output",
+            "known_facts": [],
+            "missing_points": [],
+        }
+
+    query = task.strip()
+    if current_url.startswith("https://html.duckduckgo.com/"):
+        query = f"{task.strip()} answer"
+    elif blocked:
+        query = f"{task.strip()} alternate source"
+
+    return {
+        "action": "SEARCH",
+        "value": query,
+        "answer": "",
+        "reason": "Fallback search because the model returned unstructured output",
+        "known_facts": [],
+        "missing_points": [],
+    }

app/agents/browser_dom.py

@@ -0,0 +1,142 @@
+"""Helpers for DOM-based extraction in browser agents."""
+
+from __future__ import annotations
+
+import html
+from html.parser import HTMLParser
+import json
+import re
+
+
+class _DomSnapshotParser(HTMLParser):
+    """Extract visible text and absolute links from HTML."""
+
+    def __init__(self, max_links: int = 12):
+        super().__init__()
+        self.max_links = max_links
+        self._skip_depth = 0
+        self._text_parts: list[str] = []
+        self._links: list[str] = []
+
+    def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None:
+        lowered = tag.lower()
+        if lowered in {"script", "style", "noscript"}:
+            self._skip_depth += 1
+            return
+
+        if self._skip_depth:
+            return
+
+        if lowered == "a":
+            href = dict(attrs).get("href") or ""
+            href = href.strip()
+            if href.startswith("http") and href not in self._links and len(self._links) < self.max_links:
+                self._links.append(href)
+
+    def handle_endtag(self, tag: str) -> None:
+        if tag.lower() in {"script", "style", "noscript"} and self._skip_depth:
+            self._skip_depth -= 1
+
+    def handle_data(self, data: str) -> None:
+        if self._skip_depth:
+            return
+
+        text = data.strip()
+        if text:
+            self._text_parts.append(text)
+
+    def snapshot(self, max_chars: int = 6000) -> tuple[str, list[str]]:
+        text = html.unescape(" ".join(self._text_parts))
+        text = re.sub(r"\s+", " ", text).strip()
+        return text[:max_chars], self._links[: self.max_links]
+
+
+def extract_dom_snapshot(html_text: str, max_chars: int = 6000, max_links: int = 12) -> tuple[str, list[str]]:
+    """Extract visible text and absolute links from a rendered DOM snapshot."""
+    parser = _DomSnapshotParser(max_links=max_links)
+    parser.feed(html_text or "")
+    parser.close()
+    return parser.snapshot(max_chars=max_chars)
+
+
+def build_visual_dom_extract_script(url: str, max_chars: int = 6000, max_links: int = 12) -> str:
+    """Build a sandbox-side Python script that extracts rendered DOM via headless Chrome."""
+    quoted_url = json.dumps(url)
+    return f'''
+import json
+import subprocess
+import sys
+from html.parser import HTMLParser
+import html as html_lib
+import re
+
+URL = {quoted_url}
+MAX_CHARS = {max_chars}
+MAX_LINKS = {max_links}
+
+class DomSnapshotParser(HTMLParser):
+    def __init__(self):
+        super().__init__()
+        self.skip_depth = 0
+        self.text_parts = []
+        self.links = []
+
+    def handle_starttag(self, tag, attrs):
+        lowered = tag.lower()
+        if lowered in ("script", "style", "noscript"):
+            self.skip_depth += 1
+            return
+        if self.skip_depth:
+            return
+        if lowered == "a":
+            href = dict(attrs).get("href") or ""
+            href = href.strip()
+            if href.startswith("http") and href not in self.links and len(self.links) < MAX_LINKS:
+                self.links.append(href)
+
+    def handle_endtag(self, tag):
+        if tag.lower() in ("script", "style", "noscript") and self.skip_depth:
+            self.skip_depth -= 1
+
+    def handle_data(self, data):
+        if self.skip_depth:
+            return
+        text = data.strip()
+        if text:
+            self.text_parts.append(text)
+
+    def snapshot(self):
+        text = html_lib.unescape(" ".join(self.text_parts))
+        text = re.sub(r"\\s+", " ", text).strip()
+        return text[:MAX_CHARS], self.links[:MAX_LINKS]
+
+def is_blocked(text):
+    lowered = text.lower()
+    markers = ["checking your browser", "cloudflare", "access denied", "just a moment", "enable javascript"]
+    return len(text) < 800 and any(marker in lowered for marker in markers)
+
+def dump_dom():
+    commands = [
+        ["google-chrome", "--headless=new", "--disable-gpu", "--no-sandbox", "--dump-dom", URL],
+        ["google-chrome", "--headless", "--disable-gpu", "--no-sandbox", "--dump-dom", URL],
+    ]
+    last_error = None
+    for command in commands:
+        try:
+            result = subprocess.run(command, capture_output=True, text=True, timeout=25, check=True)
+            return result.stdout
+        except Exception as exc:
+            last_error = exc
+    raise last_error or RuntimeError("Could not dump DOM")
+
+try:
+    dom_html = dump_dom()
+    parser = DomSnapshotParser()
+    parser.feed(dom_html)
+    parser.close()
+    content, links = parser.snapshot()
+    print(json.dumps({{"content": content, "links": links, "blocked": is_blocked(content)}}))
+except Exception as exc:
+    print(json.dumps({{"error": str(exc), "content": "", "links": [], "blocked": False}}))
+    sys.exit(0)
+'''

app/agents/browser_search.py

@@ -0,0 +1,56 @@
+"""Search URL helpers for browser agents."""
+
+from __future__ import annotations
+
+from urllib.parse import quote_plus
+
+
+SEARCH_ENGINES: tuple[tuple[str, str], ...] = (
+    ("bing", "https://www.bing.com/search?q={query}"),
+    ("wikipedia", "https://en.wikipedia.org/w/index.php?search={query}"),
+    ("duckduckgo", "https://html.duckduckgo.com/html/?q={query}"),
+)
+
+
+def build_search_url(query: str, engine: str = "bing") -> str:
+    """Build a search URL for the requested engine."""
+    normalized_query = quote_plus((query or "").strip())
+    for engine_name, template in SEARCH_ENGINES:
+        if engine_name == engine:
+            return template.format(query=normalized_query)
+    return SEARCH_ENGINES[0][1].format(query=normalized_query)
+
+
+def detect_search_engine(url: str) -> str | None:
+    """Infer which configured engine a URL belongs to."""
+    lowered = (url or "").lower()
+    if "bing.com/search" in lowered:
+        return "bing"
+    if "wikipedia.org/w/index.php?search=" in lowered:
+        return "wikipedia"
+    if "duckduckgo.com/html/" in lowered:
+        return "duckduckgo"
+    return None
+
+
+def choose_search_url(
+    query: str,
+    visited_urls: list[str],
+    current_url: str = "",
+    blocked: bool = False,
+) -> str:
+    """Choose the next search URL, rotating engines to avoid dead loops."""
+    current_engine = detect_search_engine(current_url)
+    engine_order = [engine for engine, _ in SEARCH_ENGINES]
+
+    if blocked and current_engine in engine_order:
+        engine_order = [engine for engine in engine_order if engine != current_engine] + [current_engine]
+
+    for engine in engine_order:
+        candidate = build_search_url(query, engine=engine)
+        if candidate not in visited_urls:
+            return candidate
+
+    if current_engine:
+        return build_search_url(query, engine=current_engine)
+    return build_search_url(query, engine="bing")

app/agents/browser_stealth.py

@@ -0,0 +1,312 @@
+"""Stealth browser agent - Camoufox with full agentic navigation.
+
+Camoufox = Firefox stealth que passa anti-bot.
+Roda DENTRO do E2B sandbox.
+Full agentic loop with LLM-driven navigation.
+Time limit: 5 minutes (300 seconds)
+"""
+
+import os
+import json
+import logging
+import time
+from typing import AsyncGenerator, Optional
+
+from app.agents.browser_decision import decide_browser_action
+from app.agents.browser_search import choose_search_url
+from app.config import get_settings
+from app.agents.llm_client import generate_completion
+from app.agents.graph.state import AgentState
+
+logger = logging.getLogger(__name__)
+
+MAX_TIME_SECONDS = 300  # 5 minutes
+MAX_PAGES = 5
+STEALTH_SANDBOX_TIMEOUT_SECONDS = 600
+
+
+async def run_browser_stealth_agent(
+    task: str,
+    url: Optional[str] = None,
+) -> AsyncGenerator[dict, None]:
+    """Run the stealth browser agent with Camoufox."""
+    settings = get_settings()
+    
+    if not settings.e2b_api_key:
+        yield {"type": "error", "message": "E2B_API_KEY not configured"}
+        return
+    
+    # Initialize agent state
+    state = AgentState(
+        task=task,
+        url=url,
+        timeout_seconds=MAX_TIME_SECONDS,
+        start_time=time.time()
+    )
+    
+    yield {"type": "status", "message": "🚀 Initializing stealth agent..."}
+    
+    desktop = None
+    
+    try:
+        from e2b_desktop import Sandbox
+        
+        os.environ["E2B_API_KEY"] = settings.e2b_api_key
+        
+        yield {"type": "status", "message": "🖥️ Creating sandbox..."}
+        desktop = Sandbox.create(timeout=STEALTH_SANDBOX_TIMEOUT_SECONDS)
+        state.desktop = desktop
+        
+        # Install Camoufox
+        yield {"type": "status", "message": "📦 Installing stealth browser..."}
+        
+        try:
+            desktop.commands.run("pip install --user camoufox playwright -q", timeout=120)
+            yield {"type": "status", "message": "🔽 Downloading Firefox stealth (~30s)..."}
+            desktop.commands.run("camoufox fetch", timeout=180)
+            desktop.commands.run("sudo apt-get update -qq && sudo apt-get install -y -qq libgtk-3-0 libasound2 libdbus-glib-1-2 2>/dev/null || true", timeout=60)
+            yield {"type": "status", "message": "✅ Browser ready!"}
+        except Exception as e:
+            logger.error(f"Camoufox install failed: {e}")
+            yield {"type": "error", "message": f"Install failed: {e}"}
+            return
+        
+        # Build initial URL
+        if url:
+            start_url = url
+        else:
+            start_url = choose_search_url(task, visited_urls=state.visited_urls)
+            state.add_query(task)
+        
+        state.visited_urls.append(start_url)
+        state.add_action({"type": "start", "url": start_url})
+        
+        # Agentic loop
+        while state.should_continue() and state.step_count < MAX_PAGES:
+            state.step_count += 1
+            elapsed = int(state.get_elapsed_time())
+            remaining = int(state.get_remaining_time())
+            
+            current_url = state.visited_urls[-1]
+            
+            yield {"type": "status", "message": f"🔍 Step {state.step_count}: Fetching {current_url[:40]}... ({elapsed}s)"}
+            
+            # Fetch page with Camoufox
+            script = _build_fetch_script(current_url)
+            desktop.commands.run(f"cat > /tmp/fetch.py << 'EOF'\n{script}\nEOF", timeout=10)
+            
+            result = desktop.commands.run("python3 /tmp/fetch.py", timeout=60)
+            output = result.stdout.strip() if hasattr(result, 'stdout') else ""
+            
+            # Parse result
+            page_content = ""
+            page_links = []
+            is_blocked = False
+            
+            try:
+                data = json.loads(output)
+                page_content = data.get("content", "")
+                page_links = data.get("links", [])
+                is_blocked = data.get("blocked", False)
+                
+                if data.get("error"):
+                    state.add_error(data["error"])
+            except json.JSONDecodeError:
+                page_content = output[:3000]
+            
+            if is_blocked:
+                yield {"type": "status", "message": f"🚫 Blocked at {current_url[:30]}..., trying next..."}
+                state.add_error(f"Blocked: {current_url}")
+            else:
+                state.extracted_data.append({
+                    "url": current_url,
+                    "content_length": len(page_content),
+                    "links_found": len(page_links),
+                    "preview": page_content[:300]
+                })
+            
+            decision = await decide_browser_action(
+                task=task,
+                current_url=current_url,
+                state=state,
+                content_preview=page_content[:2000] if page_content else "(empty)",
+                blocked=is_blocked,
+                allow_scroll=False,
+                mode_label="stealth Camoufox",
+                step_label=f"{state.step_count}/{MAX_PAGES}, {remaining}s remaining",
+                links=page_links,
+                max_tokens=650,
+            )
+            
+            action = decision.get("action", "DONE")
+            value = decision.get("value", "")
+            final_answer = decision.get("answer", "")
+            reason = decision.get("reason", "")
+            known_facts = decision.get("known_facts", [])
+            missing_points = decision.get("missing_points", [])
+
+            if action == "SEARCH":
+                state.add_query(value)
+
+            if isinstance(known_facts, list) or isinstance(missing_points, list):
+                state.update_research_progress(
+                    known_facts=known_facts if isinstance(known_facts, list) else None,
+                    missing_points=missing_points if isinstance(missing_points, list) else None,
+                )
+            
+            state.add_action({"type": action.lower(), "value": value, "reason": reason})
+            
+            yield {"type": "status", "message": f"🤔 {action}: {reason[:40]}"}
+            
+            yield {
+                "type": "progress",
+                "known_facts": state.known_facts[-8:],
+                "missing_points": state.missing_points[-8:],
+                "last_queries": state.last_queries[-8:],
+            }
+
+            if action == "DONE":
+                state.success = True
+                
+                if not final_answer:
+                    # Generate from memory
+                    all_content = "\n\n".join([
+                        f"Source: {d['url']}\n{d.get('preview', '')}"
+                        for d in state.extracted_data[-5:]
+                    ])
+                    known_summary = "\n".join([f"- {f}" for f in state.known_facts[-8:]]) or "(none)"
+                    missing_summary = "\n".join([f"- {m}" for m in state.missing_points[-8:]]) or "(none)"
+                    final_prompt = (
+                        f"Answer this: {task}\n\n"
+                        f"Known facts:\n{known_summary}\n\n"
+                        f"Missing points:\n{missing_summary}\n\n"
+                        f"Content:\n{all_content}"
+                    )
+                    final_answer = await generate_completion(
+                        messages=[{"role": "user", "content": final_prompt}],
+                        max_tokens=1200
+                    )
+                
+                state.final_result = final_answer
+                
+                yield {"type": "stream_end", "message": "Done"}
+                yield {
+                    "type": "result",
+                    "content": final_answer,
+                    "links": state.visited_urls,
+                    "steps": state.step_count,
+                    "success": True
+                }
+                yield {"type": "complete", "message": f"Done in {int(state.get_elapsed_time())}s (stealth)"}
+                return
+            
+            elif action == "NAVIGATE":
+                if value and value.startswith("http"):
+                    if value not in state.visited_urls:
+                        state.visited_urls.append(value)
+                    else:
+                        state.add_error(f"Tried revisit: {value}")
+            
+            elif action == "SEARCH":
+                new_url = choose_search_url(
+                    value,
+                    visited_urls=state.visited_urls,
+                    current_url=current_url,
+                    blocked=is_blocked,
+                )
+                if new_url not in state.visited_urls:
+                    state.visited_urls.append(new_url)
+        
+        # Timeout or max pages - generate from memory
+        yield {"type": "status", "message": "⏰ Generating final answer from memory..."}
+        
+        all_content = "\n\n".join([
+            f"Source: {d['url']}\n{d.get('preview', '')}"
+            for d in state.extracted_data[-5:]
+        ])
+        known_summary = "\n".join([f"- {f}" for f in state.known_facts[-8:]]) or "(none)"
+        missing_summary = "\n".join([f"- {m}" for m in state.missing_points[-8:]]) or "(none)"
+        final_prompt = (
+            f"Answer this: {task}\n\n"
+            f"Known facts:\n{known_summary}\n\n"
+            f"Missing points:\n{missing_summary}\n\n"
+            f"Content:\n{all_content}"
+        )
+        final_answer = await generate_completion(
+            messages=[{"role": "user", "content": final_prompt}],
+            max_tokens=1200
+        )
+        
+        state.final_result = final_answer
+        
+        yield {"type": "stream_end", "message": "Done"}
+        yield {
+            "type": "result",
+            "content": final_answer,
+            "links": state.visited_urls,
+            "steps": state.step_count,
+            "success": True
+        }
+        yield {"type": "complete", "message": f"Done in {int(state.get_elapsed_time())}s (stealth, {state.step_count} pages)"}
+        
+    except ImportError:
+        yield {"type": "error", "message": "e2b-desktop not installed"}
+    except Exception as e:
+        logger.exception("Stealth agent error")
+        yield {"type": "error", "message": str(e)}
+    finally:
+        if desktop:
+            try:
+                desktop.kill()
+            except:
+                pass
+
+
+def _build_fetch_script(url: str) -> str:
+    """Build Python script to fetch a URL with Camoufox."""
+    return f'''
+import json
+import sys
+
+try:
+    from camoufox.sync_api import Camoufox
+except:
+    print(json.dumps({{"error": "Camoufox not found"}}))
+    sys.exit(1)
+
+def is_blocked(text):
+    t = text.lower()
+    blocks = ["checking your browser", "cloudflare", "access denied", "just a moment", "enable javascript"]
+    return len(text) < 800 and any(b in t for b in blocks)
+
+try:
+    with Camoufox(headless=True) as browser:
+        page = browser.new_page()
+        page.goto("{url}", timeout=30000)
+        page.wait_for_timeout(2000)
+        
+        # Extract text
+        content = page.evaluate("""() => {{
+            document.querySelectorAll('script,style,noscript').forEach(e => e.remove());
+            return document.body.innerText || '';
+        }}""")[:5000]
+        
+        # Extract links
+        links = page.evaluate("""() => {{
+            return Array.from(document.querySelectorAll('a[href^="http"]'))
+                .map(a => a.href)
+                .filter(h => !h.includes('duckduckgo') && !h.includes('google'))
+                .slice(0, 10);
+        }}""")
+        
+        blocked = is_blocked(content)
+        
+        print(json.dumps({{
+            "content": content,
+            "links": links,
+            "blocked": blocked
+        }}))
+
+except Exception as e:
+    print(json.dumps({{"error": str(e)}}))
+'''

app/agents/browser_tools.py

@@ -0,0 +1,237 @@
+"""Reusable tooling for browser-agent action selection.
+
+This module gives the browser agents a shared, validated action surface.
+It supports both:
+- native tool calling when the LLM/provider can emit tool calls
+- JSON fallback when the model only returns plain text
+"""
+
+from __future__ import annotations
+
+import json
+from typing import Any
+
+from app.agents.tooling import ToolCall, ToolRegistry, tool
+
+
+def _normalize_points(values: list[str] | str | None) -> list[str]:
+    """Trim and deduplicate short research memory lists.
+
+    Models occasionally return a plain string instead of a list. Treat that as a
+    single item instead of iterating character-by-character.
+    """
+    cleaned: list[str] = []
+    if values is None:
+        iterable: list[Any] = []
+    elif isinstance(values, str):
+        iterable = [values]
+    else:
+        iterable = list(values)
+
+    for value in iterable:
+        text = str(value).strip()
+        if len(text) <= 1 and text.isalpha():
+            continue
+        if text and text not in cleaned:
+            cleaned.append(text)
+    return cleaned
+
+
+@tool(description="Search the web with a fresh query when the current page is insufficient.")
+def search_web(
+    query: str,
+    reason: str = "",
+    known_facts: list[str] | None = None,
+    missing_points: list[str] | None = None,
+) -> dict[str, Any]:
+    """Search the web.
+
+    Args:
+        query: Query terms to search for next.
+        reason: Why a new search is needed.
+        known_facts: Short facts already established.
+        missing_points: What information is still missing.
+    """
+    return {
+        "action": "SEARCH",
+        "value": query.strip(),
+        "reason": reason.strip(),
+        "known_facts": _normalize_points(known_facts),
+        "missing_points": _normalize_points(missing_points),
+    }
+
+
+@tool(description="Open a new URL that was discovered in the current page or results.")
+def navigate_to_url(
+    url: str,
+    reason: str = "",
+    known_facts: list[str] | None = None,
+    missing_points: list[str] | None = None,
+) -> dict[str, Any]:
+    """Navigate to a specific URL.
+
+    Args:
+        url: Absolute URL to visit next.
+        reason: Why that URL is the best next step.
+        known_facts: Short facts already established.
+        missing_points: What information is still missing.
+    """
+    return {
+        "action": "NAVIGATE",
+        "value": url.strip(),
+        "reason": reason.strip(),
+        "known_facts": _normalize_points(known_facts),
+        "missing_points": _normalize_points(missing_points),
+    }
+
+
+@tool(description="Scroll the current page to reveal more content.")
+def scroll_page(
+    reason: str = "",
+    known_facts: list[str] | None = None,
+    missing_points: list[str] | None = None,
+) -> dict[str, Any]:
+    """Scroll the current page.
+
+    Args:
+        reason: Why scrolling is useful right now.
+        known_facts: Short facts already established.
+        missing_points: What information is still missing.
+    """
+    return {
+        "action": "SCROLL",
+        "value": "",
+        "reason": reason.strip(),
+        "known_facts": _normalize_points(known_facts),
+        "missing_points": _normalize_points(missing_points),
+    }
+
+
+@tool(description="Finish the task and provide the final answer based on the collected evidence.")
+def finish_task(
+    answer: str,
+    reason: str = "",
+    known_facts: list[str] | None = None,
+    missing_points: list[str] | None = None,
+) -> dict[str, Any]:
+    """Finish the task.
+
+    Args:
+        answer: Final user-facing answer.
+        reason: Why the task is complete.
+        known_facts: Short facts already established.
+        missing_points: Remaining uncertainty, if any.
+    """
+    return {
+        "action": "DONE",
+        "value": "",
+        "answer": answer.strip(),
+        "reason": reason.strip(),
+        "known_facts": _normalize_points(known_facts),
+        "missing_points": _normalize_points(missing_points),
+    }
+
+
+BROWSER_TOOL_REGISTRY = ToolRegistry([
+    search_web,
+    navigate_to_url,
+    scroll_page,
+    finish_task,
+])
+
+
+def get_browser_tools(allow_scroll: bool = True) -> list[dict[str, Any]]:
+    """Return OpenAI-compatible tool schemas for the browser agent."""
+    tools = []
+    for schema in BROWSER_TOOL_REGISTRY.schemas:
+        if not allow_scroll and schema.name == "scroll_page":
+            continue
+        tools.append(schema.to_openai_tool())
+    return tools
+
+
+def execute_browser_tool_call(tool_call: ToolCall, allow_scroll: bool = True) -> dict[str, Any]:
+    """Execute a browser decision tool call and validate mode-specific constraints."""
+    if not allow_scroll and tool_call.name == "scroll_page":
+        raise ValueError("scroll_page is not allowed for this browser mode")
+
+    result = BROWSER_TOOL_REGISTRY.execute(tool_call)
+    return validate_browser_decision(result, allow_scroll=allow_scroll)
+
+
+def parse_browser_json_response(text: str, allow_scroll: bool = True) -> dict[str, Any]:
+    """Parse legacy JSON action output into the normalized browser-decision shape."""
+    snippet = _extract_json_object(text)
+    data = json.loads(snippet)
+
+    action = str(data.get("action", "DONE")).strip().upper()
+    normalized = {
+        "action": action,
+        "value": str(data.get("value", "")).strip(),
+        "answer": str(data.get("answer", "")).strip(),
+        "reason": str(data.get("reason", "")).strip(),
+        "known_facts": _normalize_points(data.get("known_facts")),
+        "missing_points": _normalize_points(data.get("missing_points")),
+    }
+
+    if action == "SEARCH":
+        normalized["value"] = str(data.get("query", normalized["value"])).strip()
+    elif action == "NAVIGATE":
+        normalized["value"] = str(data.get("url", normalized["value"])).strip()
+    elif action == "DONE":
+        normalized["answer"] = str(data.get("answer", data.get("result", normalized["answer"]))).strip()
+    elif action == "SCROLL":
+        normalized["value"] = ""
+
+    return validate_browser_decision(normalized, allow_scroll=allow_scroll)
+
+
+def validate_browser_decision(decision: dict[str, Any], allow_scroll: bool = True) -> dict[str, Any]:
+    """Validate and normalize a browser agent decision."""
+    action = str(decision.get("action", "")).strip().upper()
+    value = str(decision.get("value", "")).strip()
+    answer = str(decision.get("answer", "")).strip()
+
+    normalized = {
+        "action": action or "DONE",
+        "value": value,
+        "answer": answer,
+        "reason": str(decision.get("reason", "")).strip(),
+        "known_facts": _normalize_points(decision.get("known_facts")),
+        "missing_points": _normalize_points(decision.get("missing_points")),
+    }
+
+    if normalized["action"] == "SEARCH":
+        if not normalized["value"]:
+            raise ValueError("SEARCH decision requires a non-empty query")
+    elif normalized["action"] == "NAVIGATE":
+        if not normalized["value"].startswith("http"):
+            raise ValueError("NAVIGATE decision requires an absolute URL")
+    elif normalized["action"] == "SCROLL":
+        if not allow_scroll:
+            raise ValueError("SCROLL is not supported in this browser mode")
+        normalized["value"] = ""
+    elif normalized["action"] == "DONE":
+        pass
+    else:
+        raise ValueError(f"Unsupported browser action '{normalized['action']}'")
+
+    return normalized
+
+
+def _extract_json_object(text: str) -> str:
+    """Extract the first JSON object-looking slice from a model response."""
+    raw = (text or "").strip()
+    if raw.startswith("```"):
+        parts = raw.split("```")
+        if len(parts) >= 2:
+            raw = parts[1]
+            if raw.startswith("json"):
+                raw = raw[4:]
+            raw = raw.strip()
+
+    start = raw.find("{")
+    end = raw.rfind("}")
+    if start == -1 or end == -1 or end <= start:
+        raise ValueError("Model response did not contain a JSON object")
+    return raw[start:end + 1]

app/agents/browser_visual.py

@@ -0,0 +1,293 @@
+"""Visual browser agent - Chrome with live stream and agent memory.
+
+Uses E2B Desktop sandbox with Chrome browser.
+Time limit: 5 minutes (300 seconds)
+Shows live video stream.
+Includes full memory/history tracking via AgentState.
+"""
+
+import os
+import shlex
+import logging
+import time
+import json
+from typing import AsyncGenerator, Optional
+
+from app.agents.browser_dom import build_visual_dom_extract_script
+from app.agents.browser_search import choose_search_url
+from app.agents.browser_decision import decide_browser_action
+from app.config import get_settings
+from app.agents.llm_client import generate_completion
+from app.agents.graph.state import AgentState
+from app.agents.flaresolverr import is_cloudflare_blocked
+
+logger = logging.getLogger(__name__)
+
+MAX_TIME_SECONDS = 300  # 5 minutes
+
+
+async def run_browser_visual_agent(
+    task: str,
+    url: Optional[str] = None,
+) -> AsyncGenerator[dict, None]:
+    """Run the visual browser agent with Chrome and live stream."""
+    settings = get_settings()
+    
+    if not settings.e2b_api_key:
+        yield {"type": "error", "message": "E2B_API_KEY not configured"}
+        return
+    
+    # Initialize agent state with memory
+    state = AgentState(
+        task=task,
+        url=url,
+        timeout_seconds=MAX_TIME_SECONDS,
+        start_time=time.time()
+    )
+    
+    yield {"type": "status", "message": "🚀 Initializing agent..."}
+    
+    desktop = None
+    
+    try:
+        from e2b_desktop import Sandbox
+        
+        os.environ["E2B_API_KEY"] = settings.e2b_api_key
+        
+        yield {"type": "status", "message": "🖥️ Creating virtual desktop..."}
+        desktop = Sandbox.create(timeout=600)
+        state.desktop = desktop
+        
+        # Start streaming
+        stream_url = None
+        try:
+            desktop.stream.start(require_auth=True)
+            auth_key = desktop.stream.get_auth_key()
+            stream_url = desktop.stream.get_url(auth_key=auth_key)
+            yield {"type": "stream", "url": stream_url}
+            logger.info(f"Stream started: {stream_url}")
+            desktop.wait(2000)
+        except Exception as e:
+            logger.warning(f"Could not start stream: {e}")
+        
+        # Launch Chrome
+        yield {"type": "status", "message": "🌐 Launching browser..."}
+        
+        if url:
+            start_url = url
+        else:
+            start_url = choose_search_url(task, visited_urls=state.visited_urls)
+            state.add_query(task)
+        
+        chrome_flags = "--no-sandbox --disable-gpu --start-maximized --no-first-run --disable-default-apps --disable-popup-blocking --disable-translate --no-default-browser-check"
+        desktop.commands.run(f"google-chrome {chrome_flags} {shlex.quote(start_url)} &", background=True)
+        desktop.wait(3000)
+        
+        # Close dialogs
+        desktop.press("enter")
+        desktop.wait(1000)
+        
+        # Add to memory
+        state.visited_urls.append(start_url)
+        state.add_action({"type": "navigate", "url": start_url})
+        
+        # Main loop - time based with memory
+        while state.should_continue():
+            state.step_count += 1
+            elapsed = int(state.get_elapsed_time())
+            remaining = int(state.get_remaining_time())
+            
+            yield {"type": "status", "message": f"🔍 Step {state.step_count}: Analyzing... ({elapsed}s / {MAX_TIME_SECONDS}s)"}
+            
+            # Get page content
+            current_url = state.visited_urls[-1]
+            page_content = ""
+            page_links: list[str] = []
+            extracted_blocked = False
+            
+            try:
+                script = build_visual_dom_extract_script(current_url)
+                desktop.commands.run(f"cat > /tmp/visual_dom_extract.py << 'EOF'\n{script}\nEOF", timeout=10)
+                result = desktop.commands.run("python3 /tmp/visual_dom_extract.py", timeout=45)
+                output = result.stdout.strip() if hasattr(result, "stdout") else ""
+                data = json.loads(output) if output else {}
+                page_content = str(data.get("content", "") or "")
+                page_links = [
+                    link for link in (data.get("links", []) or [])
+                    if isinstance(link, str) and link.startswith("http")
+                ]
+                extracted_blocked = bool(data.get("blocked", False))
+                state.page_content = page_content
+                if data.get("error"):
+                    state.add_error(f"DOM extraction warning: {data['error']}")
+            except Exception as e:
+                logger.warning(f"DOM extraction failed: {e}")
+                state.add_error(f"DOM extraction failed: {e}")
+
+            preview_text = page_content[:2000] if page_content else "(empty page)"
+            
+            # Check for Cloudflare block
+            is_blocked = extracted_blocked or (is_cloudflare_blocked(page_content) if page_content else False)
+            
+            if is_blocked:
+                yield {"type": "status", "message": f"🚫 Cloudflare at {current_url[:40]}..., trying next link..."}
+                state.add_error(f"Cloudflare blocked: {current_url}")
+            else:
+                # Add to memory
+                state.extracted_data.append({
+                    "url": current_url,
+                    "content_length": len(page_content),
+                    "links_found": len(page_links),
+                    "preview": page_content[:200]
+                })
+            
+            decision = await decide_browser_action(
+                task=task,
+                current_url=current_url,
+                state=state,
+                content_preview=preview_text,
+                blocked=is_blocked,
+                allow_scroll=False,
+                mode_label="visual Chrome",
+                step_label=f"{state.step_count}, {remaining}s remaining",
+                links=page_links,
+                max_tokens=600,
+            )
+            
+            action = decision.get("action", "DONE")
+            value = decision.get("value", "")
+            final_answer = decision.get("answer", "")
+            reason = decision.get("reason", "")
+            known_facts = decision.get("known_facts", [])
+            missing_points = decision.get("missing_points", [])
+
+            if action == "SEARCH":
+                state.add_query(value)
+
+            if isinstance(known_facts, list) or isinstance(missing_points, list):
+                state.update_research_progress(
+                    known_facts=known_facts if isinstance(known_facts, list) else None,
+                    missing_points=missing_points if isinstance(missing_points, list) else None,
+                )
+            
+            # Record action in memory
+            state.add_action({"type": action.lower(), "value": value, "reason": reason})
+            
+            yield {"type": "status", "message": f"🤔 Action: {action} - {reason[:50]}"}
+            
+            yield {
+                "type": "progress",
+                "known_facts": state.known_facts[-8:],
+                "missing_points": state.missing_points[-8:],
+                "last_queries": state.last_queries[-8:],
+            }
+
+            if action == "DONE":
+                state.success = True
+                
+                if not final_answer:
+                    # Generate from memory
+                    all_content = "\n\n".join([
+                        f"Source: {d['url']}\n{d.get('preview', '')}" 
+                        for d in state.extracted_data[-5:]
+                    ])
+                    known_summary = "\n".join([f"- {f}" for f in state.known_facts[-8:]]) or "(none)"
+                    missing_summary = "\n".join([f"- {m}" for m in state.missing_points[-8:]]) or "(none)"
+                    final_prompt = (
+                        f"Based on this content, answer: {task}\n\n"
+                        f"Known facts:\n{known_summary}\n\n"
+                        f"Missing points:\n{missing_summary}\n\n"
+                        f"Content:\n{all_content}"
+                    )
+                    final_answer = await generate_completion(
+                        messages=[{"role": "user", "content": final_prompt}],
+                        max_tokens=1000
+                    )
+                
+                state.final_result = final_answer
+                
+                yield {"type": "stream_end", "message": "Done"}
+                yield {
+                    "type": "result",
+                    "content": final_answer,
+                    "links": state.visited_urls,
+                    "steps": state.step_count,
+                    "success": True
+                }
+                
+                yield {"type": "complete", "message": f"Completed in {int(state.get_elapsed_time())}s with {state.step_count} steps"}
+                return
+            
+            elif action == "SEARCH":
+                new_url = choose_search_url(
+                    value,
+                    visited_urls=state.visited_urls,
+                    current_url=current_url,
+                    blocked=is_blocked,
+                )
+                
+                if new_url not in state.visited_urls:
+                    desktop.commands.run(f"google-chrome {shlex.quote(new_url)} &", background=True)
+                    desktop.wait(3000)
+                    state.visited_urls.append(new_url)
+            
+            elif action == "NAVIGATE":
+                if value and value.startswith("http"):
+                    if value in state.visited_urls:
+                        yield {"type": "status", "message": f"⏭️ Already visited, skipping..."}
+                        state.add_error(f"Tried to revisit: {value}")
+                    else:
+                        desktop.commands.run(f"google-chrome {shlex.quote(value)} &", background=True)
+                        desktop.wait(3000)
+                        state.visited_urls.append(value)
+            
+            # Small delay
+            desktop.wait(1000)
+        
+        # Timeout - generate from memory
+        yield {"type": "status", "message": "⏰ Time limit reached, generating final answer from memory..."}
+        
+        all_content = "\n\n".join([
+            f"Source: {d['url']}\n{d.get('preview', '')}" 
+            for d in state.extracted_data[-5:]
+        ])
+        known_summary = "\n".join([f"- {f}" for f in state.known_facts[-8:]]) or "(none)"
+        missing_summary = "\n".join([f"- {m}" for m in state.missing_points[-8:]]) or "(none)"
+        final_prompt = (
+            f"Based on this content, answer: {task}\n\n"
+            f"Known facts:\n{known_summary}\n\n"
+            f"Missing points:\n{missing_summary}\n\n"
+            f"Content:\n{all_content}"
+        )
+        final_answer = await generate_completion(
+            messages=[{"role": "user", "content": final_prompt}],
+            max_tokens=1000
+        )
+        
+        state.final_result = final_answer
+        
+        yield {"type": "stream_end", "message": "Done"}
+        yield {
+            "type": "result",
+            "content": final_answer,
+            "links": state.visited_urls,
+            "steps": state.step_count,
+            "success": True
+        }
+        yield {"type": "complete", "message": f"Completed in {MAX_TIME_SECONDS}s (timeout) with {state.step_count} steps"}
+        
+    except ImportError as e:
+        yield {"type": "error", "message": "e2b-desktop not installed"}
+    except Exception as e:
+        logger.exception("Browser agent error")
+        yield {"type": "error", "message": f"Error: {str(e)}"}
+    finally:
+        if desktop:
+            try:
+                desktop.stream.stop()
+            except:
+                pass
+            try:
+                desktop.kill()
+            except:
+                pass

app/agents/deep_research.py

@@ -0,0 +1,236 @@
+"""Deep Research Orchestrator.
+
+Coordinates the full deep research pipeline:
+1. Planning (query decomposition)
+2. Parallel searching (multiple dimensions)
+3. Report synthesis
+"""
+
+import asyncio
+import json
+import time
+from typing import AsyncIterator, Optional
+
+from app.agents.planner import create_research_plan, ResearchPlan, ResearchDimension
+from app.agents.llm_client import generate_completion_stream
+from app.reranking.pipeline import rerank_results
+from app.config import get_settings
+
+
+class DimensionResult:
+    """Results from researching a single dimension."""
+    
+    def __init__(self, dimension: ResearchDimension):
+        self.dimension = dimension
+        self.results: list[dict] = []
+        self.error: Optional[str] = None
+
+
+async def run_deep_research(
+    query: str,
+    max_dimensions: int = 6,
+    max_sources_per_dim: int = 5,
+    max_total_searches: int = 20,
+) -> AsyncIterator[str]:
+    """
+    Run a deep research pipeline with streaming progress.
+    
+    Yields SSE-formatted events as the research progresses.
+    
+    Args:
+        query: The research query
+        max_dimensions: Maximum dimensions to research
+        max_sources_per_dim: Max results per dimension
+        max_total_searches: Total Tavily API calls allowed
+        
+    Yields:
+        SSE event strings in format: data: {json}\n\n
+    """
+    start_time = time.perf_counter()
+    settings = get_settings()
+    
+    try:
+        # === PHASE 1: PLANNING ===
+        yield _sse_event("status", {"phase": "planning", "message": "Analyzing query..."})
+        
+        plan = await create_research_plan(query, max_dimensions)
+        
+        yield _sse_event("plan_ready", {
+            "refined_query": plan.refined_query,
+            "dimensions": [
+                {"name": d.name, "description": d.description, "priority": d.priority}
+                for d in plan.dimensions
+            ],
+            "estimated_sources": plan.estimated_sources,
+        })
+        
+        # === PHASE 2: PARALLEL SEARCHING ===
+        yield _sse_event("status", {"phase": "searching", "message": "Researching dimensions..."})
+        
+        # Distribute search budget across dimensions
+        num_dimensions = len(plan.dimensions)
+        searches_per_dim = max(1, max_total_searches // num_dimensions)
+        
+        dimension_results: list[DimensionResult] = []
+        
+        # Search dimensions in parallel batches
+        for i, dimension in enumerate(plan.dimensions):
+            yield _sse_event("dimension_start", {
+                "index": i + 1,
+                "total": num_dimensions,
+                "name": dimension.name,
+                "query": dimension.search_query,
+            })
+            
+            # Search this dimension
+            result = await _search_dimension(
+                dimension=dimension,
+                max_results=max_sources_per_dim,
+                max_searches=searches_per_dim,
+            )
+            dimension_results.append(result)
+            
+            yield _sse_event("dimension_complete", {
+                "index": i + 1,
+                "name": dimension.name,
+                "results_count": len(result.results),
+                "error": result.error,
+            })
+            
+            # Small delay to avoid rate limits
+            await asyncio.sleep(0.1)
+        
+        # === PHASE 3: SYNTHESIS ===
+        yield _sse_event("status", {"phase": "synthesizing", "message": "Generating report..."})
+        yield _sse_event("synthesis_start", {})
+        
+        # Stream the report generation
+        async for chunk in _synthesize_report_stream(query, plan, dimension_results):
+            yield _sse_event("report_chunk", {"content": chunk})
+        
+        # === COMPLETE ===
+        total_time = time.perf_counter() - start_time
+        total_sources = sum(len(r.results) for r in dimension_results)
+        
+        yield _sse_event("done", {
+            "total_sources": total_sources,
+            "total_dimensions": num_dimensions,
+            "total_time_seconds": round(total_time, 2),
+        })
+        
+    except Exception as e:
+        yield _sse_event("error", {"message": str(e)})
+
+
+async def _search_dimension(
+    dimension: ResearchDimension,
+    max_results: int = 5,
+    max_searches: int = 2,
+) -> DimensionResult:
+    """Search a single dimension using the aggregator."""
+    from app.sources.aggregator import aggregate_search
+    
+    result = DimensionResult(dimension)
+    
+    try:
+        # Use aggregator to search all sources
+        all_results = await aggregate_search(
+            query=dimension.search_query,
+            max_results=max_results + 3,  # Get extra for reranking
+            include_wikipedia=True,
+        )
+        
+        # Light reranking (use embeddings when we have many results from SearXNG)
+        if all_results:
+            use_embeddings = len(all_results) > 15
+            ranked = await rerank_results(
+                query=dimension.search_query,
+                results=all_results,
+                temporal_urgency=0.5,
+                max_results=max_results,
+                use_embeddings=use_embeddings,
+            )
+            result.results = ranked
+        
+    except Exception as e:
+        result.error = str(e)
+    
+    return result
+
+
+async def _synthesize_report_stream(
+    original_query: str,
+    plan: ResearchPlan,
+    dimension_results: list[DimensionResult],
+) -> AsyncIterator[str]:
+    """Stream the synthesis of the final report."""
+    
+    # Build context from all dimension results
+    context_parts = []
+    all_sources = []
+    source_index = 1
+    
+    for dr in dimension_results:
+        if dr.results:
+            context_parts.append(f"\n## {dr.dimension.name}\n")
+            for r in dr.results:
+                context_parts.append(
+                    f"[{source_index}] {r.get('title', 'Untitled')}\n"
+                    f"   URL: {r.get('url', '')}\n"
+                    f"   Content: {r.get('content', '')[:400]}...\n"
+                )
+                all_sources.append({
+                    "index": source_index,
+                    "title": r.get("title", ""),
+                    "url": r.get("url", ""),
+                })
+                source_index += 1
+    
+    context = "\n".join(context_parts)
+    
+    # Build synthesis prompt
+    prompt = f"""You are a research analyst. Create a comprehensive research report based on the gathered information.
+
+ORIGINAL QUERY: {original_query}
+REFINED QUERY: {plan.refined_query}
+
+RESEARCH DIMENSIONS:
+{', '.join(d.name for d in plan.dimensions)}
+
+GATHERED INFORMATION:
+{context}
+
+INSTRUCTIONS:
+1. Write a comprehensive research report in Markdown format
+2. Start with an Executive Summary (2-3 paragraphs)
+3. Create a section for each research dimension
+4. Use citations [1], [2], etc. to reference sources
+5. Include a Conclusion section
+6. Be thorough but concise
+7. Write in the same language as the query
+8. Use headers (##) to organize sections
+
+Generate the report:"""
+
+    messages = [
+        {"role": "system", "content": "You are a research analyst creating detailed reports."},
+        {"role": "user", "content": prompt},
+    ]
+    
+    try:
+        async for chunk in generate_completion_stream(messages, temperature=0.4):
+            yield chunk
+        
+        # Append sources at the end
+        yield "\n\n---\n\n## Sources\n\n"
+        for src in all_sources:
+            yield f"[{src['index']}] [{src['title']}]({src['url']})\n"
+            
+    except Exception as e:
+        yield f"\n\n**Error generating report:** {e}"
+
+
+def _sse_event(event_type: str, data: dict) -> str:
+    """Format an SSE event."""
+    payload = {"type": event_type, **data}
+    return f"data: {json.dumps(payload)}\n\n"

app/agents/flaresolverr.py

@@ -0,0 +1,128 @@
+"""FlareSolverr client for Cloudflare bypass.
+
+FlareSolverr uses undetected-chromedriver to solve Cloudflare challenges.
+Must be running at http://localhost:8191 in the E2B sandbox.
+"""
+
+import logging
+import json
+import shlex
+from typing import Optional, Tuple
+
+logger = logging.getLogger(__name__)
+
+FLARESOLVERR_URL = "http://localhost:8191/v1"
+
+
+async def solve_cloudflare(desktop, url: str, timeout: int = 60) -> Tuple[bool, str]:
+    """
+    Use FlareSolverr to bypass Cloudflare protection.
+    
+    Args:
+        desktop: E2B desktop instance
+        url: URL to fetch through FlareSolverr
+        timeout: Max seconds to wait for solution
+        
+    Returns:
+        (success: bool, content: str)
+    """
+    try:
+        # Make request to FlareSolverr - properly escape the JSON payload
+        payload = json.dumps({
+            "cmd": "request.get",
+            "url": url,
+            "maxTimeout": timeout * 1000
+        })
+        
+        result = desktop.commands.run(
+            f"curl -s -X POST {shlex.quote(FLARESOLVERR_URL)} "
+            f"-H 'Content-Type: application/json' "
+            f"-d {shlex.quote(payload)} 2>/dev/null",
+            timeout=timeout + 10
+        )
+        
+        if not hasattr(result, 'stdout') or not result.stdout:
+            return False, ""
+        
+        response = json.loads(result.stdout)
+        
+        if response.get("status") == "ok":
+            solution = response.get("solution", {})
+            html = solution.get("response", "")
+            
+            # Strip HTML tags - use base64 to safely pass content
+            if html:
+                import base64
+                html_b64 = base64.b64encode(html[:10000].encode()).decode()
+                clean_result = desktop.commands.run(
+                    f"echo {shlex.quote(html_b64)} | base64 -d | sed 's/<[^>]*>//g' | tr -s ' \\n' ' ' | head -c 6000",
+                    timeout=5
+                )
+                content = clean_result.stdout.strip() if hasattr(clean_result, 'stdout') else html[:6000]
+                logger.info(f"FlareSolverr solved: {url[:50]}")
+                return True, content
+        
+        logger.warning(f"FlareSolverr failed: {response.get('message', 'unknown')}")
+        return False, ""
+        
+    except Exception as e:
+        logger.warning(f"FlareSolverr error: {e}")
+        return False, ""
+
+
+def is_cloudflare_blocked(content: str) -> bool:
+    """Check if page content indicates Cloudflare block.
+    
+    Only returns True for actual Cloudflare challenge pages,
+    not just pages that mention Cloudflare.
+    """
+    content_lower = content.lower()
+    
+    # Must have multiple strong indicators to be considered blocked
+    strong_indicators = [
+        "checking your browser before accessing",
+        "please wait while we verify",
+        "ray id:",
+        "cloudflare ray id",
+        "enable javascript and cookies",
+        "attention required! | cloudflare",
+        "just a moment...",
+        "ddos protection by cloudflare",
+    ]
+    
+    # Check for strong indicators (need at least 1)
+    has_strong = any(ind in content_lower for ind in strong_indicators)
+    
+    # Also check if content is suspiciously short (challenge pages are small)
+    is_short = len(content) < 500
+    
+    # Only block if we have strong indicator AND page is short
+    # (real content pages that mention cloudflare will be longer)
+    if has_strong and is_short:
+        return True
+    
+    # Very specific patterns that are definitely challenge pages
+    definite_blocks = [
+        "checking if the site connection is secure",
+        "please turn javascript on and reload the page",
+        "please enable cookies",
+    ]
+    
+    return any(block in content_lower for block in definite_blocks)
+
+
+def is_login_wall(content: str) -> bool:
+    """Check if page requires login."""
+    login_indicators = [
+        "sign in",
+        "log in",
+        "login",
+        "create account",
+        "register",
+        "enter your password",
+        "authentication required",
+    ]
+    
+    content_lower = content.lower()
+    # Check for login indicators but make sure it's not just a login link
+    return sum(1 for ind in login_indicators if ind in content_lower) >= 2

app/agents/graph/__init__.py

@@ -0,0 +1 @@
+# Agent Graph Package

app/agents/graph/nodes.py

@@ -0,0 +1,338 @@
+"""Graph nodes for the agent execution.
+
+Each node represents a step in the agent's decision process:
+- PlanNode: Decomposes the task into subtasks
+- SearchNode: Performs web searches
+- NavigateNode: Navigates to URLs
+- ExtractNode: Extracts content from pages
+- VerifyNode: Verifies if goal is achieved
+- RespondNode: Generates final response
+"""
+
+import json
+import logging
+import shlex
+import base64
+from abc import ABC, abstractmethod
+from typing import Tuple
+
+from app.agents.graph.state import AgentState, NodeType
+from app.agents.llm_client import generate_completion
+
+logger = logging.getLogger(__name__)
+
+
+class BaseNode(ABC):
+    """Base class for all graph nodes."""
+    
+    node_type: NodeType = NodeType.START
+    
+    @abstractmethod
+    async def execute(self, state: AgentState) -> Tuple[AgentState, NodeType]:
+        """Execute the node logic and return updated state + next node."""
+        pass
+
+
+class PlanNode(BaseNode):
+    """Decomposes task into subtasks."""
+    
+    node_type = NodeType.PLAN
+    
+    async def execute(self, state: AgentState) -> Tuple[AgentState, NodeType]:
+        prompt = f"""Você é um planejador de tarefas. Decomponha a tarefa em passos simples.
+
+TAREFA: {state.task}
+URL inicial: {state.url or 'Nenhuma - começar com busca'}
+
+Responda com JSON:
+{{
+    "goal": "objetivo principal",
+    "steps": [
+        {{"action": "search", "query": "termos de busca"}},
+        {{"action": "navigate", "description": "onde navegar"}},
+        {{"action": "extract", "what": "o que extrair"}}
+    ],
+    "success_criteria": "critério de sucesso"
+}}
+
+Responda APENAS o JSON, sem explicação."""
+
+        try:
+            response = await generate_completion(
+                messages=[{"role": "user", "content": prompt}],
+                max_tokens=500
+            )
+            
+            # Parse JSON
+            response = response.strip()
+            if response.startswith("```"):
+                response = response.split("```")[1]
+                if response.startswith("json"):
+                    response = response[4:]
+            
+            plan = json.loads(response)
+            state.plan = plan
+            logger.info(f"Plan created: {plan.get('goal', 'No goal')}")
+            
+            # Decide next node based on plan
+            if plan.get("steps") and plan["steps"][0].get("action") == "navigate" and state.url:
+                return state, NodeType.NAVIGATE
+            return state, NodeType.SEARCH
+            
+        except Exception as e:
+            logger.error(f"Planning failed: {e}")
+            state.add_error(f"Planning failed: {e}")
+            # Fallback to search
+            state.plan = {"goal": state.task, "steps": [{"action": "search", "query": state.task}]}
+            return state, NodeType.SEARCH
+
+
+class SearchNode(BaseNode):
+    """Performs web search."""
+    
+    node_type = NodeType.SEARCH
+    
+    async def execute(self, state: AgentState) -> Tuple[AgentState, NodeType]:
+        desktop = state.desktop
+        
+        # Determine search query
+        query = state.task
+        if state.plan.get("steps"):
+            for step in state.plan["steps"]:
+                if step.get("action") == "search" and step.get("query"):
+                    query = step["query"]
+                    break
+        
+        # Execute search
+        search_url = f"https://html.duckduckgo.com/html/?q={query.replace(' ', '+')}"
+        
+        try:
+            desktop.commands.run(f"google-chrome {shlex.quote(search_url)} &", background=True)
+            state.visited_urls.append(search_url)
+            desktop.wait(3000)
+            
+            state.add_action({"type": "search", "query": query})
+            logger.info(f"Searched: {query}")
+            
+            return state, NodeType.EXTRACT
+            
+        except Exception as e:
+            state.add_error(f"Search failed: {e}")
+            return state, NodeType.VERIFY
+
+
+class NavigateNode(BaseNode):
+    """Navigates to a URL."""
+    
+    node_type = NodeType.NAVIGATE
+    
+    async def execute(self, state: AgentState) -> Tuple[AgentState, NodeType]:
+        desktop = state.desktop
+        
+        # Get URL to navigate
+        url = state.url
+        if not url and state.extracted_data:
+            # Try to get URL from extracted links
+            last_data = state.extracted_data[-1]
+            if "links" in last_data.get("data", {}):
+                links = last_data["data"]["links"]
+                if links:
+                    url = links[0]
+        
+        if not url:
+            return state, NodeType.SEARCH
+        
+        try:
+            desktop.commands.run(f"google-chrome {shlex.quote(url)} &", background=True)
+            if url not in state.visited_urls:
+                state.visited_urls.append(url)
+            desktop.wait(3000)
+            
+            state.add_action({"type": "navigate", "url": url})
+            logger.info(f"Navigated to: {url[:50]}")
+            
+            return state, NodeType.EXTRACT
+            
+        except Exception as e:
+            state.add_error(f"Navigation failed: {e}")
+            return state, NodeType.SEARCH
+
+
+class ExtractNode(BaseNode):
+    """Extracts content from current page."""
+    
+    node_type = NodeType.EXTRACT
+    
+    async def execute(self, state: AgentState) -> Tuple[AgentState, NodeType]:
+        desktop = state.desktop
+        current_url = state.visited_urls[-1] if state.visited_urls else ""
+        
+        try:
+            # Get window title
+            result = desktop.commands.run("xdotool getactivewindow getwindowname 2>/dev/null", timeout=5)
+            state.window_title = result.stdout.strip() if hasattr(result, 'stdout') else ""
+            
+            # Extract page content via curl
+            if current_url.startswith("http"):
+                result = desktop.commands.run(
+                    f"curl -sL --max-time 10 --connect-timeout 5 "
+                    f"-A 'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36' "
+                    f"'{current_url}' 2>/dev/null | "
+                    "sed -e 's/<script[^>]*>.*<\\/script>//g' -e 's/<style[^>]*>.*<\\/style>//g' | "
+                    "sed 's/<[^>]*>//g' | "
+                    "tr -s ' \\n' ' ' | "
+                    "head -c 6000",
+                    timeout=15
+                )
+                state.page_content = result.stdout.strip() if hasattr(result, 'stdout') else ""
+            
+            state.add_action({"type": "extract", "content_length": len(state.page_content)})
+            logger.info(f"Extracted {len(state.page_content)} chars from {current_url[:50]}")
+            
+            return state, NodeType.VERIFY
+            
+        except Exception as e:
+            state.add_error(f"Extraction failed: {e}")
+            return state, NodeType.VERIFY
+
+
+class VerifyNode(BaseNode):
+    """Verifies if goal is achieved and decides next action."""
+    
+    node_type = NodeType.VERIFY
+    
+    async def execute(self, state: AgentState) -> Tuple[AgentState, NodeType]:
+        context = state.get_context_for_llm()
+        page_preview = state.page_content[:4000] if state.page_content else "(No content)"
+        
+        prompt = f"""Você é um agente de navegação web. Analise o conteúdo e decida o próximo passo.
+
+TAREFA: {state.task}
+PLANO: {state.plan.get('goal', 'Nenhum')}
+CRITÉRIO DE SUCESSO: {state.plan.get('success_criteria', 'Encontrar a informação pedida')}
+
+HISTÓRICO:
+{context}
+
+CONTEÚDO DA PÁGINA ATUAL:
+{page_preview}
+
+TEMPO RESTANTE: {int(state.get_remaining_time())}s
+
+Decida:
+1. Se encontrou a resposta, retorne: {{"status": "complete", "result": "Sua resposta formatada com **negrito** para valores importantes"}}
+2. Se precisa buscar mais, retorne: {{"action": "search", "query": "nova busca"}}
+3. Se precisa navegar para um link, retorne: {{"action": "navigate", "url": "https://..."}}
+4. Se precisa rolar a página, retorne: {{"action": "scroll"}}
+
+REGRAS:
+- Use **negrito** para preços e valores importantes
+- Cite as fontes
+- Se página pede login, tente outra fonte
+- Seja eficiente
+
+Responda APENAS com JSON válido."""
+
+        try:
+            response = await generate_completion(
+                messages=[{"role": "user", "content": prompt}],
+                max_tokens=800
+            )
+            
+            # Parse response
+            response = response.strip()
+            if response.startswith("```"):
+                response = response.split("```")[1]
+                if response.startswith("json"):
+                    response = response[4:]
+            
+            decision = json.loads(response)
+            state.add_action({"type": "verify", "decision": decision})
+            
+            # Route based on decision
+            if decision.get("status") == "complete":
+                state.final_result = decision.get("result", "")
+                state.success = True
+                logger.info("Goal achieved!")
+                return state, NodeType.RESPOND
+            
+            action = decision.get("action", "")
+            if action == "search":
+                # Update plan with new search
+                state.plan["steps"] = [{"action": "search", "query": decision.get("query", state.task)}]
+                return state, NodeType.SEARCH
+            elif action == "navigate":
+                state.url = decision.get("url", "")
+                return state, NodeType.NAVIGATE
+            elif action == "scroll":
+                state.desktop.scroll(-3)
+                state.desktop.wait(1000)
+                return state, NodeType.EXTRACT
+            
+            # Default: try another search
+            return state, NodeType.SEARCH
+            
+        except Exception as e:
+            logger.error(f"Verify failed: {e}")
+            state.add_error(f"Verify failed: {e}")
+            
+            # If we have some content, try to respond anyway
+            if state.get_remaining_time() < 30:
+                return state, NodeType.RESPOND
+            return state, NodeType.SEARCH
+
+
+class RespondNode(BaseNode):
+    """Generates final response."""
+    
+    node_type = NodeType.RESPOND
+    
+    async def execute(self, state: AgentState) -> Tuple[AgentState, NodeType]:
+        # If we already have a result, we're done
+        if state.final_result:
+            state.success = True
+            return state, NodeType.RESPOND
+        
+        # Generate response from collected data
+        context = state.get_context_for_llm()
+        page_content = state.page_content[:3000] if state.page_content else "(Nenhum conteúdo extraído)"
+        
+        prompt = f"""Você realizou uma tarefa de navegação web. Sintetize os resultados.
+
+TAREFA: {state.task}
+
+DADOS COLETADOS:
+{context}
+
+ÚLTIMO CONTEÚDO DA PÁGINA:
+{page_content}
+
+URLs VISITADAS:
+{chr(10).join(state.visited_urls[:5]) if state.visited_urls else '(Nenhuma)'}
+
+INSTRUÇÕES:
+- Gere uma resposta útil baseada no que foi encontrado
+- Use **negrito** para valores importantes (preços, números, nomes)
+- Cite as fontes quando possível
+- Se não encontrou o que foi pedido, explique o que encontrou ou diga honestamente que não encontrou
+
+Responda em português de forma clara e organizada."""
+
+        try:
+            response = await generate_completion(
+                messages=[{"role": "user", "content": prompt}],
+                max_tokens=1000
+            )
+            state.final_result = response.strip()
+            state.success = bool(state.final_result)
+            logger.info(f"Generated response: {len(state.final_result)} chars")
+            
+        except Exception as e:
+            logger.error(f"Response generation failed: {e}")
+            # Fallback: create response from available data
+            if state.page_content:
+                state.final_result = f"**Informação encontrada:**\n\n{state.page_content[:500]}...\n\n*Fonte: {state.visited_urls[-1] if state.visited_urls else 'desconhecida'}*"
+            else:
+                state.final_result = f"Não foi possível completar a tarefa. Erro: {e}"
+        
+        return state, NodeType.RESPOND

app/agents/graph/runner.py

@@ -0,0 +1,133 @@
+"""Graph runner - executes the agent graph.
+
+The runner orchestrates node execution, manages state transitions,
+and yields status updates for streaming.
+
+Uses timeout-based execution instead of fixed iteration count.
+"""
+
+import logging
+import time
+from typing import AsyncGenerator, Dict, Type
+
+from app.agents.graph.state import AgentState, NodeType
+from app.agents.graph.nodes import (
+    BaseNode,
+    PlanNode,
+    SearchNode,
+    NavigateNode,
+    ExtractNode,
+    VerifyNode,
+    RespondNode,
+)
+
+logger = logging.getLogger(__name__)
+
+# Node registry
+NODE_REGISTRY: Dict[NodeType, Type[BaseNode]] = {
+    NodeType.PLAN: PlanNode,
+    NodeType.SEARCH: SearchNode,
+    NodeType.NAVIGATE: NavigateNode,
+    NodeType.EXTRACT: ExtractNode,
+    NodeType.VERIFY: VerifyNode,
+    NodeType.RESPOND: RespondNode,
+}
+
+# Status messages with emojis
+STATUS_MESSAGES = {
+    NodeType.PLAN: "🎯 Planning task...",
+    NodeType.SEARCH: "🔍 Searching...",
+    NodeType.NAVIGATE: "🌐 Navigating...",
+    NodeType.EXTRACT: "📊 Extracting content...",
+    NodeType.VERIFY: "🤔 Analyzing...",
+    NodeType.RESPOND: "✅ Generating response...",
+}
+
+
+async def run_graph(state: AgentState) -> AsyncGenerator[dict, None]:
+    """Run the agent graph and yield status updates.
+    
+    Args:
+        state: Initial agent state with task, url, and desktop
+        
+    Yields:
+        Status updates and final result
+    """
+    # Initialize timing
+    state.start_time = time.time()
+    current_node_type = NodeType.PLAN
+    state.current_node = current_node_type
+    
+    logger.info(f"Starting graph execution for task: {state.task[:50]}, timeout: {state.timeout_seconds}s")
+    
+    while state.should_continue():
+        state.step_count += 1
+        state.current_node = current_node_type
+        
+        # Get node instance
+        node_class = NODE_REGISTRY.get(current_node_type)
+        if not node_class:
+            logger.error(f"Unknown node type: {current_node_type}")
+            break
+        
+        node = node_class()
+        
+        # Calculate remaining time
+        remaining = int(state.get_remaining_time())
+        elapsed = int(state.get_elapsed_time())
+        
+        # Yield status update
+        status_msg = STATUS_MESSAGES.get(current_node_type, "Processing...")
+        if current_node_type == NodeType.SEARCH and state.plan.get("steps"):
+            for step in state.plan["steps"]:
+                if step.get("action") == "search":
+                    status_msg = f"🔍 Searching: {step.get('query', state.task)[:40]}..."
+                    break
+        elif current_node_type == NodeType.NAVIGATE and state.url:
+            status_msg = f"🌐 Navigating to {state.url[:40]}..."
+        
+        yield {
+            "type": "status",
+            "message": f"{status_msg} (step {state.step_count}, {remaining}s remaining)"
+        }
+        
+        # Execute node
+        try:
+            state, next_node_type = await node.execute(state)
+            logger.info(f"Step {state.step_count}: {current_node_type.value} -> {next_node_type.value} ({elapsed}s elapsed)")
+            
+            # Check if we're done
+            if current_node_type == NodeType.RESPOND:
+                break
+            
+            # Transition to next node
+            current_node_type = next_node_type
+            
+        except Exception as e:
+            logger.exception(f"Node execution failed: {e}")
+            state.add_error(str(e))
+            
+            # If running low on time, try to respond
+            if state.get_remaining_time() < 30:
+                current_node_type = NodeType.RESPOND
+            else:
+                current_node_type = NodeType.SEARCH
+    
+    # If we timed out without a result, generate one from what we have
+    if not state.final_result and not state.success:
+        logger.warning("Timeout reached, forcing response generation")
+        respond_node = RespondNode()
+        state, _ = await respond_node.execute(state)
+    
+    # Yield final result
+    yield {
+        "type": "result",
+        "content": state.final_result,
+        "links": state.visited_urls[:10],
+        "success": state.success
+    }
+    
+    yield {"type": "complete", "message": f"Task completed in {int(state.get_elapsed_time())}s"}
+    
+    logger.info(f"Graph execution complete. Success: {state.success}, Steps: {state.step_count}, Time: {state.get_elapsed_time():.1f}s")
+

app/agents/graph/simple_agent.py

@@ -0,0 +1,321 @@
+"""Simplified agent nodes - ONE LLM call per cycle.
+
+DAG:
+  START → THINK_ACT ←→ EXECUTE → RESPOND
+              ↑______________|
+
+ThinkAndAct: Analyzes content + decides action in ONE call
+Execute: Runs the action (search, navigate, scroll) - NO LLM
+Respond: Final synthesis
+"""
+
+import json
+import logging
+import shlex
+import time
+from abc import ABC, abstractmethod
+from typing import Tuple, Optional, List
+
+from app.agents.llm_client import generate_completion
+
+logger = logging.getLogger(__name__)
+
+
+class SimpleState:
+    """Minimal state for the agent."""
+    
+    def __init__(self, task: str, url: Optional[str], desktop, timeout: float = 300):
+        self.task = task
+        self.url = url
+        self.desktop = desktop
+        self.timeout = timeout
+        self.start_time = time.time()
+        
+        # Memory - content cache (URL -> content)
+        self.content_cache: dict = {}  # {url: content}
+        self.visited_urls: List[str] = []
+        self.action_history: List[str] = []
+        
+        # Accumulated knowledge
+        self.findings: List[str] = []  # Key findings extracted
+        
+        # Result
+        self.final_result = ""
+        self.done = False
+    
+    def elapsed(self) -> float:
+        return time.time() - self.start_time
+    
+    def remaining(self) -> float:
+        return max(0, self.timeout - self.elapsed())
+    
+    def should_continue(self) -> bool:
+        return not self.done and self.remaining() > 20
+    
+    def add_page(self, url: str, content: str):
+        """Add page to cache - no duplicate fetching."""
+        if url not in self.content_cache:
+            self.content_cache[url] = content[:4000]
+        if url not in self.visited_urls:
+            self.visited_urls.append(url)
+    
+    def get_cached_content(self, url: str) -> Optional[str]:
+        """Get content from cache if available."""
+        return self.content_cache.get(url)
+    
+    def add_finding(self, finding: str):
+        """Add a key finding to memory."""
+        if finding and finding not in self.findings:
+            self.findings.append(finding)
+    
+    def get_all_content(self) -> str:
+        """Get all cached content for final synthesis."""
+        parts = []
+        for url in self.visited_urls[-5:]:
+            content = self.content_cache.get(url, "")
+            if content:
+                parts.append(f"[{url[:60]}]\n{content[:1500]}")
+        return "\n\n---\n\n".join(parts)
+    
+    def get_recent_content(self) -> str:
+        """Get last 2 pages content for context."""
+        recent_urls = self.visited_urls[-2:] if self.visited_urls else []
+        parts = []
+        for url in recent_urls:
+            content = self.content_cache.get(url, "")
+            if content:
+                parts.append(f"[{url[:60]}]\n{content[:2000]}")
+        return "\n\n---\n\n".join(parts)
+
+
+async def think_and_act(state: SimpleState) -> Tuple[str, dict]:
+    """
+    ONE LLM call that analyzes current state and decides next action.
+    Returns: (action_type, action_params)
+    
+    Actions:
+    - search: {"query": "..."}
+    - navigate: {"url": "..."}
+    - scroll: {}
+    - complete: {"result": "..."}
+    """
+    
+    content = state.get_recent_content() or "(No content yet)"
+    history = ", ".join(state.action_history[-5:]) if state.action_history else "(starting)"
+    
+    # Memory: show visited URLs so LLM doesn't repeat
+    visited = "\n".join([f"  - {u[:70]}" for u in state.visited_urls[-10:]]) if state.visited_urls else "(none)"
+    
+    prompt = f"""You are a web research agent. Analyze the current state and decide your next action.
+
+TASK: {state.task}
+
+ALREADY VISITED (DO NOT visit again):
+{visited}
+
+CURRENT PAGE CONTENT:
+{content}
+
+HISTORY: {history}
+TIME REMAINING: {int(state.remaining())}s
+
+Decide ONE action. Return JSON:
+
+If you need to search: {{"action": "search", "query": "search terms"}}
+If you found a NEW relevant link to visit: {{"action": "navigate", "url": "https://..."}}
+If you need to scroll for more content: {{"action": "scroll"}}
+If you have enough info to answer: {{"action": "complete", "result": "Your answer with **bold** for important values. Cite sources."}}
+
+RULES:
+- DO NOT navigate to URLs already in "ALREADY VISITED" list
+- Only use URLs you see in the content above
+- If you see the answer, return complete immediately
+- Use **bold** for prices, numbers, names
+- Be efficient - don't repeat searches
+
+Return ONLY valid JSON:"""
+
+    try:
+        response = await generate_completion(
+            messages=[{"role": "user", "content": prompt}],
+            max_tokens=800
+        )
+        
+        # Parse JSON
+        response = response.strip()
+        if response.startswith("```"):
+            response = response.split("```")[1]
+            if response.startswith("json"):
+                response = response[4:]
+        
+        decision = json.loads(response)
+        action = decision.get("action", "search")
+        
+        # Safety check: prevent navigating to already visited URL
+        if action == "navigate":
+            url = decision.get("url", "").rstrip("/")
+            
+            # Check if URL already visited (normalize by removing trailing slash)
+            visited_normalized = [u.rstrip("/") for u in state.visited_urls]
+            if url in visited_normalized or url in state.visited_urls:
+                logger.warning(f"LLM tried to revisit {url}, trying different approach")
+                
+                # If we have good content, finish
+                good_content = [c for c in state.content_cache.values() 
+                               if c and c not in ["[BLOCKED]", "[LOGIN_REQUIRED]"]]
+                if good_content:
+                    return "complete", {"result": f"Informação coletada: {state.get_recent_content()[:800]}"}
+                
+                # Otherwise, search with different terms
+                return "search", {"query": f"{state.task} site:wikipedia.org OR site:gov.br"}
+        
+        logger.info(f"ThinkAndAct decision: {action}")
+        return action, decision
+        
+    except Exception as e:
+        logger.error(f"ThinkAndAct failed: {e}")
+        # Fallback: if we have content, try to respond
+        if state.content_cache:
+            return "complete", {"result": f"Based on collected data: {state.get_recent_content()[:500]}"}
+        return "search", {"query": state.task}
+
+
+async def execute_action(state: SimpleState, action: str, params: dict) -> bool:
+    """
+    Execute action WITHOUT LLM call.
+    Uses cache to avoid repeated requests.
+    Returns True if should continue, False if done.
+    """
+    desktop = state.desktop
+    
+    if action == "complete":
+        state.final_result = params.get("result", "")
+        state.done = True
+        return False
+    
+    elif action == "search":
+        query = params.get("query", state.task)
+        search_url = f"https://html.duckduckgo.com/html/?q={query.replace(' ', '+')}"
+        
+        # Check cache first
+        cached = state.get_cached_content(search_url)
+        if cached:
+            logger.info(f"Using cached content for search: {query[:30]}")
+            state.action_history.append(f"search(cached):{query[:30]}")
+            return True
+        
+        desktop.commands.run(f"google-chrome {shlex.quote(search_url)} &", background=True)
+        desktop.wait(3000)
+        
+        content = await _extract_content(desktop, search_url)
+        state.add_page(search_url, content)
+        state.action_history.append(f"search:{query[:30]}")
+        
+        return True
+    
+    elif action == "navigate":
+        url = params.get("url", "")
+        if not url.startswith("http"):
+            return True  # Invalid URL, continue
+        
+        # Check cache first - don't re-fetch
+        cached = state.get_cached_content(url)
+        if cached:
+            logger.info(f"Using cached content for: {url[:50]}")
+            state.action_history.append(f"nav(cached):{url[:30]}")
+            return True
+        
+        desktop.commands.run(f"google-chrome {shlex.quote(url)} &", background=True)
+        desktop.wait(3000)
+        
+        content = await _extract_content(desktop, url)
+        
+        # Check for Cloudflare/bot detection - just skip if blocked
+        from app.agents.flaresolverr import is_cloudflare_blocked, is_login_wall
+        
+        if is_cloudflare_blocked(content):
+            logger.warning(f"Cloudflare block detected at {url[:50]}, skipping...")
+            # Mark as visited so LLM doesn't try again
+            if url not in state.visited_urls:
+                state.visited_urls.append(url)
+            state.content_cache[url] = "[BLOCKED]"  # Mark as blocked in cache
+            state.action_history.append(f"nav(blocked):{url[:30]}")
+            return True
+        
+        if is_login_wall(content):
+            logger.warning(f"Login wall detected at {url[:50]}, skipping...")
+            # Mark as visited so LLM doesn't try again
+            if url not in state.visited_urls:
+                state.visited_urls.append(url)
+            state.content_cache[url] = "[LOGIN_REQUIRED]"  # Mark in cache
+            state.action_history.append(f"nav(login_wall):{url[:30]}")
+            return True
+        
+        state.add_page(url, content)
+        state.action_history.append(f"nav:{url[:30]}")
+        
+        return True
+    
+    elif action == "scroll":
+        desktop.scroll(-3)
+        desktop.wait(1500)
+        
+        # Update cache for current page with new content
+        if state.visited_urls:
+            current_url = state.visited_urls[-1]
+            content = await _extract_content(desktop, current_url)
+            state.content_cache[current_url] = content[:4000]  # Update cache
+        
+        state.action_history.append("scroll")
+        return True
+    
+    return True
+
+
+async def _extract_content(desktop, url: str) -> str:
+    """Extract page content via curl."""
+    try:
+        result = desktop.commands.run(
+            f"curl -sL --max-time 8 --connect-timeout 5 "
+            f"-A 'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36' "
+            f"'{url}' 2>/dev/null | "
+            "sed -e 's/<script[^>]*>.*<\\/script>//g' -e 's/<style[^>]*>.*<\\/style>//g' | "
+            "sed 's/<[^>]*>//g' | "
+            "tr -s ' \\n' ' ' | "
+            "head -c 6000",
+            timeout=12
+        )
+        return result.stdout.strip() if hasattr(result, 'stdout') else ""
+    except Exception as e:
+        logger.warning(f"Extract failed: {e}")
+        return ""
+
+
+async def generate_final_response(state: SimpleState) -> str:
+    """Generate response if agent timed out without completing."""
+    if state.final_result:
+        return state.final_result
+    
+    content = state.get_recent_content()
+    
+    prompt = f"""Based on the research done, answer the question.
+
+TASK: {state.task}
+
+COLLECTED DATA:
+{content if content else "(No data collected)"}
+
+SOURCES VISITED: {', '.join(state.visited_urls[:5]) if state.visited_urls else 'None'}
+
+Provide a helpful answer based on what was found. Use **bold** for important values. If you couldn't find the answer, say so honestly.
+
+Answer in Portuguese:"""
+
+    try:
+        response = await generate_completion(
+            messages=[{"role": "user", "content": prompt}],
+            max_tokens=1000
+        )
+        return response.strip()
+    except Exception as e:
+        return f"Não foi possível completar a pesquisa. Erro: {e}"

app/agents/graph/state.py

@@ -0,0 +1,187 @@
+"""Agent state management for graph-based execution.
+
+The state is passed between nodes and accumulates information
+throughout the agent's execution.
+"""
+
+from dataclasses import dataclass, field
+from typing import Optional, Any
+from enum import Enum
+
+
+class NodeType(Enum):
+    """Types of nodes in the agent graph."""
+    START = "start"
+    PLAN = "plan"
+    SEARCH = "search"
+    NAVIGATE = "navigate"
+    EXTRACT = "extract"
+    VERIFY = "verify"
+    RESPOND = "respond"
+    ERROR = "error"
+
+
+@dataclass
+class AgentState:
+    """Shared state passed between graph nodes."""
+    
+    # Task info
+    task: str = ""
+    url: Optional[str] = None
+    
+    # Planning
+    plan: dict = field(default_factory=dict)
+    current_subtask: int = 0
+    
+    # Execution
+    current_node: NodeType = NodeType.START
+    step_count: int = 0
+    start_time: float = field(default_factory=lambda: 0.0)
+    timeout_seconds: float = 300.0  # 5 minutes default
+    
+    # Memory
+    visited_urls: list = field(default_factory=list)
+    extracted_data: list = field(default_factory=list)
+    page_content: str = ""
+    window_title: str = ""
+    known_facts: list = field(default_factory=list)
+    missing_points: list = field(default_factory=list)
+    last_queries: list = field(default_factory=list)
+    
+    # History
+    action_history: list = field(default_factory=list)
+    error_history: list = field(default_factory=list)
+    
+    # Results
+    final_result: str = ""
+    success: bool = False
+    
+    # Desktop reference (set at runtime)
+    desktop: Any = None
+    
+    def add_action(self, action: dict):
+        """Add action to history."""
+        self.action_history.append({
+            "step": self.step_count,
+            "node": self.current_node.value,
+            "action": action
+        })
+    
+    def add_error(self, error: str):
+        """Add error to history."""
+        self.error_history.append({
+            "step": self.step_count,
+            "error": error
+        })
+    
+    def add_extracted_data(self, source: str, data: dict):
+        """Add extracted data from a source."""
+        self.extracted_data.append({
+            "source": source,
+            "url": self.visited_urls[-1] if self.visited_urls else "",
+            "data": data
+        })
+
+    def add_query(self, query: str):
+        """Track recent search queries used by the agent."""
+        query = (query or "").strip()
+        if not query:
+            return
+        if query not in self.last_queries:
+            self.last_queries.append(query)
+        self.last_queries = self.last_queries[-8:]
+
+    def update_research_progress(self, known_facts: list | None = None, missing_points: list | None = None):
+        """Update short-term research memory from LLM output."""
+        if isinstance(known_facts, str):
+            known_facts = [known_facts]
+        if isinstance(missing_points, str):
+            missing_points = [missing_points]
+
+        if known_facts:
+            for fact in known_facts:
+                text = str(fact).strip()
+                if len(text) <= 1 and text.isalpha():
+                    continue
+                if text and text not in self.known_facts:
+                    self.known_facts.append(text)
+            self.known_facts = self.known_facts[-10:]
+
+        if missing_points:
+            cleaned = []
+            for point in missing_points:
+                text = str(point).strip()
+                if len(text) <= 1 and text.isalpha():
+                    continue
+                if text:
+                    cleaned.append(text)
+            # Keep latest gaps as "active" missing info.
+            self.missing_points = cleaned[-8:]
+    
+    def get_context_for_llm(self) -> str:
+        """Get formatted context for LLM prompts."""
+        context_parts = []
+        
+        if self.action_history:
+            recent = self.action_history[-5:]
+            context_parts.append("Recent actions:")
+            for h in recent:
+                action_str = h.get('action', h)
+                node_str = h.get('node', 'action')
+                context_parts.append(f"  - {node_str}: {action_str}")
+        
+        if self.extracted_data:
+            context_parts.append("\nExtracted data:")
+            for d in self.extracted_data[-5:]:
+                # Support both old format (source/data) and new format (url/preview)
+                source = d.get('source') or d.get('url', 'unknown')
+                data = d.get('data') or d.get('preview', '')[:100]
+                context_parts.append(f"  - {source[:50]}: {data[:100]}...")
+
+        if self.known_facts:
+            context_parts.append("\nKnown facts:")
+            for fact in self.known_facts[-5:]:
+                context_parts.append(f"  - {fact}")
+
+        if self.missing_points:
+            context_parts.append("\nMissing points:")
+            for point in self.missing_points[-5:]:
+                context_parts.append(f"  - {point}")
+
+        if self.last_queries:
+            context_parts.append("\nRecent queries:")
+            for query in self.last_queries[-5:]:
+                context_parts.append(f"  - {query}")
+        
+        if self.error_history:
+            context_parts.append("\nErrors encountered:")
+            for e in self.error_history[-3:]:
+                context_parts.append(f"  - {e.get('error', str(e))}")
+        
+        return "\n".join(context_parts)
+    
+    def should_continue(self) -> bool:
+        """Check if agent should continue execution based on timeout."""
+        import time
+        if self.start_time == 0:
+            self.start_time = time.time()
+        
+        elapsed = time.time() - self.start_time
+        time_ok = elapsed < self.timeout_seconds
+        
+        return (
+            not self.success and
+            time_ok and
+            self.current_node != NodeType.ERROR
+        )
+    
+    def get_elapsed_time(self) -> float:
+        """Get elapsed time in seconds."""
+        import time
+        if self.start_time == 0:
+            return 0.0
+        return time.time() - self.start_time
+    
+    def get_remaining_time(self) -> float:
+        """Get remaining time in seconds."""
+        return max(0, self.timeout_seconds - self.get_elapsed_time())

app/agents/heavy_search.py

@@ -0,0 +1,192 @@
+"""Heavy Search Agent.
+
+Middle-ground between Quick Search and Deep Research.
+Scrapes full content from top results for richer answers.
+"""
+
+import json
+import time
+from typing import AsyncIterator
+
+from app.agents.llm_client import generate_completion_stream
+from app.sources.aggregator import aggregate_search
+from app.sources.scraper import scrape_multiple_urls
+from app.reranking.pipeline import rerank_results
+from app.temporal.intent_detector import detect_temporal_intent
+
+
+async def run_heavy_search(
+    query: str,
+    max_results: int = 15,
+    max_scrape: int = 8,
+    freshness: str = "any",
+) -> AsyncIterator[str]:
+    """
+    Run heavy search with content scraping.
+    
+    Steps:
+    1. Aggregate search from multiple sources
+    2. Rerank results
+    3. Scrape full content from top N results
+    4. Stream synthesized answer
+    
+    Yields:
+        SSE event strings
+    """
+    start_time = time.perf_counter()
+    
+    try:
+        # Step 1: Status
+        yield _sse_event("status", {"phase": "searching", "message": "Searching multiple sources..."})
+        
+        # Step 2: Aggregate search
+        temporal_intent, temporal_urgency = detect_temporal_intent(query)
+        
+        raw_results = await aggregate_search(
+            query=query,
+            max_results=max_results + 5,
+            freshness=freshness,
+            include_wikipedia=True,
+        )
+        
+        if not raw_results:
+            yield _sse_event("error", {"message": "No results found"})
+            return
+        
+        yield _sse_event("search_complete", {
+            "results_count": len(raw_results),
+            "sources": list(set(r.get("source", "unknown") for r in raw_results)),
+        })
+        
+        # Step 3: Rerank (use embeddings when we have many results from SearXNG)
+        yield _sse_event("status", {"phase": "ranking", "message": "Ranking results..."})
+        
+        # Enable embeddings when we have many results (SearXNG provides volume)
+        use_embeddings = len(raw_results) > 20
+        
+        ranked_results = await rerank_results(
+            query=query,
+            results=raw_results,
+            temporal_urgency=temporal_urgency,
+            max_results=max_results,
+            use_embeddings=use_embeddings,
+        )
+        
+        # Step 4: Scrape top results
+        yield _sse_event("status", {"phase": "scraping", "message": f"Reading top {max_scrape} sources..."})
+        
+        urls_to_scrape = [r.get("url") for r in ranked_results[:max_scrape] if r.get("url")]
+        scraped_content = await scrape_multiple_urls(
+            urls=urls_to_scrape,
+            max_chars_per_url=4000,
+            max_concurrent=3,
+        )
+        
+        # Merge scraped content into results
+        for result in ranked_results:
+            url = result.get("url", "")
+            if url in scraped_content and scraped_content[url]:
+                result["full_content"] = scraped_content[url]
+                result["scraped"] = True
+            else:
+                result["full_content"] = result.get("content", "")
+                result["scraped"] = False
+        
+        scraped_count = sum(1 for r in ranked_results if r.get("scraped"))
+        yield _sse_event("scrape_complete", {
+            "scraped_count": scraped_count,
+            "total": len(urls_to_scrape),
+        })
+        
+        # Step 5: Send results
+        yield _sse_event("results", {
+            "results": [
+                {
+                    "title": r.get("title", ""),
+                    "url": r.get("url", ""),
+                    "score": r.get("score", 0),
+                    "source": r.get("source", ""),
+                    "scraped": r.get("scraped", False),
+                }
+                for r in ranked_results
+            ],
+            "temporal_intent": temporal_intent,
+            "temporal_urgency": temporal_urgency,
+        })
+        
+        # Step 6: Synthesize answer
+        yield _sse_event("status", {"phase": "synthesizing", "message": "Generating answer..."})
+        yield _sse_event("answer_start", {})
+        
+        async for chunk in _synthesize_heavy_answer(query, ranked_results, temporal_intent):
+            yield _sse_event("answer_chunk", {"content": chunk})
+        
+        # Done
+        total_time = time.perf_counter() - start_time
+        yield _sse_event("done", {
+            "total_sources": len(ranked_results),
+            "scraped_sources": scraped_count,
+            "total_time_seconds": round(total_time, 2),
+        })
+        
+    except Exception as e:
+        yield _sse_event("error", {"message": str(e)})
+
+
+async def _synthesize_heavy_answer(
+    query: str,
+    results: list[dict],
+    temporal_intent: str,
+) -> AsyncIterator[str]:
+    """Synthesize answer from scraped content."""
+    
+    # Build context with full content
+    context_parts = []
+    for i, r in enumerate(results[:8], 1):
+        content = r.get("full_content", r.get("content", ""))[:3000]
+        scraped_tag = "[FULL]" if r.get("scraped") else "[SNIPPET]"
+        
+        context_parts.append(
+            f"[{i}] {r.get('title', 'Untitled')} {scraped_tag}\n"
+            f"URL: {r.get('url', '')}\n"
+            f"Content:\n{content}\n"
+        )
+    
+    context = "\n---\n".join(context_parts)
+    
+    prompt = f"""You are a research assistant providing comprehensive answers.
+
+QUERY: {query}
+TEMPORAL INTENT: {temporal_intent}
+
+SOURCES (some with full content [FULL], some with snippets [SNIPPET]):
+{context}
+
+INSTRUCTIONS:
+1. Provide a comprehensive, well-structured answer
+2. Use information from [FULL] sources more extensively
+3. Cite sources using [1], [2], etc.
+4. Write in the same language as the query
+5. Be thorough but clear
+
+Answer:"""
+
+    messages = [
+        {"role": "system", "content": "You are a helpful research assistant."},
+        {"role": "user", "content": prompt},
+    ]
+    
+    async for chunk in generate_completion_stream(messages, temperature=0.3):
+        yield chunk
+    
+    # Add citations
+    yield "\n\n---\n**Sources:**\n"
+    for i, r in enumerate(results[:8], 1):
+        scraped = "📄" if r.get("scraped") else "📋"
+        yield f"{scraped} [{i}] [{r.get('title', 'Untitled')}]({r.get('url', '')})\n"
+
+
+def _sse_event(event_type: str, data: dict) -> str:
+    """Format an SSE event."""
+    payload = {"type": event_type, **data}
+    return f"data: {json.dumps(payload)}\n\n"

app/agents/llm_client.py

@@ -0,0 +1,522 @@
+"""LLM client abstraction for multiple providers.
+
+Supports Groq and OpenRouter for LLM inference.
+"""
+
+from dataclasses import dataclass, field
+import httpx
+import json
+from typing import Optional, AsyncIterator, Any
+
+from tenacity import (
+    retry,
+    stop_after_attempt,
+    wait_exponential,
+    retry_if_exception_type,
+)
+
+from app.config import get_settings
+from app.agents.tooling import ToolCall
+
+
+class RetryableError(Exception):
+    """Error that should trigger a retry."""
+    pass
+
+
+@dataclass
+class CompletionResponse:
+    """Normalized LLM response across providers."""
+
+    content: str = ""
+    tool_calls: list[ToolCall] = field(default_factory=list)
+    reasoning: list[str] = field(default_factory=list)
+    model: str | None = None
+    finish_reason: str | None = None
+    raw: dict[str, Any] | None = None
+
+
+async def generate_completion(
+    messages: list[dict],
+    model: Optional[str] = None,
+    temperature: float = 0.3,
+    max_tokens: int = 2048,
+    tools: list[dict[str, Any]] | None = None,
+    tool_choice: str | dict[str, Any] | None = None,
+    reasoning_effort: str | None = None,
+    prefer_responses_api: bool = False,
+) -> str:
+    """Generate a completion using the configured LLM provider."""
+    response = await generate_completion_response(
+        messages=messages,
+        model=model,
+        temperature=temperature,
+        max_tokens=max_tokens,
+        tools=tools,
+        tool_choice=tool_choice,
+        reasoning_effort=reasoning_effort,
+        prefer_responses_api=prefer_responses_api,
+    )
+    return response.content
+
+
+async def generate_completion_response(
+    messages: list[dict],
+    model: Optional[str] = None,
+    temperature: float = 0.3,
+    max_tokens: int = 2048,
+    tools: list[dict[str, Any]] | None = None,
+    tool_choice: str | dict[str, Any] | None = None,
+    reasoning_effort: str | None = None,
+    prefer_responses_api: bool = False,
+) -> CompletionResponse:
+    """Generate a normalized completion response with optional tool calls."""
+    settings = get_settings()
+    provider = settings.llm_provider
+    model = model or settings.llm_model
+    
+    if provider == "groq":
+        return await _call_groq(messages, model, temperature, max_tokens, tools, tool_choice)
+    elif provider == "openrouter":
+        if prefer_responses_api:
+            return await _call_openrouter_responses(
+                messages=messages,
+                model=model,
+                temperature=temperature,
+                max_tokens=max_tokens,
+                tools=tools,
+                tool_choice=tool_choice,
+                reasoning_effort=reasoning_effort,
+            )
+        return await _call_openrouter(messages, model, temperature, max_tokens, tools, tool_choice)
+    else:
+        raise ValueError(f"Unknown LLM provider: {provider}")
+
+
+def _build_payload(
+    model: str,
+    messages: list[dict],
+    temperature: float,
+    max_tokens: int,
+    tools: list[dict[str, Any]] | None = None,
+    tool_choice: str | dict[str, Any] | None = None,
+    stream: bool = False,
+) -> dict[str, Any]:
+    """Build an OpenAI-compatible chat payload."""
+    payload: dict[str, Any] = {
+        "model": model,
+        "messages": messages,
+        "temperature": temperature,
+        "max_tokens": max_tokens,
+    }
+    if stream:
+        payload["stream"] = True
+    if tools:
+        payload["tools"] = tools
+        if tool_choice is not None:
+            payload["tool_choice"] = tool_choice
+    return payload
+
+
+def _parse_tool_calls(message: dict[str, Any]) -> list[ToolCall]:
+    """Parse tool calls from an OpenAI-compatible response message."""
+    parsed_calls: list[ToolCall] = []
+
+    for index, raw_tool_call in enumerate(message.get("tool_calls", []) or []):
+        function_payload = raw_tool_call.get("function", {}) or {}
+        raw_arguments = function_payload.get("arguments", "{}")
+        try:
+            arguments = json.loads(raw_arguments) if raw_arguments else {}
+        except json.JSONDecodeError:
+            arguments = {"raw": raw_arguments}
+
+        parsed_calls.append(
+            ToolCall(
+                id=str(raw_tool_call.get("id", f"tool_call_{index}")),
+                name=str(function_payload.get("name", "")),
+                arguments=arguments,
+            )
+        )
+
+    return parsed_calls
+
+
+def _parse_completion_response(data: dict[str, Any]) -> CompletionResponse:
+    """Parse an OpenAI-compatible completion payload into a normalized response."""
+    choice = (data.get("choices") or [{}])[0]
+    message = choice.get("message") or {}
+
+    return CompletionResponse(
+        content=message.get("content") or "",
+        tool_calls=_parse_tool_calls(message),
+        reasoning=[],
+        model=data.get("model"),
+        finish_reason=choice.get("finish_reason"),
+        raw=data,
+    )
+
+
+def _chat_messages_to_responses_input(messages: list[dict]) -> list[dict[str, Any]]:
+    """Convert chat-style messages into OpenRouter Responses API input items."""
+    converted: list[dict[str, Any]] = []
+
+    for message in messages:
+        role = str(message.get("role", "user"))
+        content = message.get("content", "")
+        if content is None:
+            content = ""
+
+        content_type = "output_text" if role == "assistant" else "input_text"
+        converted.append(
+            {
+                "type": "message",
+                "role": role,
+                "content": [
+                    {
+                        "type": content_type,
+                        "text": str(content),
+                    }
+                ],
+            }
+        )
+
+    return converted
+
+
+def _responses_tools_from_chat_tools(tools: list[dict[str, Any]] | None) -> list[dict[str, Any]] | None:
+    """Convert chat-completions tool format into Responses API tool format."""
+    if not tools:
+        return None
+
+    converted: list[dict[str, Any]] = []
+    for tool in tools:
+        if tool.get("type") != "function":
+            continue
+        function_def = tool.get("function", {}) or {}
+        converted.append(
+            {
+                "type": "function",
+                "name": function_def.get("name", ""),
+                "description": function_def.get("description", ""),
+                "parameters": function_def.get("parameters", {"type": "object", "properties": {}}),
+                "strict": None,
+            }
+        )
+    return converted
+
+
+def _parse_openrouter_responses_api(data: dict[str, Any]) -> CompletionResponse:
+    """Parse OpenRouter Responses API output into a normalized response."""
+    content_parts: list[str] = []
+    tool_calls: list[ToolCall] = []
+    reasoning_parts: list[str] = []
+
+    for index, item in enumerate(data.get("output", []) or []):
+        item_type = item.get("type")
+
+        if item_type == "reasoning":
+            summaries = item.get("summary", []) or []
+            for summary in summaries:
+                if isinstance(summary, str):
+                    reasoning_parts.append(summary.strip())
+                elif isinstance(summary, dict):
+                    text = summary.get("text") or summary.get("summary") or ""
+                    if text:
+                        reasoning_parts.append(str(text).strip())
+            continue
+
+        if item_type == "function_call":
+            raw_arguments = item.get("arguments", "{}")
+            try:
+                arguments = json.loads(raw_arguments) if raw_arguments else {}
+            except json.JSONDecodeError:
+                arguments = {"raw": raw_arguments}
+
+            tool_calls.append(
+                ToolCall(
+                    id=str(item.get("call_id") or item.get("id") or f"tool_call_{index}"),
+                    name=str(item.get("name", "")),
+                    arguments=arguments,
+                )
+            )
+            continue
+
+        if item_type == "message":
+            for content_item in item.get("content", []) or []:
+                if content_item.get("type") == "output_text":
+                    text = content_item.get("text") or ""
+                    if text:
+                        content_parts.append(str(text))
+
+    return CompletionResponse(
+        content="\n".join(part for part in content_parts if part).strip(),
+        tool_calls=tool_calls,
+        reasoning=[part for part in reasoning_parts if part],
+        model=data.get("model"),
+        finish_reason=data.get("status"),
+        raw=data,
+    )
+
+
+@retry(
+    stop=stop_after_attempt(3),
+    wait=wait_exponential(multiplier=1, min=2, max=10),
+    retry=retry_if_exception_type(RetryableError),
+    reraise=True,
+)
+async def _call_groq(
+    messages: list[dict],
+    model: str,
+    temperature: float,
+    max_tokens: int,
+    tools: list[dict[str, Any]] | None = None,
+    tool_choice: str | dict[str, Any] | None = None,
+) -> CompletionResponse:
+    """Call Groq API with retry logic."""
+    settings = get_settings()
+    
+    if not settings.groq_api_key:
+        raise ValueError("GROQ_API_KEY not configured")
+    
+    try:
+        async with httpx.AsyncClient(timeout=60.0) as client:
+            payload = _build_payload(
+                model=model,
+                messages=messages,
+                temperature=temperature,
+                max_tokens=max_tokens,
+                tools=tools,
+                tool_choice=tool_choice,
+            )
+            response = await client.post(
+                "https://api.groq.com/openai/v1/chat/completions",
+                headers={
+                    "Authorization": f"Bearer {settings.groq_api_key}",
+                    "Content-Type": "application/json",
+                },
+                json=payload,
+            )
+            
+            # Retry on rate limit or server errors
+            if response.status_code in (429, 502, 503, 504):
+                raise RetryableError(f"Groq error {response.status_code}")
+            
+            if response.status_code == 400 and tools:
+                # Some model/provider combinations reject tools. Fall back to plain text.
+                fallback_payload = _build_payload(
+                    model=model,
+                    messages=messages,
+                    temperature=temperature,
+                    max_tokens=max_tokens,
+                )
+                response = await client.post(
+                    "https://api.groq.com/openai/v1/chat/completions",
+                    headers={
+                        "Authorization": f"Bearer {settings.groq_api_key}",
+                        "Content-Type": "application/json",
+                    },
+                    json=fallback_payload,
+                )
+
+            response.raise_for_status()
+            data = response.json()
+            
+        return _parse_completion_response(data)
+    except httpx.TimeoutException as e:
+        raise RetryableError(f"Groq timeout: {e}")
+
+
+@retry(
+    stop=stop_after_attempt(3),
+    wait=wait_exponential(multiplier=1, min=2, max=10),
+    retry=retry_if_exception_type(RetryableError),
+    reraise=True,
+)
+async def _call_openrouter(
+    messages: list[dict],
+    model: str,
+    temperature: float,
+    max_tokens: int,
+    tools: list[dict[str, Any]] | None = None,
+    tool_choice: str | dict[str, Any] | None = None,
+) -> CompletionResponse:
+    """Call OpenRouter API with retry logic."""
+    settings = get_settings()
+    
+    if not settings.openrouter_api_key:
+        raise ValueError("OPENROUTER_API_KEY not configured")
+    
+    headers = {
+        "Authorization": f"Bearer {settings.openrouter_api_key}",
+        "Content-Type": "application/json",
+        "HTTP-Referer": "https://madras1-lancer.hf.space",
+        "X-Title": "Lancer Search API",
+    }
+    
+    try:
+        async with httpx.AsyncClient(timeout=120.0) as client:
+            payload = _build_payload(
+                model=model,
+                messages=messages,
+                temperature=temperature,
+                max_tokens=max_tokens,
+                tools=tools,
+                tool_choice=tool_choice,
+            )
+            response = await client.post(
+                "https://openrouter.ai/api/v1/chat/completions",
+                headers=headers,
+                json=payload,
+            )
+            
+            # Retry on rate limit or server errors
+            if response.status_code in (429, 502, 503, 504):
+                raise RetryableError(f"OpenRouter error {response.status_code}")
+            
+            if response.status_code == 400 and tools:
+                fallback_payload = _build_payload(
+                    model=model,
+                    messages=messages,
+                    temperature=temperature,
+                    max_tokens=max_tokens,
+                )
+                response = await client.post(
+                    "https://openrouter.ai/api/v1/chat/completions",
+                    headers=headers,
+                    json=fallback_payload,
+                )
+
+            if response.status_code != 200:
+                error_text = response.text
+                raise ValueError(f"OpenRouter error {response.status_code}: {error_text}")
+            
+            data = response.json()
+            return _parse_completion_response(data)
+    except httpx.TimeoutException as e:
+        raise RetryableError(f"OpenRouter timeout: {e}")
+
+
+@retry(
+    stop=stop_after_attempt(3),
+    wait=wait_exponential(multiplier=1, min=2, max=10),
+    retry=retry_if_exception_type(RetryableError),
+    reraise=True,
+)
+async def _call_openrouter_responses(
+    messages: list[dict],
+    model: str,
+    temperature: float,
+    max_tokens: int,
+    tools: list[dict[str, Any]] | None = None,
+    tool_choice: str | dict[str, Any] | None = None,
+    reasoning_effort: str | None = None,
+) -> CompletionResponse:
+    """Call OpenRouter Responses API and keep reasoning separated from content."""
+    settings = get_settings()
+
+    if not settings.openrouter_api_key:
+        raise ValueError("OPENROUTER_API_KEY not configured")
+
+    headers = {
+        "Authorization": f"Bearer {settings.openrouter_api_key}",
+        "Content-Type": "application/json",
+        "HTTP-Referer": "https://madras1-lancer.hf.space",
+        "X-Title": "Lancer Search API",
+    }
+
+    payload: dict[str, Any] = {
+        "model": model,
+        "input": _chat_messages_to_responses_input(messages),
+        "max_output_tokens": max_tokens,
+        "temperature": temperature,
+    }
+
+    responses_tools = _responses_tools_from_chat_tools(tools)
+    if responses_tools:
+        payload["tools"] = responses_tools
+        if tool_choice is not None:
+            if isinstance(tool_choice, str):
+                payload["tool_choice"] = tool_choice
+            elif isinstance(tool_choice, dict):
+                function_name = (
+                    tool_choice.get("function", {}) or {}
+                ).get("name") or tool_choice.get("name")
+                if function_name:
+                    payload["tool_choice"] = {"type": "function", "name": function_name}
+
+    if reasoning_effort:
+        payload["reasoning"] = {"effort": reasoning_effort}
+
+    try:
+        async with httpx.AsyncClient(timeout=120.0) as client:
+            response = await client.post(
+                "https://openrouter.ai/api/v1/responses",
+                headers=headers,
+                json=payload,
+            )
+
+            if response.status_code in (429, 502, 503, 504):
+                raise RetryableError(f"OpenRouter responses error {response.status_code}")
+
+            if response.status_code != 200:
+                error_text = response.text
+                raise ValueError(f"OpenRouter responses error {response.status_code}: {error_text}")
+
+            data = response.json()
+            return _parse_openrouter_responses_api(data)
+    except httpx.TimeoutException as e:
+        raise RetryableError(f"OpenRouter responses timeout: {e}")
+
+
+async def generate_completion_stream(
+    messages: list[dict],
+    model: Optional[str] = None,
+    temperature: float = 0.3,
+    max_tokens: int = 2048,
+) -> AsyncIterator[str]:
+    """Generate a streaming completion using OpenRouter."""
+    settings = get_settings()
+    model = model or settings.llm_model
+    
+    if not settings.openrouter_api_key:
+        raise ValueError("OPENROUTER_API_KEY not configured")
+    
+    headers = {
+        "Authorization": f"Bearer {settings.openrouter_api_key}",
+        "Content-Type": "application/json",
+        "HTTP-Referer": "https://madras1-lancer.hf.space",
+        "X-Title": "Lancer Search API",
+    }
+    
+    payload = _build_payload(
+        model=model,
+        messages=messages,
+        temperature=temperature,
+        max_tokens=max_tokens,
+        stream=True,
+    )
+    
+    async with httpx.AsyncClient(timeout=120.0) as client:
+        async with client.stream(
+            "POST",
+            "https://openrouter.ai/api/v1/chat/completions",
+            headers=headers,
+            json=payload,
+        ) as response:
+            if response.status_code != 200:
+                error_text = await response.aread()
+                raise ValueError(f"OpenRouter streaming error {response.status_code}: {error_text}")
+            
+            async for line in response.aiter_lines():
+                if line.startswith("data: "):
+                    data_str = line[6:]
+                    if data_str.strip() == "[DONE]":
+                        break
+                    try:
+                        data = json.loads(data_str)
+                        delta = data.get("choices", [{}])[0].get("delta", {})
+                        content = delta.get("content", "")
+                        if content:
+                            yield content
+                    except json.JSONDecodeError:
+                        continue

app/agents/planner.py

@@ -0,0 +1,133 @@
+"""Research Planner Agent.
+
+Decomposes complex queries into multiple research dimensions.
+"""
+
+import json
+from typing import Optional
+
+from pydantic import BaseModel, Field
+
+from app.agents.llm_client import generate_completion
+from app.config import get_settings
+
+
+class ResearchDimension(BaseModel):
+    """A single dimension/aspect to research."""
+    
+    name: str = Field(..., description="Short name for this dimension")
+    description: str = Field(..., description="What this dimension covers")
+    search_query: str = Field(..., description="Optimized search query for this dimension")
+    priority: int = Field(default=1, ge=1, le=3, description="1=high, 2=medium, 3=low")
+
+
+class ResearchPlan(BaseModel):
+    """Complete research plan with all dimensions."""
+    
+    original_query: str
+    refined_query: str = Field(..., description="Clarified version of the query")
+    dimensions: list[ResearchDimension]
+    estimated_sources: int = Field(default=20)
+
+
+PLANNER_PROMPT = """You are a research planning assistant. Your job is to decompose a complex query into multiple research dimensions.
+
+USER QUERY: {query}
+
+INSTRUCTIONS:
+1. Analyze the query and identify 2-6 key dimensions/aspects that need to be researched
+2. Each dimension should be distinct and cover a different angle
+3. Create an optimized search query for each dimension
+4. Assign priority (1=high, 2=medium, 3=low) based on relevance to the main query
+5. Respond ONLY with valid JSON, no other text
+
+OUTPUT FORMAT:
+{{
+    "refined_query": "A clearer version of the user's query",
+    "dimensions": [
+        {{
+            "name": "Short name",
+            "description": "What this covers",
+            "search_query": "Optimized search query",
+            "priority": 1
+        }}
+    ]
+}}
+
+Generate the research plan:"""
+
+
+async def create_research_plan(
+    query: str,
+    max_dimensions: int = 6,
+) -> ResearchPlan:
+    """
+    Create a research plan by decomposing a query into dimensions.
+    
+    Args:
+        query: The user's research query
+        max_dimensions: Maximum number of dimensions to generate
+        
+    Returns:
+        ResearchPlan with dimensions to investigate
+    """
+    settings = get_settings()
+    
+    messages = [
+        {"role": "system", "content": "You are a research planning assistant. Always respond with valid JSON only."},
+        {"role": "user", "content": PLANNER_PROMPT.format(query=query)},
+    ]
+    
+    try:
+        response = await generate_completion(messages, temperature=0.3)
+        
+        # Parse JSON response
+        # Try to extract JSON if there's extra text
+        json_start = response.find("{")
+        json_end = response.rfind("}") + 1
+        if json_start >= 0 and json_end > json_start:
+            response = response[json_start:json_end]
+        
+        data = json.loads(response)
+        
+        # Build dimensions
+        dimensions = []
+        for dim_data in data.get("dimensions", [])[:max_dimensions]:
+            dimensions.append(ResearchDimension(
+                name=dim_data.get("name", "Unknown"),
+                description=dim_data.get("description", ""),
+                search_query=dim_data.get("search_query", query),
+                priority=dim_data.get("priority", 2),
+            ))
+        
+        # Sort by priority
+        dimensions.sort(key=lambda d: d.priority)
+        
+        return ResearchPlan(
+            original_query=query,
+            refined_query=data.get("refined_query", query),
+            dimensions=dimensions,
+            estimated_sources=len(dimensions) * 5,
+        )
+        
+    except (json.JSONDecodeError, KeyError) as e:
+        # Fallback: create a simple 2-dimension plan
+        return ResearchPlan(
+            original_query=query,
+            refined_query=query,
+            dimensions=[
+                ResearchDimension(
+                    name="Main Research",
+                    description=f"Primary research on: {query}",
+                    search_query=query,
+                    priority=1,
+                ),
+                ResearchDimension(
+                    name="Background",
+                    description=f"Background and context for: {query}",
+                    search_query=f"{query} background overview",
+                    priority=2,
+                ),
+            ],
+            estimated_sources=10,
+        )

app/agents/synthesizer.py

@@ -0,0 +1,173 @@
+"""Answer synthesizer agent.
+
+Generates a coherent answer from search results with citations.
+"""
+
+from datetime import datetime
+from typing import Optional, AsyncIterator
+
+from app.api.schemas import SearchResult, TemporalContext, Citation
+from app.agents.llm_client import generate_completion, generate_completion_stream
+
+
+SYNTHESIS_PROMPT = """You are a research assistant that synthesizes information from search results.
+
+CURRENT DATE: {current_date}
+
+USER QUERY: {query}
+
+TEMPORAL CONTEXT:
+- Query intent: {temporal_intent} (the user {intent_explanation})
+- Temporal urgency: {temporal_urgency:.0%} (how important freshness is)
+
+SEARCH RESULTS:
+{formatted_results}
+
+INSTRUCTIONS:
+1. Synthesize a comprehensive answer based on the search results
+2. ALWAYS cite your sources using [1], [2], etc. format
+3. If the query requires current information, prioritize the most recent results
+4. If there are conflicting dates or versions mentioned, use the most recent accurate information
+5. Be concise but thorough
+6. If information seems outdated compared to current date ({current_date}), note this
+7. Write in the same language as the query
+
+Generate your answer:"""
+
+
+async def synthesize_answer(
+    query: str,
+    results: list[SearchResult],
+    temporal_context: Optional[TemporalContext] = None,
+) -> tuple[str, list[Citation]]:
+    """
+    Synthesize an answer from search results.
+    
+    Args:
+        query: Original search query
+        results: List of search results to synthesize from
+        temporal_context: Temporal analysis context
+        
+    Returns:
+        Tuple of (answer_text, citations_list)
+    """
+    if not results:
+        return "No results found to synthesize an answer.", []
+    
+    messages = _build_messages(query, results, temporal_context)
+    
+    try:
+        answer = await generate_completion(messages, temperature=0.3)
+    except Exception as e:
+        # Fallback: return a simple summary without LLM
+        answer = f"Error generating synthesis: {e}. Please review the search results directly."
+    
+    # Build citations list
+    citations = _build_citations(results)
+    
+    return answer, citations
+
+
+async def synthesize_answer_stream(
+    query: str,
+    results: list[SearchResult],
+    temporal_context: Optional[TemporalContext] = None,
+) -> AsyncIterator[str]:
+    """
+    Synthesize an answer with streaming output.
+    
+    Yields chunks of the answer as they are generated.
+    
+    Args:
+        query: Original search query
+        results: List of search results to synthesize from
+        temporal_context: Temporal analysis context
+        
+    Yields:
+        Chunks of the answer text
+    """
+    if not results:
+        yield "No results found to synthesize an answer."
+        return
+    
+    messages = _build_messages(query, results, temporal_context)
+    
+    try:
+        async for chunk in generate_completion_stream(messages, temperature=0.3):
+            yield chunk
+    except Exception as e:
+        yield f"Error generating synthesis: {e}. Please review the search results directly."
+
+
+def _build_messages(
+    query: str,
+    results: list[SearchResult],
+    temporal_context: Optional[TemporalContext] = None,
+) -> list[dict]:
+    """Build messages for LLM prompt."""
+    # Format results for the prompt
+    formatted_results = format_results_for_prompt(results[:10])  # Top 10 only
+    
+    # Prepare temporal context
+    current_date = datetime.now().strftime("%Y-%m-%d")
+    temporal_intent = "neutral"
+    temporal_urgency = 0.5
+    
+    if temporal_context:
+        temporal_intent = temporal_context.query_temporal_intent
+        temporal_urgency = temporal_context.temporal_urgency
+        current_date = temporal_context.current_date
+    
+    # Map intent to explanation
+    intent_explanations = {
+        "current": "is looking for the most recent/current information",
+        "historical": "is interested in historical or background information",
+        "neutral": "has no specific temporal preference",
+    }
+    
+    prompt = SYNTHESIS_PROMPT.format(
+        current_date=current_date,
+        query=query,
+        temporal_intent=temporal_intent,
+        intent_explanation=intent_explanations.get(temporal_intent, ""),
+        temporal_urgency=temporal_urgency,
+        formatted_results=formatted_results,
+    )
+    
+    return [
+        {"role": "system", "content": "You are a helpful research assistant."},
+        {"role": "user", "content": prompt},
+    ]
+
+
+def _build_citations(results: list[SearchResult]) -> list[Citation]:
+    """Build citations list from results."""
+    citations = []
+    for i, result in enumerate(results[:10], 1):
+        citations.append(
+            Citation(
+                index=i,
+                url=result.url,
+                title=result.title,
+            )
+        )
+    return citations
+
+
+def format_results_for_prompt(results: list[SearchResult]) -> str:
+    """Format search results for inclusion in the LLM prompt."""
+    formatted = []
+    
+    for i, result in enumerate(results, 1):
+        date_str = ""
+        if result.published_date:
+            date_str = f" (Published: {result.published_date.strftime('%Y-%m-%d')})"
+        
+        formatted.append(
+            f"[{i}] {result.title}{date_str}\n"
+            f"    URL: {result.url}\n"
+            f"    Freshness: {result.freshness_score:.0%} | Authority: {result.authority_score:.0%}\n"
+            f"    Content: {result.content[:500]}..."
+        )
+    
+    return "\n\n".join(formatted)

app/agents/tooling.py

@@ -0,0 +1,221 @@
+"""Lightweight tooling primitives for Lancer agents.
+
+This module provides a small subset of the ideas from JadeAgent:
+- declarative tool schemas from Python signatures
+- a registry for reusable tools
+- structured tool-call objects that can be returned by LLM backends
+
+It intentionally stays small and synchronous so the existing agent code can
+adopt it incrementally without turning Lancer into a generic framework.
+"""
+
+from __future__ import annotations
+
+import inspect
+from dataclasses import dataclass, field
+from typing import Any, Callable, get_args, get_origin, get_type_hints
+
+
+JSON_TYPE_MAP = {
+    str: "string",
+    int: "integer",
+    float: "number",
+    bool: "boolean",
+    list: "array",
+    dict: "object",
+}
+
+
+def _python_type_to_json_schema(py_type: type) -> dict[str, Any]:
+    """Map simple Python annotations to JSON schema fragments."""
+    origin = get_origin(py_type)
+    if origin is not None:
+        args = get_args(py_type)
+        if origin is list:
+            item_type = args[0] if args else str
+            return {"type": "array", "items": _python_type_to_json_schema(item_type)}
+        if origin is dict:
+            return {"type": "object"}
+        non_none = [arg for arg in args if arg is not type(None)]
+        if non_none:
+            return _python_type_to_json_schema(non_none[0])
+
+    return {"type": JSON_TYPE_MAP.get(py_type, "string")}
+
+
+def _extract_param_description(docstring: str | None, param_name: str) -> str | None:
+    """Extract a simple parameter description from an Args: section."""
+    if not docstring:
+        return None
+
+    lines = docstring.splitlines()
+    in_args = False
+
+    for raw_line in lines:
+        line = raw_line.strip()
+        if line.lower().startswith("args:"):
+            in_args = True
+            continue
+        if not in_args:
+            continue
+        if not line:
+            continue
+        if line.startswith(f"{param_name}:"):
+            return line.split(":", 1)[1].strip() or None
+        if line.startswith(f"{param_name} "):
+            parts = line.split(":", 1)
+            if len(parts) > 1:
+                return parts[1].strip() or None
+        # Stop when we leave the Args section.
+        if not raw_line.startswith((" ", "\t")):
+            break
+
+    return None
+
+
+@dataclass(frozen=True)
+class ToolCall:
+    """Structured tool call emitted by the LLM layer."""
+
+    id: str
+    name: str
+    arguments: dict[str, Any]
+
+
+@dataclass(frozen=True)
+class ToolSchema:
+    """OpenAI-compatible tool schema."""
+
+    name: str
+    description: str
+    parameters: dict[str, Any]
+
+    def to_openai_tool(self) -> dict[str, Any]:
+        """Convert schema into OpenAI-compatible tool format."""
+        return {
+            "type": "function",
+            "function": {
+                "name": self.name,
+                "description": self.description,
+                "parameters": self.parameters,
+            },
+        }
+
+
+@dataclass
+class Tool:
+    """Registered callable tool with generated schema."""
+
+    func: Callable[..., Any]
+    name: str | None = None
+    description: str | None = None
+    schema: ToolSchema = field(init=False)
+
+    def __post_init__(self):
+        if self.name is None:
+            self.name = self.func.__name__
+        if self.description is None:
+            self.description = self.func.__doc__ or f"Tool: {self.name}"
+        self.schema = self._build_schema()
+
+    def _build_schema(self) -> ToolSchema:
+        sig = inspect.signature(self.func)
+        hints = get_type_hints(self.func)
+        properties: dict[str, Any] = {}
+        required: list[str] = []
+
+        for param_name, param in sig.parameters.items():
+            if param_name in {"self", "cls"}:
+                continue
+
+            py_type = hints.get(param_name, str)
+            prop = _python_type_to_json_schema(py_type)
+            description = _extract_param_description(self.func.__doc__, param_name)
+            if description:
+                prop["description"] = description
+            properties[param_name] = prop
+
+            if param.default is inspect.Parameter.empty:
+                required.append(param_name)
+
+        parameters: dict[str, Any] = {
+            "type": "object",
+            "properties": properties,
+        }
+        if required:
+            parameters["required"] = required
+
+        return ToolSchema(
+            name=str(self.name),
+            description=str(self.description),
+            parameters=parameters,
+        )
+
+    def execute(self, arguments: dict[str, Any]) -> Any:
+        """Execute the tool with validated arguments."""
+        return self.func(**arguments)
+
+
+def tool(
+    func: Callable[..., Any] | None = None,
+    *,
+    name: str | None = None,
+    description: str | None = None,
+) -> Tool | Callable[[Callable[..., Any]], Tool]:
+    """Decorator for creating Tool objects from plain Python callables."""
+
+    def decorator(inner: Callable[..., Any]) -> Tool:
+        return Tool(func=inner, name=name, description=description)
+
+    if func is not None:
+        return decorator(func)
+
+    return decorator
+
+
+class ToolRegistry:
+    """Small registry of reusable tools."""
+
+    def __init__(self, tools: list[Tool | Callable[..., Any]] | None = None):
+        self._tools: dict[str, Tool] = {}
+        for item in tools or []:
+            self.register(item)
+
+    def register(self, item: Tool | Callable[..., Any]):
+        """Register a Tool or plain callable."""
+        if isinstance(item, Tool):
+            tool_obj = item
+        elif callable(item):
+            tool_obj = Tool(func=item)
+        else:
+            raise TypeError(f"Expected Tool or callable, got {type(item)!r}")
+
+        self._tools[str(tool_obj.name)] = tool_obj
+
+    def get(self, name: str) -> Tool | None:
+        """Get a tool by name."""
+        return self._tools.get(name)
+
+    @property
+    def names(self) -> list[str]:
+        """Registered tool names."""
+        return list(self._tools.keys())
+
+    @property
+    def schemas(self) -> list[ToolSchema]:
+        """Structured schemas for all tools."""
+        return [tool_obj.schema for tool_obj in self._tools.values()]
+
+    def as_openai_tools(self) -> list[dict[str, Any]]:
+        """Serialize all tools into OpenAI-compatible schema format."""
+        return [schema.to_openai_tool() for schema in self.schemas]
+
+    def execute(self, tool_call: ToolCall) -> Any:
+        """Execute a structured tool call."""
+        tool_obj = self.get(tool_call.name)
+        if tool_obj is None:
+            raise KeyError(f"Unknown tool '{tool_call.name}'. Available: {self.names}")
+        return tool_obj.execute(tool_call.arguments)
+
+    def __len__(self) -> int:
+        return len(self._tools)

app/api/__init__.py

@@ -0,0 +1 @@
+"""API routes package."""

app/api/routes/__init__.py

@@ -0,0 +1 @@
+"""API routes package."""

app/api/routes/search.py

@@ -0,0 +1,579 @@
+"""Search API routes."""
+
+import json
+import time
+from datetime import datetime
+
+from fastapi import APIRouter, HTTPException, Request
+from fastapi.responses import StreamingResponse
+
+from app.api.schemas import (
+    SearchRequest,
+    SearchResponse,
+    SearchResult,
+    TemporalContext,
+    Citation,
+    ErrorResponse,
+    DeepResearchRequest,
+    BrowseRequest,
+)
+from app.config import get_settings
+from app.temporal.intent_detector import detect_temporal_intent
+from app.temporal.freshness_scorer import calculate_freshness_score
+from app.sources.tavily import search_tavily
+from app.sources.duckduckgo import search_duckduckgo
+from app.reranking.pipeline import rerank_results
+from app.agents.synthesizer import synthesize_answer, synthesize_answer_stream
+from app.middleware.rate_limiter import limiter
+
+router = APIRouter()
+
+
+@router.post(
+    "/search",
+    response_model=SearchResponse,
+    responses={500: {"model": ErrorResponse}},
+    summary="Search with AI synthesis",
+    description="Perform a search with temporal intelligence and return an AI-synthesized answer.",
+)
+@limiter.limit("30/minute")
+async def search(request: Request, body: SearchRequest) -> SearchResponse:
+    """
+    Perform an intelligent search with:
+    - Temporal intent detection
+    - Multi-source search
+    - Multi-stage reranking
+    - AI-powered answer synthesis
+    """
+    start_time = time.perf_counter()
+    settings = get_settings()
+    
+    try:
+        # Step 1: Analyze temporal intent
+        temporal_intent, temporal_urgency = detect_temporal_intent(body.query)
+        
+        temporal_context = TemporalContext(
+            query_temporal_intent=temporal_intent,
+            temporal_urgency=temporal_urgency,
+            current_date=datetime.now().strftime("%Y-%m-%d"),
+        )
+        
+        # Step 2: Search multiple sources
+        raw_results = []
+        
+        # Try Tavily first (best quality)
+        if settings.tavily_api_key:
+            tavily_results = await search_tavily(
+                query=body.query,
+                max_results=settings.max_search_results,
+                freshness=body.freshness,
+                include_domains=body.include_domains,
+                exclude_domains=body.exclude_domains,
+            )
+            raw_results.extend(tavily_results)
+        
+        # Fallback to DuckDuckGo if needed
+        if not raw_results:
+            ddg_results = await search_duckduckgo(
+                query=body.query,
+                max_results=settings.max_search_results,
+            )
+            raw_results.extend(ddg_results)
+        
+        if not raw_results:
+            return SearchResponse(
+                query=body.query,
+                answer="No results found for your query.",
+                results=[],
+                citations=[],
+                temporal_context=temporal_context,
+                processing_time_ms=(time.perf_counter() - start_time) * 1000,
+            )
+        
+        # Step 3: Apply multi-stage reranking
+        ranked_results = await rerank_results(
+            query=body.query,
+            results=raw_results,
+            temporal_urgency=temporal_urgency,
+            max_results=body.max_results,
+        )
+        
+        # Step 4: Convert to SearchResult models
+        search_results = []
+        for i, result in enumerate(ranked_results):
+            freshness = calculate_freshness_score(result.get("published_date"))
+            search_results.append(
+                SearchResult(
+                    title=result.get("title", ""),
+                    url=result.get("url", ""),
+                    content=result.get("content", ""),
+                    score=result.get("score", 0.5),
+                    published_date=result.get("published_date"),
+                    freshness_score=freshness,
+                    authority_score=result.get("authority_score", 0.5),
+                )
+            )
+        
+        # Step 5: Synthesize answer (if requested)
+        answer = None
+        citations = []
+        
+        if body.include_answer and search_results:
+            answer, citations = await synthesize_answer(
+                query=body.query,
+                results=search_results,
+                temporal_context=temporal_context,
+            )
+        
+        processing_time = (time.perf_counter() - start_time) * 1000
+        
+        return SearchResponse(
+            query=body.query,
+            answer=answer,
+            results=search_results,
+            citations=citations,
+            temporal_context=temporal_context,
+            processing_time_ms=processing_time,
+        )
+        
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))
+
+
+
+@router.post(
+    "/search/raw",
+    response_model=SearchResponse,
+    summary="Search without synthesis",
+    description="Perform a search and return raw results without AI synthesis (faster).",
+)
+@limiter.limit("30/minute")
+async def search_raw(request: Request, body: SearchRequest) -> SearchResponse:
+    """Fast search without answer synthesis."""
+    body.include_answer = False
+    return await search(request, body)
+
+
+@router.post(
+    "/search/stream",
+    summary="Search with streaming synthesis",
+    description="Perform a search and stream the AI-synthesized answer in real-time using SSE.",
+)
+@limiter.limit("30/minute")
+async def search_stream(request: Request, body: SearchRequest):
+    """
+    Streaming search with Server-Sent Events.
+    
+    Returns results first, then streams the answer as it's generated.
+    """
+    settings = get_settings()
+    
+    async def event_generator():
+        try:
+            # Step 1: Analyze temporal intent
+            temporal_intent, temporal_urgency = detect_temporal_intent(body.query)
+            
+            temporal_context = TemporalContext(
+                query_temporal_intent=temporal_intent,
+                temporal_urgency=temporal_urgency,
+                current_date=datetime.now().strftime("%Y-%m-%d"),
+            )
+            
+            # Step 2: Search sources
+            raw_results = []
+            
+            if settings.tavily_api_key:
+                tavily_results = await search_tavily(
+                    query=body.query,
+                    max_results=settings.max_search_results,
+                    freshness=body.freshness,
+                    include_domains=body.include_domains,
+                    exclude_domains=body.exclude_domains,
+                )
+                raw_results.extend(tavily_results)
+            
+            if not raw_results:
+                ddg_results = await search_duckduckgo(
+                    query=body.query,
+                    max_results=settings.max_search_results,
+                )
+                raw_results.extend(ddg_results)
+            
+            if not raw_results:
+                yield f"data: {json.dumps({'type': 'error', 'content': 'No results found'})}\n\n"
+                return
+            
+            # Step 3: Rerank
+            ranked_results = await rerank_results(
+                query=body.query,
+                results=raw_results,
+                temporal_urgency=temporal_urgency,
+                max_results=body.max_results,
+            )
+            
+            # Step 4: Convert to SearchResult models
+            search_results = []
+            for result in ranked_results:
+                freshness = calculate_freshness_score(result.get("published_date"))
+                search_results.append(
+                    SearchResult(
+                        title=result.get("title", ""),
+                        url=result.get("url", ""),
+                        content=result.get("content", ""),
+                        score=result.get("score", 0.5),
+                        published_date=result.get("published_date"),
+                        freshness_score=freshness,
+                        authority_score=result.get("authority_score", 0.5),
+                    )
+                )
+            
+            # Send results first
+            results_data = {
+                "type": "results",
+                "results": [r.model_dump(mode="json") for r in search_results],
+                "temporal_context": temporal_context.model_dump(),
+            }
+            yield f"data: {json.dumps(results_data)}\n\n"
+            
+            # Step 5: Stream answer
+            yield f"data: {json.dumps({'type': 'answer_start'})}\n\n"
+            
+            async for chunk in synthesize_answer_stream(
+                query=body.query,
+                results=search_results,
+                temporal_context=temporal_context,
+            ):
+                yield f"data: {json.dumps({'type': 'answer_chunk', 'content': chunk})}\n\n"
+            
+            yield f"data: {json.dumps({'type': 'done'})}\n\n"
+            
+        except Exception as e:
+            yield f"data: {json.dumps({'type': 'error', 'content': str(e)})}\n\n"
+    
+    return StreamingResponse(
+        event_generator(),
+        media_type="text/event-stream",
+        headers={
+            "Cache-Control": "no-cache",
+            "Connection": "keep-alive",
+            "X-Accel-Buffering": "no",
+        },
+    )
+
+
+# === Deep Research Endpoints ===
+
+@router.post(
+    "/research/deep",
+    summary="Deep research with multi-dimensional analysis",
+    description="Decompose a query into dimensions, search each in parallel, and generate a comprehensive report.",
+)
+@limiter.limit("5/minute")
+async def deep_research(request: Request, body: DeepResearchRequest):
+    """
+    Run deep research with streaming progress updates.
+    
+    Returns SSE events:
+    - plan_ready: Research plan with dimensions
+    - dimension_start/complete: Progress per dimension
+    - report_chunk: Streaming report content
+    - done: Final summary
+    """
+    from app.agents.deep_research import run_deep_research
+    
+    return StreamingResponse(
+        run_deep_research(
+            query=body.query,
+            max_dimensions=body.max_dimensions,
+            max_sources_per_dim=body.max_sources_per_dim,
+            max_total_searches=body.max_total_searches,
+        ),
+        media_type="text/event-stream",
+        headers={
+            "Cache-Control": "no-cache",
+            "Connection": "keep-alive",
+            "X-Accel-Buffering": "no",
+        },
+    )
+
+
+@router.post(
+    "/search/heavy",
+    summary="Heavy search with content scraping",
+    description="Search with full content extraction from top sources for richer answers.",
+)
+@limiter.limit("10/minute")
+async def heavy_search(request: Request, body: SearchRequest):
+    """
+    Heavy search with content scraping.
+    
+    Scrapes full content from top results instead of just snippets,
+    providing richer context for answer generation.
+    """
+    from app.agents.heavy_search import run_heavy_search
+    
+    return StreamingResponse(
+        run_heavy_search(
+            query=body.query,
+            max_results=body.max_results,
+            max_scrape=5,
+            freshness=body.freshness,
+        ),
+        media_type="text/event-stream",
+        headers={
+            "Cache-Control": "no-cache",
+            "Connection": "keep-alive",
+            "X-Accel-Buffering": "no",
+        },
+    )
+
+
+@router.get(
+    "/images",
+    summary="Search for images",
+    description="Search for images related to a query using Brave Image Search.",
+)
+@limiter.limit("60/minute")
+async def image_search(request: Request, query: str, max_results: int = 6):
+    """
+    Search for images related to a query.
+    
+    Returns a list of image results with thumbnails and source URLs.
+    """
+    from app.sources.images import search_images
+    
+    if not query:
+        raise HTTPException(status_code=400, detail="Query is required")
+    
+    images = await search_images(query=query, max_results=max_results)
+    
+    return {"query": query, "images": images}
+
+
+# === SearXNG Search (pure - no LLM) ===
+
+@router.post(
+    "/search/searxng",
+    summary="Search using SearXNG + embedding reranking",
+    description="Uses SearXNG meta-search with embedding reranking. No LLM synthesis.",
+)
+@limiter.limit("20/minute")
+async def searxng_search(request: Request, body: SearchRequest):
+    """
+    Search using SearXNG with embedding reranking only.
+    
+    This endpoint uses your SearXNG instance for 50+ results
+    and reranks with embeddings. No LLM synthesis.
+    """
+    import json
+    from app.sources.searxng import search_searxng
+    from app.reranking.embeddings import compute_bi_encoder_scores
+    
+    async def event_generator():
+        try:
+            # Step 1: Search SearXNG
+            yield f"data: {json.dumps({'type': 'status', 'message': 'Searching SearXNG...'})}\n\n"
+            
+            time_range = {"day": "day", "week": "week", "month": "month"}.get(body.freshness)
+            raw_results = await search_searxng(
+                query=body.query,
+                max_results=50,
+                time_range=time_range,
+            )
+            
+            if not raw_results:
+                yield f"data: {json.dumps({'type': 'error', 'message': 'No results from SearXNG'})}\n\n"
+                return
+            
+            yield f"data: {json.dumps({'type': 'searxng_complete', 'count': len(raw_results)})}\n\n"
+            
+            # Step 2: Rerank with embeddings
+            yield f"data: {json.dumps({'type': 'status', 'message': 'Reranking with embeddings...'})}\n\n"
+            
+            docs = [f"{r.get('title', '')}. {r.get('content', '')[:500]}" for r in raw_results]
+            scores = compute_bi_encoder_scores(body.query, docs)
+            
+            for i, result in enumerate(raw_results):
+                result["embedding_score"] = scores[i]
+                orig_score = result.get("score", 0.5)
+                result["score"] = (scores[i] * 0.7) + (orig_score * 0.3)
+            
+            raw_results.sort(key=lambda x: x["score"], reverse=True)
+            final_results = raw_results[:body.max_results]
+            
+            # Step 3: Return results (no LLM)
+            yield f"data: {json.dumps({'type': 'results', 'results': [{'title': r.get('title'), 'url': r.get('url'), 'content': r.get('content', '')[:300], 'score': round(r.get('score', 0), 3), 'source': r.get('source')} for r in final_results]})}\n\n"
+            
+            yield f"data: {json.dumps({'type': 'done', 'total_raw': len(raw_results), 'returned': len(final_results)})}\n\n"
+            
+        except Exception as e:
+            yield f"data: {json.dumps({'type': 'error', 'message': str(e)})}\n\n"
+    
+    return StreamingResponse(
+        event_generator(),
+        media_type="text/event-stream",
+        headers={
+            "Cache-Control": "no-cache",
+            "Connection": "keep-alive",
+        },
+    )
+
+
+# === Code Search (GitHub, StackOverflow) ===
+
+@router.post(
+    "/search/code",
+    summary="Search code repositories and programming Q&A",
+    description="Uses SearXNG with GitHub, StackOverflow, and code-focused engines.",
+)
+@limiter.limit("20/minute")
+async def code_search(request: Request, body: SearchRequest):
+    """
+    Search for code, programming solutions, and documentation.
+    Uses GitHub, StackOverflow, GitLab, and other code-focused engines.
+    """
+    import json
+    from app.sources.searxng import search_searxng
+    from app.reranking.embeddings import compute_bi_encoder_scores
+    
+    async def event_generator():
+        try:
+            yield f"data: {json.dumps({'type': 'status', 'message': 'Searching code repositories...'})}\n\n"
+            
+            # Use code-specific engines
+            raw_results = await search_searxng(
+                query=body.query,
+                max_results=50,
+                categories=["it"],  # IT category includes code engines
+                engines=["github", "stackoverflow", "gitlab", "npm", "pypi", "crates.io", "packagist"],
+            )
+            
+            if not raw_results:
+                yield f"data: {json.dumps({'type': 'error', 'message': 'No code results found'})}\n\n"
+                return
+            
+            yield f"data: {json.dumps({'type': 'search_complete', 'count': len(raw_results)})}\n\n"
+            
+            # Rerank with embeddings
+            yield f"data: {json.dumps({'type': 'status', 'message': 'Ranking by relevance...'})}\n\n"
+            
+            docs = [f"{r.get('title', '')}. {r.get('content', '')[:500]}" for r in raw_results]
+            scores = compute_bi_encoder_scores(body.query, docs)
+            
+            for i, result in enumerate(raw_results):
+                result["embedding_score"] = scores[i]
+                orig_score = result.get("score", 0.5)
+                result["score"] = (scores[i] * 0.7) + (orig_score * 0.3)
+            
+            raw_results.sort(key=lambda x: x["score"], reverse=True)
+            final_results = raw_results[:body.max_results]
+            
+            yield f"data: {json.dumps({'type': 'results', 'results': [{'title': r.get('title'), 'url': r.get('url'), 'content': r.get('content', '')[:300], 'score': round(r.get('score', 0), 3), 'source': r.get('source')} for r in final_results]})}\n\n"
+            yield f"data: {json.dumps({'type': 'done', 'total_raw': len(raw_results), 'returned': len(final_results)})}\n\n"
+            
+        except Exception as e:
+            yield f"data: {json.dumps({'type': 'error', 'message': str(e)})}\n\n"
+    
+    return StreamingResponse(
+        event_generator(),
+        media_type="text/event-stream",
+        headers={"Cache-Control": "no-cache", "Connection": "keep-alive"},
+    )
+
+
+# === Academic Search (arXiv, Google Scholar) ===
+
+@router.post(
+    "/search/academic",
+    summary="Search academic papers and research",
+    description="Uses SearXNG with arXiv, Google Scholar, Semantic Scholar, and academic engines.",
+)
+@limiter.limit("20/minute")
+async def academic_search(request: Request, body: SearchRequest):
+    """
+    Search for academic papers, research, and scientific content.
+    Uses arXiv, Google Scholar, Semantic Scholar, PubMed, and other academic engines.
+    """
+    import json
+    from app.sources.searxng import search_searxng
+    from app.reranking.embeddings import compute_bi_encoder_scores
+    
+    async def event_generator():
+        try:
+            yield f"data: {json.dumps({'type': 'status', 'message': 'Searching academic sources...'})}\n\n"
+            
+            # Use academic engines
+            raw_results = await search_searxng(
+                query=body.query,
+                max_results=50,
+                categories=["science"],
+                engines=["arxiv", "google scholar", "semantic scholar", "pubmed", "base", "crossref"],
+            )
+            
+            if not raw_results:
+                yield f"data: {json.dumps({'type': 'error', 'message': 'No academic results found'})}\n\n"
+                return
+            
+            yield f"data: {json.dumps({'type': 'search_complete', 'count': len(raw_results)})}\n\n"
+            
+            # Rerank with embeddings
+            yield f"data: {json.dumps({'type': 'status', 'message': 'Ranking by relevance...'})}\n\n"
+            
+            docs = [f"{r.get('title', '')}. {r.get('content', '')[:500]}" for r in raw_results]
+            scores = compute_bi_encoder_scores(body.query, docs)
+            
+            for i, result in enumerate(raw_results):
+                result["embedding_score"] = scores[i]
+                orig_score = result.get("score", 0.5)
+                result["score"] = (scores[i] * 0.7) + (orig_score * 0.3)
+            
+            raw_results.sort(key=lambda x: x["score"], reverse=True)
+            final_results = raw_results[:body.max_results]
+            
+            yield f"data: {json.dumps({'type': 'results', 'results': [{'title': r.get('title'), 'url': r.get('url'), 'content': r.get('content', '')[:300], 'score': round(r.get('score', 0), 3), 'source': r.get('source')} for r in final_results]})}\n\n"
+            yield f"data: {json.dumps({'type': 'done', 'total_raw': len(raw_results), 'returned': len(final_results)})}\n\n"
+            
+        except Exception as e:
+            yield f"data: {json.dumps({'type': 'error', 'message': str(e)})}\n\n"
+    
+    return StreamingResponse(
+        event_generator(),
+        media_type="text/event-stream",
+        headers={"Cache-Control": "no-cache", "Connection": "keep-alive"},
+    )
+
+
+# === Browser Agent ===
+
+@router.post(
+    "/agent/browse",
+    summary="Browser agent - navigate and extract from websites",
+    description="Uses E2B sandbox. stream_visual=true runs the visual Chrome agent, false runs the stealth Camoufox agent.",
+)
+@limiter.limit("10/minute")
+async def browser_agent(request: Request, body: BrowseRequest):
+    """
+    Browser agent with two modes:
+    - stream_visual=true: visual Chrome agent with live video stream
+    - stream_visual=false: stealth Camoufox headless agent
+    """
+    
+    async def event_generator():
+        try:
+            if body.stream_visual:
+                from app.agents.browser_visual import run_browser_visual_agent
+                async for event in run_browser_visual_agent(body.task, body.url):
+                    yield f"data: {json.dumps(event)}\n\n"
+            else:
+                from app.agents.browser_stealth import run_browser_stealth_agent
+                async for event in run_browser_stealth_agent(body.task, body.url):
+                    yield f"data: {json.dumps(event)}\n\n"
+        except Exception as e:
+            yield f"data: {json.dumps({'type': 'error', 'message': str(e)})}\n\n"
+    
+    return StreamingResponse(
+        event_generator(),
+        media_type="text/event-stream",
+        headers={"Cache-Control": "no-cache", "Connection": "keep-alive"},
+    )
+