our-claude-skills/docs/superpowers/plans/2026-04-28-notion-semantic-search.md

Notion Semantic Search Implementation Plan

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: Build a CLI skill that searches Notion workspaces semantically, using LLM-based query expansion + rerank to close the gap left by Notion's keyword-only native search. Output is a terminal table for browsing or JSON for piping into downstream tools.

Architecture: Eight independent modules composed by a thin CLI entrypoint. Query expansion and rerank both call Claude Haiku via a small _search_llm.py abstraction (SDK preferred, claude -p CLI fallback). Notion search uses the existing _notion_compat.py resolver and client factory shipped in Phase 2. A SHA256-keyed JSON-file cache short-circuits repeat reranks. All LLM and Notion I/O is dependency-injected so unit tests run without real API keys.

Tech Stack: Python 3.11+, notion-client v3 SDK (already in 32-notion-writer venv), optional anthropic SDK (Haiku 4.5 — claude-haiku-4-5), unittest.mock for test doubles. Tests via python3 test_notion_search.py (flat-function _assert style matching 32-notion-writer/code/scripts/test_parser.py).

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

---

File Structure

File	Purpose	Approx LOC
custom-skills/31-notion-organizer/code/scripts/_search_llm.py	Claude client abstraction (SDK + CLI fallback)	~120
custom-skills/31-notion-organizer/code/scripts/_search_cache.py	SHA256-keyed JSON file cache	~80
custom-skills/31-notion-organizer/code/scripts/notion_search.py	Main pipeline: expand, search, enrich, rerank, output, CLI	~450
custom-skills/31-notion-organizer/code/scripts/test_notion_search.py	Unit tests, all I/O mocked	~300
custom-skills/31-notion-organizer/code/scripts/_notion_compat.py	Symlink to 32's helper (already exists in 32)	(link only)
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)	+1 line

The plan keeps notion_search.py as one focused file (~450 LOC) rather than splitting into expand.py/search.py/rerank.py/output.py. Reason: each module is small (~80 LOC), they share helpers, and the existing 31 codebase uses single-file scripts (async_organizer.py, schema_migrator.py). Splitting would over-engineer for the LOC budget.

---

Test Conventions

The 31-notion-organizer codebase has no existing test file, but 32-notion-writer's test_parser.py sets the repo convention. Follow it:

• Flat function tests: def test_X(): ...
• _assert(cond, msg) helper at the top — prints ✓ or ✗ and exits on failure
• run_all() lists tests by name and calls each
• if __name__ == "__main__": run_all()
• Run via python3 test_notion_search.py

For test doubles:

• LLM: dependency-injected via llm_caller= parameter — tests pass lambda prompt, **kw: '["fake response"]'
• Notion client: unittest.mock.MagicMock constructed locally per test
• Cache: pass cache_dir= parameter — tests use tempfile.TemporaryDirectory

---

Task 1: Create test scaffolding + _search_llm.py (LLM client abstraction)

Files:

• Create: custom-skills/31-notion-organizer/code/scripts/_search_llm.py
• Create: custom-skills/31-notion-organizer/code/scripts/test_notion_search.py
• Create: custom-skills/31-notion-organizer/code/scripts/_notion_compat.py (symlink)

The LLM abstraction has three responsibilities: detect available client (SDK or CLI), dispatch the call, surface a clear error if neither is available. Tests cover dispatch logic via dependency injection — no real API calls.

• Step 1: Create the symlink to _notion_compat.py

The compat helper already exists in 32-notion-writer. Symlink it into 31 so both skills share a single source of truth.

cd /Users/ourdigital/Project/our-claude-skills/custom-skills/31-notion-organizer/code/scripts
ln -sf ../../../32-notion-writer/code/scripts/_notion_compat.py _notion_compat.py
ls -la _notion_compat.py

Expected: _notion_compat.py -> ../../../32-notion-writer/code/scripts/_notion_compat.py

• Step 2: Create the test scaffolding (test_notion_search.py)

#!/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 run_all():
    tests = [
        # tests added per task
    ]
    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()

• Step 3: Write the failing tests for _search_llm.py

Add to test_notion_search.py BEFORE run_all:

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")

Add the three tests to tests list in run_all():

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,
    ]
    # ... rest unchanged

• Step 4: Run tests to verify they fail

cd /Users/ourdigital/Project/our-claude-skills/custom-skills/31-notion-organizer/code/scripts
python3 test_notion_search.py 2>&1 | head -10

Expected: ImportError or ModuleNotFoundError on _search_llm.

