feat: perplexity: add direct API modes and async Deep Research (#629)

* perplexity: add direct API modes and async Deep Research * perplexity: avoid conflicting search date filters * perplexity: preserve malformed async metadata * Address greptile comments
2026-06-20 21:00:25 -07:00
parent bed9679699
commit 859a63146f
19 changed files with 1348 additions and 97 deletions
@@ -20,6 +20,11 @@ mise.toml
 .coverage
 htmlcov/

+# Local secrets/config. Keep tracked examples if added later.
+.env
+.env.*
+!.env.example
+
 # Root vendor/ is accidental - real vendored client lives at scripts/lib/vendor/bird-search/
 /vendor/

@@ -79,7 +79,7 @@ The project-scoped file is the cleanest pattern for **per-client setups**: drop
 | Bluesky | `BSKY_HANDLE` + `BSKY_APP_PASSWORD` | Bluesky items | yes (app password at bsky.app) |
 | TruthSocial | `TRUTHSOCIAL_TOKEN` | TruthSocial items | yes |
 | Web search | one of: `BRAVE_API_KEY`, `EXA_API_KEY`, `SERPER_API_KEY`, `PARALLEL_API_KEY` | `--auto-resolve` and Step 2 supplements | Brave has a free tier; native WebSearch on Claude Code / Codex / Gemini works as a fallback |
-| Perplexity Deep Research | `OPENROUTER_API_KEY` | `--deep-research` flag (~$0.90/query) | no |
+| Perplexity Sonar / Search API / Deep Research | `PERPLEXITY_API_KEY` (preferred) or `OPENROUTER_API_KEY` (Sonar fallback) | `INCLUDE_SOURCES=perplexity`; `--deep-research` flag (~$0.90/query) | no |
 | Caption-free transcription | `GROQ_API_KEY` (free tier, preferred) or `OPENAI_API_KEY` (paid backstop); requires `ffmpeg` | Whisper transcription for audio/video without captions (groundwork: module shipped, not yet auto-invoked by the engine) | Groq free tier is generous; needs ffmpeg installed |
 | Jobs / careers pages | none for public ATS pages; web backend improves fallback discovery | `--hiring-signals` and strong Hiring Signals in standard company reports | yes |
 | Apify (alternate scraper) | `APIFY_API_TOKEN` | fallback for Reddit/TikTok/Instagram when ScrapeCreators is exhausted | yes (limited) |
@@ -96,6 +96,11 @@ BRAVE_API_KEY=<your-brave-key>
 # Optional sources
 SCRAPECREATORS_API_KEY=<your-scrapecreators-key>
 INCLUDE_SOURCES=tiktok,instagram
+# Add perplexity to INCLUDE_SOURCES when you want the paid Perplexity source.
+# PERPLEXITY_API_KEY=<your-perplexity-key>
+# INCLUDE_SOURCES=tiktok,instagram,perplexity
+# LAST30DAYS_PERPLEXITY_MODE=sonar  # sonar | search | both
+# LAST30DAYS_PERPLEXITY_MODEL=sonar-pro  # sonar | sonar-pro | sonar-reasoning-pro

 # X authentication (one option only)
 AUTH_TOKEN=<your-auth-token>
@@ -120,6 +125,36 @@ After editing: `chmod 600 ~/.config/last30days/.env` (or `chmod 600 .claude/last

 **Troubleshooting:** if a source you expected to see isn't appearing in results, run `python3 scripts/last30days.py --diagnose`. It prints a per-source availability report (which keys were detected, which CLIs are installed, which backends are reachable) without running a full search.

+### Perplexity source modes
+
+Perplexity is a paid opt-in source. A direct `PERPLEXITY_API_KEY` unlocks first-party Perplexity features. `OPENROUTER_API_KEY` remains a Sonar compatibility fallback only; Perplexity Search API and async Deep Research call Perplexity directly.
+
+`LAST30DAYS_PERPLEXITY_MODE` controls normal `perplexity` source runs:
+
+| Value | Behavior | Calls |
+|---|---|---|
+| `sonar` (default) | Sonar synthesis plus citations. | one Sonar call |
+| `search` | Raw ranked Search API rows; best when you want source aggregation over prose. | one Search API call |
+| `both` | Sonar synthesis plus raw ranked Search API rows, deduped by URL. | one Search API call and one Sonar call |
+
+`--deep-research` ignores `LAST30DAYS_PERPLEXITY_MODE` and uses `sonar-deep-research`. With `PERPLEXITY_API_KEY`, it submits to Perplexity's async Sonar endpoint and polls with a hard wall-clock timeout. The async request uses a deterministic idempotency key derived from the request body. If the request is still running at timeout, fails remotely, or polling hits a transport/rate-limit error after the async id exists, the raw artifact records the async request id, idempotency key, last status, lifecycle timestamps returned by Perplexity, poll count, and timeout/error fields so you can inspect or resume by id outside the run. With only `OPENROUTER_API_KEY`, it keeps the OpenRouter synchronous fallback.
+
+Perplexity-specific env vars:
+
+| Env var | Default | Applies to | Notes |
+|---|---|---|---|
+| `LAST30DAYS_PERPLEXITY_MODE` | `sonar` | normal Perplexity source runs | `sonar`, `search`, or `both`; `search` and `both` require `PERPLEXITY_API_KEY`. |
+| `LAST30DAYS_PERPLEXITY_MODEL` | `sonar-pro` | direct Sonar only | Supported: `sonar`, `sonar-pro`, `sonar-reasoning-pro`. `--deep-research` forces `sonar-deep-research`. |
+| `LAST30DAYS_PERPLEXITY_MAX_RESULTS` | `10` | Search API | Clamped to Perplexity's 1..20 range. |
+| `LAST30DAYS_PERPLEXITY_SEARCH_CONTEXT_SIZE` | provider default | Search API | `low`, `medium`, or `high`; omitted unless set. |
+| `LAST30DAYS_PERPLEXITY_SEARCH_MODE` | provider default | direct Sonar | `web`, `academic`, or `sec`. |
+| `LAST30DAYS_PERPLEXITY_DOMAIN_FILTER` | unset | Search API and direct Sonar | Comma-separated domains, max 20. |
+| `LAST30DAYS_PERPLEXITY_LANGUAGE_FILTER` | unset | Search API and direct Sonar | Comma-separated ISO 639-1 language codes, max 20. |
+| `LAST30DAYS_PERPLEXITY_COUNTRY` | unset | Search API | Two-letter country code such as `US`. |
+| `LAST30DAYS_PERPLEXITY_RECENCY_FILTER` | unset | Search API and direct Sonar | `hour`, `day`, `week`, `month`, or `year`. |
+| `LAST30DAYS_PERPLEXITY_REASONING_EFFORT` | unset | direct Sonar | `minimal`, `low`, `medium`, or `high`. |
+| `LAST30DAYS_PERPLEXITY_DEEP_TIMEOUT_SECONDS` | `600` | direct async Deep Research | Wall-clock polling deadline. |
+
 ### Encrypted credential sources (Keychain / pass)

 If you'd rather not keep keys in a plaintext `.env`, the loader has two
@@ -184,7 +219,7 @@ Accepts the same comma-separated names and aliases as `--search` (`web` → grou
 1. **Gemini** - `GOOGLE_API_KEY` / `GEMINI_API_KEY` / `GOOGLE_GENAI_API_KEY`
 2. **OpenAI** - `OPENAI_API_KEY` (or Codex auth at `~/.codex/auth.json`)
 3. **xAI** - `XAI_API_KEY`
-4. **OpenRouter** - `OPENROUTER_API_KEY` (also unlocks `--deep-research`)
+4. **OpenRouter** - `OPENROUTER_API_KEY` (Sonar fallback for the Perplexity source / `--deep-research`; also usable as a reasoning provider)
 5. **Local / deterministic** - always available, lowest quality

 When you invoke `/last30days` from Claude Code, Codex, or Gemini, the host model **is** the reasoning provider for plan + synthesis - you don't need any of the keys above unless you also run the script headlessly (cron, CI, watchlist).
@@ -70,7 +70,7 @@ If you're meeting with a CEO, have you read all their tweets and YouTube transcr
 | **Threads** | The post-Twitter text layer. Conversations from creators and brands. |
 | **Pinterest** | Visual discovery. Pins, saves, and comments on products and ideas. |
 | **Bluesky** | The decentralized social layer. AT Protocol posts from the post-Twitter migration. |
-| **Perplexity** | Grounded web search with citations via Sonar Pro. |
+| **Perplexity** | Grounded Sonar synthesis, raw Search API rows, and Deep Research. |
 | **Web** | The editorial coverage, the blog comparisons. One signal of many, not the only one. |

 Community contributors keep adding more. Truth Social, Xiaohongshu (RED), and others are in the engine with more on the way.
@@ -158,7 +158,7 @@ Say "eli5 on" after any research run. The synthesis rewrites in plain language.
 - **TikTok, Instagram, Threads.** All three activate automatically once `SCRAPECREATORS_API_KEY` is set — same key, same per-call cost. Suppress any of them with `EXCLUDE_SOURCES=tiktok,instagram,threads` (any comma-separated subset).
 - **Pinterest.** Per-query opt-in (visual pins, narrow utility): the model passes `--search=pinterest` for the runs that need it. Requires `SCRAPECREATORS_API_KEY`.
 - **YouTube comments + transcript fallback.** Both activate automatically once `SCRAPECREATORS_API_KEY` is set, the same default-on backup tier. Transcripts only fall back to ScrapeCreators when yt-dlp fails (no credit spent on success); comments are bounded to the top few videos (~3 extra calls per run). Suppress comments with `EXCLUDE_SOURCES=youtube_comments`. **TikTok comments** stay opt-in via `INCLUDE_SOURCES=tiktok_comments`. Surface top comments with vote counts the same way Reddit does.
- **Perplexity Sonar.** Grounded web search with citations via OpenRouter. Add `OPENROUTER_API_KEY` and `INCLUDE_SOURCES=perplexity` (it's a separate paid API — opt-in keeps you from being surprise-billed).
+- **Perplexity Sonar / Search API / Deep Research.** Grounded web search via direct Perplexity (`PERPLEXITY_API_KEY`) or OpenRouter Sonar fallback (`OPENROUTER_API_KEY`). Add one of those keys plus `INCLUDE_SOURCES=perplexity` (it's a separate paid API - opt-in keeps you from being surprise-billed). Direct Perplexity can return Sonar synthesis, raw ranked Search API rows, or both.
 - **Polymarket noise filtering.** Common-word disambiguation prevents "Apple" from matching "Will Apple release a car?"
 - **Resilient Reddit.** Timeout budgets and runtime fallback. One slow thread doesn't kill the whole run.
 - **Fun judge v2.** Humor scoring baked into the narrative. Reddit's cleverest one-liners mixed into the synthesis where they fit, not dumped in a separate section.
@@ -287,7 +287,7 @@ These platforms don't have relationships with each other. X doesn't know what Re
 | YouTube | `brew install yt-dlp` | Free |
 | Bluesky | App password from bsky.app | Free |
 | TikTok + Instagram + Threads + Pinterest + YouTube comments | ScrapeCreators key | 100 free credits, then PAYG |
-| Perplexity Sonar | OpenRouter key | Pay as you go |
+| Perplexity Sonar / Search API / Deep Research | Perplexity key, or OpenRouter key as Sonar fallback | Pay as you go |
 | Web search | Brave Search key | 2,000 free queries/month |

 ### macOS Keychain (optional)
@@ -6,6 +6,13 @@ set -euo pipefail

 PROJECT_ENV=".claude/last30days.env"
 GLOBAL_ENV="$HOME/.config/last30days/.env"
+if [[ "${LAST30DAYS_CONFIG_DIR+x}" == "x" ]]; then
+  if [[ -n "$LAST30DAYS_CONFIG_DIR" ]]; then
+    GLOBAL_ENV="$LAST30DAYS_CONFIG_DIR/.env"
+  else
+    GLOBAL_ENV=""
+  fi
+fi

 # Ensure LAST30DAYS_MEMORY_DIR exists for HTML-brief / raw-markdown saves.
 # SKILL.md and the engine default this via the same env-var fallback. Fresh
@@ -33,6 +40,25 @@ check_perms() {
  fi
 }

+trim_ws() {
+  local s="$1"
+  s="${s#"${s%%[![:space:]]*}"}"
+  s="${s%"${s##*[![:space:]]}"}"
+  printf '%s' "$s"
+}
+
+strip_outer_quotes() {
+  local s="$1"
+  if [[ ${#s} -ge 2 ]]; then
+    if [[ "${s:0:1}" == '"' && "${s: -1}" == '"' ]]; then
+      s="${s:1:${#s}-2}"
+    elif [[ "${s:0:1}" == "'" && "${s: -1}" == "'" ]]; then
+      s="${s:1:${#s}-2}"
+    fi
+  fi
+  printf '%s' "$s"
+}
+
 # Load env file into variables for inspection (without exporting)
 load_env_vars() {
  local file="$1"
@@ -41,8 +67,8 @@ load_env_vars() {
      # Skip comments, empty lines
      [[ "$key" =~ ^[[:space:]]*# ]] && continue
      [[ -z "$key" ]] && continue
-      key=$(echo "$key" | sed 's/^[[:space:]]*//;s/[[:space:]]*$//')
-      value=$(echo "$value" | sed -e 's/^[[:space:]]*//;s/[[:space:]]*$//' -e 's/^["'\''"]//;s/["'\''"]$//')
+      key="$(trim_ws "$key")"
+      value="$(strip_outer_quotes "$(trim_ws "$value")")"
      # Strip inline comments (# preceded by whitespace) to prevent
      # command substitution in backtick-containing comments
      value="${value%%[[:space:]]#*}"
@@ -174,16 +200,15 @@ if [[ -n "$HAS_BSKY" ]]; then
 fi
 if [[ -n "$HAS_SCRAPECREATORS" ]]; then
  # Start with Reddit comments + TikTok + Instagram, subtract any in EXCLUDE_SOURCES.
-  # Normalise EXCLUDED (lowercase + collapse whitespace around commas + strip outer
-  # whitespace) so the matching mirrors pipeline.py's .strip().lower() parsing.
+  # Normalise EXCLUDED by removing whitespace; case-insensitive matches below
+  # mirror pipeline.py's .strip().lower() parsing without requiring sed/tr.
  SC_ADD=3
  EXCLUDED="${ENV_EXCLUDE_SOURCES:-${EXCLUDE_SOURCES:-}}"
-  EXCLUDED_NORM=$(printf '%s' "$EXCLUDED" | tr '[:upper:]' '[:lower:]' \
-    | sed -E 's/[[:space:]]*,[[:space:]]*/,/g; s/^[[:space:]]+//; s/[[:space:]]+$//')
-  if [[ ",$EXCLUDED_NORM," == *",tiktok,"* ]]; then
+  EXCLUDED_NORM="${EXCLUDED//[[:space:]]/}"
+  if [[ ",$EXCLUDED_NORM," == *",[Tt][Ii][Kk][Tt][Oo][Kk],"* ]]; then
    SC_ADD=$((SC_ADD - 1))
  fi
-  if [[ ",$EXCLUDED_NORM," == *",instagram,"* ]]; then
+  if [[ ",$EXCLUDED_NORM," == *",[Ii][Nn][Ss][Tt][Aa][Gg][Rr][Aa][Mm],"* ]]; then
    SC_ADD=$((SC_ADD - 1))
  fi
  SOURCE_COUNT=$((SOURCE_COUNT + SC_ADD))
@@ -19,6 +19,7 @@ metadata:
        - OPENAI_API_KEY
        - XAI_API_KEY
        - OPENROUTER_API_KEY
+        - PERPLEXITY_API_KEY
        - PARALLEL_API_KEY
        - BRAVE_API_KEY
        - APIFY_API_TOKEN
@@ -396,6 +397,8 @@ SKILL_DIR="<absolute path of the directory containing the SKILL.md you just Read

 - If EXCLUDE_SOURCES is set (comma-separated, case-insensitive): drop any matching source from ACTIVE_SOURCES_LIST before displaying

+**Perplexity source:** use it only when the user asks for Perplexity, Deep Research, or paid grounded synthesis, or when `perplexity` is already enabled in `INCLUDE_SOURCES` / `--search`. Direct `PERPLEXITY_API_KEY` supports Sonar synthesis, Search API rows, and async Deep Research. `OPENROUTER_API_KEY` is only a Sonar fallback. Normal runs default to `LAST30DAYS_PERPLEXITY_MODE=sonar`; use `search` for raw ranked web rows, `both` for synthesis plus rows, and `--deep-research` for `sonar-deep-research` with a 600s default wall timeout. A local Deep Research timeout is not a failed API key; inspect the raw artifact's async request id/status and resume by id if needed.
+

 Then display (use "and more" if 5+ sources, otherwise list all with Oxford comma):

@@ -741,7 +744,7 @@ Topic A (the main topic, first in the vs-string) uses outer `--x-handle`, `--x-r

 **Why --competitors-plan over --competitors-list:** without per-entity handles/subs, peer sub-runs run with deterministic single-word planner queries and produce visibly thinner evidence than the main topic. The Resolved Entities block in stdout makes the gap visible — dashes for a peer = you skipped its Step 0.55.

-**Engine-internal auto-resolve (headless fallback):** if the engine detects BRAVE_API_KEY / EXA_API_KEY / SERPER_API_KEY / PARALLEL_API_KEY / OPENROUTER_API_KEY, it runs its own per-entity `resolve.auto_resolve()` before each sub-run. The hosting-model path does NOT need those keys — you are the WebSearch. The engine's auto-resolve is the cron/CI fallback for when no reasoning model is driving.
+**Engine-internal auto-resolve (headless fallback):** if the engine detects BRAVE_API_KEY / EXA_API_KEY / SERPER_API_KEY / PARALLEL_API_KEY / PERPLEXITY_API_KEY / OPENROUTER_API_KEY, it runs its own per-entity `resolve.auto_resolve()` before each sub-run. The hosting-model path does NOT need those keys — you are the WebSearch. The engine's auto-resolve is the cron/CI fallback for when no reasoning model is driving.

 **Output:** one `{slug}-raw.md` per entity in `--save-dir` plus the merged comparison on stdout. Synthesis contract identical to the vs-mode protocol above.

@@ -1791,7 +1794,7 @@ Want another prompt? Just tell me what you're creating next.
 - Sends search queries to Polymarket Gamma API (`gamma-api.polymarket.com`) for prediction market discovery (free, no auth)
 - Runs `yt-dlp` locally for YouTube search and transcript extraction (no API key, public data)
 - Sends search queries to ScrapeCreators API (`api.scrapecreators.com`) for TikTok and Instagram search, transcript/caption extraction (PAYG after 100 free credits)
- Optionally sends search queries to Brave Search API, Parallel AI API, or OpenRouter API for web search
+- Optionally sends search queries to Brave Search API, Parallel AI API, Perplexity API (`api.perplexity.ai`), or OpenRouter API for web search / synthesis
 - Fetches public Reddit thread data from `reddit.com` for engagement metrics
 - Stores research findings in local SQLite database (watchlist mode only)
 - Saves research briefings as .md files to `LAST30DAYS_MEMORY_DIR` (defaults to `~/Documents/Last30Days`)
@@ -298,7 +298,7 @@ def build_parser() -> argparse.ArgumentParser:
                        choices=["auto", "brave", "exa", "serper", "parallel", "none"],
                        help="Web search backend (default: auto, tries Brave then Exa then Serper then Parallel)")
    parser.add_argument("--deep-research", action="store_true",
-                        help="Use Perplexity Deep Research (~$0.90/query) for in-depth analysis. Requires OPENROUTER_API_KEY.")
+                        help="Use Perplexity Deep Research (~$0.90/query) for in-depth analysis. Requires PERPLEXITY_API_KEY or OPENROUTER_API_KEY.")
    parser.add_argument("--hiring-signals", action="store_true",
                        help="Analyze public jobs/careers postings as evidence-backed company focus signals.")
    parser.add_argument("--plan", help="JSON query plan (skips internal LLM planner). Can be a JSON string or a file path.")
