Agentic Search

# agentic-search secrets template
#
# Copy this file to .env.search.local in this same directory and fill in real
# values. scripts/_http.py auto-loads it via a relative-path walk-up from its
# own __file__ location, so the skill is fully portable — drop agentic-search/
# anywhere and it just works. No absolute paths, no shell sourcing.
# .env.search.local is gitignored.
#
# Precedence: shell exports always win over file values, so an `export
# GROK_API_KEY=...` in your shell will override whatever's in this file.

# ── Required: Grok provider (xAI direct, Responses API) ─────────────────────
# v3 targets xAI's Responses API at /v1/responses with the web_search tool.
# Set GROK_API_URL to the xAI v1 base URL — the script appends /responses.
# (The legacy /chat/completions path with search_parameters is deprecated and
# returns 410 Gone; the skill no longer supports it.)
GROK_API_URL=https://api.x.ai/v1
GROK_API_KEY=

# Optional model override. Leave unset to use the built-in default:
# grok-4-1-fast-reasoning. See references/provider-quirks.md before changing.
# GROK_MODEL=grok-4-1-fast-reasoning

# ── Optional: Tavily provider ────────────────────────────────────────────────
# Used for: agentic_search --extra-sources (supplementary URLs),
#           agentic_fetch (primary extractor),
#           agentic_map (REQUIRED — no fallback).
TAVILY_API_KEY=
# TAVILY_API_URL=https://api.tavily.com

# ── Optional: Firecrawl provider ─────────────────────────────────────────────
# Used for: agentic_search --extra-sources (breadth-prioritized when both
#           Tavily and Firecrawl keys are set),
#           agentic_fetch (fallback when Tavily fails or returns empty).
FIRECRAWL_API_KEY=
# FIRECRAWL_API_URL=https://api.firecrawl.dev/v2

# ── Optional: retry tuning ───────────────────────────────────────────────────
# GROK_RETRY_MAX_ATTEMPTS=3      # number of retries (so 3 = 4 total attempts)
# GROK_RETRY_MULTIPLIER=1        # exponential backoff multiplier
# GROK_RETRY_MAX_WAIT=10         # cap on wait between retries (seconds)

# ── Optional: debug logging ──────────────────────────────────────────────────
# Set to true/1/yes to log diagnostic info to stderr (stdout stays parseable).
# GROK_DEBUG=true

# Environment files
.env
.env.*
*.env
*.env.*
!.env.template
!.env.*.template
!*.env.template

# Python bytecode
__pycache__/
*.pyc

"""Shared HTTP utilities for agentic-search scripts.

Provides:
- Async retry wrapper with Retry-After header parsing.
- Env var helpers for config (mirroring upstream GrokSearch v1.9.2 surface).
- Debug logging to stderr (gated by GROK_DEBUG).

Pure stdlib + httpx + tenacity. No fastmcp, no pydantic, no upstream package.
"""

from __future__ import annotations

import json
import os
import sys
from datetime import datetime, timezone
from email.utils import parsedate_to_datetime
from pathlib import Path
from typing import Optional

import httpx
from tenacity import (
    AsyncRetrying,
    retry_if_exception,
    stop_after_attempt,
    wait_random_exponential,
)
from tenacity.wait import wait_base


# ---------- dotenv loader (no python-dotenv dependency) ----------

_DOTENV_FILENAME = ".env.search.local"


def _load_dotenv() -> Optional[Path]:
    """Walk up from this file's directory looking for .env.search.local and
    load KEY=value pairs into os.environ. Existing env vars take precedence,
    so shell exports always win over file values. Returns the loaded path
    (for debug logging) or None if no file was found."""
    here = Path(__file__).resolve().parent
    candidates = [here] + list(here.parents)[:6]  # cap walk depth
    for parent in candidates:
        candidate = parent / _DOTENV_FILENAME
        if not candidate.is_file():
            continue
        try:
            for raw in candidate.read_text(encoding="utf-8").splitlines():
                line = raw.strip()
                if not line or line.startswith("#"):
                    continue
                if "=" not in line:
                    continue
                key, _, value = line.partition("=")
                key = key.strip()
                # strip surrounding quotes (single or double) if matched
                value = value.strip()
                if len(value) >= 2 and value[0] == value[-1] and value[0] in ("'", '"'):
                    value = value[1:-1]
                if key and key not in os.environ:
                    os.environ[key] = value
        except OSError:
            return None
        return candidate
    return None


_LOADED_DOTENV = _load_dotenv()


# ---------- env / config ----------

RETRYABLE_STATUS_CODES = {408, 429, 500, 502, 503, 504}


def env_str(name: str, default: Optional[str] = None) -> Optional[str]:
    val = os.getenv(name)
    return val if val is not None else default


def env_bool(name: str, default: bool = False) -> bool:
    val = os.getenv(name)
    if val is None:
        return default
    return val.lower() in ("true", "1", "yes")


def env_int(name: str, default: int) -> int:
    try:
        return int(os.getenv(name, str(default)))
    except (TypeError, ValueError):
        return default


def env_float(name: str, default: float) -> float:
    try:
        return float(os.getenv(name, str(default)))
    except (TypeError, ValueError):
        return default


def debug_enabled() -> bool:
    return env_bool("GROK_DEBUG", False)


def debug(msg: str) -> None:
    if debug_enabled():
        print(f"[agentic-search] {msg}", file=sys.stderr, flush=True)


def _emit_dotenv_debug_once() -> None:
    if debug_enabled() and _LOADED_DOTENV is not None:
        # Print once at first debug call so we know which file was sourced.
        print(f"[agentic-search] loaded dotenv: {_LOADED_DOTENV}", file=sys.stderr, flush=True)


_emit_dotenv_debug_once()


def retry_max_attempts() -> int:
    return env_int("GROK_RETRY_MAX_ATTEMPTS", 3)


def retry_multiplier() -> float:
    return env_float("GROK_RETRY_MULTIPLIER", 1.0)


def retry_max_wait() -> int:
    return env_int("GROK_RETRY_MAX_WAIT", 10)


# ---------- Grok config ----------

def grok_api_url() -> str:
    url = env_str("GROK_API_URL")
    if not url:
        raise SystemExit(
            "error: GROK_API_URL is not set. "
            "Set it to your OpenAI-compatible endpoint (e.g. https://api.x.ai/v1)."
        )
    return url


def grok_api_key() -> str:
    key = env_str("GROK_API_KEY")
    if not key:
        raise SystemExit("error: GROK_API_KEY is not set.")
    return key


def grok_model(override: Optional[str] = None) -> str:
    model = override or env_str("GROK_MODEL") or "grok-4-1-fast-reasoning"
    # OpenRouter requires :online suffix for live web access
    try:
        url = grok_api_url()
    except SystemExit:
        return model
    if "openrouter" in url and ":online" not in model:
        return f"{model}:online"
    return model


# ---------- Tavily config ----------

def tavily_api_url() -> str:
    return env_str("TAVILY_API_URL", "https://api.tavily.com") or "https://api.tavily.com"


def tavily_api_key() -> Optional[str]:
    return env_str("TAVILY_API_KEY")


# ---------- Firecrawl config ----------

def firecrawl_api_url() -> str:
    return env_str("FIRECRAWL_API_URL", "https://api.firecrawl.dev/v2") or "https://api.firecrawl.dev/v2"


def firecrawl_api_key() -> Optional[str]:
    return env_str("FIRECRAWL_API_KEY")


# ---------- retry strategy ----------

def is_retryable_exception(exc: BaseException) -> bool:
    if isinstance(exc, (httpx.TimeoutException, httpx.NetworkError, httpx.ConnectError, httpx.RemoteProtocolError)):
        return True
    if isinstance(exc, httpx.HTTPStatusError):
        return exc.response.status_code in RETRYABLE_STATUS_CODES
    return False


class WaitWithRetryAfter(wait_base):
    """Honor Retry-After on 429s, otherwise exponential backoff with a flat
    +3s bump on RemoteProtocolError to give the connection time to recover."""

    def __init__(self, multiplier: float, max_wait: int):
        self._base_wait = wait_random_exponential(multiplier=multiplier, max=max_wait)
        self._protocol_error_base = 3.0

    def __call__(self, retry_state):
        if retry_state.outcome and retry_state.outcome.failed:
            exc = retry_state.outcome.exception()
            if isinstance(exc, httpx.HTTPStatusError) and exc.response.status_code == 429:
                retry_after = self._parse_retry_after(exc.response)
                if retry_after is not None:
                    return retry_after
            if isinstance(exc, httpx.RemoteProtocolError):
                return self._base_wait(retry_state) + self._protocol_error_base
        return self._base_wait(retry_state)

    @staticmethod
    def _parse_retry_after(response: httpx.Response) -> Optional[float]:
        header = response.headers.get("Retry-After")
        if not header:
            return None
        header = header.strip()
        if header.isdigit():
            return float(header)
        try:
            retry_dt = parsedate_to_datetime(header)
            if retry_dt.tzinfo is None:
                retry_dt = retry_dt.replace(tzinfo=timezone.utc)
            delay = (retry_dt - datetime.now(timezone.utc)).total_seconds()
            return max(0.0, delay)
        except (TypeError, ValueError):
            return None


def make_retrying() -> AsyncRetrying:
    return AsyncRetrying(
        stop=stop_after_attempt(retry_max_attempts() + 1),
        wait=WaitWithRetryAfter(retry_multiplier(), retry_max_wait()),
        retry=retry_if_exception(is_retryable_exception),
        reraise=True,
    )


# ---------- xAI Responses API (v3 — primary Grok call path) ----------
#
# As of January 2026, xAI deprecated the `search_parameters` field on the
# `/v1/chat/completions` endpoint and migrated live web search to the new
# `/v1/responses` endpoint with server-side `web_search` and `x_search` tools.
# All Grok-mediated scripts in this skill (search, fetch, extract, rank)
# call into `grok_responses_call` below to get real grounded citations.
#
# Docs: https://docs.x.ai/docs/guides/live-search
#       https://docs.x.ai/docs/guides/tools/search-tools


