Add project-driven tiered domain discovery (discover)

Derive the harvest taxonomy from the user's own repos instead of a hand-written
topics.yaml. `clawlibrary discover` scans git repos under a root, has local
gemma4 analyze each (README+manifests+tree+source headers → purpose/tech/
features/architecture/domain_keywords), then synthesizes a 3-tier taxonomy
(general→related→niche) and derives the harvest queues.

- clawlibrary/discover.py: find_repos, gather_repo_context (pruned, budgeted),
  analyze_repo (gemma4 json_mode, mirrors summarize.py), sidecars, optional
  embed_analyses (projects become searchable; excluded from the paper pipeline
  via extract.iter_documents skip of projects/).
- clawlibrary/taxonomy.py: Domain/Taxonomy models, synthesize() with TaxonomyError
  guard, to_topics (tier→priority/max_results, arXiv-category validation +
  general fallback, OA-books pass via work_types), to_youtube_queue (ytsearchN:
  reusing existing ingestion), yaml writers.
- config.py: Topic.work_types + write_topics (round-trips load_topics).
  openalex._build_filters threads work_types (type:book|book-chapter); default
  type:article unchanged.
- CLI: `discover` command + `--use-generated` sugar on run/refresh.

22 new tests (112 passed, 1 skipped); ruff clean. README documents the workflow.

Co-Authored-By: Claude Opus 4.8 <[email protected]>

README.md

@@ -124,6 +124,35 @@ clawlibrary ask "What did the talks say about agentic planning?"
 (only), `whisper` (force). On a GPU node, `--whisper-device auto` picks CUDA;
 it falls back to CPU/int8 when no GPU is present.
 
+## Project-driven domains (`discover`)
+
+Instead of hand-writing `topics.yaml`, derive the research domains from your own
+code. `clawlibrary discover` scans the git repos under a root (default
+`~/projects`), has local **gemma4** read each one (README + manifests + pruned
+file tree + source headers) to infer `{purpose, technologies, features,
+architecture, domain_keywords}`, then synthesizes a **tiered taxonomy**
+(general → related → niche) and derives the harvest queues:
+
+```bash
+clawlibrary discover --root ~/projects            # analyze repos → artifacts
+#   writes:  domains.yaml              (the tiered taxonomy — REVIEW this)
+#            topics.generated.yaml     (papers + OA-books queries)
+#            youtube.generated.yaml    (ytsearch video queries)
+#            vault/projects/*.json     (per-project analyses, embedded for search)
+
+# review/edit domains.yaml, then harvest the generated queues:
+clawlibrary refresh --use-generated               # papers + books, all tiers
+clawlibrary youtube --queue youtube.generated.yaml
+```
+
+Tiers map to harvest depth: **general** (priority 1, broad queries, full arXiv
+categories) is harvested first, **niche** (priority 9, laser queries) is drilled
+deepest — so you capture resources at every level on the way down. A parallel
+**books pass** per domain pulls OA book/book-chapter works (OpenAlex
+`type:book`). The flow is review-then-apply: `discover` only writes artifacts;
+harvesting is a separate, explicit step. Project analyses are embedded into the
+vault too, so agents can ask "which of my projects relate to X".
+
 ## Scheduled refresh (keeping the vault current)
 
 `clawlibrary refresh` re-polls every tracked domain and folds only the **new**

clawlibrary/__main__.py

@@ -13,6 +13,7 @@ Commands:
   catalog [--topics|--recent N] [--json]   vault catalog slices
   youtube [--queue F] [--url U --topic T]  ingest YouTube videos as transcripts
   refresh [--source both] [--no-whisper]   incremental: harvest+ingest+process new
