fix(notion-search): correct --filter example syntax in docs

Final review caught that both --filter example locations used simplified
JSON ({"Status": "Done"}) that Notion's data_sources.query API rejects
with a 400. The script passes --filter verbatim, so users copy-pasting
the example would hit a confusing error.

Replace with Notion's actual filter shape:
  {"property": "Status", "status": {"equals": "Done"}}

Also added a compound (and/or) example in CLAUDE.md so users have a
reference for combining filters.

30/30 tests still pass.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

custom-skills/31-notion-organizer/code/CLAUDE.md

@@ -16,6 +16,46 @@ python scripts/schema_migrator.py --source [DB_ID] --target [DB_ID] --dry-run
 python scripts/async_organizer.py --database [DB_ID] --action cleanup
 ```
 
+## Semantic Search
+
+```bash
+# Default browse mode (terminal table)
+python scripts/notion_search.py "AI agents in 2026"
+
+# JSON output for piping
+python scripts/notion_search.py "AI agents" --json | jq '.[].id'
+
+# Constrain to specific databases
+python scripts/notion_search.py "MCP" --databases f8f19ede-32bd-43ac-9f60-0651f6f40afe
+
+# Property filter (Notion's filter shape — passed verbatim to data_sources.query)
+python scripts/notion_search.py "MCP" --databases ID \
+  --filter '{"property": "Status", "status": {"equals": "Done"}}'
+
+# Compound filter (and/or)
+python scripts/notion_search.py "MCP" --databases ID \
+  --filter '{"and": [{"property": "Status", "status": {"equals": "Done"}}, {"property": "Topic", "multi_select": {"contains": "AI"}}]}'
+
+# Fast mode (skip LLM stages)
+python scripts/notion_search.py "exact term" --no-rerank --no-expand
+```
+
+The search runs four stages:
+
+1. **Expand** — Claude Haiku generates up to 5 query variants (synonyms + cross-language KR↔EN)
+2. **Search** — Notion API searched per variant; results unioned + deduped at 30 candidates
+3. **Enrich** — title, properties, and 200-char excerpt fetched per candidate
+4. **Rerank** — Claude Haiku scores candidates against the *original* query; top N returned
+
+Results are cached for 24h (SHA256 of query + candidate IDs). Bypass with `--no-cache`.
+
+### Requirements
+
+| Env var | Purpose |
+|---------|---------|
+| `NOTION_API_KEY` (or legacy `NOTION_TOKEN`) | Notion integration token |
+| `ANTHROPIC_API_KEY` (optional) | Use Claude SDK directly. If missing, the skill falls back to `claude -p` CLI. |
+
 ## Scripts
 
 | Script | Purpose |

custom-skills/31-notion-organizer/code/scripts/_notion_compat.py

@@ -0,0 +1 @@
+../../../32-notion-writer/code/scripts/_notion_compat.py

custom-skills/31-notion-organizer/code/scripts/_search_cache.py

@@ -0,0 +1,73 @@
+"""Filesystem cache for rerank results, keyed by SHA256(query + candidate_ids)."""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import time
+from pathlib import Path
+from typing import Any, Dict, List, Optional
+
+DEFAULT_CACHE_DIR = Path.home() / ".cache" / "notion-search"
+DEFAULT_TTL_SECONDS = 86400  # 24 hours
+
+
+def _cache_key(query: str, candidate_ids: List[str]) -> str:
+    payload = f"{query}|{','.join(sorted(candidate_ids))}"
+    return hashlib.sha256(payload.encode("utf-8")).hexdigest()
+
+
+def cache_get(
+    query: str,
+    candidate_ids: List[str],
+    *,
+    cache_dir: Path = DEFAULT_CACHE_DIR,
+    ttl_seconds: int = DEFAULT_TTL_SECONDS,
+) -> Optional[List[Dict[str, Any]]]:
+    """Return cached results if file exists and is fresh, else None.
+
+    Corrupted cache files are silently treated as misses (and removed).
+    `ttl_seconds <= 0` always returns None (deterministic "always expired").
+    """
+    if ttl_seconds <= 0:
+        return None
+
+    key = _cache_key(query, candidate_ids)
+    path = cache_dir / f"{key}.json"
+    if not path.exists():
+        return None
+
+    try:
+        data = json.loads(path.read_text(encoding="utf-8"))
+    except (json.JSONDecodeError, OSError):
+        try:
+            path.unlink()
+        except OSError:
+            pass
+        return None
+
+    cached_at = data.get("cached_at_epoch", 0)
+    if time.time() - cached_at > ttl_seconds:
+        return None
+
+    return data.get("results")
+
+
+def cache_put(
+    query: str,
+    candidate_ids: List[str],
+    results: List[Dict[str, Any]],
+    *,
+    cache_dir: Path = DEFAULT_CACHE_DIR,
+) -> None:
+    """Write results to cache. Creates cache_dir if missing."""
+    cache_dir.mkdir(parents=True, exist_ok=True)
+    key = _cache_key(query, candidate_ids)
+    path = cache_dir / f"{key}.json"
+    payload = {
+        "query": query,
+        "candidate_ids": list(candidate_ids),
+        "results": results,
+        "cached_at_epoch": time.time(),
+    }
+    path.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")

custom-skills/31-notion-organizer/code/scripts/_search_llm.py

@@ -0,0 +1,71 @@
+"""Claude client abstraction. SDK preferred, `claude -p` CLI fallback."""
+
+from __future__ import annotations
+
+import os
+import shutil
+import subprocess
+
+# Anthropic SDK requires fully-qualified dated model IDs (the bare alias
+# `claude-haiku-4-5` works in Claude Code CLI but may not resolve via SDK).
+DEFAULT_MODEL = "claude-haiku-4-5-20251001"
+
+
+def _have_anthropic_sdk() -> bool:
+    try:
+        import anthropic  # noqa: F401
+        return True
+    except ImportError:
+        return False
+
+
+def _have_claude_cli() -> bool:
+    return shutil.which("claude") is not None
+
+
+def _call_via_sdk(prompt: str, model: str, max_tokens: int) -> str:
+    import anthropic
+    client = anthropic.Anthropic()
+    response = client.messages.create(
+        model=model,
+        max_tokens=max_tokens,
+        messages=[{"role": "user", "content": prompt}],
+    )
+    return response.content[0].text
+
+
+def _call_via_cli(prompt: str, model: str, max_tokens: int) -> str:
+    """Shell out to Claude Code CLI. max_tokens is ignored (CLI handles defaults)."""
+    result = subprocess.run(
+        ["claude", "-p", prompt, "--model", model],
+        capture_output=True,
+        text=True,
+        check=True,
+        timeout=60,
+    )
+    return result.stdout.strip()
+
+
+def call_claude(
+    prompt: str,
+    *,
+    model: str = DEFAULT_MODEL,
+    max_tokens: int = 1000,
+) -> str:
+    """Send a prompt to Claude.
+
+    Tries the anthropic SDK first (requires ANTHROPIC_API_KEY), falls back to
+    the `claude -p` CLI if available. Raises RuntimeError if neither works.
+
+    Note: ``max_tokens`` is honored by the SDK path but ignored by the CLI
+    path (the CLI sets its own defaults).
+    """
+    if _have_anthropic_sdk() and os.getenv("ANTHROPIC_API_KEY"):
+        return _call_via_sdk(prompt, model, max_tokens)
+    if _have_claude_cli():
+        return _call_via_cli(prompt, model, max_tokens)
+    raise RuntimeError(
+        "No Claude client available. Either:\n"
+        "  - Install the `anthropic` SDK and set ANTHROPIC_API_KEY, or\n"
+        "  - Install Claude Code CLI (`claude` on PATH)"
+    )

custom-skills/31-notion-organizer/code/scripts/notion_search.py

@@ -0,0 +1,474 @@
+#!/usr/bin/env python3
+"""Notion semantic search — query expansion + LLM rerank over Notion API search.
+
+CLI: python3 notion_search.py "query" [--databases ID,...] [--filter JSON] \
+                                       [--limit N] [--no-rerank] [--no-expand] \
+                                       [--no-cache] [--json]
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import re
+import sys
+from pathlib import Path
+from typing import Callable, Dict, List, Optional
+
+import _notion_compat as compat
+import _search_cache
+from _search_llm import call_claude as default_llm_caller
+
+EXPAND_PROMPT = """You are a query expander for a Notion semantic search tool. Generate up to {n} variants of the user's query that capture related concepts, synonyms, and cross-language alternates (especially Korean ↔ English).
+
+Rules:
+- Always include the original query verbatim as the first variant.
+- Variants should help find pages that the original query might miss due to keyword-only search.
+- For Korean queries, include English synonyms; for English queries, include Korean alternates if the topic has common Korean usage.
+- Keep variants concise (under 8 words each).
+
+Query: {query}
+
+Return ONLY a JSON array of strings, no prose. Example:
+["original query", "synonym variant", "cross-language variant"]"""
+
+
+def expand_query(
+    query: str,
+    *,
+    llm_caller: Optional[Callable[..., str]] = None,
+    max_variants: int = 5,
+) -> List[str]:
+    """Expand a query into related variants. Returns [query] on any failure."""
+    if llm_caller is None:
+        llm_caller = default_llm_caller
+
+    prompt = EXPAND_PROMPT.format(n=max_variants, query=query)
+
+    try:
+        response = llm_caller(prompt)
+    except Exception as exc:
+        print(f"Warning: query expansion failed ({exc}); using original query only",
+              file=sys.stderr)
+        return [query]
+
+    # Be permissive: extract the first JSON array from the response.
+    match = re.search(r'\[.*\]', response, re.DOTALL)
+    if not match:
+        print("Warning: query expansion returned no JSON array; using original query only",
+              file=sys.stderr)
+        return [query]
+
+    try:
+        variants = json.loads(match.group(0))
+    except json.JSONDecodeError:
+        print("Warning: query expansion returned invalid JSON; using original query only",
+              file=sys.stderr)
+        return [query]
+
+    if not isinstance(variants, list) or not all(isinstance(v, str) for v in variants):
+        print("Warning: query expansion returned non-string-list shape; using original query only",
+              file=sys.stderr)
+        return [query]
+
+    # Always include the original query first; dedupe; cap at max_variants.
+    seen = set()
+    result = []
+    for v in [query] + variants:
+        v = v.strip()
+        if v and v not in seen:
+            seen.add(v)
+            result.append(v)
+        if len(result) >= max_variants:
+            break
+    return result
+
+
+MAX_CANDIDATES = 30
+
+
+def search_candidates(
+    notion,
+    queries: List[str],
+    *,
+    databases: Optional[List[str]] = None,
+    prop_filter: Optional[Dict] = None,
+) -> List[Dict]:
+    """Run each query against Notion (workspace or per-DB), dedupe, cap at MAX_CANDIDATES.
+
+    Returns a list of page result objects (whatever Notion returned), order preserves
+    first-seen position across all variants — used as fallback ordering when rerank is off.
+    """
+    seen_ids = set()
+    candidates: List[Dict] = []
+
+    for query in queries:
+        if databases:
+            # Per-database query (data_sources.query)
+            for db_id in databases:
+                try:
+                    data_source_id = compat.resolve_data_source_id(notion, db_id)
+                except Exception as exc:
+                    print(f"Warning: could not resolve database {db_id}: {exc}",
+                          file=sys.stderr)
+                    continue
+
+                params = {"data_source_id": data_source_id, "page_size": 100}
+                if prop_filter:
+                    params["filter"] = prop_filter
+
+                try:
+                    response = notion.data_sources.query(**params)
+                except Exception as exc:
+                    print(f"Warning: query failed for database {db_id}: {exc}",
+                          file=sys.stderr)
+                    continue
+                for page in response.get("results") or []:
+                    page_id = page.get("id")
+                    if page_id and page_id not in seen_ids:
+                        seen_ids.add(page_id)
+                        candidates.append(page)
+                        if len(candidates) >= MAX_CANDIDATES:
+                            return candidates
+        else:
+            # Workspace-wide search
+            try:
+                response = notion.search(query=query, page_size=100)
+            except Exception as exc:
+                print(f"Warning: workspace search failed for variant {query!r}: {exc}",
+                      file=sys.stderr)
+                continue
+            for item in response.get("results") or []:
+                if item.get("object") != "page":
+                    continue
+                page_id = item.get("id")
+                if page_id and page_id not in seen_ids:
+                    seen_ids.add(page_id)
+                    candidates.append(item)
+                    if len(candidates) >= MAX_CANDIDATES:
+                        return candidates
+
+    return candidates
+
+
+EXCERPT_MAX_CHARS = 200
+TEXT_BEARING_BLOCK_TYPES = {"paragraph", "heading_1", "heading_2", "heading_3",
+                             "quote", "callout", "bulleted_list_item",
+                             "numbered_list_item", "to_do", "toggle"}
+
+
+def _flatten_property(prop: Dict):
+    """Flatten a Notion property to a Python value suitable for display/rerank."""
+    ptype = prop.get("type")
+    if ptype == "title":
+        return "".join(t.get("plain_text", "") for t in prop.get("title", []))
+    if ptype == "rich_text":
+        return "".join(t.get("plain_text", "") for t in prop.get("rich_text", []))
+    if ptype == "select":
+        sel = prop.get("select")
+        return sel.get("name") if sel else None
+    if ptype == "status":
+        st = prop.get("status")
+        return st.get("name") if st else None
+    if ptype == "multi_select":
+        return [o.get("name") for o in prop.get("multi_select", [])]
+    if ptype == "date":
+        return prop.get("date")
+    if ptype == "checkbox":
+        return prop.get("checkbox")
+    if ptype == "number":
+        return prop.get("number")
+    if ptype == "url":
+        return prop.get("url")
+    if ptype == "email":
+        return prop.get("email")
+    if ptype == "phone_number":
+        return prop.get("phone_number")
+    return None
+
+
+def _extract_title(properties: Dict) -> str:
+    for prop in properties.values():
+        if prop.get("type") == "title":
+            return "".join(t.get("plain_text", "") for t in prop.get("title", []))
+    return ""
+
+
+def _block_text(block: Dict) -> str:
+    """Concatenate plain_text from a block's rich_text array, if any."""
+    btype = block.get("type")
+    if btype not in TEXT_BEARING_BLOCK_TYPES:
+        return ""
+    body = block.get(btype, {})
+    rich_text = body.get("rich_text", [])
+    return "".join(t.get("plain_text", "") for t in rich_text)
+
+
+def _fetch_excerpt(notion, page_id: str) -> str:
+    """Fetch first text-bearing block; return its plain text capped at EXCERPT_MAX_CHARS."""
+    try:
+        response = notion.blocks.children.list(block_id=page_id, page_size=5)
+    except Exception:
+        return ""
+    for block in response.get("results") or []:
+        text = _block_text(block).strip()
+        if text:
+            return text[:EXCERPT_MAX_CHARS]
+    return ""
+
+
+def enrich_candidates(notion, candidates: List[Dict]) -> List[Dict]:
+    """Add title, flattened properties, and 200-char excerpt to each candidate."""
+    enriched = []
+    for c in candidates:
+        properties = c.get("properties", {})
+        title = _extract_title(properties)
+        flat_props = {}
+        for name, prop in properties.items():
+            if prop.get("type") == "title":
+                continue
+            value = _flatten_property(prop)
+            if value not in (None, [], ""):
+                flat_props[name] = value
+        excerpt = _fetch_excerpt(notion, c["id"])
+        enriched.append({
+            "id": c["id"],
+            "url": c.get("url", ""),
+            "title": title,
+            "properties": flat_props,
+            "excerpt": excerpt,
+        })
+    return enriched
+
+
+RERANK_PROMPT = """You are a reranker for Notion semantic search. Score each candidate 0.0-1.0 by how relevant it is to the user's ORIGINAL query (not any expanded variants).
+
+User's query: {query}
+
+Candidates:
+{candidates}
+
+Return ONLY a JSON array, ordered however you like. Each object has:
+- "index": integer matching the candidate's [N] number
+- "score": float 0.0-1.0
+- "why": one short sentence (under 80 chars) explaining the score
+
+Example output:
+[{{"index": 0, "score": 0.95, "why": "Direct match — covers exactly this topic"}},
+ {{"index": 2, "score": 0.6, "why": "Adjacent — shares context but not topic"}}]"""
+
+
+def _format_candidate_for_rerank(idx: int, c: Dict) -> str:
+    parts = [f"[{idx}] {c['title']}"]
+    props_str = ", ".join(f"{k}: {v}" for k, v in c.get("properties", {}).items())
+    if props_str:
+        parts.append(f"  Properties: {props_str}")
+    if c.get("excerpt"):
+        parts.append(f"  Excerpt: {c['excerpt']}")
+    return "\n".join(parts)
+
+
+def _fallback_rank(candidates: List[Dict], limit: int) -> List[Dict]:
+    """Return candidates in input order with null relevance/snippet."""
+    return [
+        {**c, "relevance": None, "snippet": None}
+        for c in candidates[:limit]
+    ]
+
+
+def rerank(
+    query: str,
+    candidates: List[Dict],
+    *,
+    llm_caller: Optional[Callable[..., str]] = None,
+    limit: int = 10,
+) -> List[Dict]:
+    """Rerank candidates against the original query. Fallback to input order on failure."""
+    if llm_caller is None:
+        llm_caller = default_llm_caller
+    if not candidates:
+        return []
+
+    formatted = "\n\n".join(
+        _format_candidate_for_rerank(i, c) for i, c in enumerate(candidates)
+    )
+    prompt = RERANK_PROMPT.format(query=query, candidates=formatted)
+
+    try:
+        response = llm_caller(prompt)
+    except Exception as exc:
+        print(f"Warning: rerank failed ({exc}); returning unranked results",
+              file=sys.stderr)
+        return _fallback_rank(candidates, limit)
+
+    match = re.search(r'\[.*\]', response, re.DOTALL)
+    if not match:
+        print("Warning: rerank returned no JSON array; returning unranked results",
+              file=sys.stderr)
+        return _fallback_rank(candidates, limit)
+
+    try:
+        scored = json.loads(match.group(0))
+    except json.JSONDecodeError:
+        print("Warning: rerank returned invalid JSON; returning unranked results",
+              file=sys.stderr)
+        return _fallback_rank(candidates, limit)
+
+    if not isinstance(scored, list):
+        print("Warning: rerank returned non-list shape; returning unranked results",
+              file=sys.stderr)
+        return _fallback_rank(candidates, limit)
+
+    # Sort by score descending and map back to candidates
+    scored.sort(key=lambda s: s.get("score", 0), reverse=True)
+    out = []
+    for entry in scored[:limit]:
+        idx = entry.get("index")
+        if not isinstance(idx, int) or idx < 0 or idx >= len(candidates):
+            continue
+        c = candidates[idx]
+        out.append({
+            **c,
+            "relevance": float(entry.get("score", 0.0)),
+            "snippet": entry.get("why", ""),
+        })
+    return out
+
+
+def run_search(
+    notion,
+    query: str,
+    *,
+    databases: Optional[List[str]] = None,
+    prop_filter: Optional[Dict] = None,
+    limit: int = 10,
+    no_rerank: bool = False,
+    no_expand: bool = False,
+    use_cache: bool = True,
+    cache_dir: Optional[Path] = None,
+    expand_llm: Optional[Callable[..., str]] = None,
+    rerank_llm: Optional[Callable[..., str]] = None,
+) -> List[Dict]:
+    """Full pipeline: expand → search → enrich → rerank → return.
+
+    expand_llm and rerank_llm are dependency-injected for tests.
+    Output schema (each result dict): id, url, title, relevance (or None),
+    snippet (or None), excerpt, properties.
+    """
+    # Stage 1: expand
+    if no_expand:
+        queries = [query]
+    else:
+        queries = expand_query(query, llm_caller=expand_llm)
+
+    # Stage 2: search
+    candidates = search_candidates(
+        notion, queries, databases=databases, prop_filter=prop_filter,
+    )
+    if not candidates:
+        return []
+
+    # Stage 3: enrich
+    enriched = enrich_candidates(notion, candidates)
+
+    # Stage 4: rerank (or skip)
+    if no_rerank:
+        return [
+            {**c, "relevance": None, "snippet": None}
+            for c in enriched[:limit]
+        ]
+
+    candidate_ids = [c["id"] for c in enriched]
+    # Pass cache_dir only if explicitly set; otherwise let _search_cache use its default.
+    cache_kwargs = {"cache_dir": cache_dir} if cache_dir else {}
+
+    if use_cache:
+        cached = _search_cache.cache_get(query, candidate_ids, **cache_kwargs)
+        if cached is not None:
+            return cached
+
+    ranked = rerank(query, enriched, llm_caller=rerank_llm, limit=limit)
+
+    if use_cache:
+        _search_cache.cache_put(query, candidate_ids, ranked, **cache_kwargs)
+
+    return ranked
+
+
+def format_terminal(results: List[Dict]) -> str:
+    """Human-readable terminal output."""
+    if not results:
+        return "No matches.\n"
+    lines = []
+    for i, r in enumerate(results, 1):
+        rel = f"(rel: {r['relevance']:.2f}) " if r.get("relevance") is not None else ""
+        lines.append(f"[{i}] {rel}{r['title']}")
+        props = r.get("properties", {})
+        if props:
+            prop_strs = []
+            for k, v in props.items():
+                if isinstance(v, list):
+                    v = ", ".join(str(x) for x in v)
+                prop_strs.append(f"{k}: {v}")
+            lines.append(f"    {' · '.join(prop_strs)}")
+        if r.get("snippet"):
+            lines.append(f"    Why: {r['snippet']}")
+        if r.get("url"):
+            lines.append(f"    {r['url']}")
+        lines.append("")
+    return "\n".join(lines).rstrip() + "\n"
+
+
+def main():
+    parser = argparse.ArgumentParser(
+        prog="notion-search",
+        description="Semantic search across Notion workspace via LLM expand + rerank.",
+    )
+    parser.add_argument("query", help="Search query (natural language)")
+    parser.add_argument("--databases", "-d", default=None,
+                        help="Comma-separated database/data_source IDs (default: workspace-wide)")
+    parser.add_argument("--filter", "-f", default=None,
+                        help="JSON property filter (per-database mode only)")
+    parser.add_argument("--limit", "-l", type=int, default=10,
+                        help="Max results after rerank (default: 10)")
+    parser.add_argument("--no-rerank", action="store_true",
+                        help="Skip Claude rerank stage")
+    parser.add_argument("--no-expand", action="store_true",
+                        help="Skip query-variant generation")
+    parser.add_argument("--no-cache", action="store_true",
+                        help="Bypass result cache")
+    parser.add_argument("--json", action="store_true",
+                        help="Output JSON array instead of terminal table")
+    args = parser.parse_args()
+
+    # Parse databases and filter
+    databases = args.databases.split(",") if args.databases else None
+    prop_filter = json.loads(args.filter) if args.filter else None
+    if prop_filter and not databases:
+        print("Warning: --filter is only applied in per-database mode; ignored without --databases",
+              file=sys.stderr)
+        prop_filter = None
+
+    # Build notion client
+    api_key = os.getenv("NOTION_API_KEY") or os.getenv("NOTION_TOKEN")
+    if not api_key:
+        print("Error: NOTION_API_KEY (or NOTION_TOKEN) not set", file=sys.stderr)
+        sys.exit(1)
+    notion = compat.make_client(api_key)
+
+    results = run_search(
+        notion, args.query,
+        databases=databases, prop_filter=prop_filter, limit=args.limit,
+        no_rerank=args.no_rerank, no_expand=args.no_expand,
+        use_cache=not args.no_cache,
+    )
+
+    if args.json:
+        print(json.dumps(results, ensure_ascii=False, indent=2))
+    else:
+        print(format_terminal(results))
+
+
+if __name__ == "__main__":
+    main()

custom-skills/31-notion-organizer/code/scripts/requirements.txt

@@ -24,3 +24,7 @@ tqdm==4.66.1
 
 # Optional: Fuzzy matching for duplicates
 # rapidfuzz==3.5.2
+
+# Optional: required only for direct Anthropic SDK use.
+# If missing, the search skill falls back to `claude -p` CLI.
+anthropic>=0.40.0

custom-skills/31-notion-organizer/code/scripts/test_notion_search.py

@@ -0,0 +1,639 @@
+#!/usr/bin/env python3
+"""Tests for notion_search.py — run with `python3 test_notion_search.py`."""
+
+import sys
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).parent))
+
+
+def _assert(cond, msg):
+    if not cond:
+        print(f"  ✗ FAIL: {msg}")
+        raise SystemExit(1)
+    print(f"  ✓ {msg}")
+
+
+def test_call_claude_dispatches_to_sdk_when_available():
+    """When SDK is available and ANTHROPIC_API_KEY is set, use SDK path."""
+    import os
+    from unittest.mock import patch
+    import _search_llm
+
+    with patch.object(_search_llm, "_have_anthropic_sdk", return_value=True), \
+         patch.dict(os.environ, {"ANTHROPIC_API_KEY": "sk-fake"}), \
+         patch.object(_search_llm, "_call_via_sdk", return_value="sdk-response") as sdk_mock, \
+         patch.object(_search_llm, "_call_via_cli", return_value="cli-response") as cli_mock:
+        result = _search_llm.call_claude("hello")
+        _assert(result == "sdk-response", "SDK path returned its response")
+        _assert(sdk_mock.called, "SDK path was invoked")
+        _assert(not cli_mock.called, "CLI path was NOT invoked")
+
+
+def test_call_claude_falls_back_to_cli_when_no_sdk():
+    """When SDK is missing but CLI is available, use CLI path."""
+    from unittest.mock import patch
+    import _search_llm
+
+    with patch.object(_search_llm, "_have_anthropic_sdk", return_value=False), \
+         patch.object(_search_llm, "_have_claude_cli", return_value=True), \
+         patch.object(_search_llm, "_call_via_cli", return_value="cli-response") as cli_mock:
+        result = _search_llm.call_claude("hello")
+        _assert(result == "cli-response", "CLI path returned its response")
+        _assert(cli_mock.called, "CLI path was invoked")
+
+
+def test_call_claude_raises_when_neither_available():
+    """Both clients missing → RuntimeError with setup hint."""
+    from unittest.mock import patch
+    import _search_llm
+
+    with patch.object(_search_llm, "_have_anthropic_sdk", return_value=False), \
+         patch.object(_search_llm, "_have_claude_cli", return_value=False):
+        try:
+            _search_llm.call_claude("hello")
+            _assert(False, "should have raised RuntimeError")
+        except RuntimeError as exc:
+            _assert("ANTHROPIC_API_KEY" in str(exc) or "claude" in str(exc).lower(),
+                    "error mentions setup options")
+
+
+def test_cache_miss_returns_none():
+    import tempfile
+    from pathlib import Path
+    import _search_cache
+
+    with tempfile.TemporaryDirectory() as tmpdir:
+        result = _search_cache.cache_get("query", ["id1", "id2"], cache_dir=Path(tmpdir))
+        _assert(result is None, "cache miss returns None")
+
+
+def test_cache_hit_returns_cached_results():
+    import tempfile
+    from pathlib import Path
+    import _search_cache
+
+    with tempfile.TemporaryDirectory() as tmpdir:
+        cache_dir = Path(tmpdir)
+        results = [{"id": "abc", "title": "Test"}]
+        _search_cache.cache_put("query", ["id1", "id2"], results, cache_dir=cache_dir)
+        hit = _search_cache.cache_get("query", ["id1", "id2"], cache_dir=cache_dir)
+        _assert(hit == results, "cache hit returns the stored results")
+
+
+def test_cache_miss_on_different_candidates():
+    import tempfile
+    from pathlib import Path
+    import _search_cache
+
+    with tempfile.TemporaryDirectory() as tmpdir:
+        cache_dir = Path(tmpdir)
+        _search_cache.cache_put("query", ["id1", "id2"], [{"x": 1}], cache_dir=cache_dir)
+        miss = _search_cache.cache_get("query", ["id1", "id3"], cache_dir=cache_dir)
+        _assert(miss is None, "different candidate set hashes to different key")
+
+
+def test_cache_miss_after_ttl_expires():
+    import tempfile
+    from pathlib import Path
+    import _search_cache
+
+    with tempfile.TemporaryDirectory() as tmpdir:
+        cache_dir = Path(tmpdir)
+        _search_cache.cache_put("query", ["id1"], [{"x": 1}], cache_dir=cache_dir)
+        # TTL=0 means immediately expired
+        miss = _search_cache.cache_get("query", ["id1"], cache_dir=cache_dir, ttl_seconds=0)
+        _assert(miss is None, "expired cache returns None")
+
+
+def test_cache_handles_corrupted_file():
+    import tempfile
+    import hashlib
+    from pathlib import Path
+    import _search_cache
+
+    with tempfile.TemporaryDirectory() as tmpdir:
+        cache_dir = Path(tmpdir)
+        cache_dir.mkdir(exist_ok=True)
+        # Write garbage to the expected key path
+        key = hashlib.sha256("query|id1".encode()).hexdigest()
+        (cache_dir / f"{key}.json").write_text("not json {{{")
+        result = _search_cache.cache_get("query", ["id1"], cache_dir=cache_dir)
+        _assert(result is None, "corrupted cache file returns None")
+
+
+def test_expand_query_returns_variants_with_original_first():
+    """LLM mock returns variants; expansion includes original verbatim as first item."""
+    import notion_search
+
+    fake_llm = lambda prompt, **kw: '["AI agents", "multi-agent systems", "autonomous LLMs"]'
+    variants = notion_search.expand_query("AI agents", llm_caller=fake_llm)
+    _assert(variants[0] == "AI agents", "original query is first variant")
+    _assert(len(variants) >= 2, "at least 2 variants returned")
+    _assert("multi-agent systems" in variants, "LLM-suggested variant included")
+
+
+def test_expand_query_dedupes_and_caps():
+    """Duplicates removed; total ≤ max_variants."""
+    import notion_search
+
+    fake_llm = lambda prompt, **kw: '["AI", "AI", "agents", "agents", "systems", "tools", "models"]'
+    variants = notion_search.expand_query("AI", llm_caller=fake_llm, max_variants=3)
+    _assert(len(variants) <= 3, "capped at max_variants=3")
+    _assert(len(variants) == len(set(variants)), "no duplicates in result")
+
+
+def test_expand_query_falls_back_on_invalid_json():
+    """LLM returns prose instead of JSON → fall back to [original]."""
+    import notion_search
+
+    fake_llm = lambda prompt, **kw: "I'm sorry, I can't help with that."
+    variants = notion_search.expand_query("AI agents", llm_caller=fake_llm)
+    _assert(variants == ["AI agents"], "non-JSON response falls back to original only")
+
+
+def test_expand_query_falls_back_on_llm_exception():
+    """LLM raises → fall back to [original]."""
+    import notion_search
+
+    def boom(prompt, **kw):
+        raise RuntimeError("API error")
+
+    variants = notion_search.expand_query("AI agents", llm_caller=boom)
+    _assert(variants == ["AI agents"], "LLM exception falls back to original only")
+
+
+def test_search_unions_and_dedupes_workspace():
+    """Two variants returning overlapping pages produce a deduped candidate set."""
+    from unittest.mock import MagicMock
+    import notion_search
+
+    notion = MagicMock()
+    notion.search.side_effect = [
+        {"results": [
+            {"id": "page-a", "object": "page", "url": "https://notion.so/a"},
+            {"id": "page-b", "object": "page", "url": "https://notion.so/b"},
+        ]},
+        {"results": [
+            {"id": "page-b", "object": "page", "url": "https://notion.so/b"},
+            {"id": "page-c", "object": "page", "url": "https://notion.so/c"},
+        ]},
+    ]
+    candidates = notion_search.search_candidates(
+        notion, ["query1", "query2"], databases=None
+    )
+    ids = [c["id"] for c in candidates]
+    _assert(set(ids) == {"page-a", "page-b", "page-c"}, "union of pages from both variants")
+    _assert(len(ids) == 3, "duplicate page-b deduped")
+
+
+def test_search_caps_candidates_at_30():
+    """If total candidates exceed 30, cap to 30."""
+    from unittest.mock import MagicMock
+    import notion_search
+
+    notion = MagicMock()
+    notion.search.return_value = {
+        "results": [
+            {"id": f"page-{i:02d}", "object": "page", "url": f"https://notion.so/{i}"}
+            for i in range(40)
+        ]
+    }
+    candidates = notion_search.search_candidates(notion, ["query"], databases=None)
+    _assert(len(candidates) == 30, f"capped to 30 (got {len(candidates)})")
+
+
+def test_search_uses_data_source_query_when_databases_specified():
+    """--databases flag → per-DB data_sources.query instead of workspace search."""
+    from unittest.mock import MagicMock, patch
+    import notion_search
+
+    notion = MagicMock()
+    notion.data_sources.query.return_value = {
+        "results": [{"id": "page-a", "url": "https://notion.so/a"}]
+    }
+
+    with patch("notion_search.compat.resolve_data_source_id",
+               side_effect=lambda c, ds: ds):
+        candidates = notion_search.search_candidates(
+            notion, ["query"], databases=["db-id-1"]
+        )
+    _assert(notion.data_sources.query.called, "data_sources.query was called")
+    _assert(not notion.search.called, "workspace search NOT called when databases specified")
+    _assert(len(candidates) == 1, "candidate from data source returned")
+
+
+def test_search_filters_only_pages():
+    """search() may return databases too; we keep only object='page'."""
+    from unittest.mock import MagicMock
+    import notion_search
+
+    notion = MagicMock()
+    notion.search.return_value = {"results": [
+        {"id": "page-a", "object": "page", "url": "https://notion.so/a"},
+        {"id": "db-1", "object": "database", "url": "https://notion.so/db1"},
+    ]}
+    candidates = notion_search.search_candidates(notion, ["query"], databases=None)
+    ids = [c["id"] for c in candidates]
+    _assert("page-a" in ids, "page kept")
+    _assert("db-1" not in ids, "database object filtered out")
+
+
+def test_search_passes_property_filter_to_data_sources_query():
+    """--filter JSON is passed through to data_sources.query as the `filter` parameter."""
+    from unittest.mock import MagicMock, patch
+    import notion_search
+
+    notion = MagicMock()
+    notion.data_sources.query.return_value = {"results": []}
+
+    prop_filter = {"property": "Status", "status": {"equals": "Done"}}
+
+    with patch("notion_search.compat.resolve_data_source_id",
+               side_effect=lambda c, ds: ds):
+        notion_search.search_candidates(
+            notion, ["query"], databases=["db-id-1"], prop_filter=prop_filter,
+        )
+
+    _assert(notion.data_sources.query.called, "data_sources.query was called")
+    call_kwargs = notion.data_sources.query.call_args.kwargs
+    _assert(call_kwargs.get("filter") == prop_filter,
+            "prop_filter passed through as `filter` keyword argument")
+
+
+def test_enrich_extracts_title_and_properties():
+    """Candidate with properties → title and properties extracted."""
+    from unittest.mock import MagicMock
+    import notion_search
+
+    notion = MagicMock()
+    notion.blocks.children.list.return_value = {"results": []}  # no body
+
+    candidate = {
+        "id": "page-a",
+        "url": "https://notion.so/a",
+        "properties": {
+            "Name": {
+                "type": "title",
+                "title": [{"plain_text": "Test Page"}],
+            },
+            "Status": {
+                "type": "status",
+                "status": {"name": "Done"},
+            },
+            "Topic": {
+                "type": "multi_select",
+                "multi_select": [{"name": "AI"}, {"name": "MCP"}],
+            },
+        },
+    }
+    enriched = notion_search.enrich_candidates(notion, [candidate])
+    _assert(len(enriched) == 1, "one enriched candidate")
+    e = enriched[0]
+    _assert(e["title"] == "Test Page", "title extracted from title property")
+    _assert(e["properties"]["Status"] == "Done", "status flattened to name")
+    _assert(e["properties"]["Topic"] == ["AI", "MCP"], "multi_select flattened to names list")
+
+
+def test_enrich_extracts_first_paragraph_excerpt():
+    """First paragraph block → excerpt (200-char max)."""
+    from unittest.mock import MagicMock
+    import notion_search
+
+    notion = MagicMock()
+    notion.blocks.children.list.return_value = {
+        "results": [
+            {
+                "type": "paragraph",
+                "paragraph": {
+                    "rich_text": [{"plain_text": "This is the first paragraph of the page."}]
+                },
+            }
+        ]
+    }
+    candidate = {
+        "id": "page-a", "url": "https://notion.so/a",
+        "properties": {"Name": {"type": "title", "title": [{"plain_text": "Test"}]}},
+    }
+    enriched = notion_search.enrich_candidates(notion, [candidate])
+    _assert(enriched[0]["excerpt"] == "This is the first paragraph of the page.",
+            "excerpt is first paragraph plain text")
+
+
+def test_enrich_falls_back_to_empty_excerpt():
+    """Page leads with image/table only → excerpt is empty string, no crash."""
+    from unittest.mock import MagicMock
+    import notion_search
+
+    notion = MagicMock()
+    notion.blocks.children.list.return_value = {
+        "results": [
+            {"type": "image", "image": {}},
+            {"type": "divider", "divider": {}},
+        ]
+    }
+    candidate = {
+        "id": "page-a", "url": "https://notion.so/a",
+        "properties": {"Name": {"type": "title", "title": [{"plain_text": "Test"}]}},
+    }
+    enriched = notion_search.enrich_candidates(notion, [candidate])
+    _assert(enriched[0]["excerpt"] == "", "empty excerpt when no text-bearing block")
+
+
+def test_enrich_truncates_long_excerpt():
+    """Excerpt is capped at 200 chars."""
+    from unittest.mock import MagicMock
+    import notion_search
+
+    notion = MagicMock()
+    long_text = "x" * 500
+    notion.blocks.children.list.return_value = {
+        "results": [
+            {"type": "paragraph", "paragraph": {"rich_text": [{"plain_text": long_text}]}}
+        ]
+    }
+    candidate = {
+        "id": "page-a", "url": "https://notion.so/a",
+        "properties": {"Name": {"type": "title", "title": [{"plain_text": "Test"}]}},
+    }
+    enriched = notion_search.enrich_candidates(notion, [candidate])
+    _assert(len(enriched[0]["excerpt"]) == 200, "excerpt truncated to 200 chars")
+
+
+def test_enrich_keeps_falsy_but_meaningful_values():
+    """checkbox=False and number=0 are meaningful, not 'empty' — must survive the filter."""
+    from unittest.mock import MagicMock
+    import notion_search
+
+    notion = MagicMock()
+    notion.blocks.children.list.return_value = {"results": []}
+
+    candidate = {
+        "id": "page-a", "url": "https://notion.so/a",
+        "properties": {
+            "Name": {"type": "title", "title": [{"plain_text": "Test"}]},
+            "Active": {"type": "checkbox", "checkbox": False},
+            "Score": {"type": "number", "number": 0},
+        },
+    }
+    enriched = notion_search.enrich_candidates(notion, [candidate])
+    _assert(enriched[0]["properties"].get("Active") is False,
+            "checkbox=False kept (not filtered as 'empty')")
+    _assert(enriched[0]["properties"].get("Score") == 0,
+            "number=0 kept (not filtered as 'empty')")
+
+
+def test_rerank_orders_by_score_descending():
+    """LLM returns scores; output sorted score-desc, top N."""
+    import notion_search
+
+    candidates = [
+        {"id": "p1", "url": "u1", "title": "Page 1", "properties": {}, "excerpt": "x"},
+        {"id": "p2", "url": "u2", "title": "Page 2", "properties": {}, "excerpt": "x"},
+        {"id": "p3", "url": "u3", "title": "Page 3", "properties": {}, "excerpt": "x"},
+    ]
+
+    # LLM returns: p3 most relevant, then p1, then p2
+    fake_llm = lambda prompt, **kw: '''[
+        {"index": 2, "score": 0.95, "why": "best match"},
+        {"index": 0, "score": 0.7, "why": "ok match"},
+        {"index": 1, "score": 0.3, "why": "weak match"}
+    ]'''
+
+    ranked = notion_search.rerank("query", candidates, llm_caller=fake_llm, limit=3)
+    _assert(ranked[0]["id"] == "p3", "highest score first")
+    _assert(ranked[0]["relevance"] == 0.95, "score attached to result")
+    _assert(ranked[0]["snippet"] == "best match", "why text attached as snippet")
+    _assert(ranked[1]["id"] == "p1", "second place correct")
+    _assert(ranked[2]["id"] == "p2", "lowest score last")
+
+
+def test_rerank_respects_limit():
+    """limit=2 → returns top 2 only."""
+    import notion_search
+
+    candidates = [
+        {"id": f"p{i}", "url": f"u{i}", "title": f"Page {i}", "properties": {}, "excerpt": "x"}
+        for i in range(5)
+    ]
+    fake_llm = lambda prompt, **kw: '''[
+        {"index": 0, "score": 0.9, "why": "x"},
+        {"index": 1, "score": 0.8, "why": "x"},
+        {"index": 2, "score": 0.7, "why": "x"},
+        {"index": 3, "score": 0.6, "why": "x"},
+        {"index": 4, "score": 0.5, "why": "x"}
+    ]'''
+    ranked = notion_search.rerank("query", candidates, llm_caller=fake_llm, limit=2)
+    _assert(len(ranked) == 2, "exactly limit results returned")
+
+
+def test_rerank_falls_back_on_parse_error():
+    """Non-JSON response → return candidates in input order, unranked (no scores)."""
+    import notion_search
+
+    candidates = [
+        {"id": "p1", "url": "u1", "title": "Page 1", "properties": {}, "excerpt": "x"},
+        {"id": "p2", "url": "u2", "title": "Page 2", "properties": {}, "excerpt": "x"},
+    ]
+    fake_llm = lambda prompt, **kw: "I cannot help with that."
+    ranked = notion_search.rerank("query", candidates, llm_caller=fake_llm, limit=10)
+    _assert(len(ranked) == 2, "all candidates returned in fallback")
+    _assert(ranked[0]["id"] == "p1", "fallback preserves input order")
+    _assert(ranked[0]["relevance"] is None, "no score in fallback")
+    _assert(ranked[0]["snippet"] is None, "no snippet in fallback")
+
+
+def test_rerank_falls_back_on_llm_exception():
+    """LLM raises → fallback unranked."""
+    import notion_search
+
+    candidates = [{"id": "p1", "url": "u1", "title": "Page 1", "properties": {}, "excerpt": "x"}]
+
+    def boom(prompt, **kw):
+        raise RuntimeError("API down")
+
+    ranked = notion_search.rerank("query", candidates, llm_caller=boom, limit=10)
+    _assert(len(ranked) == 1, "candidate returned despite LLM failure")
+    _assert(ranked[0]["relevance"] is None, "no score in fallback")
+
+
+def test_pipeline_search_returns_json_serializable_results():
+    """End-to-end pipeline returns results matching the spec's JSON schema."""
+    from unittest.mock import MagicMock
+    import json as json_mod
+    import notion_search
+
+    notion = MagicMock()
+    notion.search.return_value = {
+        "results": [
+            {
+                "id": "page-a", "object": "page", "url": "https://notion.so/a",
+                "properties": {
+                    "Name": {"type": "title", "title": [{"plain_text": "AI Agents Page"}]},
+                    "Status": {"type": "status", "status": {"name": "Done"}},
+                },
+            }
+        ]
+    }
+    notion.blocks.children.list.return_value = {
+        "results": [
+            {"type": "paragraph",
+             "paragraph": {"rich_text": [{"plain_text": "Notes about AI agents."}]}}
+        ]
+    }
+
+    expand_llm = lambda prompt, **kw: '["AI agents"]'
+    rerank_llm = lambda prompt, **kw: '[{"index": 0, "score": 0.9, "why": "match"}]'
+
+    results = notion_search.run_search(
+        notion, "AI agents",
+        expand_llm=expand_llm, rerank_llm=rerank_llm,
+        limit=10, use_cache=False,
+    )
+
+    _assert(len(results) == 1, "one result")
+    r = results[0]
+    _assert(r["id"] == "page-a", "id present")
+    _assert(r["title"] == "AI Agents Page", "title extracted")
+    _assert(r["relevance"] == 0.9, "relevance score from rerank")
+    _assert(r["properties"]["Status"] == "Done", "property included")
+    _assert(r["excerpt"] == "Notes about AI agents.", "excerpt included")
+    # Schema must be JSON-serializable
+    json_mod.dumps(results)
+
+
+def test_pipeline_no_rerank_skips_llm():
+    """--no-rerank flag → results have null relevance, no rerank LLM call."""
+    from unittest.mock import MagicMock
+    import notion_search
+
+    notion = MagicMock()
+    notion.search.return_value = {
+        "results": [
+            {"id": "p1", "object": "page", "url": "u1",
+             "properties": {"Name": {"type": "title", "title": [{"plain_text": "Page 1"}]}}}
+        ]
+    }
+    notion.blocks.children.list.return_value = {"results": []}
+
+    expand_llm = lambda prompt, **kw: '["query"]'
+
+    rerank_calls = []
+    def rerank_llm(prompt, **kw):
+        rerank_calls.append(prompt)
+        return "[]"
+
+    results = notion_search.run_search(
+        notion, "query",
+        expand_llm=expand_llm, rerank_llm=rerank_llm,
+        no_rerank=True, use_cache=False,
+    )
+    _assert(len(rerank_calls) == 0, "rerank LLM not called when --no-rerank")
+    _assert(results[0]["relevance"] is None, "no relevance score when no rerank")
+
+
+def test_pipeline_no_expand_uses_single_query():
+    """--no-expand → expand_llm not called, Notion gets only original query."""
+    from unittest.mock import MagicMock
+    import notion_search
+
+    notion = MagicMock()
+    notion.search.return_value = {"results": []}
+
+    expand_calls = []
+    def expand_llm(prompt, **kw):
+        expand_calls.append(prompt)
+        return '["should not be used"]'
+
+    rerank_llm = lambda prompt, **kw: "[]"
+
+    notion_search.run_search(
+        notion, "exact term",
+        expand_llm=expand_llm, rerank_llm=rerank_llm,
+        no_expand=True, use_cache=False,
+    )
+    _assert(len(expand_calls) == 0, "expand LLM not called when --no-expand")
+    _assert(notion.search.call_count == 1, "Notion search called exactly once")
+
+
+def test_pipeline_uses_cache_on_second_call():
+    """Two identical pipeline calls → second one skips rerank LLM."""
+    import tempfile
+    from pathlib import Path
+    from unittest.mock import MagicMock
+    import notion_search
+
+    notion = MagicMock()
+    notion.search.return_value = {
+        "results": [
+            {"id": "p1", "object": "page", "url": "u1",
+             "properties": {"Name": {"type": "title", "title": [{"plain_text": "Page 1"}]}}}
+        ]
+    }
+    notion.blocks.children.list.return_value = {"results": []}
+
+    expand_llm = lambda prompt, **kw: '["query"]'
+
+    rerank_calls = []
+    def rerank_llm(prompt, **kw):
+        rerank_calls.append(prompt)
+        return '[{"index": 0, "score": 0.5, "why": "x"}]'
+
+    with tempfile.TemporaryDirectory() as tmpdir:
+        cache_dir = Path(tmpdir)
+
+        notion_search.run_search(
+            notion, "query", expand_llm=expand_llm, rerank_llm=rerank_llm,
+            use_cache=True, cache_dir=cache_dir,
+        )
+        notion_search.run_search(
+            notion, "query", expand_llm=expand_llm, rerank_llm=rerank_llm,
+            use_cache=True, cache_dir=cache_dir,
+        )
+
+    _assert(len(rerank_calls) == 1, "second call skipped rerank LLM (cache hit)")
+
+
+def run_all():
+    tests = [
+        test_call_claude_dispatches_to_sdk_when_available,
+        test_call_claude_falls_back_to_cli_when_no_sdk,
+        test_call_claude_raises_when_neither_available,
+        test_cache_miss_returns_none,
+        test_cache_hit_returns_cached_results,
+        test_cache_miss_on_different_candidates,
+        test_cache_miss_after_ttl_expires,
+        test_cache_handles_corrupted_file,
+        test_expand_query_returns_variants_with_original_first,
+        test_expand_query_dedupes_and_caps,
+        test_expand_query_falls_back_on_invalid_json,
+        test_expand_query_falls_back_on_llm_exception,
+        test_search_unions_and_dedupes_workspace,
+        test_search_caps_candidates_at_30,
+        test_search_uses_data_source_query_when_databases_specified,
+        test_search_filters_only_pages,
+        test_search_passes_property_filter_to_data_sources_query,
+        test_enrich_extracts_title_and_properties,
+        test_enrich_extracts_first_paragraph_excerpt,
+        test_enrich_falls_back_to_empty_excerpt,
+        test_enrich_truncates_long_excerpt,
+        test_enrich_keeps_falsy_but_meaningful_values,
+        test_rerank_orders_by_score_descending,
+        test_rerank_respects_limit,
+        test_rerank_falls_back_on_parse_error,
+        test_rerank_falls_back_on_llm_exception,
+        test_pipeline_search_returns_json_serializable_results,
+        test_pipeline_no_rerank_skips_llm,
+        test_pipeline_no_expand_uses_single_query,
+        test_pipeline_uses_cache_on_second_call,
+    ]
+    for t in tests:
+        print(f"\n{t.__name__}")
+        t()
+    print("\n" + "=" * 50)
+    print(f"✅ All {len(tests)} tests passed")
+    print("=" * 50)
+
+
+if __name__ == "__main__":
+    run_all()

custom-skills/31-notion-organizer/commands/notion-search.md

@@ -0,0 +1,50 @@
+---
+description: Search Notion workspace semantically (LLM-expanded + reranked).
+argument-hint: "<query> [--databases ID,...] [--filter JSON] [--limit N] [--no-rerank] [--no-expand] [--no-cache] [--json]"
+---
+
+Search the user's Notion workspace using semantic search powered by Claude Haiku query expansion and result reranking. Closes the gap left by Notion's keyword-only native search — handles synonyms, related concepts, and Korean ↔ English cross-language queries.
+
+## Usage
+
+Run the search script with the user's arguments:
+
+```bash
+cd ~/Project/our-claude-skills/custom-skills/31-notion-organizer/code/scripts
+python3 notion_search.py "$ARGUMENTS"
+```
+
+## Common patterns
+
+- **Default browse mode** (terminal table): `notion-search "AI agents in 2026"`
+- **JSON for piping** (e.g. into the future notion-to-notebooklm push skill): `notion-search "AI agents" --json | jq '.[].id'`
+- **Constrain to specific databases**: `notion-search "MCP" --databases f8f19ede-32bd-43ac-9f60-0651f6f40afe`
+- **Property filter** (per-database mode, Notion's native filter shape): `notion-search "MCP" --databases ID --filter '{"property": "Status", "status": {"equals": "Done"}}'`
+- **Fast mode (no LLM)**: `notion-search "exact phrase" --no-rerank --no-expand`
+
+## Requirements
+
+- `NOTION_API_KEY` or `NOTION_TOKEN` env var
+- One of:
+  - `anthropic` SDK installed + `ANTHROPIC_API_KEY` env var, or
+  - Claude Code CLI (`claude` on PATH)
+
+The skill auto-detects which is available; SDK is preferred when both are present.
+
+## Output schema (JSON mode)
+
+```json
+[
+  {
+    "id": "abc-def-...",
+    "url": "https://notion.so/...",
+    "title": "Page title",
+    "relevance": 0.94,
+    "snippet": "Why it matched",
+    "excerpt": "First paragraph text...",
+    "properties": {"Status": "Done", "Topic": ["AI", "MCP"]}
+  }
+]
+```
+
+`relevance` and `snippet` are `null` when `--no-rerank` is set. This schema is the contract for downstream tools (e.g., the future `notion-to-notebooklm` push skill).

docs/superpowers/specs/2026-04-28-notion-semantic-search-design.md

@@ -0,0 +1,282 @@
+# Notion Semantic Search (Skill 31) — Design
+
+> **Date**: 2026-04-28
+> **Status**: Approved (brainstorming)
+> **Scope**: Phase 3b-i — semantic search across Notion workspaces with query expansion + LLM rerank
+> **Sequence**: First of two sub-projects derived from the original Phase 3b "Notion-as-RAG export". The push-to-NotebookLM skill is the second sub-project, gets its own design after this ships.
+> **Predecessor**: Phase 3c (commit `6446f00` — extended block coverage in `32-notion-writer`)
+
+---
+
+## Goal
+
+Build a CLI skill that searches the user's Notion workspace semantically. Output is either a terminal-friendly table for browsing or JSON for piping into the (future) `notion-to-notebooklm` push skill.
+
+The load-bearing problem: Notion's native search is keyword-only and weak — "AI agents" doesn't surface a page titled "Multi-agent orchestration", and Korean ↔ English queries don't cross-match. We close the gap with two LLM stages over Notion's API search: query expansion (Claude generates synonym/cross-language variants) and rerank (Claude scores candidates against the original query).
+
+This is the foundation for the rest of the 31-notion-organizer vision — aggregation, move/reorganize, and exports all operate on search results, so getting search right unblocks everything else.
+
+---
+
+## Non-goals
+
+- **Block-level result granularity.** Page-level only in v1. Surfacing matching block snippets within pages adds complexity and most retrieval is page-level anyway. Defer until a real use case demands it.
+- **Cross-workspace search.** Notion's integration model is per-workspace; abstracting over multiple workspaces means installing the integration in each. Document the limitation, don't try to hide it.
+- **Embedding-based search.** Vector store + sync pipeline pushes this back into "build a tool" territory. Out of scope for the skill model.
+- **The push-to-NotebookLM skill.** Separate spec, separate plan. Search outputs JSON; the push skill consumes it.
+- **Automatic pagination of search results.** Cap at ~30 candidates pre-rerank; this is enough for top-10 reranked results. If a query genuinely has 100+ relevant matches, the user can constrain with `--databases` or `--filter`.
+
+---
+
+## Architectural decisions
+
+| Decision | Choice | Rationale |
+|---|---|---|
+| Search strategy | **Strategy C — query expansion + rerank** | Notion's keyword search alone is the gap we're filling. Rerank-only catches synonyms but misses cross-language. Expansion + rerank covers both. |
+| Workflow shape | **Standalone search skill** (browse-friendly), JSON output for downstream chaining | User searches frequently for memory-refresh; pushing to NotebookLM is occasional. Two independent skills, composable via JSON. |
+| LLM model | **Claude Haiku 4.5** for both expansion and rerank | Plenty good for "rank these 30 titles + properties by relevance"; ~$0.005/query, 8-15s total latency. |
+| LLM client | **`anthropic` SDK preferred, `claude -p` CLI fallback** | SDK gives structured output; CLI works without separate API key for users on Claude Code subscriptions. |
+| Caching | **SHA256(query + sorted_candidate_ids) → JSON file** in `~/.cache/notion-search/` | Cheap, no infra, invalidates naturally when Notion content changes (different candidates → cache miss). 1-day TTL. |
+| Property filter syntax | **JSON object** matching `32-notion-writer`'s `--properties` form | Consistency across the skill suite. |
+| Failure mode | **Permissive degradation** | Rerank fail → return raw expanded-search; expansion fail → original query only. Match the parser philosophy from Phase 3c. |
+
+---
+
+## CLI surface
+
+```bash
+# Default — rerank + expand, terminal output
+notion-search "AI agents in 2026"
+
+# Pipe to JSON for the future push skill
+notion-search "AI agents" --json | jq '.[].id'
+
+# Constrain to specific databases
+notion-search "AI agents" --databases f8f19ede-32bd-43ac-9f60-0651f6f40afe,abc-def-...
+
+# Property filter (same JSON form as notion-writer's --properties)
+notion-search "MCP" --filter '{"Status": "Done", "Topic": "AI"}'
+
+# Fast paths
+notion-search "exact term" --no-rerank   # Notion API only, no LLM
+notion-search "exact term" --no-expand   # Skip variant generation
+notion-search "AI agents" --limit 5      # Default 10
+notion-search "AI agents" --no-cache     # Bypass result cache
+```
+
+**Flags:**
+
+| Flag | Default | Effect |
+|---|---|---|
+| `--json` | off | Emit JSON array instead of terminal table |
+| `--databases <ids>` | all accessible | Comma-separated database/data_source IDs to constrain search |
+| `--filter <json>` | none | Property filter applied per-database (skipped for workspace-wide search) |
+| `--no-rerank` | off | Skip Claude rerank stage |
+| `--no-expand` | off | Skip query-variant generation |
+| `--limit <n>` | 10 | Number of results to return after rerank |
+| `--no-cache` | off | Bypass cache lookup AND don't write cache |
+
+---
+
+## Pipeline
+
+### 1. Query expansion (skipped if `--no-expand`)
+
+Claude Haiku takes the original query and produces 3-5 variants covering:
+- Synonyms ("AI agents" → "multi-agent orchestration", "autonomous LLM agents")
+- Cross-language KR↔EN ("AI agents" → "AI 에이전트")
+- Related concepts ("AI agents" → "agent SDK", "tool use")
+
+The original query is always included. Total variants ≤ 5 to bound API calls.
+
+Prompt approach: ask Claude for a JSON array of variants, parse with `json.loads`. On parse failure, fall back to original query only.
+
+### 2. Search execution
+
+For each variant, hit Notion API:
+- Workspace-wide (`client.search`) when no `--databases` flag
+- Per-database (`client.data_sources.query`) when `--databases` is provided. The existing `_notion_compat.resolve_data_source_id` resolves DB IDs.
+
+Calls are made sequentially (no concurrent dispatch) to stay well under Notion's 3 req/sec average rate limit. With ≤5 variants × ≤2 DBs (typical case), that's at most 10 sequential calls — under 4 seconds total even at the slower end.
+
+Each call returns up to 100 results; we cap candidates at 30 total (after dedup across all variants and DBs) to keep rerank costs predictable.
+
+Dedupe by page ID. Preserve the highest position-rank if the same page appears for multiple variants (used as fallback ordering when rerank is skipped).
+
+### 3. Property + excerpt fetch
+
+For each candidate, gather:
+- Title (from page.properties[title_prop])
+- Key properties (Status, Topic, Type, Account Code, etc. — whatever exists)
+- 200-char excerpt: fetch the first text-bearing block via `client.blocks.children.list` with `page_size=5`, walk the returned blocks and pick the first paragraph/heading/quote whose rich-text concatenates to non-empty content
+
+Excerpt fallback: if none of the first 5 blocks have text content (e.g., page leads with an image, table-only, or empty), excerpt is the empty string. Rerank still runs — title + properties usually carry enough signal.
+
+Properties often arrive inline with `client.search` results; we only re-fetch when properties are stripped (data source queries return full properties; workspace search doesn't).
+
+### 4. Rerank (skipped if `--no-rerank`)
+
+Cache lookup first: SHA256(original_query + sorted_candidate_ids).
+
+Cache miss → call Claude Haiku with:
+- Original (un-expanded) query
+- Numbered list of candidates: `[N] Title — Properties — Excerpt`
+- Asks for a JSON array of objects: `{index, score, why}` ordered by score descending
+
+Parse, map back to candidate page objects, take top `--limit`.
+
+Failure modes:
+- LLM call error → return raw expanded-search results in candidate order, warn on stderr
+- JSON parse error → same as above
+- Rerank returns fewer than requested → take what's there, warn
+
+### 5. Output
+
+**Terminal** (default): formatted table with title, relevance, snippet, key properties, URL. ANSI color via `rich` library if available, plain text otherwise.
+
+**JSON** (`--json`): array of objects with stable schema:
+```json
+[
+  {
+    "id": "abc-def-...",                    // dashed UUID
+    "url": "https://notion.so/...",
+    "title": "MCP server architecture...",
+    "relevance": 0.94,                       // 0.0-1.0; null if --no-rerank
+    "snippet": "Direct match — covers...",   // null if --no-rerank
+    "excerpt": "First paragraph text...",    // 200-char from page body
+    "properties": {                          // only properties present on the page
+      "Status": "Done",
+      "Topic": ["AI", "MCP"],
+      "Account Code": "D.intelligence"
+    }
+  }
+]
+```
+
+The schema is stable: any future tool consuming search output (notion-to-notebooklm push, aggregation, etc.) reads this format.
+
+---
+
+## File structure
+
+| File | Purpose | LOC |
+|---|---|---|
+| `custom-skills/31-notion-organizer/code/scripts/notion_search.py` | Main CLI + pipeline | ~400 |
+| `custom-skills/31-notion-organizer/code/scripts/test_notion_search.py` | Unit tests with mocked Claude/Notion | ~250 |
+| `custom-skills/31-notion-organizer/code/scripts/_search_llm.py` | LLM client abstraction (SDK + CLI fallback, prompt construction) | ~150 |
+| `custom-skills/31-notion-organizer/commands/notion-search.md` | Slash command definition | ~50 |
+| `custom-skills/31-notion-organizer/code/CLAUDE.md` | Add usage section | +60 lines |
+| `custom-skills/31-notion-organizer/code/scripts/requirements.txt` | Add `anthropic` (optional, falls back to CLI) | +1 line |
+
+**Total: ~850 LOC + tests + docs.** ~3 days focused work.
+
+The `_notion_compat` helper from Phase 2 is reused (client factory, error explanation, ID resolution). No new compatibility layer needed.
+
+---
+
+## LLM client (`_search_llm.py`)
+
+Abstraction so callers don't care whether SDK or CLI is in use.
+
+```python
+def call_claude(prompt: str, model: str = "claude-haiku-4-5", max_tokens: int = 1000) -> str:
+    """Return Claude's text response. Raises on failure."""
+    if _have_anthropic_sdk() and os.getenv("ANTHROPIC_API_KEY"):
+        return _call_via_sdk(prompt, model, max_tokens)
+    if _have_claude_cli():
+        return _call_via_cli(prompt, model)
+    raise RuntimeError(
+        "No Claude client available. Install `anthropic` and set "
+        "ANTHROPIC_API_KEY, or install Claude Code CLI."
+    )
+```
+
+Two thin implementations behind it:
+- `_call_via_sdk`: standard `anthropic.Anthropic().messages.create(...)`
+- `_call_via_cli`: `subprocess.run(["claude", "-p", prompt, "--model", model], ...)`, capture stdout
+
+Both return the assistant's text content as a single string. Callers (`expand_query`, `rerank_candidates`) parse JSON out of it.
+
+---
+
+## Caching
+
+**Key:** SHA256 hex digest of `f"{query}|{','.join(sorted(candidate_ids))}"`.
+
+**Storage:** `~/.cache/notion-search/<hash>.json`. Each file contains:
+```json
+{
+  "query": "AI agents",
+  "candidate_ids": ["abc...", "def..."],
+  "results": [...],         // the reranked output
+  "cached_at": "2026-04-28T14:23:00Z"
+}
+```
+
+**Lookup:** if file exists and `now - cached_at < TTL` (default 24h), return cached results. Else run rerank and write fresh.
+
+**Invalidation:** TTL-only for simplicity. `--no-cache` bypasses both read and write.
+
+**Why not invalidate on Notion content changes?** That requires tracking last-edited timestamps and computing a content hash — significant complexity for marginal benefit. With a 1-day TTL, stale results are bounded; user can `--no-cache` for fresh runs.
+
+---
+
+## Error handling
+
+| Failure | Behavior |
+|---|---|
+| Notion API: `ObjectNotFound` on `--databases` ID | Error out with friendly message via `_notion_compat.explain_api_error` |
+| Notion API: `Unauthorized` | Same |
+| Query expansion fails (LLM error or JSON parse) | Use original query only, warn on stderr, continue |
+| Rerank fails (LLM error or JSON parse) | Return raw expanded-search results in candidate order, warn on stderr, continue |
+| Cache file corrupted | Delete cache file, treat as miss |
+| Empty Notion results | Print "No matches for '<query>'" and exit 0 (not an error) |
+| LLM client unavailable (no SDK + no CLI) | Error out with setup instructions if `--no-rerank` and `--no-expand` aren't both set; otherwise proceed |
+| `--filter` JSON parse error | Error out before any API calls |
+
+---
+
+## Tests
+
+10 tests covering the public surface, all using mocked Claude and Notion clients (no real API calls in tests):
+
+| Test | Verifies |
+|---|---|
+| `test_expand_query_returns_variants` | LLM mock returns variant list including original; expansion produces 3-5 unique strings |
+| `test_search_unions_and_dedupes` | Two variants returning overlapping pages produce deduped candidate set |
+| `test_rerank_orders_by_relevance` | Mock rerank returns scores; output sorted score-descending |
+| `test_no_rerank_returns_raw` | `--no-rerank` skips LLM, returns Notion's native order, no relevance scores |
+| `test_no_expand_uses_single_query` | `--no-expand` calls Notion once with original query, no LLM expansion |
+| `test_json_output_parseable` | `--json` emits valid JSON conforming to the schema in Section 5 |
+| `test_cache_hit_on_repeat` | Second identical query skips LLM; cache file exists |
+| `test_cache_miss_on_different_candidates` | Different candidate set hashes different → fresh rerank |
+| `test_property_filter_applied` | `--filter` JSON passed to data_sources.query verbatim |
+| `test_database_scope_constrains` | `--databases` causes per-DB queries instead of workspace search |
+
+Tests live in `test_notion_search.py`. Run via `python3 test_notion_search.py` (matching the existing test_parser.py convention in 32).
+
+---
+
+## Out-of-scope follow-ups
+
+- **3b-ii — `notion-to-notebooklm` push skill**: consumes search-output JSON, pushes pages as NotebookLM sources. Separate spec after this ships.
+- **Block-level granularity**: surface matching blocks within pages. Adds chunking logic and richer rerank prompts. Defer.
+- **Aggregation skill**: "find pages on X, write a summary page with citations." Builds on this. Future Phase 3 sub-project.
+- **Multi-workspace search**: requires per-workspace integrations and credential management. Future.
+
+---
+
+## Implementation transition
+
+After user approval, transition to `superpowers:writing-plans` to break this into bite-sized TDD tasks. Likely shape:
+
+1. `_search_llm.py` skeleton (SDK + CLI client abstraction) + unit test
+2. Query expansion module + tests
+3. Search execution module (workspace + per-DB) + tests
+4. Property + excerpt fetch + tests
+5. Rerank module (with prompt + JSON parsing) + tests
+6. Cache layer + tests
+7. CLI assembly (argparse + output formatting) + tests
+8. Slash command + 31's CLAUDE.md update + final integration smoke test
+
+Each step independently testable. Total ~3 days estimated, mirrors Phase 3c's pacing.