async def parse_responses_streaming(
    response: httpx.Response,
) -> tuple[str, list[dict]]:
    """Parse xAI Responses API SSE stream.

    Returns (accumulated_text, annotations).

    Each SSE line is `data: {"type": "<event>", ...}`. Event types handled:
        response.created                      → ignored (start marker)
        response.output_item.added            → ignored (item begin)
        response.output_text.delta            → append `delta` field to text
        response.output_text.annotation.added → append `annotation` to citations
        response.completed                    → terminal (normal exit)
        response.incomplete                   → terminal (capped/cut off)
        response.failed                       → terminal (raise httpx.HTTPError)
        all other types                       → ignored (forward-compat)

    Falls back to non-streaming JSON parse if no events parse — some proxies
    buffer the full body into one chunk. The fallback walks the standard
    Responses API response shape: output[].content[].text + annotations[].

    Mirrors xAI's documented streaming envelope (which itself mirrors OpenAI's
    Responses API streaming events). On `response.failed`, raises httpx.HTTPError
    so the caller's `make_retrying()` loop can decide whether to retry.
    """
    content = ""
    annotations: list[dict] = []
    saw_event = False
    full_body: list[str] = []

    async for line in response.aiter_lines():
        line = line.strip()
        if not line:
            continue
        full_body.append(line)
        if not line.startswith("data:"):
            continue
        payload = line[5:].lstrip()
        if not payload or payload == "[DONE]":
            continue
        try:
            event = json.loads(payload)
        except json.JSONDecodeError:
            continue
        if not isinstance(event, dict):
            continue

        event_type = event.get("type", "")
        saw_event = True

        if event_type == "response.output_text.delta":
            delta = event.get("delta")
            if isinstance(delta, str):
                content += delta
            continue

        if event_type == "response.output_text.annotation.added":
            ann = event.get("annotation")
            if isinstance(ann, dict):
                annotations.append(ann)
            continue

        if event_type in ("response.completed", "response.incomplete"):
            # Terminal events; loop will end naturally as the stream closes
            continue

        if event_type == "response.failed":
            err = event.get("error") or {}
            err_msg = err.get("message") if isinstance(err, dict) else str(err)
            raise httpx.HTTPError(f"xAI Responses API stream failed: {err_msg}")

        # Other event types (response.created, response.output_item.added,
        # response.content_part.added/done, etc.) are ignored — we don't need
        # them for text+citation extraction. Forward-compat by design.

    # Fallback: no SSE events parsed (proxy buffered body, or non-streaming
    # response was returned despite stream=true). Walk the documented response
    # envelope: {output: [{type, content: [{type: "output_text", text, annotations}]}]}.
    if not saw_event and full_body:
        try:
            data = json.loads("".join(full_body))
        except json.JSONDecodeError:
            return content, annotations
        if not isinstance(data, dict):
            return content, annotations
        for item in data.get("output", []) or []:
            if not isinstance(item, dict) or item.get("type") != "message":
                continue
            for part in item.get("content", []) or []:
                if not isinstance(part, dict):
                    continue
                if part.get("type") == "output_text":
                    text = part.get("text")
                    if isinstance(text, str):
                        content += text
                    for ann in part.get("annotations", []) or []:
                        if isinstance(ann, dict):
                            annotations.append(ann)

    return content, annotations


# Recognized auxiliary fields on annotation objects, copied through into our
# normalized source dict shape. The `url` field is required; everything else
# is optional. We deliberately accept multiple synonyms for the same concept
# (e.g. snippet/summary/description) so we're robust to xAI/OpenAI naming drift.
_ANNOTATION_FIELD_ALIASES = {
    "title": "title",
    "snippet": "description",
    "summary": "description",
    "description": "description",
    "source": "source",
    "published_date": "published_date",
    "publishedAt": "published_date",
    "date": "published_date",
}


def normalize_responses_annotations(annotations: list[dict]) -> list[dict]:
    """Convert Responses API annotation objects to our standard source dict shape.

    Permissive parser:
      - Skips entries that aren't dicts
      - Skips entries without a `url` field starting with http(s)://
      - Strips trailing punctuation from URLs (matches `_extract_unique_urls`)
      - Dedupes by URL across the input list (first-seen wins)
      - Copies recognized auxiliary fields via `_ANNOTATION_FIELD_ALIASES`
      - Tags every result with `provider="grok-web-search"` so downstream
        logic can distinguish native Grok citations from Tavily/Firecrawl extras

    Returns a list of `{url, title?, description?, source?, published_date?, provider}`
    dicts. Output schema matches the upstream `_normalize_sources` shape so
    downstream merging/dedupe code is unchanged.
    """
    out: list[dict] = []
    seen: set[str] = set()
    for ann in annotations or []:
        if not isinstance(ann, dict):
            continue
        url = ann.get("url")
        if not isinstance(url, str):
            continue
        url = url.strip().rstrip(".,;:!?")
        if not url.startswith(("http://", "https://")):
            continue
        if url in seen:
            continue
        seen.add(url)
        rec: dict = {"url": url, "provider": "grok-web-search"}
        for src_key, dst_key in _ANNOTATION_FIELD_ALIASES.items():
            v = ann.get(src_key)
            if isinstance(v, str) and v.strip():
                rec[dst_key] = v.strip()
        out.append(rec)
    return out


# ---------- xAI Responses API (v3 — primary Grok call path) ----------
#
# As of January 2026, xAI deprecated the `search_parameters` field on the
# `/v1/chat/completions` endpoint and migrated live web search to the new
# `/v1/responses` endpoint with server-side `web_search` and `x_search` tools.
# All Grok-mediated scripts in this skill (search, fetch, extract, rank)
# call into `grok_responses_call` below to get real grounded citations.
#
# Docs: https://docs.x.ai/docs/guides/live-search
#       https://docs.x.ai/docs/guides/tools/search-tools


async def grok_responses_call(
    *,
    instructions: str,
    user_content: str,
    model: Optional[str] = None,
    enable_search: bool = True,
    search_mode: str = "auto",
    tools: Optional[list[str]] = None,
    allowed_domains: Optional[list[str]] = None,
    excluded_domains: Optional[list[str]] = None,
    timeout_read: float = 120.0,
) -> tuple[str, list[dict]]:
    """POST to xAI Responses API (`/v1/responses`) with streaming.

    Returns ``(text_content, annotations)`` where annotations is the list of
    raw Responses-API annotation objects (use `normalize_responses_annotations`
    to convert into our source-dict shape).

    Single shared helper used by all 4 Grok call sites:
      - agentic_search.py:_grok_search          (tools=["web_search", "x_search"])
      - agentic_fetch.py:_grok_fetch            (tools=["web_search"])
      - agentic_extract.py:describe_url         (tools=["web_search"])
      - agentic_rank.py:rank_sources            (enable_search=False)

    The `instructions` field is the preferred way to set the system prompt
    on the Responses API (cleaner than embedding role=system in input[]).

    Live search tools are specified via the `tools` parameter:
      - `tools=["web_search"]` — search the web and browse pages (default)
      - `tools=["x_search"]` — search X/Twitter posts, users, threads
      - `tools=["web_search", "x_search"]` — both (recommended for research)

    The legacy `search_parameters` field has been deprecated by xAI
    (returns 410 Gone with error: "Live search is deprecated. Please switch
    to the Agent Tools API"), so we use tools exclusively.

    The `search_mode` argument is a soft hint preserved for callers' clarity
    of intent — it does NOT map to a request field anymore.

    Domain filters: `allowed_domains` and `excluded_domains` are mutually
    exclusive per xAI docs. We pass at most one. Both `None` means no filter.

    Auth, retry strategy, base URL, and model resolution all reuse the
    existing helpers in this module — no duplicate config plumbing.
    """
    api_url = grok_api_url()
    api_key = grok_api_key()
    effective_model = grok_model(model)

    headers = {
        "Authorization": f"Bearer {api_key}",
        "Content-Type": "application/json",
    }

    payload: dict = {
        "model": effective_model,
        "instructions": instructions,
        "input": [
            {"role": "user", "content": user_content},
        ],
        "stream": True,
    }

    if enable_search:
        # Build the tools array. Default to web_search only for backward compat.
        if tools is None:
            tools = ["web_search"]
        
        # Validate tool types
        valid_tools = {"web_search", "x_search"}
        for t in tools:
            if t not in valid_tools:
                raise ValueError(f"Invalid tool type: {t}. Valid options: {valid_tools}")
        
        # Build tools payload
        tools_payload = [{"type": t} for t in tools]
        
        # Apply filters only to web_search (xAI requires filters on specific tool)
        if "web_search" in tools and (allowed_domains or excluded_domains):
            if allowed_domains and excluded_domains:
                debug(
                    "grok_responses_call: both allowed_domains and excluded_domains "
                    "set; allowed_domains wins (xAI requires they be mutually exclusive)"
                )
            # Find the web_search tool and add filters
            for tool in tools_payload:
                if tool["type"] == "web_search":
                    if allowed_domains:
                        tool["filters"] = {"allowed_domains": list(allowed_domains)[:5]}
                    elif excluded_domains:
                        tool["filters"] = {"excluded_domains": list(excluded_domains)[:5]}
                    break
        
        payload["tools"] = tools_payload

    debug(
        f"grok_responses_call: model={effective_model} "
        f"search={enable_search} mode={search_mode if enable_search else 'n/a'} "
        f"input_chars={len(user_content)}"
    )

    timeout = httpx.Timeout(connect=6.0, read=timeout_read, write=10.0, pool=None)
    async with httpx.AsyncClient(timeout=timeout, follow_redirects=True) as client:
        async for attempt in make_retrying():
            with attempt:
                async with client.stream(
                    "POST",
                    f"{api_url.rstrip('/')}/responses",
                    headers=headers,
                    json=payload,
                ) as response:
                    response.raise_for_status()
                    text, annotations = await parse_responses_streaming(response)
                    if annotations:
                        # One-time debug dump of the first annotation shape so
                        # we can refine the field aliases if xAI uses unexpected
                        # field names. Cheap; only fires when GROK_DEBUG=true.
                        debug(f"grok_responses_call: first annotation shape: {annotations[0]}")
                    return text, annotations

    return "", []

"""Centralized prompts for agentic-search scripts.

All prompts are ported byte-for-byte from upstream GrokSearch
(`GrokSearch/src/grok_search/utils.py`) so the skill remains
behavior-compatible with the upstream MCP server. Centralizing them here
lets multiple scripts (`agentic_search`, `agentic_fetch --engine grok`,
`agentic_extract`, `agentic_rank`) share a single source of truth.
"""

from __future__ import annotations

# ---------------------------------------------------------------------------
# SEARCH_PROMPT
# Source: GrokSearch/src/grok_search/utils.py:209-241
# Used by: agentic_search.py (Grok system prompt for the main search call)
# ---------------------------------------------------------------------------

SEARCH_PROMPT = """
# Core Instruction

1. User needs may be vague. Think divergently, infer intent from multiple angles, and leverage full conversation context to progressively clarify their true needs.
2. **Breadth-First Search**—Approach problems from multiple dimensions. Brainstorm 5+ perspectives and execute parallel searches for each. Consult as many high-quality sources as possible before responding.
3. **Depth-First Search**—After broad exploration, select ≥2 most relevant perspectives for deep investigation into specialized knowledge.
4. **Evidence-Based Reasoning & Traceable Sources**—Every claim must be followed by a citation (`citation_card` format). More credible sources strengthen arguments. If no references exist, remain silent.
5. Before responding, ensure full execution of Steps 1–4.

---

# Search Instruction

1. Think carefully before responding—anticipate the user's true intent to ensure precision.
2. Verify every claim rigorously to avoid misinformation.
3. Follow problem logic—dig deeper until clues are exhaustively clear. If a question seems simple, still infer broader intent and search accordingly. Use multiple parallel tool calls per query and ensure answers are well-sourced.
4. Search in English first (prioritizing English resources for volume/quality), but switch to Chinese if context demands.
5. Prioritize authoritative sources: Wikipedia, academic databases, books, reputable media/journalism.
6. Favor sharing in-depth, specialized knowledge over generic or common-sense content.

---

# Output Style

0. **Be direct—no unnecessary follow-ups**.
1. Lead with the **most probable solution** before detailed analysis.
2. **Define every technical term** in plain language (annotate post-paragraph).
3. Explain expertise **simply yet profoundly**.
4. **Respect facts and search results—use statistical rigor to discern truth**.
5. **Every sentence must cite sources** (`citation_card`). More references = stronger credibility. Silence if uncited.
6. Expand on key concepts—after proposing solutions, **use real-world analogies** to demystify technical terms.
7. **Strictly format outputs in polished Markdown** (LaTeX for formulas, code blocks for scripts, etc.).
"""


# ---------------------------------------------------------------------------
# FETCH_PROMPT
# Source: GrokSearch/src/grok_search/utils.py:80-186
# Used by: agentic_fetch.py --engine grok (Grok-mediated full-page extraction)
# ---------------------------------------------------------------------------