• Step 5: Create _search_llm.py

"""Claude client abstraction. SDK preferred, `claude -p` CLI fallback."""

from __future__ import annotations

import os
import shutil
import subprocess
from typing import Callable

LLMCaller = Callable[..., str]  # (prompt, *, model, max_tokens) -> response text


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 = "claude-haiku-4-5",
    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.
    """
    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)"
    )

• Step 6: Run tests to verify they pass

python3 test_notion_search.py 2>&1 | tail -5

Expected: ✅ All 3 tests passed.

• Step 7: Commit

cd /Users/ourdigital/Project/our-claude-skills
git add custom-skills/31-notion-organizer/code/scripts/_search_llm.py \
        custom-skills/31-notion-organizer/code/scripts/test_notion_search.py \
        custom-skills/31-notion-organizer/code/scripts/_notion_compat.py
git commit -m "$(cat <<'EOF'
feat(notion-search): add LLM client abstraction with SDK + CLI fallback

First task of Phase 3b-i (notion semantic search). Adds _search_llm.py
that dispatches to the anthropic SDK when ANTHROPIC_API_KEY is set,
falls back to `claude -p` CLI otherwise, and raises a clear setup
error if neither works. Symlinks _notion_compat.py from 32-notion-writer
so both skills share one source of truth.

3 tests cover the three dispatch paths.

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

---

Task 2: Cache layer (_search_cache.py)

Files:

• Create: custom-skills/31-notion-organizer/code/scripts/_search_cache.py
• Modify: custom-skills/31-notion-organizer/code/scripts/test_notion_search.py

SHA256(query + sorted_candidate_ids) → JSON file. Filesystem-only, no DB. Tests use tempfile.TemporaryDirectory for isolation.

• Step 1: Write the failing tests

Add to test_notion_search.py before run_all:

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")

Add to tests list:

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,
    ]
    # ... rest unchanged

• Step 2: Run tests to verify they fail

cd /Users/ourdigital/Project/our-claude-skills/custom-skills/31-notion-organizer/code/scripts
python3 test_notion_search.py 2>&1 | grep -A 1 "test_cache" | head -10

Expected: ImportError on _search_cache.

• Step 3: Create _search_cache.py

"""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 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]:
    """Return cached results if file exists and is fresh, else None.

    Corrupted cache files are silently treated as misses (and removed).
    """
    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,
    *,
    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")

• Step 4: Run tests to verify they pass

python3 test_notion_search.py 2>&1 | tail -5

Expected: ✅ All 8 tests passed.

• Step 5: Commit

cd /Users/ourdigital/Project/our-claude-skills
git add custom-skills/31-notion-organizer/code/scripts/_search_cache.py \
        custom-skills/31-notion-organizer/code/scripts/test_notion_search.py
git commit -m "$(cat <<'EOF'
feat(notion-search): add SHA256-keyed JSON file cache

Cache layer for rerank results. Key is SHA256 of query + sorted
candidate IDs, so changing the candidate set automatically invalidates.
1-day TTL by default; corrupted cache files are silently dropped.

5 tests cover hit, miss, different-candidates, TTL expiry, corruption.

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

---

Task 3: Query expansion module

Files:

• Create (initial scaffold): custom-skills/31-notion-organizer/code/scripts/notion_search.py
• Modify: custom-skills/31-notion-organizer/code/scripts/test_notion_search.py

This task starts notion_search.py with one function (expand_query). Subsequent tasks add the rest.

• Step 1: Write the failing tests

Add to test_notion_search.py before run_all:

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")

Add to tests list:

        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,

• Step 2: Run tests to verify they fail

cd /Users/ourdigital/Project/our-claude-skills/custom-skills/31-notion-organizer/code/scripts
python3 test_notion_search.py 2>&1 | grep "test_expand" | head -5

Expected: ImportError on notion_search.

• Step 3: Create notion_search.py with expand_query

#!/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 json
import re
import sys
from typing import Callable, List

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: 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(f"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(f"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):
        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

• Step 4: Run tests to verify they pass

python3 test_notion_search.py 2>&1 | tail -5

Expected: ✅ All 12 tests passed.

• Step 5: Commit

cd /Users/ourdigital/Project/our-claude-skills
git add custom-skills/31-notion-organizer/code/scripts/notion_search.py \
        custom-skills/31-notion-organizer/code/scripts/test_notion_search.py
git commit -m "$(cat <<'EOF'
feat(notion-search): add query expansion via Claude Haiku