@@ -792,8 +792,8 @@ def main() -> int:

        # --deep-research: auto-enable perplexity source and set deep flag
        if args.deep_research:
-            if not config.get("OPENROUTER_API_KEY"):
-                print("Error: --deep-research requires OPENROUTER_API_KEY", file=sys.stderr)
+            if not (config.get("PERPLEXITY_API_KEY") or config.get("OPENROUTER_API_KEY")):
+                print("Error: --deep-research requires PERPLEXITY_API_KEY or OPENROUTER_API_KEY", file=sys.stderr)
                sys.exit(1)
            config["_deep_research"] = True
            # Auto-enable perplexity in INCLUDE_SOURCES
@@ -889,7 +889,7 @@ def main() -> int:
                        "\n"
                        "HEADLESS / CRON PATH (no hosting model available): set "
                        "BRAVE_API_KEY / EXA_API_KEY / SERPER_API_KEY / PARALLEL_API_KEY / "
-                        "OPENROUTER_API_KEY and re-run.\n"
+                        "PERPLEXITY_API_KEY / OPENROUTER_API_KEY and re-run.\n"
                        "\n"
                        "MINIMUM ESCAPE HATCH: pass --competitors-list 'A,B,C' to skip "
                        "discovery. Without --competitors-plan, peer sub-runs fall back to "