FETCH_PROMPT = """
# Profile: Web Content Fetcher

- **Language**: 中文
- **Role**: 你是一个专业的网页内容抓取和解析专家，获取指定 URL 的网页内容，并将其转换为与原网页高度一致的结构化 Markdown 文本格式。

---

## Workflow

### 1. URL 验证与内容获取
- 验证 URL 格式有效性，检查可访问性（处理重定向/超时）
- **关键**：优先识别页面目录/大纲结构（Table of Contents），作为内容抓取的导航索引
- 全量获取 HTML 内容，确保不遗漏任何章节或动态加载内容

### 2. 智能解析与内容提取
- **结构优先**：若存在目录/大纲，严格按其层级结构进行内容提取和组织
- 解析 HTML 文档树，识别所有内容元素：
  - 标题层级（h1-h6）及其嵌套关系
  - 正文段落、文本格式（粗体/斜体/下划线）
  - 列表结构（有序/无序/嵌套）
  - 表格（包含表头/数据行/合并单元格）
  - 代码块（行内代码/多行代码块/语言标识）
  - 引用块、分隔线
  - 图片（src/alt/title 属性）
  - 链接（内部/外部/锚点）

### 3. 内容清理与语义保留
- 移除非内容标签：`<script>`、`<style>`、`<iframe>`、`<noscript>`
- 过滤干扰元素：广告模块、追踪代码、社交分享按钮
- **保留语义信息**：图片 alt/title、链接 href/title、代码语言标识
- 特殊模块标注：导航栏、侧边栏、页脚用特殊标记保留

---

## Skills

### 1. 内容精准提取与还原
- **如果存在目录或者大纲，则按照目录或者大纲的结构进行提取**
- **完整保留原始内容结构**，不遗漏任何信息
- **准确识别并提取**标题、段落、列表、表格、代码块等所有元素
- **保持原网页的内容层次和逻辑关系**
- **精确处理特殊字符**，确保无乱码和格式错误
- **还原文本内容**，包括换行、缩进、空格等细节

### 2. 结构化组织与呈现
- **标题层级**：使用 `#`、`##`、`###` 等还原标题层级
- **目录结构**：使用列表生成 Table of Contents，带锚点链接
- **内容分区**：使用 `###` 或代码块（` ```section ``` `）明确划分 Section
- **嵌套结构**：使用缩进列表或引用块（`>`）保持层次关系
- **辅助模块**：侧边栏、导航等用特殊代码块（` ```sidebar ``` `、` ```nav ``` `）包裹

### 3. 格式转换优化
- **HTML 转 Markdown**：保持 100% 内容一致性
- **表格处理**：使用 Markdown 表格语法（`|---|---|`）
- **代码片段**：用 ` ```语言标识``` ` 包裹，保留原始缩进
- **图片处理**：转换为 `![alt](url)` 格式，保留所有属性
- **链接处理**：转换为 `[文本](URL)` 格式，保持完整路径
- **强调样式**：`<strong>` → `**粗体**`，`<em>` → `*斜体*`

### 4. 内容完整性保障
- **零删减原则**：不删减任何原网页文本内容
- **元数据保留**：保留时间戳、作者信息、标签等关键信息
- **多媒体标注**：视频、音频以链接或占位符标注（`[视频: 标题](URL)`）
- **动态内容处理**：尽可能抓取完整内容

---

## Rules

### 1. 内容一致性原则（核心）
- ✅ 返回内容必须与原网页内容**完全一致**，不能有信息缺失
- ✅ 保持原网页的**所有文本、结构和语义信息**
- ❌ **不进行**内容摘要、精简、改写或总结
- ✅ 保留原始的**段落划分、换行、空格**等格式细节

### 2. 格式转换标准
| HTML | Markdown | 示例 |
|------|----------|------|
| `<h1>`-`<h6>` | `#`-`######` | `# 标题` |
| `<strong>` | `**粗体**` | **粗体** |
| `<em>` | `*斜体*` | *斜体* |
| `<a>` | `[文本](url)` | [链接](url) |
| `<img>` | `![alt](url)` | ![图](url) |
| `<code>` | `` `代码` `` | `code` |
| `<pre><code>` | ` ```\n代码\n``` ` | 代码块 |

### 3. 输出质量要求
- **元数据头部**：
  ```markdown
  ---
  source: [原始URL]
  title: [网页标题]
  fetched_at: [抓取时间]
  ---
  ```
- **编码标准**：统一使用 UTF-8
- **可用性**：输出可直接用于文档生成或阅读

---

## Initialization

当接收到 URL 时：
1. 按 Workflow 执行抓取和处理
2. 返回完整的结构化 Markdown 文档
"""


# ---------------------------------------------------------------------------
# URL_DESCRIBE_PROMPT
# Source: GrokSearch/src/grok_search/utils.py:189-200
# Used by: agentic_extract.py (LLM-mediated page → title + verbatim quotes)
# ---------------------------------------------------------------------------

URL_DESCRIBE_PROMPT = (
    "Browse the given URL. Return exactly two sections:\n\n"
    "Title: <page title from the page's own <title> tag or top heading; "
    "if missing/generic, craft one using key terms found in the page>\n\n"
    "Extracts: <copy 2-4 verbatim fragments from the page that best represent "
    "its core content. Each fragment must be the author's original words, "
    "wrapped in quotes, separated by ' | '. "
    "Do NOT paraphrase, rephrase, interpret, or describe. "
    "Do NOT write sentences like 'This page discusses...' or 'The author argues...'. "
    "You are a copy-paste machine.>\n\n"
    "Nothing else."
)


# ---------------------------------------------------------------------------
# RANK_SOURCES_PROMPT
# Source: GrokSearch/src/grok_search/utils.py:202-207
# Used by: agentic_rank.py (rerank a numbered source list by query relevance)
# ---------------------------------------------------------------------------

RANK_SOURCES_PROMPT = (
    "Given a user query and a numbered source list, output ONLY the source numbers "
    "reordered by relevance to the query (most relevant first). "
    "Format: space-separated integers on a single line (e.g., 14 12 1 3 5). "
    "Include every number exactly once. Nothing else."
)

"""Disk-backed session cache for agentic-search.

Stores search sessions as individual JSON files in the system temp directory,
under `agentic-search/sessions/`. Mirrors the LRU semantics of upstream
GrokSearch's in-memory `SourcesCache` (sources.py:32-51) — 256 sessions max,
oldest evicted by mtime when the cap is exceeded.

Pure stdlib. No locks (each session_id is unique; concurrent rewrites of the
same session are rare and tolerated by atomic write-then-rename). No TTL —
size-based eviction only, matching upstream.

Public surface:
    new_session_id() -> str
    session_dir() -> Path
    write_session(session_id, data) -> Path
    read_session(session_id) -> dict | None
    list_sessions() -> list[dict]
    prune_sessions(max_count=256) -> int
"""

from __future__ import annotations

import json
import os
import tempfile
import uuid
from pathlib import Path
from typing import Optional

_DIR_NAME = "agentic-search"
_SUBDIR_NAME = "sessions"
_DEFAULT_MAX = 256


def new_session_id() -> str:
    """12-char hex session id. Matches upstream sources.py:28-29 exactly."""
    return uuid.uuid4().hex[:12]


def session_dir() -> Path:
    """Return (and create if missing) the disk session cache directory.

    Path: tempfile.gettempdir() / agentic-search / sessions /
    On macOS this is typically /var/folders/.../T/agentic-search/sessions/.
    On Linux: /tmp/agentic-search/sessions/.
    """
    base = Path(tempfile.gettempdir()) / _DIR_NAME / _SUBDIR_NAME
    base.mkdir(parents=True, exist_ok=True)
    return base


def _session_path(session_id: str) -> Path:
    return session_dir() / f"{session_id}.json"


def write_session(session_id: str, data: dict) -> Path:
    """Atomically write session JSON to disk and return its path.

    Uses write-to-temp + os.replace for atomicity, so a crashed write never
    leaves a partial file. Existing sessions with the same id are overwritten.
    """
    path = _session_path(session_id)
    tmp = path.with_suffix(".json.tmp")
    payload = json.dumps(data, ensure_ascii=False, indent=2)
    tmp.write_text(payload, encoding="utf-8")
    os.replace(tmp, path)
    return path


def read_session(session_id: str) -> Optional[dict]:
    """Return the session dict or None if not found / unreadable."""
    path = _session_path(session_id)
    if not path.is_file():
        return None
    try:
        return json.loads(path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError):
        return None


def list_sessions() -> list[dict]:
    """Return a list of session diagnostics dicts, most-recent first.

    Each entry: {session_id, mtime, query?, model?, sources_count?}.
    Best-effort: corrupted sessions are listed with only the session_id and
    mtime; their query/model/sources_count fields are omitted.
    """
    base = session_dir()
    entries = []
    for path in base.glob("*.json"):
        if path.suffix.endswith(".tmp"):
            continue
        try:
            stat = path.stat()
        except OSError:
            continue
        entry: dict = {
            "session_id": path.stem,
            "mtime": stat.st_mtime,
        }
        try:
            data = json.loads(path.read_text(encoding="utf-8"))
            if isinstance(data, dict):
                if "query" in data:
                    entry["query"] = data["query"]
                if "model" in data:
                    entry["model"] = data["model"]
                if "sources" in data and isinstance(data["sources"], list):
                    entry["sources_count"] = len(data["sources"])
                if "created_at" in data:
                    entry["created_at"] = data["created_at"]
        except (OSError, json.JSONDecodeError):
            pass
        entries.append(entry)
    entries.sort(key=lambda e: e["mtime"], reverse=True)
    return entries


def prune_sessions(max_count: int = _DEFAULT_MAX) -> int:
    """Evict oldest sessions (by mtime) so at most `max_count` remain.

    Returns the number of sessions deleted. Mirrors upstream's LRU eviction
    in sources.py:42-43 (`while len(self._cache) > self._max_size:
    self._cache.popitem(last=False)`), adapted to disk via mtime ordering.
    """
    base = session_dir()
    paths = []
    for path in base.glob("*.json"):
        if path.suffix.endswith(".tmp"):
            continue
        try:
            paths.append((path, path.stat().st_mtime))
        except OSError:
            continue
    if len(paths) <= max_count:
        return 0
    # Oldest first
    paths.sort(key=lambda p: p[1])
    to_delete = paths[: len(paths) - max_count]
    deleted = 0
    for path, _ in to_delete:
        try:
            path.unlink()
            deleted += 1
        except OSError:
            continue
    return deleted

#!/usr/bin/env -S uv run --script
# /// script
# requires-python = ">=3.13,<3.14"
# dependencies = [
#   "httpx>=0.28.0",
#   "tenacity>=8.0.0",
# ]
# ///
"""agentic_extract — LLM-mediated "read this URL and give me verbatim quotes".

Ports upstream GrokSearch `GrokSearchProvider.describe_url()` (grok.py:236-257),
which is defined upstream but never wired to an MCP tool. It asks Grok to read
a URL and return exactly two labeled sections:

    Title: <page title>
    Extracts: <quote 1> | <quote 2> | ... (2-4 verbatim fragments)

The LLM is instructed to be "a copy-paste machine" — no paraphrasing, no
interpretation, just the author's original words.

**When to use this vs agentic_fetch:**
- `agentic_fetch`: when you want the *full* page contents (academic paper,
  docs, source code). Tavily/Firecrawl/Grok engines for 100% fidelity markdown.
- `agentic_extract`: when you want to know *what's on a page in the author's
  own words* — a title plus 2–4 key quotes. Much cheaper and smaller output.

Output: JSON `{url, title, extracts: [str, ...], model}` to stdout. On total
failure, prints `{"error": ...}` and exits 1.

CLI:
    python agentic_extract.py --url "https://..." [--session-id S]

If `--session-id` is provided, the extract result is also appended to that
session's JSON on disk under an `extracted_pages` list (for composable
workflows like `agentic_search → agentic_extract --session-id S`).

Env vars: GROK_API_URL, GROK_API_KEY (required); GROK_MODEL, GROK_DEBUG (optional).
See ../references/extract-and-rank.md and ../references/provider-quirks.md.
"""

