Add scheduled refresh + agent-facing insight library (MCP + JSON CLI)

Part A — incremental `refresh`:
- clawlibrary/refresh.py: in-process orchestrator chaining the already-idempotent
  stages (harvest papers → ingest videos → extract → summarize → embed) on only-new
  items; one run_id, per-stage StageResult, NewItem capture, and a "what's new" feed
  (vault/refresh-log.jsonl + vault/digests/{date}.md).
- clawlibrary/lock.py: O_EXCL file_lock (auto-breaks stale/dead locks) so overlapping
  runs can't corrupt state.json/index.json.
- summarize.append_insight_once/appended_dois: idempotent insights.md (no dup sections
  on re-run); cmd_summarize switched to it.
- cmd_refresh CLI (--source/--skip-youtube/--no-whisper/--max/--model/--dry-run);
  scheduled YouTube allows Whisper GPU fallback. scripts/refresh.sh + cron docs.

Part B — agent access:
- clawlibrary/insights_api.py: load-once LibraryContext wrapping embedstore.search /
  rag.ask; search_vault/ask_library/get_insight/list_topics/list_recent; get_insight
  reads sidecars (index.json is summary-less). VaultUnavailable/BackendDown errors.
- clawlibrary/mcp_server.py: FastMCP stdio server (clawlibrary-mcp), [mcp] extra.
- --json on search/ask + new insight/catalog commands, sharing the same layer.

25 new tests (97 passed, 1 skipped where mcp absent); ruff clean. README updated.

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

README.md

@@ -124,6 +124,72 @@ 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.
 
+## Scheduled refresh (keeping the vault current)
+
+`clawlibrary refresh` re-polls every tracked domain and folds only the **new**
+content into the insight layer — in one idempotent pass:
+
+```
+harvest new papers (OpenAlex/arXiv) → ingest new videos (youtube.yaml)
+   → extract → summarize → embed
+```
+
+Each stage already skips what it's done, so re-running is cheap and safe. The run
+is guarded by a lockfile (`state.json.lock`) so overlapping invocations can't
+corrupt state, and it emits a **"what's new" feed**:
+
+- `vault/refresh-log.jsonl` — one JSON line per run (counts + new items); the
+  machine-readable stream an agent can tail.
+- `vault/digests/{YYYYMMDD}.md` — a skimmable page of new papers/videos with
+  one-line summaries.
+
+```bash
+clawlibrary refresh --source both --dry-run   # what would be added; writes nothing
+clawlibrary refresh --source both             # do it (Whisper fallback ON for videos)
+clawlibrary refresh --source both --no-whisper --skip-youtube   # papers only, no GPU
+```
+
+**Cron** (on the GPU node `tank`) — `scripts/refresh.sh` activates the venv, runs
+the refresh, logs to `logs/refresh-{date}.log`, and restarts the MCP server so
+agents pick up fresh embeddings:
+
+```cron
+30 3 * * *  /home/osobh/clawlibrary/scripts/refresh.sh
+```
+
+## Agent access — MCP server + JSON CLI
+
+The vault is exposed to agents two ways, both sharing one query layer
+(`clawlibrary/insights_api.py`, which loads the embedding matrix once).
+
+**MCP server** (recommended for agents) — `pip install -e ".[mcp]"`, then register
+the `clawlibrary-mcp` stdio server. Tools: `search_vault`, `ask_library`,
+`get_insight`, `list_topics`, `list_recent`; resources: `clawlibrary://insights`,
+`clawlibrary://catalog`. Run it on `tank` (keeps the matrix local to the GPU box):
+
+```json
+{
+  "mcpServers": {
+    "clawlibrary": {
+      "command": "/home/osobh/clawlibrary/.venv/bin/clawlibrary-mcp",
+      "cwd": "/home/osobh/clawlibrary",
+      "env": { "CLAWLIBRARY_CONFIG": "config.yaml", "OLLAMA_HOST": "http://localhost:11434" }
+    }
+  }
+}
+```
+
+**JSON CLI** (shell/cron fallback) — `--json` emits clean JSON on stdout:
+
+```bash
+clawlibrary search "agentic planning" --json | jq .
+clawlibrary ask "How does the surface code reach threshold?" --json
+clawlibrary insight 10.48550/arxiv.2401.01234 --json
+clawlibrary catalog --topics --json
+```
+
+Exit codes: `0` ok, `1` not-found (`insight`), `2` vault/ollama unavailable.
+
 ## Subscription content (TDM)
 
 For papers with no open-access copy, the legitimate path is your institution's

clawlibrary/__main__.py

@@ -7,9 +7,12 @@ Commands:
   extract [--limit N] [--vlm-fallback]     extract full text from vault PDFs
   summarize [--limit N] [--model M]        LLM structured summaries (gemma4)
   embed [--limit N]                        embed documents for semantic search