@@ -42,7 +42,7 @@ KEYCHAIN_KEYS = (
    "GOOGLE_GENAI_API_KEY", "SCRAPECREATORS_API_KEY", "APIFY_API_TOKEN",
    "AUTH_TOKEN", "CT0", "BSKY_HANDLE", "BSKY_APP_PASSWORD",
    "TRUTHSOCIAL_TOKEN", "BRAVE_API_KEY", "EXA_API_KEY", "SERPER_API_KEY",
-    "OPENROUTER_API_KEY", "PARALLEL_API_KEY", "XQUIK_API_KEY",
+    "OPENROUTER_API_KEY", "PERPLEXITY_API_KEY", "PARALLEL_API_KEY", "XQUIK_API_KEY",
    "XIAOHONGSHU_API_BASE",
 )

@@ -408,6 +408,18 @@ def get_config() -> dict[str, Any]:
        ('EXA_API_KEY', None),
        ('SERPER_API_KEY', None),
        ('OPENROUTER_API_KEY', None),
+        ('PERPLEXITY_API_KEY', None),
+        ('LAST30DAYS_PERPLEXITY_MODE', 'sonar'),
+        ('LAST30DAYS_PERPLEXITY_MODEL', None),
+        ('LAST30DAYS_PERPLEXITY_MAX_RESULTS', None),
+        ('LAST30DAYS_PERPLEXITY_SEARCH_CONTEXT_SIZE', None),
+        ('LAST30DAYS_PERPLEXITY_SEARCH_MODE', None),
+        ('LAST30DAYS_PERPLEXITY_DOMAIN_FILTER', None),
+        ('LAST30DAYS_PERPLEXITY_LANGUAGE_FILTER', None),
+        ('LAST30DAYS_PERPLEXITY_COUNTRY', None),
+        ('LAST30DAYS_PERPLEXITY_RECENCY_FILTER', None),
+        ('LAST30DAYS_PERPLEXITY_REASONING_EFFORT', None),
+        ('LAST30DAYS_PERPLEXITY_DEEP_TIMEOUT_SECONDS', '600'),
        ('PARALLEL_API_KEY', None),
        ('XQUIK_API_KEY', None),
        # Host-native search signal: set by the SKILL.md agent-host path when the
@@ -1,22 +1,77 @@
-"""Perplexity Sonar Pro / Deep Research via OpenRouter API.
+"""Perplexity Sonar, Search API, and Deep Research.