from __future__ import annotations

import argparse
import asyncio
import json
import sys
from typing import Optional

sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent))

from _http import (  # noqa: E402
    debug,
    grok_model,
    grok_responses_call,
)
from _prompts import URL_DESCRIBE_PROMPT  # noqa: E402
from _session import read_session, write_session  # noqa: E402


async def describe_url(url: str, model_override: Optional[str] = None) -> dict:
    """LLM-mediated URL → title + verbatim quotes via xAI Responses API.

    Returns a dict with:
        url: the input URL
        title: extracted page title (falls back to the URL if Grok omits it)
        extracts: list of 2-4 verbatim quote strings (may be empty on parse failure)
        model: the effective model id used

    Search mode is "on" so Grok actually browses the URL via web_search rather
    than confabulating content from training data. Annotations are returned by
    the helper but discarded here — describe_url's contract is title+extracts,
    not citations.
    """
    model = grok_model(model_override)
    debug(f"describe_url: model={model} url={url}")

    result_text, _annotations = await grok_responses_call(
        instructions=URL_DESCRIBE_PROMPT,
        user_content=url,
        model=model_override,
        enable_search=True,
        search_mode="on",
    )

    # Parse the upstream-format response:
    #   Title: <title>
    #   Extracts: "quote 1" | "quote 2" | ...
    # Mirrors upstream grok.py:251-256 parsing loop, but splits the extracts
    # line on ` | ` so the caller gets a list instead of a single string.
    title = url
    extracts_raw = ""
    for line in (result_text or "").strip().splitlines():
        stripped = line.strip()
        if stripped.startswith("Title:"):
            candidate = stripped[6:].strip()
            if candidate:
                title = candidate
        elif stripped.startswith("Extracts:"):
            extracts_raw = stripped[9:].strip()

    # Split extracts on ' | ' (the separator the prompt instructs Grok to use).
    # Strip surrounding quotes per-fragment — Grok wraps quotes around each.
    extracts: list[str] = []
    if extracts_raw:
        for frag in extracts_raw.split(" | "):
            frag = frag.strip()
            if len(frag) >= 2 and frag[0] == frag[-1] and frag[0] in ('"', "'"):
                frag = frag[1:-1].strip()
            if frag:
                extracts.append(frag)

    return {
        "url": url,
        "title": title,
        "extracts": extracts,
        "model": model,
    }


def main() -> int:
    parser = argparse.ArgumentParser(
        description="agentic_extract — LLM-mediated URL → title + verbatim quotes."
    )
    parser.add_argument("--url", required=True, help="HTTP/HTTPS URL to extract.")
    parser.add_argument("--model", default="", help="Override GROK_MODEL for this call only.")
    parser.add_argument(
        "--session-id",
        default="",
        help="Optional session id. If set, the extract is appended to the session's 'extracted_pages' list.",
    )
    args = parser.parse_args()

    if not args.url.startswith(("http://", "https://")):
        print(
            json.dumps({"error": "--url must start with http:// or https://"}),
            file=sys.stderr,
        )
        return 1

    try:
        result = asyncio.run(describe_url(args.url, args.model or None))
    except SystemExit as e:
        print(json.dumps({"error": str(e) if e.code != 0 else "configuration error"}))
        return 1
    except Exception as e:
        print(json.dumps({"error": f"unexpected: {type(e).__name__}: {e}"}))
        return 1

    # Optionally merge into an existing session
    if args.session_id:
        session = read_session(args.session_id)
        if session is not None:
            extracted = session.get("extracted_pages") or []
            extracted.append(result)
            session["extracted_pages"] = extracted
            try:
                write_session(args.session_id, session)
                debug(f"appended extract to session {args.session_id}")
            except Exception as e:
                debug(f"failed to update session {args.session_id}: {e}")
        else:
            debug(f"session {args.session_id} not found; skipping append")

    print(json.dumps(result, ensure_ascii=False, indent=2))
    return 0


if __name__ == "__main__":
    sys.exit(main())

#!/usr/bin/env -S uv run --script
# /// script
# requires-python = ">=3.13,<3.14"
# dependencies = [
#   "httpx>=0.28.0",
#   "tenacity>=8.0.0",
# ]
# ///
"""agentic_fetch — full-fidelity URL → Markdown extraction.

Four engines are available:

- **auto** (default): Tavily Extract → Firecrawl Scrape fallback chain.
  Faithful to upstream GrokSearch `web_fetch` (`server.py:339-377`).
  Grok is NOT automatically added to the chain to preserve upstream parity.
- **tavily**: Only Tavily Extract.
- **firecrawl**: Only Firecrawl Scrape (with progressive waitFor retries).
- **grok**: Only Grok via `FETCH_PROMPT` (LLM-mediated 100%-fidelity markdown).
  Ports upstream `grok_provider.fetch()` (dead code in upstream but a real capability).

Output: Markdown text to stdout. On total failure, prints `error: ...` to stderr
and exits 1. Stdout contract is unchanged from v1 — provenance (which engine
succeeded) is surfaced to stderr via `GROK_DEBUG=true` only.

CLI:
    python agentic_fetch.py --url "https://..." [--engine auto|tavily|firecrawl|grok]

Env vars:
    auto / tavily:     TAVILY_API_KEY required
    auto / firecrawl:  FIRECRAWL_API_KEY required (for fallback step in auto)
    grok:              GROK_API_URL + GROK_API_KEY required
    Optional:          GROK_MODEL, GROK_DEBUG

See ../references/fetch-fidelity.md and ../references/provider-quirks.md.
"""

from __future__ import annotations

import argparse
import asyncio
import sys
from typing import Optional

import httpx

sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent))

from _http import (  # noqa: E402
    debug,
    firecrawl_api_key,
    firecrawl_api_url,
    grok_responses_call,
    retry_max_attempts,
    tavily_api_key,
    tavily_api_url,
)
from _prompts import FETCH_PROMPT  # noqa: E402


ENGINES = ("auto", "tavily", "firecrawl", "grok")


async def _tavily_extract(url: str) -> Optional[str]:
    api_key = tavily_api_key()
    if not api_key:
        debug("tavily extract skipped: TAVILY_API_KEY not set")
        return None
    endpoint = f"{tavily_api_url().rstrip('/')}/extract"
    headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}
    body = {"urls": [url], "format": "markdown"}
    try:
        async with httpx.AsyncClient(timeout=60.0) as client:
            resp = await client.post(endpoint, headers=headers, json=body)
            resp.raise_for_status()
            data = resp.json()
            results = data.get("results") or []
            if results:
                content = results[0].get("raw_content", "") or ""
                if content.strip():
                    return content
            debug("tavily extract returned empty content")
            return None
    except Exception as e:
        debug(f"tavily extract failed: {e}")
        return None


async def _firecrawl_scrape(url: str) -> Optional[str]:
    api_key = firecrawl_api_key()
    if not api_key:
        debug("firecrawl scrape skipped: FIRECRAWL_API_KEY not set")
        return None
    endpoint = f"{firecrawl_api_url().rstrip('/')}/scrape"
    headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}
    max_retries = retry_max_attempts()
    for attempt in range(max_retries):
        body = {
            "url": url,
            "formats": ["markdown"],
            "timeout": 60000,
            "waitFor": (attempt + 1) * 1500,  # 1.5s, 3s, 4.5s
        }
        try:
            async with httpx.AsyncClient(timeout=90.0) as client:
                resp = await client.post(endpoint, headers=headers, json=body)
                resp.raise_for_status()
                data = resp.json()
                markdown = (data.get("data") or {}).get("markdown", "") or ""
                if markdown.strip():
                    return markdown
                debug(f"firecrawl scrape returned empty markdown, attempt {attempt + 1}/{max_retries}")
        except Exception as e:
            debug(f"firecrawl scrape failed: {e}")
            return None
    return None


async def _grok_fetch(url: str) -> Optional[str]:
    """Grok-mediated full-page markdown extraction via xAI Responses API.

    Uses the same FETCH_PROMPT (the Chinese "Web Content Fetcher" persona with
    100% fidelity guarantee) and the same user-message composition as upstream
    GrokSearch (URL + CN instruction suffix). Search mode is "on" so Grok
    actually browses the URL via the web_search tool rather than confabulating
    page content. Returns the Grok-generated markdown string or None on failure.

    Annotations are discarded — fetch returns just the markdown text to stdout.
    """
    try:
        text, _annotations = await grok_responses_call(
            instructions=FETCH_PROMPT,
            user_content=url + "\n获取该网页内容并返回其结构化Markdown格式",
            enable_search=True,
            search_mode="on",
        )
    except SystemExit as e:
        debug(f"grok fetch skipped: {e}")
        return None
    except Exception as e:
        debug(f"grok fetch failed: {e}")
        return None
    return text if text and text.strip() else None


async def fetch_url(url: str, engine: str = "auto") -> tuple[Optional[str], Optional[str]]:
    """Fetch a URL's content as markdown via the chosen engine.

    Returns ``(markdown, engine_used)`` or ``(None, None)`` on total failure.

    Public helper — also imported by `agentic_search.py` for `--auto-fetch-top`
    to avoid duplicating the engine routing logic.

    Engine routing:
        auto:      Tavily Extract → Firecrawl Scrape (default, upstream-faithful)
        tavily:    only Tavily Extract
        firecrawl: only Firecrawl Scrape
        grok:      only Grok via FETCH_PROMPT
    """
    if engine not in ENGINES:
        raise ValueError(f"invalid engine {engine!r}; must be one of {ENGINES}")

    debug(f"fetch_url: engine={engine} url={url}")

    if engine == "tavily":
        result = await _tavily_extract(url)
        return (result, "tavily") if result else (None, None)

    if engine == "firecrawl":
        result = await _firecrawl_scrape(url)
        return (result, "firecrawl") if result else (None, None)

    if engine == "grok":
        result = await _grok_fetch(url)
        return (result, "grok") if result else (None, None)

    # engine == "auto" — Tavily → Firecrawl chain (unchanged from v1)
    result = await _tavily_extract(url)
    if result:
        return result, "tavily"
    debug("tavily failed or unavailable, trying firecrawl")
    result = await _firecrawl_scrape(url)
    if result:
        return result, "firecrawl"
    return None, None


# Backwards-compat alias — older callers may still import `fetch`
fetch = fetch_url