+  discover [--root D] [--max-repos N]      scan repos → tiered domain taxonomy
 """
 
 from __future__ import annotations
@@ -58,6 +59,8 @@ def _print_summary(stats: RunStats, *, dry_run: bool) -> None:
 
 
 def cmd_run(args: argparse.Namespace) -> int:
+    if getattr(args, "use_generated", False):
+        args.topics = "topics.generated.yaml"
     config, topics = _load(args.config, args.topics)
     if args.topic:
         topics = [t for t in topics if t.query == args.topic]
@@ -527,6 +530,9 @@ def cmd_refresh(args: argparse.Namespace) -> int:
     from .lock import LockError
     from .refresh import refresh
 
+    if args.use_generated:
+        args.topics = "topics.generated.yaml"
+        args.queue = "youtube.generated.yaml"
     config, topics = _load(args.config, args.topics)
 
     def on_event(kind: str, **data):
@@ -556,6 +562,63 @@ def cmd_refresh(args: argparse.Namespace) -> int:
     return 1 if stats.had_error else 0
 
 
+def cmd_discover(args: argparse.Namespace) -> int:
+    from . import discover as D
+    from . import taxonomy as TX
+    from .config import load_config
+    from .ollama_client import OllamaClient
+
+    config = load_config(args.config)  # config only — discover doesn't need topics.yaml
+    root = Path(config.vault.root)
+    oc = OllamaClient(text_model=args.model)
+    if not oc.is_up():
+        console.print("[red]Ollama not reachable.[/red]")
+        return 2
+
+    repos = D.find_repos(args.root)
+    if not repos:
+        console.print(f"[yellow]No git repositories found under {args.root}[/yellow]")
+        return 1
+    if args.max_repos:
+        repos = repos[: args.max_repos]
+    console.print(f"[bold]Analyzing {len(repos)} repositories under {args.root}…[/bold]")
+
+    analyses = []
+    for repo in repos:
+        a = D.analyze_repo(oc, D.gather_repo_context(repo))
+        D.write_repo_sidecar(root, a)
+        analyses.append(a)
+        tag = ("[red]parse_error[/red]" if a.get("parse_error")
+               else ", ".join(a.get("domain_keywords", [])[:4]))
+        console.print(f"  [green]✓[/green] {repo.name}  [dim]{tag}[/dim]")
+
+    try:
+        taxonomy = TX.synthesize(oc, analyses)
+    except TX.TaxonomyError as e:
+        console.print(f"[red]Taxonomy synthesis failed:[/red] {e}")
+        return 1
+
+    TX.write_domains_yaml(taxonomy, args.domains)
+    TX.write_topics_yaml(TX.to_topics(taxonomy), args.out_topics)
+    TX.write_youtube_yaml(TX.to_youtube_queue(taxonomy), args.out_youtube)
+    if not args.no_embed:
+        n = D.embed_analyses(oc, root, analyses)
+        console.print(f"  [dim]embedded {n} project analyses into the vault[/dim]")
+
+    table = Table(title="Discovered taxonomy", header_style="bold")
+    table.add_column("Tier")
+    table.add_column("Domains", justify="right")
+    table.add_column("Queries", justify="right")
+    for tier in TX.TIERS:
+        domains = getattr(taxonomy, tier)
+        table.add_row(tier, str(len(domains)), str(sum(len(d.queries) for d in domains)))
+    console.print(table)
+    console.print(f"[bold]Review[/bold] {args.domains}, then: "
+                  f"[cyan]clawlibrary refresh --use-generated[/cyan]")
+    console.print(f"[dim]wrote {args.out_topics} and {args.out_youtube}[/dim]")
+    return 0
+
+
 def build_parser() -> argparse.ArgumentParser:
     p = argparse.ArgumentParser(prog="clawlibrary", description=__doc__)
     p.add_argument("--config", default="config.yaml", help="path to config.yaml")
@@ -568,6 +631,8 @@ def build_parser() -> argparse.ArgumentParser:
     pr.add_argument("--max", type=int, default=None, help="override max downloads this run")
     pr.add_argument("--source", choices=["openalex", "arxiv", "both"], default="openalex",
                     help="discovery source (arxiv = polite arXiv API harvest)")
+    pr.add_argument("--use-generated", action="store_true",
+                    help="harvest topics.generated.yaml (from `discover`)")
     pr.set_defaults(func=cmd_run)
 
     ps = sub.add_parser("stats", help="show vault statistics")
@@ -647,7 +712,22 @@ def build_parser() -> argparse.ArgumentParser:
                      help="captions-only YouTube (no GPU whisper) for cheap scheduled runs")
     prf.add_argument("--model", default="gemma4:E4B", help="Ollama model for summaries")
     prf.add_argument("--dry-run", action="store_true", help="report would-be additions; no writes")
+    prf.add_argument("--use-generated", action="store_true",
+                     help="use topics.generated.yaml + youtube.generated.yaml (from `discover`)")
     prf.set_defaults(func=cmd_refresh)
+
+    pdisc = sub.add_parser("discover",
+                           help="scan code repos + synthesize a tiered domain taxonomy")
+    pdisc.add_argument("--root", default=str(Path.home() / "projects"),
+                       help="directory of project repos to scan")
+    pdisc.add_argument("--model", default="gemma4:E4B", help="Ollama model for analysis")
+    pdisc.add_argument("--max-repos", type=int, default=None, help="cap repos analyzed")
+    pdisc.add_argument("--no-embed", action="store_true",
+                       help="do not embed project analyses into the vault")
+    pdisc.add_argument("--domains", default="domains.yaml", help="taxonomy output (review this)")
+    pdisc.add_argument("--out-topics", default="topics.generated.yaml")
+    pdisc.add_argument("--out-youtube", default="youtube.generated.yaml")
+    pdisc.set_defaults(func=cmd_discover)
     return p
 
 

clawlibrary/config.py

@@ -34,6 +34,9 @@ class Topic(BaseModel):
     # Optional arXiv category restriction for `run --source arxiv`
     # (e.g. ["quant-ph"]); defaults to a broad AI+quantum set if unset.
     arxiv_categories: list[str] | None = None
+    # OpenAlex work types to harvest (None => "article"). A books pass sets
+    # e.g. ["book", "book-chapter"]; see sources/openalex._build_filters.
+    work_types: list[str] | None = None
 
     @field_validator("query")
     @classmethod
@@ -167,3 +170,27 @@ def load_topics(path: str | Path = "topics.yaml") -> TopicQueue:
         merged.append(topic)
 
     return TopicQueue(defaults=defaults, topics=[Topic.model_validate(t) for t in merged])
+
+
+def _drop_none(d: dict) -> dict:
+    """Recursively drop keys whose value is None (keeps generated YAML clean)."""
+    out: dict = {}
+    for k, v in d.items():
+        if v is None:
+            continue
+        out[k] = _drop_none(v) if isinstance(v, dict) else v
+    return out
+
+
+def write_topics(topics: list[Topic], path: str | Path, *, defaults: dict | None = None) -> None:
+    """Serialize topics to a YAML file that round-trips through load_topics().
+
+    Each topic is dumped in full (no reliance on a `defaults` block) so the file
+    reloads identically even without defaults. Used by the taxonomy generator to
+    emit topics.generated.yaml.
+    """
+    data: dict = {}
+    if defaults:
+        data["defaults"] = defaults
+    data["topics"] = [_drop_none(t.model_dump(mode="json")) for t in topics]
+    Path(path).write_text(yaml.safe_dump(data, sort_keys=False, allow_unicode=True))

clawlibrary/discover.py

@@ -0,0 +1,270 @@
+"""Scan code repositories and analyze each with a local LLM (gemma4:E4B).
+
+Given a root directory (e.g. ~/projects), enumerate the git project dirs and,
+for each, gather a compact signal — README, manifests, a pruned file tree, and a
+few source-file headers — then ask gemma4 (JSON mode) to infer what the project
+is, its technologies, features, architecture, and the research domains it
+implies. Results are persisted as `vault/projects/{slug}.json` sidecars and feed
+the taxonomy synthesizer.
+
+Mirrors `summarize.py`: a prompt constant, char budgets to keep analysis fast,
+json_mode generation, and a `{"raw":..., "parse_error":True}` fallback.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+from pathlib import Path
+
+from .ollama_client import OllamaClient
+from .vault import slugify
+
+README_NAMES = ("README.md", "README.rst", "README.txt", "README")
+MANIFESTS = (
+    "pyproject.toml", "package.json", "Cargo.toml", "go.mod",
+    "requirements.txt", "setup.py", "pom.xml", "build.gradle", "Gemfile",
+)
+SKIP_DIRS = {
+    ".git", "node_modules", ".venv", "venv", "__pycache__", "dist", "build",
+    "target", ".next", ".mypy_cache", ".ruff_cache", ".pytest_cache", ".idea",
+}
+SOURCE_EXTS = (".py", ".rs", ".go", ".ts", ".tsx", ".js", ".java", ".c", ".cpp", ".rb")
+
+MAX_README_CHARS = 2000
+MAX_MANIFEST_CHARS = 1200
+MAX_TREE_CHARS = 1500
+MAX_TREE_ENTRIES = 200
+MAX_SOURCE_HEADERS = 6
+MAX_SOURCE_HEAD_CHARS = 400
+
+_PROMPT = """You are a software analyst cataloging a code repository to infer the \
+research domains its author works in.
+
+Respond with ONLY a JSON object with these keys:
+- "name": short project name (string)
+- "purpose": one to two sentences on what the project does
+- "technologies": array of 3-10 frameworks/languages/libraries it uses
+- "features": array of 2-6 short capability strings
+- "architecture": one sentence on the structural approach (or "")
+- "domain_keywords": array of 4-10 research/topic keywords this project implies \
+(e.g. "vector search", "quantum error correction", "LLM agents")
+
+Repository: {name}
+
+README (truncated):
+{readme}
+
+Manifests:
+{manifests}
+
+File tree (pruned):
+{tree}
+
+Key source headers:
+{source_headers}
+"""
+
+
+def find_repos(root: str | Path) -> list[Path]:
+    """Top-level directories under `root` that are git repos (contain .git).
+
+    Does not recurse into a repo, and skips a repo's nested submodules.
+    """
+    root = Path(root).expanduser()
+    if not root.is_dir():
+        return []
+    return [
+        child for child in sorted(root.iterdir())
+        if child.is_dir() and (child / ".git").exists()
+    ]
+
+
+def _read_clip(path: Path, n: int) -> str:
+    try:
+        return path.read_text(encoding="utf-8", errors="ignore")[:n]
+    except OSError:
+        return ""
+
+
+def _first_readme(repo: Path) -> str:
+    for name in README_NAMES:
+        p = repo / name
+        if p.is_file():
+            return _read_clip(p, MAX_README_CHARS)
+    return ""
+
+
+def _walk_tree(repo: Path) -> tuple[list[str], list[Path]]:
+    entries: list[str] = []
+    sources: list[Path] = []
+    for dirpath, dirnames, filenames in os.walk(repo):
+        dirnames[:] = [d for d in dirnames if d not in SKIP_DIRS and not d.startswith(".")]
+        rel_dir = Path(dirpath).relative_to(repo)
+        for fn in sorted(filenames):
+            rel = fn if str(rel_dir) == "." else str(rel_dir / fn)
+            entries.append(rel)
+            if fn.endswith(SOURCE_EXTS):
+                sources.append(repo / rel)
+            if len(entries) >= MAX_TREE_ENTRIES:
+                break
+        if len(entries) >= MAX_TREE_ENTRIES:
+            break
+    # Prefer shallow files for headers (entry points tend to live near the root).
+    sources.sort(key=lambda p: (len(p.relative_to(repo).parts), str(p)))
+    return entries, sources
+
+
+def gather_repo_context(repo: str | Path) -> dict:
+    """Collect a compact, truncated signal about a repository for the LLM."""
+    repo = Path(repo)
+    manifests = {
+        m: _read_clip(repo / m, MAX_MANIFEST_CHARS)
+        for m in MANIFESTS if (repo / m).is_file()
+    }
+    entries, sources = _walk_tree(repo)
+    source_headers = {
+        str(sp.relative_to(repo)): _read_clip(sp, MAX_SOURCE_HEAD_CHARS)
+        for sp in sources[:MAX_SOURCE_HEADERS]
+    }
+    return {
+        "name": repo.name,
+        "readme": _first_readme(repo),
+        "manifests": manifests,
+        "tree": entries,
+        "source_headers": source_headers,
+    }
+
+
+def build_repo_prompt(context: dict) -> str:
+    manifests = "\n".join(
+        f"--- {name} ---\n{text}" for name, text in context.get("manifests", {}).items()
+    )[: MAX_MANIFEST_CHARS * 2] or "(none)"
+    tree = "\n".join(context.get("tree", []))[:MAX_TREE_CHARS] or "(empty)"
+    headers = "\n".join(
+        f"--- {path} ---\n{text}" for path, text in context.get("source_headers", {}).items()
+    )[: MAX_SOURCE_HEAD_CHARS * MAX_SOURCE_HEADERS] or "(none)"
+    return _PROMPT.format(
+        name=context.get("name", "(unknown)"),
+        readme=(context.get("readme") or "(none)"),
+        manifests=manifests,
+        tree=tree,
+        source_headers=headers,
+    )
+
+
+def analyze_repo(client: OllamaClient, context: dict) -> dict:
+    """gemma4 json_mode → normalized analysis dict. Mirrors summarize_paper."""
+    name = context.get("name", "")
+    raw = client.generate(build_repo_prompt(context), json_mode=True, num_predict=600)
+    try:
+        data = json.loads(raw)
+    except (json.JSONDecodeError, TypeError):
+        return {"name": name, "raw": raw.strip()[:1000], "parse_error": True}
+    if not data.get("name"):
+        data["name"] = name
+    for k in ("technologies", "features", "domain_keywords"):
+        v = data.get(k)
+        if isinstance(v, str):
+            data[k] = [v]
+        elif not isinstance(v, list):
+            data[k] = []
+    for k in ("purpose", "architecture"):
+        if not isinstance(data.get(k), str):
+            data[k] = str(data.get(k) or "")
+    return data
+
+
+def repo_slug(name: str) -> str:
+    return slugify(name or "project")
+
+
+def analyze_all(client: OllamaClient, repos: list[Path], *, max_repos: int | None = None,
+                on_event=None) -> list[dict]:
+    if max_repos:
+        repos = repos[:max_repos]
+    analyses: list[dict] = []
+    for repo in repos:
+        analysis = analyze_repo(client, gather_repo_context(repo))
+        analyses.append(analysis)
+        if on_event:
+            on_event("repo_analyzed", name=analysis.get("name", ""),
+                     parse_error=bool(analysis.get("parse_error")))
+    return analyses
+
+
+def _projects_dir(vault_root: str | Path) -> Path:
+    return Path(vault_root) / "projects"
+
+
+def write_repo_sidecar(vault_root: str | Path, analysis: dict) -> Path:
+    d = _projects_dir(vault_root)
+    d.mkdir(parents=True, exist_ok=True)
+    path = d / f"{repo_slug(analysis.get('name'))}.json"
+    path.write_text(json.dumps(analysis, indent=2))
+    return path
+
+
+def load_repo_sidecars(vault_root: str | Path) -> list[dict]:
+    """Read back project analyses so the taxonomy can be re-synthesized without
+    re-running the LLM over every repo."""
+    d = _projects_dir(vault_root)
+    out: list[dict] = []
+    if not d.is_dir():
+        return out
+    for p in sorted(d.glob("*.json")):
+        try:
+            out.append(json.loads(p.read_text()))
+        except (OSError, json.JSONDecodeError):
+            continue
+    return out
+
+
+def _analysis_text(a: dict) -> str:
+    parts = [
+        a.get("name", ""),
+        a.get("purpose", ""),
+        "Technologies: " + ", ".join(a.get("technologies", [])),
+        "Features: " + ", ".join(a.get("features", [])),
+        "Architecture: " + (a.get("architecture") or ""),
+        "Domains: " + ", ".join(a.get("domain_keywords", [])),
+    ]
+    return "\n".join(p for p in parts if p.strip())
+
+
+def embed_analyses(client: OllamaClient, vault_root: str | Path, analyses: list[dict]) -> int:
+    """Embed project analyses into the main vault store so agents can query
+    'which projects use X' alongside papers/videos. Returns count embedded.
+
+    Each project becomes a searchable doc keyed `project:{slug}`, with its prose
+    written to `vault/projects/{slug}.txt`. The projects/ dir is excluded from
+    extract/summarize/embed discovery (see extract.iter_documents), so this is
+    the only path that embeds them — no double-processing.
+    """
+    from .embedstore import EmbedStore, add_document, merge
+
+    root = Path(vault_root)
+    store = EmbedStore.load(root)
+    seen = store.dois
+    d = _projects_dir(root)
+    d.mkdir(parents=True, exist_ok=True)
+    added = 0
+    for a in analyses:
+        if a.get("parse_error"):
+            continue
+        slug = repo_slug(a.get("name"))
+        doi = f"project:{slug}"
+        if doi in seen:
+            continue
+        text = _analysis_text(a)
+        txt_path = d / f"{slug}.txt"
+        txt_path.write_text(text)
+        vecs, rows = add_document(
+            client, text=text, doi=doi, title=a.get("name", slug),
+            topic="project", text_path=str(txt_path),
+        )
+        store = merge(store, vecs, rows)
+        seen.add(doi)
+        added += 1
+    store.save(root)
+    return added

