Deploy Awesome-WAM long-context Q&A app

.gitattributes

@@ -33,3 +33,4 @@ saved_model/**/* filter=lfs diff=lfs merge=lfs -text
 *.zip filter=lfs diff=lfs merge=lfs -text
 *.zst filter=lfs diff=lfs merge=lfs -text
 *tfevents* filter=lfs diff=lfs merge=lfs -text
+data/wam.db filter=lfs diff=lfs merge=lfs -text

README.md

@@ -1,20 +1,15 @@
 ---
 title: Awesome Embodied MM
-emoji: 🚀
-colorFrom: red
-colorTo: red
-sdk: docker
-app_port: 8501
-tags:
-- streamlit
+emoji: 🤖
+colorFrom: blue
+colorTo: indigo
+sdk: streamlit
+app_file: src/wam/webapp/app.py
 pinned: false
-short_description: Interactive QA about Awesome-Embodied&MM
-license: apache-2.0
+short_description: Q&A over a curated World Action Models knowledge base
 ---
 
-# Welcome to Streamlit!
+# Awesome-WAM Knowledge Base
 
-Edit `/src/streamlit_app.py` to customize this app to your heart's desire. :heart:
-
-If you have any questions, checkout our [documentation](https://docs.streamlit.io) and [community
-forums](https://discuss.streamlit.io).
+Ask questions about World Action Models (world models, VLA, video-generation, robot
+foundation models) — answers are grounded in a curated corpus with citations.

config/config.yaml

@@ -0,0 +1,136 @@
+# Awesome-WAM system configuration.
+# Everything domain/tuning-related lives here so behavior changes need no code edits.
+
+provider:
+  # OpenRouter is OpenAI-compatible. Key read from this env var.
+  base_url: "https://openrouter.ai/api/v1"
+  api_key_env: "OPENROUTER_API_KEY"
+  # Sent as OpenRouter attribution headers (optional but recommended).
+  http_referer: "https://github.com/your-org/Awesome-WAM"
+  app_title: "Awesome-WAM"
+
+models:
+  # Tiered routing. Cheap = filtering/first-pass (high volume); mid = analysis; strong =
+  # scoring. Chosen via a blind Opus-4.8 ranking of anonymized outputs on real papers (see
+  # scripts/bench_models.py). Latency is NOT a selection criterion for these batch stages
+  # (only interactive Q&A cares about latency, and its model is deferred):
+  #   - cheap: deepseek-v4-flash — #1 on the scoring task (best skeptical N/A calibration,
+  #     no fabrication) AND cheapest by ~50x; ideal for the high-volume relevance filter.
+  #   - mid/strong: glm-5.1 — #1 overall in the blind ranking, low cost; its only weakness
+  #     (27s latency) is irrelevant for batch work.
+  # Excluded: deepseek-v4-pro (abdicated scoring -> all-N/A), glm-4.7-flash (4/6 parse
+  # failures), gemini-3.1-flash-lite (fabricated scores).
+  tiers:
+    cheap: "deepseek/deepseek-v4-flash"
+    mid: "z-ai/glm-5.1"
+    strong: "z-ai/glm-5.1"
+    xstrong: "openai/gpt-5.5"   # top tier for the most quality-critical stage (scoring)
+  # Optional per-stage overrides (stage name -> tier). Falls back to the tier passed in code.
+  stage_tiers:
+    filter: cheap
+    summarize: cheap
+    analyze: mid
+    extract: cheap      # deepseek-v4-flash: fast + JSON-reliable + faithful for table extraction
+    score: xstrong      # flagship two-layer scoring -> gpt-5.5
+    innovation: mid
+    front_summary: strong
+    qa: strong          # TODO: pick a low-latency Q&A model in Phase 7 (deferred)
+  defaults:
+    temperature: 0.2
+    max_tokens: 4096
+  # USD per 1,000,000 tokens, {input, output}. Used for cost tracking/logging.
+  cost_per_million:
+    "deepseek/deepseek-v4-flash": {input: 0.098, output: 0.197}
+    "z-ai/glm-5.1": {input: 0.98, output: 3.08}
+    "openai/gpt-5.5": {input: 5.00, output: 30.00}
+    # alternates considered: gemini-3.5-flash {1.50, 9.00}, claude-sonnet-4.6 {3.00, 15.00}
+
+constants:
+  request_timeout: 90        # seconds per LLM call
+  max_retries: 4
+  retry_backoff: 3.0         # seconds, exponential base
+  lookback_days: 60          # ignore papers older than this on fetch
+  relevance_threshold: 0.5   # below this -> dropped
+  analyze_cap: 40            # max papers fully analyzed per run (cost guard)
+
+logging:
+  level: "INFO"              # console level; debug file sink enabled via --debug/WAM_DEBUG
+  log_dir: "logs"
+
+email:
+  feature_threshold: 6.0        # core papers with weighted_total >= this get a detailed card
+  max_featured: 8               # cap on featured (detailed) papers
+  max_grouped: 40               # cap on the grouped lower tier
+  group_lower_by_direction: true  # group the lower tier by research direction (else flat list)
+
+paths:
+  db: "data/wam.db"
+  pdf_cache: "cache/pdfs"
+  index_dir: "data/index"
+
+# Two-layer scoring weights. Top-4 WAM metrics weighted 2x; computed over non-N/A metrics.
+scoring:
+  general_weights:
+    novelty: 1.0
+    soundness: 1.0
+    impact: 1.0
+  wam_weights:
+    generalist: 2.0
+    inference_speed: 2.0
+    specialist: 2.0
+    inference_cost: 2.0
+    trustworthiness: 1.0
+    collaborative: 1.0
+    controlled_generation: 1.0
+    other: 1.0
+
+# Source + search config is consumed in Phase 2 (fetch).
+sources:
+  arxiv:
+    enabled: true
+    categories: ["cs.RO", "cs.AI", "cs.LG", "cs.CV"]
+    max_results: 300
+  semantic_scholar:
+    enabled: true
+    api_key_env: "SEMANTIC_SCHOLAR_API_KEY"   # optional; higher rate limits if set
+  papers_with_code:
+    enabled: true
+  news:
+    enabled: true
+    feeds:
+      - "https://www.therobotreport.com/feed/"
+      - "https://spectrum.ieee.org/feeds/topic/robotics.rss"
+      - "https://huggingface.co/blog/feed.xml"
+
+# Search keyword sets. The LLM filter makes the final core/adjacent/drop call,
+# but these seed the source queries (broad net by design).
+keywords:
+  core:        # World Action Models proper
+    - "world action model"
+    - "action world model"
+    - "embodied foundation model"
+    - "robot foundation model"
+    - "generalist robot policy"
+  adjacent:    # transferable techniques — kept in the innovation track
+    - "vision-language-action"
+    - "world model"
+    - "video generation"
+    - "action-conditioned video"
+    - "interactive world model"
+
+qa:
+  # Knowledge-base Q&A is long-context (no embeddings/vector DB): a compact pack of
+  # paper summaries + scores + leaderboard + authors + trends is built from the DB and
+  # passed to the qa-tier model. See stage_tiers.qa for the model.
+  max_papers: 400            # cap papers in the pack (most-scored first) to bound context
+  include_dropped: true      # include filtered-out papers (title-only) so the KB still covers them
+
+trends:
+  sim_threshold: 0.62        # cosine edge threshold for the direction graph
+  min_front_size: 3          # ignore clusters smaller than this
+  window_days: 30            # momentum = recent window vs prior window
+
+people:
+  min_papers: 2              # influential if on >= N tracked papers
+  min_citations: 2000        # (reserved) citation-based influence via Semantic Scholar
+  max_per_run: 30            # cap influential authors processed per run (most prolific first)

config/profile.md

@@ -0,0 +1,45 @@
+# WAM Research Interest Profile
+
+This profile is read verbatim by the relevance-filter and scoring prompts. It defines
+what "World Action Models" means for this project and what we care about. Edit freely —
+no code changes needed.
+
+## What counts as a World Action Model (WAM)
+
+A WAM is a learned model that maps perception + goals to **actions** in the physical or
+simulated world, often grounded in a learned **world model** (predictive dynamics). The
+umbrella includes:
+
+- Generalist robot/agent policies and embodied foundation models
+- Vision-Language-Action (VLA) models that output actions
+- World models / action-conditioned video & dynamics models used for planning or control
+- Models that unify perception, prediction, and action for embodied agents
+
+## Two relevance tracks
+
+- **core** — the paper *is* a WAM (or directly proposes/evaluates one). Gets the full
+  two-layer rubric and benchmark extraction.
+- **adjacent** — VLA / world model / video-generation work that is *not* a WAM itself but
+  carries a technique plausibly **transferable to WAM**. Gets an innovation note only
+  (key idea + why it could transfer); no rubric scores.
+- **drop** — unrelated to the above.
+
+## What we care most about (drives WAM-specific scoring)
+
+Top-4 (weighted 2x):
+1. **inference_speed** — real-time or faster control; latency/throughput on stated hardware.
+2. **generalist** — generalizes across tasks, benchmarks, and especially *embodiments*.
+3. **specialist** — extreme accuracy/robustness on a specific task or task set.
+4. **inference_cost** — compute/$ to run; small models that punch above their weight.
+
+Also tracked (weighted 1x): trustworthiness/safety, collaborative (multi-robot control),
+controlled/steerable generation of videos or actions, and other features (e.g. async
+inference, streaming, on-device).
+
+## Scoring guidance
+
+- Score each metric 0–10 from evidence in the paper; use **"N/A"** when the paper does not
+  address a metric — do not guess or penalize with a 0.
+- Be **skeptical of self-reported numbers**. Note the dataset a model was trained/finetuned
+  on; the same model name on a different dataset is a different system.
+- Reward reproducibility (released code/data/weights) under `impact`.

data/wam.db

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:f7e93e5fdd9460355f30b196fe25fbf8bcf4b0f2fc6e40354e4ad6a797175324
+size 1957888

requirements.txt

@@ -1,3 +1,10 @@
-altair
-pandas
-streamlit
+# Runtime deps for the Hugging Face Space (Streamlit Q&A web app).
+# The daily pipeline uses pyproject (pip install -e .); this file is for the Space.
+openai>=1.40.0
+pydantic>=2.7.0
+pyyaml>=6.0
+python-dotenv>=1.0.0
+requests>=2.31.0
+networkx>=3.2
+streamlit>=1.36.0
+# No embeddings/vector DB: Q&A is long-context over a knowledge pack built from data/wam.db.

src/awesome_wam.egg-info/PKG-INFO

@@ -0,0 +1,24 @@
+Metadata-Version: 2.4
+Name: awesome-wam
+Version: 0.1.0
+Summary: Daily intelligence system for World Action Models (WAM) research: search, summarize, analyze, score, digest, and a RAG knowledge base.
+Requires-Python: >=3.10
+Description-Content-Type: text/markdown
+Requires-Dist: openai>=1.40.0
+Requires-Dist: pydantic>=2.7.0
+Requires-Dist: pyyaml>=6.0
+Requires-Dist: python-dotenv>=1.0.0
+Requires-Dist: requests>=2.31.0
+Requires-Dist: feedparser>=6.0.0
+Requires-Dist: tenacity>=8.2.0
+Requires-Dist: PyMuPDF>=1.24.0
+Requires-Dist: networkx>=3.2
+Provides-Extra: rag
+Requires-Dist: sentence-transformers>=3.0.0; extra == "rag"
+Requires-Dist: faiss-cpu>=1.8.0; extra == "rag"
+Requires-Dist: streamlit>=1.36.0; extra == "rag"
+Provides-Extra: dev
+Requires-Dist: pytest>=8.0.0; extra == "dev"
+Requires-Dist: ruff>=0.5.0; extra == "dev"
+
+# Awesome-WAM

src/awesome_wam.egg-info/SOURCES.txt

@@ -0,0 +1,15 @@
+README.md
+pyproject.toml
+src/awesome_wam.egg-info/PKG-INFO
+src/awesome_wam.egg-info/SOURCES.txt
+src/awesome_wam.egg-info/dependency_links.txt
+src/awesome_wam.egg-info/requires.txt
+src/awesome_wam.egg-info/top_level.txt
+src/wam/__init__.py
+src/wam/config.py
+src/wam/logging.py
+src/wam/llm/__init__.py
+src/wam/llm/client.py
+src/wam/store/__init__.py
+src/wam/store/db.py
+src/wam/store/schema.sql

src/awesome_wam.egg-info/dependency_links.txt

@@ -0,0 +1 @@
+

src/awesome_wam.egg-info/requires.txt

@@ -0,0 +1,18 @@
+openai>=1.40.0
+pydantic>=2.7.0
+pyyaml>=6.0
+python-dotenv>=1.0.0
+requests>=2.31.0
+feedparser>=6.0.0
+tenacity>=8.2.0
+PyMuPDF>=1.24.0
+networkx>=3.2
+
+[dev]
+pytest>=8.0.0
+ruff>=0.5.0
+
+[rag]
+sentence-transformers>=3.0.0
+faiss-cpu>=1.8.0
+streamlit>=1.36.0

src/awesome_wam.egg-info/top_level.txt

@@ -0,0 +1 @@
+wam

src/wam/__init__.py

@@ -0,0 +1,3 @@
+"""Awesome-WAM: daily intelligence system for World Action Models research."""
+
+__version__ = "0.1.0"

src/wam/config.py

@@ -0,0 +1,86 @@
+"""Configuration loading.
+
+Loads ``config/config.yaml`` plus the human-readable ``config/profile.md`` and resolves
+the project root. Environment variables (e.g. API keys) are loaded from a local ``.env``
+when present so dev runs work without exporting anything.
+"""
+
+from __future__ import annotations
+
+import os
+from functools import lru_cache
+from pathlib import Path
+from typing import Any
+
+import yaml
+
+try:  # optional: only needed for local dev convenience
+    from dotenv import load_dotenv
+except ImportError:  # pragma: no cover
+    load_dotenv = None
+
+
+def project_root() -> Path:
+    """Repo root = two levels up from this file (src/wam/config.py)."""
+    return Path(__file__).resolve().parents[2]
+
+
+class Config:
+    """Thin wrapper over the parsed config dict with dotted-path access."""
+
+    def __init__(self, data: dict[str, Any], root: Path):
+        self._data = data
+        self.root = root
+
+    def get(self, path: str, default: Any = None) -> Any:
+        """Fetch a nested value by dotted path, e.g. ``config.get("models.tiers.cheap")``."""
+        node: Any = self._data
+        for part in path.split("."):
+            if not isinstance(node, dict) or part not in node:
+                return default
+            node = node[part]
+        return node
+
+    def __getitem__(self, key: str) -> Any:
+        return self._data[key]
+
+    @property
+    def data(self) -> dict[str, Any]:
+        return self._data
+
+    # --- convenience resolvers -------------------------------------------------
+    def path(self, key: str) -> Path:
+        """Resolve a configured relative path (under ``paths.*``) against the repo root."""
+        rel = self.get(f"paths.{key}")
+        if rel is None:
+            raise KeyError(f"paths.{key} not configured")
+        return (self.root / rel).resolve()
+
+    def require_env(self, env_var: str) -> str:
+        val = os.environ.get(env_var)
+        if not val:
+            raise RuntimeError(f"Required environment variable {env_var} is not set")
+        return val
+
+    def profile_text(self) -> str:
+        """The research-interest profile fed to relevance/scoring prompts."""
+        return (self.root / "config" / "profile.md").read_text(encoding="utf-8")
+
+    def model_for_stage(self, stage: str, fallback_tier: str = "mid") -> str:
+        """Resolve the OpenRouter model id for a pipeline stage via stage_tiers -> tiers."""
+        tier = self.get(f"models.stage_tiers.{stage}", fallback_tier)
+        model = self.get(f"models.tiers.{tier}")
+        if not model:
+            raise KeyError(f"No model configured for tier '{tier}'")
+        return model
+
+
+@lru_cache(maxsize=1)
+def load_config(config_path: str | None = None) -> Config:
+    """Load and cache the project config. Also loads ``.env`` if available."""
+    root = project_root()
+    if load_dotenv is not None:
+        load_dotenv(root / ".env")
+    path = Path(config_path) if config_path else root / "config" / "config.yaml"
+    data = yaml.safe_load(path.read_text(encoding="utf-8"))
+    return Config(data, root)

src/wam/llm/__init__.py

@@ -0,0 +1,3 @@
+from wam.llm.client import LLMClient
+
+__all__ = ["LLMClient"]

src/wam/llm/client.py

@@ -0,0 +1,191 @@
+"""OpenRouter LLM client with tiered routing, validated JSON output, and cost tracking.
+
+OpenRouter exposes an OpenAI-compatible API, so we drive it with the ``openai`` SDK pointed
+at the OpenRouter base URL. Three usage paths:
+
+- :meth:`complete` — free-form text.
+- :meth:`complete_json` — forces a JSON object, validates it against a Pydantic model, and
+  retries with the validation error fed back to the model on malformed output.
+
+Every call logs a structured record (tier, model, tokens, latency, cost) and updates the
+shared :data:`wam.logging.COST` tracker.
+"""
+
+from __future__ import annotations
+
+import json
+import re
+import time
+from typing import Type, TypeVar
+
+from openai import (APIConnectionError, APITimeoutError, APIStatusError, OpenAI,
+                    RateLimitError)
+from pydantic import BaseModel, ValidationError
+from tenacity import retry, retry_if_exception, stop_after_attempt, wait_exponential
+
+from wam.config import Config, load_config
+from wam.logging import COST, get_logger
+
+log = get_logger("llm")
+T = TypeVar("T", bound=BaseModel)
+
+
+class LLMError(RuntimeError):
+    pass
+
+
+def _is_transient(exc: BaseException) -> bool:
+    """Retry only on rate limits, timeouts, connection drops, and 5xx — not 4xx."""
+    if isinstance(exc, (RateLimitError, APITimeoutError, APIConnectionError)):
+        return True
+    if isinstance(exc, APIStatusError):
+        return exc.status_code >= 500
+    return False
+
+
+class LLMClient:
+    def __init__(self, config: Config | None = None):
+        self.cfg = config or load_config()
+        api_key = self.cfg.require_env(self.cfg.get("provider.api_key_env", "OPENROUTER_API_KEY"))
+        self.client = OpenAI(
+            base_url=self.cfg.get("provider.base_url", "https://openrouter.ai/api/v1"),
+            api_key=api_key,
+            timeout=self.cfg.get("constants.request_timeout", 90),
+            default_headers={
+                # OpenRouter attribution headers (optional).
+                "HTTP-Referer": self.cfg.get("provider.http_referer", ""),
+                "X-Title": self.cfg.get("provider.app_title", "Awesome-WAM"),
+            },
+        )
+        self._cost_table = self.cfg.get("models.cost_per_million", {}) or {}
+        self._max_retries = int(self.cfg.get("constants.max_retries", 4))
+        self._backoff = float(self.cfg.get("constants.retry_backoff", 3.0))
+
+    # --- model resolution ------------------------------------------------------
+    def resolve_model(self, tier_or_stage: str) -> str:
+        """Accept a tier name ('cheap'/'mid'/'strong') or a stage name."""
+        tiers = self.cfg.get("models.tiers", {}) or {}
+        if tier_or_stage in tiers:
+            return tiers[tier_or_stage]
+        return self.cfg.model_for_stage(tier_or_stage)
+
+    def _estimate_cost(self, model: str, in_tok: int, out_tok: int) -> float:
+        rates = self._cost_table.get(model)
+        if not rates:
+            return 0.0
+        return (in_tok * rates.get("input", 0.0) + out_tok * rates.get("output", 0.0)) / 1_000_000
+
+    # --- core call -------------------------------------------------------------
+    def _call(self, model: str, messages: list[dict], *, temperature: float, max_tokens: int,
+              json_mode: bool, label: str) -> tuple[str, int, int]:
+        @retry(
+            reraise=True,
+            stop=stop_after_attempt(self._max_retries),
+            wait=wait_exponential(multiplier=self._backoff, max=60),
+            retry=retry_if_exception(_is_transient),
+        )
+        def _do() -> tuple[str, int, int]:
+            kwargs: dict = {"model": model, "messages": messages, "temperature": temperature}
+            if max_tokens:
+                kwargs["max_tokens"] = max_tokens
+            if json_mode:
+                kwargs["response_format"] = {"type": "json_object"}
+            t0 = time.monotonic()
+            resp = self.client.chat.completions.create(**kwargs)
+            dt = time.monotonic() - t0
+            choice = resp.choices[0]
+            content = choice.message.content or ""
+            usage = resp.usage
+            in_tok = getattr(usage, "prompt_tokens", 0) or 0
+            out_tok = getattr(usage, "completion_tokens", 0) or 0
+            cost = self._estimate_cost(model, in_tok, out_tok)
+            COST.record(model, in_tok, out_tok, cost)
+            log.debug("call[%s] model=%s tokens=%d+%d cost=$%.5f %.2fs finish=%s",
+                      label, model, in_tok, out_tok, cost, dt, choice.finish_reason)
+            return content, in_tok, out_tok
+
+        return _do()
+
+    # --- public API ------------------------------------------------------------
+    def complete(self, tier: str, system: str, user: str, *, temperature: float | None = None,
+                 max_tokens: int | None = None, label: str = "complete",
+                 model: str | None = None) -> str:
+        model = model or self.resolve_model(tier)
+        temperature = self.cfg.get("models.defaults.temperature", 0.2) if temperature is None else temperature
+        max_tokens = self.cfg.get("models.defaults.max_tokens", 4096) if max_tokens is None else max_tokens
+        messages = [{"role": "system", "content": system}, {"role": "user", "content": user}]
+        content, _, _ = self._call(model, messages, temperature=temperature,
+                                    max_tokens=max_tokens, json_mode=False, label=label)
+        return content
+
+    def complete_json(self, tier: str, system: str, user: str, schema: Type[T], *,
+                      temperature: float | None = None, max_tokens: int | None = None,
+                      label: str = "json", max_repair: int = 2, model: str | None = None) -> T:
+        """Return an instance of ``schema``. Retries with the validation error on failure."""
+        model = model or self.resolve_model(tier)
+        temperature = self.cfg.get("models.defaults.temperature", 0.2) if temperature is None else temperature
+        max_tokens = self.cfg.get("models.defaults.max_tokens", 4096) if max_tokens is None else max_tokens
+        schema_json = json.dumps(schema.model_json_schema(), indent=2)
+        keys = ", ".join(schema.model_fields.keys())
+        sys_full = (f"{system}\n\nReturn ONLY a JSON object with exactly these top-level keys: "
+                    f"{keys}. Put the values directly at the top level — do NOT wrap them in a "
+                    f"'properties' object and do NOT echo the schema. Schema for reference:\n"
+                    f"{schema_json}")
+        messages = [{"role": "system", "content": sys_full}, {"role": "user", "content": user}]
+
+        last_err: Exception | None = None
+        for attempt in range(max_repair + 1):
+            content, _, _ = self._call(model, messages, temperature=temperature,
+                                        max_tokens=max_tokens, json_mode=True,
+                                        label=f"{label}#{attempt}")
+            try:
+                return schema.model_validate(_extract_obj(content, schema))
+            except (ValidationError, json.JSONDecodeError, ValueError) as e:
+                last_err = e
+                log.warning("JSON validation failed for %s (attempt %d): %s", label, attempt,
+                            str(e)[:200])
+                messages.append({"role": "assistant", "content": content})
+                messages.append({"role": "user", "content":
+                                 f"That did not validate: {e}. Return ONLY the corrected JSON "
+                                 f"object with top-level keys {keys}."})
+        raise LLMError(f"Could not get valid {schema.__name__} from {model}: {last_err}")
+
+
+def _strip_fences(text: str) -> str:
+    """Remove ```json ... ``` fences some models add despite json_mode."""
+    t = text.strip()
+    if t.startswith("```"):
+        t = t.split("\n", 1)[1] if "\n" in t else t
+        if t.endswith("```"):
+            t = t[: t.rfind("```")]
+    return t.strip()
+
+
+# Envelope keys some models wrap the real object in (e.g. deepseek echoes the JSON schema's
+# "properties"; others use "output"/"result").
+_ENVELOPES = ("properties", "output", "result", "data", "response", "json")
+
+
+def _loads_lenient(raw: str) -> object:
+    """json.loads, retrying after stripping trailing commas (a common LLM malformation)."""
+    try:
+        return json.loads(raw)
+    except json.JSONDecodeError:
+        repaired = re.sub(r",(\s*[}\]])", r"\1", raw)
+        return json.loads(repaired)
+
+
+def _extract_obj(content: str, schema: type[BaseModel]) -> dict:
+    """Parse model content to the target dict, tolerating common wrapper envelopes."""
+    raw = _strip_fences(content)
+    if not raw:
+        raise ValueError("empty content (model likely spent the token budget on reasoning)")
+    data = _loads_lenient(raw)
+    if isinstance(data, dict):
+        required = set(schema.model_fields.keys())
+        if not (required & data.keys()):  # none of the expected keys at top level → unwrap
+            for key in _ENVELOPES:
+                inner = data.get(key)
+                if isinstance(inner, dict) and (required & inner.keys()):
+                    return inner
+    return data

src/wam/logging.py

@@ -0,0 +1,93 @@
+"""Central logging.
+
+A single console handler at the configured level; in **debug mode** an additional file
+sink captures *everything* for the run under ``logs/run-<date>.log`` (gitignored). The LLM
+client logs a structured record per call (tier, model, tokens, cost, latency) so a debug
+run is a full, replayable trace. A module-level :class:`CostTracker` accumulates spend and
+emits a per-run summary.
+"""
+
+from __future__ import annotations
+
+import logging
+import os
+from dataclasses import dataclass, field
+from datetime import date
+from logging.handlers import RotatingFileHandler
+from pathlib import Path
+
+_CONFIGURED = False
+_FMT = "%(asctime)s %(levelname)-7s %(name)s | %(message)s"
+
+
+def is_debug() -> bool:
+    return os.environ.get("WAM_DEBUG", "").lower() in {"1", "true", "yes"}
+
+
+def setup_logging(level: str = "INFO", log_dir: str | Path = "logs", debug: bool | None = None,
+                  run_date: str | None = None) -> logging.Logger:
+    """Configure root logging once. Returns the ``wam`` logger.
+
+    When ``debug`` (or ``WAM_DEBUG`` env) is set, attaches a rotating file handler at DEBUG
+    level writing the full run trace to ``<log_dir>/run-<date>.log``.
+    """
+    global _CONFIGURED
+    debug = is_debug() if debug is None else debug
+    root = logging.getLogger()
+    base_level = logging.DEBUG if debug else getattr(logging, level.upper(), logging.INFO)
+    root.setLevel(logging.DEBUG if debug else base_level)
+
+    if not _CONFIGURED:
+        console = logging.StreamHandler()
+        console.setLevel(base_level)
+        console.setFormatter(logging.Formatter(_FMT))
+        root.addHandler(console)
+
+        if debug:
+            d = Path(log_dir)
+            d.mkdir(parents=True, exist_ok=True)
+            stamp = run_date or date.today().isoformat()
+            fh = RotatingFileHandler(d / f"run-{stamp}.log", maxBytes=20_000_000,
+                                     backupCount=5, encoding="utf-8")
+            fh.setLevel(logging.DEBUG)
+            fh.setFormatter(logging.Formatter(_FMT))
+            root.addHandler(fh)
+        _CONFIGURED = True
+
+    return logging.getLogger("wam")
+
+
+def get_logger(name: str) -> logging.Logger:
+    return logging.getLogger(f"wam.{name}")
+
+
+@dataclass
+class CostTracker:
+    """Accumulates token usage + estimated cost across a run."""
+
+    calls: int = 0
+    input_tokens: int = 0
+    output_tokens: int = 0
+    cost_usd: float = 0.0
+    by_model: dict[str, dict[str, float]] = field(default_factory=dict)
+
+    def record(self, model: str, in_tok: int, out_tok: int, cost: float) -> None:
+        self.calls += 1
+        self.input_tokens += in_tok
+        self.output_tokens += out_tok
+        self.cost_usd += cost
+        m = self.by_model.setdefault(model, {"calls": 0, "in": 0, "out": 0, "cost": 0.0})
+        m["calls"] += 1
+        m["in"] += in_tok
+        m["out"] += out_tok
+        m["cost"] += cost
+
+    def summary(self) -> str:
+        parts = [f"{self.calls} calls", f"{self.input_tokens}+{self.output_tokens} tok",
+                 f"${self.cost_usd:.4f}"]
+        per = ", ".join(f"{k}: ${v['cost']:.4f}" for k, v in sorted(self.by_model.items()))
+        return f"LLM usage: {' / '.join(parts)}" + (f" [{per}]" if per else "")
+
+
+# One shared tracker per process; the LLM client writes to it.
+COST = CostTracker()

src/wam/models.py

@@ -0,0 +1,38 @@
+"""Shared Pydantic models.
+
+``PaperRecord`` is the normalized in-memory shape produced by the fetch sources and stored
+in the ``papers`` table. LLM output schemas (summary, analysis, scores, ...) are added in
+later phases next to the stages that use them.
+"""
+
+from __future__ import annotations
+
+from pydantic import BaseModel, Field
+
+
+class Links(BaseModel):
+    abs: str | None = None
+    pdf: str | None = None
+    project_page: str | None = None
+    code: str | None = None
+    doi: str | None = None
+
+
+class PaperRecord(BaseModel):
+    """A fetched, source-normalized item (paper or news) before any LLM processing."""
+
+    id: str = Field(description="stable id, e.g. 'arxiv:2506.01234' or 'news:<sha1>'")
+    source: str
+    title: str
+    authors: list[str] = Field(default_factory=list)
+    published: str | None = None          # ISO date
+    abstract: str | None = None
+    categories: list[str] = Field(default_factory=list)
+    links: Links = Field(default_factory=Links)
+    citations: int = 0
+    influential_citations: int = 0
+    has_code: bool = False
+
+    @property
+    def arxiv_id(self) -> str | None:
+        return self.id.split("arxiv:", 1)[1] if self.id.startswith("arxiv:") else None

src/wam/notify/__init__.py

@@ -0,0 +1 @@
+"""Email digest delivery (Gmail SMTP)."""

src/wam/notify/mailer.py

@@ -0,0 +1,318 @@
+"""Build + send the daily digest email via Gmail SMTP.
+
+Clean and scannable: the key points (counts, what's hot, top picks) sit above the fold, with
+compact cards and obvious links. The recipient list lives in the ``SUBSCRIBERS`` env var
+(comma-separated or a JSON array) and is NEVER committed. Uses inline CSS for client support.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import re
+import smtplib
+import sqlite3
+from datetime import date
+from email.mime.multipart import MIMEMultipart
+from email.mime.text import MIMEText
+
+from wam.config import Config
+from wam.logging import get_logger
+
+log = get_logger("notify.mailer")
+
+
+def _subscribers() -> list[str]:
+    raw = os.environ.get("SUBSCRIBERS", "").strip()
+    if not raw:
+        return []
+    if raw.startswith("["):
+        try:
+            return [e.strip() for e in json.loads(raw) if e.strip()]
+        except Exception:  # noqa: BLE001
+            pass
+    return [e.strip() for e in raw.split(",") if e.strip()]
+
+
+def _links(lj: str | None) -> dict:
+    try:
+        return json.loads(lj or "{}")
+    except Exception:  # noqa: BLE001
+        return {}
+
+
+def _clip(text: str, n: int) -> str:
+    text = (text or "").strip()
+    if len(text) <= n:
+        return text
+    return text[:n].rsplit(" ", 1)[0].rstrip(".,;: ") + "…"
+
+
+_GH = re.compile(r"https?://github\.com/[\w.\-/]+")
+_HF = re.compile(r"https?://huggingface\.co/[\w.\-/]+")
+
+
+def _title_links(links: dict, abstract: str | None) -> str:
+    """Small inline links next to a title: code (GitHub), HF, project page, pdf.
+
+    GitHub/HF URLs are pulled from links or, failing that, from the abstract text.
+    """
+    abstract = abstract or ""
+    gh = links.get("code") if (links.get("code") and "github.com" in links["code"]) else None
+    if not gh:
+        m = _GH.search(abstract)
+        gh = m.group(0).rstrip(".)") if m else None
+    hf = links.get("hf")
+    if not hf:
+        if links.get("code") and "huggingface.co" in links["code"]:
+            hf = links["code"]
+        else:
+            m = _HF.search(abstract)
+            hf = m.group(0).rstrip(".)") if m else None
+    chips = []
+    style = 'style="color:#1a4fcc;font-size:11px;text-decoration:none;margin-left:6px"'
+    if gh:
+        chips.append(f'<a href="{gh}" {style}>⌨ code</a>')
+    if hf:
+        chips.append(f'<a href="{hf}" {style}>🤗 HF</a>')
+    if links.get("project_page"):
+        chips.append(f'<a href="{links["project_page"]}" {style}>🔗 page</a>')
+    if links.get("pdf"):
+        chips.append(f'<a href="{links["pdf"]}" {style}>📄 pdf</a>')
+    return "".join(chips)
+
+
+def _wt(row) -> float:
+    try:
+        return float(json.loads(row["scores_json"]).get("weighted_total") or 0)
+    except Exception:  # noqa: BLE001
+        return 0.0
+
+
+def _score_badge(v) -> str:
+    return (f'<span style="background:#1a4fcc;color:#fff;border-radius:4px;padding:1px 6px;'
+            f'font-size:12px;margin-left:6px">{v}</span>')
+
+
+_GEN_LABELS = [("novelty", "nov"), ("soundness", "snd"), ("impact", "imp")]
+# WAM order: top-4 (weighted 2x) first, then the rest.
+_WAM_LABELS = [("inference_speed", "spd"), ("generalist", "gen"), ("specialist", "spec"),
+               ("inference_cost", "cost"), ("trustworthiness", "trust"),
+               ("collaborative", "collab"), ("controlled_generation", "ctrl"), ("other", "other")]
+_WAM_TOP4 = {"inference_speed", "generalist", "specialist", "inference_cost"}
+
+
+def _v(x) -> str:
+    return str(x) if isinstance(x, int) else "–"
+
+
+def _chip(label: str, val: str, strong: bool = False) -> str:
+    bg, weight = ("#e8f0ff", "600") if strong else ("#f2f2f2", "400")
+    return (f'<span style="background:{bg};border-radius:4px;padding:1px 5px;margin:0 4px 2px 0;'
+            f'font-size:11px;color:#333;font-weight:{weight};display:inline-block">{label} {val}</span>')
+
+
+def _metric_badges(scores: dict) -> str:
+    """Full two-layer dimension badges (general + all WAM; top-4 WAM emphasized)."""
+    g, w = scores.get("general", {}) or {}, scores.get("wam", {}) or {}
+    gen = "".join(_chip(l, _v(g.get(k))) for k, l in _GEN_LABELS)
+    wam = "".join(_chip(l, _v(w.get(k)), k in _WAM_TOP4) for k, l in _WAM_LABELS)
+    return (f'<div style="margin:5px 0 0"><span style="color:#999;font-size:11px">general</span> '
+            f'{gen}</div>'
+            f'<div style="margin:2px 0 0"><span style="color:#999;font-size:11px">WAM</span> '
+            f'{wam}</div>')
+
+
+def _top4_inline(scores: dict) -> str:
+    w = scores.get("wam", {}) or {}
+    return " · ".join(f"{l} {_v(w.get(k))}" for k, l in
+                      [("inference_speed", "spd"), ("generalist", "gen"),
+                       ("specialist", "spec"), ("inference_cost", "cost")])
+
+
+def _card(row, direction: str | None = None) -> str:
+    """Detailed featured card: title (+ links), direction tag, blurb, metrics."""
+    s = json.loads(row["scores_json"])
+    sm = json.loads(row["summary_json"] or "{}")
+    links = _links(row["links_json"])
+    link = links.get("abs") or links.get("pdf") or "#"
+    tldr = sm.get("tldr", "")
+    extra = _clip(" ".join(x for x in (sm.get("method", ""), sm.get("results", "")) if x), 300)
+    dir_html = ""
+    if direction:
+        dir_html = (f'<div style="margin-top:4px"><span style="background:#eef3ff;color:#1a4fcc;'
+                    f'border-radius:10px;padding:1px 9px;font-size:11px">📂 {direction}</span>'
+                    f'</div>')
+    blurb = f'<div style="font-size:13px;color:#333;margin-top:5px">{tldr}</div>'
+    if extra:
+        blurb += f'<div style="font-size:12px;color:#666;margin-top:3px">{extra}</div>'
+    return (f'<div style="margin:12px 0;padding:10px 14px;border-left:3px solid #1a4fcc;'
+            f'background:#fafbff;border-radius:0 6px 6px 0">'
+            f'<a href="{link}" style="font-weight:600;color:#1a4fcc;text-decoration:none;'
+            f'font-size:14px">{row["title"]}</a>{_score_badge(s.get("weighted_total","?"))}'
+            f'{_title_links(links, row["abstract"])}'
+            f'{dir_html}{blurb}'
+            f'<div style="margin-top:6px">{_metric_badges(s)}</div></div>')
+
+
+def _compact(row) -> str:
+    """Compact grouped-tier item: score + title link + short description."""
+    s = json.loads(row["scores_json"])
+    tldr = json.loads(row["summary_json"] or "{}").get("tldr", "")
+    if len(tldr) > 160:
+        tldr = tldr[:160].rsplit(" ", 1)[0].rstrip(".,;: ") + "…"
+    link = (_links(row["links_json"]).get("abs") or _links(row["links_json"]).get("pdf") or "#")
+    desc = f' — <span style="color:#444">{tldr}</span>' if tldr else ""
+    return (f'<div style="font-size:13px;margin:5px 0">'
+            f'<b>{s.get("weighted_total","?")}</b> · '
+            f'<a href="{link}" style="color:#1a4fcc;text-decoration:none">{row["title"]}</a>'
+            f'{_title_links(_links(row["links_json"]), row["abstract"])}'
+            f'{desc}'
+            f'<div style="font-size:11px;color:#888;margin-top:1px">{_top4_inline(s)}</div></div>')
+
+
+def _teams_html(conn: sqlite3.Connection) -> str:
+    """Short closing paragraph on the most influential teams / contributors."""
+    groups = conn.execute("SELECT name FROM groups ORDER BY json_array_length(member_ids_json) "
+                          "DESC LIMIT 3").fetchall()
+    authors = conn.execute("SELECT name, affiliation FROM authors ORDER BY "
+                           "json_array_length(paper_ids_json) DESC LIMIT 5").fetchall()
+    if not authors and not groups:
+        return ""
+    if groups:
+        body = ", ".join(f"<b>{g['name']}</b>" for g in groups)
+        lead = f"Most active groups in the tracked corpus: {body}."
+    else:
+        body = ", ".join(f"<b>{a['name']}</b>" + (f" ({a['affiliation']})" if a["affiliation"]
+                         else "") for a in authors)
+        lead = f"Most active contributors right now: {body}."
+    return ('<h2 style="font-size:15px;border-bottom:1px solid #eee;padding-bottom:4px;'
+            f'margin-top:18px">👥 Influential teams</h2>'
+            f'<p style="font-size:13px;color:#444;margin:6px 0 0">{lead}</p>')
+
+
+def build_html(cfg: Config, conn: sqlite3.Connection, today: str | None = None) -> tuple[str, str]:
+    today = today or date.today().isoformat()
+    counts = dict(conn.execute("SELECT track, count(*) FROM papers GROUP BY track").fetchall())
+    new_core = conn.execute(
+        "SELECT count(*) FROM papers WHERE track='core' AND first_seen=?", (today,)).fetchone()[0]
+
+    # Core papers: new today; fall back to recent if none new.
+    sel = ("SELECT id, title, abstract, links_json, scores_json, summary_json FROM papers WHERE "
+           "track='core' AND scores_json IS NOT NULL")
+    order = " ORDER BY json_extract(scores_json,'$.weighted_total') DESC"
+    rows = conn.execute(sel + " AND first_seen=?" + order, (today,)).fetchall()
+    fallback = not rows
+    if fallback:
+        rows = conn.execute(sel + order + " LIMIT 60").fetchall()
+
+    snap = conn.execute("SELECT max(snapshot_date) FROM fronts").fetchone()[0]
+    # Rising directions, excluding the catch-all "Miscellaneous" bucket.
+    hot = conn.execute(
+        "SELECT name, size, summary FROM fronts WHERE snapshot_date=? AND momentum='rising' "
+        "AND name NOT LIKE '%iscellaneous%' ORDER BY size DESC LIMIT 6", (snap,)).fetchall() \
+        if snap else []
+    # paper id -> research direction (smaller/more-specific fronts win, via size ASC)
+    dir_of: dict[str, str] = {}
+    if snap:
+        for fr in conn.execute("SELECT name, member_ids_json FROM fronts WHERE "
+                               "snapshot_date=? ORDER BY size", (snap,)):
+            for pid in json.loads(fr["member_ids_json"] or "[]"):
+                dir_of[pid] = fr["name"]
+
+    feat_thr = float(cfg.get("email.feature_threshold", 7.0))
+    featured = [r for r in rows if _wt(r) >= feat_thr][: int(cfg.get("email.max_featured", 8))]
+    feat_ids = {r["id"] for r in featured}
+    rest = [r for r in rows if r["id"] not in feat_ids][: int(cfg.get("email.max_grouped", 40))]
+
+    S = "font-family:-apple-system,Segoe UI,Roboto,Helvetica,Arial,sans-serif"
+    parts = [f'<div style="max-width:640px;margin:auto;{S};color:#1a1a1a">']
+    parts.append(f'<h1 style="font-size:20px;margin:0 0 8px">🤖 WAM Daily — {today}</h1>')
+    new_adj = conn.execute("SELECT count(*) FROM papers WHERE track='adjacent' AND first_seen=?",
+                           (today,)).fetchone()[0]
+    total_core, total_adj = counts.get("core", 0), counts.get("adjacent", 0)
+
+    def _stat(n, lbl):
+        return (f'<td style="text-align:center;padding:8px 6px">'
+                f'<div style="font-size:22px;font-weight:700;color:#1a4fcc">{n}</div>'
+                f'<div style="font-size:11px;color:#777">{lbl}</div></td>')
+    parts.append(
+        '<table style="width:100%;border-collapse:collapse;margin:0 0 18px;background:#f5f8ff;'
+        'border-radius:8px"><tr>'
+        + _stat(new_core + new_adj, "relevant today") + _stat(new_core, "core today")
+        + _stat(total_core + total_adj, "relevant total") + _stat(total_core, "core total")
+        + '</tr></table>')
+    if hot:
+        parts.append('<h2 style="font-size:15px;border-bottom:1px solid #eee;'
+                     'padding-bottom:4px">📈 What\'s hot</h2>')
+        items = "".join(
+            f'<div style="font-size:13px;margin:5px 0"><b>{h["name"]}</b> '
+            f'<span style="color:#999">({h["size"]})</span> — '
+            f'<span style="color:#444">{_clip(h["summary"] or "", 160)}</span></div>'
+            for h in hot)
+        parts.append(items)
+
+    # Tier 1 — featured (detailed cards)
+    label = "Top recent papers" if fallback else "Top new papers today"
+    parts.append(f'<h2 style="font-size:15px;border-bottom:1px solid #eee;padding-bottom:4px">'
+                 f'⭐ {label} (score ≥ {feat_thr:g})</h2>')
+    if not featured:
+        parts.append('<p style="font-size:13px;color:#666">Nothing above the feature threshold '
+                     '— see the rest below.</p>')
+    parts.extend(_card(r, dir_of.get(r["id"])) for r in featured)
+
+    # Tier 2 — the rest, grouped by research direction
+    if rest:
+        parts.append('<h2 style="font-size:15px;border-bottom:1px solid #eee;padding-bottom:4px;'
+                     'margin-top:18px">More core papers</h2>')
+        if bool(cfg.get("email.group_lower_by_direction", True)):
+            groups: dict[str, list] = {}
+            for r in rest:
+                groups.setdefault(dir_of.get(r["id"], "Other"), []).append(r)
+            # Miscellaneous / Other always go last; otherwise larger groups first.
+            def _gkey(k):
+                last = k == "Other" or "iscellaneous" in k
+                return (last, -len(groups[k]))
+            for gname in sorted(groups, key=_gkey):
+                parts.append(f'<p style="font-weight:600;font-size:13px;margin:12px 0 2px">'
+                             f'{gname} <span style="color:#999;font-weight:400">'
+                             f'({len(groups[gname])})</span></p>')
+                parts.extend(_compact(r) for r in groups[gname])
+        else:
+            parts.extend(_compact(r) for r in rest)
+    parts.append(_teams_html(conn))
+
+    parts.append(
+        '<p style="color:#aaa;font-size:11px;margin-top:18px;border-top:1px solid #eee;'
+        'padding-top:8px">Dimensions — general: nov=novelty, snd=soundness, imp=impact · '
+        'WAM (bold = weighted 2×): spd=inference speed, gen=generalist, spec=specialist, '
+        'cost=inference cost, trust=trustworthiness, collab=collaborative, '
+        'ctrl=controlled generation. “–” = the paper doesn’t address it.</p>')
+    parts.append(f'<p style="color:#999;font-size:12px;margin-top:8px">Awesome-WAM · '
+                 f'<a href="https://github.com/your-org/Awesome-WAM">repo</a></p></div>')
+    subject = f"WAM Daily — {today}: {new_core} new" + ("" if not fallback else " (recap)")
+    return subject, "".join(parts)
+
+
+def send(cfg: Config, conn: sqlite3.Connection, today: str | None = None,
+         test_to: str | None = None) -> int:
+    user = os.environ.get("GMAIL_USER")
+    pw = os.environ.get("GMAIL_APP_PASSWORD")
+    if not (user and pw):
+        log.warning("GMAIL_USER / GMAIL_APP_PASSWORD not set; skipping email send")
+        return 0
+    recipients = [test_to] if test_to else _subscribers()
+    if not recipients:
+        log.warning("no SUBSCRIBERS configured; skipping send")
+        return 0
+    subject, html = build_html(cfg, conn, today)
+    msg = MIMEMultipart("alternative")
+    msg["Subject"], msg["From"] = subject, user
+    msg["To"] = ", ".join(recipients)
+    msg.attach(MIMEText("This digest is best viewed as HTML.", "plain"))
+    msg.attach(MIMEText(html, "html"))
+    with smtplib.SMTP_SSL("smtp.gmail.com", 465) as server:
+        server.login(user, pw)
+        server.sendmail(user, recipients, msg.as_string())
+    log.info("sent digest to %d recipient(s)", len(recipients))
+    return len(recipients)

src/wam/pipeline/__init__.py

@@ -0,0 +1 @@
+"""Pipeline stages and the daily orchestrator."""

src/wam/pipeline/analyze.py

@@ -0,0 +1,45 @@
+"""Analyze stage — deep technical analysis for core-track papers (mid tier).
+
+Capped at ``constants.analyze_cap`` papers per run (highest-relevance first) to bound cost.
+"""
+
+from __future__ import annotations
+
+import sqlite3
+
+from wam.config import Config
+from wam.llm import LLMClient
+from wam.logging import get_logger
+from wam.pipeline.schemas import PaperAnalysis
+from wam.store import papers as ps
+
+log = get_logger("pipeline.analyze")
+
+SYSTEM = ("You are an expert reviewer of World Action Models (embodied/robot foundation "
+          "models, VLA, world models). Analyze the paper rigorously and skeptically. Always "
+          "write in English, regardless of the paper's language.\n\n"
+          "--- PROFILE ---\n{profile}")
+
+
+def run(cfg: Config, client: LLMClient, conn: sqlite3.Connection,
+        limit: int | None = None) -> int:
+    cap = limit or int(cfg.get("constants.analyze_cap", 40))
+    system = SYSTEM.format(profile=cfg.profile_text())
+    todo = ps.needs_analysis(conn, limit=cap)
+    log.info("analyzing %d core papers (cap=%d)", len(todo), cap)
+    done = 0
+    for row in todo:
+        user = f"Title: {row['title']}\n\nAbstract: {row['abstract'] or '(none)'}"
+        try:
+            a = client.complete_json("analyze", system, user, PaperAnalysis,
+                                     label="analyze", max_tokens=5000)
+        except Exception as e:  # noqa: BLE001
+            log.warning("analyze failed for %s: %s", row["id"], e)
+            continue
+        ps.set_analysis(conn, row["id"], a.model_dump_json())
+        done += 1
+        if done % 10 == 0:
+            conn.commit()
+    conn.commit()
+    log.info("analyzed %d papers", done)
+    return done

src/wam/pipeline/extract.py

@@ -0,0 +1,64 @@
+"""Extract stage — normalized benchmark rows + model cards for core papers (mid tier).
+
+Reads the full PDF text (cached) since benchmark numbers live in tables, not the abstract.
+Extraction is deliberately conservative: only numbers actually stated, each tagged with the
+model variant (name + training dataset) and whether it's the authors' own claim. Capped at
+``constants.analyze_cap`` per run.
+"""
+
+from __future__ import annotations
+
+import json
+import sqlite3
+
+from wam.config import Config
+from wam.llm import LLMClient
+from wam.logging import get_logger
+from wam.pipeline.schemas import ExtractionResult
+from wam.sources import pdf
+from wam.store import benchmarks as bm
+from wam.store import papers as ps
+
+log = get_logger("pipeline.extract")
+
+SYSTEM = (
+    "Extract structured experimental results from this World Action Models paper. Rules:\n"
+    "- Only report numbers ACTUALLY stated in the text — never estimate or invent.\n"
+    "- Each model/system is identified by (model_name, training_dataset). The SAME name "
+    "trained/finetuned on a different dataset is a DIFFERENT variant — always capture the "
+    "training/finetune dataset when stated.\n"
+    "- For each result give the benchmark, task, metric name+value, and any inference "
+    "speed/cost with their units and hardware as reported.\n"
+    "- Set claimed_by_authors=false for numbers quoted from OTHER papers (baselines/"
+    "comparisons), true for the paper's own results.\n"
+    "- BE CONCISE: report only this paper's headline results — at most ~15 benchmark rows "
+    "and ~6 model variants (the proposed method + key baselines). Keep every 'notes' field "
+    "under 12 words; omit notes if not needed.\n"
+    "- Empty lists are fine if the paper reports no quantitative results.")
+
+
+def run(cfg: Config, client: LLMClient, conn: sqlite3.Connection,
+        limit: int | None = None) -> int:
+    cap = limit or int(cfg.get("constants.analyze_cap", 40))
+    todo = ps.needs_extract(conn, limit=cap)
+    log.info("extracting benchmarks for %d core papers (cap=%d)", len(todo), cap)
+    done = 0
+    for row in todo:
+        links = json.loads(row["links_json"] or "{}")
+        # 35k chars (~9k tokens) comfortably covers intro+results+tables while staying fast.
+        text = pdf.get_text(cfg, row["id"], links.get("pdf") or "", max_chars=35000)
+        body = text or row["abstract"] or ""
+        user = (f"Title: {row['title']}\n\nAbstract: {row['abstract'] or ''}\n\n"
+                f"Paper text (may be truncated):\n{body}")
+        try:
+            res = client.complete_json("extract", SYSTEM, user, ExtractionResult,
+                                       label="extract", max_tokens=12000)
+        except Exception as e:  # noqa: BLE001
+            log.warning("extract failed for %s: %s", row["id"], e)
+            continue
+        n_models, n_rows = bm.store_extraction(conn, row["id"], res.models, res.benchmarks)
+        done += 1
+        log.debug("%s -> %d variants, %d benchmark rows", row["id"], n_models, n_rows)
+        conn.commit()
+    log.info("extracted benchmarks from %d papers", done)
+    return done

src/wam/pipeline/fetch.py

@@ -0,0 +1,51 @@
+"""Fetch + dedup stage.
+
+Gathers candidates from all enabled sources, enriches them (Semantic Scholar citations),
+dedups against what's already in the DB, persists the new ones, and updates enrichment on
+all seen ones. PwC code-link lookups are deferred to the post-filter shortlist (Phase 3) to
+keep this stage cheap. Returns the list of *newly inserted* records.
+"""
+
+from __future__ import annotations
+
+import sqlite3
+
+from wam.config import Config
+from wam.logging import get_logger
+from wam.models import PaperRecord
+from wam.sources import arxiv, news, semantic_scholar
+from wam.store import papers as paper_store
+
+log = get_logger("pipeline.fetch")
+
+
+def gather(cfg: Config) -> list[PaperRecord]:
+    """Pull + enrich candidates from all sources (no DB writes). Deduped by id."""
+    records: dict[str, PaperRecord] = {}
+    for rec in arxiv.fetch(cfg):
+        records.setdefault(rec.id, rec)
+    for rec in news.fetch(cfg):
+        records.setdefault(rec.id, rec)
+    candidates = list(records.values())
+    log.info("gathered %d unique candidates across sources", len(candidates))
+    semantic_scholar.enrich(cfg, [r for r in candidates if r.source != "news"])
+    return candidates
+
+
+def persist(conn: sqlite3.Connection, candidates: list[PaperRecord]) -> list[PaperRecord]:
+    """Insert new records, refresh enrichment on existing. Returns the new ones."""
+    seen = paper_store.existing_ids(conn)
+    new = [r for r in candidates if r.id not in seen]
+    for rec in candidates:
+        if rec.id in seen:
+            paper_store.update_enrichment(conn, rec)
+        else:
+            paper_store.insert_new(conn, rec)
+    conn.commit()
+    log.info("persisted: %d new, %d already known", len(new), len(candidates) - len(new))
+    return new
+
+
+def run(cfg: Config, conn: sqlite3.Connection) -> list[PaperRecord]:
+    candidates = gather(cfg)
+    return persist(conn, candidates)

src/wam/pipeline/filter.py

@@ -0,0 +1,57 @@
+"""Relevance filter — the cheap-model gate.
+
+Classifies each unfiltered paper into core / adjacent / drop and assigns a 0..1 relevance,
+reading the research-interest profile so the call is personalized. News items bypass the LLM
+(tagged ``news``). A failed call stores relevance=-1 so it retries next run. The DB `track`
+column is the cache: filtered papers are never re-classified.
+"""
+
+from __future__ import annotations
+
+import sqlite3
+
+from wam.config import Config
+from wam.llm import LLMClient
+from wam.logging import COST, get_logger
+from wam.pipeline.schemas import RelevanceVerdict
+from wam.store import papers as ps
+
+log = get_logger("pipeline.filter")
+
+SYSTEM = ("You triage papers for a World Action Models (WAM) intelligence digest. Use the "
+          "profile below to decide the track. Be decisive but inclusive of adjacent work with "
+          "transferable techniques. Write the reason in English.\n\n--- PROFILE ---\n{profile}")
+
+
+def run(cfg: Config, client: LLMClient, conn: sqlite3.Connection,
+        limit: int | None = None) -> dict[str, int]:
+    profile = cfg.profile_text()
+    system = SYSTEM.format(profile=profile)
+    threshold = float(cfg.get("constants.relevance_threshold", 0.5))
+    counts = {"core": 0, "adjacent": 0, "drop": 0, "news": 0, "error": 0}
+
+    todo = ps.needs_filter(conn, limit=limit)
+    log.info("filtering %d papers (threshold=%.2f)", len(todo), threshold)
+    for i, row in enumerate(todo):
+        if i and i % 25 == 0:
+            conn.commit()  # checkpoint so a long run doesn't lose progress on crash
+            log.info("filter progress: %d/%d", i, len(todo))
+        if row["source"] == "news":
+            ps.set_filter(conn, row["id"], "news", 1.0, "news item")
+            counts["news"] += 1
+            continue
+        user = f"Title: {row['title']}\n\nAbstract: {row['abstract'] or '(none)'}"
+        try:
+            v = client.complete_json("cheap", system, user, RelevanceVerdict,
+                                     label="filter", max_tokens=1500)
+        except Exception as e:  # noqa: BLE001
+            log.warning("filter failed for %s: %s", row["id"], e)
+            ps._touch(conn, row["id"], relevance=-1)
+            counts["error"] += 1
+            continue
+        track = "drop" if (v.track == "drop" or v.relevance < threshold) else v.track
+        ps.set_filter(conn, row["id"], track, v.relevance, v.reason)
+        counts[track] += 1
+    conn.commit()
+    log.info("filter result: %s", counts)
+    return counts

src/wam/pipeline/innovation.py

@@ -0,0 +1,43 @@
+"""Innovation stage — for adjacent-track papers (VLA / world model / video gen).
+
+No rubric scoring: just the key technical idea and why it could transfer to WAM (mid tier).
+"""
+
+from __future__ import annotations
+
+import sqlite3
+
+from wam.config import Config
+from wam.llm import LLMClient
+from wam.logging import get_logger
+from wam.pipeline.schemas import InnovationNote
+from wam.store import papers as ps
+
+log = get_logger("pipeline.innovation")
+
+SYSTEM = ("This paper is adjacent to World Action Models (it is VLA / a world model / video "
+          "generation, not a WAM itself). Identify its core technical innovation and explain "
+          "concretely why/how it could transfer to World Action Models. Always write in "
+          "English, regardless of the paper's language.")
+
+
+def run(cfg: Config, client: LLMClient, conn: sqlite3.Connection,
+        limit: int | None = None) -> int:
+    todo = ps.needs_innovation(conn, limit=limit)
+    log.info("extracting innovation notes for %d adjacent papers", len(todo))
+    done = 0
+    for row in todo:
+        user = f"Title: {row['title']}\n\nAbstract: {row['abstract'] or '(none)'}"
+        try:
+            note = client.complete_json("innovation", SYSTEM, user, InnovationNote,
+                                        label="innovation", max_tokens=2000)
+        except Exception as e:  # noqa: BLE001
+            log.warning("innovation failed for %s: %s", row["id"], e)
+            continue
+        ps.set_innovation(conn, row["id"], note.model_dump_json())
+        done += 1
+        if done % 20 == 0:
+            conn.commit()
+    conn.commit()
+    log.info("innovation notes for %d papers", done)
+    return done

src/wam/pipeline/links.py

@@ -0,0 +1,84 @@
+"""Links enrichment — find each paper's GitHub / Hugging Face repo.
+
+Repo URLs are almost always in the PDF body ("Code is available at https://github.com/…"),
+not the abstract — so we scan the (cached) full text. A keyword-proximity heuristic picks the
+authors' own repo over baselines/comparisons. Found URLs are written into ``links_json`` as
+``code`` (GitHub) and ``hf`` (Hugging Face); the README and email then surface them.
+"""
+
+from __future__ import annotations
+
+import json
+import re
+import sqlite3
+
+from wam.config import Config
+from wam.logging import get_logger
+from wam.sources import pdf
+
+log = get_logger("pipeline.links")
+
+_GH = re.compile(r"https?://github\.com/[A-Za-z0-9_.\-]+/[A-Za-z0-9_.\-]+")
+_HF = re.compile(r"https?://huggingface\.co/[A-Za-z0-9_.\-/]+")
+# repos that are almost always dependencies/baselines, not the paper's own.
+_GH_DENY = ("github.com/huggingface", "github.com/pytorch", "github.com/openai",
+            "github.com/google", "github.com/facebookresearch/detectron")
+_KEYWORDS = ("code", "project", "available", "github", "hugging", "weights", "released",
+             "release", "repo", "page", "checkpoint")
+
+
+def _clean(url: str) -> str:
+    return url.rstrip(").,;:'\"]}").rstrip("/")
+
+
+def _best(text: str, pattern: re.Pattern, deny: tuple = ()) -> str | None:
+    cands = [(m.start(), _clean(m.group(0))) for m in pattern.finditer(text)]
+    cands = [(p, u) for p, u in cands if not any(d in u.lower() for d in deny)]
+    if not cands:
+        return None
+    low = text.lower()
+    for pos, url in cands:  # prefer a URL sitting right after a "code/project/…" cue
+        if any(k in low[max(0, pos - 60):pos] for k in _KEYWORDS):
+            return url
+    return cands[0][1]
+
+
+def run(cfg: Config, conn: sqlite3.Connection, limit: int | None = None,
+        download: bool = True) -> int:
+    q = ("SELECT id, abstract, links_json FROM papers WHERE track IN ('core','adjacent')")
+    if limit:
+        q += f" LIMIT {int(limit)}"
+    rows = conn.execute(q).fetchall()
+    log.info("links: scanning %d papers (download=%s)", len(rows), download)
+    found = 0
+    for i, r in enumerate(rows):
+        links = json.loads(r["links_json"] or "{}")
+        if links.get("code") and links.get("hf"):
+            continue
+        pdf_url = links.get("pdf") or (f"https://arxiv.org/pdf/{r['id'].split(':',1)[1]}"
+                                       if r["id"].startswith("arxiv:") else "")
+        text = pdf.get_text(cfg, r["id"], pdf_url if download else "", max_chars=60000)
+        blob = (text or "") + "\n" + (r["abstract"] or "")
+        if not blob.strip():
+            continue
+        changed = False
+        if not links.get("code"):
+            gh = _best(blob, _GH, _GH_DENY)
+            if gh:
+                links["code"] = gh
+                changed = True
+        if not links.get("hf"):
+            hf = _best(blob, _HF)
+            if hf:
+                links["hf"] = hf
+                changed = True
+        if changed:
+            conn.execute("UPDATE papers SET links_json=? WHERE id=?",
+                         (json.dumps(links), r["id"]))
+            found += 1
+        if i and i % 25 == 0:
+            conn.commit()
+            log.info("links progress: %d/%d (found %d)", i, len(rows), found)
+    conn.commit()
+    log.info("links: enriched %d papers with code/HF", found)
+    return found

src/wam/pipeline/people.py

@@ -0,0 +1,118 @@
+"""People stage — influential authors + research groups.
+
+Identity comes from **Semantic Scholar author IDs** attached to each paper (reliable; avoids
+name collisions), not name search. Influential authors (on >= N tracked papers) get an
+LLM-summarized research direction. **Groups are co-authorship clusters**: influential authors
+who publish together form a lab/group (connected components of the co-authorship graph) —
+this works even though S2's affiliation data is sparse.
+"""
+
+from __future__ import annotations
+
+import json
+import sqlite3
+from collections import defaultdict
+
+from pydantic import BaseModel, Field
+
+from wam.config import Config
+from wam.llm import LLMClient
+from wam.logging import get_logger
+from wam.sources import semantic_scholar as s2
+from wam.store import people as store
+from wam.store.people import slug
+
+log = get_logger("pipeline.people")
+
+
+class Directions(BaseModel):
+    directions: str = Field(description="1-2 sentences on this author's main WAM-related "
+                                        "research directions, grounded in the listed papers")
+
+
+def run(cfg: Config, client: LLMClient, conn: sqlite3.Connection,
+        limit: int | None = None) -> dict[str, int]:
+    min_papers = int(cfg.get("people.min_papers", 2))
+    cap = limit or int(cfg.get("people.max_per_run", 30))
+
+    rows = conn.execute(
+        "SELECT id, title, summary_json FROM papers WHERE track IN ('core','adjacent')").fetchall()
+    meta = {r["id"]: {"title": r["title"],
+                      "tldr": (json.loads(r["summary_json"] or "{}").get("tldr", "")
+                               if r["summary_json"] else "")} for r in rows}
+    arxiv_ids = [r["id"].split("arxiv:", 1)[1] for r in rows if r["id"].startswith("arxiv:")]
+    authors_by_paper = s2.fetch_authors(cfg, arxiv_ids)
+    log.info("got authors for %d papers", len(authors_by_paper))
+    if not authors_by_paper:
+        # Don't wipe the existing registry on a transient S2 failure (e.g. 429).
+        log.warning("no author data from S2 (rate-limited?); keeping existing registry")
+        return {"authors": 0, "groups": 0}
+
+    # Aggregate by S2 authorId.
+    agg: dict[str, dict] = {}
+    for pid, alist in authors_by_paper.items():
+        for a in alist:
+            aid = a.get("authorId")
+            if not aid:
+                continue
+            e = agg.setdefault(aid, {"name": a.get("name") or "?", "pids": set(),
+                                     "affiliation": None, "h_index": a.get("hIndex"),
+                                     "citations": a.get("citationCount")})
+            e["pids"].add(pid)
+            affs = a.get("affiliations") or []
+            if affs and not e["affiliation"]:
+                e["affiliation"] = affs[0]
+
+    influential = {aid: e for aid, e in agg.items() if len(e["pids"]) >= min_papers}
+    ranked = sorted(influential, key=lambda x: -len(influential[x]["pids"]))[:cap]
+    log.info("%d distinct S2 authors; %d influential (>=%d papers); processing top %d",
+             len(agg), len(influential), min_papers, len(ranked))
+
+    # Fresh rebuild each run.
+    conn.execute("DELETE FROM authors")
+    conn.execute("DELETE FROM groups")
+
+    kept = set(ranked)
+    for aid in ranked:
+        e = influential[aid]
+        titles = "\n".join(f"- {meta[p]['title']}: {meta[p]['tldr']}" for p in e["pids"]
+                           if p in meta)
+        try:
+            directions = client.complete_json(
+                "cheap", "Summarize an author's research directions for a WAM digest. Write "
+                "in English.", f"Author: {e['name']}\nTracked papers:\n{titles}", Directions,
+                label="author-directions", max_tokens=2000).directions
+        except Exception as ex:  # noqa: BLE001
+            log.warning("directions failed for %s: %s", e["name"], ex)
+            directions = None
+        store.upsert_author(
+            conn, author_id=aid, name=e["name"], affiliation=e["affiliation"],
+            s2_url=f"https://www.semanticscholar.org/author/{aid}", citations=e["citations"],
+            h_index=e["h_index"], paper_ids=sorted(e["pids"]), directions=directions)
+    conn.commit()
+
+    # Groups = co-authorship clusters among the influential authors.
+    import networkx as nx
+    g = nx.Graph()
+    g.add_nodes_from(kept)
+    for pid, alist in authors_by_paper.items():
+        ids = [a["authorId"] for a in alist if a.get("authorId") in kept]
+        for i in range(len(ids)):
+            for j in range(i + 1, len(ids)):
+                g.add_edge(ids[i], ids[j])
+    n_grp = 0
+    for comp in nx.connected_components(g):
+        members = [m for m in comp]
+        if len(members) < 2:
+            continue
+        lead = max(members, key=lambda m: len(influential[m]["pids"]))
+        aff = next((influential[m]["affiliation"] for m in members
+                    if influential[m]["affiliation"]), None)
+        name = f"{influential[lead]['name']} group" + (f" · {aff}" if aff else "")
+        notable = sorted({p for m in members for p in influential[m]["pids"]})[:8]
+        store.upsert_group(conn, group_id=slug(name), name=name, affiliation=aff,
+                           member_ids=members, directions=None, notable=notable)
+        n_grp += 1
+    conn.commit()
+    log.info("stored %d authors, %d co-authorship groups", len(ranked), n_grp)
+    return {"authors": len(ranked), "groups": n_grp}

src/wam/pipeline/schemas.py

@@ -0,0 +1,100 @@
+"""Pydantic schemas for LLM pipeline outputs (Phase 3+).
+
+Each stage forces the model to return one of these; the LLM client validates and retries on
+malformed output. ``"N/A"`` is allowed for any WAM metric the paper does not address so we
+never penalize with a fake 0.
+"""
+
+from __future__ import annotations
+
+from typing import Literal
+
+from pydantic import BaseModel, Field
+
+Track = Literal["core", "adjacent", "drop"]
+# Scores are 0-10 ints, or the string "N/A" when unaddressed.
+Score = int | Literal["N/A"]
+
+
+class RelevanceVerdict(BaseModel):
+    track: Track = Field(description="core = is a WAM; adjacent = VLA/world-model/video-gen "
+                                     "with transferable ideas; drop = unrelated")
+    relevance: float = Field(ge=0.0, le=1.0, description="0..1 confidence this matters to WAM")
+    reason: str = Field(description="one sentence justification")
+
+
+class PaperSummary(BaseModel):
+    tldr: str = Field(description="one-sentence takeaway")
+    problem: str
+    method: str
+    results: str
+
+
+class PaperAnalysis(BaseModel):
+    contributions: list[str] = Field(description="key contributions, most important first")
+    limitations: list[str]
+    wam_relevance: str = Field(description="why this matters for World Action Models")
+
+
+class GeneralScores(BaseModel):
+    novelty: Score
+    soundness: Score
+    impact: Score
+
+
+class WAMScores(BaseModel):
+    generalist: Score
+    inference_speed: Score
+    specialist: Score
+    inference_cost: Score
+    trustworthiness: Score
+    collaborative: Score
+    controlled_generation: Score
+    other: Score
+
+
+class ScoreCard(BaseModel):
+    general: GeneralScores
+    wam: WAMScores
+    rationale: str = Field(description="brief justification grounded in the paper")
+
+
+class InnovationNote(BaseModel):
+    key_idea: str = Field(description="the core technical innovation")
+    transferable_to_wam: str = Field(description="why/how it could transfer to World Action Models")
+
+
+# --- benchmark extraction (Phase 3b) ----------------------------------------
+class ModelVariant(BaseModel):
+    model_name: str = Field(description="model/system name as called in the paper")
+    training_dataset: str | None = Field(
+        default=None, description="dataset(s) it was trained/finetuned on — part of its "
+        "identity; same name on a different dataset is a different system")
+    base_model: str | None = None
+    params: str | None = Field(default=None, description="e.g. '7B', '300M'")
+    modality: str | None = None
+
+
+class BenchmarkRow(BaseModel):
+    model_name: str
+    training_dataset: str | None = None
+    benchmark: str = Field(description="benchmark/dataset the result is on")
+    task: str | None = None
+    split: str | None = None
+    metric_name: str | None = Field(default=None, description="e.g. 'success rate', 'accuracy'")
+    metric_value: float | None = None
+    inference_speed: float | None = Field(default=None, description="numeric value if stated")
+    speed_unit: str | None = Field(default=None, description="e.g. 'Hz', 'fps', 'ms', 'tok/s'")
+    inference_cost: float | None = None
+    cost_unit: str | None = Field(default=None, description="e.g. 'GPU-hours', 'FLOPs', '$'")
+    hardware: str | None = Field(default=None, description="hardware as reported, if any")
+    claimed_by_authors: bool = Field(
+        default=True, description="True if this is the paper's own reported number (vs a "
+        "third-party / comparison number quoted from another paper)")
+    notes: str | None = None
+
+
+class ExtractionResult(BaseModel):
+    """Only extract numbers ACTUALLY stated in the text. Empty lists are fine."""
+    models: list[ModelVariant] = Field(default_factory=list)
+    benchmarks: list[BenchmarkRow] = Field(default_factory=list)

src/wam/pipeline/score.py

@@ -0,0 +1,71 @@
+"""Score stage — two-layer WAM rubric for core papers (strong tier).
+
+Produces ``scores_json = {general, wam, weighted_total, rationale}``. The weighted total is
+computed over *available* (non-"N/A") metrics using config weights; the top-4 WAM metrics are
+weighted 2x by default so they dominate. Capped at ``constants.analyze_cap`` per run.
+"""
+
+from __future__ import annotations
+
+import json
+import sqlite3
+
+from wam.config import Config
+from wam.llm import LLMClient
+from wam.logging import get_logger
+from wam.pipeline.schemas import ScoreCard
+from wam.store import papers as ps
+
+log = get_logger("pipeline.score")
+
+SYSTEM = ("Score the paper on the two-layer WAM rubric. Use the profile's definitions and "
+          "scoring guidance. Output 0-10 per metric, or \"N/A\" when the paper does not address "
+          "a metric (do NOT guess). Be skeptical of self-reported numbers. Write the "
+          "rationale in English.\n\n"
+          "--- PROFILE ---\n{profile}")
+
+
+def weighted_total(card: ScoreCard, cfg: Config) -> float:
+    gw = cfg.get("scoring.general_weights", {}) or {}
+    ww = cfg.get("scoring.wam_weights", {}) or {}
+    num = den = 0.0
+    for metric, weight in gw.items():
+        v = getattr(card.general, metric, "N/A")
+        if isinstance(v, int):
+            num += weight * v
+            den += weight
+    for metric, weight in ww.items():
+        v = getattr(card.wam, metric, "N/A")
+        if isinstance(v, int):
+            num += weight * v
+            den += weight
+    return round(num / den, 2) if den else 0.0
+
+
+def run(cfg: Config, client: LLMClient, conn: sqlite3.Connection,
+        limit: int | None = None) -> int:
+    cap = limit or int(cfg.get("constants.analyze_cap", 40))
+    system = SYSTEM.format(profile=cfg.profile_text())
+    todo = ps.needs_score(conn, limit=cap)
+    log.info("scoring %d analyzed papers (cap=%d)", len(todo), cap)
+    done = 0
+    for row in todo:
+        analysis = conn.execute("SELECT summary_json, analysis_json FROM papers WHERE id=?",
+                                (row["id"],)).fetchone()
+        ctx = (f"Title: {row['title']}\n\nAbstract: {row['abstract'] or '(none)'}\n\n"
+               f"Summary: {analysis['summary_json']}\n\nAnalysis: {analysis['analysis_json']}")
+        try:
+            card = client.complete_json("score", system, ctx, ScoreCard,
+                                        label="score", max_tokens=5000)
+        except Exception as e:  # noqa: BLE001
+            log.warning("score failed for %s: %s", row["id"], e)
+            continue
+        payload = card.model_dump()
+        payload["weighted_total"] = weighted_total(card, cfg)
+        ps.set_scores(conn, row["id"], json.dumps(payload))
+        done += 1
+        if done % 10 == 0:
+            conn.commit()
+    conn.commit()
+    log.info("scored %d papers", done)
+    return done

src/wam/pipeline/summarize.py

@@ -0,0 +1,39 @@
+"""Summarize stage — structured TL;DR for core + adjacent papers (cheap tier)."""
+
+from __future__ import annotations
+
+import sqlite3
+
+from wam.config import Config
+from wam.llm import LLMClient
+from wam.logging import get_logger
+from wam.pipeline.schemas import PaperSummary
+from wam.store import papers as ps
+
+log = get_logger("pipeline.summarize")
+
+SYSTEM = ("Summarize the paper for a World Action Models research digest. Be faithful to the "
+          "abstract; do not invent results. Always write in English, regardless of the "
+          "paper's language.")
+
+
+def run(cfg: Config, client: LLMClient, conn: sqlite3.Connection,
+        limit: int | None = None) -> int:
+    todo = ps.needs_summary(conn, limit=limit)
+    log.info("summarizing %d papers", len(todo))
+    done = 0
+    for row in todo:
+        user = f"Title: {row['title']}\n\nAbstract: {row['abstract'] or '(none)'}"
+        try:
+            s = client.complete_json("summarize", SYSTEM, user, PaperSummary,
+                                     label="summarize", max_tokens=2000)
+        except Exception as e:  # noqa: BLE001
+            log.warning("summarize failed for %s: %s", row["id"], e)
+            continue
+        ps.set_summary(conn, row["id"], s.model_dump_json())
+        done += 1
+        if done % 20 == 0:
+            conn.commit()
+    conn.commit()
+    log.info("summarized %d papers", done)
+    return done

src/wam/pipeline/trends.py

@@ -0,0 +1,108 @@
+"""Trends stage — popular research directions with momentum (long-context, no embeddings).
+
+Feeds the full list of tracked papers (id + title + tldr) to a strong long-context model and
+asks it to cluster them into research directions. Momentum (rising/steady/cooling) is then
+computed in code from members' publication dates (recent vs prior window). Results land in
+the ``fronts`` table and render into the README.
+"""
+
+from __future__ import annotations
+
+import json
+import sqlite3
+from collections import Counter
+from datetime import date, datetime, timedelta
+
+from pydantic import BaseModel, Field
+
+from wam.config import Config
+from wam.llm import LLMClient
+from wam.logging import get_logger
+
+log = get_logger("pipeline.trends")
+
+
+class Direction(BaseModel):
+    name: str = Field(description="a 4-8 word research-direction name")
+    summary: str = Field(description="one sentence describing the direction")
+    members: list[int] = Field(description="indices (from the numbered list) of papers in it")
+
+
+class TrendResult(BaseModel):
+    directions: list[Direction]
+
+
+SYSTEM = ("Cluster these World Action Models papers into coherent research directions "
+          "(aim for 6-12 directions). Every paper index should belong to exactly one "
+          "direction; put genuine outliers in a 'Miscellaneous' direction. Use the numbered "
+          "list; refer to papers only by their index. Write direction names and summaries in "
+          "English.")
+
+
+def _month(d: str | None) -> str:
+    return (d or "")[:7] or "unknown"
+
+
+def _momentum(dates: list[str], window: int, today: date) -> str:
+    def n_between(lo, hi):
+        c = 0
+        for d in dates:
+            if not d:
+                continue
+            try:
+                dd = date.fromisoformat(d)
+            except ValueError:
+                continue
+            if lo < dd <= hi:
+                c += 1
+        return c
+    recent = n_between(today - timedelta(days=window), today)
+    prior = n_between(today - timedelta(days=2 * window), today - timedelta(days=window))
+    if recent > prior * 1.3 and recent >= 2:
+        return "rising"
+    if recent < prior * 0.7:
+        return "cooling"
+    return "steady"
+
+
+def run(cfg: Config, client: LLMClient, conn: sqlite3.Connection) -> int:
+    rows = conn.execute(
+        "SELECT id, title, published, summary_json FROM papers WHERE track IN ('core','adjacent') "
+        "ORDER BY published DESC").fetchall()
+    min_size = int(cfg.get("trends.min_front_size", 3))
+    if len(rows) < min_size:
+        log.info("not enough papers for trends (%d)", len(rows))
+        return 0
+
+    lines = []
+    for i, r in enumerate(rows):
+        tldr = json.loads(r["summary_json"] or "{}").get("tldr", "") if r["summary_json"] else ""
+        lines.append(f"{i}. {r['title']} — {tldr}")
+    log.info("clustering %d papers into directions (long-context)", len(rows))
+    # Clustering 200+ papers is a large-output task: use the fast, JSON-reliable cheap model
+    # (glm-5.1 was too slow here and truncated). Generous token ceiling for the index lists.
+    res = client.complete_json("cheap", SYSTEM, "\n".join(lines), TrendResult,
+                               label="trends", max_tokens=12000)
+    log.info("model returned %d directions, member counts: %s", len(res.directions),
+             [len(d.members) for d in res.directions])
+
+    today = date.today()
+    window = int(cfg.get("trends.window_days", 30))
+    snapshot = today.isoformat()
+    conn.execute("DELETE FROM fronts WHERE snapshot_date=?", (snapshot,))
+    n = 0
+    for ci, d in enumerate(res.directions):
+        members = [rows[i] for i in d.members if 0 <= i < len(rows)]
+        if len(members) < min_size:
+            continue
+        dates = [m["published"] for m in members]
+        conn.execute(
+            "INSERT INTO fronts (front_id, snapshot_date, name, summary, member_ids_json, "
+            "size, momentum, volume_json) VALUES (?,?,?,?,?,?,?,?)",
+            (f"front-{ci}", snapshot, d.name, d.summary,
+             json.dumps([m["id"] for m in members]), len(members),
+             _momentum(dates, window, today), json.dumps(dict(Counter(_month(x) for x in dates)))))
+        n += 1
+    conn.commit()
+    log.info("stored %d research fronts (snapshot %s)", n, snapshot)
+    return n

src/wam/render/__init__.py

@@ -0,0 +1 @@
+"""Rendering: generate README, daily digests, and CSV exports from the SQLite store."""

src/wam/render/readme.py

@@ -0,0 +1,259 @@
+"""Generate the README (awesome list), a daily digest, and a benchmarks CSV export.
+
+All outputs are derived from the SQLite store. Dropped papers stay in the DB/KB but are
+excluded from these human-facing surfaces. A link-integrity check flags records missing the
+required abstract link.
+"""
+
+from __future__ import annotations
+
+import csv
+import json
+import sqlite3
+from datetime import date
+
+from wam.config import Config
+from wam.logging import get_logger
+
+log = get_logger("render")
+
+WAM_TOP4 = ["inference_speed", "generalist", "specialist", "inference_cost"]
+TOP4_LABEL = {"inference_speed": "spd", "generalist": "gen", "specialist": "spec",
+              "inference_cost": "cost"}
+
+
+def _clip(text: str, n: int) -> str:
+    """Truncate at a word boundary with an ellipsis, never mid-word."""
+    text = (text or "").strip()
+    if len(text) <= n:
+        return text
+    cut = text[:n].rsplit(" ", 1)[0]
+    return cut.rstrip(".,;: ") + "…"
+
+
+def _links(links_json: str | None) -> dict:
+    try:
+        return json.loads(links_json or "{}")
+    except Exception:  # noqa: BLE001
+        return {}
+
+
+def _link_md(links: dict) -> str:
+    parts = []
+    for key, label in (("abs", "abs"), ("pdf", "pdf"), ("project_page", "site"),
+                       ("code", "code"), ("doi", "doi")):
+        if links.get(key):
+            parts.append(f"[{label}]({links[key]})")
+    return " · ".join(parts) or "—"
+
+
+def _top4_badge(scores: dict) -> str:
+    wam = (scores or {}).get("wam", {})
+    cells = []
+    for m in WAM_TOP4:
+        v = wam.get(m)
+        cells.append(f"{TOP4_LABEL[m]} {v if isinstance(v, int) else '–'}")
+    return " · ".join(cells)
+
+
+def _fmt_authors(authors_json: str | None, n: int = 3) -> str:
+    a = json.loads(authors_json or "[]")
+    return ", ".join(a[:n]) + (" et al." if len(a) > n else "")
+
+
+# --- sections ---------------------------------------------------------------
+def _core_table(conn: sqlite3.Connection, limit: int = 50) -> str:
+    rows = conn.execute(
+        "SELECT id, title, published, links_json, scores_json FROM papers "
+        "WHERE track='core' AND scores_json IS NOT NULL "
+        "ORDER BY json_extract(scores_json,'$.weighted_total') DESC LIMIT ?", (limit,)
+    ).fetchall()
+    if not rows:
+        return "_No scored papers yet._\n"
+    out = ["| Score | Paper | Published | Top-4 (spd·gen·spec·cost) | Links |",
+           "|------:|-------|-----------|---------------------------|-------|"]
+    for r in rows:
+        s = json.loads(r["scores_json"])
+        out.append(f"| **{s.get('weighted_total','?')}** | {r['title']} | "
+                   f"{r['published'] or '—'} | {_top4_badge(s)} | {_link_md(_links(r['links_json']))} |")
+    return "\n".join(out) + "\n"
+
+
+def _leaderboard(conn: sqlite3.Connection, limit: int = 40) -> str:
+    rows = conn.execute(
+        "SELECT model_name, training_dataset, benchmark, task, metric_name, metric_value, "
+        "claimed_by_authors FROM benchmarks WHERE metric_value IS NOT NULL "
+        "ORDER BY benchmark, metric_value DESC LIMIT ?", (limit,)).fetchall()
+    if not rows:
+        return "_No benchmark results extracted yet._\n"
+    out = ["| Benchmark | Task | Model (training data) | Metric | Value | Source |",
+           "|-----------|------|-----------------------|--------|------:|:------:|"]
+    for r in rows:
+        td = f" _({r['training_dataset']})_" if r["training_dataset"] else ""
+        src = "authors" if r["claimed_by_authors"] else "3rd-party"
+        out.append(f"| {r['benchmark']} | {r['task'] or '—'} | {r['model_name']}{td} | "
+                   f"{r['metric_name'] or '—'} | {r['metric_value']} | {src} |")
+    return "\n".join(out) + "\n"
+
+
+def _innovation(conn: sqlite3.Connection, limit: int = 30) -> str:
+    rows = conn.execute(
+        "SELECT title, links_json, innovation_json FROM papers WHERE track='adjacent' "
+        "AND innovation_json IS NOT NULL ORDER BY relevance DESC LIMIT ?", (limit,)).fetchall()
+    if not rows:
+        return "_No adjacent-track innovations captured yet._\n"
+    out = []
+    for r in rows:
+        inv = json.loads(r["innovation_json"])
+        out.append(f"- **{r['title']}** — {_clip(inv.get('key_idea',''), 320)} "
+                   f"_(→ WAM: {_clip(inv.get('transferable_to_wam',''), 260)})_ "
+                   f"{_link_md(_links(r['links_json']))}")
+    return "\n".join(out) + "\n"
+
+
+def _authors(conn: sqlite3.Connection, limit: int = 25) -> str:
+    rows = conn.execute(
+        "SELECT name, affiliation, citations, paper_ids_json, directions, s2_url FROM authors "
+        "ORDER BY json_array_length(paper_ids_json) DESC LIMIT ?", (limit,)).fetchall()
+    if not rows:
+        return "_No authors aggregated yet._\n"
+    out = []
+    for r in rows:
+        n = len(json.loads(r["paper_ids_json"] or "[]"))
+        name = f"[{r['name']}]({r['s2_url']})" if r["s2_url"] else r["name"]
+        aff = f" · {r['affiliation']}" if r["affiliation"] else ""
+        out.append(f"- **{name}** ({n} papers{aff}) — {_clip(r['directions'], 260)}")
+    return "\n".join(out) + "\n"
+
+
+def _news(conn: sqlite3.Connection, limit: int = 15) -> str:
+    rows = conn.execute(
+        "SELECT title, authors_json, links_json FROM papers WHERE track='news' "
+        "ORDER BY published DESC LIMIT ?", (limit,)).fetchall()
+    if not rows:
+        return "_No news items yet._\n"
+    out = []
+    for r in rows:
+        outlet = (json.loads(r["authors_json"] or "[]") or ["—"])[0]
+        link = _links(r["links_json"]).get("abs", "")
+        out.append(f"- [{r['title']}]({link}) — _{outlet}_")
+    return "\n".join(out) + "\n"
+
+
+_MOM_ICON = {"rising": "📈 rising", "cooling": "📉 cooling", "steady": "➡️ steady"}
+
+
+def _trends(conn: sqlite3.Connection, limit: int = 12) -> str:
+    snap = conn.execute("SELECT max(snapshot_date) FROM fronts").fetchone()[0]
+    if not snap:
+        return "_No trend snapshot yet._\n"
+    rows = conn.execute(
+        "SELECT name, summary, size, momentum FROM fronts WHERE snapshot_date=? "
+        "ORDER BY size DESC LIMIT ?", (snap, limit)).fetchall()
+    if not rows:
+        return "_No research fronts detected._\n"
+    out = ["| Direction | Papers | Momentum | Summary |", "|-----------|-------:|----------|---------|"]
+    for r in rows:
+        out.append(f"| **{r['name']}** | {r['size']} | {_MOM_ICON.get(r['momentum'], r['momentum'])} "
+                   f"| {_clip(r['summary'], 120)} |")
+    return "\n".join(out) + "\n"
+
+
+def _counts(conn: sqlite3.Connection) -> dict:
+    d = dict(conn.execute("SELECT track, count(*) FROM papers GROUP BY track").fetchall())
+    return {"core": d.get("core", 0), "adjacent": d.get("adjacent", 0),
+            "drop": d.get("drop", 0), "news": d.get("news", 0),
+            "benchmarks": conn.execute("SELECT count(*) FROM benchmarks").fetchone()[0],
+            "variants": conn.execute("SELECT count(*) FROM model_variants").fetchone()[0],
+            "authors": conn.execute("SELECT count(*) FROM authors").fetchone()[0]}
+
+
+# --- public -----------------------------------------------------------------
+def link_integrity(conn: sqlite3.Connection) -> list[str]:
+    issues = []
+    for r in conn.execute("SELECT id, links_json FROM papers WHERE track IN ('core','adjacent')"):
+        if not _links(r["links_json"]).get("abs"):
+            issues.append(f"{r['id']}: missing abstract link")
+    return issues
+
+
+def render_readme(cfg: Config, conn: sqlite3.Connection, today: str | None = None) -> str:
+    today = today or date.today().isoformat()
+    c = _counts(conn)
+    md = f"""# Awesome-WAM
+
+> Daily-updated intelligence on **World Action Models** — world models, vision-language-action
+> (VLA) models, action-conditioned video/world generation, robot foundation models, and
+> embodied/physical AI. Auto-generated; do not edit by hand.
+
+**Last updated:** {today} · **Tracked:** {c['core']} core · {c['adjacent']} adjacent ·
+{c['news']} news · **{c['benchmarks']}** benchmark rows across **{c['variants']}** model
+variants · **{c['authors']}** authors
+
+> Scoring: two layers — general (novelty/soundness/impact) + WAM-specific. Top-4 WAM metrics
+> (inference **speed**, **gen**eralist, **spec**ialist, inference **cost**) are weighted 2×.
+> `–` means the paper does not address that metric (we never fabricate a score).
+
+## 📈 Trends & Popular Directions
+{_trends(conn)}
+## 🏆 Top World Action Model Papers
+{_core_table(conn)}
+## 📊 Benchmark Leaderboard
+_Model identity = (name, training dataset); the same name on different data is a distinct row.
+Numbers are as reported; `authors` = self-reported, `3rd-party` = quoted comparison._
+{_leaderboard(conn)}
+## 🔬 Innovation Watch — adjacent fields (VLA / world models / video generation)
+_Not scored; surfaced for techniques transferable to WAM._
+{_innovation(conn)}
+## 👥 Influential Authors & Groups
+{_authors(conn)}
+## 📰 Embodied / Physical-AI News
+{_news(conn)}
+---
+_Generated by [Awesome-WAM](https://github.com/your-org/Awesome-WAM)._
+"""
+    (cfg.root / "README.md").write_text(md, encoding="utf-8")
+    log.info("wrote README.md")
+    return md
+
+
+def render_digest(cfg: Config, conn: sqlite3.Connection, today: str | None = None) -> str:
+    today = today or date.today().isoformat()
+    new_core = conn.execute(
+        "SELECT count(*) FROM papers WHERE track='core' AND first_seen=?", (today,)).fetchone()[0]
+    new_adj = conn.execute(
+        "SELECT count(*) FROM papers WHERE track='adjacent' AND first_seen=?", (today,)).fetchone()[0]
+    rows = conn.execute(
+        "SELECT title, links_json, scores_json, summary_json FROM papers WHERE track='core' "
+        "AND first_seen=? AND scores_json IS NOT NULL "
+        "ORDER BY json_extract(scores_json,'$.weighted_total') DESC LIMIT 15", (today,)).fetchall()
+    lines = [f"# WAM Daily Digest — {today}\n",
+             f"**New today:** {new_core} core · {new_adj} adjacent papers\n",
+             "## Top new papers\n"]
+    if not rows:
+        lines.append("_No new scored core papers today._\n")
+    for r in rows:
+        s = json.loads(r["scores_json"])
+        tldr = json.loads(r["summary_json"] or "{}").get("tldr", "")
+        lines.append(f"### {r['title']}  ·  **{s.get('weighted_total','?')}**\n"
+                     f"{tldr}\n\n_{_top4_badge(s)}_ · {_link_md(_links(r['links_json']))}\n")
+    md = "\n".join(lines)
+    out = cfg.root / "data" / "digests"
+    out.mkdir(parents=True, exist_ok=True)
+    (out / f"{today}.md").write_text(md, encoding="utf-8")
+    log.info("wrote digest %s.md", today)
+    return md
+
+
+def export_benchmarks_csv(cfg: Config, conn: sqlite3.Connection) -> None:
+    cols = ["variant_key", "model_name", "training_dataset", "benchmark", "task", "split",
+            "metric_name", "metric_value", "inference_speed", "speed_unit", "inference_cost",
+            "cost_unit", "hardware", "source_paper_id", "claimed_by_authors", "notes",
+            "extracted_on"]
+    rows = conn.execute(f"SELECT {','.join(cols)} FROM benchmarks ORDER BY benchmark, variant_key")
+    path = cfg.root / "data" / "benchmarks.csv"
+    with open(path, "w", newline="", encoding="utf-8") as f:
+        w = csv.writer(f)
+        w.writerow(cols)
+        w.writerows(rows)
+    log.info("exported %s", path)

src/wam/sources/__init__.py

@@ -0,0 +1 @@
+"""Paper/news source adapters. Each returns normalized ``PaperRecord`` objects."""

src/wam/sources/arxiv.py

@@ -0,0 +1,83 @@
+"""arXiv source — the core candidate feed.
+
+Queries the arXiv Atom API for recent papers in the configured categories matching a broad
+keyword net (WAM + adjacent fields). The LLM filter makes the final core/adjacent/drop call
+later; here we just cast a wide, recall-oriented net within the lookback window.
+"""
+
+from __future__ import annotations
+
+import time
+from datetime import date, datetime, timedelta
+
+import feedparser
+import requests
+
+from wam.config import Config
+from wam.logging import get_logger
+from wam.models import Links, PaperRecord
+
+log = get_logger("source.arxiv")
+
+API = "http://export.arxiv.org/api/query"
+
+
+def _build_query(categories: list[str], keywords: list[str]) -> str:
+    cats = " OR ".join(f"cat:{c}" for c in categories)
+    kws = " OR ".join(f'all:"{k}"' for k in keywords)
+    return f"({cats}) AND ({kws})"
+
+
+def _arxiv_id(entry) -> str:
+    # entry.id like http://arxiv.org/abs/2506.01234v2 -> 2506.01234
+    raw = entry.id.rsplit("/abs/", 1)[-1]
+    return raw.split("v")[0] if "v" in raw else raw
+
+
+def fetch(cfg: Config, *, max_results: int | None = None,
+          lookback_days: int | None = None) -> list[PaperRecord]:
+    scfg = cfg.get("sources.arxiv", {}) or {}
+    if not scfg.get("enabled", True):
+        return []
+    categories = scfg.get("categories", ["cs.RO", "cs.AI", "cs.LG", "cs.CV"])
+    keywords = (cfg.get("keywords.core", []) or []) + (cfg.get("keywords.adjacent", []) or [])
+    max_results = max_results or scfg.get("max_results", 300)
+    lookback_days = lookback_days or cfg.get("constants.lookback_days", 60)
+    cutoff = date.today() - timedelta(days=lookback_days)
+
+    params = {
+        "search_query": _build_query(categories, keywords),
+        "start": 0,
+        "max_results": max_results,
+        "sortBy": "submittedDate",
+        "sortOrder": "descending",
+    }
+    log.info("arxiv query: %s (max=%d, since=%s)", params["search_query"], max_results, cutoff)
+    resp = requests.get(API, params=params, timeout=cfg.get("constants.request_timeout", 90))
+    resp.raise_for_status()
+    feed = feedparser.parse(resp.text)
+
+    records: list[PaperRecord] = []
+    for e in feed.entries:
+        try:
+            published = datetime(*e.published_parsed[:6]).date()
+        except Exception:  # noqa: BLE001
+            published = None
+        if published and published < cutoff:
+            continue
+        aid = _arxiv_id(e)
+        pdf = next((l.href for l in getattr(e, "links", []) if l.get("type") == "application/pdf"),
+                   f"https://arxiv.org/pdf/{aid}")
+        records.append(PaperRecord(
+            id=f"arxiv:{aid}",
+            source="arxiv",
+            title=" ".join(e.title.split()),
+            authors=[a.name for a in getattr(e, "authors", [])],
+            published=published.isoformat() if published else None,
+            abstract=" ".join(e.summary.split()) if hasattr(e, "summary") else None,
+            categories=[t.term for t in getattr(e, "tags", [])],
+            links=Links(abs=f"https://arxiv.org/abs/{aid}", pdf=pdf),
+        ))
+    log.info("arxiv returned %d entries, %d within lookback", len(feed.entries), len(records))
+    time.sleep(3)  # be polite to arXiv
+    return records

src/wam/sources/news.py

@@ -0,0 +1,58 @@
+"""Embodied / physical-AI news via RSS.
+
+News items are normalized into ``PaperRecord`` with ``source='news'`` so they flow through
+the same store. They get a light summary later but not the full WAM rubric.
+"""
+
+from __future__ import annotations
+
+import hashlib
+from datetime import date, datetime, timedelta
+
+import feedparser
+
+from wam.config import Config
+from wam.logging import get_logger
+from wam.models import Links, PaperRecord
+
+log = get_logger("source.news")
+
+
+def fetch(cfg: Config, *, lookback_days: int | None = None) -> list[PaperRecord]:
+    ncfg = cfg.get("sources.news", {}) or {}
+    if not ncfg.get("enabled", True):
+        return []
+    feeds = ncfg.get("feeds", []) or []
+    lookback_days = lookback_days or cfg.get("constants.lookback_days", 60)
+    cutoff = date.today() - timedelta(days=lookback_days)
+
+    records: list[PaperRecord] = []
+    for url in feeds:
+        try:
+            feed = feedparser.parse(url)
+        except Exception as e:  # noqa: BLE001
+            log.warning("news feed failed %s: %s", url, e)
+            continue
+        outlet = feed.feed.get("title", url) if hasattr(feed, "feed") else url
+        for e in feed.entries:
+            try:
+                published = datetime(*e.published_parsed[:6]).date()
+            except Exception:  # noqa: BLE001
+                published = None
+            if published and published < cutoff:
+                continue
+            link = getattr(e, "link", "")
+            uid = hashlib.sha1(link.encode("utf-8")).hexdigest()[:16]
+            summary = getattr(e, "summary", "") or ""
+            records.append(PaperRecord(
+                id=f"news:{uid}",
+                source="news",
+                title=" ".join(getattr(e, "title", "(untitled)").split()),
+                authors=[outlet],
+                published=published.isoformat() if published else None,
+                abstract=" ".join(summary.split())[:2000],
+                categories=["news"],
+                links=Links(abs=link),
+            ))
+    log.info("news: %d items within lookback from %d feeds", len(records), len(feeds))
+    return records

src/wam/sources/papers_with_code.py

@@ -0,0 +1,58 @@
+"""Papers with Code enrichment — has-code / repo link signal.
+
+Best-effort lookup by arXiv id. This is one HTTP call per paper, so callers should pass only
+the shortlist (e.g. post-filter core/adjacent papers) and respect ``max_lookups``.
+"""
+
+from __future__ import annotations
+
+import time
+
+import requests
+
+from wam.config import Config
+from wam.logging import get_logger
+from wam.models import PaperRecord
+
+log = get_logger("source.pwc")
+
+API = "https://paperswithcode.com/api/v1/papers/"
+
+
+def _repo_for_arxiv(aid: str, timeout: int) -> str | None:
+    try:
+        r = requests.get(API, params={"arxiv_id": aid}, timeout=timeout)
+        r.raise_for_status()
+        results = r.json().get("results") or []
+        if not results:
+            return None
+        pid = results[0]["id"]
+        rr = requests.get(f"{API}{pid}/repositories/", timeout=timeout)
+        rr.raise_for_status()
+        repos = rr.json().get("results") or []
+        if not repos:
+            return None
+        # Prefer the official repo if flagged, else the most-starred.
+        repos.sort(key=lambda x: (x.get("is_official", False), x.get("stars", 0)), reverse=True)
+        return repos[0].get("url")
+    except Exception as e:  # noqa: BLE001
+        log.debug("pwc lookup failed for %s: %s", aid, e)
+        return None
+
+
+def enrich(cfg: Config, records: list[PaperRecord], *, max_lookups: int = 60) -> list[PaperRecord]:
+    if not (cfg.get("sources.papers_with_code", {}) or {}).get("enabled", True):
+        return records
+    timeout = cfg.get("constants.request_timeout", 90)
+    found = 0
+    for rec in records[:max_lookups]:
+        if not rec.arxiv_id or rec.links.code:
+            continue
+        url = _repo_for_arxiv(rec.arxiv_id, timeout)
+        if url:
+            rec.links.code = url
+            rec.has_code = True
+            found += 1
+        time.sleep(0.5)
+    log.info("pwc found code for %d papers", found)
+    return records

src/wam/sources/pdf.py

@@ -0,0 +1,53 @@
+"""PDF text extraction with on-disk caching.
+
+Benchmark numbers live in the paper body (tables), not the abstract, so the extractor needs
+full text. We download each PDF once, extract text with PyMuPDF, and cache it keyed by a
+content hash so arXiv revisions are re-extracted but repeats are free.
+"""
+
+from __future__ import annotations
+
+import hashlib
+from pathlib import Path
+
+import requests
+
+from wam.config import Config
+from wam.logging import get_logger
+
+log = get_logger("source.pdf")
+
+
+def _cache_dir(cfg: Config) -> Path:
+    d = cfg.path("pdf_cache")
+    d.mkdir(parents=True, exist_ok=True)
+    return d
+
+
+def get_text(cfg: Config, paper_id: str, pdf_url: str, *, max_chars: int = 60000) -> str | None:
+    """Return cached/extracted text for a paper (truncated to max_chars), or None on failure."""
+    safe = paper_id.replace("/", "_").replace(":", "_")
+    cache = _cache_dir(cfg) / f"{safe}.txt"
+    if cache.exists():
+        return cache.read_text(encoding="utf-8")[:max_chars]
+    if not pdf_url:
+        return None
+    try:
+        import fitz  # PyMuPDF
+    except ImportError:  # pragma: no cover
+        log.warning("PyMuPDF not installed; skipping PDF extraction")
+        return None
+    try:
+        resp = requests.get(pdf_url, timeout=cfg.get("constants.request_timeout", 90),
+                            headers={"User-Agent": "Awesome-WAM/0.1"})
+        resp.raise_for_status()
+        with fitz.open(stream=resp.content, filetype="pdf") as doc:
+            text = "\n".join(page.get_text() for page in doc)
+    except Exception as e:  # noqa: BLE001
+        log.warning("PDF extraction failed for %s: %s", paper_id, e)
+        return None
+    cache.write_text(text, encoding="utf-8")
+    h = hashlib.sha256(resp.content).hexdigest()
+    cache.with_suffix(".hash").write_text(h, encoding="utf-8")
+    log.debug("cached PDF text for %s (%d chars)", paper_id, len(text))
+    return text[:max_chars]

src/wam/sources/semantic_scholar.py

@@ -0,0 +1,142 @@
+"""Semantic Scholar enrichment.
+
+Adds citation signals (total + influential) and a DOI/open-access PDF link to existing
+records, keyed by arXiv id via the batch endpoint. Best-effort: failures leave records as-is.
+"""
+
+from __future__ import annotations
+
+import os
+import time
+
+import requests
+
+from wam.config import Config
+from wam.logging import get_logger
+from wam.models import PaperRecord
+
+log = get_logger("source.s2")
+
+BATCH = "https://api.semanticscholar.org/graph/v1/paper/batch"
+FIELDS = "citationCount,influentialCitationCount,externalIds,openAccessPdf"
+AUTHOR_SEARCH = "https://api.semanticscholar.org/graph/v1/author/search"
+
+
+def _headers(cfg: Config) -> dict:
+    scfg = cfg.get("sources.semantic_scholar", {}) or {}
+    key = os.environ.get(scfg.get("api_key_env", "SEMANTIC_SCHOLAR_API_KEY") or "")
+    return {"x-api-key": key} if key else {}
+
+
+def fetch_authors(cfg: Config, arxiv_ids: list[str]) -> dict[str, list[dict]]:
+    """Map arxiv_id -> list of authors with reliable S2 ids (authorId, name, affiliations,
+    hIndex, citationCount), via the batch paper endpoint. Best-effort; missing -> absent."""
+    if not arxiv_ids:
+        return {}
+    fields = "authors.authorId,authors.name,authors.affiliations,authors.hIndex,authors.citationCount"
+    out: dict[str, list[dict]] = {}
+    headers = _headers(cfg)
+    timeout = cfg.get("constants.request_timeout", 90)
+    # Small chunks + generous backoff because the keyless S2 limit is tight (429-prone).
+    for start in range(0, len(arxiv_ids), 100):
+        chunk = arxiv_ids[start:start + 100]
+        ids = [f"ARXIV:{a}" for a in chunk]
+        data = None
+        for attempt, wait in enumerate((0, 6, 15, 30)):
+            if wait:
+                time.sleep(wait)
+            try:
+                r = requests.post(BATCH, params={"fields": fields}, json={"ids": ids},
+                                  headers=headers, timeout=timeout)
+                if r.status_code == 429:
+                    log.info("s2 429, backing off (attempt %d)", attempt + 1)
+                    continue
+                r.raise_for_status()
+                data = r.json()
+                break
+            except Exception as e:  # noqa: BLE001
+                log.warning("s2 author batch error (attempt %d): %s", attempt + 1, e)
+        if not data:
+            continue
+        for aid, item in zip(chunk, data):
+            if item and item.get("authors"):
+                out[f"arxiv:{aid}"] = item["authors"]
+        time.sleep(1.5)
+    log.info("s2 fetched authors for %d/%d papers", len(out), len(arxiv_ids))
+    return out
+
+
+def search_author(cfg: Config, name: str) -> dict | None:
+    """Best-effort: return {name, affiliation, citations, h_index, url} for a name, or None."""
+    try:
+        r = requests.get(AUTHOR_SEARCH, params={
+            "query": name, "fields": "name,affiliations,hIndex,citationCount,url", "limit": 1},
+            headers=_headers(cfg), timeout=cfg.get("constants.request_timeout", 90))
+        if r.status_code == 429:
+            time.sleep(3)
+            return None
+        r.raise_for_status()
+        data = (r.json().get("data") or [])
+        if not data:
+            return None
+        a = data[0]
+        affs = a.get("affiliations") or []
+        return {"id": a.get("authorId"), "name": a.get("name"),
+                "affiliation": affs[0] if affs else None,
+                "citations": a.get("citationCount"), "h_index": a.get("hIndex"),
+                "url": a.get("url")}
+    except Exception as e:  # noqa: BLE001
+        log.debug("s2 author search failed for %s: %s", name, e)
+        return None
+
+
+def enrich(cfg: Config, records: list[PaperRecord]) -> list[PaperRecord]:
+    scfg = cfg.get("sources.semantic_scholar", {}) or {}
+    if not scfg.get("enabled", True):
+        return records
+    arxiv_recs = [r for r in records if r.arxiv_id]
+    if not arxiv_recs:
+        return records
+    by_aid = {r.arxiv_id: r for r in arxiv_recs}
+    ids = [f"ARXIV:{aid}" for aid in by_aid]
+
+    headers = {}
+    key = os.environ.get(scfg.get("api_key_env", "SEMANTIC_SCHOLAR_API_KEY") or "")
+    if key:
+        headers["x-api-key"] = key
+
+    enriched = 0
+    for chunk_start in range(0, len(ids), 500):  # batch endpoint caps at 500 ids
+        chunk = ids[chunk_start:chunk_start + 500]
+        try:
+            resp = requests.post(BATCH, params={"fields": FIELDS}, json={"ids": chunk},
+                                 headers=headers, timeout=cfg.get("constants.request_timeout", 90))
+            if resp.status_code == 429:
+                log.warning("s2 rate-limited; backing off")
+                time.sleep(5)
+                resp = requests.post(BATCH, params={"fields": FIELDS}, json={"ids": chunk},
+                                     headers=headers, timeout=90)
+            resp.raise_for_status()
+            data = resp.json()
+        except Exception as e:  # noqa: BLE001
+            log.warning("s2 batch failed: %s", e)
+            continue
+        for s2_id, item in zip(chunk, data):
+            if not item:
+                continue
+            aid = s2_id.split("ARXIV:", 1)[1]
+            rec = by_aid.get(aid)
+            if not rec:
+                continue
+            rec.citations = item.get("citationCount") or 0
+            rec.influential_citations = item.get("influentialCitationCount") or 0
+            ext = item.get("externalIds") or {}
+            if ext.get("DOI"):
+                rec.links.doi = f"https://doi.org/{ext['DOI']}"
+            oa = item.get("openAccessPdf") or {}
+            if oa.get("url") and not rec.links.pdf:
+                rec.links.pdf = oa["url"]
+            enriched += 1
+        time.sleep(1)
+    log.info("s2 enriched %d/%d records", enriched, len(arxiv_recs))
+    return records

src/wam/store/__init__.py

@@ -0,0 +1,3 @@
+from wam.store.db import Database, connect
+
+__all__ = ["Database", "connect"]

src/wam/store/benchmarks.py

@@ -0,0 +1,70 @@
+"""Benchmark + model-variant persistence.
+
+Model identity = (model_name, training_dataset) -> a slugified ``variant_key``, so the same
+model name finetuned on a different dataset is tracked as a distinct system. Benchmark rows
+are append-and-merge (UNIQUE constraint dedups identical rows) so conflicting numbers from
+different papers stay visible side by side.
+"""
+
+from __future__ import annotations
+
+import re
+import sqlite3
+from datetime import date, datetime
+
+from wam.pipeline.schemas import BenchmarkRow, ModelVariant
+
+
+def variant_key(model_name: str, training_dataset: str | None) -> str:
+    base = f"{model_name}|{training_dataset or 'unknown'}".lower()
+    return re.sub(r"[^a-z0-9]+", "-", base).strip("-")
+
+
+def upsert_variant(conn: sqlite3.Connection, v: ModelVariant, source_paper_id: str) -> str:
+    vk = variant_key(v.model_name, v.training_dataset)
+    conn.execute(
+        """INSERT INTO model_variants
+           (variant_key, model_name, training_dataset, base_model, params, modality,
+            source_paper_id, updated_at)
+           VALUES (?,?,?,?,?,?,?,?)
+           ON CONFLICT(variant_key) DO UPDATE SET
+             base_model=COALESCE(excluded.base_model, base_model),
+             params=COALESCE(excluded.params, params),
+             modality=COALESCE(excluded.modality, modality),
+             updated_at=excluded.updated_at""",
+        (vk, v.model_name, v.training_dataset, v.base_model, v.params, v.modality,
+         source_paper_id, datetime.now().isoformat(timespec="seconds")),
+    )
+    return vk
+
+
+def insert_benchmark(conn: sqlite3.Connection, row: BenchmarkRow, source_paper_id: str) -> None:
+    vk = variant_key(row.model_name, row.training_dataset)
+    conn.execute(
+        """INSERT OR IGNORE INTO benchmarks
+           (variant_key, model_name, training_dataset, benchmark, task, split, metric_name,
+            metric_value, inference_speed, speed_unit, inference_cost, cost_unit, hardware,
+            source_paper_id, claimed_by_authors, notes, extracted_on)
+           VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""",
+        (vk, row.model_name, row.training_dataset, row.benchmark, row.task, row.split,
+         row.metric_name, row.metric_value, row.inference_speed, row.speed_unit,
+         row.inference_cost, row.cost_unit, row.hardware, source_paper_id,
+         int(row.claimed_by_authors), row.notes, date.today().isoformat()),
+    )
+
+
+def store_extraction(conn: sqlite3.Connection, paper_id: str, models: list[ModelVariant],
+                     rows: list[BenchmarkRow]) -> tuple[int, int]:
+    for v in models:
+        upsert_variant(conn, v, paper_id)
+    # Ensure any variant referenced by a row exists even if not in `models`.
+    seen = {variant_key(v.model_name, v.training_dataset) for v in models}
+    for r in rows:
+        vk = variant_key(r.model_name, r.training_dataset)
+        if vk not in seen:
+            upsert_variant(conn, ModelVariant(model_name=r.model_name,
+                                              training_dataset=r.training_dataset), paper_id)
+            seen.add(vk)
+        insert_benchmark(conn, r, paper_id)
+    conn.execute("UPDATE papers SET benchmarks_extracted=1 WHERE id=?", (paper_id,))
+    return len(models), len(rows)

src/wam/store/db.py

@@ -0,0 +1,69 @@
+"""SQLite access layer.
+
+The DB is the canonical store; everything else (README, web app, digests) is exported from
+it. This module just handles connection + schema init + a couple of low-level helpers; the
+domain-specific upserts (papers, benchmarks, people, fronts) live in their own store modules
+added in later phases.
+"""
+
+from __future__ import annotations
+
+import sqlite3
+from datetime import date, datetime
+from pathlib import Path
+
+from wam.config import Config, load_config
+from wam.logging import get_logger
+
+log = get_logger("store")
+
+SCHEMA_PATH = Path(__file__).with_name("schema.sql")
+
+
+def connect(db_path: str | Path) -> sqlite3.Connection:
+    """Open a connection with row access by name and schema applied."""
+    db_path = Path(db_path)
+    db_path.parent.mkdir(parents=True, exist_ok=True)
+    conn = sqlite3.connect(db_path)
+    conn.row_factory = sqlite3.Row
+    conn.executescript(SCHEMA_PATH.read_text(encoding="utf-8"))
+    _migrate(conn)
+    return conn
+
+
+def _migrate(conn: sqlite3.Connection) -> None:
+    """Idempotent additive migrations for DBs created before a column existed."""
+    cols = {r["name"] for r in conn.execute("PRAGMA table_info(papers)")}
+    if "benchmarks_extracted" not in cols:
+        conn.execute("ALTER TABLE papers ADD COLUMN benchmarks_extracted INTEGER DEFAULT 0")
+    conn.commit()
+
+
+class Database:
+    """Convenience wrapper around a connection bound to the configured DB path."""
+
+    def __init__(self, config: Config | None = None, db_path: str | Path | None = None):
+        self.cfg = config or load_config()
+        self.path = Path(db_path) if db_path else self.cfg.path("db")
+        self.conn = connect(self.path)
+        log.debug("opened db at %s", self.path)
+
+    def __enter__(self) -> "Database":
+        return self
+
+    def __exit__(self, *exc) -> None:
+        self.close()
+
+    def close(self) -> None:
+        self.conn.commit()
+        self.conn.close()
+
+    def log_run(self, stage: str, n_in: int, n_out: int, cost_usd: float = 0.0,
+                notes: str = "") -> None:
+        self.conn.execute(
+            "INSERT INTO runs (run_date, stage, n_in, n_out, cost_usd, notes, created_at) "
+            "VALUES (?,?,?,?,?,?,?)",
+            (date.today().isoformat(), stage, n_in, n_out, cost_usd, notes,
+             datetime.now().isoformat(timespec="seconds")),
+        )
+        self.conn.commit()

src/wam/store/papers.py

@@ -0,0 +1,121 @@
+"""Paper persistence: map ``PaperRecord`` <-> the ``papers`` table."""
+
+from __future__ import annotations
+
+import json
+import sqlite3
+from datetime import date, datetime
+
+from wam.models import PaperRecord
+
+
+def existing_ids(conn: sqlite3.Connection) -> set[str]:
+    return {row["id"] for row in conn.execute("SELECT id FROM papers")}
+
+
+def insert_new(conn: sqlite3.Connection, rec: PaperRecord, *, first_seen: str | None = None) -> None:
+    """Insert a freshly-fetched record. No-op if the id already exists."""
+    first_seen = first_seen or date.today().isoformat()
+    conn.execute(
+        """INSERT OR IGNORE INTO papers
+           (id, source, title, authors_json, published, first_seen, abstract, categories_json,
+            links_json, citations, influential_citations, has_code, status, updated_at)
+           VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?)""",
+        (
+            rec.id, rec.source, rec.title, json.dumps(rec.authors), rec.published, first_seen,
+            rec.abstract, json.dumps(rec.categories), rec.links.model_dump_json(),
+            rec.citations, rec.influential_citations, int(rec.has_code), "new",
+            datetime.now().isoformat(timespec="seconds"),
+        ),
+    )
+
+
+def update_enrichment(conn: sqlite3.Connection, rec: PaperRecord) -> None:
+    """Refresh citation/code/link fields for an existing record (idempotent)."""
+    conn.execute(
+        """UPDATE papers SET citations=?, influential_citations=?, has_code=?, links_json=?,
+           updated_at=? WHERE id=?""",
+        (rec.citations, rec.influential_citations, int(rec.has_code), rec.links.model_dump_json(),
+         datetime.now().isoformat(timespec="seconds"), rec.id),
+    )
+
+
+def count(conn: sqlite3.Connection) -> int:
+    return conn.execute("SELECT count(*) AS c FROM papers").fetchone()["c"]
+
+
+# --- stage work-queues -------------------------------------------------------
+# Each returns rows (id, title, abstract, ...) that still need a given stage. Relevance
+# acts as a cache: a paper with track already set is not re-filtered.
+
+def _rows(conn: sqlite3.Connection, where: str, params: tuple = ()) -> list[sqlite3.Row]:
+    return conn.execute(
+        f"SELECT id, source, title, abstract, published, categories_json FROM papers "
+        f"WHERE {where}", params).fetchall()
+
+
+def _limit_clause(where: str, limit: int | None) -> str:
+    return where + (f" LIMIT {int(limit)}" if limit else "")
+
+
+def needs_filter(conn: sqlite3.Connection, limit: int | None = None) -> list[sqlite3.Row]:
+    # track NULL = unfiltered; relevance=-1 = a prior failure to retry.
+    return _rows(conn, _limit_clause("track IS NULL OR relevance = -1", limit))
+
+
+def needs_summary(conn: sqlite3.Connection, limit: int | None = None) -> list[sqlite3.Row]:
+    return _rows(conn, _limit_clause(
+        "track IN ('core','adjacent') AND summary_json IS NULL", limit))
+
+
+def needs_analysis(conn: sqlite3.Connection, limit: int | None = None) -> list[sqlite3.Row]:
+    q = "track = 'core' AND analysis_json IS NULL"
+    if limit:
+        q += f" ORDER BY relevance DESC LIMIT {int(limit)}"
+    return _rows(conn, q)
+
+
+def needs_score(conn: sqlite3.Connection, limit: int | None = None) -> list[sqlite3.Row]:
+    q = "track = 'core' AND analysis_json IS NOT NULL AND scores_json IS NULL"
+    if limit:
+        q += f" ORDER BY relevance DESC LIMIT {int(limit)}"
+    return _rows(conn, q)
+
+
+def needs_innovation(conn: sqlite3.Connection, limit: int | None = None) -> list[sqlite3.Row]:
+    return _rows(conn, _limit_clause("track = 'adjacent' AND innovation_json IS NULL", limit))
+
+
+def needs_extract(conn: sqlite3.Connection, limit: int | None = None) -> list[sqlite3.Row]:
+    # core papers that have been analyzed but not yet benchmark-extracted
+    return conn.execute(_limit_clause(
+        "SELECT id, source, title, abstract, links_json FROM papers WHERE track='core' "
+        "AND analysis_json IS NOT NULL AND benchmarks_extracted=0", limit)).fetchall()
+
+
+# --- stage writers -----------------------------------------------------------
+def _touch(conn: sqlite3.Connection, pid: str, **cols) -> None:
+    cols["updated_at"] = datetime.now().isoformat(timespec="seconds")
+    sets = ", ".join(f"{k}=?" for k in cols)
+    conn.execute(f"UPDATE papers SET {sets} WHERE id=?", (*cols.values(), pid))
+
+
+def set_filter(conn, pid: str, track: str, relevance: float, reason: str) -> None:
+    _touch(conn, pid, track=track, relevance=relevance, relevance_reason=reason,
+           status="filtered")
+
+
+def set_summary(conn, pid: str, summary_json: str) -> None:
+    _touch(conn, pid, summary_json=summary_json, status="summarized")
+
+
+def set_analysis(conn, pid: str, analysis_json: str) -> None:
+    _touch(conn, pid, analysis_json=analysis_json, status="analyzed")
+
+
+def set_scores(conn, pid: str, scores_json: str) -> None:
+    _touch(conn, pid, scores_json=scores_json, status="scored")
+
+
+def set_innovation(conn, pid: str, innovation_json: str) -> None:
+    _touch(conn, pid, innovation_json=innovation_json, status="done")

src/wam/store/people.py

@@ -0,0 +1,50 @@
+"""Authors + groups persistence."""
+
+from __future__ import annotations
+
+import json
+import re
+import sqlite3
+from datetime import datetime
+
+
+def slug(name: str) -> str:
+    return re.sub(r"[^a-z0-9]+", "-", name.lower()).strip("-")
+
+
+def upsert_author(conn: sqlite3.Connection, *, author_id: str, name: str,
+                  affiliation: str | None, s2_url: str | None, citations: int | None,
+                  h_index: int | None, paper_ids: list[str], directions: str | None) -> None:
+    conn.execute(
+        """INSERT INTO authors
+           (id, name, affiliation, s2_url, citations, h_index, paper_ids_json, directions,
+            updated_at)
+           VALUES (?,?,?,?,?,?,?,?,?)
+           ON CONFLICT(id) DO UPDATE SET
+             affiliation=COALESCE(excluded.affiliation, affiliation),
+             s2_url=COALESCE(excluded.s2_url, s2_url),
+             citations=COALESCE(excluded.citations, citations),
+             h_index=COALESCE(excluded.h_index, h_index),
+             paper_ids_json=excluded.paper_ids_json,
+             directions=COALESCE(excluded.directions, directions),
+             updated_at=excluded.updated_at""",
+        (author_id, name, affiliation, s2_url, citations, h_index, json.dumps(paper_ids),
+         directions, datetime.now().isoformat(timespec="seconds")),
+    )
+
+
+def upsert_group(conn: sqlite3.Connection, *, group_id: str, name: str,
+                 affiliation: str | None, member_ids: list[str], directions: str | None,
+                 notable: list[str]) -> None:
+    conn.execute(
+        """INSERT INTO groups (id, name, affiliation, member_ids_json, directions,
+            notable_json, updated_at)
+           VALUES (?,?,?,?,?,?,?)
+           ON CONFLICT(id) DO UPDATE SET
+             member_ids_json=excluded.member_ids_json,
+             directions=COALESCE(excluded.directions, directions),
+             notable_json=excluded.notable_json,
+             updated_at=excluded.updated_at""",
+        (group_id, name, affiliation, json.dumps(member_ids), directions, json.dumps(notable),
+         datetime.now().isoformat(timespec="seconds")),
+    )

src/wam/store/schema.sql

@@ -0,0 +1,127 @@
+-- Canonical SQLite store for Awesome-WAM.
+-- JSON/markdown exports (README, web app, digests) are generated FROM this DB.
+
+PRAGMA journal_mode = WAL;
+PRAGMA foreign_keys = ON;
+
+-- One row per discovered item (paper or news).
+CREATE TABLE IF NOT EXISTS papers (
+    id                     TEXT PRIMARY KEY,         -- e.g. "arxiv:2506.01234", "news:<hash>"
+    source                 TEXT NOT NULL,            -- arxiv | semantic_scholar | papers_with_code | news
+    title                  TEXT NOT NULL,
+    authors_json           TEXT,                     -- JSON array of author names
+    published              TEXT,                     -- ISO date
+    first_seen             TEXT NOT NULL,            -- ISO date we first ingested it
+    abstract               TEXT,
+    categories_json        TEXT,                     -- JSON array
+    track                  TEXT,                     -- core | adjacent | news | drop (NULL = unfiltered)
+    links_json             TEXT,                     -- JSON {abs,pdf,project_page,code,doi}
+    citations              INTEGER DEFAULT 0,
+    influential_citations  INTEGER DEFAULT 0,
+    has_code               INTEGER DEFAULT 0,        -- bool
+    relevance              REAL,                     -- 0..1; -1 means scoring failed (retry next run)
+    relevance_reason       TEXT,
+    summary_json           TEXT,                     -- JSON {tldr,problem,method,results}
+    analysis_json          TEXT,                     -- JSON {contributions,limitations,wam_relevance}
+    innovation_json        TEXT,                     -- JSON {key_idea,transferable_to_wam} (adjacent)
+    scores_json            TEXT,                     -- JSON {general,wam,weighted_total,rationale}
+    status                 TEXT DEFAULT 'new',       -- new|filtered|summarized|analyzed|scored|done
+    pdf_hash               TEXT,
+    benchmarks_extracted   INTEGER DEFAULT 0,        -- bool: benchmark/model extraction done
+    updated_at             TEXT
+);
+CREATE INDEX IF NOT EXISTS idx_papers_track ON papers(track);
+CREATE INDEX IF NOT EXISTS idx_papers_published ON papers(published);
+CREATE INDEX IF NOT EXISTS idx_papers_status ON papers(status);
+
+-- Normalized benchmark rows. Model identity = (model_name, training_dataset) -> variant_key.
+CREATE TABLE IF NOT EXISTS benchmarks (
+    id                 INTEGER PRIMARY KEY AUTOINCREMENT,
+    variant_key        TEXT NOT NULL,
+    model_name         TEXT NOT NULL,
+    training_dataset   TEXT,
+    benchmark          TEXT NOT NULL,
+    task               TEXT,
+    split              TEXT,
+    metric_name        TEXT,
+    metric_value       REAL,
+    inference_speed    REAL,
+    speed_unit         TEXT,
+    inference_cost     REAL,
+    cost_unit          TEXT,
+    hardware           TEXT,
+    source_paper_id    TEXT REFERENCES papers(id),
+    claimed_by_authors INTEGER DEFAULT 1,
+    notes              TEXT,
+    extracted_on       TEXT,
+    UNIQUE(variant_key, benchmark, task, metric_name, source_paper_id)
+);
+CREATE INDEX IF NOT EXISTS idx_bench_variant ON benchmarks(variant_key);
+
+-- Model variant registry (the disambiguated identity).
+CREATE TABLE IF NOT EXISTS model_variants (
+    variant_key      TEXT PRIMARY KEY,
+    model_name       TEXT NOT NULL,
+    training_dataset TEXT,
+    base_model       TEXT,
+    params           TEXT,
+    modality         TEXT,
+    source_paper_id  TEXT REFERENCES papers(id),
+    updated_at       TEXT
+);
+
+-- Influential authors.
+CREATE TABLE IF NOT EXISTS authors (
+    id              TEXT PRIMARY KEY,                -- semantic scholar id or slug
+    name            TEXT NOT NULL,
+    affiliation     TEXT,
+    s2_url          TEXT,
+    citations       INTEGER,
+    h_index         INTEGER,
+    paper_ids_json  TEXT,                            -- JSON array of tracked paper ids
+    directions      TEXT,                            -- LLM summary of research directions
+    updated_at      TEXT
+);
+
+-- Research groups / labs.
+CREATE TABLE IF NOT EXISTS groups (
+    id              TEXT PRIMARY KEY,
+    name            TEXT NOT NULL,
+    affiliation     TEXT,
+    member_ids_json TEXT,
+    directions      TEXT,
+    notable_json    TEXT,                            -- JSON array of notable paper ids
+    updated_at      TEXT
+);
+
+-- Research fronts (trends), snapshot-versioned.
+CREATE TABLE IF NOT EXISTS fronts (
+    front_id          TEXT NOT NULL,
+    snapshot_date     TEXT NOT NULL,
+    name              TEXT,
+    summary           TEXT,
+    member_ids_json   TEXT,
+    size              INTEGER,
+    momentum          TEXT,                          -- rising | steady | cooling
+    volume_json       TEXT,                          -- per-window counts
+    PRIMARY KEY (front_id, snapshot_date)
+);
+CREATE TABLE IF NOT EXISTS front_lineage (
+    current_id    TEXT,
+    previous_id   TEXT,
+    snapshot_date TEXT,
+    relationship  TEXT,                              -- continuation | merge | split | new
+    overlap       REAL
+);
+
+-- Run audit log.
+CREATE TABLE IF NOT EXISTS runs (
+    id           INTEGER PRIMARY KEY AUTOINCREMENT,
+    run_date     TEXT,
+    stage        TEXT,
+    n_in         INTEGER,
+    n_out        INTEGER,
+    cost_usd     REAL,
+    notes        TEXT,
+    created_at   TEXT
+);

src/wam/webapp/__init__.py

@@ -0,0 +1 @@
+"""Streamlit RAG Q&A web app over the WAM knowledge base."""

src/wam/webapp/app.py

@@ -0,0 +1,42 @@
+"""Streamlit Q&A app — the WAM knowledge base front-end (long-context, no RAG).
+
+Run locally:   streamlit run src/wam/webapp/app.py
+On HF Spaces:  set OPENROUTER_API_KEY as a Space secret; this file is the entry point.
+
+The OpenRouter key stays server-side. Answers are grounded in a knowledge pack built from the
+committed data/wam.db (paper summaries + scores + leaderboard + authors + trends).
+"""
+
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
+
+import streamlit as st  # noqa: E402
+
+from wam.config import load_config  # noqa: E402
+from wam.webapp.qa import answer  # noqa: E402
+
+st.set_page_config(page_title="Awesome-WAM Knowledge Base", page_icon="🤖", layout="centered")
+st.title("🤖 Awesome-WAM Knowledge Base")
+st.caption("Ask about World Action Models — papers, benchmarks, authors, trends. "
+           "Answers are grounded in the tracked corpus and cite paper ids.")
+
+cfg = load_config()
+db_exists = cfg.path("db").exists()
+if not db_exists:
+    st.warning("No data found yet. Run the pipeline to populate data/wam.db.")
+
+examples = ["Which VLA models report real-time inference?",
+            "What are the rising research directions?",
+            "Who works on world models for autonomous driving?"]
+q = st.text_input("Your question", placeholder=examples[0])
+st.caption("Try: " + " · ".join(f"_{e}_" for e in examples))
+
+if q and db_exists:
+    with st.spinner("Reading the corpus and composing an answer…"):
+        res = answer(q, cfg=cfg)
+    st.markdown(res["answer"])
+    st.caption(f"Grounded in a {res['pack_chars']:,}-char knowledge pack.")

src/wam/webapp/qa.py

@@ -0,0 +1,103 @@
+"""Long-context Q&A over the WAM knowledge base (no RAG / no embeddings).
+
+At our scale (~hundreds of tracked papers) the whole corpus's *summaries* + scores +
+benchmark leaderboard + author directions + trends fit in a long-context model's window, so
+we build a compact "knowledge pack" from SQLite and let the model answer directly with
+citations — no retrieval step to miss documents, no vector index to maintain.
+"""
+
+from __future__ import annotations
+
+import json
+import sqlite3
+
+from wam.config import Config, load_config
+from wam.llm import LLMClient
+from wam.logging import get_logger
+from wam.store import Database
+
+log = get_logger("webapp.qa")
+
+SYSTEM = (
+    "You answer questions about World Action Models research using ONLY the knowledge pack "
+    "below (tracked papers with scores, a benchmark leaderboard, author directions, and "
+    "trends). Cite papers by their id like (arxiv:2606.01234). If the pack doesn't contain "
+    "the answer, say so — never invent papers, numbers, or citations. Be concise and concrete.")
+
+
+def build_pack(conn: sqlite3.Connection, max_papers: int = 400,
+               include_dropped: bool = True) -> str:
+    """Compact, citeable corpus snapshot for the long-context Q&A model."""
+    sections: list[str] = []
+
+    papers = conn.execute(
+        "SELECT id, title, track, published, scores_json, summary_json FROM papers "
+        "WHERE track IN ('core','adjacent') ORDER BY "
+        "COALESCE(json_extract(scores_json,'$.weighted_total'),0) DESC LIMIT ?", (max_papers,)
+    ).fetchall()
+    lines = ["## PAPERS"]
+    for r in papers:
+        s = json.loads(r["scores_json"]) if r["scores_json"] else {}
+        tldr = json.loads(r["summary_json"] or "{}").get("tldr", "") if r["summary_json"] else ""
+        score = f" score={s['weighted_total']}" if s.get("weighted_total") is not None else ""
+        wam = json.dumps(s.get("wam", {})) if s else ""
+        lines.append(f"- ({r['id']}) [{r['track']}{score}] {r['title']} — {tldr}"
+                     + (f" wam_scores={wam}" if wam else ""))
+    sections.append("\n".join(lines))
+
+    # Dropped papers stay in the KB (title-only) so it still covers them, per project policy.
+    if include_dropped:
+        drops = conn.execute("SELECT id, title FROM papers WHERE track='drop' "
+                             "ORDER BY first_seen DESC LIMIT 200").fetchall()
+        if drops:
+            sections.append("## OTHER (filtered-out) PAPERS — title only\n"
+                            + "\n".join(f"- ({d['id']}) {d['title']}" for d in drops))
+
+    bench = conn.execute(
+        "SELECT model_name, training_dataset, benchmark, task, metric_name, metric_value, "
+        "claimed_by_authors, source_paper_id FROM benchmarks WHERE metric_value IS NOT NULL "
+        "ORDER BY benchmark LIMIT 400").fetchall()
+    if bench:
+        bl = ["## BENCHMARK LEADERBOARD (model | training data | benchmark | task | metric=value | source)"]
+        for b in bench:
+            src = "authors" if b["claimed_by_authors"] else "3rd-party"
+            bl.append(f"- {b['model_name']} | {b['training_dataset'] or '?'} | {b['benchmark']} | "
+                      f"{b['task'] or '-'} | {b['metric_name']}={b['metric_value']} | {src} "
+                      f"({b['source_paper_id']})")
+        sections.append("\n".join(bl))
+
+    authors = conn.execute(
+        "SELECT name, affiliation, directions FROM authors ORDER BY "
+        "json_array_length(paper_ids_json) DESC LIMIT 60").fetchall()
+    if authors:
+        al = ["## INFLUENTIAL AUTHORS"]
+        for a in authors:
+            al.append(f"- {a['name']}{' ('+a['affiliation']+')' if a['affiliation'] else ''}: "
+                      f"{a['directions'] or ''}")
+        sections.append("\n".join(al))
+
+    snap = conn.execute("SELECT max(snapshot_date) FROM fronts").fetchone()[0]
+    if snap:
+        fronts = conn.execute("SELECT name, size, momentum, summary FROM fronts WHERE "
+                              "snapshot_date=? ORDER BY size DESC", (snap,)).fetchall()
+        fl = ["## TRENDS / DIRECTIONS (name | papers | momentum | summary)"]
+        for f in fronts:
+            fl.append(f"- {f['name']} | {f['size']} | {f['momentum']} | {f['summary']}")
+        sections.append("\n".join(fl))
+
+    return "\n\n".join(sections)
+
+
+def answer(question: str, cfg: Config | None = None, client: LLMClient | None = None) -> dict:
+    cfg = cfg or load_config()
+    client = client or LLMClient(cfg)
+    with Database(cfg) as db:
+        pack = build_pack(db.conn, max_papers=int(cfg.get("qa.max_papers", 400)),
+                          include_dropped=bool(cfg.get("qa.include_dropped", True)))
+    if not pack.strip():
+        return {"answer": "The knowledge base is empty — run the pipeline first.", "pack_chars": 0}
+    # Generous ceiling: the qa-tier model may be a reasoning model (burns tokens before the
+    # answer); too low a cap yields empty content.
+    ans = client.complete("qa", SYSTEM, f"KNOWLEDGE PACK:\n{pack}\n\nQUESTION: {question}",
+                          label="qa", max_tokens=6000)
+    return {"answer": ans, "pack_chars": len(pack)}