-Queries Perplexity models through OpenRouter for AI-synthesized research
-with citation annotations. Returns normalized items with synthesis text
-and individual citation entries.
+Direct Perplexity keys are preferred so the source can use first-party Search
+API results and async Deep Research. OpenRouter remains a Sonar compatibility
+fallback when no direct Perplexity key is configured.
 """

 from __future__ import annotations

+import hashlib
+import json
+import random
 import sys
+import time
+from datetime import datetime
 from urllib.parse import urlparse

 from . import http, log


 OPENROUTER_URL = "https://openrouter.ai/api/v1/chat/completions"
+PERPLEXITY_URL = "https://api.perplexity.ai/v1/sonar"
+PERPLEXITY_SEARCH_URL = "https://api.perplexity.ai/search"
+PERPLEXITY_ASYNC_URL = "https://api.perplexity.ai/v1/async/sonar"

-MODEL_SONAR_PRO = "perplexity/sonar-pro"
-MODEL_DEEP_RESEARCH = "perplexity/sonar-deep-research"
+OPENROUTER_MODEL_SONAR_PRO = "perplexity/sonar-pro"
+OPENROUTER_MODEL_DEEP_RESEARCH = "perplexity/sonar-deep-research"
+PERPLEXITY_MODEL_SONAR = "sonar"
+PERPLEXITY_MODEL_SONAR_PRO = "sonar-pro"
+PERPLEXITY_MODEL_REASONING_PRO = "sonar-reasoning-pro"
+PERPLEXITY_MODEL_DEEP_RESEARCH = "sonar-deep-research"
+PERPLEXITY_MODE_SONAR = "sonar"
+PERPLEXITY_MODE_SEARCH = "search"
+PERPLEXITY_MODE_BOTH = "both"
+PERPLEXITY_DEFAULT_DEEP_TIMEOUT_SECONDS = 600
+PERPLEXITY_DEEP_INITIAL_POLL_DELAY_SECONDS = 5.0
+PERPLEXITY_DEEP_MAX_POLL_DELAY_SECONDS = 60.0
+
+DIRECT_MODELS = {
+    PERPLEXITY_MODEL_SONAR,
+    PERPLEXITY_MODEL_SONAR_PRO,
+    PERPLEXITY_MODEL_REASONING_PRO,
+    PERPLEXITY_MODEL_DEEP_RESEARCH,
+}
+DIRECT_MODES = {
+    PERPLEXITY_MODE_SONAR,
+    PERPLEXITY_MODE_SEARCH,
+    PERPLEXITY_MODE_BOTH,
+}
+SEARCH_CONTEXT_SIZES = {"low", "medium", "high"}
+SEARCH_RECENCY_FILTERS = {"hour", "day", "week", "month", "year"}
+SONAR_SEARCH_MODES = {"web", "academic", "sec"}
+REASONING_EFFORTS = {"minimal", "low", "medium", "high"}
+
+
+class AsyncDeepResearchTimeout(TimeoutError):
+    def __init__(self, metadata: dict):
+        timeout_seconds = metadata.get("asyncTimeoutSeconds") or "unknown"
+        super().__init__(f"Async Deep Research exceeded {timeout_seconds}s wall timeout")
+        self.metadata = metadata
+
+
+class AsyncDeepResearchFailed(RuntimeError):
+    def __init__(self, metadata: dict):
+        message = metadata.get("asyncErrorMessage") or "Async Deep Research failed"
+        super().__init__(str(message))
+        self.metadata = metadata
+
+
+class AsyncDeepResearchPollError(RuntimeError):
+    def __init__(self, metadata: dict):
+        message = metadata.get("asyncPollError") or "Async Deep Research poll failed"
+        super().__init__(str(message))
+        self.metadata = metadata


 def _log(msg: str):
@@ -27,30 +82,433 @@ def _domain(url: str) -> str:
    return urlparse(url).netloc.strip().lower()


-def search(
+def _provider(config: dict, deep: bool) -> tuple[str, str, str, str] | None:
+    """Return (provider, api_key, url, model), preferring direct Perplexity."""
+    if config.get("PERPLEXITY_API_KEY"):
+        model = _direct_model(config, deep)
+        url = PERPLEXITY_ASYNC_URL if deep else PERPLEXITY_URL
+        return "perplexity", config["PERPLEXITY_API_KEY"], url, model
+    if config.get("OPENROUTER_API_KEY"):
+        model = OPENROUTER_MODEL_DEEP_RESEARCH if deep else OPENROUTER_MODEL_SONAR_PRO
+        return "openrouter", config["OPENROUTER_API_KEY"], OPENROUTER_URL, model
+    return None
+
+
+def _config_text(config: dict, key: str) -> str:
+    return str(config.get(key) or "").strip()
+
+
+def _csv_values(raw: str, limit: int | None = None) -> list[str]:
+    values = [part.strip() for part in raw.split(",") if part.strip()]
+    return values[:limit] if limit is not None else values
+
+
+def _direct_model(config: dict, deep: bool) -> str:
+    if deep:
+        return PERPLEXITY_MODEL_DEEP_RESEARCH
+    model = _config_text(config, "LAST30DAYS_PERPLEXITY_MODEL") or PERPLEXITY_MODEL_SONAR_PRO
+    if model not in DIRECT_MODELS:
+        _log(f"Unsupported LAST30DAYS_PERPLEXITY_MODEL={model!r}; using sonar-pro")
+        return PERPLEXITY_MODEL_SONAR_PRO
+    if model == PERPLEXITY_MODEL_DEEP_RESEARCH:
+        return PERPLEXITY_MODEL_SONAR_PRO
+    return model
+
+
+def _mode(config: dict, provider: str, deep: bool) -> str:
+    if deep:
+        return PERPLEXITY_MODE_SONAR
+    mode = (_config_text(config, "LAST30DAYS_PERPLEXITY_MODE") or PERPLEXITY_MODE_SONAR).lower()
+    if mode not in DIRECT_MODES:
+        _log(f"Unsupported LAST30DAYS_PERPLEXITY_MODE={mode!r}; using sonar")
+        return PERPLEXITY_MODE_SONAR
+    if provider != "perplexity" and mode != PERPLEXITY_MODE_SONAR:
+        _log("Search API modes require PERPLEXITY_API_KEY; using OpenRouter Sonar fallback")
+        return PERPLEXITY_MODE_SONAR
+    return mode
+
+
+def _positive_int(raw: object, default: int, min_value: int, max_value: int | None = None) -> int:
+    try:
+        value = int(str(raw).strip())
+    except (TypeError, ValueError):
+        return default
+    value = max(value, min_value)
+    if max_value is not None:
+        value = min(value, max_value)
+    return value
+
+
+def _mmddyyyy(date: str | None) -> str | None:
+    if not date:
+        return None
+    try:
+        return datetime.strptime(date, "%Y-%m-%d").strftime("%m/%d/%Y")
+    except ValueError:
+        return None
+
+
+def _usage(data: dict) -> dict:
+    usage = data.get("usage")
+    return usage if isinstance(usage, dict) else {}
+
+
+def _idempotency_key(json_data: dict) -> str:
+    payload = json.dumps(json_data, sort_keys=True, separators=(",", ":"))
+    digest = hashlib.sha256(payload.encode("utf-8")).hexdigest()[:32]
+    return f"last30days:{digest}"
+
+
+def _async_metadata(
+    data: dict,
+    request_id: str,
+    timeout_seconds: int,
+    idempotency_key: str,
+    poll_count: int,
+    local_status: str,
+) -> dict:
+    metadata = {
+        "async": True,
+        "asyncRequestId": request_id,
+        "asyncStatus": data.get("status"),
+        "asyncTimeoutSeconds": timeout_seconds,
+        "asyncIdempotencyKey": idempotency_key,
+        "asyncPollCount": poll_count,
+        "asyncLocalStatus": local_status,
+        "asyncCreatedAt": data.get("created_at"),
+        "asyncStartedAt": data.get("started_at"),
+        "asyncCompletedAt": data.get("completed_at"),
+        "asyncFailedAt": data.get("failed_at"),
+        "asyncErrorMessage": data.get("error_message"),
+    }
+    return {k: v for k, v in metadata.items() if v is not None}
+
+
+def _error_artifact(exc: Exception) -> dict:
+    artifact = {
+        "error": type(exc).__name__,
+        "message": str(exc)[:200],
+    }
+    if isinstance(exc, http.HTTPError):
+        artifact["statusCode"] = exc.status_code
+    return artifact
+
+
+def _empty_async_sonar_artifact(
+    provider: str,
+    model: str,
+    deep: bool,
+    query: str,
+    data: dict,
+    async_artifact: dict,
+    error: str,
+    message: str,
+) -> dict:
+    if not async_artifact:
+        return {}
+    artifact = {
+        "label": "perplexity",
+        "provider": provider,
+        "mode": PERPLEXITY_MODE_SONAR,
+        "endpoint": "async-sonar",
+        "model": model,
+        "deep": deep,
+        "query": query,
+        "error": error,
+        "synthesisLength": 0,
+        "citationCount": 0,
+        "usage": _usage(data),
+        **async_artifact,
+    }
+    if not artifact.get("asyncErrorMessage"):
+        artifact["asyncErrorMessage"] = message
+    return artifact
+
+
+def _build_sonar_payload(prompt: str, model: str, date_range: tuple[str, str], config: dict) -> dict:
+    payload = {
+        "model": model,
+        "messages": [{"role": "user", "content": prompt}],
+    }
+
+    from_date, to_date = date_range
+    web_options: dict[str, object] = {}
+    search_mode = _config_text(config, "LAST30DAYS_PERPLEXITY_SEARCH_MODE").lower()
+    if search_mode in SONAR_SEARCH_MODES:
+        web_options["search_mode"] = search_mode
+
+    domains = _csv_values(_config_text(config, "LAST30DAYS_PERPLEXITY_DOMAIN_FILTER"), limit=20)
+    if domains:
+        web_options["search_domain_filter"] = domains
+
+    languages = _csv_values(_config_text(config, "LAST30DAYS_PERPLEXITY_LANGUAGE_FILTER"), limit=20)
+    if languages:
+        web_options["search_language_filter"] = languages
+
+    recency = _config_text(config, "LAST30DAYS_PERPLEXITY_RECENCY_FILTER").lower()
+    if recency in SEARCH_RECENCY_FILTERS:
+        web_options["search_recency_filter"] = recency
+
+    after = _mmddyyyy(from_date)
+    before = _mmddyyyy(to_date)
+    if after:
+        web_options["search_after_date_filter"] = after
+    if before:
+        web_options["search_before_date_filter"] = before
+
+    if web_options:
+        payload["web_search_options"] = web_options
+
+    effort = _config_text(config, "LAST30DAYS_PERPLEXITY_REASONING_EFFORT").lower()
+    if effort in REASONING_EFFORTS:
+        payload["reasoning_effort"] = effort
+
+    return payload
+
+
+def _build_search_payload(query: str, date_range: tuple[str, str], config: dict) -> dict:
+    from_date, to_date = date_range
+    payload: dict[str, object] = {
+        "query": query,
+        "max_results": _positive_int(config.get("LAST30DAYS_PERPLEXITY_MAX_RESULTS"), 10, 1, 20),
+    }
+
+    context_size = _config_text(config, "LAST30DAYS_PERPLEXITY_SEARCH_CONTEXT_SIZE").lower()
+    if context_size in SEARCH_CONTEXT_SIZES:
+        payload["search_context_size"] = context_size
+
+    country = _config_text(config, "LAST30DAYS_PERPLEXITY_COUNTRY").upper()
+    if len(country) == 2:
+        payload["country"] = country
+
+    domains = _csv_values(_config_text(config, "LAST30DAYS_PERPLEXITY_DOMAIN_FILTER"), limit=20)
+    if domains:
+        payload["search_domain_filter"] = domains
+
+    languages = _csv_values(_config_text(config, "LAST30DAYS_PERPLEXITY_LANGUAGE_FILTER"), limit=20)
+    if languages:
+        payload["search_language_filter"] = languages
+
+    after = _mmddyyyy(from_date)
+    before = _mmddyyyy(to_date)
+    if after:
+        payload["search_after_date_filter"] = after
+    if before:
+        payload["search_before_date_filter"] = before
+
+    # Perplexity Search API rejects search_recency_filter when explicit
+    # published-date filters are present. last30days already passes an exact
+    # date range, so prefer that and keep recency only for undated callers.
+    recency = _config_text(config, "LAST30DAYS_PERPLEXITY_RECENCY_FILTER").lower()
+    if recency in SEARCH_RECENCY_FILTERS and not (after or before):
+        payload["search_recency_filter"] = recency
+
+    return payload
+
+
+def _append_citation(citations: list[dict], seen_urls: set[str], citation: dict) -> None:
+    url = (citation.get("url") or "").strip()
+    if not url or url in seen_urls:
+        return
+    seen_urls.add(url)
+    citations.append({
+        "url": url,
+        "title": citation.get("title") or "",
+        "snippet": citation.get("snippet") or "",
+        "date": citation.get("date"),
+    })
+
+
+def _extract_citations(data: dict, choice: dict) -> list[dict]:
+    """Extract citations from direct Perplexity and OpenRouter response shapes."""
+    citations: list[dict] = []
+    seen_urls: set[str] = set()
+
+    search_results_by_url: dict[str, dict] = {}
+    for result in data.get("search_results") or []:
+        if not isinstance(result, dict):
+            continue
+        url = (result.get("url") or "").strip()
+        if not url:
+            continue
+        search_results_by_url[url] = result
+        _append_citation(citations, seen_urls, result)
+
+    for url in data.get("citations") or []:
+        if not isinstance(url, str):
+            continue
+        result = search_results_by_url.get(url, {})
+        _append_citation(citations, seen_urls, {
+            "url": url,
+            "title": result.get("title") or _domain(url),
+            "snippet": result.get("snippet") or "",
+            "date": result.get("date"),
+        })
+
+    annotations = choice.get("message", {}).get("annotations", [])
+    for ann in annotations or []:
+        if not isinstance(ann, dict):
+            continue
+        url_citation = ann.get("url_citation", {})
+        if not isinstance(url_citation, dict):
+            continue
+        _append_citation(citations, seen_urls, {
+            "url": url_citation.get("url") or "",
+            "title": url_citation.get("title") or "",
+        })
+
+    return citations
+
+
+def _poll_async_sonar(json_data: dict, headers: dict, config: dict) -> tuple[dict, dict]:
+    timeout_seconds = _positive_int(
+        config.get("LAST30DAYS_PERPLEXITY_DEEP_TIMEOUT_SECONDS"),
+        PERPLEXITY_DEFAULT_DEEP_TIMEOUT_SECONDS,
+        1,
+        None,
+    )
+    idempotency_key = _idempotency_key(json_data)
+    created = http.post(
+        PERPLEXITY_ASYNC_URL,
+        {"request": json_data, "idempotency_key": idempotency_key},
+        headers=headers,
+        timeout=30,
+        retries=2,
+    )
+    request_id = created.get("id")
+    if not request_id:
+        raise http.HTTPError("Async Deep Research response missing id")
+
+    deadline = time.monotonic() + timeout_seconds
+    poll_url = f"{PERPLEXITY_ASYNC_URL}/{request_id}"
+    delay = PERPLEXITY_DEEP_INITIAL_POLL_DELAY_SECONDS
+    last_status = created.get("status")
+    poll_count = 0
+    last_data = created
+    if last_status:
+        _log(f"Deep Research async status: {last_status}")
+
+    while time.monotonic() < deadline:
+        try:
+            data = http.get(poll_url, headers=headers, timeout=30, retries=2)
+        except http.HTTPError as e:
+            metadata = _async_metadata(
+                last_data, request_id, timeout_seconds, idempotency_key, poll_count + 1,
+                "POLL_ERROR",
+            )
+            metadata["asyncPollError"] = str(e)
+            if e.status_code is not None:
+                metadata["asyncPollStatusCode"] = e.status_code
+            raise AsyncDeepResearchPollError(metadata)
+        poll_count += 1
+        last_data = data
+        status = data.get("status")
+        if status and status != last_status:
+            _log(f"Deep Research async status: {status}")
+            last_status = status
+        if status == "COMPLETED":
+            response = data.get("response")
+            if not isinstance(response, dict):
+                metadata = _async_metadata(
+                    data, request_id, timeout_seconds, idempotency_key, poll_count,
+                    "FAILED_REMOTE",
+                )
+                metadata["asyncErrorMessage"] = "Async Deep Research completed without response"
+                raise AsyncDeepResearchFailed(metadata)
+            return response, _async_metadata(
+                data, request_id, timeout_seconds, idempotency_key, poll_count,
+                "COMPLETED_REMOTE",
+            )
+        if status == "FAILED":
+            raise AsyncDeepResearchFailed(_async_metadata(
+                data, request_id, timeout_seconds, idempotency_key, poll_count,
+                "FAILED_REMOTE",
+            ))
+        remaining = deadline - time.monotonic()
+        if remaining <= 0:
+            break
+        jitter = random.uniform(0, 2)
+        time.sleep(min(delay + jitter, max(0.1, remaining)))
+        delay = min(delay * 1.5, PERPLEXITY_DEEP_MAX_POLL_DELAY_SECONDS)
+
+    raise AsyncDeepResearchTimeout(_async_metadata(
+        last_data, request_id, timeout_seconds, idempotency_key, poll_count,
+        "PENDING_REMOTE",
+    ))
+
+
+def _search_api(
    query: str,
    date_range: tuple[str, str],
    config: dict,
-    deep: bool = False,
+    api_key: str,
+) -> tuple[list[dict], dict]:
+    from_date, to_date = date_range
+    headers = {
+        "Authorization": f"Bearer {api_key}",
+        "Content-Type": "application/json",
+    }
+    payload = _build_search_payload(query, date_range, config)
+    _log(f"Querying Perplexity Search API for '{query}' ({from_date} to {to_date})")
+
+    data = http.post(PERPLEXITY_SEARCH_URL, payload, headers=headers, timeout=30)
+    results = data.get("results") or []
+    if not isinstance(results, list):
+        results = []
+
+    items = []
+    for i, result in enumerate(results):
+        if not isinstance(result, dict):
+            continue
+        url = (result.get("url") or "").strip()
+        if not url:
+            continue
+        items.append({
+            "id": f"PXS{i + 1}",
+            "title": result.get("title") or _domain(url),
+            "url": url,
+            "source_domain": _domain(url),
+            "snippet": result.get("snippet") or "",
+            "date": result.get("date"),
+            "relevance": max(0.55, 0.85 - (i * 0.03)),
+            "why_relevant": f"Ranked by Perplexity Search API for '{query}'",
+            "engagement": {},
+            "metadata": {
+                "last_updated": result.get("last_updated"),
+                "perplexity_search_id": data.get("id"),
+            },
+        })
+
+    artifact = {
+        "label": "perplexity",
+        "provider": "perplexity",
+        "mode": PERPLEXITY_MODE_SEARCH,
+        "endpoint": "search",
+        "query": query,
+        "resultCount": len(items),
+        "request": {
+            k: v
+            for k, v in payload.items()
+            if k not in {"query"}
+        },
+        "responseId": data.get("id"),
+        "serverTime": data.get("server_time"),
+    }
+    _log(f"Got {len(items)} Search API results")
+    return items, artifact
+
+
+def _sonar_search(
+    query: str,
+    date_range: tuple[str, str],
+    config: dict,
+    provider: str,
+    api_key: str,
+    url: str,
+    model: str,
+    deep: bool,
 ) -> tuple[list[dict], dict]:
-    """Search via Perplexity Sonar Pro or Deep Research through OpenRouter.
-
-    Args:
-        query: Search topic
-        date_range: (from_date, to_date) as YYYY-MM-DD strings
-        config: Must contain OPENROUTER_API_KEY
-        deep: Use Deep Research model (~$0.90/query) instead of Sonar Pro
-
-    Returns:
-        Tuple of (items list, artifact dict).
-    """
-    api_key = config.get("OPENROUTER_API_KEY")
-    if not api_key:
-        _log("No OPENROUTER_API_KEY configured, skipping")
-        return [], {}
-
    from_date, to_date = date_range
-    model = MODEL_DEEP_RESEARCH if deep else MODEL_SONAR_PRO
    timeout = 120 if deep else 30

    if deep:
@@ -66,56 +524,44 @@ def search(
        "Content-Type": "application/json",
    }

-    json_data = {
-        "model": model,
-        "messages": [{"role": "user", "content": prompt}],
-    }
+    json_data = _build_sonar_payload(prompt, model, date_range, config)
+    if provider != "perplexity":
+        json_data.pop("web_search_options", None)
+        json_data.pop("reasoning_effort", None)

-    _log(f"Querying {model} for '{query}' ({from_date} to {to_date})")
+    _log(f"Querying {provider} {model} for '{query}' ({from_date} to {to_date})")

-    try:
-        data = http.post(OPENROUTER_URL, json_data, headers=headers, timeout=timeout)
-    except http.HTTPError as e:
-        if e.status_code == 401:
-            _log("Invalid OpenRouter API key (401)")
-        elif e.status_code == 429:
-            _log("Rate limited by OpenRouter (429)")
-        else:
-            _log(f"HTTP error: {e}")
-        return [], {}
-    except Exception as e:
-        _log(f"Request failed: {e}")
-        return [], {}
+    async_artifact = {}
+    if provider == "perplexity" and deep:
+        data, async_artifact = _poll_async_sonar(json_data, headers, config)
+    else:
+        data = http.post(url, json_data, headers=headers, timeout=timeout)

    # Parse response
    choices = data.get("choices", [])
    if not choices:
        _log("No choices in response")
-        return [], {}
+        return [], _empty_async_sonar_artifact(
+            provider, model, deep, query, data, async_artifact,
+            "empty_choices",
+            "Async Deep Research completed without choices",
+        )

-    synthesis = choices[0].get("message", {}).get("content", "")
+    choice = choices[0] if isinstance(choices[0], dict) else {}
+    message = choice.get("message")
+    message = message if isinstance(message, dict) else {}
+    synthesis = message.get("content") or ""
+    if not isinstance(synthesis, str):
+        synthesis = ""
    if not synthesis:
        _log("Empty synthesis content")
-        return [], {}
+        return [], _empty_async_sonar_artifact(
+            provider, model, deep, query, data, async_artifact,
+            "empty_synthesis",
+            "Async Deep Research completed with empty synthesis",
+        )

-    # Extract citations from annotations
-    annotations = choices[0].get("message", {}).get("annotations", [])
-    citations = []
-    for ann in annotations:
-        url_citation = ann.get("url_citation", {})
-        url = url_citation.get("url", "")
-        title = url_citation.get("title", "")
-        if url:
-            citations.append({"url": url, "title": title})
-
-    # Deduplicate citations by URL
-    seen_urls = set()
-    unique_citations = []
-    for c in citations:
-        if c["url"] not in seen_urls:
-            seen_urls.add(c["url"])
-            unique_citations.append(c)
-    citations = unique_citations
+    citations = _extract_citations(data, choice)

    _log(f"Got synthesis ({len(synthesis)} chars) with {len(citations)} citations")

@@ -126,7 +572,7 @@ def search(
    snippet = synthesis[:2000]
    items.append({
        "id": "PX1",
-        "title": f"Perplexity {'Deep Research' if deep else 'Sonar Pro'}: {query}",
+        "title": f"Perplexity {'Deep Research' if deep else 'Sonar'}: {query}",
        "url": "",
        "source_domain": "perplexity.ai",
        "snippet": snippet,
@@ -134,7 +580,11 @@ def search(
        "relevance": 0.9,
        "why_relevant": f"AI synthesis of recent activity for '{query}'",
        "engagement": {"citations": len(citations)},
-        "metadata": {"citations": citations},
+        "metadata": {
+            "citations": citations,
+            "usage": _usage(data),
+            **async_artifact,
+        },
    })

    # Individual items for each citation
@@ -144,8 +594,8 @@ def search(
            "title": cit["title"] or _domain(cit["url"]),
            "url": cit["url"],
            "source_domain": _domain(cit["url"]),
-            "snippet": "",
-            "date": None,
+            "snippet": cit.get("snippet") or "",
+            "date": cit.get("date"),
            "relevance": 0.7,
            "why_relevant": f"Cited in Perplexity synthesis for '{query}'",
            "engagement": {"citations": 1},
@@ -154,11 +604,141 @@ def search(

    artifact = {
        "label": "perplexity",
+        "provider": provider,
+        "mode": PERPLEXITY_MODE_SONAR,
+        "endpoint": "async-sonar" if async_artifact else "sonar",
        "model": model,
        "deep": deep,
        "query": query,
        "synthesisLength": len(synthesis),
        "citationCount": len(citations),
+        "usage": _usage(data),
+        **async_artifact,
    }

    return items, artifact
+
+
+def _merge_sonar_and_search(sonar_items: list[dict], search_items: list[dict]) -> list[dict]:
+    if not sonar_items:
+        return search_items
+    merged = sonar_items[:1]
+    seen_urls = {item.get("url") for item in merged if item.get("url")}
+    for item in [*search_items, *sonar_items[1:]]:
+        url = item.get("url")
+        if url and url in seen_urls:
+            continue
+        if url:
+            seen_urls.add(url)
+        merged.append(item)
+    return merged
+
+
+def search(
+    query: str,
+    date_range: tuple[str, str],
+    config: dict,
+    deep: bool = False,
+) -> tuple[list[dict], dict]:
+    """Search via Perplexity Sonar Pro or Deep Research.
+
+    Args:
+        query: Search topic
+        date_range: (from_date, to_date) as YYYY-MM-DD strings
+        config: Must contain PERPLEXITY_API_KEY or OPENROUTER_API_KEY
+        deep: Use Deep Research model (~$0.90/query) instead of Sonar Pro
+
+    Returns:
+        Tuple of (items list, artifact dict).
+    """
+    resolved = _provider(config, deep)
+    if not resolved:
+        _log("No PERPLEXITY_API_KEY or OPENROUTER_API_KEY configured, skipping")
+        return [], {}
+    provider, api_key, url, model = resolved
+    mode = _mode(config, provider, deep)
+
+    try:
+        if mode == PERPLEXITY_MODE_SEARCH:
+            return _search_api(query, date_range, config, api_key)
+        if mode == PERPLEXITY_MODE_BOTH:
+            search_items: list[dict] = []
+            sonar_items: list[dict] = []
+            search_artifact: dict = {}
+            sonar_artifact: dict = {}
+            try:
+                search_items, search_artifact = _search_api(query, date_range, config, api_key)
+            except Exception as e:
+                _log(f"Search API leg failed in both mode: {e}")
+                search_artifact = _error_artifact(e)
+            try:
+                sonar_items, sonar_artifact = _sonar_search(
+                    query, date_range, config, provider, api_key, url, model, deep
+                )
+            except Exception as e:
+                _log(f"Sonar leg failed in both mode: {e}")
+                sonar_artifact = _error_artifact(e)
+            items = _merge_sonar_and_search(sonar_items, search_items)
+            return items, {
+                "label": "perplexity",
+                "provider": "perplexity",
+                "mode": PERPLEXITY_MODE_BOTH,
+                "query": query,
+                "search": search_artifact,
+                "sonar": sonar_artifact,
+                "itemCount": len(items),
+            }
+        return _sonar_search(query, date_range, config, provider, api_key, url, model, deep)
+    except http.HTTPError as e:
+        if e.status_code == 401:
+            _log(f"Invalid {provider} API key (401)")
+        elif e.status_code == 429:
+            _log(f"Rate limited by {provider} (429)")
+        else:
+            _log(f"HTTP error: {e}")
+        return [], {}
+    except AsyncDeepResearchTimeout as e:
+        _log(f"Request timed out: {e}")
+        return [], {
+            "label": "perplexity",
+            "provider": provider,
+            "mode": PERPLEXITY_MODE_SONAR,
+            "endpoint": "async-sonar",
+            "model": model,
+            "deep": deep,
+            "query": query,
+            "error": "timeout",
+            **e.metadata,
+        }
+    except AsyncDeepResearchFailed as e:
+        _log(f"Deep Research failed: {e}")
+        return [], {
+            "label": "perplexity",
+            "provider": provider,
+            "mode": PERPLEXITY_MODE_SONAR,
+            "endpoint": "async-sonar",
+            "model": model,
+            "deep": deep,
+            "query": query,
+            "error": "failed",
+            **e.metadata,
+        }
+    except AsyncDeepResearchPollError as e:
+        _log(f"Deep Research poll failed: {e}")
+        return [], {
+            "label": "perplexity",
+            "provider": provider,
+            "mode": PERPLEXITY_MODE_SONAR,
+            "endpoint": "async-sonar",
+            "model": model,
+            "deep": deep,
+            "query": query,
+            "error": "poll_error",
+            **e.metadata,
+        }
+    except TimeoutError as e:
+        _log(f"Request timed out: {e}")
+        return [], {"label": "perplexity", "provider": provider, "error": "timeout"}
+    except Exception as e:
+        _log(f"Request failed: {e}")
+        return [], {}
@@ -76,6 +76,10 @@ FROM_LANE_COUNT_PER = 8
 MENTION_LANE_COUNT_PER = 5
 RELATED_HANDLE_COUNT_PER = 3

+
+def _has_perplexity_provider(config: dict[str, Any]) -> bool:
+    return bool(config.get("PERPLEXITY_API_KEY") or config.get("OPENROUTER_API_KEY"))
+
 MOCK_AVAILABLE_SOURCES = [
    "reddit",
    "x",
@@ -141,7 +145,7 @@ def available_sources(config: dict[str, Any], requested_sources: list[str] | Non
        available.append("jobs")
    # Perplexity Sonar: opt-in additive source via INCLUDE_SOURCES=perplexity
    include_sources = (config.get("INCLUDE_SOURCES") or "").lower().split(",")
-    if config.get("OPENROUTER_API_KEY") and (
+    if _has_perplexity_provider(config) and (
        "perplexity" in include_sources or (requested_sources and "perplexity" in requested_sources)
    ):
        available.append("perplexity")
@@ -177,10 +181,14 @@ def diagnose(config: dict[str, Any], requested_sources: list[str] | None = None)
        "openai": bool(config.get("OPENAI_API_KEY")) and config.get("OPENAI_AUTH_STATUS") == env.AUTH_STATUS_OK,
        "xai": bool(config.get("XAI_API_KEY")),
        "openrouter": bool(config.get("OPENROUTER_API_KEY")),
+        "perplexity": bool(config.get("PERPLEXITY_API_KEY")),
    }
+    reasoning_provider_available = any(
+        providers_status[name] for name in ("google", "openai", "xai", "openrouter")
+    )
    return {
        "providers": providers_status,
-        "local_mode": not any(providers_status.values()),
+        "local_mode": not reasoning_provider_available,
        "reasoning_provider": (config.get("LAST30DAYS_REASONING_PROVIDER") or "auto").lower(),
        "x_backend": x_status["source"],
        "bird_installed": x_status["bird_installed"],
@@ -76,6 +76,7 @@ def _has_backend(config: dict) -> bool:
        or config.get("SERPER_API_KEY")
        or config.get("PARALLEL_API_KEY")
        or config.get("OPENROUTER_API_KEY")
+        or config.get("PERPLEXITY_API_KEY")
    )


@@ -36,6 +36,7 @@ ALL_KEYS=(
  EXA_API_KEY
  SERPER_API_KEY
  OPENROUTER_API_KEY
+  PERPLEXITY_API_KEY
  PARALLEL_API_KEY
  XQUIK_API_KEY
  XIAOHONGSHU_API_BASE
@@ -38,6 +38,7 @@ ALL_KEYS=(
  EXA_API_KEY
  SERPER_API_KEY
  OPENROUTER_API_KEY
+  PERPLEXITY_API_KEY
  PARALLEL_API_KEY
  XQUIK_API_KEY
  XIAOHONGSHU_API_BASE
@@ -90,6 +90,13 @@ class CliV3Tests(unittest.TestCase):
        )
        self.assertIn("perplexity", available)

+    def test_explicit_perplexity_search_uses_direct_key_without_include_sources(self):
+        available = cli.pipeline.available_sources(
+            {"PERPLEXITY_API_KEY": "test-key", "INCLUDE_SOURCES": ""},
+            requested_sources=["perplexity"],
+        )
+        self.assertIn("perplexity", available)
+
    def test_parse_search_flag_rejects_invalid_or_empty_inputs(self):
        with self.assertRaises(SystemExit):
            cli.parse_search_flag("unknown")
@@ -101,7 +101,7 @@ def clean_env(monkeypatch, tmp_path):
        "OPENAI_API_KEY", "XAI_API_KEY", "BRAVE_API_KEY", "AUTH_TOKEN", "CT0",
        "SCRAPECREATORS_API_KEY", "APIFY_API_TOKEN", "BSKY_HANDLE",
        "BSKY_APP_PASSWORD", "TRUTHSOCIAL_TOKEN", "EXA_API_KEY",
-        "SERPER_API_KEY", "OPENROUTER_API_KEY", "PARALLEL_API_KEY",
+        "SERPER_API_KEY", "OPENROUTER_API_KEY", "PERPLEXITY_API_KEY", "PARALLEL_API_KEY",
        "XQUIK_API_KEY", "GOOGLE_API_KEY", "GEMINI_API_KEY",
        "GOOGLE_GENAI_API_KEY", "INCLUDE_SOURCES", "FROM_BROWSER",
    ]:
@@ -130,7 +130,7 @@ def clean_env(monkeypatch, tmp_path):
        "OPENAI_API_KEY", "XAI_API_KEY", "BRAVE_API_KEY", "AUTH_TOKEN", "CT0",
        "SCRAPECREATORS_API_KEY", "APIFY_API_TOKEN", "BSKY_HANDLE",
        "BSKY_APP_PASSWORD", "TRUTHSOCIAL_TOKEN", "EXA_API_KEY",
-        "SERPER_API_KEY", "OPENROUTER_API_KEY", "PARALLEL_API_KEY",
+        "SERPER_API_KEY", "OPENROUTER_API_KEY", "PERPLEXITY_API_KEY", "PARALLEL_API_KEY",
        "XQUIK_API_KEY", "GOOGLE_API_KEY", "GEMINI_API_KEY",
        "GOOGLE_GENAI_API_KEY", "INCLUDE_SOURCES", "FROM_BROWSER",
    ]:
@@ -46,6 +46,30 @@ class EnvV3Tests(unittest.TestCase):
        path.stat.assert_not_called()
        write.assert_not_called()

+    def test_get_config_includes_perplexity_knobs(self):
+        overrides = {
+            "LAST30DAYS_PERPLEXITY_MODE": "search",
+            "LAST30DAYS_PERPLEXITY_MODEL": "sonar-reasoning-pro",
+            "LAST30DAYS_PERPLEXITY_MAX_RESULTS": "3",
+            "LAST30DAYS_PERPLEXITY_SEARCH_CONTEXT_SIZE": "low",
+            "LAST30DAYS_PERPLEXITY_SEARCH_MODE": "academic",
+            "LAST30DAYS_PERPLEXITY_DOMAIN_FILTER": "example.com",
+            "LAST30DAYS_PERPLEXITY_LANGUAGE_FILTER": "en",
+            "LAST30DAYS_PERPLEXITY_COUNTRY": "US",
+            "LAST30DAYS_PERPLEXITY_RECENCY_FILTER": "week",
+            "LAST30DAYS_PERPLEXITY_REASONING_EFFORT": "high",
+            "LAST30DAYS_PERPLEXITY_DEEP_TIMEOUT_SECONDS": "600",
+        }
+        with mock.patch.object(env, "CONFIG_FILE", None), \
+             mock.patch.object(env, "_find_project_env", return_value=None), \
+             mock.patch("lib.env._load_keychain", return_value={}), \
+             mock.patch("lib.env._load_pass", return_value={}), \
+             mock.patch.dict(os.environ, overrides, clear=False):
+            config = env.get_config()
+
+        for key, value in overrides.items():
+            self.assertEqual(value, config[key])
+

 class ThreadsAvailabilityTests(unittest.TestCase):
    """Threads is in the SC default-on family: same key, same per-call cost