clawlibrary/extract.py

@@ -120,8 +120,12 @@ def iter_documents(vault_root: Path):
     This is the discovery surface for embed/summarize: PDFs gain a .txt after
     `extract`, YouTube transcripts have one from the start — both look the same
     here, so video knowledge is embedded and summarized alongside the papers.
-    Callers use ``txt.with_suffix(".json")`` for the sidecar.
+    Callers use ``txt.with_suffix(".json")`` for the sidecar. The ``projects/``
+    dir (LLM project analyses from `discover`) is excluded — those are embedded
+    by discover.embed_analyses and must not be re-summarized as papers.
     """
     for txt in sorted(vault_root.glob("*/*.txt")):
+        if txt.parent.name == "projects":
+            continue
         if txt.with_suffix(".json").exists():
             yield txt

clawlibrary/sources/openalex.py

@@ -22,7 +22,10 @@ API_URL = "https://api.openalex.org/works"
 
 
 def _build_filters(topic: Topic) -> str:
-    parts = ["type:article"]
+    # None => articles (default); a books pass sets ["book", "book-chapter"].
+    # OpenAlex ORs values within one key via "|": type:book|book-chapter
+    types = topic.work_types or ["article"]
+    parts = ["type:" + "|".join(types)]
     f = topic.filters
     if f.open_access_only:
         parts.append("is_oa:true")

clawlibrary/taxonomy.py

@@ -0,0 +1,189 @@
+"""Synthesize a tiered research-domain taxonomy from project analyses, and
+derive the harvest queues (papers/books topics + a YouTube search queue).
+
+The taxonomy has three tiers — general → related → niche — so harvesting drills
+from broad fields down to the specific subtopics a person's projects are about,
+capturing resources at every tier. gemma4 (JSON mode) does the synthesis; the
+derivation to Topic objects / YouTube `ytsearch` sources is deterministic.
+"""
+
+from __future__ import annotations
+
+import json
+import re
+from pathlib import Path
+
+import yaml
+from pydantic import BaseModel, Field
+
+from .config import Topic, TopicFilters, write_topics
+from .ollama_client import OllamaClient
+from .sources.arxiv_search import DEFAULT_CATEGORIES
+from .vault import slugify
+
+TIERS = ("general", "related", "niche")
+
+# tier → harvest knobs. general is harvested first (lowest priority number) and
+# broad; niche is laser-focused and drilled deeper per query.
+TIER_PRIORITY = {"general": 1, "related": 5, "niche": 9}
+TIER_MAX_RESULTS = {"general": 120, "related": 80, "niche": 60}
+
+_CAT_RE = re.compile(r"^[a-z][a-z-]*(\.[A-Za-z-]+)?$")  # e.g. cs.AI, quant-ph, math.OC
+MAX_ANALYSES_CHARS = 5000
+
+
+class TaxonomyError(RuntimeError):
+    pass
+
+
+class Domain(BaseModel):
+    name: str
+    tier: str
+    queries: list[str] = Field(default_factory=list)
+    arxiv_categories: list[str] = Field(default_factory=list)
+    rationale: str = ""
+
+
+class Taxonomy(BaseModel):
+    general: list[Domain] = Field(default_factory=list)
+    related: list[Domain] = Field(default_factory=list)
+    niche: list[Domain] = Field(default_factory=list)
+
+    def all_domains(self) -> list[Domain]:
+        return [*self.general, *self.related, *self.niche]
+
+
+_PROMPT = """You are a research librarian. Given analyses of a person's code \
+projects, design a TIERED taxonomy of research domains to harvest literature for.
+
+Tiers:
+- "general": broad parent fields the projects sit in (3-6 domains)
+- "related": adjacent fields worth tracking (4-8 domains)
+- "niche": specific, laser-focused subtopics derived from the projects (5-12 domains)
+
+Respond with ONLY a JSON object: {{"general": [...], "related": [...], "niche": [...]}}
+where each domain is:
+{{"name": "<domain name>", "queries": ["<2-4 search query strings>"], \
+"arxiv_categories": ["<0-4 arXiv codes like cs.AI, quant-ph>"], \
+"rationale": "<one sentence tying it to the projects>"}}
+
+Make general queries broad and niche queries specific. Use arXiv codes only from \
+the standard taxonomy; omit if unsure.
+
+Project analyses:
+{analyses}
+"""
+
+
+def _as_list(v) -> list[str]:
+    if isinstance(v, str):
+        return [v]
+    if isinstance(v, list):
+        return [str(x) for x in v if str(x).strip()]
+    return []
+
+
+def _valid_cat(c: str) -> bool:
+    return bool(_CAT_RE.match(c.strip()))
+
+
+def build_taxonomy_prompt(analyses: list[dict]) -> str:
+    compact = [
+        {
+            "name": a.get("name", ""),
+            "purpose": a.get("purpose", ""),
+            "technologies": a.get("technologies", []),
+            "domain_keywords": a.get("domain_keywords", []),
+        }
+        for a in analyses if not a.get("parse_error")
+    ]
+    blob = json.dumps(compact, indent=1)[:MAX_ANALYSES_CHARS]
+    return _PROMPT.format(analyses=blob)
+
+
+def synthesize(client: OllamaClient, analyses: list[dict]) -> Taxonomy:
+    """gemma4 json_mode → validated Taxonomy. Raises TaxonomyError on bad output
+    (we never want to write a broken taxonomy)."""
+    raw = client.generate(build_taxonomy_prompt(analyses), json_mode=True, num_predict=1400)
+    try:
+        data = json.loads(raw)
+    except (json.JSONDecodeError, TypeError) as e:
+        raise TaxonomyError(f"could not parse taxonomy JSON: {e}") from e
+    if not isinstance(data, dict):
+        raise TaxonomyError("taxonomy response was not a JSON object")
+
+    tiers: dict[str, list[Domain]] = {}
+    for tier in TIERS:
+        domains: list[Domain] = []
+        for d in data.get(tier) or []:
+            if not isinstance(d, dict) or not d.get("name"):
+                continue
+            queries = _as_list(d.get("queries")) or [str(d["name"])]
+            cats = [c.strip() for c in _as_list(d.get("arxiv_categories")) if _valid_cat(c)]
+            domains.append(Domain(
+                name=str(d["name"]).strip(), tier=tier, queries=queries,
+                arxiv_categories=cats, rationale=str(d.get("rationale") or ""),
+            ))
+        tiers[tier] = domains
+
+    taxonomy = Taxonomy(**tiers)
+    if not taxonomy.all_domains():
+        raise TaxonomyError("taxonomy contained no usable domains")
+    return taxonomy
+
+
+def to_topics(taxonomy: Taxonomy, *, include_books: bool = True,
+              date_from: int | None = 2019) -> list[Topic]:
+    """Derive harvest topics from the taxonomy. One Topic per (domain, query),
+    tier→priority/max_results, plus an optional OA-books pass per domain."""
+    topics: list[Topic] = []
+    for domain in taxonomy.all_domains():
+        tier = domain.tier
+        # general tier with no model categories falls back to the broad default set.
+        cats = domain.arxiv_categories or (DEFAULT_CATEGORIES if tier == "general" else None)
+        for query in domain.queries:
+            topics.append(Topic(
+                query=query,
+                priority=TIER_PRIORITY[tier],
+                max_results=TIER_MAX_RESULTS[tier],
+                arxiv_categories=cats,
+                filters=TopicFilters(open_access_only=True, language="en", date_from=date_from),
+            ))
+        if include_books:
+            topics.append(Topic(
+                query=domain.queries[0] if domain.queries else domain.name,
+                priority=TIER_PRIORITY[tier],
+                max_results=40,
+                arxiv_categories=cats,
+                work_types=["book", "book-chapter"],
+                filters=TopicFilters(open_access_only=True, language="en"),
+            ))
+    return topics
+
+
+def to_youtube_queue(taxonomy: Taxonomy, *, per_query_videos: int = 8) -> dict:
+    """Build a youtube.yaml-shaped queue of `ytsearchN:` sources (one per query)."""
+    sources = []
+    for domain in taxonomy.all_domains():
+        slug = slugify(domain.name)
+        for query in domain.queries:
+            sources.append({
+                "name": domain.name,
+                "url": f"ytsearch{per_query_videos}:{query}",
+                "topic": slug,
+                "max_videos": per_query_videos,
+            })
+    return {"defaults": {"max_videos": per_query_videos, "mode": "auto"}, "sources": sources}
+
+
+def write_domains_yaml(taxonomy: Taxonomy, path: str | Path) -> None:
+    Path(path).write_text(yaml.safe_dump(taxonomy.model_dump(), sort_keys=False,
+                                         allow_unicode=True))
+
+
+def write_topics_yaml(topics: list[Topic], path: str | Path) -> None:
+    write_topics(topics, path)
+
+
+def write_youtube_yaml(queue: dict, path: str | Path) -> None:
+    Path(path).write_text(yaml.safe_dump(queue, sort_keys=False, allow_unicode=True))

tests/test_cli.py

@@ -158,3 +158,42 @@ def test_cli_search_backend_down(monkeypatch, capsys):
     assert cli.main(["search", "rust", "--json"]) == 2
     out = json.loads(capsys.readouterr().out)
     assert "error" in out
+
+
+def test_cli_discover_dispatch(project, monkeypatch, tmp_path):
+    import clawlibrary.discover as D
+    import clawlibrary.taxonomy as TX
+    from clawlibrary.ollama_client import OllamaClient
+
+    monkeypatch.setattr(OllamaClient, "is_up", lambda self: True)
+    monkeypatch.setattr(D, "find_repos", lambda root: [tmp_path / "repoA"])
+    monkeypatch.setattr(D, "gather_repo_context", lambda r: {"name": "repoA"})
+    monkeypatch.setattr(D, "analyze_repo", lambda c, ctx: {"name": "repoA",
+                        "domain_keywords": ["RAG"], "technologies": [], "features": []})
+    monkeypatch.setattr(D, "write_repo_sidecar", lambda root, a: None)
+    monkeypatch.setattr(D, "embed_analyses", lambda c, root, a: 1)
+    tax = TX.Taxonomy(general=[TX.Domain(name="AI", tier="general", queries=["ml"])])
+    monkeypatch.setattr(TX, "synthesize", lambda c, a: tax)
+    writes = {}
+    monkeypatch.setattr(TX, "write_domains_yaml", lambda t, p: writes.update(domains=p))
+    monkeypatch.setattr(TX, "write_topics_yaml", lambda t, p: writes.update(topics=p))
+    monkeypatch.setattr(TX, "write_youtube_yaml", lambda q, p: writes.update(yt=p))
+
+    rc = cli.main(["discover", "--root", str(tmp_path), "--no-embed"])
+    assert rc == 0
+    assert writes == {"domains": "domains.yaml", "topics": "topics.generated.yaml",
+                      "yt": "youtube.generated.yaml"}
+
+
+def test_cli_run_use_generated_rewrites_topics(project, monkeypatch):
+    captured = {}
+
+    def fake_run(config, topics, **kwargs):
+        captured["called"] = True
+        return RunStats(run_id="r1")
+
+    # write a minimal generated topics file so _load succeeds
+    (project / "topics.generated.yaml").write_text("topics:\n  - query: gen-domain\n")
+    monkeypatch.setattr(cli, "run", fake_run)
+    rc = cli.main(["run", "--use-generated"])
+    assert rc == 0 and captured["called"]

tests/test_config.py

@@ -1,6 +1,6 @@
 import pytest
 
-from clawlibrary.config import load_config, load_topics
+from clawlibrary.config import Topic, TopicFilters, load_config, load_topics, write_topics
 
 
 def _write(tmp_path, name, text):
@@ -69,3 +69,21 @@ def test_load_config_env_overlay(tmp_path, monkeypatch):
 def test_load_config_missing_file(tmp_path):
     with pytest.raises(FileNotFoundError):
         load_config(tmp_path / "nope.yaml", load_env=False)
+
+
+def test_write_topics_round_trips(tmp_path):
+    topics = [
+        Topic(query="LLM agents tool use", priority=1, max_results=120,
+              arxiv_categories=["cs.AI", "cs.LG"],
+              filters=TopicFilters(date_from=2019, open_access_only=True)),
+        Topic(query="quantum error correction books", priority=9, max_results=60,
+              work_types=["book", "book-chapter"]),
+    ]
+    path = tmp_path / "topics.generated.yaml"
+    write_topics(topics, path)
+    reloaded = load_topics(path).active()  # priority order: 1 then 9
+    assert [t.query for t in reloaded] == [topics[0].query, topics[1].query]
+    assert reloaded[0].arxiv_categories == ["cs.AI", "cs.LG"]
+    assert reloaded[0].filters.date_from == 2019
+    assert reloaded[1].work_types == ["book", "book-chapter"]
+    assert reloaded[1].max_results == 60

tests/test_discover.py

@@ -0,0 +1,83 @@
+import json
+
+from clawlibrary import discover
+from clawlibrary.extract import iter_documents
+
+
+class FakeClient:
+    def __init__(self, gen_response="{}"):
+        self.gen_response = gen_response
+        self.last_prompt = None
+
+    def generate(self, prompt, **kw):
+        self.last_prompt = prompt
+        return self.gen_response
+
+    def embed_documents(self, texts):
+        return [[float(len(t)), 1.0] for t in texts]
+
+
+def _make_repo(root, name, *, git=True, files=None):
+    repo = root / name
+    repo.mkdir(parents=True)
+    if git:
+        (repo / ".git").mkdir()
+    for rel, content in (files or {}).items():
+        p = repo / rel
+        p.parent.mkdir(parents=True, exist_ok=True)
+        p.write_text(content)
+    return repo
+
+
+def test_find_repos_top_level_only(tmp_path):
+    _make_repo(tmp_path, "alpha")
+    beta = _make_repo(tmp_path, "beta")
+    _make_repo(beta, "nested")          # nested repo under beta
+    _make_repo(tmp_path, "plain", git=False)  # no .git
+    repos = discover.find_repos(tmp_path)
+    assert [r.name for r in repos] == ["alpha", "beta"]  # nested + plain excluded
+
+
+def test_gather_repo_context_excludes_junk_and_includes_signal(tmp_path):
+    repo = _make_repo(tmp_path, "proj", files={
+        "README.md": "A tool for vector search over papers.",
+        "pyproject.toml": "[project]\nname='proj'\n",
+        "src/main.py": "import numpy\n",
+        "node_modules/junk/index.js": "x",
+    })
+    ctx = discover.gather_repo_context(repo)
+    assert "vector search" in ctx["readme"]
+    assert "pyproject.toml" in ctx["manifests"]
+    assert any("main.py" in t for t in ctx["tree"])
+    assert not any("node_modules" in t for t in ctx["tree"])  # pruned
+    assert any("main.py" in k for k in ctx["source_headers"])
+
+
+def test_analyze_repo_normalizes_and_handles_bad_json(tmp_path):
+    ctx = {"name": "proj", "readme": "r", "manifests": {}, "tree": [], "source_headers": {}}
+    fc = FakeClient(gen_response=json.dumps({
+        "purpose": "P", "technologies": "Python",  # str -> list
+        "features": ["a"], "domain_keywords": ["vector search", "RAG"],
+    }))
+    out = discover.analyze_repo(fc, ctx)
+    assert out["name"] == "proj"  # injected from context
+    assert out["technologies"] == ["Python"]
+    assert out["domain_keywords"] == ["vector search", "RAG"]
+
+    bad = discover.analyze_repo(FakeClient(gen_response="not json"), ctx)
+    assert bad["parse_error"] is True and bad["name"] == "proj"
+
+
+def test_sidecar_round_trip_and_excluded_from_iter_documents(tmp_path):
+    vault = tmp_path / "vault"
+    analysis = {"name": "Proj X", "purpose": "P", "technologies": ["Python"],
+                "features": [], "architecture": "", "domain_keywords": ["RAG"]}
+    side = discover.write_repo_sidecar(vault, analysis)
+    assert side.parent.name == "projects"
+    loaded = discover.load_repo_sidecars(vault)
+    assert loaded and loaded[0]["name"] == "Proj X"
+
+    # embed writes a projects/{slug}.txt, but iter_documents must skip projects/
+    discover.embed_analyses(FakeClient(), vault, [analysis])
+    assert (vault / "projects" / "proj-x.txt").exists()
+    assert list(iter_documents(vault)) == []  # projects excluded from paper pipeline

tests/test_sources.py

@@ -48,6 +48,12 @@ def test_openalex_search_paginates(openalex_work):
     assert len(results) == 2
 
 
+def test_openalex_build_filters_default_and_books():
+    assert openalex._build_filters(Topic(query="x")).startswith("type:article")
+    books = openalex._build_filters(Topic(query="x", work_types=["book", "book-chapter"]))
+    assert "type:book|book-chapter" in books
+
+
 def test_unpaywall_resolve():
     def handler(request):
         return httpx.Response(200, json={

tests/test_taxonomy.py

@@ -0,0 +1,93 @@
+import json
+
+import pytest
+
+from clawlibrary import taxonomy as T
+from clawlibrary.config import load_topics, write_topics
+from clawlibrary.sources.arxiv_search import DEFAULT_CATEGORIES
+from clawlibrary.youtube import load_queue
+
+
+class FakeClient:
+    def __init__(self, gen_response):
+        self.gen_response = gen_response
+
+    def generate(self, prompt, **kw):
+        return self.gen_response
+
+
+_GOOD = json.dumps({
+    "general": [
+        {"name": "Artificial Intelligence", "queries": ["machine learning", "deep learning"],
+         "arxiv_categories": ["cs.AI", "bogus!!"], "rationale": "core"},
+    ],
+    "related": [
+        {"name": "Information Retrieval", "queries": ["semantic search"],
+         "arxiv_categories": ["cs.IR"], "rationale": "adjacent"},
+    ],
+    "niche": [
+        {"name": "Retrieval Augmented Generation", "queries": ["RAG citations"],
+         "arxiv_categories": ["cs.CL"], "rationale": "specific"},
+    ],
+})
+
+
+def test_synthesize_three_tiers_and_filters_bad_categories():
+    tax = T.synthesize(FakeClient(_GOOD), [{"name": "p", "domain_keywords": ["RAG"]}])
+    assert [d.name for d in tax.general] == ["Artificial Intelligence"]
+    assert tax.general[0].arxiv_categories == ["cs.AI"]  # "bogus!!" dropped
+    assert tax.related and tax.niche
+    assert len(tax.all_domains()) == 3
+
+
+def test_synthesize_raises_on_bad_json():
+    with pytest.raises(T.TaxonomyError):
+        T.synthesize(FakeClient("not json at all"), [])
+
+
+def test_synthesize_raises_when_no_domains():
+    with pytest.raises(T.TaxonomyError):
+        T.synthesize(FakeClient(json.dumps({"general": [], "related": [], "niche": []})), [])
+
+
+def test_to_topics_tiering_and_books(tmp_path):
+    tax = T.synthesize(FakeClient(_GOOD), [{"name": "p"}])
+    topics = T.to_topics(tax, include_books=True)
+    papers = {t.query: t for t in topics if t.work_types is None}
+    # general tier -> priority 1, niche -> 9
+    assert papers["machine learning"].priority == T.TIER_PRIORITY["general"]
+    assert papers["RAG citations"].priority == T.TIER_PRIORITY["niche"]
+    assert papers["machine learning"].max_results == T.TIER_MAX_RESULTS["general"]
+    # a books-pass topic exists with book work types
+    books = [t for t in topics if t.work_types == ["book", "book-chapter"]]
+    assert books and all(t.max_results == 40 for t in books)
+
+
+def test_to_topics_general_category_fallback():
+    # a general domain with NO valid categories falls back to DEFAULT_CATEGORIES
+    data = json.dumps({"general": [{"name": "X", "queries": ["x"], "arxiv_categories": []}],
+                       "related": [], "niche": []})
+    tax = T.synthesize(FakeClient(data), [])
+    topics = T.to_topics(tax, include_books=False)
+    assert topics[0].arxiv_categories == DEFAULT_CATEGORIES
+
+
+def test_to_topics_round_trips_through_write_topics(tmp_path):
+    tax = T.synthesize(FakeClient(_GOOD), [{"name": "p"}])
+    topics = T.to_topics(tax)
+    path = tmp_path / "topics.generated.yaml"
+    write_topics(topics, path)
+    reloaded = load_topics(path).active()
+    assert reloaded[0].priority == 1  # general first
+    assert any(t.work_types == ["book", "book-chapter"] for t in reloaded)
+
+
+def test_to_youtube_queue_emits_ytsearch(tmp_path):
+    tax = T.synthesize(FakeClient(_GOOD), [{"name": "p"}])
+    queue = T.to_youtube_queue(tax, per_query_videos=5)
+    assert all(s["url"].startswith("ytsearch5:") for s in queue["sources"])
+    # round-trips through the youtube queue loader
+    path = tmp_path / "youtube.generated.yaml"
+    T.write_youtube_yaml(queue, path)
+    defaults, sources = load_queue(path)
+    assert sources and sources[0].url.startswith("ytsearch5:")