Last30days

interface:
  display_name: "Last 30 Days"
  short_description: "Research any topic across Reddit, X, YouTube, and the web from the last 30 days. Returns synthesized expert answers and copy-paste prompts."
  default_prompt: "Research this topic from the last 30 days across Reddit, X, YouTube, and web. Synthesize what people are actually saying, upvoting, and sharing right now."
  brand_color: "#FF6B35"

policy:
  allow_implicit_invocation: true

#!/usr/bin/env python3
"""Morning briefing generator for last30days.

Synthesizes accumulated findings into formatted briefings.
The Python script collects the data; the agent (via SKILL.md) does the
beautiful synthesis. This script provides the structured data.

Usage:
    python3 briefing.py generate              # Daily briefing data
    python3 briefing.py generate --weekly     # Weekly digest data
    python3 briefing.py show [--date DATE]    # Show saved briefing
"""

import argparse
import json
import sys
from datetime import datetime, timedelta, timezone
from pathlib import Path

SCRIPT_DIR = Path(__file__).parent.resolve()
sys.path.insert(0, str(SCRIPT_DIR))

import store

BRIEFS_DIR = Path.home() / ".local" / "share" / "last30days" / "briefs"


def _parse_sqlite_utc_timestamp(value: str) -> datetime:
    return datetime.strptime(value, "%Y-%m-%d %H:%M:%S").replace(tzinfo=timezone.utc)


def generate_daily(since: str = None) -> dict:
    """Generate daily briefing data.

    Returns structured data for the agent to synthesize into a beautiful briefing.
    """
    store.init_db()
    topics = store.list_topics()

    if not topics:
        return {
            "status": "no_topics",
            "message": "No watchlist topics yet. Add one with: last30days watch add \"your topic\"",
        }

    enabled = [t for t in topics if t["enabled"]]
    if not enabled:
        return {
            "status": "no_enabled",
            "message": "All topics are paused. Enable a topic to generate briefings.",
        }

    # Default: findings since yesterday
    if not since:
        since = (datetime.now() - timedelta(days=1)).strftime("%Y-%m-%d")

    briefing_topics = []
    total_new = 0

    for topic in enabled:
        findings = store.get_new_findings(topic["id"], since)
        last_run = topic.get("last_run")
        last_status = topic.get("last_status", "unknown")

        # Calculate staleness
        stale = False
        hours_ago = None
        if last_run:
            try:
                run_dt = _parse_sqlite_utc_timestamp(last_run)
                hours_ago = (datetime.now(timezone.utc) - run_dt).total_seconds() / 3600
                stale = hours_ago > 36  # Stale if > 36 hours
            except (ValueError, TypeError):
                stale = True

        topic_data = {
            "name": topic["name"],
            "findings": findings,
            "new_count": len(findings),
            "last_run": last_run,
            "last_status": last_status,
            "stale": stale,
            "hours_ago": round(hours_ago, 1) if hours_ago else None,
        }

        # Extract top finding by engagement
        if findings:
            top = max(findings, key=lambda f: f.get("engagement_score", 0))
            topic_data["top_finding"] = {
                "title": top.get("source_title", ""),
                "source": top.get("source", ""),
                "author": top.get("author", ""),
                "engagement": top.get("engagement_score", 0),
                "content": top.get("content", "")[:300],
            }

        briefing_topics.append(topic_data)
        total_new += len(findings)

    # Cost info
    daily_cost = store.get_daily_cost()
    budget = float(store.get_setting("daily_budget", "5.00"))

    # Find the single top finding across all topics (for TL;DR)
    all_findings = []
    for t in briefing_topics:
        for f in t["findings"]:
            f["_topic"] = t["name"]
            all_findings.append(f)

    top_overall = None
    if all_findings:
        top_overall = max(all_findings, key=lambda f: f.get("engagement_score", 0))

    result = {
        "status": "ok",
        "date": datetime.now().strftime("%Y-%m-%d"),
        "since": since,
        "topics": briefing_topics,
        "total_new": total_new,
        "total_topics": len(briefing_topics),
        "top_finding": {
            "title": top_overall.get("source_title", ""),
            "topic": top_overall.get("_topic", ""),
            "engagement": top_overall.get("engagement_score", 0),
        } if top_overall else None,
        "cost": {
            "daily": daily_cost,
            "budget": budget,
        },
        "failed_topics": [
            t["name"] for t in briefing_topics if t["last_status"] == "failed"
        ],
    }

    # Save briefing data
    _save_briefing(result)

    return result


def generate_weekly() -> dict:
    """Generate weekly digest data with trend analysis."""
    store.init_db()

    week_ago = (datetime.now() - timedelta(days=7)).strftime("%Y-%m-%d")
    two_weeks_ago = (datetime.now() - timedelta(days=14)).strftime("%Y-%m-%d")

    topics = store.list_topics()
    if not topics:
        return {"status": "no_topics", "message": "No watchlist topics."}

    weekly_topics = []

    for topic in topics:
        if not topic["enabled"]:
            continue

        # This week's findings
        this_week = store.get_new_findings(topic["id"], week_ago)

        # Last week's findings (for comparison)
        conn = store._connect()
        try:
            last_week_rows = conn.execute(
                """SELECT * FROM findings
                   WHERE topic_id = ? AND first_seen >= ? AND first_seen < ? AND dismissed = 0
                   ORDER BY engagement_score DESC""",
                (topic["id"], two_weeks_ago, week_ago),
            ).fetchall()
            last_week = [dict(r) for r in last_week_rows]
        finally:
            conn.close()

        this_engagement = sum(f.get("engagement_score", 0) for f in this_week)
        last_engagement = sum(f.get("engagement_score", 0) for f in last_week)

        # Trend calculation
        if last_engagement > 0:
            engagement_change = ((this_engagement - last_engagement) / last_engagement) * 100
        else:
            engagement_change = 100 if this_engagement > 0 else 0

        weekly_topics.append({
            "name": topic["name"],
            "this_week_count": len(this_week),
            "last_week_count": len(last_week),
            "this_week_engagement": this_engagement,
            "last_week_engagement": last_engagement,
            "engagement_change_pct": round(engagement_change, 1),
            "top_findings": this_week[:5],  # Top 5 by engagement (already sorted)
        })

    result = {
        "status": "ok",
        "type": "weekly",
        "week_of": week_ago,
        "topics": weekly_topics,
    }

    _save_briefing(result, suffix="-weekly")

    return result


def show_briefing(date: str = None) -> dict:
    """Load a saved briefing by date."""
    if not date:
        date = datetime.now().strftime("%Y-%m-%d")

    path = BRIEFS_DIR / f"{date}.json"
    if not path.exists():
        # Try weekly
        path = BRIEFS_DIR / f"{date}-weekly.json"

    if not path.exists():
        return {"status": "not_found", "message": f"No briefing found for {date}."}

    with open(path, encoding="utf-8") as f:
        return json.load(f)


def _save_briefing(data: dict, suffix: str = ""):
    """Save briefing data to local archive."""
    BRIEFS_DIR.mkdir(parents=True, exist_ok=True)
    date = datetime.now().strftime("%Y-%m-%d")
    path = BRIEFS_DIR / f"{date}{suffix}.json"
    with open(path, "w", encoding="utf-8") as f:
        json.dump(data, f, indent=2, default=str)


def main():
    parser = argparse.ArgumentParser(description="Generate last30days briefings")
    sub = parser.add_subparsers(dest="command")

    # generate
    g = sub.add_parser("generate", help="Generate a briefing")
    g.add_argument("--weekly", action="store_true", help="Weekly digest")
    g.add_argument("--since", help="Findings since date (YYYY-MM-DD)")

    # show
    s = sub.add_parser("show", help="Show a saved briefing")
    s.add_argument("--date", help="Date (YYYY-MM-DD, default: today)")

    args = parser.parse_args()

    if args.command == "generate":
        if args.weekly:
            result = generate_weekly()
        else:
            result = generate_daily(since=args.since)
        print(json.dumps(result, indent=2, default=str))

    elif args.command == "show":
        result = show_briefing(date=args.date)
        print(json.dumps(result, indent=2, default=str))

    else:
        parser.print_help()
        sys.exit(1)


if __name__ == "__main__":
    main()

#!/usr/bin/env bash
# build-skill.sh - package this repo as a claude.ai-upload-ready .skill file
# Usage: bash skills/last30days/scripts/build-skill.sh  (run from repo root)
#
# Produces dist/last30days.skill, a zip with a single top-level `last30days/`
# directory containing SKILL.md and the scripts/ runtime from skills/last30days.
# See
# docs/plans/2026-04-14-001-fix-skill-upload-200-file-limit-plan.md.
set -euo pipefail

REPO_ROOT="$(cd "$(dirname "$0")/../../.." && pwd)"
cd "$REPO_ROOT"

if ! git diff --quiet || ! git diff --cached --quiet; then
  echo "error: working tree is dirty; commit or stash before building" >&2
  exit 1
fi