Generates up to 5 query variants (synonyms + cross-language KR↔EN) so
later Notion API search can union over them. Permissive failure modes:
LLM error or non-JSON response falls back to [original] with stderr
warning. Dedupes and caps variants.

4 tests: variants list, dedup+cap, JSON-parse fallback, exception
fallback.

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

---

Task 4: Search execution (Notion API)

Files:

• Modify: custom-skills/31-notion-organizer/code/scripts/notion_search.py
• Modify: custom-skills/31-notion-organizer/code/scripts/test_notion_search.py

For each query variant, hit Notion API. Workspace-wide via client.search when no --databases flag, per-database via client.data_sources.query otherwise. Union and dedupe results by page ID, cap at 30.

• Step 1: Write the failing tests

Add to test_notion_search.py before run_all:

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()
    # Variant 1 returns pages A, B
    # Variant 2 returns pages B, C
    # Final union should be A, B, C
    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()
    # Mock the resolver and the per-DB query
    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):  # passthrough
        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")

Add to tests list:

        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,

• Step 2: Run tests to verify they fail

python3 test_notion_search.py 2>&1 | grep "test_search" | head -5

Expected: AttributeError on notion_search.search_candidates.

• Step 3: Add search_candidates to notion_search.py

Add these imports at the top of notion_search.py, after the existing imports:

from typing import Callable, Dict, List, Optional

import _notion_compat as compat

Add the function (place after expand_query):

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

                response = notion.data_sources.query(**params)
                for page in response.get("results", []):
                    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
            response = notion.search(query=query, page_size=100)
            for item in response.get("results", []):
                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

• Step 4: Run tests to verify they pass

python3 test_notion_search.py 2>&1 | tail -5

Expected: ✅ All 17 tests passed.

• Step 5: Commit

cd /Users/ourdigital/Project/our-claude-skills
git add custom-skills/31-notion-organizer/code/scripts/notion_search.py \
        custom-skills/31-notion-organizer/code/scripts/test_notion_search.py
git commit -m "$(cat <<'EOF'
feat(notion-search): add Notion search execution with workspace/DB modes

For each expanded query variant: hit Notion's workspace search OR
per-database data_sources.query (when --databases is specified).
Union and dedupe by page ID, cap at 30 candidates total. Filters out
non-page objects (databases) from workspace search results.
Property filters (--filter JSON) pass through to data_sources.query
when in per-database mode.

5 tests: workspace dedup, 30-cap, DB-mode dispatch, page-only filter,
property-filter passthrough.

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

---

Task 5: Property + excerpt enrichment

Files:

• Modify: custom-skills/31-notion-organizer/code/scripts/notion_search.py
• Modify: custom-skills/31-notion-organizer/code/scripts/test_notion_search.py

For each candidate: extract title from properties, gather key properties, fetch first text-bearing block (paragraph/heading/quote) as a 200-char excerpt. Empty excerpt is acceptable.

• Step 1: Write the failing tests

Add to test_notion_search.py:

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")

Add to tests list:

        test_enrich_extracts_title_and_properties,
        test_enrich_extracts_first_paragraph_excerpt,
        test_enrich_falls_back_to_empty_excerpt,
        test_enrich_truncates_long_excerpt,

• Step 2: Run tests to verify they fail

python3 test_notion_search.py 2>&1 | grep "test_enrich" | head -5

Expected: AttributeError on notion_search.enrich_candidates.

• Step 3: Add enrich_candidates to notion_search.py