def main() -> int:
    parser = argparse.ArgumentParser(
        description="agentic_fetch — full-fidelity URL → Markdown extraction."
    )
    parser.add_argument("--url", required=True, help="HTTP/HTTPS URL to fetch.")
    parser.add_argument(
        "--engine",
        default="auto",
        choices=ENGINES,
        help="Extraction engine: 'auto' (Tavily→Firecrawl, default), 'tavily', 'firecrawl', or 'grok'.",
    )
    args = parser.parse_args()

    if not args.url.startswith(("http://", "https://")):
        print("error: --url must start with http:// or https://", file=sys.stderr)
        return 1

    try:
        markdown, engine_used = asyncio.run(fetch_url(args.url, args.engine))
    except Exception as e:
        print(f"error: unexpected: {type(e).__name__}: {e}", file=sys.stderr)
        return 1

    if markdown is None:
        if args.engine == "auto":
            if not tavily_api_key() and not firecrawl_api_key():
                print(
                    "error: neither TAVILY_API_KEY nor FIRECRAWL_API_KEY is set; cannot fetch",
                    file=sys.stderr,
                )
            else:
                print("error: all extraction providers failed to return content", file=sys.stderr)
        else:
            print(
                f"error: engine '{args.engine}' failed to return content (check API key and network)",
                file=sys.stderr,
            )
        return 1

    debug(f"fetch succeeded via {engine_used}, {len(markdown)} chars")
    sys.stdout.write(markdown)
    if not markdown.endswith("\n"):
        sys.stdout.write("\n")
    return 0


if __name__ == "__main__":
    sys.exit(main())

#!/usr/bin/env -S uv run --script
# /// script
# requires-python = ">=3.13,<3.14"
# dependencies = [
#   "httpx>=0.28.0",
#   "tenacity>=8.0.0",
# ]
# ///
"""agentic_get_sources — retrieve cached sessions from disk.

Simple wrapper around `_session.read_session` and `_session.list_sessions`.
Use it to:

- Retrieve a full session (content, sources, fetched_pages, etc.) from a prior
  `agentic_search` call by its `session_id`.
- List recent sessions for diagnostics / picking which one to rerank or extend.

CLI:
    # Retrieve a single session
    python agentic_get_sources.py --session-id abc123def456

    # List all cached sessions, most-recent first
    python agentic_get_sources.py --list

Output: JSON to stdout. On failure, prints `{"error": ...}` to stderr and
exits 1.

The v2 session cache replaces upstream GrokSearch's in-memory SourcesCache
(sources.py:32-51) with a disk-backed equivalent in the system temp dir. LRU
eviction is by file mtime, capped at 256 sessions (matching upstream).

See ../references/provider-quirks.md for the session cache schema and
storage path.
"""

from __future__ import annotations

import argparse
import json
import sys

sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent))

from _session import list_sessions, read_session, session_dir  # noqa: E402


def main() -> int:
    parser = argparse.ArgumentParser(
        description="agentic_get_sources — retrieve a cached session or list recent sessions."
    )
    group = parser.add_mutually_exclusive_group(required=True)
    group.add_argument(
        "--session-id",
        default="",
        help="Session id returned by a previous agentic_search call.",
    )
    group.add_argument(
        "--list",
        action="store_true",
        help="List recent sessions (most recent first) with diagnostic metadata.",
    )
    args = parser.parse_args()

    if args.list:
        sessions = list_sessions()
        print(
            json.dumps(
                {"cache_dir": str(session_dir()), "count": len(sessions), "sessions": sessions},
                ensure_ascii=False,
                indent=2,
            )
        )
        return 0

    session = read_session(args.session_id)
    if session is None:
        print(
            json.dumps({"error": f"session {args.session_id} not found or expired"}),
            file=sys.stderr,
        )
        return 1

    print(json.dumps(session, ensure_ascii=False, indent=2))
    return 0


if __name__ == "__main__":
    sys.exit(main())

#!/usr/bin/env -S uv run --script
# /// script
# requires-python = ">=3.13,<3.14"
# dependencies = [
#   "httpx>=0.28.0",
#   "tenacity>=8.0.0",
# ]
# ///
"""agentic_map — discover URLs under a site root via Tavily Map.

Traverses a site like a graph, returning a structured list of discovered URLs.
Use natural-language `--instructions` to filter (e.g., "only API reference
pages"). Tunable depth/breadth/limit/timeout.

Output: JSON `{base_url, results, response_time}` to stdout. On error, prints
`error: ...` to stderr and exits 1.

CLI:
    python agentic_map.py --url "https://docs.example.com" \\
        [--instructions "only docs"] [--max-depth 2] [--max-breadth 30] \\
        [--limit 100] [--timeout 150]

Env vars: TAVILY_API_KEY (required), TAVILY_API_URL (optional).
See ../references/provider-quirks.md.
"""

from __future__ import annotations

import argparse
import asyncio
import json
import sys

import httpx

sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent))

from _http import debug, tavily_api_key, tavily_api_url  # noqa: E402


async def map_site(
    url: str,
    instructions: str,
    max_depth: int,
    max_breadth: int,
    limit: int,
    timeout: int,
) -> dict:
    api_key = tavily_api_key()
    if not api_key:
        raise SystemExit("error: TAVILY_API_KEY is not set; agentic_map requires it")

    endpoint = f"{tavily_api_url().rstrip('/')}/map"
    headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}
    body: dict = {
        "url": url,
        "max_depth": max_depth,
        "max_breadth": max_breadth,
        "limit": limit,
        "timeout": timeout,
    }
    if instructions:
        body["instructions"] = instructions

    debug(f"map request: url={url} depth={max_depth} breadth={max_breadth} limit={limit}")

    try:
        async with httpx.AsyncClient(timeout=float(timeout + 10)) as client:
            resp = await client.post(endpoint, headers=headers, json=body)
            resp.raise_for_status()
            data = resp.json()
            return {
                "base_url": data.get("base_url", "") or "",
                "results": data.get("results", []) or [],
                "response_time": data.get("response_time", 0),
            }
    except httpx.TimeoutException:
        raise SystemExit(f"error: map timed out after {timeout}s")
    except httpx.HTTPStatusError as e:
        raise SystemExit(f"error: HTTP {e.response.status_code}: {e.response.text[:200]}")
    except Exception as e:
        raise SystemExit(f"error: {type(e).__name__}: {e}")


def main() -> int:
    parser = argparse.ArgumentParser(
        description="agentic_map — discover URLs under a site root via Tavily Map."
    )
    parser.add_argument("--url", required=True, help="Root URL to begin mapping.")
    parser.add_argument(
        "--instructions",
        default="",
        help="Natural-language filter (e.g., 'only documentation pages').",
    )
    parser.add_argument("--max-depth", type=int, default=1, help="Traversal depth (1-5). Default 1.")
    parser.add_argument(
        "--max-breadth", type=int, default=20, help="Max links per page (1-500). Default 20."
    )
    parser.add_argument("--limit", type=int, default=50, help="Total link cap (1-500). Default 50.")
    parser.add_argument(
        "--timeout", type=int, default=150, help="Server-side timeout in seconds (10-150). Default 150."
    )
    args = parser.parse_args()

    if not args.url.startswith(("http://", "https://")):
        print("error: --url must start with http:// or https://", file=sys.stderr)
        return 1

    try:
        result = asyncio.run(
            map_site(
                args.url,
                args.instructions,
                args.max_depth,
                args.max_breadth,
                args.limit,
                args.timeout,
            )
        )
    except SystemExit as e:
        print(str(e), file=sys.stderr)
        return 1
    except Exception as e:
        print(f"error: unexpected: {type(e).__name__}: {e}", file=sys.stderr)
        return 1

    print(json.dumps(result, ensure_ascii=False, indent=2))
    return 0


if __name__ == "__main__":
    sys.exit(main())

#!/usr/bin/env -S uv run --script
# /// script
# requires-python = ">=3.13,<3.14"
# dependencies = [
#   "httpx>=0.28.0",
#   "tenacity>=8.0.0",
# ]
# ///
"""agentic_rank — rerank a list of sources by relevance to a refined query.

Ports upstream GrokSearch `GrokSearchProvider.rank_sources()` (grok.py:259-288),
which is defined upstream but never wired to an MCP tool. It takes a numbered
source list, asks Grok to output the numbers reordered by relevance to a query,
and returns the sources in the new order (with missing indices filled at the
end — robust to partial model outputs).

**Typical workflow**:
    # 1. Run a broad search
    uv run scripts/agentic_search.py --query "best agentic AI frameworks" > /tmp/search.json
    SID=$(python -c "import json; print(json.load(open('/tmp/search.json'))['session_id'])")

    # 2. Rerank by a more specific lens
    uv run scripts/agentic_rank.py --query "production-ready open source" --session-id "$SID"

The rerank mutates the session's sources list in place so subsequent scripts
(`agentic_get_sources`, `agentic_search --auto-fetch-top`) see the new order.

Two input modes:
    --session-id S            Read sources from session cache, rerank, write back.
    --sources-json -          Read a JSON list of sources from stdin (ad-hoc mode).

Output: JSON `{ordered_sources, model, session_id?}` to stdout.

Env vars: GROK_API_URL, GROK_API_KEY (required); GROK_MODEL, GROK_DEBUG (optional).
See ../references/extract-and-rank.md.
"""

from __future__ import annotations

import argparse
import asyncio
import json
import sys
from typing import Optional

sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent))

from _http import (  # noqa: E402
    debug,
    grok_model,
    grok_responses_call,
)
from _prompts import RANK_SOURCES_PROMPT  # noqa: E402
from _session import read_session, write_session  # noqa: E402


def _format_sources_for_ranking(sources: list[dict]) -> str:
    """Format sources as a numbered list for the Grok prompt.

    Format (one per line, 1-indexed):
        1. [Title] - URL
           description (if any)
        2. ...
    """
    lines = []
    for i, src in enumerate(sources, start=1):
        url = src.get("url", "") or ""
        title = src.get("title") or "Untitled"
        lines.append(f"{i}. [{title}] - {url}")
        desc = src.get("description") or ""
        if desc:
            lines.append(f"   {desc[:200]}")
    return "\n".join(lines)


async def rank_sources(query: str, sources: list[dict], model_override: Optional[str] = None) -> dict:
    """Rerank a list of sources by relevance to a refined query via xAI Responses API.

    Returns a dict:
        ordered_sources: [...]  (same dicts, new order)
        model: effective model id
        original_count: int

    **Search is disabled** for this call (`enable_search=False`). Ranking is
    pure reasoning over a given list of sources — no new web information is
    needed, and disabling search saves cost and latency. The Responses API
    helper omits both the `web_search` tool and the `search_parameters` field
    when `enable_search=False`.

    Mirrors upstream's 'fill missing indices' tail logic so partial model
    outputs degrade gracefully (unranked indices get appended at the end).
    """
    if not sources:
        return {"ordered_sources": [], "model": grok_model(model_override), "original_count": 0}

    model = grok_model(model_override)
    total = len(sources)
    sources_text = _format_sources_for_ranking(sources)

    debug(f"rank_sources: model={model} total={total} (search disabled)")

    result_text, _annotations = await grok_responses_call(
        instructions=RANK_SOURCES_PROMPT,
        user_content=f"Query: {query}\n\n{sources_text}",
        model=model_override,
        enable_search=False,
    )

    # Parse the response: space-separated integers, dedupe, validate range.
    # Mirrors upstream grok.py:274-287 byte-for-byte.
    order: list[int] = []
    seen: set[int] = set()
    for token in (result_text or "").strip().split():
        try:
            n = int(token)
            if 1 <= n <= total and n not in seen:
                seen.add(n)
                order.append(n)
        except ValueError:
            continue
    # Fill any missing indices at the end
    for i in range(1, total + 1):
        if i not in seen:
            order.append(i)

    # Reorder the original sources list (1-indexed → 0-indexed)
    ordered = [sources[n - 1] for n in order]

    return {
        "ordered_sources": ordered,
        "model": model,
        "original_count": total,
    }