mkdir -p dist
OUT="dist/last30days.skill"
git archive --format=zip --prefix=last30days/ --output="$OUT" HEAD:skills/last30days

COUNT=$(unzip -l "$OUT" | tail -1 | awk '{print $2}')
SIZE=$(du -h "$OUT" | cut -f1)

if [ "$COUNT" -gt 200 ]; then
  echo "error: $COUNT files in zip, claude.ai's cap is 200" >&2
  echo "       check .gitattributes export-ignore entries and this script's zip -d excludes" >&2
  exit 1
fi

SKILL_MD_COUNT=$(unzip -l "$OUT" | grep -c "SKILL.md" || true)
if [ "$SKILL_MD_COUNT" -ne 1 ]; then
  echo "error: expected exactly one SKILL.md, found $SKILL_MD_COUNT" >&2
  exit 1
fi

echo "built $OUT ($COUNT files, $SIZE)"
echo "upload via the claude.ai skill UI"

#!/bin/bash
# A/B test runner: public release vs private beta
# Usage: bash skills/last30days/scripts/compare.sh "Kanye West"
#
# Runs /last30days (public release) and /last30days-beta (private beta)
# sequentially with a 30s gap, saves raw results with distinct suffixes,
# prints file paths for comparison.

set -e

if [ $# -eq 0 ]; then
  echo "Usage: bash skills/last30days/scripts/compare.sh <topic>"
  echo "  Example: bash skills/last30days/scripts/compare.sh Kevin Rose"
  exit 1
fi
TOPIC="$*"
SLUG=$(echo "$TOPIC" | tr '[:upper:]' '[:lower:]' | sed 's/[^a-z0-9]/-/g' | sed 's/--*/-/g' | sed 's/^-//' | sed 's/-$//')
LAST30DAYS_MEMORY_DIR="${LAST30DAYS_MEMORY_DIR:-$HOME/Documents/Last30Days}"
DIR="$LAST30DAYS_MEMORY_DIR"
DATE=$(date +%Y-%m-%d)

echo "=============================================="
echo " A/B Test: $TOPIC"
echo " Date: $DATE"
echo "=============================================="
echo ""

# Run 1: public release
echo "[1/2] Running /last30days (public release)..."
echo "  This takes 2-4 minutes..."
claude -p --dangerously-skip-permissions "/last30days $TOPIC" > /dev/null 2>&1 || true
RELEASE_FILE="$DIR/${SLUG}-raw.md"
[ -f "$RELEASE_FILE" ] && echo "  Done: $RELEASE_FILE" || echo "  FAILED: no output file"
echo ""

echo "  Waiting 30s for API rate limits..."
sleep 30

# Run 2: private beta
echo "[2/2] Running /last30days-beta (private beta)..."
echo "  This takes 2-4 minutes..."
claude -p --dangerously-skip-permissions "/last30days-beta $TOPIC" > /dev/null 2>&1 || true
BETA_FILE="$DIR/${SLUG}-raw-beta.md"
[ -f "$BETA_FILE" ] && echo "  Done: $BETA_FILE" || echo "  FAILED: no output file"
echo ""

echo "=============================================="
echo " Both complete. Raw files:"
echo "=============================================="
echo ""
ls -la "$DIR/${SLUG}-raw"*.md 2>/dev/null || echo "  (no files found - check if skills saved correctly)"
echo ""
echo "To compare, run in Claude Code:"
echo "  Read and compare these raw research files, produce a detailed report:"
echo "  $RELEASE_FILE"
echo "  $BETA_FILE"
echo ""
echo "Beta output should start with a line like:"
echo "  🧪 last30days-beta · branch <name> · synced $DATE"
echo "If that line is missing, the beta badge regressed. See docs/plans/2026-04-17-005-*-plan.md."
echo ""

#!/usr/bin/env python3
"""Compare two last30days revisions on the v3 ranked candidate output."""

from __future__ import annotations

import argparse
import json
import math
import os
import subprocess
import sys
import tempfile
from datetime import datetime
from pathlib import Path
from typing import Any
from urllib.error import HTTPError, URLError
from urllib.request import Request, urlopen

sys.path.insert(0, str(Path(__file__).parent))

from lib import env as envlib
from lib import schema
from lib.providers import GEMINI_FLASH_LITE


SKILL_ROOT = Path(__file__).resolve().parents[1]
REPO_ROOT = Path(__file__).resolve().parents[3]
EVAL_TOPICS_FILE = REPO_ROOT / "fixtures" / "eval_topics.json"


def _load_default_topics() -> list[tuple[str, str]]:
    if EVAL_TOPICS_FILE.exists():
        rows = json.loads(EVAL_TOPICS_FILE.read_text())
        return [(row["topic"], row["query_type"]) for row in rows]
    return [
        ("nano banana pro prompting", "product"),
        ("codex vs claude code", "comparison"),
        ("openclaw vs nanoclaw vs ironclaw", "comparison"),
        ("anthropic odds", "prediction"),
        ("kanye west", "breaking_news"),
        ("remotion animations for Claude Code", "how_to"),
    ]


DEFAULT_TOPICS = _load_default_topics()
DEFAULT_SEARCH = ""
DEFAULT_JUDGE_MODEL = GEMINI_FLASH_LITE
GEMINI_API_URL = "https://generativelanguage.googleapis.com/v1beta/models/{model}:generateContent?key={api_key}"


def stable_item_key(item: dict[str, Any]) -> str:
    return str(item.get("candidate_id") or item.get("url") or item.get("title") or "")


def row_sources(row: dict[str, Any]) -> list[str]:
    candidate = schema.candidate_from_dict(row)
    return schema.candidate_sources(candidate)


def row_best_date(row: dict[str, Any]) -> str | None:
    candidate = schema.candidate_from_dict(row)
    return schema.candidate_best_published_at(candidate)


V2_SOURCE_KEYS = [
    ("reddit", "title"),
    ("x", "text"),
    ("youtube", "title"),
    ("tiktok", "text"),
    ("instagram", "text"),
    ("hackernews", "title"),
    ("bluesky", "text"),
    ("truthsocial", "text"),
    ("polymarket", "question"),
    ("web", "title"),
]


def build_ranked_items(report: dict[str, Any], limit: int) -> list[dict[str, Any]]:
    # v3 format: ranked_candidates list
    if report.get("ranked_candidates"):
        ranked = []
        for row in report["ranked_candidates"][:limit]:
            candidate_sources = row_sources(row)
            ranked.append({
                "key": stable_item_key(row),
                "source": ", ".join(candidate_sources),
                "sources": candidate_sources,
                "url": str(row.get("url") or ""),
                "text": str(row.get("title") or ""),
                "date": row_best_date(row),
                "score": float(row.get("final_score") or 0.0),
            })
        return ranked

    # v2 format: per-source lists (reddit, x, youtube, etc.)
    all_items = []
    for source_key, text_field in V2_SOURCE_KEYS:
        for item in report.get(source_key) or []:
            if not isinstance(item, dict):
                continue
            all_items.append({
                "key": str(item.get("url") or item.get("id") or item.get(text_field) or ""),
                "source": source_key,
                "sources": [source_key],
                "url": str(item.get("url") or ""),
                "text": str(item.get(text_field) or item.get("title") or ""),
                "date": item.get("date"),
                "score": float(item.get("score") or 0.0),
            })
    all_items.sort(key=lambda x: x["score"], reverse=True)
    return all_items[:limit]


def source_sets(report: dict[str, Any], limit: int) -> dict[str, set[str]]:
    grouped: dict[str, set[str]] = {}
    for item in build_ranked_items(report, limit):
        for source in item["sources"]:
            grouped.setdefault(source, set()).add(item["key"])
    return grouped


def jaccard(left: set[str], right: set[str]) -> float:
    if not left and not right:
        return 1.0
    union = left | right
    if not union:
        return 1.0
    return len(left & right) / len(union)


def retention(left: set[str], right: set[str]) -> float:
    if not left:
        return 1.0
    return len(left & right) / len(left)


def precision_at_k(ranking: list[dict[str, Any]], judgments: dict[str, int], k: int) -> float:
    top = ranking[:k]
    if not top:
        return 0.0
    return sum(1 for item in top if judgments.get(item["key"], 0) >= 2) / len(top)


def ndcg_at_k(ranking: list[dict[str, Any]], judgments: dict[str, int], k: int, judged_pool: list[dict[str, Any]]) -> float:
    top = ranking[:k]
    if not top:
        return 0.0

    def dcg(grades: list[int]) -> float:
        total = 0.0
        for index, grade in enumerate(grades, start=1):
            total += (2**grade - 1) / math.log2(index + 1)
        return total

    actual = [judgments.get(item["key"], 0) for item in top]
    ideal = sorted((judgments.get(item["key"], 0) for item in judged_pool), reverse=True)[: len(top)]
    ideal_score = dcg(ideal)
    if ideal_score == 0:
        return 0.0
    return dcg(actual) / ideal_score


def source_coverage_recall(ranking: list[dict[str, Any]], judged_pool: list[dict[str, Any]], judgments: dict[str, int]) -> float:
    good_sources = {
        source
        for item in judged_pool
        if judgments.get(item["key"], 0) >= 2
        for source in item["sources"]
    }
    if not good_sources:
        return 1.0
    hit_sources = {
        source
        for item in ranking
        if judgments.get(item["key"], 0) >= 2
        for source in item["sources"]
    }
    return len(hit_sources & good_sources) / len(good_sources)


def resolve_google_judge_api_key(config: dict[str, Any]) -> str | None:
    return (
        os.environ.get("GOOGLE_API_KEY")
        or config.get("GOOGLE_API_KEY")
        or os.environ.get("GEMINI_API_KEY")
        or config.get("GEMINI_API_KEY")
        or os.environ.get("GOOGLE_GENAI_API_KEY")
        or config.get("GOOGLE_GENAI_API_KEY")
    )


def extract_gemini_text(payload: dict[str, Any]) -> str:
    for candidate in payload.get("candidates") or []:
        content = candidate.get("content") or {}
        for part in content.get("parts") or []:
            if part.get("text"):
                return part["text"]
    raise ValueError("Gemini response did not contain text.")


def call_gemini_judge(api_key: str, model: str, prompt: str) -> dict[str, Any]:
    body = {
        "contents": [{"parts": [{"text": prompt}]}],
        "generationConfig": {"temperature": 0, "responseMimeType": "application/json"},
    }
    request = Request(
        GEMINI_API_URL.format(model=model, api_key=api_key),
        data=json.dumps(body).encode("utf-8"),
        headers={"Content-Type": "application/json"},
        method="POST",
    )
    try:
        with urlopen(request, timeout=120) as response:
            payload = json.loads(response.read().decode("utf-8"))
    except HTTPError as exc:
        detail = exc.read().decode("utf-8", errors="replace")
        raise RuntimeError(f"Gemini HTTP {exc.code}: {detail}") from exc
    except URLError as exc:
        raise RuntimeError(f"Gemini request failed: {exc}") from exc
    return json.loads(extract_gemini_text(payload))


def build_judge_prompt(topic: str, query_type: str, items: list[dict[str, Any]]) -> str:
    item_lines = []
    for item in items:
        item_lines.append(
            "\n".join([
                f"- id: {item['key']}",
                f"  source: {item['source']}",
                f"  title: {item['text'][:220]}",
                f"  url: {item['url']}",
                f"  date: {item.get('date') or 'unknown'}",
            ])
        )
    return f"""
Judge search-result relevance for a last-30-days research tool.

Topic: {topic}
Query type: {query_type}

Score each item on this 0-3 scale:
- 0 = off-topic or clearly bad
- 1 = weak or tangential
- 2 = relevant and useful
- 3 = highly relevant, one of the best results

Return JSON only:
{{
  "judgments": [
    {{"id": "ITEM_ID", "grade": 0}}
  ]
}}

Items:
{chr(10).join(item_lines)}
""".strip()


def get_judgments(
    *,
    output_dir: Path,
    slug: str,
    topic: str,
    query_type: str,
    items: list[dict[str, Any]],
    judge_model: str,
    gemini_api_key: str | None,
) -> dict[str, int]:
    cache_file = output_dir / "judgments" / f"{slug}.json"
    cache_file.parent.mkdir(parents=True, exist_ok=True)
    if cache_file.exists():
        payload = json.loads(cache_file.read_text())
        return {row["id"]: int(row["grade"]) for row in payload.get("judgments") or []}
    if not gemini_api_key or not items:
        return {}
    payload = call_gemini_judge(gemini_api_key, judge_model, build_judge_prompt(topic, query_type, items))
    cache_file.write_text(json.dumps(payload, indent=2))
    return {row["id"]: int(row["grade"]) for row in payload.get("judgments") or []}


def create_eval_env() -> dict[str, str]:
    config = envlib.get_config()
    passthrough = {
        "PATH": os.environ.get("PATH", ""),
        "LANG": os.environ.get("LANG", "en_US.UTF-8"),
        "LC_ALL": os.environ.get("LC_ALL", ""),
        "TMPDIR": os.environ.get("TMPDIR", ""),
        "PYTHONUTF8": "1",
        "LAST30DAYS_CONFIG_DIR": "",
    }
    for key in (
        "GOOGLE_API_KEY",
        "GEMINI_API_KEY",
        "GOOGLE_GENAI_API_KEY",
        "OPENAI_API_KEY",
        "XAI_API_KEY",
        "SCRAPECREATORS_API_KEY",
        "BSKY_HANDLE",
        "BSKY_APP_PASSWORD",
        "TRUTHSOCIAL_TOKEN",
        "AUTH_TOKEN",
        "CT0",
    ):
        value = os.environ.get(key) or config.get(key)
        if value:
            passthrough[key] = value
    return passthrough


def run_last30days(repo_dir: Path, topic: str, *, search: str, timeout_seconds: int, quick: bool, mock: bool, env: dict[str, str]) -> dict[str, Any]:
    engine = repo_dir / "skills" / "last30days" / "scripts" / "last30days.py"
    if not engine.exists():
        engine = repo_dir / "scripts" / "last30days.py"
    cmd = [sys.executable, str(engine), topic, "--emit=json"]
    if search:
        cmd.extend(["--search", search])
    if quick:
        cmd.append("--quick")
    if mock:
        cmd.append("--mock")
    result = subprocess.run(
        cmd,
        cwd=repo_dir,
        env=env,
        capture_output=True,
        text=True,
        timeout=timeout_seconds,
        check=False,
    )
    if result.returncode != 0:
        raise RuntimeError(f"{repo_dir.name} failed for '{topic}' with exit {result.returncode}\n{result.stderr.strip()}")
    return json.loads(result.stdout)


def create_worktree(rev: str) -> Path:
    worktree_dir = Path(tempfile.mkdtemp(prefix="last30days-eval-"))
    subprocess.run(
        ["git", "worktree", "add", "--detach", str(worktree_dir), rev],
        cwd=REPO_ROOT,
        check=True,
        capture_output=True,
        text=True,
    )
    return worktree_dir


def resolve_repo_dir(label: str) -> tuple[Path, bool]:
    """Resolve a benchmark label into a repo directory and whether it is temporary."""
    if label == "WORKTREE":
        return REPO_ROOT, False
    return create_worktree(label), True


def remove_worktree(path: Path) -> None:
    subprocess.run(
        ["git", "worktree", "remove", "--force", str(path)],
        cwd=REPO_ROOT,
        check=False,
        capture_output=True,
        text=True,
    )
    try:
        os.rmdir(path)
    except OSError:
        pass


def summarize_topic(topic: str, query_type: str, baseline_report: dict[str, Any], candidate_report: dict[str, Any], judgments: dict[str, int], judged_pool: list[dict[str, Any]], limit: int) -> dict[str, Any]:
    baseline_ranked = build_ranked_items(baseline_report, limit)
    candidate_ranked = build_ranked_items(candidate_report, limit)
    baseline_sets = source_sets(baseline_report, limit)
    candidate_sets = source_sets(candidate_report, limit)
    overall_left = set().union(*baseline_sets.values()) if baseline_sets else set()
    overall_right = set().union(*candidate_sets.values()) if candidate_sets else set()
    sources = sorted(set(baseline_sets) | set(candidate_sets))
    return {
        "topic": topic,
        "query_type": query_type,
        "baseline": {
            "precision_at_5": precision_at_k(baseline_ranked, judgments, 5),
            "ndcg_at_5": ndcg_at_k(baseline_ranked, judgments, 5, judged_pool),
            "source_coverage_recall": source_coverage_recall(baseline_ranked, judged_pool, judgments),
        },
        "candidate": {
            "precision_at_5": precision_at_k(candidate_ranked, judgments, 5),
            "ndcg_at_5": ndcg_at_k(candidate_ranked, judgments, 5, judged_pool),
            "source_coverage_recall": source_coverage_recall(candidate_ranked, judged_pool, judgments),
        },
        "stability": {
            "overall_jaccard": jaccard(overall_left, overall_right),
            "overall_retention_vs_baseline": retention(overall_left, overall_right),
            "per_source": {
                source: {
                    "baseline_count": len(baseline_sets.get(source, set())),
                    "candidate_count": len(candidate_sets.get(source, set())),
                    "jaccard": jaccard(baseline_sets.get(source, set()), candidate_sets.get(source, set())),
                    "retention_vs_baseline": retention(baseline_sets.get(source, set()), candidate_sets.get(source, set())),
                }
                for source in sources
            },
        },
    }


def write_summary(output_dir: Path, baseline_label: str, candidate_label: str, summaries: list[dict[str, Any]]) -> None:
    output_dir.mkdir(parents=True, exist_ok=True)
    payload = {
        "generated_at": datetime.now().isoformat(timespec="seconds"),
        "baseline": baseline_label,
        "candidate": candidate_label,
        "topics": summaries,
    }
    (output_dir / "metrics.json").write_text(json.dumps(payload, indent=2))

    lines = [
        "# Search Quality Evaluation",
        "",
        f"- Baseline: `{baseline_label}`",
        f"- Candidate: `{candidate_label}`",
        f"- Generated: {payload['generated_at']}",
        "",
        "| Topic | Base P@5 | Cand P@5 | Base nDCG@5 | Cand nDCG@5 | Jaccard | Retention |",
        "|---|---:|---:|---:|---:|---:|---:|",
    ]
    for row in summaries:
        lines.append(
            "| {topic} | {bp:.2f} | {cp:.2f} | {bn:.2f} | {cn:.2f} | {jac:.2f} | {ret:.2f} |".format(
                topic=row["topic"],
                bp=row["baseline"]["precision_at_5"],
                cp=row["candidate"]["precision_at_5"],
                bn=row["baseline"]["ndcg_at_5"],
                cn=row["candidate"]["ndcg_at_5"],
                jac=row["stability"]["overall_jaccard"],
                ret=row["stability"]["overall_retention_vs_baseline"],
            )
        )
    (output_dir / "summary.md").write_text("\n".join(lines) + "\n")


def write_failure_summary(
    output_dir: Path,
    baseline_label: str,
    candidate_label: str,
    summaries: list[dict[str, Any]],
    failures: list[dict[str, Any]],
) -> None:
    write_summary(output_dir, baseline_label, candidate_label, summaries)
    metrics_path = output_dir / "metrics.json"
    payload = json.loads(metrics_path.read_text()) if metrics_path.exists() else {
        "generated_at": datetime.now().isoformat(timespec="seconds"),
        "baseline": baseline_label,
        "candidate": candidate_label,
        "topics": [],
    }
    payload["failures"] = failures
    metrics_path.write_text(json.dumps(payload, indent=2))

    summary_path = output_dir / "summary.md"
    lines = summary_path.read_text().splitlines() if summary_path.exists() else ["# Search Quality Evaluation", ""]
    if failures:
        lines.extend([
            "",
            "## Failures",
            "",
        ])
        for failure in failures:
            lines.append(f"- `{failure['topic']}`: {failure['error']}")
    summary_path.write_text("\n".join(lines).rstrip() + "\n")


def parse_topics_file(path: Path) -> list[tuple[str, str]]:
    rows = json.loads(path.read_text())
    return [(str(row["topic"]), str(row.get("query_type") or "general")) for row in rows]


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(description="Compare two last30days revisions on ranked candidate quality")
    parser.add_argument("--baseline", default="HEAD~1")
    parser.add_argument("--candidate", default="WORKTREE")
    parser.add_argument("--search", default=DEFAULT_SEARCH)
    parser.add_argument("--output-dir", default="tmp/search-quality")
    parser.add_argument("--judge-model", default=DEFAULT_JUDGE_MODEL)
    parser.add_argument("--timeout", type=int, default=240)
    parser.add_argument("--limit", type=int, default=20)
    parser.add_argument("--mock", action="store_true")
    parser.add_argument("--quick", action="store_true")
    parser.add_argument("--topics-file")
    return parser


def main() -> int:
    args = build_parser().parse_args()
    topics = parse_topics_file(Path(args.topics_file)) if args.topics_file else DEFAULT_TOPICS
    output_dir = Path(args.output_dir).resolve()
    config = envlib.get_config()
    gemini_api_key = resolve_google_judge_api_key(config)
    run_env = create_eval_env()

    baseline_dir, baseline_temp = resolve_repo_dir(args.baseline)
    candidate_dir, candidate_temp = resolve_repo_dir(args.candidate)
    try:
        summaries = []
        failures = []
        for topic, query_type in topics:
            try:
                baseline_report = run_last30days(
                    baseline_dir,
                    topic,
                    search=args.search,
                    timeout_seconds=args.timeout,
                    quick=args.quick,
                    mock=args.mock,
                    env=run_env,
                )
                candidate_report = run_last30days(
                    candidate_dir,
                    topic,
                    search=args.search,
                    timeout_seconds=args.timeout,
                    quick=args.quick,
                    mock=args.mock,
                    env=run_env,
                )
                judged_pool_map = {
                    item["key"]: item
                    for item in build_ranked_items(baseline_report, args.limit) + build_ranked_items(candidate_report, args.limit)
                }
                judged_pool = list(judged_pool_map.values())
                judgments = get_judgments(
                    output_dir=output_dir,
                    slug="".join(char.lower() if char.isalnum() else "-" for char in topic).strip("-"),
                    topic=topic,
                    query_type=query_type,
                    items=judged_pool,
                    judge_model=args.judge_model,
                    gemini_api_key=gemini_api_key,
                )
                summaries.append(summarize_topic(topic, query_type, baseline_report, candidate_report, judgments, judged_pool, args.limit))
            except Exception as exc:
                failures.append({"topic": topic, "query_type": query_type, "error": str(exc)})
        write_failure_summary(output_dir, args.baseline, args.candidate, summaries, failures)
    finally:
        if baseline_temp:
            remove_worktree(baseline_dir)
        if candidate_temp:
            remove_worktree(candidate_dir)
    result = {"output_dir": str(output_dir), "topics": len(topics), "failures": len(failures)}
    print(json.dumps(result, indent=2))
    return 1 if failures else 0


if __name__ == "__main__":
    raise SystemExit(main())

#!/usr/bin/env python3
# ruff: noqa: E402
"""last30days CLI."""

from __future__ import annotations

import argparse
import atexit
import datetime
import json
import os
import re
import signal
import sys
import threading
from pathlib import Path

MIN_PYTHON = (3, 12)


def ensure_supported_python(version_info: tuple[int, int, int] | object | None = None) -> None:
    if version_info is None:
        version_info = sys.version_info
    major, minor, micro = tuple(version_info[:3])
    if (major, minor) >= MIN_PYTHON:
        return
    sys.stderr.write(
        "last30days v3 requires Python 3.12+.\n"
        f"Detected Python {major}.{minor}.{micro}.\n"
        "Install and use python3.12 or python3.13, then rerun this command.\n"
    )
    raise SystemExit(1)


ensure_supported_python()

if os.name == "nt":
    for stream in (sys.stdout, sys.stderr):
        if hasattr(stream, "reconfigure"):
            stream.reconfigure(encoding="utf-8", errors="replace")

SCRIPT_DIR = Path(__file__).parent.resolve()
sys.path.insert(0, str(SCRIPT_DIR))

from lib import env, html_render, pipeline, render, schema, ui

_child_pids: set[int] = set()
_child_pids_lock = threading.Lock()


def register_child_pid(pid: int) -> None:
    with _child_pids_lock:
        _child_pids.add(pid)


def unregister_child_pid(pid: int) -> None:
    with _child_pids_lock:
        _child_pids.discard(pid)


def _cleanup_children() -> None:
    with _child_pids_lock:
        pids = list(_child_pids)
    for pid in pids:
        try:
            if hasattr(os, "killpg"):
                os.killpg(os.getpgid(pid), signal.SIGTERM)
            else:
                os.kill(pid, signal.SIGTERM)
        except (ProcessLookupError, PermissionError, OSError):
            continue


atexit.register(_cleanup_children)


def parse_search_flag(raw: str) -> list[str]:
    sources = []
    for source in raw.split(","):
        source = source.strip().lower()
        if not source:
            continue
        normalized = pipeline.SEARCH_ALIAS.get(source, source)
        if normalized not in pipeline.MOCK_AVAILABLE_SOURCES:
            raise SystemExit(f"Unknown search source: {source}")
        if normalized not in sources:
            sources.append(normalized)
    if not sources:
        raise SystemExit("--search requires at least one source.")
    return sources


def slugify(value: str) -> str:
    slug = re.sub(r"[^a-z0-9]+", "-", value.lower()).strip("-")
    return slug or "last30days"


def save_output(
    report: schema.Report,
    emit: str,
    save_dir: str,
    suffix: str = "",
    synthesis_md: str | None = None,
    topic_override: str | None = None,
    rendered_content: str | None = None,
) -> Path:
    from datetime import datetime
    path = Path(save_dir).expanduser().resolve()
    path.mkdir(parents=True, exist_ok=True)
    slug = slugify(topic_override or report.topic)
    extension = "json" if emit == "json" else "html" if emit == "html" else "md"
    raw_label = "raw-html" if emit == "html" else "raw"
    suffix_part = f"-{suffix}" if suffix else ""
    out_path = path / f"{slug}-{raw_label}{suffix_part}.{extension}"
    if out_path.exists():
        out_path = path / f"{slug}-{raw_label}{suffix_part}-{datetime.now().strftime('%Y-%m-%d')}.{extension}"
    # Markdown saves keep the complete debug artifact. JSON and HTML preserve
    # their requested wire format so file extensions match their content.
    if rendered_content is not None:
        content = rendered_content
    elif emit in {"json", "html"}:
        content = emit_output(report, emit, synthesis_md=synthesis_md)
    else:
        content = render.render_full(report)
    out_path.write_text(content, encoding="utf-8")
    return out_path


def emit_output(
    report: schema.Report,
    emit: str,
    fun_level: str = "medium",
    save_path: str | None = None,
    synthesis_md: str | None = None,
) -> str:
    if emit == "json":
        return json.dumps(schema.to_dict(report), indent=2, sort_keys=True)
    if emit == "html":
        return html_render.render_html(
            report, fun_level=fun_level, save_path=save_path, synthesis_md=synthesis_md,
        )
    if emit in {"compact", "md"}:
        return render.render_compact(report, fun_level=fun_level, save_path=save_path)
    if emit == "context":
        return render.render_context(report)
    raise SystemExit(f"Unsupported emit mode: {emit}")


def emit_comparison_output(
    entity_reports: list[tuple[str, schema.Report]],
    emit: str,
    fun_level: str = "medium",
    save_path: str | None = None,
    synthesis_md: str | None = None,
) -> str:
    if emit == "json":
        payload = {
            "comparison": True,
            "entities": [label for label, _ in entity_reports],
            "reports": [
                {"entity": label, "report": schema.to_dict(report)}
                for label, report in entity_reports
            ],
        }
        return json.dumps(payload, indent=2, sort_keys=True)
    if emit == "html":
        return html_render.render_html_comparison(
            entity_reports,
            fun_level=fun_level,
            save_path=save_path,
            synthesis_md=synthesis_md,
        )
    if emit in {"compact", "md"}:
        return render.render_comparison_multi(
            entity_reports, fun_level=fun_level, save_path=save_path,
        )
    if emit == "context":
        return render.render_comparison_multi_context(entity_reports)
    raise SystemExit(f"Unsupported emit mode: {emit}")


def comparison_topic(entity_reports: list[tuple[str, schema.Report]]) -> str:
    return " vs ".join(label for label, _ in entity_reports)


def compute_save_path_display(save_dir: str, topic: str, suffix: str, emit: str) -> str:
    """Compute the user-friendly save path string that will be shown in the footer.

    Uses ~ when the saved file is under the user's home directory; otherwise
    returns the absolute path.
    """
    from pathlib import Path as _Path
    path = _Path(save_dir).expanduser().resolve()
    slug = slugify(topic)
    extension = "json" if emit == "json" else "html" if emit == "html" else "md"
    raw_label = "raw-html" if emit == "html" else "raw"
    suffix_part = f"-{suffix}" if suffix else ""
    raw = path / f"{slug}-{raw_label}{suffix_part}.{extension}"
    try:
        home = _Path.home().resolve()
        relative = raw.relative_to(home)
        return f"~/{relative.as_posix()}"
    except ValueError:
        return raw.as_posix()


def read_synthesis_file(path: str) -> str:
    try:
        return Path(path).expanduser().read_text(encoding="utf-8")
    except OSError as exc:
        sys.stderr.write(f"[last30days] Cannot read --synthesis-file: {exc}\n")
        raise SystemExit(2)


def persist_report(report: schema.Report) -> dict[str, int]:
    import store

    store.init_db()
    topic_row = store.add_topic(report.topic)
    topic_id = topic_row["id"]
    source_mode = ",".join(sorted(report.items_by_source)) or "v3"
    run_id = store.record_run(topic_id, source_mode=source_mode, status="running")
    try:
        findings = store.findings_from_report(report)
        counts = store.store_findings(run_id, topic_id, findings)
        store.update_run(
            run_id,
            status="completed",
            findings_new=counts["new"],
            findings_updated=counts["updated"],
        )
        return counts
    except Exception as exc:
        store.update_run(run_id, status="failed", error_message=str(exc)[:500])
        raise


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(description="Research a topic across live social, market, and grounded web sources.")
    parser.add_argument("topic", nargs="*", help="Research topic")
    parser.add_argument("--emit", default="compact", choices=["compact", "json", "context", "md", "html"])
    parser.add_argument("--search", help="Comma-separated source list")
    parser.add_argument("--quick", action="store_true", help="Lower-latency retrieval profile")
    parser.add_argument("--deep", action="store_true", help="Higher-recall retrieval profile")
    parser.add_argument("--debug", action="store_true", help="Enable HTTP debug logging")
    parser.add_argument("--mock", action="store_true", help="Use mock retrieval fixtures")
    parser.add_argument("--diagnose", action="store_true", help="Print provider and source availability")
    parser.add_argument("--save-dir", help="Optional directory for saving the rendered output")
    parser.add_argument("--synthesis-file", help="Markdown synthesis to embed in --emit=html output")
    parser.add_argument("--store", action="store_true", help="Persist ranked findings to the SQLite research store")
    parser.add_argument("--x-handle", help="X handle for targeted supplemental search")
    parser.add_argument("--x-related", help="Comma-separated related X handles (searched with lower weight)")
    parser.add_argument("--web-backend", default="auto",
                        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.")
    parser.add_argument("--plan", help="JSON query plan (skips internal LLM planner). Can be a JSON string or a file path.")
    parser.add_argument("--save-suffix", help="Suffix for saved output filename (e.g., 'gemini' → kanye-west-raw-gemini.md)")
    parser.add_argument("--subreddits", help="Comma-separated subreddit names to search (e.g., SaaS,Entrepreneur)")
    parser.add_argument("--tiktok-hashtags", help="Comma-separated TikTok hashtags without # (e.g., tella,screenrecording)")
    parser.add_argument("--tiktok-creators", help="Comma-separated TikTok creator handles (e.g., TellaHQ,taborplace)")
    parser.add_argument("--ig-creators", help="Comma-separated Instagram creator handles (e.g., tella.tv,laborstories)")
    parser.add_argument(
        "--days",
        "--lookback-days",
        dest="lookback_days",
        type=int,
        default=30,
        help="Number of days to look back for research (default: 30, watchlist uses 90)",
    )
    parser.add_argument("--auto-resolve", action="store_true",
                        help="Use web search to discover subreddits/handles before planning (for platforms without WebSearch)")
    parser.add_argument("--github-user", help="GitHub username for person-mode search (e.g., steipete)")
    parser.add_argument("--github-repo", help="Comma-separated owner/repo for project-mode search (e.g., openclaw/openclaw,paperclipai/paperclip)")
    parser.add_argument(
        "--competitors",
        nargs="?",
        const=2,
        type=int,
        default=None,
        metavar="N",
        help="Auto-discover N competitor entities and fan out last30days across all of them as a comparison (default N=2 → 3-way: original + 2 peers; range 1..6). Use --competitors-list to override discovery.",
    )
    parser.add_argument(
        "--competitors-list",
        dest="competitors_list",
        help="Comma-separated competitor entities to skip discovery (e.g., 'Anthropic,xAI,Google Gemini'). Implies --competitors.",
    )
    parser.add_argument(
        "--polymarket-keywords",
        dest="polymarket_keywords",
        help=(
            "Comma-separated keywords that Polymarket market titles must match "
            "to be included. Use for ambiguous single-token topics like 'Warriors' "
            "(nba,gsw,golden-state) to filter out Glasgow Warriors rugby, Honor "
            "of Kings Rogue Warriors, etc. When omitted, Polymarket returns all "
            "matching markets — so expect cross-entity noise on generic topics."
        ),
    )
    parser.add_argument(
        "--competitors-plan",
        dest="competitors_plan",
        help=(
            "JSON mapping of per-entity Step 0.55 targeting for competitor / vs-mode "
            "sub-runs. Schema: {entity_name: {x_handle?, x_related?, subreddits?, "
            "github_user?, github_repos?, context?}}. Accepts inline JSON or a file "
            "path. Implies --competitors. Preferred over --competitors-list when the "
            "hosting model has already resolved per-entity handles and subs."
        ),
    )
    return parser


def parse_competitors_plan(raw: str | None) -> dict[str, dict]:
    """Parse a --competitors-plan argument into a {entity_name_lower: plan_entry} dict.

    Accepts inline JSON or a file path (matches --plan). Returns {} on None/empty.
    Validation: top-level must be a dict; each value must be a dict. Unknown fields
    in entry values log a warning but do not abort. Invalid JSON or non-dict shape
    raises SystemExit(2) with a clear stderr message.
    """
    if not raw:
        return {}
    plan_str = raw
    if os.path.isfile(plan_str):
        try:
            plan_str = open(plan_str).read()
        except OSError as exc:
            sys.stderr.write(f"[CompetitorsPlan] Cannot read plan file: {exc}\n")
            raise SystemExit(2)
    try:
        parsed = json.loads(plan_str)
    except json.JSONDecodeError as exc:
        sys.stderr.write(f"[CompetitorsPlan] Invalid JSON: {exc}\n")
        raise SystemExit(2)
    if not isinstance(parsed, dict):
        sys.stderr.write(
            f"[CompetitorsPlan] Top-level must be a dict of "
            f"{{entity: {{targeting}}}}, got {type(parsed).__name__}\n"
        )
        raise SystemExit(2)
    known_fields = {
        "x_handle", "x_related", "subreddits",
        "github_user", "github_repos", "context",
    }
    normalized: dict[str, dict] = {}
    for entity, entry in parsed.items():
        if not isinstance(entry, dict):
            sys.stderr.write(
                f"[CompetitorsPlan] Entry for {entity!r} must be a dict, "
                f"got {type(entry).__name__}; skipping.\n"
            )
            continue
        unknown = set(entry.keys()) - known_fields
        if unknown:
            sys.stderr.write(
                f"[CompetitorsPlan] Unknown fields in {entity!r}: "
                f"{sorted(unknown)}; ignoring.\n"
            )
        normalized[entity.strip().lower()] = {
            k: v for k, v in entry.items() if k in known_fields
        }
    return normalized


def subrun_kwargs_for(
    entity: str,
    plan_entry: dict,
    *,
    resolved: dict,
) -> dict:
    """Build an explicit per-entity kwargs dict for pipeline.run().

    Plan values win over auto_resolve values. Returns keys for all per-entity
    targeting flags so callers never fall through to closure defaults.

    This helper is the single source of truth for sub-run kwargs — main-topic
    flags can only leak if a caller bypasses it.
    """
    def _choose(plan_key: str, resolved_key: str | None = None):
        if plan_key in plan_entry and plan_entry[plan_key]:
            return plan_entry[plan_key]
        if resolved_key is not None and resolved.get(resolved_key):
            return resolved[resolved_key]
        return None

    x_handle = _choose("x_handle", "x_handle")
    if isinstance(x_handle, str):
        x_handle = x_handle.lstrip("@") or None

    subreddits = _choose("subreddits", "subreddits")
    if isinstance(subreddits, list):
        subreddits = [s.strip().removeprefix("r/") for s in subreddits if s.strip()] or None

    x_related = plan_entry.get("x_related")
    if isinstance(x_related, list):
        x_related = [h.strip().lstrip("@") for h in x_related if h.strip()] or None
    else:
        x_related = None

    github_user = _choose("github_user", "github_user")
    if isinstance(github_user, str):
        github_user = github_user.lstrip("@").lower() or None

    github_repos = _choose("github_repos", "github_repos")
    if isinstance(github_repos, list):
        github_repos = [r.strip() for r in github_repos if r.strip() and "/" in r.strip()] or None

    context = plan_entry.get("context") or resolved.get("context") or ""

    return {
        "x_handle": x_handle,
        "x_related": x_related,
        "subreddits": subreddits,
        "github_user": github_user,
        "github_repos": github_repos,
        "_context": context,
    }


COMPETITORS_MIN = 1
COMPETITORS_MAX = 6
COMPETITORS_DEFAULT = 2


def resolve_competitors_args(args: argparse.Namespace) -> tuple[bool, int, list[str]]:
    """Normalize --competitors / --competitors-list into (enabled, count, explicit_list).

    - (False, 0, []) when neither flag is set.
    - An explicit list always wins; count is derived from list length.
    - A numeric count outside [1, 6] is clamped with a stderr warning.
    - count <= 0 (explicit) raises SystemExit(2).
    """
    explicit_list: list[str] = []
    list_flag_provided = args.competitors_list is not None
    if list_flag_provided:
        explicit_list = [
            entity.strip()
            for entity in args.competitors_list.split(",")
            if entity.strip()
        ]
        if not explicit_list:
            sys.stderr.write("[Competitors] --competitors-list is empty.\n")
            raise SystemExit(2)

    competitors_flag = args.competitors
    list_present = bool(explicit_list)
    flag_present = competitors_flag is not None

    if not list_present and not flag_present:
        return False, 0, []

    if list_present:
        count = len(explicit_list)
        if flag_present and competitors_flag != count:
            sys.stderr.write(
                f"[Competitors] --competitors={competitors_flag} ignored; using "
                f"{count} entries from --competitors-list.\n"
            )
        if count > COMPETITORS_MAX:
            sys.stderr.write(
                f"[Competitors] --competitors-list has {count} entries, clamping to {COMPETITORS_MAX}.\n"
            )
            explicit_list = explicit_list[:COMPETITORS_MAX]
            count = COMPETITORS_MAX
        return True, count, explicit_list

    # flag_present, no explicit list
    count = competitors_flag
    if count < COMPETITORS_MIN:
        sys.stderr.write(
            f"[Competitors] --competitors must be >= {COMPETITORS_MIN} (got {count}).\n"
        )
        raise SystemExit(2)
    if count > COMPETITORS_MAX:
        sys.stderr.write(
            f"[Competitors] --competitors={count} exceeds max {COMPETITORS_MAX}; clamping.\n"
        )
        count = COMPETITORS_MAX
    return True, count, []


def _missing_sources_for_promo(diag: dict[str, object]) -> str | None:
    available = set(diag.get("available_sources") or [])
    missing = []
    if "reddit" not in available:
        missing.append("reddit")
    if "x" not in available:
        missing.append("x")
    if "grounding" not in available:
        missing.append("web")
    if not missing:
        return None
    if "reddit" in missing and "x" in missing:
        return "both"
    return missing[0]


def _show_runtime_ui(
    report: schema.Report,
    progress: ui.ProgressDisplay,
    diag: dict[str, object],
    suppress_web_promo: bool = False,
) -> None:
    counts = {source: len(items) for source, items in report.items_by_source.items()}
    display_sources = list(
        dict.fromkeys(
            [
                *report.query_plan.source_weights.keys(),
                *report.items_by_source.keys(),
                *report.errors_by_source.keys(),
            ]
        )
    )
    progress.end_processing()
    progress.show_complete(
        source_counts=counts,
        display_sources=display_sources,
    )
    promo = _missing_sources_for_promo(diag)
    # The `web` promo nudges users to set BRAVE_API_KEY / SERPER_API_KEY, which
    # is wrong advice when a hosting reasoning model (Claude Code, Codex,
    # Hermes, Gemini) is driving — those already have WebSearch and can
    # pre-resolve Step 0.55 themselves. Suppress the web promo when a hosting
    # model signal is present (--plan or --competitors-plan was passed).
    if promo:
        if suppress_web_promo and promo == "web":
            return
        if suppress_web_promo and promo == "both":
            # "both" means reddit + web both missing; still nudge reddit but
            # skip the web line. show_promo has a per-source variant.
            progress.show_promo("reddit", diag=diag)
            return
        progress.show_promo(promo, diag=diag)


def _write_last_run(topic: str, report: "schema.Report") -> None:
    try:
        if env.CONFIG_DIR is None:
            return
        target = env.CONFIG_DIR
        target.mkdir(parents=True, exist_ok=True)
        counts = {source: len(items) for source, items in report.items_by_source.items()}
        payload = {
            "topic": topic,
            "timestamp": datetime.datetime.now(datetime.timezone.utc).isoformat(),
            "sources": counts,
            "total": sum(counts.values()),
        }
        (target / "last-run.json").write_text(json.dumps(payload, indent=2))
    except Exception:
        pass


def main() -> int:
    parser = build_parser()
    # Use parse_known_args so setup sub-flags (--device-auth, --github,
    # --openclaw) pass through without argparse hard-exiting.
    args, extra_argv = parser.parse_known_args()
    if args.debug:
        os.environ["LAST30DAYS_DEBUG"] = "1"

    config = env.get_config()

    # Surface SSH-routing config as an env var so library modules (e.g.
    # youtube_yt) can read it without taking a config dependency. This
    # routes yt-dlp through `ssh <host>` to bypass YouTube's bot-wall on
    # datacenter IPs (see lib/youtube_yt.py for details).
    if config.get("LAST30DAYS_YOUTUBE_SSH_HOST") and "LAST30DAYS_YOUTUBE_SSH_HOST" not in os.environ:
        os.environ["LAST30DAYS_YOUTUBE_SSH_HOST"] = config["LAST30DAYS_YOUTUBE_SSH_HOST"]

    # Handle setup subcommand
    topic = " ".join(args.topic).strip()
    if topic.lower() == "setup":
        from lib import setup_wizard
        if "--openclaw" in extra_argv:
            results = setup_wizard.run_openclaw_setup(config)
            print(json.dumps(results))
            return 0
        if "--github" in extra_argv:
            results = setup_wizard.run_github_auth()
            print(json.dumps(results))
            return 0
        if "--device-auth" in extra_argv:
            results = setup_wizard.run_full_device_auth()
            print(json.dumps(results))
            return 0
        sys.stderr.write("Running auto-setup...\n")
        results = setup_wizard.run_auto_setup(config)
        from_browser = "auto"
        if results.get("cookies_found"):
            first_browser = next(iter(results["cookies_found"].values()))
            from_browser = first_browser
        setup_wizard.write_setup_config(env.CONFIG_FILE, from_browser=from_browser)
        results["env_written"] = True
        sys.stderr.write(setup_wizard.get_setup_status_text(results) + "\n")
        return 0

    requested_sources = parse_search_flag(args.search) if args.search else None
    diag = pipeline.diagnose(config, requested_sources)

    if args.diagnose:
        print(json.dumps(diag, indent=2, sort_keys=True))
        return 0

    if not topic:
        parser.print_usage(sys.stderr)
        return 2

    synthesis_md = None
    if args.synthesis_file:
        if args.emit == "html":
            synthesis_md = read_synthesis_file(args.synthesis_file)
        else:
            sys.stderr.write("[last30days] Warning: --synthesis-file is only used with --emit=html; ignoring.\n")

    if not os.environ.get("LAST30DAYS_SKIP_PREFLIGHT"):
        from lib import preflight
        refuse_msg = preflight.check_class_1_trap(topic)
        if refuse_msg:
            sys.stderr.write(refuse_msg)
            return 2

    progress = ui.ProgressDisplay(topic, show_banner=True)
    progress.start_processing()

    depth = "deep" if args.deep else "quick" if args.quick else "default"
    try:
        x_related = [h.strip() for h in args.x_related.split(",") if h.strip()] if args.x_related else None
        subreddits = [s.strip().removeprefix("r/") for s in args.subreddits.split(",") if s.strip()] if args.subreddits else None
        tiktok_hashtags = [h.strip().lstrip("#") for h in args.tiktok_hashtags.split(",") if h.strip()] if args.tiktok_hashtags else None
        tiktok_creators = [c.strip().lstrip("@") for c in args.tiktok_creators.split(",") if c.strip()] if args.tiktok_creators else None
        ig_creators = [c.strip().lstrip("@") for c in args.ig_creators.split(",") if c.strip()] if args.ig_creators else None
        # Parse external plan if provided via --plan flag
        external_plan = None
        if args.plan:
            import json as _json
            plan_str = args.plan
            if os.path.isfile(plan_str):
                plan_str = open(plan_str).read()
            try:
                external_plan = _json.loads(plan_str)
            except _json.JSONDecodeError as exc:
                sys.stderr.write(f"[Planner] Invalid --plan JSON: {exc}\n")

        # Auto-resolve: use web search to discover subreddits/handles before planning.
        # This is the engine-side equivalent of SKILL.md Steps 0.55/0.75 for platforms
        # without WebSearch (OpenClaw, Codex, raw CLI).
        repos_from_auto_resolve = False
        if args.auto_resolve and not external_plan:
            from lib import resolve
            resolution = resolve.auto_resolve(topic, config)
            if resolution.get("subreddits") and not subreddits:
                subreddits = resolution["subreddits"]
                sys.stderr.write(f"[AutoResolve] Subreddits: {', '.join(subreddits)}\n")
            if resolution.get("x_handle") and not args.x_handle:
                args.x_handle = resolution["x_handle"]
                sys.stderr.write(f"[AutoResolve] X handle: @{args.x_handle}\n")
            if resolution.get("github_user") and not args.github_user:
                args.github_user = resolution["github_user"]
                sys.stderr.write(f"[AutoResolve] GitHub user: @{args.github_user}\n")
            if resolution.get("github_repos") and not args.github_repo:
                args.github_repo = ",".join(resolution["github_repos"])
                # auto_resolve already canonicalized via canonicalize_github_repos(cap=5);
                # mark so we don't re-canonicalize below and clobber its relevance order.
                repos_from_auto_resolve = True
                sys.stderr.write(f"[AutoResolve] GitHub repos: {args.github_repo}\n")
            if resolution.get("context"):
                # Inject context into external_plan metadata for the planner to use
                if not external_plan:
                    external_plan = None  # planner will use its own, but with context
                # Store context for the planner prompt injection
                config["_auto_resolve_context"] = resolution["context"]
                sys.stderr.write(f"[AutoResolve] Context: {resolution['context'][:80]}...\n")

        github_user = args.github_user.lstrip("@").lower() if args.github_user else None
        github_repos = [r.strip() for r in args.github_repo.split(",") if r.strip() and "/" in r.strip()] if args.github_repo else None

        # Only canonicalize when repos came from a user-supplied --github-repo flag.
        # When repos_from_auto_resolve is True, auto_resolve already ran
        # canonicalize_github_repos(cap=5) and ranked by relevance; re-running here
        # with cap=None can re-sort by topic-slug match and lose that ordering.
        if github_repos and not repos_from_auto_resolve:
            from lib import resolve as resolve_lib
            original_github_repos = github_repos[:]
            github_repos = resolve_lib.canonicalize_github_repos(topic, github_repos, cap=None)
            if github_repos != original_github_repos:
                sys.stderr.write(
                    "[GitHub] Canonicalized repos: "
                    f"{','.join(original_github_repos)} -> {','.join(github_repos)}\n"
                )

        # --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)
                sys.exit(1)
            config["_deep_research"] = True
            # Auto-enable perplexity in INCLUDE_SOURCES
            include = config.get("INCLUDE_SOURCES") or ""
            if "perplexity" not in include.lower():
                config["INCLUDE_SOURCES"] = f"{include},perplexity" if include else "perplexity"

        comp_enabled, comp_count, comp_explicit = resolve_competitors_args(args)
        comp_plan = parse_competitors_plan(args.competitors_plan)

        # Polymarket disambiguation: if user passed --polymarket-keywords,
        # store on config so the polymarket adapter can filter matches.
        if args.polymarket_keywords:
            keywords = [
                k.strip().lower()
                for k in args.polymarket_keywords.split(",")
                if k.strip()
            ]
            if keywords:
                config["_polymarket_keywords"] = keywords

        # vs-mode: if the topic string contains " vs " / " versus " and the
        # planner can split it into >=2 entities, route through the same
        # N-pass fanout path as --competitors. The first entity becomes the
        # main topic; remaining entities become the competitor list. User's
        # outer --x-handle / --subreddits apply to the first entity unless
        # --competitors-plan covers it.
        from lib import planner as _planner
        vs_entities = _planner._comparison_entities(topic)
        if len(vs_entities) >= 2 and not comp_enabled:
            topic = vs_entities[0]
            comp_enabled = True
            comp_count = len(vs_entities) - 1
            comp_explicit = vs_entities[1:]
            sys.stderr.write(
                f"[Competitors] vs-mode: routing to N-pass fanout: "
                f"{' vs '.join(vs_entities)}\n"
            )

        def _main_runner() -> schema.Report:
            r = pipeline.run(
                topic=topic,
                config=config,
                depth=depth,
                requested_sources=requested_sources,
                mock=args.mock,
                x_handle=args.x_handle,
                x_related=x_related,
                web_backend=args.web_backend,
                external_plan=external_plan,
                subreddits=subreddits,
                tiktok_hashtags=tiktok_hashtags,
                tiktok_creators=tiktok_creators,
                ig_creators=ig_creators,
                lookback_days=args.lookback_days,
                github_user=github_user,
                github_repos=github_repos,
            )
            r.artifacts["resolved"] = {
                "entity": topic,
                "x_handle": (args.x_handle or "").lstrip("@"),
                "subreddits": list(subreddits or []),
                "github_user": (github_user or ""),
                "github_repos": list(github_repos or []),
                "context": config.get("_auto_resolve_context", "") or "",
            }
            return r

        if comp_enabled:
            from lib import competitors as competitors_mod
            from lib import fanout, resolve as resolve_mod

            if comp_explicit:
                discovered = comp_explicit
            else:
                if not resolve_mod._has_backend(config) and not args.mock:
                    sys.stderr.write(
                        "[Competitors] Cannot auto-discover peers without help.\n"
                        "\n"
                        "RECOMMENDED PATH (hosting reasoning models — Claude Code, Codex, "
                        "Hermes, Gemini, any agent with a WebSearch tool): YOU have "
                        "WebSearch. Use it to run full Step 0.55 per entity, then invoke "
                        "the engine with a vs-topic plus --competitors-plan:\n"
                        "  1. WebSearch for '{topic} competitors' or '{topic} alternatives'.\n"
                        "  2. For each peer, WebSearch for handles/subs/github (Step 0.55).\n"
                        "  3. Re-invoke: /last30days '{topic} vs {peer1} vs {peer2}' "
                        "--competitors-plan '{\"Peer1\":{\"x_handle\":\"h1\",\"subreddits\":"
                        "[\"s1\"],...},\"Peer2\":{...}}'.\n"
                        "See SKILL.md 'Competitor mode' for the full protocol.\n"
                        "\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"
                        "\n"
                        "MINIMUM ESCAPE HATCH: pass --competitors-list 'A,B,C' to skip "
                        "discovery. Without --competitors-plan, peer sub-runs fall back to "
                        "planner defaults and produce visibly thinner data than the main.\n"
                    )
                    return 2
                discovered = competitors_mod.discover_competitors(
                    topic, comp_count, config, lookback_days=args.lookback_days,
                )
                if not discovered:
                    sys.stderr.write(
                        f"[Competitors] No peers discovered for {topic!r}; aborting "
                        "comparison run. Pass --competitors-list to override.\n"
                    )
                    return 2

            sys.stderr.write(
                f"[Competitors] Comparing: {topic} vs " + " vs ".join(discovered) + "\n"
            )

            def _competitor_runner(entity: str) -> schema.Report:
                # Deep-copy config so per-entity auto_resolve context does not
                # leak across sub-runs. Each sub-run writes its own
                # `_auto_resolve_context` into its local config copy.
                entity_config = dict(config)
                plan_entry = comp_plan.get(entity.strip().lower(), {})
                resolved = {
                    "entity": entity,
                    "x_handle": "",
                    "subreddits": [],
                    "github_user": "",
                    "github_repos": [],
                    "context": "",
                }
                # Skip engine-internal auto_resolve when the hosting model
                # pre-resolved via --competitors-plan (saves a redundant
                # round-trip and makes per-entity Step 0.55 purely
                # hosting-model-driven).
                plan_covers_fully = bool(plan_entry.get("x_handle")) and bool(
                    plan_entry.get("subreddits")
                )
                if (
                    not args.mock
                    and not plan_covers_fully
                    and resolve_mod._has_backend(entity_config)
                ):
                    try:
                        r = resolve_mod.auto_resolve(entity, entity_config)
                    except Exception as exc:
                        sys.stderr.write(
                            f"[Competitors] auto_resolve failed for {entity!r}: "
                            f"{type(exc).__name__}: {exc}\n"
                        )
                        r = {}
                    resolved["x_handle"] = r.get("x_handle", "") or ""
                    resolved["subreddits"] = list(r.get("subreddits") or [])
                    resolved["github_user"] = r.get("github_user", "") or ""
                    resolved["github_repos"] = list(r.get("github_repos") or [])
                    resolved["context"] = r.get("context", "") or ""
                kwargs = subrun_kwargs_for(entity, plan_entry, resolved=resolved)
                # Record effective per-entity targeting for the Resolved block.
                resolved_effective = {
                    "entity": entity,
                    "x_handle": kwargs["x_handle"] or "",
                    "subreddits": kwargs["subreddits"] or [],
                    "github_user": kwargs["github_user"] or "",
                    "github_repos": kwargs["github_repos"] or [],
                    "context": kwargs["_context"],
                }
                if kwargs["_context"]:
                    entity_config["_auto_resolve_context"] = kwargs["_context"]
                sys.stderr.write(
                    f"[Competitors] {entity}: "
                    f"x=@{resolved_effective['x_handle'] or '-'} "
                    f"subs={len(resolved_effective['subreddits'])} "
                    f"gh={resolved_effective['github_user'] or '-'} "
                    f"({'plan' if plan_entry else 'auto'})\n"
                )
                report = pipeline.run(
                    topic=entity,
                    config=entity_config,
                    depth=depth,
                    requested_sources=requested_sources,
                    mock=args.mock,
                    x_handle=kwargs["x_handle"],
                    x_related=kwargs["x_related"],
                    subreddits=kwargs["subreddits"],
                    github_user=kwargs["github_user"],
                    github_repos=kwargs["github_repos"],
                    web_backend=args.web_backend,
                    lookback_days=args.lookback_days,
                    internal_subrun=True,
                )
                report.artifacts["resolved"] = resolved_effective
                return report

            entity_reports = fanout.run_competitor_fanout(
                main_topic=topic,
                main_runner=_main_runner,
                competitors=discovered,
                competitor_runner=_competitor_runner,
            )
            if len(entity_reports) < 2:
                progress.end_processing()
                sys.stderr.write(
                    f"[Competitors] Fewer than 2 sub-runs survived ({len(entity_reports)}); "
                    "cannot render a comparison. Re-run without --competitors or check the "
                    "warnings above.\n"
                )
                return 1
            report = entity_reports[0][1]
        else:
            entity_reports = None
            report = _main_runner()
    except Exception as exc:
        progress.end_processing()
        progress.show_error(str(exc))
        raise
    _show_runtime_ui(
        report, progress, diag,
        suppress_web_promo=bool(external_plan or comp_plan),
    )
    _write_last_run(topic, report)
    # LAST30DAYS_STORE env var = persistence default-on. Read both os.environ
    # (for shell-exported users) and config (for users who set it in
    # ~/.config/last30days/.env, which env.py loads but does not propagate
    # to os.environ). Mirrors the LAST30DAYS_DEBUG / LAST30DAYS_SKIP_PREFLIGHT
    # convention; env-var or config wins, with `--store` flag still working.
    _store_env = (
        os.environ.get("LAST30DAYS_STORE")
        or config.get("LAST30DAYS_STORE")
        or ""
    ).lower()
    if args.store or _store_env in ("1", "true", "yes"):
        counts = persist_report(report)
        sys.stderr.write(
            f"[last30days] Stored {counts['new']} new, {counts['updated']} updated findings\n"
        )
        sys.stderr.flush()

    # Show quality nudge if applicable
    try:
        from lib import quality_nudge
        # Populate transcript-fetch ratio so quality_nudge can detect the
        # degraded-YouTube failure mode (videos returned but transcripts
        # silently failed - typically a stale yt-dlp binary).
        youtube_items = report.items_by_source.get("youtube") or []
        instagram_items = report.items_by_source.get("instagram") or []
        research_results = {
            "youtube_videos_count": len(youtube_items),
            "youtube_transcripts_count": sum(
                1 for it in youtube_items
                if (it.metadata.get("transcript_highlights") or it.metadata.get("transcript_snippet"))
            ),
            "youtube_error": report.errors_by_source.get("youtube"),
            "x_error": report.errors_by_source.get("x"),
            # Captions-disabled videos can never produce a transcript regardless
            # of yt-dlp version; subtract them from the degraded-ratio
            # denominator so a single uploader-disabled video does not trip the
            # "stale yt-dlp" nudge.
            "youtube_captions_disabled_count": sum(
                1 for it in youtube_items if it.metadata.get("captions_disabled")
            ),
            # Track Instagram returned-zero-items so quality_nudge can detect
            # the silent-failure case (SC configured but the v2 reels endpoint
            # 500'd through both the original query and the hashtag retry).
            "instagram_items_count": len(instagram_items),
        }
        quality = quality_nudge.compute_quality_score(config, research_results)
        if quality.get("nudge_text"):
            sys.stderr.write(f"\n{quality['nudge_text']}\n")
            sys.stderr.flush()
    except Exception:
        pass

    fun_level = config.get("FUN_LEVEL", "medium").lower()
    # Comparison HTML is the one case where the saved file's title and content
    # have to be overridden away from the leading entity's report. Compute the
    # gate once so the footer-display and save-output paths can't disagree.
    is_comparison_html = bool(entity_reports) and args.emit == "html"
    footer_save_path = None
    if args.save_dir:
        save_topic_for_display = comparison_topic(entity_reports) if is_comparison_html else report.topic
        footer_save_path = compute_save_path_display(
            args.save_dir, save_topic_for_display, args.save_suffix or "", args.emit
        )

    # Signal to render_compact whether pre-research flags were supplied.
    # Used to emit a Pre-Research Status warning when the model skipped
    # Step 0.5 / 0.55 and invoked the engine bare on an eligible topic.
    pre_research_flags_present = bool(
        args.x_handle
        or args.github_user
        or args.subreddits
        or args.plan
        or args.auto_resolve
        or args.tiktok_creators
        or args.ig_creators
    )
    report.artifacts["pre_research_flags_present"] = pre_research_flags_present

    if entity_reports:
        rendered = emit_comparison_output(
            entity_reports,
            args.emit,
            fun_level=fun_level,
            save_path=footer_save_path,
            synthesis_md=synthesis_md,
        )
    else:
        rendered = emit_output(
            report,
            args.emit,
            fun_level=fun_level,
            save_path=footer_save_path,
            synthesis_md=synthesis_md,
        )
    if args.save_dir:
        # Save the main topic's raw file (single-entity or comparison main).
        save_path = save_output(
            report,
            args.emit,
            args.save_dir,
            suffix=args.save_suffix or "",
            synthesis_md=synthesis_md,
            topic_override=comparison_topic(entity_reports) if is_comparison_html else None,
            rendered_content=rendered if is_comparison_html else None,
        )
        sys.stderr.write(f"[last30days] Saved output to {save_path}\n")
        # Competitor / vs-mode: also save a per-entity raw file for each peer.
        # Matches historical vs-mode behavior (N passes → N save files).
        if entity_reports and len(entity_reports) > 1:
            for label, entity_report in entity_reports[1:]:
                peer_path = save_output(
                    entity_report, args.emit, args.save_dir,
                    suffix=args.save_suffix or "",
                    synthesis_md=synthesis_md,
                )
                sys.stderr.write(f"[last30days] Saved output to {peer_path}\n")
        sys.stderr.flush()
    print(rendered)
    return 0


if __name__ == "__main__":
    raise SystemExit(main())