Add the function (place after search_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(name: str, 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", []):
        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(name, 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

• Step 4: Run tests to verify they pass

python3 test_notion_search.py 2>&1 | tail -5

Expected: ✅ All 21 tests passed.

• Step 5: Commit

cd /Users/ourdigital/Project/our-claude-skills
git add custom-skills/31-notion-organizer/code/scripts/notion_search.py \
        custom-skills/31-notion-organizer/code/scripts/test_notion_search.py
git commit -m "$(cat <<'EOF'
feat(notion-search): add candidate enrichment (title, properties, excerpt)

For each candidate page, extract the title, flatten common property
types (status/select/multi_select/date/checkbox/number/url/etc.) to
display values, and fetch the first text-bearing block as a 200-char
excerpt. Empty excerpt is acceptable when the page has no leading text.

4 tests: title+properties, paragraph excerpt, empty fallback,
truncation.

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

---

Task 6: Rerank module

Files:

• Modify: custom-skills/31-notion-organizer/code/scripts/notion_search.py
• Modify: custom-skills/31-notion-organizer/code/scripts/test_notion_search.py

Build the rerank prompt, call Claude, parse JSON, sort by score, take top N. Permissive on parse errors.

• Step 1: Write the failing tests

Add to test_notion_search.py:

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")

Add to tests list:

        test_rerank_orders_by_score_descending,
        test_rerank_respects_limit,
        test_rerank_falls_back_on_parse_error,
        test_rerank_falls_back_on_llm_exception,

• Step 2: Run tests to verify they fail

python3 test_notion_search.py 2>&1 | grep "test_rerank" | head -5

Expected: AttributeError on notion_search.rerank.

• Step 3: Add rerank to notion_search.py

Add (place after enrich_candidates):

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: 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(f"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(f"Warning: rerank returned invalid JSON; returning unranked results",
              file=sys.stderr)
        return _fallback_rank(candidates, limit)

    if not isinstance(scored, list):
        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

• Step 4: Run tests to verify they pass

python3 test_notion_search.py 2>&1 | tail -5

Expected: ✅ All 25 tests passed.

• Step 5: Commit

cd /Users/ourdigital/Project/our-claude-skills
git add custom-skills/31-notion-organizer/code/scripts/notion_search.py \
        custom-skills/31-notion-organizer/code/scripts/test_notion_search.py
git commit -m "$(cat <<'EOF'
feat(notion-search): add Claude Haiku rerank module

Builds a rerank prompt with title + flattened properties + excerpt for
each candidate, calls Claude, parses JSON, sorts by score descending,
takes top N. On any failure (LLM error, missing JSON, parse error),
falls back to candidates in input order with null relevance/snippet.

4 tests: ordering, limit, parse-error fallback, exception fallback.

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

---

Task 7: CLI assembly + output formatting

Files:

• Modify: custom-skills/31-notion-organizer/code/scripts/notion_search.py
• Modify: custom-skills/31-notion-organizer/code/scripts/test_notion_search.py

Tie the modules together. argparse, cache wiring, terminal table or JSON output. Tests use mocked notion + LLM.

• Step 1: Write the failing tests

Add to test_notion_search.py:

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)")

Add to tests list:

        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,

• Step 2: Run tests to verify they fail

python3 test_notion_search.py 2>&1 | grep "test_pipeline" | head -5

Expected: AttributeError on notion_search.run_search.

• Step 3: Add run_search, output formatters, and main to notion_search.py

Add at the top (after existing imports):

import argparse
import os
from pathlib import Path

import _search_cache

Add (place after rerank):

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: Callable[..., str] = None,
    rerank_llm: Callable[..., str] = None,
) -> List[Dict]:
    """Full pipeline: expand → search → enrich → rerank → return.

    expand_llm and rerank_llm are dependency-injected for tests.
    """
    # 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]
    if use_cache:
        kwargs = {"cache_dir": cache_dir} if cache_dir else {}
        cached = _search_cache.cache_get(query, candidate_ids, **kwargs)
        if cached is not None:
            return cached

    ranked = rerank(query, enriched, llm_caller=rerank_llm, limit=limit)

    if use_cache:
        kwargs = {"cache_dir": cache_dir} if cache_dir else {}
        _search_cache.cache_put(query, candidate_ids, ranked, **kwargs)

    return ranked


def format_terminal(results: List[Dict]) -> str:
    """Human-readable terminal output."""
    if not results:
        return "No matches."
    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

    # 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()

• Step 4: Run tests to verify they pass

python3 test_notion_search.py 2>&1 | tail -5

Expected: ✅ All 29 tests passed.

• Step 5: Verify CLI help works

python3 notion_search.py --help 2>&1 | head -20

Expected: argparse help text listing all flags.

• Step 6: Commit

cd /Users/ourdigital/Project/our-claude-skills
git add custom-skills/31-notion-organizer/code/scripts/notion_search.py \
        custom-skills/31-notion-organizer/code/scripts/test_notion_search.py
git commit -m "$(cat <<'EOF'
feat(notion-search): add CLI entrypoint with argparse + output formatting

Wires together the four stages (expand → search → enrich → rerank) into
run_search(). CLI flags: --databases, --filter, --limit, --no-rerank,
--no-expand, --no-cache, --json. Terminal output renders as a numbered
table with title, relevance, properties, snippet, URL.

Cache lookup happens BEFORE rerank, with cache_put after success.
NOTION_API_KEY (or NOTION_TOKEN) env var required.

4 end-to-end pipeline tests (mocked Notion + LLM): JSON-serializable
output, --no-rerank skip, --no-expand skip, cache hit on repeat.

Total: 29 tests passing.

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

---

Task 8: Slash command + docs

Files:

• Create: custom-skills/31-notion-organizer/commands/notion-search.md
• Modify: custom-skills/31-notion-organizer/code/CLAUDE.md
• Modify: custom-skills/31-notion-organizer/code/scripts/requirements.txt

Wire up the slash command, document the new tool in CLAUDE.md, list anthropic as an optional dependency.

• Step 1: Create the slash command file

Create directory if it doesn't exist, then write the file:

mkdir -p /Users/ourdigital/Project/our-claude-skills/custom-skills/31-notion-organizer/commands

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

---
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-search "MCP" --databases ID --filter '{"Status": "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)

[
  {
    "id": "abc-def-...",
    "url": "https://notion.so/...",
    "title": "Page title",
    "relevance": 0.94,           // null when --no-rerank
    "snippet": "Why it matched", // null when --no-rerank
    "excerpt": "First paragraph text...",
    "properties": {"Status": "Done", "Topic": ["AI", "MCP"]}
  }
]

This schema is the contract for downstream tools (e.g., the future notion-to-notebooklm push skill).

- [ ] **Step 2: Add Notion search section to CLAUDE.md**

Read the current `custom-skills/31-notion-organizer/code/CLAUDE.md` and find the "Quick Start" section. Insert the following section AFTER "Quick Start" and BEFORE "Scripts":

```markdown
## 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
python scripts/notion_search.py "MCP" --databases ID \
  --filter '{"Status": "Done", "Topic": "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.

- [ ] **Step 3: Add `anthropic` to requirements.txt**

Read the current `custom-skills/31-notion-organizer/code/scripts/requirements.txt` and append:

Optional: required only for direct Anthropic SDK use.

If missing, the search skill falls back to claude -p CLI.

anthropic>=0.40.0

- [ ] **Step 4: Run the full test suite as a final sanity check**

```bash
cd /Users/ourdigital/Project/our-claude-skills/custom-skills/31-notion-organizer/code/scripts
python3 test_notion_search.py 2>&1 | tail -5

Expected: ✅ All 29 tests passed.

• Step 5: Verify the slash command file is well-formed

python3 -c "
content = open('/Users/ourdigital/Project/our-claude-skills/custom-skills/31-notion-organizer/commands/notion-search.md').read()
assert 'description:' in content, 'frontmatter description present'
assert 'python3 notion_search.py' in content, 'script invocation present'
assert 'NOTION_API_KEY' in content, 'env requirement documented'
print('Slash command file looks valid')
"

Expected: Slash command file looks valid.

• Step 6: Commit

cd /Users/ourdigital/Project/our-claude-skills
git add custom-skills/31-notion-organizer/commands/notion-search.md \
        custom-skills/31-notion-organizer/code/CLAUDE.md \
        custom-skills/31-notion-organizer/code/scripts/requirements.txt
git commit -m "$(cat <<'EOF'
docs(notion-search): add /notion-search slash command + CLAUDE.md section

Slash command at custom-skills/31-notion-organizer/commands/notion-search.md
documents the CLI surface and JSON output schema. CLAUDE.md gains a
Semantic Search section explaining the 4-stage pipeline and env var
requirements. requirements.txt notes the optional anthropic SDK
dependency (the skill falls back to the claude CLI if missing).

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

---

Final verification

After all eight tasks:

• Run the full test suite: 29 tests passing
• Sanity-check imports:

python3 -c "
import sys
sys.path.insert(0, '/Users/ourdigital/Project/our-claude-skills/custom-skills/31-notion-organizer/code/scripts')
from notion_search import expand_query, search_candidates, enrich_candidates, rerank, run_search, format_terminal, main
print('All public functions importable')
"

• Confirm git log shows 8 new commits since 08e3dd7 (the spec commit), one per task

---

Out-of-scope follow-ups

• Phase 3b-ii (notion-to-notebooklm push skill): consumes the JSON schema produced by --json. Separate brainstorm + plan when this ships.
• Block-level result granularity: surface matching block snippets within pages instead of (or in addition to) page-level results. Defer.
• Cross-workspace search: requires per-workspace integrations and credential management. Future.
• Embedding-based search: vector store + sync pipeline. Belongs in a separate tool, not a skill.
• Live API smoke test against the AI database: not in tests (mocked only). The user can run the real CLI against their workspace once credentials are in .env.