def main() -> int:
    parser = argparse.ArgumentParser(
        description="agentic_rank — rerank a source list by relevance to a refined query."
    )
    parser.add_argument("--query", required=True, help="The refined query to rank sources against.")
    parser.add_argument("--model", default="", help="Override GROK_MODEL for this call only.")
    group = parser.add_mutually_exclusive_group(required=True)
    group.add_argument(
        "--session-id",
        default="",
        help="Session id from a previous agentic_search call. Sources are loaded from the session cache and the reranked order is written back.",
    )
    group.add_argument(
        "--sources-json",
        default="",
        help="Read a JSON list of sources from this file, or '-' for stdin.",
    )
    args = parser.parse_args()

    # Load sources
    sources: list[dict]
    session_data: Optional[dict] = None

    if args.session_id:
        session_data = read_session(args.session_id)
        if session_data is None:
            print(
                json.dumps({"error": f"session {args.session_id} not found or expired"}),
                file=sys.stderr,
            )
            return 1
        sources = session_data.get("sources") or []
    else:
        try:
            if args.sources_json == "-":
                raw = sys.stdin.read()
            else:
                raw = __import__("pathlib").Path(args.sources_json).read_text(encoding="utf-8")
            sources = json.loads(raw)
            if not isinstance(sources, list):
                raise ValueError("sources JSON must be a list")
        except Exception as e:
            print(
                json.dumps({"error": f"failed to read sources JSON: {type(e).__name__}: {e}"}),
                file=sys.stderr,
            )
            return 1

    if not sources:
        print(json.dumps({"error": "no sources to rank"}), file=sys.stderr)
        return 1

    # Run the rank
    try:
        result = asyncio.run(rank_sources(args.query, sources, args.model or None))
    except SystemExit as e:
        print(json.dumps({"error": str(e) if e.code != 0 else "configuration error"}))
        return 1
    except Exception as e:
        print(json.dumps({"error": f"unexpected: {type(e).__name__}: {e}"}))
        return 1

    # If invoked via session_id, write the reranked sources back to the session
    if args.session_id and session_data is not None:
        session_data["sources"] = result["ordered_sources"]
        # Preserve sources_count (still the same count, just reordered)
        session_data["sources_count"] = len(result["ordered_sources"])
        session_data["ranked_by_query"] = args.query
        try:
            write_session(args.session_id, session_data)
            result["session_id"] = args.session_id
            debug(f"wrote reranked sources back to session {args.session_id}")
        except Exception as e:
            debug(f"failed to write back to session: {e}")

    print(json.dumps(result, ensure_ascii=False, indent=2))
    return 0


if __name__ == "__main__":
    sys.exit(main())

#!/usr/bin/env -S uv run --script
# /// script
# requires-python = ">=3.13,<3.14"
# dependencies = [
#   "httpx>=0.28.0",
#   "tenacity>=8.0.0",
# ]
# ///
"""agentic_search — deep web research via Grok + optional Tavily/Firecrawl fusion.

Calls Grok's chat-completions endpoint with a discipline-loaded system prompt,
optionally fans out to Tavily and/or Firecrawl in parallel for supplementary
sources, parses citations out of Grok's free-form answer, dedupes sources by
URL, and prints a JSON object to stdout.

Output schema:
    {
        "content": str,                 # Grok's answer with sources stripped
        "sources": [                    # deduped, ordered: grok first, then extras
            {"url": str, "title"?: str, "description"?: str, "provider"?: str},
            ...
        ],
        "sources_count": int,
        "model": str                    # the effective model used
    }

On configuration error, prints {"error": "..."} and exits 1.

CLI:
    python agentic_search.py --query "..." [--platform "..."] [--model "..."] [--extra-sources N]

Env vars: GROK_API_URL, GROK_API_KEY (required); GROK_MODEL, TAVILY_API_KEY,
FIRECRAWL_API_KEY (optional). See ../references/provider-quirks.md.
"""

from __future__ import annotations

import argparse
import ast
import asyncio
import json
import re
import sys
from datetime import datetime, timezone
from typing import Any, Optional

import httpx

# Allow running this file directly: `python scripts/agentic_search.py`
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent))

from _http import (  # noqa: E402
    debug,
    firecrawl_api_key,
    firecrawl_api_url,
    grok_model,
    grok_responses_call,
    normalize_responses_annotations,
    tavily_api_key,
    tavily_api_url,
)
from _prompts import SEARCH_PROMPT  # noqa: E402
from _session import (  # noqa: E402
    new_session_id,
    prune_sessions,
    write_session,
)


# ---------- time-context heuristic ----------

_CN_TIME_KEYWORDS = (
    "当前", "现在", "今天", "明天", "昨天",
    "本周", "上周", "下周", "这周",
    "本月", "上月", "下月", "这个月",
    "今年", "去年", "明年",
    "最新", "最近", "近期", "刚刚", "刚才",
    "实时", "即时", "目前",
)
_EN_TIME_KEYWORDS = (
    "current", "now", "today", "tomorrow", "yesterday",
    "this week", "last week", "next week",
    "this month", "last month", "next month",
    "this year", "last year", "next year",
    "latest", "recent", "recently", "just now",
    "real-time", "realtime", "up-to-date",
)
_CN_WEEKDAYS = ("星期一", "星期二", "星期三", "星期四", "星期五", "星期六", "星期日")


def _needs_time_context(query: str) -> bool:
    if any(kw in query for kw in _CN_TIME_KEYWORDS):
        return True
    lower = query.lower()
    return any(kw in lower for kw in _EN_TIME_KEYWORDS)


def _local_time_block() -> str:
    try:
        local_tz = datetime.now().astimezone().tzinfo
        local_now = datetime.now(local_tz)
    except Exception:
        local_now = datetime.now(timezone.utc)
    weekday = _CN_WEEKDAYS[local_now.weekday()]
    return (
        f"[Current Time Context]\n"
        f"- Date: {local_now.strftime('%Y-%m-%d')} ({weekday})\n"
        f"- Time: {local_now.strftime('%H:%M:%S')}\n"
        f"- Timezone: {local_now.tzname() or 'Local'}\n"
    )


# ---------- Grok call (xAI Responses API with native web search) ----------

async def _grok_search(
    query: str, platform: str, model: Optional[str]
) -> tuple[str, list[dict]]:
    """Call Grok via the xAI Responses API with web_search enabled.

    Returns ``(text_content, raw_annotations)``. Raw annotations are
    Responses-API citation objects; pass them through
    `normalize_responses_annotations` to convert to source dicts.

    Time-context injection (CN/EN keyword detection) and platform-focus
    suffix are preserved verbatim from v1/v2 — they're prompt-discipline
    decisions, not API-shape decisions.
    """
    user_content = ""
    if _needs_time_context(query):
        user_content += _local_time_block() + "\n"
    user_content += query
    if platform:
        user_content += (
            f"\n\nYou should search the web for the information you need, "
            f"and focus on these platform: {platform}\n"
        )

    return await grok_responses_call(
        instructions=SEARCH_PROMPT,
        user_content=user_content,
        model=model,
        enable_search=True,
        search_mode="auto",
        tools=["web_search", "x_search"],  # Use both web and X search for research
    )


# ---------- Tavily / Firecrawl supplementary search ----------

async def _tavily_search(query: str, max_results: int) -> Optional[list[dict]]:
    api_key = tavily_api_key()
    if not api_key or max_results <= 0:
        return None
    endpoint = f"{tavily_api_url().rstrip('/')}/search"
    headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}
    body = {
        "query": query,
        "max_results": max_results,
        "search_depth": "advanced",
        "include_raw_content": False,
        "include_answer": False,
    }
    try:
        async with httpx.AsyncClient(timeout=90.0) as client:
            resp = await client.post(endpoint, headers=headers, json=body)
            resp.raise_for_status()
            data = resp.json()
            results = data.get("results") or []
            return [
                {
                    "title": r.get("title", "") or "",
                    "url": r.get("url", "") or "",
                    "content": r.get("content", "") or "",
                    "score": r.get("score", 0),
                }
                for r in results
            ] or None
    except Exception as e:
        debug(f"tavily search failed: {e}")
        return None


async def _firecrawl_search(query: str, limit: int) -> Optional[list[dict]]:
    api_key = firecrawl_api_key()
    if not api_key or limit <= 0:
        return None
    endpoint = f"{firecrawl_api_url().rstrip('/')}/search"
    headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}
    body = {"query": query, "limit": limit}
    try:
        async with httpx.AsyncClient(timeout=90.0) as client:
            resp = await client.post(endpoint, headers=headers, json=body)
            resp.raise_for_status()
            data = resp.json()
            results = (data.get("data") or {}).get("web") or []
            return [
                {
                    "title": r.get("title", "") or "",
                    "url": r.get("url", "") or "",
                    "description": r.get("description", "") or "",
                }
                for r in results
            ] or None
    except Exception as e:
        debug(f"firecrawl search failed: {e}")
        return None


# ---------- source extraction (port of upstream sources.py) ----------

_URL_PATTERN = re.compile(r'https?://[^\s<>"\'`，。、；：！？》）】\)]+')
_MD_LINK_PATTERN = re.compile(r"\[([^\]]+)\]\((https?://[^)]+)\)")
_SOURCES_HEADING_PATTERN = re.compile(
    r"(?im)^"
    r"(?:#{1,6}\s*)?"
    r"(?:\*\*|__)?\s*"
    r"(sources?|references?|citations?|信源|参考资料|参考|引用|来源列表|来源)"
    r"\s*(?:\*\*|__)?"
    r"(?:\s*[（(][^)\n]*[)）])?"
    r"\s*[:：]?\s*$"
)
_SOURCES_FUNCTION_PATTERN = re.compile(
    r"(?im)(^|\n)\s*(sources|source|citations|citation|references|reference|citation_card|source_cards|source_card)\s*\("
)

# v2: parser for inline `(\`citation_card\`: Author, "Title," Source, Year, URL, description)`
# parentheticals that Grok uses mid-sentence. Reuses _URL_PATTERN for URL extraction.
_CITATION_CARD_PATTERN = re.compile(
    r"\(\s*`?citation_card`?\s*:\s*([^)]+?)\s*\)",
    re.IGNORECASE | re.DOTALL,
)
_QUOTED_TITLE_PATTERN = re.compile(r'"([^"]+?)"')
_YEAR_PATTERN = re.compile(r"\b(?:19|20)\d{2}\b")

# v2.1: matches fenced code blocks of any language tag (or none). Captures the
# language tag as group 1 and the body as group 2. Used by
# _parse_fenced_citation_blocks below.
#
# In real grok-4-1-fast-reasoning output we've observed at least three citation
# fence variants, ALL of which this regex catches:
#   ```json\n{"url":..., "title":...}\n```                    (JSON dict)
#   ```\n{"url":..., "title":...}\n```                        (JSON, no lang)
#   ```citation_card\nurl: ...\ntitle: "..."\nsummary: ...```  (YAML-ish kv)
_FENCED_BLOCK_PATTERN = re.compile(
    r"```(\w*)\s*\n?(.*?)\n?```",
    re.DOTALL,
)


def _extract_unique_urls(text: str) -> list[str]:
    seen: set[str] = set()
    urls: list[str] = []
    for m in _URL_PATTERN.finditer(text or ""):
        url = m.group().rstrip(".,;:!?")
        if url not in seen:
            seen.add(url)
            urls.append(url)
    return urls