@@ -42,7 +42,7 @@ class FooterNudgeSuppressionTests(unittest.TestCase):
        # triggers deterministically in mock + no-backend. Also strip X cookie
        # credentials so XAI_API_KEY is the unambiguous X backend.
        for key in ("BRAVE_API_KEY", "EXA_API_KEY", "SERPER_API_KEY",
-                    "PARALLEL_API_KEY", "OPENROUTER_API_KEY",
+                    "PARALLEL_API_KEY", "OPENROUTER_API_KEY", "PERPLEXITY_API_KEY",
                    "AUTH_TOKEN", "CT0", "LAST30DAYS_X_BACKEND"):
            env.pop(key, None)
        # Run from a tmpdir so _find_project_env() can't walk up into any
@@ -0,0 +1,534 @@
+import unittest
+from unittest.mock import patch
+
+from lib import perplexity
+
+
+class PerplexityProviderTests(unittest.TestCase):
+    def test_direct_perplexity_key_wins_and_parses_search_results(self):
+        response = {
+            "choices": [{"message": {"content": "Direct synthesis"}}],
+            "citations": ["https://example.com/a"],
+            "search_results": [
+                {
+                    "title": "Example A",
+                    "url": "https://example.com/a",
+                    "date": "2026-06-01",
+                    "snippet": "Direct source snippet",
+                    "source": "web",
+                }
+            ],
+        }
+        with patch("lib.perplexity.http.post", return_value=response) as post:
+            items, artifact = perplexity.search(
+                "test topic",
+                ("2026-05-01", "2026-06-01"),
+                {
+                    "PERPLEXITY_API_KEY": "pplx-test",
+                    "OPENROUTER_API_KEY": "or-test",
+                },
+            )
+
+        url, payload = post.call_args.args[:2]
+        headers = post.call_args.kwargs["headers"]
+        self.assertEqual(perplexity.PERPLEXITY_URL, url)
+        self.assertEqual("Bearer pplx-test", headers["Authorization"])
+        self.assertEqual("sonar-pro", payload["model"])
+        self.assertEqual(
+            "05/01/2026",
+            payload["web_search_options"]["search_after_date_filter"],
+        )
+        self.assertEqual(
+            "06/01/2026",
+            payload["web_search_options"]["search_before_date_filter"],
+        )
+        self.assertEqual("perplexity", artifact["provider"])
+        self.assertEqual("sonar-pro", artifact["model"])
+        self.assertEqual("Example A", items[1]["title"])
+        self.assertEqual("Direct source snippet", items[1]["snippet"])
+
+    def test_direct_model_config_selects_supported_sonar_model(self):
+        response = {
+            "choices": [{"message": {"content": "Reasoned synthesis"}}],
+            "citations": [],
+            "search_results": [],
+        }
+        with patch("lib.perplexity.http.post", return_value=response) as post:
+            _, artifact = perplexity.search(
+                "test topic",
+                ("2026-05-01", "2026-06-01"),
+                {
+                    "PERPLEXITY_API_KEY": "pplx-test",
+                    "LAST30DAYS_PERPLEXITY_MODEL": "sonar-reasoning-pro",
+                    "LAST30DAYS_PERPLEXITY_REASONING_EFFORT": "high",
+                },
+            )
+
+        payload = post.call_args.args[1]
+        self.assertEqual("sonar-reasoning-pro", payload["model"])
+        self.assertEqual("high", payload["reasoning_effort"])
+        self.assertEqual("sonar-reasoning-pro", artifact["model"])
+
+    def test_openrouter_fallback_uses_openrouter_models_and_annotations(self):
+        response = {
+            "choices": [
+                {
+                    "message": {
+                        "content": "OpenRouter synthesis",
+                        "annotations": [
+                            {
+                                "url_citation": {
+                                    "url": "https://example.com/b",
+                                    "title": "Example B",
+                                }
+                            }
+                        ],
+                    }
+                }
+            ],
+        }
+        with patch("lib.perplexity.http.post", return_value=response) as post:
+            items, artifact = perplexity.search(
+                "test topic",
+                ("2026-05-01", "2026-06-01"),
+                {"OPENROUTER_API_KEY": "or-test"},
+                deep=True,
+            )
+
+        url, payload = post.call_args.args[:2]
+        headers = post.call_args.kwargs["headers"]
+        self.assertEqual(perplexity.OPENROUTER_URL, url)
+        self.assertEqual("Bearer or-test", headers["Authorization"])
+        self.assertEqual("perplexity/sonar-deep-research", payload["model"])
+        self.assertEqual(120, post.call_args.kwargs["timeout"])
+        self.assertEqual("openrouter", artifact["provider"])
+        self.assertEqual("perplexity/sonar-deep-research", artifact["model"])
+        self.assertEqual("Example B", items[1]["title"])
+
+    def test_search_api_mode_returns_ranked_rows_with_filters(self):
+        response = {
+            "id": "search-1",
+            "server_time": "2026-06-01T00:00:00Z",
+            "results": [
+                {
+                    "title": "Ranked result",
+                    "url": "https://example.com/ranked",
+                    "snippet": "Search API snippet",
+                    "date": "2026-05-15",
+                    "last_updated": "2026-05-20",
+                }
+            ],
+        }
+        config = {
+            "PERPLEXITY_API_KEY": "pplx-test",
+            "LAST30DAYS_PERPLEXITY_MODE": "search",
+            "LAST30DAYS_PERPLEXITY_MAX_RESULTS": "3",
+            "LAST30DAYS_PERPLEXITY_SEARCH_CONTEXT_SIZE": "low",
+            "LAST30DAYS_PERPLEXITY_COUNTRY": "us",
+            "LAST30DAYS_PERPLEXITY_DOMAIN_FILTER": "example.com,example.org",
+            "LAST30DAYS_PERPLEXITY_LANGUAGE_FILTER": "en",
+            "LAST30DAYS_PERPLEXITY_RECENCY_FILTER": "year",
+        }
+        with patch("lib.perplexity.http.post", return_value=response) as post:
+            items, artifact = perplexity.search(
+                "test topic",
+                ("2026-05-01", "2026-06-01"),
+                config,
+            )
+
+        url, payload = post.call_args.args[:2]
+        self.assertEqual(perplexity.PERPLEXITY_SEARCH_URL, url)
+        self.assertEqual("test topic", payload["query"])
+        self.assertEqual(3, payload["max_results"])
+        self.assertEqual("low", payload["search_context_size"])
+        self.assertEqual("US", payload["country"])
+        self.assertEqual(["example.com", "example.org"], payload["search_domain_filter"])
+        self.assertEqual("05/01/2026", payload["search_after_date_filter"])
+        self.assertEqual("06/01/2026", payload["search_before_date_filter"])
+        self.assertNotIn("search_recency_filter", payload)
+        self.assertEqual("search", artifact["mode"])
+        self.assertEqual("Ranked result", items[0]["title"])
+        self.assertEqual("2026-05-20", items[0]["metadata"]["last_updated"])
+
+    def test_search_api_keeps_recency_filter_when_no_exact_dates_are_available(self):
+        payload = perplexity._build_search_payload(
+            "test topic",
+            ("not-a-date", "also-not-a-date"),
+            {"LAST30DAYS_PERPLEXITY_RECENCY_FILTER": "week"},
+        )
+
+        self.assertEqual("week", payload["search_recency_filter"])
+        self.assertNotIn("search_after_date_filter", payload)
+        self.assertNotIn("search_before_date_filter", payload)
+
+    def test_both_mode_keeps_synthesis_and_dedupes_raw_rows(self):
+        search_response = {
+            "id": "search-1",
+            "results": [
+                {
+                    "title": "Duplicate ranked result",
+                    "url": "https://example.com/a",
+                    "snippet": "Raw row",
+                    "date": "2026-05-15",
+                },
+                {
+                    "title": "Unique ranked result",
+                    "url": "https://example.com/unique",
+                    "snippet": "Unique raw row",
+                    "date": "2026-05-16",
+                },
+            ],
+        }
+        sonar_response = {
+            "choices": [{"message": {"content": "Sonar synthesis"}}],
+            "citations": ["https://example.com/a"],
+            "search_results": [
+                {
+                    "title": "Citation result",
+                    "url": "https://example.com/a",
+                    "snippet": "Citation row",
+                    "date": "2026-05-15",
+                }
+            ],
+        }
+        with patch(
+            "lib.perplexity.http.post",
+            side_effect=[search_response, sonar_response],
+        ) as post:
+            items, artifact = perplexity.search(
+                "test topic",
+                ("2026-05-01", "2026-06-01"),
+                {
+                    "PERPLEXITY_API_KEY": "pplx-test",
+                    "LAST30DAYS_PERPLEXITY_MODE": "both",
+                },
+            )
+
+        self.assertEqual(perplexity.PERPLEXITY_SEARCH_URL, post.call_args_list[0].args[0])
+        self.assertEqual(perplexity.PERPLEXITY_URL, post.call_args_list[1].args[0])
+        self.assertEqual("both", artifact["mode"])
+        self.assertEqual(3, artifact["itemCount"])
+        self.assertEqual("perplexity.ai", items[0]["source_domain"])
+        urls = [item["url"] for item in items if item["url"]]
+        self.assertEqual(["https://example.com/a", "https://example.com/unique"], urls)
+
+    def test_both_mode_keeps_search_rows_when_sonar_leg_fails(self):
+        search_response = {
+            "id": "search-1",
+            "results": [
+                {
+                    "title": "Raw result",
+                    "url": "https://example.com/raw",
+                    "snippet": "Raw row",
+                    "date": "2026-05-15",
+                },
+            ],
+        }
+        with patch(
+            "lib.perplexity.http.post",
+            side_effect=[
+                search_response,
+                perplexity.http.HTTPError("HTTP 500: Server Error", status_code=500),
+            ],
+        ):
+            items, artifact = perplexity.search(
+                "test topic",
+                ("2026-05-01", "2026-06-01"),
+                {
+                    "PERPLEXITY_API_KEY": "pplx-test",
+                    "LAST30DAYS_PERPLEXITY_MODE": "both",
+                },
+            )
+
+        self.assertEqual("Raw result", items[0]["title"])
+        self.assertEqual(1, artifact["itemCount"])
+        self.assertEqual("HTTPError", artifact["sonar"]["error"])
+        self.assertEqual(500, artifact["sonar"]["statusCode"])
+
+    def test_direct_deep_research_uses_async_api_and_wall_timeout_config(self):
+        create_response = {"id": "async-1", "status": "CREATED", "created_at": 123}
+        complete_response = {
+            "id": "async-1",
+            "status": "COMPLETED",
+            "created_at": 123,
+            "started_at": 124,
+            "completed_at": 130,
+            "response": {
+                "choices": [{"message": {"content": "Deep synthesis"}}],
+                "citations": ["https://example.com/deep"],
+                "search_results": [
+                    {
+                        "title": "Deep citation",
+                        "url": "https://example.com/deep",
+                        "snippet": "Deep snippet",
+                    }
+                ],
+                "usage": {
+                    "total_tokens": 123,
+                    "cost": {"total_cost": 0.12},
+                },
+            },
+        }
+        with patch("lib.perplexity.http.post", return_value=create_response) as post, \
+             patch("lib.perplexity.http.get", return_value=complete_response) as get:
+            items, artifact = perplexity.search(
+                "test topic",
+                ("2026-05-01", "2026-06-01"),
+                {
+                    "PERPLEXITY_API_KEY": "pplx-test",
+                    "LAST30DAYS_PERPLEXITY_DEEP_TIMEOUT_SECONDS": "300",
+                },
+                deep=True,
+            )
+
+        self.assertEqual(perplexity.PERPLEXITY_ASYNC_URL, post.call_args.args[0])
+        self.assertEqual(
+            perplexity.PERPLEXITY_ASYNC_URL,
+            perplexity._provider({"PERPLEXITY_API_KEY": "pplx-test"}, deep=True)[2],
+        )
+        create_payload = post.call_args.args[1]
+        self.assertEqual("sonar-deep-research", create_payload["request"]["model"])
+        self.assertTrue(create_payload["idempotency_key"].startswith("last30days:"))
+        self.assertEqual(f"{perplexity.PERPLEXITY_ASYNC_URL}/async-1", get.call_args.args[0])
+        self.assertEqual("async-sonar", artifact["endpoint"])
+        self.assertEqual(True, artifact["async"])
+        self.assertEqual(300, artifact["asyncTimeoutSeconds"])
+        self.assertEqual(create_payload["idempotency_key"], artifact["asyncIdempotencyKey"])
+        self.assertEqual(1, artifact["asyncPollCount"])
+        self.assertEqual("COMPLETED_REMOTE", artifact["asyncLocalStatus"])
+        self.assertEqual(123, artifact["asyncCreatedAt"])
+        self.assertEqual(124, artifact["asyncStartedAt"])
+        self.assertEqual(130, artifact["asyncCompletedAt"])
+        self.assertEqual(123, items[0]["metadata"]["usage"]["total_tokens"])
+
+    def test_direct_deep_research_timeout_returns_empty_result(self):
+        with patch("lib.perplexity.http.post", return_value={"id": "async-1", "status": "CREATED", "created_at": 123}), \
+             patch("lib.perplexity.http.get", return_value={
+                 "id": "async-1",
+                 "status": "IN_PROGRESS",
+                 "created_at": 123,
+                 "started_at": 124,
+             }), \
+             patch("lib.perplexity.time.monotonic", side_effect=[0, 0, 2, 2]), \
+             patch("lib.perplexity.time.sleep"):
+            items, artifact = perplexity.search(
+                "test topic",
+                ("2026-05-01", "2026-06-01"),
+                {
+                    "PERPLEXITY_API_KEY": "pplx-test",
+                    "LAST30DAYS_PERPLEXITY_DEEP_TIMEOUT_SECONDS": "1",
+                },
+                deep=True,
+            )
+
+        self.assertEqual([], items)
+        self.assertEqual("timeout", artifact["error"])
+        self.assertEqual("async-1", artifact["asyncRequestId"])
+        self.assertEqual("IN_PROGRESS", artifact["asyncStatus"])
+        self.assertEqual(1, artifact["asyncTimeoutSeconds"])
+        self.assertEqual(1, artifact["asyncPollCount"])
+        self.assertEqual("PENDING_REMOTE", artifact["asyncLocalStatus"])
+        self.assertEqual(123, artifact["asyncCreatedAt"])
+        self.assertEqual(124, artifact["asyncStartedAt"])
+
+    def test_direct_deep_research_failed_status_returns_failure_artifact(self):
+        with patch("lib.perplexity.http.post", return_value={"id": "async-1", "status": "CREATED"}), \
+             patch("lib.perplexity.http.get", return_value={
+                 "id": "async-1",
+                 "status": "FAILED",
+                 "failed_at": 130,
+                 "error_message": "provider failure",
+             }):
+            items, artifact = perplexity.search(
+                "test topic",
+                ("2026-05-01", "2026-06-01"),
+                {"PERPLEXITY_API_KEY": "pplx-test"},
+                deep=True,
+            )
+
+        self.assertEqual([], items)
+        self.assertEqual("failed", artifact["error"])
+        self.assertEqual("async-1", artifact["asyncRequestId"])
+        self.assertEqual("FAILED", artifact["asyncStatus"])
+        self.assertEqual("FAILED_REMOTE", artifact["asyncLocalStatus"])
+        self.assertEqual(130, artifact["asyncFailedAt"])
+        self.assertEqual("provider failure", artifact["asyncErrorMessage"])
+
+    def test_direct_deep_research_poll_error_preserves_async_id(self):
+        with patch("lib.perplexity.http.post", return_value={
+            "id": "async-1",
+            "status": "CREATED",
+            "created_at": 123,
+        }), \
+             patch("lib.perplexity.http.get", side_effect=perplexity.http.HTTPError(
+                 "HTTP 429: Too Many Requests",
+                 status_code=429,
+             )):
+            items, artifact = perplexity.search(
+                "test topic",
+                ("2026-05-01", "2026-06-01"),
+                {"PERPLEXITY_API_KEY": "pplx-test"},
+                deep=True,
+            )
+
+        self.assertEqual([], items)
+        self.assertEqual("poll_error", artifact["error"])
+        self.assertEqual("async-1", artifact["asyncRequestId"])
+        self.assertEqual("CREATED", artifact["asyncStatus"])
+        self.assertEqual("POLL_ERROR", artifact["asyncLocalStatus"])
+        self.assertEqual(1, artifact["asyncPollCount"])
+        self.assertEqual(429, artifact["asyncPollStatusCode"])
+
+    def test_direct_deep_research_malformed_completed_preserves_async_id(self):
+        with patch("lib.perplexity.http.post", return_value={
+            "id": "async-1",
+            "status": "CREATED",
+            "created_at": 123,
+        }), \
+             patch("lib.perplexity.http.get", return_value={
+                 "id": "async-1",
+                 "status": "COMPLETED",
+                 "created_at": 123,
+                 "completed_at": 130,
+                 "response": None,
+             }):
+            items, artifact = perplexity.search(
+                "test topic",
+                ("2026-05-01", "2026-06-01"),
+                {"PERPLEXITY_API_KEY": "pplx-test"},
+                deep=True,
+            )
+
+        self.assertEqual([], items)
+        self.assertEqual("failed", artifact["error"])
+        self.assertEqual("async-1", artifact["asyncRequestId"])
+        self.assertEqual("COMPLETED", artifact["asyncStatus"])
+        self.assertEqual("FAILED_REMOTE", artifact["asyncLocalStatus"])
+        self.assertEqual(1, artifact["asyncPollCount"])
+        self.assertEqual(130, artifact["asyncCompletedAt"])
+        self.assertEqual(
+            "Async Deep Research completed without response",
+            artifact["asyncErrorMessage"],
+        )
+
+    def test_direct_deep_research_empty_choices_preserves_async_id(self):
+        with patch("lib.perplexity.http.post", return_value={
+            "id": "async-1",
+            "status": "CREATED",
+            "created_at": 123,
+        }) as post, \
+             patch("lib.perplexity.http.get", return_value={
+                 "id": "async-1",
+                 "status": "COMPLETED",
+                 "created_at": 123,
+                 "completed_at": 130,
+                 "response": {
+                     "choices": [],
+                     "usage": {"total_tokens": 321},
+                 },
+             }):
+            items, artifact = perplexity.search(
+                "test topic",
+                ("2026-05-01", "2026-06-01"),
+                {"PERPLEXITY_API_KEY": "pplx-test"},
+                deep=True,
+            )
+
+        self.assertEqual([], items)
+        self.assertEqual("empty_choices", artifact["error"])
+        self.assertEqual("async-1", artifact["asyncRequestId"])
+        self.assertEqual("COMPLETED", artifact["asyncStatus"])
+        self.assertEqual("COMPLETED_REMOTE", artifact["asyncLocalStatus"])
+        self.assertEqual(1, artifact["asyncPollCount"])
+        self.assertEqual(130, artifact["asyncCompletedAt"])
+        self.assertEqual(
+            post.call_args.args[1]["idempotency_key"],
+            artifact["asyncIdempotencyKey"],
+        )
+        self.assertEqual(321, artifact["usage"]["total_tokens"])
+        self.assertEqual(
+            "Async Deep Research completed without choices",
+            artifact["asyncErrorMessage"],
+        )
+
+    def test_direct_deep_research_empty_synthesis_preserves_async_id(self):
+        with patch("lib.perplexity.http.post", return_value={
+            "id": "async-1",
+            "status": "CREATED",
+            "created_at": 123,
+        }) as post, \
+             patch("lib.perplexity.http.get", return_value={
+                 "id": "async-1",
+                 "status": "COMPLETED",
+                 "created_at": 123,
+                 "completed_at": 130,
+                 "response": {
+                     "choices": [{"message": {"content": ""}}],
+                 },
+             }):
+            items, artifact = perplexity.search(
+                "test topic",
+                ("2026-05-01", "2026-06-01"),
+                {"PERPLEXITY_API_KEY": "pplx-test"},
+                deep=True,
+            )
+
+        self.assertEqual([], items)
+        self.assertEqual("empty_synthesis", artifact["error"])
+        self.assertEqual("async-1", artifact["asyncRequestId"])
+        self.assertEqual("COMPLETED", artifact["asyncStatus"])
+        self.assertEqual("COMPLETED_REMOTE", artifact["asyncLocalStatus"])
+        self.assertEqual(1, artifact["asyncPollCount"])
+        self.assertEqual(130, artifact["asyncCompletedAt"])
+        self.assertEqual(
+            post.call_args.args[1]["idempotency_key"],
+            artifact["asyncIdempotencyKey"],
+        )
+        self.assertEqual(
+            "Async Deep Research completed with empty synthesis",
+            artifact["asyncErrorMessage"],
+        )
+
+    def test_direct_deep_research_malformed_choice_preserves_async_id(self):
+        with patch("lib.perplexity.http.post", return_value={
+            "id": "async-1",
+            "status": "CREATED",
+            "created_at": 123,
+        }) as post, \
+             patch("lib.perplexity.http.get", return_value={
+                 "id": "async-1",
+                 "status": "COMPLETED",
+                 "created_at": 123,
+                 "completed_at": 130,
+                 "response": {
+                     "choices": [None],
+                 },
+             }):
+            items, artifact = perplexity.search(
+                "test topic",
+                ("2026-05-01", "2026-06-01"),
+                {"PERPLEXITY_API_KEY": "pplx-test"},
+                deep=True,
+            )
+
+        self.assertEqual([], items)
+        self.assertEqual("empty_synthesis", artifact["error"])
+        self.assertEqual("async-1", artifact["asyncRequestId"])
+        self.assertEqual("COMPLETED", artifact["asyncStatus"])
+        self.assertEqual("COMPLETED_REMOTE", artifact["asyncLocalStatus"])
+        self.assertEqual(
+            post.call_args.args[1]["idempotency_key"],
+            artifact["asyncIdempotencyKey"],
+        )
+
+    def test_missing_keys_skip_without_http(self):
+        with patch("lib.perplexity.http.post") as post:
+            items, artifact = perplexity.search(
+                "test topic",
+                ("2026-05-01", "2026-06-01"),
+                {},
+            )
+
+        post.assert_not_called()
+        self.assertEqual([], items)
+        self.assertEqual({}, artifact)
@@ -198,7 +198,6 @@ class TestSourceFetchCap(unittest.TestCase):

    def test_cap_logic_limits_source_submissions(self):
        """Verify the cap logic skips submissions beyond the limit."""