-  search QUERY [--k N]                     semantic search across the vault
+  search QUERY [--k N] [--json]            semantic search across the vault
-  ask QUESTION [--k N]                     RAG answer from retrieved passages
+  ask QUESTION [--k N] [--json]            RAG answer from retrieved passages
+  insight DOI [--json]                     structured insight for one document
+  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
 """
 
 from __future__ import annotations
@@ -189,7 +192,7 @@ def cmd_summarize(args: argparse.Namespace) -> int:
 
     from .extract import iter_documents
     from .ollama_client import OllamaClient
-    from .summarize import append_insight, has_summary, summarize_one
+    from .summarize import append_insight_once, appended_dois, has_summary, summarize_one
 
     config, _ = _load(args.config, args.topics)
     oc = OllamaClient(text_model=args.model)
@@ -197,6 +200,7 @@ def cmd_summarize(args: argparse.Namespace) -> int:
         console.print("[red]Ollama not reachable.[/red]")
         return 2
     insights = Path(config.vault.root) / "insights.md"
+    seen_insights = appended_dois(insights)  # dedup digest sections across runs
     docs = list(iter_documents(Path(config.vault.root)))
     if args.limit:
         docs = docs[: args.limit]
@@ -215,7 +219,7 @@ def cmd_summarize(args: argparse.Namespace) -> int:
             console.print(f"[red]✗[/red] {doc.name}: {e}")
             continue
         if summary is not None:
-            append_insight(insights, _json.loads(side.read_text()))
+            append_insight_once(insights, _json.loads(side.read_text()), seen=seen_insights)
             done += 1
         if i % 10 == 0:
             console.print(f"  summarize {i}/{total}  (done {done}, skip {skipped}, fail {failed})")
@@ -276,55 +280,126 @@ def cmd_embed(args: argparse.Namespace) -> int:
 
 
 def cmd_search(args: argparse.Namespace) -> int:
-    from .embedstore import EmbedStore, search
+    import json as _json
-    from .ollama_client import OllamaClient
 
-    config, _ = _load(args.config, args.topics)
+    from .insights_api import BackendDown, LibraryContext, VaultUnavailable, search_vault
-    store = EmbedStore.load(Path(config.vault.root))
+    from .ollama_client import OllamaError
-    if store.vectors.size == 0:
+
-        console.print("[yellow]No embeddings yet. Run `clawlibrary embed` first.[/yellow]")
+    ctx = LibraryContext.create(args.config)
-        return 0
+    try:
-    oc = OllamaClient()
+        rows = search_vault(ctx, args.query, k=args.k)
-    if not oc.is_up():
+    except (VaultUnavailable, BackendDown, OllamaError) as e:
-        console.print("[red]Ollama not reachable.[/red]")
+        if args.json:
+            print(_json.dumps({"error": str(e)}))
+        else:
+            console.print(f"[red]{e}[/red]")
         return 2
-    hits = search(store, oc, args.query, k=args.k)
+    if args.json:
+        print(_json.dumps(rows))
+        return 0
     table = Table(title=f'Search: "{args.query}"', header_style="bold")
     table.add_column("Score", justify="right")
     table.add_column("Title")
     table.add_column("Topic", style="dim")
-    for h in hits:
+    for r in rows:
-        table.add_row(f"{h.score:.3f}", h.title[:70], h.topic[:30])
+        table.add_row(f"{r['score']:.3f}", r["title"][:70], r["topic"][:30])
     console.print(table)
     if args.snippets:
-        for h in hits:
+        for r in rows:
-            console.print(f"\n[bold]{h.title[:80]}[/bold] [dim]{h.doi}[/dim]")
+            console.print(f"\n[bold]{r['title'][:80]}[/bold] [dim]{r['doi']}[/dim]")
-            console.print(f"  {h.snippet}")
+            console.print(f"  {r['snippet']}")
+    return 0
+
+
+def cmd_insight(args: argparse.Namespace) -> int:
+    import json as _json
+
+    from .insights_api import LibraryContext, get_insight
+
+    ctx = LibraryContext.create(args.config)
+    data = get_insight(ctx, args.doi)
+    if data is None:
+        if args.json:
+            print(_json.dumps({"error": "not found"}))
+        else:
+            console.print(f"[yellow]No insight for {args.doi}[/yellow]")
+        return 1
+    if args.json:
+        print(_json.dumps(data))
+        return 0
+    s = data.get("summary") or {}
+    console.print(f"[bold]{data.get('title', '')}[/bold] [dim]{data.get('doi', '')}[/dim]")
+    console.print(f"[dim]{data.get('topic', '')} · {data.get('source', '')} · "
+                  f"{data.get('year', '')}[/dim]")
+    if s.get("problem"):
+        console.print(f"\n[bold]Problem:[/bold] {s['problem']}")
+    if s.get("method"):
+        console.print(f"[bold]Method:[/bold] {s['method']}")
+    if s.get("key_findings"):
+        console.print("[bold]Key findings:[/bold]")
+        for f in s["key_findings"]:
+            console.print(f"  - {f}")
+    if s.get("relevance"):
+        console.print(f"[bold]Relevance:[/bold] {s['relevance']}")
+    if s.get("keywords"):
+        console.print(f"[bold]Keywords:[/bold] {', '.join(s['keywords'])}")
+    return 0
+
+
+def cmd_catalog(args: argparse.Namespace) -> int:
+    import json as _json
+
+    from .insights_api import LibraryContext, list_recent, list_topics
+
+    ctx = LibraryContext.create(args.config)
+    rows = list_topics(ctx) if args.topics else list_recent(ctx, n=args.recent)
+    if args.json:
+        print(_json.dumps(rows))
+        return 0
+    if args.topics:
+        table = Table(title="Topics", header_style="bold")
+        table.add_column("Topic")
+        table.add_column("Chunks", justify="right")
+        for r in rows:
+            table.add_row(r["topic"][:50], str(r["chunks"]))
+    else:
+        table = Table(title=f"{len(rows)} most recent", header_style="bold")
+        table.add_column("Title")
+        table.add_column("Topic", style="dim")
+        table.add_column("Source")
+        for r in rows:
+            table.add_row((r.get("title") or "")[:60], (r.get("topic") or "")[:25],
+                          r.get("source") or "")
+    console.print(table)
     return 0
 
 
 def cmd_ask(args: argparse.Namespace) -> int:
-    from .embedstore import EmbedStore
+    import json as _json
-    from .ollama_client import OllamaClient
-    from .rag import ask
 
-    config, _ = _load(args.config, args.topics)
+    from .insights_api import BackendDown, LibraryContext, VaultUnavailable, ask_library
-    store = EmbedStore.load(Path(config.vault.root))
+    from .ollama_client import OllamaError
-    if store.vectors.size == 0:
+
-        console.print("[yellow]No embeddings yet. Run `clawlibrary embed` first.[/yellow]")
+    ctx = LibraryContext.create(args.config)
-        return 0
+    if args.model:
-    oc = OllamaClient(text_model=args.model)
+        ctx.client.text_model = args.model
-    if not oc.is_up():
+    try:
-        console.print("[red]Ollama not reachable.[/red]")
+        result = ask_library(ctx, args.question, k=args.k)
+    except (VaultUnavailable, BackendDown, OllamaError) as e:
+        if args.json:
+            print(_json.dumps({"error": str(e)}))
+        else:
+            console.print(f"[red]{e}[/red]")
         return 2
-    with console.status("Thinking..."):
+    if args.json:
-        result = ask(store, oc, args.question, k=args.k)
+        print(_json.dumps(result))
+        return 0
     console.print(f"\n[bold]{args.question}[/bold]\n")
-    console.print(result.answer)
+    console.print(result["answer"])
-    if result.sources:
+    if result["sources"]:
         console.print("\n[dim]Sources:[/dim]")
-        for s in result.sources:
+        for s in result["sources"]:
-            console.print(f"  [cyan][{s.n}][/cyan] {s.title[:75]} [dim]{s.doi}[/dim]")
+            console.print(f"  [cyan][{s['n']}][/cyan] {s['title'][:75]} [dim]{s['doi']}[/dim]")
     return 0
 
 
@@ -422,6 +497,65 @@ def cmd_youtube(args: argparse.Namespace) -> int:
     return 0
 
 
+def _print_refresh_summary(stats, *, dry_run: bool) -> None:
+    verb = "Would add" if dry_run else "Added"
+    table = Table(title=f"Refresh {stats.run_id}", header_style="bold")
+    table.add_column("Stage")
+    table.add_column(verb, justify="right")
+    table.add_column("Skipped", justify="right")
+    table.add_column("Failed", justify="right")
+    table.add_column("Status")
+    for s in stats.stages:
+        status = ("[red]error[/red]" if s.error
+                  else "skipped" if not s.ran else "[green]ok[/green]")
+        pending = s.detail.get("pending")
+        added = str(pending) if (dry_run and pending is not None) else str(s.added)
+        table.add_row(s.name, added, str(s.skipped), str(s.failed), status)
+    console.print(table)
+    papers = sum(1 for i in stats.new_items if i.kind == "paper")
+    videos = sum(1 for i in stats.new_items if i.kind == "video")
+    if dry_run:
+        console.print("[dim]Dry run — nothing written.[/dim]")
+    else:
+        console.print(
+            f"[bold]New this run:[/bold] {papers} papers, {videos} videos  "
+            f"[dim]→ vault/digests/{stats.run_id[:8]}.md[/dim]"
+        )
+
+
+def cmd_refresh(args: argparse.Namespace) -> int:
+    from .lock import LockError
+    from .refresh import refresh
+
+    config, topics = _load(args.config, args.topics)
+
+    def on_event(kind: str, **data):
+        if kind == "topic_start":
+            console.print(f"[bold cyan]→ {data['topic']}[/bold cyan]")
+        elif kind == "downloaded":
+            console.print(f"  [green]✓[/green] {data.get('title', '')[:80]}  "
+                          f"[dim]({data.get('source', '')})[/dim]")
+        elif kind == "would_download":
+            console.print(f"  [yellow]·[/yellow] {data.get('title', '')[:80]}  "
+                          f"[dim]({data.get('source', '')})[/dim]")
+        elif kind == "stage_error":
+            console.print(f"[red]Stage {data['stage']} failed:[/red] {data['error']}")
+        elif kind == "refresh_start":
+            console.print("[bold]Refreshing all tracked domains…[/bold]")
+
+    try:
+        stats = refresh(
+            config, topics, source=args.source, skip_youtube=args.skip_youtube,
+            youtube_queue=args.queue, max_override=args.max, no_whisper=args.no_whisper,
+            model=args.model, dry_run=args.dry_run, on_event=on_event,
+        )
+    except LockError as e:
+        console.print(f"[red]Refresh already running:[/red] {e}")
+        return 3
+    _print_refresh_summary(stats, dry_run=args.dry_run)
+    return 1 if stats.had_error else 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")
@@ -465,14 +599,27 @@ def build_parser() -> argparse.ArgumentParser:
     pq.add_argument("query", help="natural-language query")
     pq.add_argument("--k", type=int, default=10, help="number of results")
     pq.add_argument("--snippets", action="store_true", help="show matching snippets")
+    pq.add_argument("--json", action="store_true", help="emit JSON to stdout (for agents)")
     pq.set_defaults(func=cmd_search)
 
     pa = sub.add_parser("ask", help="ask a question, answered from retrieved passages (RAG)")
     pa.add_argument("question", help="natural-language question")
     pa.add_argument("--k", type=int, default=6, help="passages to retrieve")
     pa.add_argument("--model", default="gemma4:E4B", help="Ollama model for the answer")
+    pa.add_argument("--json", action="store_true", help="emit JSON to stdout (for agents)")
     pa.set_defaults(func=cmd_ask)
 
+    pin = sub.add_parser("insight", help="structured insight for one DOI / youtube:<id>")
+    pin.add_argument("doi", help="DOI or youtube:<id> key")
+    pin.add_argument("--json", action="store_true", help="emit JSON to stdout (for agents)")
+    pin.set_defaults(func=cmd_insight)
+
+    pcat = sub.add_parser("catalog", help="vault catalog: topics or recent items")
+    pcat.add_argument("--topics", action="store_true", help="list topics + counts (else recent)")
+    pcat.add_argument("--recent", type=int, default=20, help="number of recent items")
+    pcat.add_argument("--json", action="store_true", help="emit JSON to stdout (for agents)")
+    pcat.set_defaults(func=cmd_catalog)
+
     py = sub.add_parser("youtube", help="ingest YouTube videos as transcripts into the vault")
     py.add_argument("--queue", default="youtube.yaml", help="path to the youtube.yaml queue")
     py.add_argument("--url", help="ad-hoc channel/playlist/video URL (requires --topic)")
@@ -488,6 +635,19 @@ def build_parser() -> argparse.ArgumentParser:
     py.add_argument("--whisper-compute", default=None,
                     help="ctranslate2 compute type (e.g. float16, int8); default by device")
     py.set_defaults(func=cmd_youtube)
+
+    prf = sub.add_parser("refresh",
+                         help="incremental harvest→ingest→extract→summarize→embed of new content")
+    prf.add_argument("--source", choices=["openalex", "arxiv", "both"], default="openalex",
+                     help="paper discovery source (scheduler uses 'both')")
+    prf.add_argument("--skip-youtube", action="store_true", help="papers only, skip video queue")
+    prf.add_argument("--queue", default="youtube.yaml", help="path to the youtube.yaml queue")
+    prf.add_argument("--max", type=int, default=None, help="cap new items per topic/source")
+    prf.add_argument("--no-whisper", action="store_true",
+                     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.set_defaults(func=cmd_refresh)
     return p
 
 

clawlibrary/insights_api.py

@@ -0,0 +1,152 @@
+"""Agent-facing query layer over the vault: load-once retrieval + structured insights.
+
+Transport-agnostic. Both the MCP server (`mcp_server.py`) and the CLI `--json`
+modes call these functions, so retrieval is defined exactly once — they reuse
+`embedstore.search` / `rag.ask`, never reimplement it.
+
+`LibraryContext` loads the (~41k-vector) embedding matrix and an Ollama client
+ONCE and reuses them across calls; the MCP server holds a single context for its
+whole lifetime, while the CLI builds one per invocation.
+"""
+
+from __future__ import annotations
+
+import json
+from dataclasses import dataclass
+from functools import lru_cache
+from pathlib import Path
+
+from .config import load_config
+from .embedstore import EmbedStore, search
+from .ollama_client import OllamaClient
+from .rag import ask as rag_ask
+
+
+class VaultUnavailable(RuntimeError):
+    """No embeddings on disk yet (run `clawlibrary embed`)."""
+
+
+class BackendDown(RuntimeError):
+    """Ollama is not reachable."""
+
+
+@dataclass
+class LibraryContext:
+    root: Path
+    store: EmbedStore
+    client: OllamaClient
+
+    @classmethod
+    def create(cls, config_path: str = "config.yaml", *, host: str | None = None) -> LibraryContext:
+        cfg = load_config(config_path)
+        root = Path(cfg.vault.root)
+        store = EmbedStore.load(root)  # one np.load + jsonl parse, reused thereafter
+        client = OllamaClient(host=host) if host else OllamaClient()
+        return cls(root=root, store=store, client=client)
+
+    def require_ready(self) -> None:
+        if self.store.vectors.size == 0:
+            raise VaultUnavailable("no embeddings yet; run `clawlibrary embed` first")
+        if not self.client.is_up():
+            raise BackendDown("ollama not reachable")
+
+
+# --------------------------------------------------------------------------- #
+# Tool implementations — return plain JSON-ready dict/list
+# --------------------------------------------------------------------------- #
+def search_vault(ctx: LibraryContext, query: str, k: int = 10) -> list[dict]:
+    ctx.require_ready()
+    hits = search(ctx.store, ctx.client, query, k=k)
+    return [
+        {"score": round(h.score, 4), "title": h.title, "doi": h.doi,
+         "topic": h.topic, "snippet": h.snippet}
+        for h in hits
+    ]
+
+
+def ask_library(ctx: LibraryContext, question: str, k: int = 6) -> dict:
+    ctx.require_ready()
+    ans = rag_ask(ctx.store, ctx.client, question, k=k)
+    return {
+        "answer": ans.answer,
+        "sources": [
+            {"n": s.n, "title": s.title, "doi": s.doi, "topic": s.topic}
+            for s in ans.sources
+        ],
+    }
+
+
+def get_insight(ctx: LibraryContext, doi: str) -> dict | None:
+    """Structured per-paper/video insight from its sidecar (NOT index.json, which
+    carries no summary). Returns None if the DOI isn't in the vault."""
+    side = _sidecar_for_doi(ctx, doi)
+    if side is None:
+        return None
+    try:
+        data = json.loads(side.read_text())
+    except (OSError, json.JSONDecodeError):
+        return None
+    return {
+        "doi": data.get("doi"), "title": data.get("title"), "topic": data.get("topic"),
+        "year": data.get("year"), "authors": data.get("authors", []),
+        "source": data.get("source"), "summary": data.get("summary", {}),
+        "word_count": data.get("word_count"),
+    }
+
+
+def list_topics(ctx: LibraryContext) -> list[dict]:
+    counts: dict[str, int] = {}
+    for m in ctx.store.meta:
+        t = m.get("topic", "")
+        if t:
+            counts[t] = counts.get(t, 0) + 1
+    return [{"topic": t, "chunks": n} for t, n in sorted(counts.items(), key=lambda kv: -kv[1])]
+
+
+def list_recent(ctx: LibraryContext, n: int = 20) -> list[dict]:
+    entries = _load_index(ctx.root)
+    entries.sort(key=lambda e: e.get("downloaded_at", ""), reverse=True)
+    return [
+        {"doi": e.get("doi"), "title": e.get("title"), "topic": e.get("topic"),
+         "year": e.get("year"), "source": e.get("source")}
+        for e in entries[:n]
+    ]
+
+
+# --------------------------------------------------------------------------- #
+# Helpers
+# --------------------------------------------------------------------------- #
+def _sidecar_for_doi(ctx: LibraryContext, doi: str) -> Path | None:
+    target = doi.lower().strip()
+    # Fast path: meta rows already in memory carry text_path.
+    for m in ctx.store.meta:
+        if (m.get("doi") or "").lower() == target:
+            side = Path(m["text_path"]).with_suffix(".json")
+            if side.exists():
+                return side
+    # Fallback: doi→sidecar map built by globbing sidecars (cached per root).
+    return _doi_index(ctx.root).get(target)
+
+
+@lru_cache(maxsize=4)
+def _doi_index(root: Path) -> dict[str, Path]:
+    out: dict[str, Path] = {}
+    for side in root.glob("*/*.json"):
+        try:
+            d = json.loads(side.read_text())
+        except (OSError, json.JSONDecodeError):
+            continue
+        key = (d.get("doi") or d.get("id") or "").lower()
+        if key:
+            out[key] = side
+    return out
+
+
+def _load_index(root: Path) -> list[dict]:
+    p = root / "index.json"
+    if not p.exists():
+        return []
+    try:
+        return json.loads(p.read_text())
+    except (OSError, json.JSONDecodeError):
+        return []