def _normalize_sources(data: Any) -> list[dict]:
    items: list[Any]
    if isinstance(data, (list, tuple)):
        items = list(data)
    elif isinstance(data, dict):
        items = [data]
    else:
        items = [data]

    out: list[dict] = []
    seen: set[str] = set()
    for item in items:
        if isinstance(item, str):
            for url in _extract_unique_urls(item):
                if url not in seen:
                    seen.add(url)
                    out.append({"url": url})
            continue
        if isinstance(item, (list, tuple)) and len(item) >= 2:
            title, url = item[0], item[1]
            if isinstance(url, str) and url.startswith(("http://", "https://")) and url not in seen:
                seen.add(url)
                rec: dict = {"url": url}
                if isinstance(title, str) and title.strip():
                    rec["title"] = title.strip()
                out.append(rec)
            continue
        if isinstance(item, dict):
            url = item.get("url") or item.get("href") or item.get("link")
            if not isinstance(url, str) or not url.startswith(("http://", "https://")):
                continue
            if url in seen:
                continue
            seen.add(url)
            rec = {"url": url}
            title = item.get("title") or item.get("name") or item.get("label")
            if isinstance(title, str) and title.strip():
                rec["title"] = title.strip()
            desc = item.get("description") or item.get("snippet") or item.get("content")
            if isinstance(desc, str) and desc.strip():
                rec["description"] = desc.strip()
            out.append(rec)
    return out


# v2.1: known citation field keys, used to slice key:value bodies regardless
# of whether they're newline-separated or all on one line. Word-boundary anchored
# so titles containing tokens like "Reasoning:" don't match (case-insensitive).
_CITATION_KEY_PATTERN = re.compile(
    r"\b(title|author|authors|url|date|year|section|snippet|summary|description|published|publisher|source|venue)\s*:\s*",
    re.IGNORECASE,
)


def _parse_keyvalue_citation_body(body: str) -> Optional[dict]:
    """Parse a key:value citation block body into a dict.

    Used by `_parse_fenced_citation_blocks` when `json.loads` fails on the
    body. Handles both newline-separated AND single-line layouts that Grok
    emits — sliced by known field-key positions, not by `\\n`.

    Strips surrounding quotes from values. Returns None if no `url` key is
    found or no recognized fields parse.

    Multi-line example:
        title: "What is a 'harness' in agent benchmarks?"
        author: Nathan Lambert
        url: https://www.interconnects.ai/p/what-is-a-harness-in-agent-benchmarks
        date: 2024-10-15
        summary: Defines harness as ...

    Single-line example (also seen in real Grok output):
        title: "Harnesses" author: Stanford CRFM url: https://crfm.stanford.edu/helm/latest/ date: 2024 section: Harness Definition
    """
    matches = list(_CITATION_KEY_PATTERN.finditer(body))
    if not matches:
        return None

    result: dict = {}
    for i, m in enumerate(matches):
        key = m.group(1).lower()
        value_start = m.end()
        value_end = matches[i + 1].start() if i + 1 < len(matches) else len(body)
        value = body[value_start:value_end].strip()
        # Strip trailing comma/semicolon noise that may bleed in from
        # comma-separated layouts
        value = value.rstrip(",;").strip()
        # Strip surrounding matched quotes
        if len(value) >= 2 and value[0] == value[-1] and value[0] in ('"', "'"):
            value = value[1:-1].strip()
        if value:
            # Last value wins on duplicate keys (rare; usually a malformed citation)
            result[key] = value

    if "url" not in result:
        return None
    return result


def _parse_fenced_citation_blocks(text: str) -> list[dict]:
    """Parse fenced code blocks containing citation records, in either JSON
    dict form OR YAML-ish `key: value` form.

    v2.1: handles the formats `grok-4-1-fast-reasoning` actually uses in real
    output. Three observed variants are all caught:

        ```json
        {"url": "...", "title": "...", "snippet": "..."}
        ```

        ```
        {"url": "...", "title": "..."}
        ```

        ```citation_card
        title: "..."
        author: ...
        url: ...
        date: 2024-10-15
        summary: ...
        ```

    For each fenced block:
      1. Pre-filter on the cheap substring `url` — skip blocks that obviously
         can't be citations (e.g. code examples).
      2. Try `json.loads(body)` first; on success, treat result as dict or list of dicts.
      3. On JSON failure, try the line-based key:value parser.
      4. Extract a normalized record with these field aliases:
            url                   -> url (required, must start with http(s)://)
            title                 -> title
            author / authors      -> author
            year (or date's YYYY) -> year
            snippet / summary
              / description       -> description
            citation_id           -> dropped (Grok-internal)
      5. Dedupe by URL across the entire input.

    Output schema matches `_parse_citation_cards` so downstream consumers
    don't need to know which extractor produced a record.
    """
    out: list[dict] = []
    seen: set[str] = set()
    for m in _FENCED_BLOCK_PATTERN.finditer(text or ""):
        body = m.group(2).strip()
        if not body or "url" not in body:
            continue  # cheap pre-filter — skip non-citation code blocks

        # Try JSON first; fall back to key:value parser.
        items: list[dict] = []
        try:
            data = json.loads(body)
            if isinstance(data, list):
                items = [d for d in data if isinstance(d, dict)]
            elif isinstance(data, dict):
                items = [data]
        except (json.JSONDecodeError, ValueError):
            kv = _parse_keyvalue_citation_body(body)
            if kv is not None:
                items = [kv]

        for item in items:
            url = item.get("url")
            if not isinstance(url, str) or not url.startswith(("http://", "https://")):
                continue
            url = url.rstrip(".,;:!?")
            if url in seen:
                continue
            seen.add(url)

            rec: dict = {"url": url}

            # title
            title = item.get("title")
            if isinstance(title, str) and title.strip():
                rec["title"] = title.strip()

            # author / authors
            author = item.get("author") or item.get("authors")
            if isinstance(author, str) and author.strip():
                rec["author"] = author.strip()
            elif isinstance(author, list) and author:
                rec["author"] = ", ".join(str(a) for a in author if a)

            # year directly, or year from a date string like "2024-10-15"
            year = item.get("year")
            if isinstance(year, str) and year.strip():
                rec["year"] = year.strip()
            elif isinstance(year, int):
                rec["year"] = str(year)
            else:
                date = item.get("date")
                if isinstance(date, str):
                    ym = _YEAR_PATTERN.search(date)
                    if ym:
                        rec["year"] = ym.group()

            # description: snippet | summary | description
            for k in ("snippet", "summary", "description"):
                v = item.get(k)
                if isinstance(v, str) and v.strip():
                    rec["description"] = v.strip()
                    break

            out.append(rec)
    return out


# Backwards-compat alias — earlier v2.1 plan referred to it as JSON-only.
_parse_json_citation_blocks = _parse_fenced_citation_blocks


def _parse_citation_cards(text: str) -> list[dict]:
    r"""Parse Grok's inline `(\`citation_card\`: Author, "Title," Source, Year, URL, description)`
    annotations into rich source dicts.

    Returns a list of `{url, title?, author?, year?, description?}` dicts.
    Best-effort: any field that can't be parsed is omitted (never None).
    Dedupes by URL across the input. Robust to missing fields and to the
    backtick-wrapped vs bare `citation_card` variants.

    v2 deviation from upstream GrokSearch — upstream's `sources.py` only
    parses citation_card as a top-level function call (pattern 1 in
    `split_answer_and_sources`), not as inline parentheticals. This parser
    catches the inline form Grok actually emits in v2 model output.
    """
    out: list[dict] = []
    seen: set[str] = set()
    for m in _CITATION_CARD_PATTERN.finditer(text or ""):
        inner = m.group(1).strip()

        # URL — required; if absent, skip this annotation
        url_match = _URL_PATTERN.search(inner)
        if not url_match:
            continue
        url = url_match.group().rstrip(".,;:!?")
        if url in seen:
            continue
        seen.add(url)
        rec: dict = {"url": url}

        # Title — first quoted string
        title_match = _QUOTED_TITLE_PATTERN.search(inner)
        if title_match:
            title = title_match.group(1).strip().rstrip(",").strip()
            if title:
                rec["title"] = title

        # Year — first 4-digit year-like token (matches the standalone year
        # before the URL most of the time; falls back to year inside the URL
        # path if no standalone year is present)
        year_match = _YEAR_PATTERN.search(inner)
        if year_match:
            rec["year"] = year_match.group()

        # Author — text before the first comma, if it's not the URL or
        # the (already-extracted) title
        first_comma = inner.find(",")
        if first_comma > 0:
            author_candidate = inner[:first_comma].strip()
            if (
                author_candidate
                and "http" not in author_candidate
                and author_candidate != rec.get("title", "")
                and not _YEAR_PATTERN.fullmatch(author_candidate)
            ):
                rec["author"] = author_candidate

        # Description — text after the URL up to the closing paren
        url_end = url_match.end()
        tail = inner[url_end:].strip()
        # Strip leading punctuation/comma noise
        while tail and tail[0] in ",;:.- ":
            tail = tail[1:]
        if tail:
            rec["description"] = tail.strip()

        out.append(rec)
    return out


def _extract_sources_from_text(text: str) -> list[dict]:
    r"""Harvest sources from free-form text using a four-tier extraction order.

    Tiers, richest first (each one's URLs win over the next, deduped by URL):
        1. Fenced JSON citation blocks  (v2.1) — `_parse_json_citation_blocks`
           handles ```json {"url":...,"title":...,"snippet":...} ``` blocks,
           which is the format `grok-4-1-fast-reasoning` actually emits.
        2. citation_card parentheticals  (v2)  — `_parse_citation_cards`
           handles `(\`citation_card\`: Author, "Title," ..., URL, desc)`.
        3. Markdown links                (v1)  — `[anchor](url)`. Only `title`
           field populated (the link's anchor text).
        4. Bare URLs                     (v1)  — no metadata.
    """
    sources: list[dict] = []
    seen: set[str] = set()

    # Tier 1 (v2.1): fenced citation blocks (JSON or YAML-ish key:value form)
    for rec in _parse_fenced_citation_blocks(text or ""):
        url = rec["url"]
        if url not in seen:
            seen.add(url)
            sources.append(rec)

    # Tier 2 (v2): citation_card parentheticals
    for rec in _parse_citation_cards(text or ""):
        url = rec["url"]
        if url in seen:
            continue
        seen.add(url)
        sources.append(rec)

    # Tier 3 (v1): markdown links (poorer metadata: just title)
    for title, url in _MD_LINK_PATTERN.findall(text or ""):
        url = (url or "").strip()
        if not url or url in seen:
            continue
        seen.add(url)
        title = (title or "").strip()
        sources.append({"title": title, "url": url} if title else {"url": url})

    # Finally bare URLs (no metadata)
    for url in _extract_unique_urls(text or ""):
        if url in seen:
            continue
        seen.add(url)
        sources.append({"url": url})

    return sources


def _parse_sources_payload(payload: str) -> list[dict]:
    payload = (payload or "").strip().rstrip(";")
    if not payload:
        return []
    data: Any = None
    try:
        data = json.loads(payload)
    except Exception:
        try:
            data = ast.literal_eval(payload)
        except Exception:
            data = None
    if data is None:
        return _extract_sources_from_text(payload)
    if isinstance(data, dict):
        for key in ("sources", "citations", "references", "urls"):
            if key in data:
                return _normalize_sources(data[key])
        return _normalize_sources(data)
    return _normalize_sources(data)


def _extract_balanced_call(text: str, open_idx: int) -> Optional[tuple[int, str]]:
    if open_idx < 0 or open_idx >= len(text) or text[open_idx] != "(":
        return None
    depth = 1
    in_string: Optional[str] = None
    escape = False
    for idx in range(open_idx + 1, len(text)):
        ch = text[idx]
        if in_string:
            if escape:
                escape = False
                continue
            if ch == "\\":
                escape = True
                continue
            if ch == in_string:
                in_string = None
            continue
        if ch in ("'", '"'):
            in_string = ch
            continue
        if ch == "(":
            depth += 1
            continue
        if ch == ")":
            depth -= 1
            if depth == 0:
                if text[idx + 1:].strip():
                    return None
                return idx, text[open_idx + 1: idx]
    return None