-        cap = pipeline.MAX_SOURCE_FETCHES.get("x", float("inf"))
        subquery_sources = [
            ["x", "reddit", "youtube"],
            ["x", "reddit", "youtube"],
@@ -228,7 +227,7 @@ class TestSourceFetchCap(unittest.TestCase):
        mock_retrieve.side_effect = lambda **kwargs: pipeline._mock_stream_results(
            kwargs["source"], kwargs["subquery"]
        )
-        report = pipeline.run(
+        pipeline.run(
            topic="compare iPhone vs Android vs Pixel vs Samsung",
            config={"LAST30DAYS_REASONING_PROVIDER": "gemini"},
            depth="quick",
@@ -295,7 +294,7 @@ class TestRateLimitSharing(unittest.TestCase):
        self.assertEqual(artifact, {})


-class TestThinSourceRetry(unittest.TestCase):
+class TestThinSourceRetryPlannedSource(unittest.TestCase):
    @patch("lib.pipeline._retrieve_stream")
    def test_retry_includes_planned_source_with_zero_initial_items(self, mock_retrieve):
        mock_retrieve.return_value = (
@@ -740,7 +739,6 @@ class TestThinSourceRetry(unittest.TestCase):
            )
            # x (non-errored, thin) should be retried; reddit (errored) should not
            if mock_retrieve.call_count > 0:
-                retried_sources = [call.kwargs.get("source") or call.args[2] for call in mock_retrieve.call_args_list if hasattr(call, 'kwargs')]
                self.assertNotIn("reddit", [c.kwargs.get("source") for c in mock_retrieve.call_args_list])

    def test_retry_skipped_in_quick_mode(self):
@@ -1120,6 +1118,23 @@ class TestExcludeSources(unittest.TestCase):
        self.assertIn("reddit", sources)


+class TestPerplexityAvailability(unittest.TestCase):
+    def test_perplexity_source_not_available_with_direct_key_without_opt_in(self):
+        sources = pipeline.available_sources({"PERPLEXITY_API_KEY": "test-key"})
+        self.assertNotIn("perplexity", sources)
+
+    def test_perplexity_source_available_with_direct_key(self):
+        sources = pipeline.available_sources(
+            {"PERPLEXITY_API_KEY": "test-key", "INCLUDE_SOURCES": "perplexity"}
+        )
+        self.assertIn("perplexity", sources)
+
+    def test_perplexity_diagnose_reports_direct_provider(self):
+        diag = pipeline.diagnose({"PERPLEXITY_API_KEY": "test-key"})
+        self.assertTrue(diag["providers"]["perplexity"])
+        self.assertTrue(diag["local_mode"])
+
+
 class TestKeylessGroundingAvailability(unittest.TestCase):
    """Grounding (general web) availability is host-aware.