clawlibrary/lock.py

@@ -0,0 +1,96 @@
+"""A tiny, dependency-free cross-run lockfile.
+
+Guards the files a `refresh` rewrites (state.json, vault/index.json) against a
+second overlapping run — a manual invocation racing the scheduled one, say.
+Atomic create via O_EXCL; auto-breaks a lock left by a dead process (same host)
+or one older than `stale_after_seconds` (covers a crash on another host).
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import socket
+import time
+from contextlib import contextmanager
+from pathlib import Path
+
+
+class LockError(RuntimeError):
+    pass
+
+
+def _pid_alive(pid: int) -> bool:
+    try:
+        os.kill(pid, 0)
+    except ProcessLookupError:
+        return False
+    except PermissionError:
+        return True  # exists, owned by someone else
+    return True
+
+
+def _read_lock(path: Path) -> dict:
+    try:
+        return json.loads(path.read_text())
+    except (OSError, json.JSONDecodeError):
+        return {}
+
+
+def _is_stale(info: dict, *, stale_after_seconds: float) -> bool:
+    started = float(info.get("started_at_epoch", 0) or 0)
+    if started and (time.time() - started) > stale_after_seconds:
+        return True
+    # Same-host liveness check (can't probe a remote pid).
+    if info.get("host") == socket.gethostname():
+        pid = int(info.get("pid", 0) or 0)
+        if pid and not _pid_alive(pid):
+            return True
+    return False
+
+
+@contextmanager
+def file_lock(path: str | Path, *, stale_after_seconds: float = 6 * 3600):
+    """Hold an exclusive lock at `path` for the duration of the context.
+
+    Raises LockError if a live, non-stale lock is already held.
+    """
+    path = Path(path)
+    path.parent.mkdir(parents=True, exist_ok=True)
+    info = {
+        "pid": os.getpid(),
+        "host": socket.gethostname(),
+        "started_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
+        "started_at_epoch": time.time(),
+    }
+    payload = json.dumps(info)
+
+    def _acquire() -> int:
+        fd = os.open(path, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o644)
+        os.write(fd, payload.encode("utf-8"))
+        return fd
+
+    try:
+        fd = _acquire()
+    except FileExistsError:
+        existing = _read_lock(path)
+        if _is_stale(existing, stale_after_seconds=stale_after_seconds):
+            try:
+                path.unlink()
+            except FileNotFoundError:
+                pass
+            try:
+                fd = _acquire()
+            except FileExistsError as e:  # lost a race to break it
+                raise LockError(f"lock held at {path}: {existing}") from e
+        else:
+            raise LockError(f"lock held at {path}: {existing}") from None
+
+    try:
+        os.close(fd)
+        yield path
+    finally:
+        try:
+            path.unlink()
+        except FileNotFoundError:
+            pass

clawlibrary/mcp_server.py

@@ -0,0 +1,94 @@
+"""ClawLibrary MCP server (stdio) — exposes the insight vault to agents.
+
+Tools: search_vault, ask_library, get_insight, list_topics, list_recent.
+Resources: clawlibrary://insights (insights.md), clawlibrary://catalog (index.json).
+
+The embedding matrix + Ollama client are loaded ONCE at startup (a single
+LibraryContext) and reused for every call. Install with `pip install -e '.[mcp]'`
+and register the `clawlibrary-mcp` entry point with your agent (see README).
+
+Env: CLAWLIBRARY_CONFIG (default config.yaml), OLLAMA_HOST (e.g. http://tank:11434).
+"""
+
+from __future__ import annotations
+
+import os
+
+from .insights_api import (
+    BackendDown,
+    LibraryContext,
+    VaultUnavailable,
+    ask_library,
+    get_insight,
+    list_recent,
+    list_topics,
+    search_vault,
+)
+from .ollama_client import OllamaError
+
+
+def _build_server():
+    from mcp.server.fastmcp import FastMCP  # noqa: PLC0415 — optional [mcp] dependency
+
+    mcp = FastMCP("clawlibrary")
+    ctx = LibraryContext.create(
+        config_path=os.environ.get("CLAWLIBRARY_CONFIG", "config.yaml"),
+        host=os.environ.get("OLLAMA_HOST"),
+    )
+
+    def _safe(fn, *a):
+        try:
+            return fn(ctx, *a)
+        except (VaultUnavailable, BackendDown, OllamaError) as e:
+            return {"error": str(e)}
+
+    @mcp.tool()
+    def search_vault_tool(query: str, k: int = 10) -> list[dict] | dict:
+        """Semantic search across the research vault (papers + video transcripts).
+        Returns scored {title, doi, topic, snippet}."""
+        return _safe(search_vault, query, k)
+
+    @mcp.tool()
+    def ask_library_tool(question: str, k: int = 6) -> dict:
+        """Answer a question grounded in retrieved vault passages, with bracketed
+        citations and a sources list."""
+        return _safe(ask_library, question, k)
+
+    @mcp.tool()
+    def get_insight_tool(doi: str) -> dict:
+        """Structured insight for one paper/video: problem, method, key_findings,
+        relevance, keywords. Accepts a DOI or a youtube:<id> key."""
+        r = _safe(get_insight, doi)
+        return r if r else {"error": f"no insight for doi {doi!r}"}
+
+    @mcp.tool()
+    def list_topics_tool() -> list[dict] | dict:
+        """All topics in the vault with their chunk counts."""
+        return _safe(list_topics)
+
+    @mcp.tool()
+    def list_recent_tool(n: int = 20) -> list[dict] | dict:
+        """The most recently ingested papers/videos."""
+        return _safe(list_recent, n)
+
+    @mcp.resource("clawlibrary://insights")
+    def insights_digest() -> str:
+        """Human-readable insight digest (vault/insights.md)."""
+        path = ctx.root / "insights.md"
+        return path.read_text() if path.exists() else "(no insights yet)"
+
+    @mcp.resource("clawlibrary://catalog")
+    def catalog() -> str:
+        """Full vault catalog (vault/index.json)."""
+        path = ctx.root / "index.json"
+        return path.read_text() if path.exists() else "[]"
+
+    return mcp
+
+
+def main() -> None:
+    _build_server().run()
+
+
+if __name__ == "__main__":
+    main()

clawlibrary/refresh.py

@@ -0,0 +1,401 @@
+"""Incremental refresh: harvest → ingest → extract → summarize → embed.
+
+One command that re-polls every tracked domain (OpenAlex/arXiv topics +
+youtube.yaml channels) and folds only the NEW items into the insight layer. The
+underlying stages are already idempotent (state.json dedup, already_extracted /
+has_summary / store.dois), so this is orchestration: it runs them in order under
+a single lockfile, captures what was newly added, and emits a "what's new"
+digest an agent or human can review.
+
+Each stage is wrapped so a failure is recorded and the chain continues —
+extract/summarize/embed reprocess whatever is on disk, so a harvest hiccup is
+not fatal.
+"""
+
+from __future__ import annotations
+
+import json
+from dataclasses import asdict, dataclass, field
+from datetime import UTC, datetime
+from pathlib import Path
+
+from loguru import logger
+
+from . import pipeline
+from .config import Config, Topic
+from .lock import file_lock
+from .pipeline import _setup_logging, generate_run_id
+
+
+# --------------------------------------------------------------------------- #
+# Result types (mirror pipeline.RunStats style)
+# --------------------------------------------------------------------------- #
+@dataclass
+class StageResult:
+    name: str
+    ran: bool = True          # False when skipped via flag (e.g. --skip-youtube)
+    added: int = 0
+    skipped: int = 0
+    failed: int = 0
+    error: str | None = None  # set if the whole stage raised / was unavailable
+    detail: dict = field(default_factory=dict)
+
+
+@dataclass
+class NewItem:
+    kind: str                 # "paper" | "video"
+    doi: str
+    title: str
+    source: str
+    topic: str = ""
+    summary_one_line: str = ""
+
+
+@dataclass
+class RefreshStats:
+    run_id: str
+    started_at: str
+    finished_at: str | None = None
+    dry_run: bool = False
+    stages: list[StageResult] = field(default_factory=list)
+    new_items: list[NewItem] = field(default_factory=list)
+
+    def stage(self, name: str) -> StageResult | None:
+        return next((s for s in self.stages if s.name == name), None)
+
+    @property
+    def had_error(self) -> bool:
+        return any(s.error for s in self.stages)
+
+
+def _utcnow_iso() -> str:
+    return datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%SZ")
+
+
+# --------------------------------------------------------------------------- #
+# Stages — each reuses existing library functions, never reimplements them
+# --------------------------------------------------------------------------- #
+def stage_papers(
+    config: Config, topics: list[Topic], *, source: str, max_override: int | None,
+    dry_run: bool, emit,
+) -> tuple[StageResult, list[NewItem]]:
+    new_items: list[NewItem] = []
+
+    def on_ev(kind: str, **data):
+        if kind in ("downloaded", "would_download"):
+            new_items.append(NewItem(
+                kind="paper", doi=(data.get("doi") or ""),
+                title=data.get("title", ""), source=data.get("source", "") or "",
+            ))
+        emit(kind, **data)
+
+    stats = pipeline.run(
+        config, topics, dry_run=dry_run, max_override=max_override,
+        source=source, on_event=on_ev,
+    )
+    sr = StageResult(
+        name="papers",
+        added=(stats.would_download if dry_run else stats.downloaded),
+        skipped=stats.skipped_dedup, failed=stats.failed,
+        detail={"skipped_no_pdf": stats.skipped_no_pdf, "suspended": stats.suspended},
+    )
+    return sr, new_items
+
+
+def stage_youtube(
+    config: Config, *, queue_path: str, mode: str, max_override: int | None,
+    dry_run: bool, run_id: str, emit,
+) -> tuple[StageResult, list[NewItem]]:
+    from . import youtube as yt
+
+    sr = StageResult(name="youtube")
+    new_items: list[NewItem] = []
+    try:
+        _defaults, sources = yt.load_queue(queue_path)
+    except (FileNotFoundError, ValueError) as e:
+        sr.ran = False
+        sr.error = str(e)
+        return sr, new_items
+
+    root = Path(config.vault.root)
+    transcribe = yt.make_transcriber(mode=mode)
+    for s in sources:
+        cap = max_override or s.max_videos
+        try:
+            videos = yt.list_source_videos(s.url, max_videos=cap)
+        except Exception as e:  # noqa: BLE001 — one source must not abort the stage
+            sr.failed += 1
+            emit("yt_list_failed", url=s.url, error=str(e))
+            continue
+        for v in videos:
+            try:
+                res = yt.ingest_video(
+                    v["url"], s.topic, root, run_id=run_id,
+                    transcribe_fn=transcribe, dry_run=dry_run,
+                )
+            except Exception as e:  # noqa: BLE001
+                sr.failed += 1
+                emit("failed", title=v.get("title", v.get("id", "")), reason=str(e))
+                continue
+            st = res["status"]
+            if st in ("ingested", "planned"):
+                sr.added += 1
+                if st == "ingested":
+                    new_items.append(NewItem(
+                        kind="video", doi=f"youtube:{res['video_id']}",
+                        title=res.get("title", ""), source="youtube", topic=s.topic,
+                    ))
+                emit("downloaded" if st == "ingested" else "would_download",
+                     title=res.get("title", v.get("title", "")), source="youtube")
+            elif st == "dedup":
+                sr.skipped += 1
+            else:
+                sr.detail[st] = sr.detail.get(st, 0) + 1
+    return sr, new_items
+
+
+def stage_extract(config: Config, *, dry_run: bool, emit) -> StageResult:
+    from .extract import already_extracted, extract_one, iter_pdfs
+
+    root = Path(config.vault.root)
+    pending = [p for p in iter_pdfs(root) if not already_extracted(p)]
+    sr = StageResult(name="extract")
+    if dry_run:
+        sr.detail["pending"] = len(pending)
+        return sr
+    for pdf in pending:
+        try:
+            extract_one(pdf)
+            sr.added += 1
+        except Exception as e:  # noqa: BLE001
+            sr.failed += 1
+            emit("extract_failed", file=pdf.name, error=str(e))
+    return sr
+
+
+def stage_summarize(config: Config, *, model: str, dry_run: bool, emit) -> StageResult:
+    from .extract import iter_documents
+    from .ollama_client import OllamaClient
+    from .summarize import append_insight_once, appended_dois, has_summary, summarize_one
+
+    root = Path(config.vault.root)
+    pending = [d for d in iter_documents(root) if not has_summary(d.with_suffix(".json"))]
+    sr = StageResult(name="summarize")
+    if dry_run:
+        sr.detail["pending"] = len(pending)
+        return sr
+    oc = OllamaClient(text_model=model)
+    if not oc.is_up():
+        sr.error = "ollama not reachable"
+        return sr
+    insights = root / "insights.md"
+    seen = appended_dois(insights)
+    for doc in pending:
+        try:
+            summary = summarize_one(doc, oc)
+        except Exception as e:  # noqa: BLE001 — one doc must not abort the batch
+            sr.failed += 1
+            emit("summarize_failed", file=doc.name, error=str(e))
+            continue
+        if summary is not None:
+            append_insight_once(insights, json.loads(doc.with_suffix(".json").read_text()),
+                                seen=seen)
+            sr.added += 1
+    return sr
+
+
+def stage_embed(config: Config, *, dry_run: bool, emit) -> StageResult:
+    from .embedstore import EmbedStore, add_document, merge
+    from .extract import iter_documents
+    from .ollama_client import OllamaClient
+
+    root = Path(config.vault.root)
+    store = EmbedStore.load(root)
+    seen = store.dois
+    pending: list[tuple[Path, dict, str]] = []
+    for d in iter_documents(root):
+        data = json.loads(d.with_suffix(".json").read_text())
+        doi = data.get("doi") or data.get("id") or d.stem
+        if doi not in seen:
+            pending.append((d, data, doi))
+    sr = StageResult(name="embed")
+    if dry_run:
+        sr.detail["pending"] = len(pending)
+        return sr
+    oc = OllamaClient()
+    if not oc.is_up():
+        sr.error = "ollama not reachable"
+        return sr
+    for d, data, doi in pending:
+        try:
+            vecs, rows = add_document(
+                oc, text=d.read_text(), doi=doi, title=data.get("title", ""),
+                topic=data.get("topic", ""), text_path=str(d),
+            )
+        except Exception as e:  # noqa: BLE001
+            sr.failed += 1
+            emit("embed_failed", file=d.name, error=str(e))
+            continue
+        store = merge(store, vecs, rows)
+        seen.add(doi)
+        sr.added += 1
+        if sr.added % 20 == 0:
+            store.save(root)  # periodic checkpoint
+    store.save(root)
+    return sr
+
+
+# --------------------------------------------------------------------------- #
+# Digest writers — the "what's new" feed
+# --------------------------------------------------------------------------- #
+def write_refresh_log(vault_root: Path, stats: RefreshStats) -> Path:
+    """Append one JSON line to vault/refresh-log.jsonl (the agent-readable stream)."""
+    vault_root.mkdir(parents=True, exist_ok=True)
+    path = vault_root / "refresh-log.jsonl"
+    rec = {
+        "run_id": stats.run_id,
+        "started_at": stats.started_at,
+        "finished_at": stats.finished_at,
+        "stages": [asdict(s) for s in stats.stages],
+        "new_items": [asdict(i) for i in stats.new_items],
+    }
+    with open(path, "a") as f:
+        f.write(json.dumps(rec) + "\n")
+    return path
+
+
+def write_digest_markdown(vault_root: Path, stats: RefreshStats, *, date: str) -> Path:
+    """Write/append vault/digests/{date}.md — a skimmable page of what was added."""
+    ddir = vault_root / "digests"
+    ddir.mkdir(parents=True, exist_ok=True)
+    path = ddir / f"{date}.md"
+    papers = [i for i in stats.new_items if i.kind == "paper"]
+    videos = [i for i in stats.new_items if i.kind == "video"]
+    lines = [
+        f"## Refresh {stats.run_id} — {stats.finished_at}",
+        "",
+        f"**New:** {len(papers)} papers, {len(videos)} videos",
+        "",
+    ]
+    for it in stats.new_items:
+        topic = f"  ·  {it.topic}" if it.topic else ""
+        lines.append(f"- **[{it.kind}]** {it.title}{topic}  ·  `{it.doi}`")
+        if it.summary_one_line:
+            lines.append(f"  - {it.summary_one_line}")
+    if not stats.new_items:
+        lines.append("- _(nothing new this run)_")
+    lines.append("")
+    with open(path, "a") as f:
+        f.write("\n".join(lines) + "\n")
+    return path
+
+
+def _enrich_new_items(vault_root: Path, items: list[NewItem]) -> None:
+    """Backfill topic + one-line summary on new items from their sidecars."""
+    if not items:
+        return
+    index: dict[str, dict] = {}
+    for side in vault_root.glob("*/*.json"):
+        try:
+            data = json.loads(side.read_text())
+        except (OSError, json.JSONDecodeError):
+            continue
+        key = (data.get("doi") or data.get("id") or "").lower()
+        if key:
+            index[key] = data
+    for it in items:
+        data = index.get(it.doi.lower())
+        if not data:
+            continue
+        if not it.topic:
+            it.topic = data.get("topic", "")
+        summ = data.get("summary") or {}
+        it.summary_one_line = (summ.get("problem") or summ.get("relevance") or "")[:300]
+
+
+# --------------------------------------------------------------------------- #
+# Orchestrator
+# --------------------------------------------------------------------------- #
+def refresh(
+    config: Config,
+    topics: list[Topic],
+    *,
+    source: str = "openalex",
+    skip_youtube: bool = False,
+    youtube_queue: str = "youtube.yaml",
+    max_override: int | None = None,
+    no_whisper: bool = False,
+    model: str = "gemma4:E4B",
+    dry_run: bool = False,
+    on_event=None,
+) -> RefreshStats:
+    """Run the full incremental chain under a lockfile. Raises LockError if a
+    refresh is already running. Returns a RefreshStats with per-stage results and
+    the list of newly added items."""
+    run_id = generate_run_id()
+    stats = RefreshStats(run_id=run_id, started_at=_utcnow_iso(), dry_run=dry_run)
+    root = Path(config.vault.root)
+    lock_path = Path(f"{config.state.state_file}.lock")
+    sink_id = _setup_logging(config, run_id)
+
+    def emit(kind: str, **data):
+        logger.bind(run_id=run_id, **data).info(kind)
+        if on_event:
+            on_event(kind, **data)
+
+    try:
+        with file_lock(lock_path):
+            emit("refresh_start", source=source, dry_run=dry_run,
+                 skip_youtube=skip_youtube, no_whisper=no_whisper)
+
+            # 1) papers
+            try:
+                sr, items = stage_papers(
+                    config, topics, source=source, max_override=max_override,
+                    dry_run=dry_run, emit=emit,
+                )
+            except Exception as e:  # noqa: BLE001
+                sr, items = StageResult(name="papers", error=str(e)), []
+                emit("stage_error", stage="papers", error=str(e))
+            stats.stages.append(sr)
+            stats.new_items += items
+
+            # 2) youtube
+            if skip_youtube:
+                stats.stages.append(StageResult(name="youtube", ran=False))
+            else:
+                mode = "captions" if no_whisper else "auto"
+                try:
+                    sr, items = stage_youtube(
+                        config, queue_path=youtube_queue, mode=mode,
+                        max_override=max_override, dry_run=dry_run, run_id=run_id, emit=emit,
+                    )
+                except Exception as e:  # noqa: BLE001
+                    sr, items = StageResult(name="youtube", error=str(e)), []
+                    emit("stage_error", stage="youtube", error=str(e))
+                stats.stages.append(sr)
+                stats.new_items += items
+
+            # 3) extract → 4) summarize → 5) embed
+            for name, thunk in (
+                ("extract", lambda: stage_extract(config, dry_run=dry_run, emit=emit)),
+                ("summarize", lambda: stage_summarize(
+                    config, model=model, dry_run=dry_run, emit=emit)),
+                ("embed", lambda: stage_embed(config, dry_run=dry_run, emit=emit)),
+            ):
+                try:
+                    stats.stages.append(thunk())
+                except Exception as e:  # noqa: BLE001
+                    stats.stages.append(StageResult(name=name, error=str(e)))
+                    emit("stage_error", stage=name, error=str(e))
+
+            stats.finished_at = _utcnow_iso()
+            _enrich_new_items(root, stats.new_items)
+            if not dry_run:
+                write_refresh_log(root, stats)
+                write_digest_markdown(root, stats, date=run_id[:8])
+            emit("refresh_end", new_items=len(stats.new_items),
+                 stages={s.name: s.added for s in stats.stages})
+    finally:
+        logger.remove(sink_id)
+    return stats

clawlibrary/summarize.py

@@ -94,7 +94,7 @@ def summarize_one(pdf_path: Path, client: OllamaClient) -> dict | None:
     return summary
 
 
-def append_insight(insights_path: Path, sidecar_data: dict) -> None:
+def _insight_lines(sidecar_data: dict) -> list[str]:
     s = sidecar_data.get("summary", {})
     title = sidecar_data.get("title", "(untitled)")
     doi = sidecar_data.get("doi", "")
@@ -114,5 +114,56 @@ def append_insight(insights_path: Path, sidecar_data: dict) -> None:
     if s.get("keywords"):
         lines.append(f"- **Keywords:** {', '.join(s['keywords'])}")
     lines.append("")
+    return lines
+
+
+def append_insight(insights_path: Path, sidecar_data: dict) -> None:
+    with open(insights_path, "a") as f:
+        f.write("\n".join(_insight_lines(sidecar_data)) + "\n")
+
+
+_MARKER_PREFIX = "<!-- doi:"
+_MARKER_SUFFIX = "-->"
+
+
+def _insight_key(sidecar_data: dict) -> str:
+    """Stable identity for a document in the digest (matches the embed convention)."""
+    key = sidecar_data.get("doi") or sidecar_data.get("id") or sidecar_data.get("title", "")
+    return key.lower().strip()
+
+
+def appended_dois(insights_path: Path) -> set[str]:
+    """The set of document keys already present in insights.md (via section markers)."""
+    if not insights_path.exists():
+        return set()
+    out: set[str] = set()
+    for line in insights_path.read_text().splitlines():
+        line = line.strip()
+        if line.startswith(_MARKER_PREFIX) and line.endswith(_MARKER_SUFFIX):
+            out.add(line[len(_MARKER_PREFIX) : -len(_MARKER_SUFFIX)].strip())
+    return out
+
+
+def append_insight_once(
+    insights_path: Path, sidecar_data: dict, *, seen: set[str] | None = None
+) -> bool:
+    """Append a digest section only if this document isn't already in the file.
+
+    Writes a hidden ``<!-- doi: {key} -->`` marker so future runs detect it.
+    `seen` is an optional in-memory set the caller maintains across a batch (it is
+    updated in place); when omitted it is seeded from the file once. Returns True
+    if a section was written, False if skipped as a duplicate.
+    """
+    key = _insight_key(sidecar_data)
+    if not key:  # no identity — fall back to unconditional append
+        append_insight(insights_path, sidecar_data)
+        return True
+    if seen is None:
+        seen = appended_dois(insights_path)
+    if key in seen:
+        return False
+    lines = [f"{_MARKER_PREFIX} {key} {_MARKER_SUFFIX}", *_insight_lines(sidecar_data)]
     with open(insights_path, "a") as f:
         f.write("\n".join(lines) + "\n")
+    seen.add(key)
+    return True

pyproject.toml

@@ -34,9 +34,15 @@ youtube = [
     "yt-dlp>=2024.1",
     "faster-whisper>=1.0",
 ]
+# MCP server: exposes the insight vault to agents over stdio. Install on the
+# serving node (tank) to enable `clawlibrary-mcp`.
+mcp = [
+    "mcp>=1.2",
+]
 
 [project.scripts]
 clawlibrary = "clawlibrary.__main__:main"
+clawlibrary-mcp = "clawlibrary.mcp_server:main"
 
 [tool.hatch.build.targets.wheel]
 packages = ["clawlibrary"]

scripts/refresh.sh

@@ -0,0 +1,29 @@
+#!/usr/bin/env bash
+# Scheduled incremental refresh for ClawLibrary (run by cron).
+# Harvests new papers (OpenAlex + arXiv) and new videos (youtube.yaml, Whisper
+# fallback ON), then extracts/summarizes/embeds only the new items. Safe to run
+# concurrently — the app holds a lockfile (state.json.lock).
+#
+# Install (on the GPU node `tank`):
+#   crontab -e
+#   30 3 * * *  /home/osobh/clawlibrary/scripts/refresh.sh
+set -euo pipefail
+
+CLAW_HOME="${CLAW_HOME:-$HOME/clawlibrary}"
+cd "$CLAW_HOME"
+# shellcheck disable=SC1091
+source .venv/bin/activate
+
+mkdir -p logs
+LOG="logs/refresh-$(date +%F).log"
+
+{
+  echo "===== refresh $(date -u +%FT%TZ) ====="
+  python -m clawlibrary refresh --source both
+  rc=$?
+  echo "----- refresh exit: ${rc} -----"
+  # Pick up the freshly-written embeddings in any running MCP server so agents
+  # query current data (no-op if the server isn't running under systemd --user).
+  systemctl --user restart clawlibrary-mcp 2>/dev/null || true
+  exit "${rc}"
+} >> "${LOG}" 2>&1

tests/test_cli.py

@@ -82,3 +82,79 @@ def test_cli_export_csv_and_json(project, tmp_path):
 def test_cli_missing_config_returns_2(tmp_path, monkeypatch):
     monkeypatch.chdir(tmp_path)
     assert cli.main(["stats"]) == 2
+
+
+def test_cli_refresh_dispatches(project, monkeypatch):
+    import clawlibrary.refresh as R
+
+    def fake_refresh(config, topics, **kwargs):
+        assert kwargs["dry_run"] is True
+        return R.RefreshStats(run_id="20260604T000000-abc", started_at="t")
+
+    monkeypatch.setattr(R, "refresh", fake_refresh)
+    assert cli.main(["refresh", "--dry-run"]) == 0
+
+
+def test_cli_refresh_lock_returns_3(project, monkeypatch):
+    import clawlibrary.lock as L
+    import clawlibrary.refresh as R
+
+    def boom(*a, **k):
+        raise L.LockError("held")
+
+    monkeypatch.setattr(R, "refresh", boom)
+    assert cli.main(["refresh"]) == 3
+
+
+# ---- JSON CLI surface (agent-facing) ---------------------------------------
+import types  # noqa: E402
+
+
+def _fake_ctx():
+    return types.SimpleNamespace(
+        client=types.SimpleNamespace(text_model="gemma4:E4B"), root="vault",
+    )
+
+
+def _patch_api(monkeypatch, **fns):
+    import clawlibrary.insights_api as API
+
+    monkeypatch.setattr(API.LibraryContext, "create", lambda *a, **k: _fake_ctx())
+    for name, fn in fns.items():
+        monkeypatch.setattr(API, name, fn)
+
+
+def test_cli_search_json(monkeypatch, capsys):
+    _patch_api(monkeypatch, search_vault=lambda ctx, q, k=10: [
+        {"score": 0.9, "title": "T", "doi": "10.1/x", "topic": "t", "snippet": "s"}])
+    assert cli.main(["search", "rust", "--json"]) == 0
+    out = json.loads(capsys.readouterr().out)
+    assert out[0]["doi"] == "10.1/x"
+
+
+def test_cli_ask_json(monkeypatch, capsys):
+    _patch_api(monkeypatch, ask_library=lambda ctx, q, k=6: {
+        "answer": "yes [1]", "sources": [{"n": 1, "title": "T", "doi": "10.1/x", "topic": "t"}]})
+    assert cli.main(["ask", "is rust safe?", "--json"]) == 0
+    out = json.loads(capsys.readouterr().out)
+    assert out["answer"] == "yes [1]" and out["sources"][0]["n"] == 1
+
+
+def test_cli_insight_json(monkeypatch, capsys):
+    _patch_api(monkeypatch, get_insight=lambda ctx, doi: {
+        "doi": doi, "title": "T", "summary": {"problem": "P"}})
+    assert cli.main(["insight", "10.1/x", "--json"]) == 0
+    out = json.loads(capsys.readouterr().out)
+    assert out["summary"]["problem"] == "P"
+
+
+def test_cli_search_backend_down(monkeypatch, capsys):
+    from clawlibrary.insights_api import BackendDown
+
+    def boom(ctx, q, k=10):
+        raise BackendDown("ollama not reachable")
+
+    _patch_api(monkeypatch, search_vault=boom)
+    assert cli.main(["search", "rust", "--json"]) == 2
+    out = json.loads(capsys.readouterr().out)
+    assert "error" in out

tests/test_insights_api.py

@@ -0,0 +1,104 @@
+import json
+
+import numpy as np
+import pytest
+
+from clawlibrary.embedstore import EmbedStore
+from clawlibrary.insights_api import (
+    BackendDown,
+    LibraryContext,
+    VaultUnavailable,
+    ask_library,
+    get_insight,
+    list_recent,
+    list_topics,
+    search_vault,
+)
+
+
+class FakeClient:
+    def __init__(self, up=True):
+        self.up = up
+        self.text_model = "gemma4:E4B"
+
+    def is_up(self):
+        return self.up
+
+    def embed_query(self, text):
+        return [1.0, 0.0]  # aligns with the first stored vector
+
+    def generate(self, prompt, **kw):
+        return '{"answer": "Rust is memory safe [1]."}'
+
+
+def _ctx(tmp_path, *, up=True, empty=False):
+    root = tmp_path / "vault"
+    (root / "t").mkdir(parents=True)
+    side = root / "t" / "p.json"
+    side.write_text(json.dumps({
+        "doi": "10.1/a", "title": "A", "topic": "rust", "year": 2024, "source": "arxiv",
+        "authors": ["Z"], "word_count": 3,
+        "summary": {"problem": "P", "method": "M", "key_findings": ["k"],
+                    "relevance": "R", "keywords": ["x"]},
+    }))
+    txt = root / "t" / "p.txt"
+    txt.write_text("rust memory safety")
+    # index.json for list_recent
+    (root / "index.json").write_text(json.dumps([
+        {"doi": "10.1/a", "title": "A", "topic": "rust", "year": 2024,
+         "source": "arxiv", "downloaded_at": "2026-06-01T00:00:00Z"},
+    ]))
+    if empty:
+        store = EmbedStore.empty()
+    else:
+        vectors = np.array([[1.0, 0.0], [0.0, 1.0]], dtype=np.float32)
+        meta = [
+            {"doi": "10.1/a", "title": "A", "topic": "rust", "chunk": 0,
+             "snippet": "rust memory", "text_path": str(txt)},
+            {"doi": "10.1/b", "title": "B", "topic": "quantum", "chunk": 0,
+             "snippet": "qubit", "text_path": str(root / "t" / "missing.txt")},
+        ]
+        store = EmbedStore(vectors=vectors, meta=meta)
+    return LibraryContext(root=root, store=store, client=FakeClient(up=up))
+
+
+def test_search_vault_shape(tmp_path):
+    ctx = _ctx(tmp_path)
+    rows = search_vault(ctx, "rust memory", k=2)
+    assert rows and rows[0]["doi"] == "10.1/a"
+    assert set(rows[0]) == {"score", "title", "doi", "topic", "snippet"}
+
+
+def test_ask_library_shape(tmp_path):
+    ctx = _ctx(tmp_path)
+    res = ask_library(ctx, "Is rust safe?", k=2)
+    assert "[1]" in res["answer"]
+    assert res["sources"] and set(res["sources"][0]) == {"n", "title", "doi", "topic"}
+
+
+def test_get_insight(tmp_path):
+    ctx = _ctx(tmp_path)
+    data = get_insight(ctx, "10.1/a")
+    assert data["summary"]["problem"] == "P"
+    assert data["title"] == "A"
+    assert get_insight(ctx, "10.1/nope") is None
+
+
+def test_list_topics_and_recent(tmp_path):
+    ctx = _ctx(tmp_path)
+    topics = list_topics(ctx)
+    assert {t["topic"] for t in topics} == {"rust", "quantum"}
+    recent = list_recent(ctx, n=5)
+    assert recent[0]["doi"] == "10.1/a"
+
+
+def test_vault_unavailable(tmp_path):
+    ctx = _ctx(tmp_path, empty=True)
+    with pytest.raises(VaultUnavailable):
+        search_vault(ctx, "anything")
+
+
+def test_backend_down(tmp_path):
+    ctx = _ctx(tmp_path, up=False)
+    with pytest.raises(BackendDown):
+        ask_library(ctx, "anything")

tests/test_lock.py

@@ -0,0 +1,32 @@
+import json
+
+import pytest
+
+from clawlibrary.lock import LockError, file_lock
+
+
+def test_file_lock_basic(tmp_path):
+    p = tmp_path / "state.json.lock"
+    with file_lock(p):
+        assert p.exists()
+        info = json.loads(p.read_text())
+        assert "pid" in info and "host" in info
+    assert not p.exists()  # released on exit
+
+
+def test_file_lock_contended_raises(tmp_path):
+    p = tmp_path / "state.json.lock"
+    with file_lock(p):
+        with pytest.raises(LockError):
+            with file_lock(p):
+                pass
+
+
+def test_file_lock_breaks_stale(tmp_path):
+    p = tmp_path / "state.json.lock"
+    # A lock from 1970 on another host => stale, should be broken and re-acquired.
+    p.write_text(json.dumps({"pid": 1, "host": "otherhost", "started_at_epoch": 1.0}))
+    with file_lock(p, stale_after_seconds=1):
+        assert p.exists()
+        assert json.loads(p.read_text())["host"] != "otherhost"
+    assert not p.exists()

tests/test_mcp_server.py

@@ -0,0 +1,28 @@
+"""Smoke test for the MCP server wiring.
+
+Skipped where the optional `mcp` package isn't installed (e.g. the dev box);
+runs on the serving node where `pip install -e '.[mcp]'` has been done.
+"""
+
+import pytest
+
+from clawlibrary.embedstore import EmbedStore
+
+
+def test_build_server_smoke(tmp_path, monkeypatch):
+    pytest.importorskip("mcp")
+    import clawlibrary.mcp_server as M
+    from clawlibrary.insights_api import LibraryContext
+
+    class _Client:
+        text_model = "gemma4:E4B"
+
+        def is_up(self):
+            return True
+
+    fake = LibraryContext(root=tmp_path, store=EmbedStore.empty(), client=_Client())
+    monkeypatch.setattr(M.LibraryContext, "create", lambda *a, **k: fake)
+
+    server = M._build_server()
+    assert server is not None
+    assert hasattr(server, "run")

tests/test_refresh.py

@@ -0,0 +1,114 @@
+import pytest
+
+import clawlibrary.lock as L
+import clawlibrary.refresh as R
+from clawlibrary.config import Config, LoggingConfig, StateConfig, Topic, VaultConfig
+
+
+def _cfg(tmp_path) -> tuple[Config, list[Topic]]:
+    config = Config(
+        vault=VaultConfig(root=tmp_path / "vault"),
+        state=StateConfig(state_file=tmp_path / "state.json"),
+        logging=LoggingConfig(log_dir=tmp_path / "logs"),
+    )
+    return config, [Topic(query="rust")]
+
+
+def _patch_stages(monkeypatch, calls, *, extras=None):
+    """Replace the five stage fns with fakes that record call order."""
+    extras = {} if extras is None else extras
+
+    def papers(config, topics, *, source, max_override, dry_run, emit):
+        calls.append("papers")
+        return (R.StageResult(name="papers", added=2),
+                [R.NewItem(kind="paper", doi="10.1/x", title="X", source="arxiv")])
+
+    def youtube(config, *, queue_path, mode, max_override, dry_run, run_id, emit):
+        calls.append("youtube")
+        extras["yt_mode"] = mode
+        return (R.StageResult(name="youtube", added=1),
+                [R.NewItem(kind="video", doi="youtube:abc", title="V", source="youtube")])
+
+    def extract(config, *, dry_run, emit):
+        calls.append("extract")
+        return R.StageResult(name="extract", added=2)
+
+    def summarize(config, *, model, dry_run, emit):
+        calls.append("summarize")
+        return R.StageResult(name="summarize", added=3)
+
+    def embed(config, *, dry_run, emit):
+        calls.append("embed")
+        return R.StageResult(name="embed", added=3)
+
+    monkeypatch.setattr(R, "stage_papers", extras.get("papers", papers))
+    monkeypatch.setattr(R, "stage_youtube", extras.get("youtube", youtube))
+    monkeypatch.setattr(R, "stage_extract", extras.get("extract", extract))
+    monkeypatch.setattr(R, "stage_summarize", extras.get("summarize", summarize))
+    monkeypatch.setattr(R, "stage_embed", extras.get("embed", embed))
+
+
+def test_refresh_chains_stages_in_order(tmp_path, monkeypatch):
+    config, topics = _cfg(tmp_path)
+    calls: list[str] = []
+    _patch_stages(monkeypatch, calls)
+    stats = R.refresh(config, topics, dry_run=False)
+    assert calls == ["papers", "youtube", "extract", "summarize", "embed"]
+    assert [s.name for s in stats.stages] == ["papers", "youtube", "extract", "summarize", "embed"]
+    assert {i.kind for i in stats.new_items} == {"paper", "video"}
+    # digest + event stream written when not a dry run
+    assert (tmp_path / "vault" / "refresh-log.jsonl").exists()
+    assert list((tmp_path / "vault" / "digests").glob("*.md"))
+    assert not stats.had_error
+
+
+def test_refresh_skip_youtube(tmp_path, monkeypatch):
+    config, topics = _cfg(tmp_path)
+    calls: list[str] = []
+    _patch_stages(monkeypatch, calls)
+    stats = R.refresh(config, topics, skip_youtube=True, dry_run=True)
+    assert "youtube" not in calls
+    yt = stats.stage("youtube")
+    assert yt is not None and yt.ran is False
+
+
+def test_refresh_no_whisper_forces_captions(tmp_path, monkeypatch):
+    config, topics = _cfg(tmp_path)
+    calls: list[str] = []
+    extras: dict = {}
+    _patch_stages(monkeypatch, calls, extras=extras)
+    R.refresh(config, topics, no_whisper=True, dry_run=True)
+    assert extras["yt_mode"] == "captions"
+
+
+def test_refresh_dry_run_writes_no_files(tmp_path, monkeypatch):
+    config, topics = _cfg(tmp_path)
+    _patch_stages(monkeypatch, [])
+    R.refresh(config, topics, dry_run=True)
+    assert not (tmp_path / "vault" / "refresh-log.jsonl").exists()
+    assert not (tmp_path / "vault" / "digests").exists()
+
+
+def test_refresh_stage_failure_continues(tmp_path, monkeypatch):
+    config, topics = _cfg(tmp_path)
+    calls: list[str] = []
+
+    def boom(config, *, dry_run, emit):
+        raise RuntimeError("disk full")
+
+    _patch_stages(monkeypatch, calls, extras={"extract": boom})
+    stats = R.refresh(config, topics, dry_run=True)
+    ex = stats.stage("extract")
+    assert ex is not None and ex.error and "disk full" in ex.error
+    # chain still reached the later stages
+    assert "summarize" in calls and "embed" in calls
+    assert stats.had_error
+
+
+def test_refresh_lock_contention(tmp_path, monkeypatch):
+    config, topics = _cfg(tmp_path)
+    _patch_stages(monkeypatch, [])
+    lock_path = f"{config.state.state_file}.lock"
+    with L.file_lock(lock_path):
+        with pytest.raises(L.LockError):
+            R.refresh(config, topics, dry_run=True)

tests/test_summarize_dedup.py

@@ -0,0 +1,37 @@
+from clawlibrary.summarize import append_insight_once, appended_dois
+
+
+def _sidecar(doi="10.1/x"):
+    return {
+        "doi": doi, "title": "T", "topic": "t",
+        "summary": {"problem": "P", "method": "M", "key_findings": ["a"],
+                    "relevance": "R", "keywords": ["k"]},
+    }
+
+
+def test_append_insight_once_dedups(tmp_path):
+    p = tmp_path / "insights.md"
+    assert append_insight_once(p, _sidecar()) is True
+    assert append_insight_once(p, _sidecar()) is False  # second time: skipped
+    text = p.read_text()
+    assert text.count("<!-- doi: 10.1/x -->") == 1
+    assert text.count("## T") == 1
+
+
+def test_appended_dois_parses_markers(tmp_path):
+    p = tmp_path / "insights.md"
+    append_insight_once(p, _sidecar("10.1/a"))
+    append_insight_once(p, _sidecar("10.1/b"))
+    assert appended_dois(p) == {"10.1/a", "10.1/b"}
+
+
+def test_append_insight_once_uses_seen_set(tmp_path):
+    p = tmp_path / "insights.md"
+    seen: set[str] = set()
+    assert append_insight_once(p, _sidecar("10.1/x"), seen=seen) is True
+    assert "10.1/x" in seen
+    assert append_insight_once(p, _sidecar("10.1/x"), seen=seen) is False
+
+
+def test_appended_dois_missing_file(tmp_path):
+    assert appended_dois(tmp_path / "nope.md") == set()