def _split_function_call(text: str) -> Optional[tuple[str, list[dict]]]:
    matches = list(_SOURCES_FUNCTION_PATTERN.finditer(text))
    if not matches:
        return None
    for m in reversed(matches):
        open_idx = m.end() - 1
        extracted = _extract_balanced_call(text, open_idx)
        if not extracted:
            continue
        _, args_text = extracted
        sources = _parse_sources_payload(args_text)
        if not sources:
            continue
        return text[: m.start()].rstrip(), sources
    return None


def _split_heading(text: str) -> Optional[tuple[str, list[dict]]]:
    matches = list(_SOURCES_HEADING_PATTERN.finditer(text))
    if not matches:
        return None
    for m in reversed(matches):
        sources = _extract_sources_from_text(text[m.start():])
        if sources:
            return text[: m.start()].rstrip(), sources
    return None


def _split_details_block(text: str) -> Optional[tuple[str, list[dict]]]:
    lower = text.lower()
    close_idx = lower.rfind("</details>")
    if close_idx == -1:
        return None
    if text[close_idx + len("</details>"):].strip():
        return None
    open_idx = lower.rfind("<details", 0, close_idx)
    if open_idx == -1:
        return None
    block = text[open_idx: close_idx + len("</details>")]
    sources = _extract_sources_from_text(block)
    if len(sources) < 2:
        return None
    return text[:open_idx].rstrip(), sources


def _is_link_only_line(line: str) -> bool:
    stripped = re.sub(r"^\s*(?:[-*]|\d+\.)\s*", "", line).strip()
    if not stripped:
        return False
    if stripped.startswith(("http://", "https://")):
        return True
    return bool(_MD_LINK_PATTERN.search(stripped))


def _split_tail_links(text: str) -> Optional[tuple[str, list[dict]]]:
    lines = text.splitlines()
    if not lines:
        return None
    idx = len(lines) - 1
    while idx >= 0 and not lines[idx].strip():
        idx -= 1
    if idx < 0:
        return None
    tail_end = idx
    link_count = 0
    while idx >= 0:
        line = lines[idx].strip()
        if not line:
            idx -= 1
            continue
        if not _is_link_only_line(line):
            break
        link_count += 1
        idx -= 1
    if link_count < 2:
        return None
    tail_start = idx + 1
    block = "\n".join(lines[tail_start: tail_end + 1])
    sources = _extract_sources_from_text(block)
    if not sources:
        return None
    return "\n".join(lines[:tail_start]).rstrip(), sources


def split_answer_and_sources(text: str) -> tuple[str, list[dict]]:
    raw = (text or "").strip()
    if not raw:
        return "", []
    for splitter in (_split_function_call, _split_heading, _split_details_block, _split_tail_links):
        result = splitter(raw)
        if result:
            return result
    # v1 deviation from upstream GrokSearch: when none of the four upstream
    # patterns match (e.g. Grok inlined citations as `(citation_card: ..., URL)`
    # annotations mid-sentence), fall back to harvesting any inline URLs from
    # the answer body without modifying it. Reuses _extract_sources_from_text.
    inline_sources = _extract_sources_from_text(raw)
    return raw, inline_sources


def _merge_sources(*lists: list[dict]) -> list[dict]:
    seen: set[str] = set()
    merged: list[dict] = []
    for sources in lists:
        for item in sources or []:
            url = (item or {}).get("url")
            if not isinstance(url, str) or not url.strip():
                continue
            url = url.strip()
            if url in seen:
                continue
            seen.add(url)
            merged.append(item)
    return merged


def _extras_to_sources(
    tavily: Optional[list[dict]], firecrawl: Optional[list[dict]]
) -> list[dict]:
    out: list[dict] = []
    seen: set[str] = set()
    # Firecrawl first to match upstream's preference (Firecrawl is breadth-prioritized)
    for r in firecrawl or []:
        url = (r.get("url") or "").strip()
        if not url or url in seen:
            continue
        seen.add(url)
        item: dict = {"url": url, "provider": "firecrawl"}
        if r.get("title"):
            item["title"] = r["title"].strip()
        if r.get("description"):
            item["description"] = r["description"].strip()
        out.append(item)
    for r in tavily or []:
        url = (r.get("url") or "").strip()
        if not url or url in seen:
            continue
        seen.add(url)
        item = {"url": url, "provider": "tavily"}
        if r.get("title"):
            item["title"] = r["title"].strip()
        if r.get("content"):
            item["description"] = r["content"].strip()
        out.append(item)
    return out


# ---------- main ----------

async def run(
    query: str,
    platform: str,
    model_override: Optional[str],
    extra_sources: int,
    auto_fetch_top: int = 0,
) -> dict:
    session_id = new_session_id()
    created_at = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
    model = grok_model(model_override)

    providers_used: list[str] = []
    providers_failed: list[str] = []

    # Allocation matches upstream server.py:153-164:
    # If both keys present, Firecrawl takes 100% of N (it's breadth-prioritized).
    has_tavily = bool(tavily_api_key())
    has_firecrawl = bool(firecrawl_api_key())
    firecrawl_count = 0
    tavily_count = 0
    if extra_sources > 0:
        if has_firecrawl and has_tavily:
            firecrawl_count = extra_sources
            tavily_count = 0
        elif has_firecrawl:
            firecrawl_count = extra_sources
        elif has_tavily:
            tavily_count = extra_sources

    async def safe_grok() -> tuple[str, list[dict]]:
        try:
            return await _grok_search(query, platform, model)
        except Exception as e:
            debug(f"grok search failed: {e}")
            return "", []

    coros: list = [safe_grok()]
    if tavily_count > 0:
        coros.append(_tavily_search(query, tavily_count))
    if firecrawl_count > 0:
        coros.append(_firecrawl_search(query, firecrawl_count))

    gathered = await asyncio.gather(*coros)
    grok_result_pair = gathered[0] or ("", [])
    grok_result: str = grok_result_pair[0] or ""
    grok_annotations: list[dict] = grok_result_pair[1] or []
    idx = 1
    tavily_results = None
    firecrawl_results = None
    if tavily_count > 0:
        tavily_results = gathered[idx]
        idx += 1
    if firecrawl_count > 0:
        firecrawl_results = gathered[idx]

    # Track providers_used / providers_failed based on result content.
    # v3: distinguish "grok" (text only, no native search citations) from
    # "grok-web-search" (text AND structured annotations from Responses API
    # web_search tool). When annotations are present, native live search ran;
    # when absent, Grok answered from training data only and we fell back to
    # heuristic citation extraction.
    if grok_result.strip():
        if grok_annotations:
            providers_used.append("grok-web-search")
        else:
            providers_used.append("grok")
    else:
        providers_failed.append("grok")
    if tavily_count > 0:
        if tavily_results:
            providers_used.append("tavily")
        else:
            providers_failed.append("tavily")
    if firecrawl_count > 0:
        if firecrawl_results:
            providers_used.append("firecrawl")
        else:
            providers_failed.append("firecrawl")

    # v3: prefer native annotations from the Responses API. They're real,
    # structured citations from Grok's actual web_search call — no
    # confabulation, no URL invention, no parsing heuristics needed.
    # If annotations is empty (rare: only when search_mode produced none, or
    # when grok_responses_call fell back to a non-streaming response that
    # didn't include annotations), fall through to the v1/v2/v2.1 heuristic
    # citation extraction chain as a safety net.
    native_sources = normalize_responses_annotations(grok_annotations)
    if native_sources:
        answer = grok_result
        grok_sources = native_sources
        debug(f"grok annotations: {len(native_sources)} native sources extracted")
    else:
        answer, grok_sources = split_answer_and_sources(grok_result)
        if grok_result.strip():
            debug(
                f"grok annotations empty; fell back to heuristic citation parser "
                f"(found {len(grok_sources)} sources)"
            )

    extras = _extras_to_sources(tavily_results, firecrawl_results)
    merged = _merge_sources(grok_sources, extras)

    # Auto-fetch-top: parallel fetch of the top N source URLs via the default
    # fetch chain (Tavily → Firecrawl). Grok is NOT used here to keep bulk
    # fetch fast and upstream-faithful. Imported locally to avoid a circular
    # import at module load.
    fetched_pages: list[dict] = []
    if auto_fetch_top > 0 and merged:
        try:
            from agentic_fetch import fetch_url  # noqa: WPS433 — intentional lazy import
        except ImportError as e:
            debug(f"auto-fetch-top: could not import agentic_fetch.fetch_url: {e}")
        else:
            top_urls = [
                s["url"] for s in merged[:auto_fetch_top] if isinstance(s.get("url"), str)
            ]
            debug(f"auto-fetch-top: fetching {len(top_urls)} pages in parallel")

            async def _fetch_one(u: str) -> dict:
                md, engine = await fetch_url(u, engine="auto")
                return {
                    "url": u,
                    "engine": engine or "none",
                    "markdown": md or "",
                    "ok": bool(md),
                }

            fetched_pages = list(await asyncio.gather(*(_fetch_one(u) for u in top_urls)))

    result = {
        "session_id": session_id,
        "created_at": created_at,
        "model": model,
        "query": query,
        "platform": platform,
        "extra_sources": extra_sources,
        "content": answer,
        "sources": merged,
        "sources_count": len(merged),
        "providers_used": providers_used,
        "providers_failed": providers_failed,
    }
    if fetched_pages:
        result["fetched_pages"] = fetched_pages

    # Persist the session to disk cache so downstream scripts (agentic_rank,
    # agentic_get_sources, agentic_extract --session-id) can compose on it.
    try:
        write_session(session_id, result)
        pruned = prune_sessions()
        if pruned:
            debug(f"pruned {pruned} old sessions")
    except Exception as e:
        debug(f"session cache write failed: {e}")

    return result


def main() -> int:
    parser = argparse.ArgumentParser(
        description="agentic_search — Grok-primary deep web search with optional Tavily/Firecrawl fusion."
    )
    parser.add_argument("--query", required=True, help="Natural-language search query.")
    parser.add_argument("--platform", default="", help="Optional platform focus (e.g., 'GitHub', 'Reddit').")
    parser.add_argument("--model", default="", help="Override GROK_MODEL for this call only.")
    parser.add_argument(
        "--extra-sources",
        type=int,
        default=0,
        help="Number of supplementary results from Tavily/Firecrawl. 0 disables.",
    )
    parser.add_argument(
        "--auto-fetch-top",
        type=int,
        default=0,
        help="After the search, fetch the top N source URLs in parallel via Tavily→Firecrawl and include their markdown in the output under 'fetched_pages'. 0 disables.",
    )
    args = parser.parse_args()

    try:
        result = asyncio.run(
            run(
                args.query,
                args.platform,
                args.model or None,
                args.extra_sources,
                auto_fetch_top=args.auto_fetch_top,
            )
        )
    except SystemExit as e:
        # Config errors raised from _http
        msg = str(e) if e.code != 0 else "configuration error"
        print(json.dumps({"error": msg}, ensure_ascii=False))
        return 1
    except Exception as e:
        print(json.dumps({"error": f"unexpected: {type(e).__name__}: {e}"}, ensure_ascii=False))
        return 1

    print(json.dumps(result, ensure_ascii=False, indent=2))
    return 0


if __name__ == "__main__":
    sys.exit(main())