fix(discover): address PR #816 review - real budget enforcement + shallow-flag guard

P1: enrichment workers now run as daemon threads behind a semaphore instead of
a ThreadPoolExecutor. Executor threads are non-daemon and joined at interpreter
shutdown, so one hung sub-run could keep the process alive long after its topic
was downgraded to nomination-only. Daemon workers make the wall-clock budget
real; abandonment is safe because internal_subrun passes write nothing to disk
and every fetch layer carries its own timeout.

P2: --discover-shallow without --discover now errors (exit 2) instead of
silently no-opping into a full research run.

CONFIGURATION.md

@@ -49,7 +49,8 @@ The engine's `.env` reader doesn't expand `$HOME` — only the tilde, via `Path(
 - `--corpus <dir>` - add a local `.md`/`.txt` directory as a private ranked source; repeat the flag for multiple directories. PDFs are extracted only when `pdftotext` is on PATH and otherwise skip with a note. File modification time supplies recency, so the normal research window applies.
 - `--corpus-all-time` - include relevant registered files whose modification time is older than the current research window. Without this flag, a 30-day run includes only files modified in those 30 days.
 - `--register {default,exec,dev,creator,eli5}` - shape a standard single-topic Markdown or HTML research brief for its audience. `exec` is decisions-first with five core findings and numbers up top; `dev` gives GitHub, code, and technical signals more room; `creator` leads with hooks, Best Takes, community reactions, and virality metrics; `eli5` keeps the established evidence layout and asks the synthesizing agent for accessible language. Registers do not change retrieval, JSON exports, discovery, drill, library feed/search, or comparison output.
-- `--discover <domain>` - topic-less trending discovery. Sweeps rising/top-week Reddit listings (category-mapped communities, with r/all as the uncategorized fallback), Hacker News front/best stories, Digg AI 1000 clusters when `digg-pp-cli` is on PATH, and broad X activity when an X backend is authenticated, then returns 5-10 engagement-velocity-ranked topics. Run without a positional topic; it is mutually exclusive with `--drill`. `--emit=json` uses the separate versioned discovery contract documented in the [JSON export reference](docs/reference/json-export.md).
+- `--discover [domain]` - trending discovery, two-stage: a river-listing sweep NOMINATES candidate topics, then each nomination gets a full research pass (Reddit with comments, X, YouTube, Techmeme, arXiv, HN, Polymarket, web) before ranking. Bare `--discover` (no domain) is **global trending**: every feed's own hot list (r/all rising/top-week, Hacker News front/best, Digg clusters when `digg-pp-cli` is on PATH) with no keyword gate; with a domain, the sweep is category-scoped and keyword-gated, and broad X activity joins when an X backend is authenticated. Every topic must clear a confidence floor (cross-source confirmation or a genuinely strong single-source spike); when nothing clears it the run reports "Nothing solid this window" instead of ranked noise. Run without a positional topic; it is mutually exclusive with `--drill`. `--emit=json` uses the separate versioned discovery contract (now with `outcome`, `weak_signal`, per-topic `top_comment` and `corroboration_count`) documented in the [JSON export reference](docs/reference/json-export.md).
+- `--discover-shallow` - skip discovery's per-topic research passes and rank on listing evidence only. Faster and thinner; the confidence floor still applies. An explicit `--search` source list bounds both the sweep and the research passes.
 - `--drill <target>` - deep follow-up over the fresh `~/.config/last30days/last-report.json` cache. Accepts a 1-based index (`--drill "cluster 3"` or `--drill "3"`) or a fuzzy cluster title/entity description. It re-fetches only sources that contributed to the matched cluster, enables their deep comment/transcript enrichment paths, merges/dedupes the evidence, and replaces the cache so drills can chain. Run it without a positional topic; if the cache is absent or expired, run a normal research pass first.
 - `--verify-freshness` - opt into an act-time verification pass for conservatively extracted, source-grounded claims (Polymarket odds/end dates, GitHub stars, StockTwits sentiment ratios, and explicit status assertions). With a topic, verification runs after research; without a topic, it re-verifies the fresh `last-report.json` cache without repeating research. Verdicts are `current`, `stale`, `contradicted`, or `unsupported` and include evidence timestamps. Set `LAST30DAYS_VERIFY_FRESHNESS=on` in `.env` to make the pass default for normal research runs.
 - `--save-suffix <name>` - distinguish runs of the same topic (e.g. per client: `--save-suffix=acme`).

docs/reference/json-export.md

@@ -33,7 +33,7 @@ Discovery mode has a separate versioned contract so its topic results do not cha
 python3 skills/last30days/scripts/last30days.py --discover "AI agents" --emit=json
 ```
 
-Its top level contains `schema_version` (`1.0`), `kind` (`"discovery"`), `domain`, `generated_at`, `window_days`, `source_status`, `feeds`, `results`, and `warnings`. Each ranked result contains `rank`, `topic`, `why_spiking`, `momentum` (`new-this-week` or `building`), `velocity_score`, `sources`, per-source native `engagement`, a ready-to-run `command`, and `evidence_urls`. The discovery contract follows the same versioning policy below but evolves independently of the normal agent export. `--json-profile=raw` returns the unversioned internal `DiscoveryReport` dataclass instead.
+Its top level contains `schema_version` (`1.0`), `kind` (`"discovery"`), `domain` (`""` for a global no-domain trending run), `generated_at`, `window_days`, `source_status`, `feeds`, `results`, `warnings`, `outcome` (`"ok"`, or `"nothing-solid"` when no topic cleared the confidence floor), and `weak_signal` (the closest sub-floor topic name on a nothing-solid run, else `null`). Each ranked result contains `rank`, `topic`, `why_spiking`, `momentum` (`new-this-week` or `building`), `velocity_score`, `sources`, per-source native `engagement`, a ready-to-run `command`, `evidence_urls`, `top_comment` (the strongest verbatim community comment from the topic's research pass, with attribution; `null` on shallow runs), and `corroboration_count` (distinct confirming sources). The discovery contract follows the same versioning policy below but evolves independently of the normal agent export. `--json-profile=raw` returns the unversioned internal `DiscoveryReport` dataclass instead.
 
 When `LAST30DAYS_API_KEY` and `LAST30DAYS_API_BASE` route a run through a configured remote API, the server does not return the local `Report` needed to build this profile. In that mode, `--json-profile=agent` exits with status 2 instead of emitting a misleading shape; use `--json-profile=raw` to retain the remote backend's existing server-response JSON contract.
 

skills/last30days/SKILL.md

@@ -127,7 +127,7 @@ Replace `{VERSION}` with the installed plugin version (`jq -r '.version' "$SKILL
 **Placement by query type:**
 - GENERAL / NEWS / PROMPTING / RECOMMENDATIONS: badge on line 1, blank line 2, `What I learned:` on line 3, then bold-lead-in paragraphs
 - COMPARISON: badge on line 1, blank line 2, `# {TOPIC_A} vs {TOPIC_B} [vs {TOPIC_C}]: What the Community Says (/Last30Days)` on line 3, then Quick Verdict section
-- DISCOVERY: pass through the engine's topic-per-section discovery brief verbatim. Its ranked headings, momentum labels, evidence counters, and `/last30days "<topic>"` handoffs are engine-owned and are an explicit exception to the GENERAL synthesis template.
+- DISCOVERY: pass through the engine's topic-per-section discovery brief verbatim. Its ranked headings, momentum labels, community-voice quotes, evidence counters, `/last30days "<topic>"` handoffs, and the "Nothing solid this window" empty state are engine-owned and are an explicit exception to the GENERAL synthesis template. A nothing-solid result is a valid final answer — relay it, never retry or fabricate topics around it.
 
 ---
 
@@ -289,7 +289,11 @@ The single most common failure mode of this skill is the model reading this file
 
 Branching rule:
 
-- **If the user asks what is trending, exploding, or worth covering in a domain** (for example, `/last30days what's exploding in AI agents?`): set `DISCOVERY_DOMAIN` to the domain phrase, complete the first-run wizard if needed, **and after the wizard finishes return to THIS branch with the saved `DISCOVERY_DOMAIN` (do NOT fall through to Parse User Intent / Step 0.45 / normal topic research - onboarding must not downgrade a discovery request into a topic run)**, then run `"${LAST30DAYS_PYTHON}" "${SKILL_DIR}/scripts/last30days.py" --discover "${DISCOVERY_DOMAIN}" --emit=compact --save-dir="${LAST30DAYS_MEMORY_DIR}"`. Do not run Step 0.5, Step 0.55, Step 0.75, WebSearch supplements, or the normal synthesis pass; the listing sweep and topic-per-section brief are the complete discovery flow. Relay stdout verbatim. If no domain was supplied, ask one short question for the domain and wait.
+- **If the user asks what is trending — globally or in a domain** (for example, `/last30days trending`, `/last30days what's hot right now?`, `/last30days what's exploding in AI agents?`): this is DISCOVERY. Complete the first-run wizard if needed, **and after the wizard finishes return to THIS branch (do NOT fall through to Parse User Intent / Step 0.45 / normal topic research - onboarding must not downgrade a discovery request into a topic run)**. Two variants:
+  - **Global trending** (no domain named — "trending", "what's hot", "what's happening"): run `"${LAST30DAYS_PYTHON}" "${SKILL_DIR}/scripts/last30days.py" --discover --emit=compact --save-dir="${LAST30DAYS_MEMORY_DIR}"` (bare `--discover`, NO domain argument, NOT a request to ask the user for a domain). It sweeps every river feed's own hot list (r/all, HN front page, Digg) with no keyword gate.
+  - **Domain trending** (a domain phrase is named): set `DISCOVERY_DOMAIN` to the domain phrase and run `"${LAST30DAYS_PYTHON}" "${SKILL_DIR}/scripts/last30days.py" --discover "${DISCOVERY_DOMAIN}" --emit=compact --save-dir="${LAST30DAYS_MEMORY_DIR}"`.
+
+  Discovery is two-stage: a listing sweep NOMINATES candidate topics, then each nomination gets a full research pass (Reddit with comments, X, YouTube, Techmeme, arXiv, HN, Polymarket, web) before ranking — expect an enriched discovery run to take a few minutes; use a Bash timeout of 600000 (10 minutes). If the user asks for a fast/rough sweep, add `--discover-shallow` (listing evidence only; thinner cards, still quality-floored). Do not run Step 0.5, Step 0.55, Step 0.75, WebSearch supplements, or the normal synthesis pass; the nominate-enrich sweep and topic-per-section brief are the complete discovery flow. Relay stdout verbatim — including a **"Nothing solid this window"** result, which is a valid, honest outcome (the confidence floor found no topic with enough cross-source confirmation or engagement; do NOT retry, work around it, or fabricate topics — relay it and suggest a narrower domain or a direct topic run).
 - **If the user provided a topic** (e.g. `/last30days Kanye West`, `/last30days nvidia earnings`): confirm the first-run gate above passed (output `1`), then proceed to `## Step 0: First-Run Setup Wizard` (or skip it if already confirmed complete), then continue to Step 0.45 / Step 0.5 / Step 0.55 / Step 0.75 / Research Execution below. Do not skip straight to WebSearch. WebSearch is a **supplement after** the Python engine runs (see Step 2). It is **not a substitute**.
 - **If the user provided no topic**: ask the user for a topic with a single short question. Do not run research. Do not run WebSearch. Wait.
 

skills/last30days/scripts/last30days.py

@@ -507,7 +507,22 @@ def build_parser() -> argparse.ArgumentParser:
     parser.add_argument(
         "--discover",
         metavar="DOMAIN",
-        help="Sweep category listings and rank 5-10 topics accelerating in a domain",
+        nargs="?",
+        const="",
+        default=None,
+        help=(
+            "Sweep river listings and rank the topics accelerating in a domain; "
+            "each survivor gets a full research pass. Bare --discover (no domain) "
+            "runs global trending across every feed's hot list"
+        ),
+    )
+    parser.add_argument(
+        "--discover-shallow",
+        action="store_true",
+        help=(
+            "Skip the per-topic research pass during --discover: rank on listing "
+            "evidence only (faster, thinner; the confidence floor still applies)"
+        ),
     )
     parser.add_argument("--debug", action="store_true", help="Enable HTTP debug logging")
     parser.add_argument("--mock", action="store_true", help="Use mock retrieval fixtures")
@@ -1263,9 +1278,8 @@ def _save_discovery_output(
 
 def _run_discover(args: argparse.Namespace, config: dict[str, object]) -> int:
     domain = " ".join(str(args.discover or "").split())
-    if not domain:
-        sys.stderr.write("[last30days] --discover requires a non-empty domain.\n")
-        return 2
+    # Empty domain = global trending: sweep every river feed's hot list with no
+    # keyword gate. The confidence floor is what keeps junk out, not a keyword.
     if args.as_of_date:
         sys.stderr.write(
             "[last30days] --as-of cannot be used with --discover because discovery "
@@ -1281,6 +1295,10 @@ def _run_discover(args: argparse.Namespace, config: dict[str, object]) -> int:
         sys.stderr.write("[last30days] Warning: --synthesis-file is not used by discovery mode.\n")
 
     requested_sources = resolve_requested_sources(args.search, config)
+    # The user's original source boundary, honored by the per-topic research
+    # passes (which reach beyond the discovery-capable listing feeds - e.g.
+    # Techmeme, arXiv, YouTube, Polymarket). None = every available source.
+    enrich_requested_sources = list(requested_sources) if requested_sources else None
     if requested_sources:
         discovery_sources = [
             source for source in requested_sources
@@ -1315,6 +1333,8 @@ def _run_discover(args: argparse.Namespace, config: dict[str, object]) -> int:
             subreddits=subreddits,
             lookback_days=args.lookback_days or 30,
             as_of_date=args.as_of_date,
+            enrich=not args.discover_shallow,
+            enrich_requested_sources=enrich_requested_sources,
         )
     except ValueError as exc:
         sys.stderr.write(f"[last30days] {exc}\n")
@@ -1332,7 +1352,7 @@ def _run_discover(args: argparse.Namespace, config: dict[str, object]) -> int:
     if args.save_dir:
         save_path = _save_discovery_output(
             rendered,
-            domain=domain,
+            domain=domain or "trending",
             emit=args.emit,
             save_dir=args.save_dir,
             suffix=args.save_suffix or "",
@@ -2071,7 +2091,9 @@ def _main(
         sys.stderr.write(setup_wizard.get_setup_status_text(results) + "\n")
         return 0
 
-    if args.discover:
+    # Bare --discover (no domain) is global trending, so the dispatch keys on
+    # "flag present" (is not None), never on the domain string's truthiness.
+    if args.discover is not None:
         if topic:
             sys.stderr.write(
                 "[last30days] --discover supplies the domain and cannot be combined "
@@ -2083,6 +2105,15 @@ def _main(
             return 2
         return _run_discover(args, config)
 
+    if args.discover_shallow:
+        # Without --discover this flag would silently no-op into a full
+        # research run - reject it instead of ignoring the requested mode.
+        sys.stderr.write(
+            "[last30days] --discover-shallow only applies to --discover runs; "
+            "add --discover [domain] or drop the flag.\n"
+        )
+        return 2
+
     if args.drill:
         if topic:
             sys.stderr.write(

skills/last30days/scripts/lib/pipeline.py

@@ -4,12 +4,14 @@ from __future__ import annotations
 
 import copy
 import math
+import queue
 import re
 import sqlite3
 import sys
 import threading
 import time
 from concurrent.futures import ThreadPoolExecutor, as_completed
+from dataclasses import dataclass, field
 from datetime import date, datetime, timedelta, timezone
 from pathlib import Path
 from shutil import which
@@ -371,7 +373,17 @@ def _fetch_discovery_source(
     depth: str,
     mock: bool,
     config: dict[str, Any],
+    keyword_gate: bool = True,
 ) -> tuple[list[dict[str, Any]], str | None]:
+    """Fetch one listing/river source for the nominate stage.
+
+    ``keyword_gate`` controls whether items are filtered to the domain by
+    ``_matches_discovery_domain``. Domain-scoped discovery (``--discover X``)
+    keeps the gate on; global trending (``--discover`` with no domain) turns it
+    off, because there is no keyword to gate against - the river feeds ARE the
+    "what is hot right now" signal, and the confidence floor downstream is what
+    keeps junk out, not a keyword match.
+    """
     if mock:
         return _mock_discovery_items(source, plan.domain, to_date), None
     if source == "reddit":
@@ -379,13 +391,14 @@ def _fetch_discovery_source(
             plan.subreddits, depth=depth, query=plan.domain,
         )
         items = result.get("items") or []
-        items = [
-            item for item in items
-            if _matches_discovery_domain(
-                plan.domain,
-                f"{item.get('title') or ''} {item.get('selftext') or ''}",
-            )
-        ]
+        if keyword_gate:
+            items = [
+                item for item in items
+                if _matches_discovery_domain(
+                    plan.domain,
+                    f"{item.get('title') or ''} {item.get('selftext') or ''}",
+                )
+            ]
         return items, "; ".join(result.get("errors") or []) or None
     if source == "hackernews":
         result = hackernews.fetch_discovery_listings(from_date, to_date, depth=depth)
@@ -395,21 +408,25 @@ def _fetch_discovery_source(
                 plan.domain,
                 str(item.get("title") or ""),
             )
-        # HN is a broad technology listing, so keep only domain-bearing stories.
-        items = [
-            item for item in items
-            if _matches_discovery_domain(plan.domain, str(item.get("title") or ""))
-        ]
+        # HN is a broad technology listing, so keep only domain-bearing stories
+        # when a domain is in play; global trending keeps the whole front page.
+        if keyword_gate:
+            items = [
+                item for item in items
+                if _matches_discovery_domain(plan.domain, str(item.get("title") or ""))
+            ]
         errors = result.get("errors") or []
         return items, "; ".join(errors) or None
     if source == "digg":
         result = digg.search_digg(plan.domain, from_date, to_date, depth=depth)
         items = digg.parse_digg_response(result, query=plan.domain)
-        # Digg is an AI-focused broad listing, so keep only domain-bearing clusters.
-        items = [
-            item for item in items
-            if _matches_discovery_domain(plan.domain, str(item.get("title") or ""))
-        ]
+        # Digg is an AI-focused broad listing, so keep only domain-bearing
+        # clusters when scoped; global trending keeps the whole feed.
+        if keyword_gate:
+            items = [
+                item for item in items
+                if _matches_discovery_domain(plan.domain, str(item.get("title") or ""))
+            ]
         return items, result.get("error")
     if source == "x":
         subquery = schema.SubQuery(
@@ -504,57 +521,29 @@ def _discovery_momentum(items: list[schema.SourceItem], to_date: str) -> str:
     return "new-this-week" if ages and max(ages) < 7 else "building"
 
 
-def run_discover(
+def nominate_candidates(
+    plan: schema.DiscoveryPlan,
     *,
-    domain: str,
+    from_date: str,
+    to_date: str,
+    depth: str,
+    mock: bool,
     config: dict[str, Any],
-    depth: str = "default",
-    requested_sources: list[str] | None = None,
-    mock: bool = False,
-    subreddits: list[str] | None = None,
-    lookback_days: int = 30,
-    as_of_date: str | None = None,
-    limit: int = 10,
-) -> schema.DiscoveryReport:
-    """Sweep category listings and rank the topics gaining velocity."""
-    from_date, to_date = dates.get_date_range(lookback_days, as_of_date=as_of_date)
-    requested = normalize_requested_sources(requested_sources)
-    unsupported = sorted(set(requested or []) - set(DISCOVERY_SOURCES))
-    if unsupported:
-        raise ValueError(
-            "Discovery supports listing sources only: reddit, hackernews, digg "
-            f"(unsupported: {', '.join(unsupported)})"
-        )
-    available = list(DISCOVERY_SOURCES) if mock else [
-        source for source in available_sources(config, requested, x_pending=False)
-        if source in DISCOVERY_SOURCES
-    ]
-    if requested:
-        available = [source for source in available if source in requested]
-    plan = planner.build_discovery_plan(
-        domain,
-        available_sources=available,
-        subreddits=subreddits,
-    )
+    lookback_days: int,
+    keyword_gate: bool = True,
+) -> schema.RetrievalBundle:
+    """Stage 1 of discovery: fetch, normalize, and bundle candidate hot items
+    from the river/listing feeds.
 
-    source_status: dict[str, schema.SourceOutcome] = {}
+    This is the topic-nomination pass. For domain discovery ``keyword_gate`` is
+    on and the feeds are filtered to the domain; for global trending it is off
+    and the feeds' own hot ranking IS the signal. The returned bundle feeds the
+    clustering + enrichment stages downstream. Every source's failure is
+    recorded on the bundle (never raised) so a single dead feed cannot sink the
+    run - the confidence floor decides whether the surviving evidence is enough.
+    """
     bundle = schema.RetrievalBundle()
-    query_plan = schema.QueryPlan(
-        intent="breaking_news",
-        freshness_mode="breaking",
-        cluster_mode="story",
-        raw_topic=plan.domain,
-        subqueries=[schema.SubQuery(
-            label="discovery-listings",
-            search_query=plan.domain,
-            ranking_query=f"What is accelerating in {plan.domain}?",
-            sources=list(plan.sources),
-        )],
-        source_weights={source: 1.0 for source in plan.sources},
-        notes=["discover-mode", "listing-sweep"],
-    )
-
-    with ThreadPoolExecutor(max_workers=len(plan.sources)) as executor:
+    with ThreadPoolExecutor(max_workers=max(1, len(plan.sources))) as executor:
         futures = {
             executor.submit(
                 _fetch_discovery_source,
@@ -565,6 +554,7 @@ def run_discover(
                 depth=depth,
                 mock=mock,
                 config=config,
+                keyword_gate=keyword_gate,
             ): source
             for source in plan.sources
         }
@@ -580,7 +570,10 @@ def run_discover(
                     to_date,
                     freshness_mode="breaking",
                 )
-                prepared = relevance.PreparedQuery(plan.domain)
+                # Global trending has no domain; annotate against a neutral
+                # phrase so snippet extraction still works without biasing
+                # relevance toward any keyword.
+                prepared = relevance.PreparedQuery(plan.domain or "trending now")
                 normalized = signals.annotate_stream(
                     normalized,
                     prepared,
@@ -606,22 +599,40 @@ def run_discover(
             except Exception as exc:
                 state, attempted = _classify_source_failure(exc)
                 bundle.record_failure(source, state, str(exc), attempted=attempted)
+    return bundle
 
-    for source in DISCOVERY_SOURCES:
-        if source in bundle.source_status:
-            continue
-        detail = (
-            "Source is not configured for discovery."
-        )
-        source_status[source] = schema.SourceOutcome(
-            source=source,
-            state=schema.SKIPPED_UNCONFIGURED,
-            attempted=False,
-            detail=detail,
-            fix_hint="doctor",
-        )
-    source_status.update(_finalize_source_status(bundle.source_status, bundle.items_by_source))
 
+@dataclass(frozen=True)
+class Nomination:
+    """A named candidate topic produced by the nominate stage.
+
+    ``seed_score`` is the cheap pre-enrichment velocity rank - enough to decide
+    WHICH candidates deserve a full pipeline pass, but not the final ranking
+    signal (that comes from enriched evidence downstream).
+    """
+
+    name: str
+    seed_score: float
+    items: list[schema.SourceItem] = field(default_factory=list)
+    summary: str = ""
+
+
+def nominate_topics(
+    bundle: schema.RetrievalBundle,
+    query_plan: schema.QueryPlan,
+    plan: schema.DiscoveryPlan,
+    *,
+    to_date: str,
+    limit: int,
+) -> list[Nomination]:
+    """Stage 1b of discovery: cluster nominated items into named candidate
+    topics and rank them by seed velocity.
+
+    Names are deduped casefold so the same story surfacing under two clusters
+    yields one nomination. Returns at most ``limit`` nominations, never padded -
+    fewer clusters than ``limit`` means a shorter list, and the confidence
+    floor downstream decides whether what survived is worth showing.
+    """
     candidates = weighted_rrf(bundle.items_by_source_and_query, query_plan, pool_limit=80)
     for candidate in candidates:
         velocity = rerank.discovery_velocity_score(candidate.source_items, as_of_date=to_date)
@@ -643,8 +654,7 @@ def run_discover(
         ranked_clusters.append((score, cluster, cluster_items))
     ranked_clusters.sort(key=lambda entry: (-entry[0], entry[1].title.lower()))
 
-    topic_limit = max(5, min(10, limit))
-    topics: list[schema.DiscoveryTopic] = []
+    nominations: list[Nomination] = []
     seen_topic_names: set[str] = set()
     for score, cluster, cluster_items in ranked_clusters:
         name = discovery_topic_name(cluster, candidate_map, plan.domain)
@@ -652,41 +662,367 @@ def run_discover(
         if name_key in seen_topic_names:
             continue
         seen_topic_names.add(name_key)
-        rank = len(topics) + 1
-        sources = sorted({item.source for item in cluster_items})
-        native_total = sum(rerank.discovery_engagement_total(item) for item in cluster_items)
+        leader = candidate_map.get(cluster.representative_ids[0]) if cluster.representative_ids else None
+        summary = (leader.snippet if leader else "") or (leader.title if leader else name)
+        nominations.append(Nomination(
+            name=name,
+            seed_score=score,
+            items=cluster_items,
+            summary=summary,
+        ))
+        if len(nominations) >= limit:
+            break
+    return nominations
+
+
+# Enrichment fan-out bounds. Sub-runs hit the same upstream APIs as a normal
+# research pass, so parallelism stays low and the whole batch runs against a
+# wall-clock budget - a slow topic is dropped, never fatal.
+ENRICH_LIMIT = 6
+ENRICH_DEPTH = "quick"
+ENRICH_MAX_WORKERS = 3
+ENRICH_BUDGET_SECONDS = 240.0
+
+
+@dataclass
+class EnrichedTopic:
+    """A nomination plus the full-pipeline evidence gathered for it.
+
+    ``report`` is None when enrichment for this topic failed or ran past the
+    batch budget - the topic survives as nomination-only and the confidence
+    floor downstream decides whether its seed evidence is enough to show.
+    """
+
+    nomination: Nomination
+    report: schema.Report | None = None
+    error: str | None = None
+
+
+def enrich_nominations(
+    nominations: list[Nomination],
+    *,
+    config: dict[str, Any],
+    requested_sources: list[str] | None = None,
+    mock: bool = False,
+    depth: str = ENRICH_DEPTH,
+    lookback_days: int = 30,
+    as_of_date: str | None = None,
+    max_workers: int = ENRICH_MAX_WORKERS,
+    budget_seconds: float = ENRICH_BUDGET_SECONDS,
+) -> list[EnrichedTopic]:
+    """Stage 2 of discovery: run the real research pipeline on each nomination.
+
+    Each nominated topic gets a full ``run()`` pass (``internal_subrun=True``,
+    same lane as comparison-mode sub-runs), which buys the whole multi-source
+    corpus - Reddit with comments, X, YouTube, Techmeme, arXiv, HN, Polymarket,
+    web - plus clustering and ranking, with zero bespoke fetch code.
+
+    Failure containment: a topic whose sub-run raises is returned with
+    ``report=None`` and the error recorded; topics still unfinished when the
+    batch budget expires are likewise dropped to nomination-only. The batch
+    never raises and preserves nomination order.
+    """
+    if not nominations:
+        return []
+
+    def _run_one(nomination: Nomination) -> schema.Report:
+        return run(
+            topic=nomination.name,
+            config=config,
+            depth=depth,
+            requested_sources=requested_sources,
+            mock=mock,
+            lookback_days=lookback_days,
+            as_of_date=as_of_date,
+            internal_subrun=True,
+        )
+
+    # Daemon threads + a semaphore instead of ThreadPoolExecutor: executor
+    # threads are non-daemon and joined at interpreter shutdown, so one hung
+    # sub-run could keep the whole process alive long after its topic was
+    # downgraded to nomination-only. Daemon workers make the wall-clock budget
+    # real - stragglers cannot delay process exit. Abandonment is safe because
+    # internal_subrun passes write nothing to disk (no save, no library sync,
+    # no store), and every fetch layer inside run() carries its own timeout.
+    enriched: dict[str, EnrichedTopic] = {}
+    results_queue: queue.Queue[tuple[Nomination, schema.Report | None, Exception | None]] = queue.Queue()
+    slots = threading.Semaphore(max(1, max_workers))
+
+    def _worker(nomination: Nomination) -> None:
+        with slots:
+            try:
+                results_queue.put((nomination, _run_one(nomination), None))
+            except Exception as exc:  # noqa: BLE001 - containment is the contract
+                results_queue.put((nomination, None, exc))
+
+    for nomination in nominations:
+        threading.Thread(
+            target=_worker,
+            args=(nomination,),
+            name=f"discover-enrich-{nomination.name[:32]}",
+            daemon=True,
+        ).start()
+
+    deadline = time.monotonic() + max(1.0, budget_seconds)
+    pending = len(nominations)
+    while pending and (remaining := deadline - time.monotonic()) > 0:
+        try:
+            nomination, report, exc = results_queue.get(timeout=min(remaining, 0.5))
+        except queue.Empty:
+            continue
+        pending -= 1
+        if exc is None:
+            enriched[nomination.name] = EnrichedTopic(
+                nomination=nomination, report=report,
+            )
+        else:
+            enriched[nomination.name] = EnrichedTopic(
+                nomination=nomination,
+                error=f"{type(exc).__name__}: {exc}",
+            )
+            print(
+                f"[Discover] enrichment failed for {nomination.name!r}: "
+                f"{type(exc).__name__}: {exc}",
+                file=sys.stderr,
+            )
+    # Budget expired (or all done): unfinished topics fall through below as
+    # nomination-only; their daemon workers are abandoned and cannot block exit.
+
+    results: list[EnrichedTopic] = []
+    for nomination in nominations:
+        entry = enriched.get(nomination.name)
+        if entry is None:
+            entry = EnrichedTopic(
+                nomination=nomination,
+                error="enrichment budget exhausted",
+            )
+            print(
+                f"[Discover] enrichment budget exhausted before {nomination.name!r} "
+                "finished; keeping nomination-only evidence",
+                file=sys.stderr,
+            )
+        results.append(entry)
+    return results
+
+
+def _enriched_evidence_items(entry: EnrichedTopic) -> list[schema.SourceItem]:
+    """The items a topic is judged on: the enriched corpus when the pipeline
+    pass succeeded, the nomination's seed items otherwise."""
+    if entry.report is not None:
+        flattened: list[schema.SourceItem] = []
+        for source_items in entry.report.items_by_source.values():
+            flattened.extend(source_items)
+        if flattened:
+            return flattened
+    return entry.nomination.items
+
+
+def _best_community_comment(items: list[schema.SourceItem]) -> str | None:
+    """The strongest verbatim community comment across a topic's evidence,
+    formatted with attribution - the voice-of-the-people line on a trend card.
+
+    Vote strength is per-platform-normalized (signals.normalized_comment_vote)
+    so one viral platform's counts don't drown out the rest.
+    """
+    best: tuple[float, str, str | None, float | int | None] | None = None
+    for item in items:
+        comments = item.metadata.get("top_comments") or []
+        for comment in comments:
+            if not isinstance(comment, dict):
+                continue
+            body = (comment.get("excerpt") or comment.get("text") or comment.get("body") or "").strip()
+            if len(body) < 12:
+                continue
+            strength = signals.normalized_comment_vote(item.source, comment.get("score"))
+            if best is None or strength > best[0]:
+                best = (strength, body, comment.get("author"), comment.get("score"))
+    if best is None:
+        return None
+    _, body, author, score = best
+    # Comment bodies that themselves start/end with quote characters would
+    # render as doubled quotes inside our wrapping quotes.
+    body = body.strip('"“”‘’\'').strip()
+    if len(body) > 200:
+        body = body[:197].rsplit(" ", 1)[0] + "..."
+    attribution = f" - {author}" if author else ""
+    votes = (
+        f" ({int(score):,} votes)"
+        if isinstance(score, (int, float)) and not isinstance(score, bool) and score > 0
+        else ""
+    )
+    return f'"{body}"{attribution}{votes}'
+
+
+def run_discover(
+    *,
+    domain: str,
+    config: dict[str, Any],
+    depth: str = "default",
+    requested_sources: list[str] | None = None,
+    mock: bool = False,
+    subreddits: list[str] | None = None,
+    lookback_days: int = 30,
+    as_of_date: str | None = None,
+    limit: int = 10,
+    enrich: bool = False,
+    enrich_requested_sources: list[str] | None = None,
+) -> schema.DiscoveryReport:
+    """Sweep category listings and rank the topics gaining velocity.
+
+    ``requested_sources`` bounds the listing sweep (discovery-capable feeds
+    only). ``enrich_requested_sources`` bounds the per-topic research passes:
+    None means every available source - which is what lets Techmeme, arXiv,
+    YouTube, Polymarket, and community comments reach discovery despite having
+    no river feed of their own. Pass the user's original --search list here so
+    an explicit source boundary holds through enrichment too.
+    """
+    from_date, to_date = dates.get_date_range(lookback_days, as_of_date=as_of_date)
+    requested = normalize_requested_sources(requested_sources)
+    unsupported = sorted(set(requested or []) - set(DISCOVERY_SOURCES))
+    if unsupported:
+        raise ValueError(
+            "Discovery supports listing sources only: reddit, hackernews, digg "
+            f"(unsupported: {', '.join(unsupported)})"
+        )
+    available = list(DISCOVERY_SOURCES) if mock else [
+        source for source in available_sources(config, requested, x_pending=False)
+        if source in DISCOVERY_SOURCES
+    ]
+    if requested:
+        available = [source for source in available if source in requested]
+    plan = planner.build_discovery_plan(
+        domain,
+        available_sources=available,
+        subreddits=subreddits,
+    )
+
+    global_mode = not plan.domain
+    domain_label = plan.domain or "everything"
+    source_status: dict[str, schema.SourceOutcome] = {}
+    query_plan = schema.QueryPlan(
+        intent="breaking_news",
+        freshness_mode="breaking",
+        cluster_mode="story",
+        raw_topic=plan.domain,
+        subqueries=[schema.SubQuery(
+            label="discovery-listings",
+            search_query=plan.domain,
+            ranking_query=f"What is accelerating in {domain_label}?",
+            sources=list(plan.sources),
+        )],
+        source_weights={source: 1.0 for source in plan.sources},
+        notes=["discover-mode", "listing-sweep"],
+    )
+
+    bundle = nominate_candidates(
+        plan,
+        from_date=from_date,
+        to_date=to_date,
+        depth=depth,
+        mock=mock,
+        config=config,
+        lookback_days=lookback_days,
+        # Global trending has no keyword to gate against - the river feeds' own
+        # hot ranking is the signal and the confidence floor culls the junk.
+        keyword_gate=not global_mode,
+    )
+
+    for source in DISCOVERY_SOURCES:
+        if source in bundle.source_status:
+            continue
+        detail = (
+            "Source is not configured for discovery."
+        )
+        source_status[source] = schema.SourceOutcome(
+            source=source,
+            state=schema.SKIPPED_UNCONFIGURED,
+            attempted=False,
+            detail=detail,
+            fix_hint="doctor",
+        )
+    source_status.update(_finalize_source_status(bundle.source_status, bundle.items_by_source))
+
+    topic_limit = max(5, min(10, limit))
+    nominations = nominate_topics(
+        bundle, query_plan, plan,
+        to_date=to_date,
+        limit=ENRICH_LIMIT if enrich else topic_limit,
+    )
+
+    if enrich and nominations:
+        enriched_entries = enrich_nominations(
+            nominations,
+            config=config,
+            requested_sources=enrich_requested_sources,
+            mock=mock,
+            lookback_days=lookback_days,
+            as_of_date=as_of_date,
+        )
+    else:
+        enriched_entries = [
+            EnrichedTopic(nomination=nomination) for nomination in nominations
+        ]
+
+    topics: list[schema.DiscoveryTopic] = []
+    weak_signal: tuple[float, str] | None = None
+    for entry in enriched_entries:
+        evidence_items = _enriched_evidence_items(entry)
+        sources = sorted({item.source for item in evidence_items})
+        native_total = sum(
+            rerank.discovery_engagement_total(item) for item in evidence_items
+        )
+        score = rerank.discovery_velocity_score(evidence_items, as_of_date=to_date)
+        if not rerank.passes_discovery_floor(
+            source_count=len(sources),
+            engagement_total=native_total,
+            item_count=len(evidence_items),
+        ):
+            # Sub-floor evidence never ranks; remember what came closest so a
+            # nothing-solid brief can still name the strongest weak signal.
+            if weak_signal is None or score > weak_signal[0]:
+                weak_signal = (score, entry.nomination.name)
+            continue
+        if len(topics) >= topic_limit:
+            break
+        nomination = entry.nomination
         source_phrase = ", ".join(sources[:-1]) + (
             f" and {sources[-1]}" if len(sources) > 1 else (sources[0] if sources else "the listings")
         )
-        leader = candidate_map.get(cluster.representative_ids[0]) if cluster.representative_ids else None
-        summary = (leader.snippet if leader else "") or (leader.title if leader else name)
+        noun = "evidence item" if entry.report is not None else "listing item"
         why = (
-            f"{len(cluster_items)} listing item{'s' if len(cluster_items) != 1 else ''} on "
+            f"{len(evidence_items)} {noun}{'s' if len(evidence_items) != 1 else ''} on "
             f"{source_phrase} generated {native_total:,.0f} native interactions. "
-            f"{summary[:220]}"
+            f"{nomination.summary[:220]}"
         )
         topics.append(schema.DiscoveryTopic(
-            rank=rank,
-            name=name,
+            rank=len(topics) + 1,
+            name=nomination.name,
             why_spiking=why,
-            momentum=_discovery_momentum(cluster_items, to_date),
+            momentum=_discovery_momentum(evidence_items, to_date),
             velocity_score=round(score, 2),
             sources=sources,
-            engagement_by_source=_discovery_engagement(cluster_items),
-            command=f'/last30days "{name.replace(chr(34), chr(39))}"',
-            evidence_urls=list(dict.fromkeys(item.url for item in cluster_items if item.url))[:5],
+            engagement_by_source=_discovery_engagement(evidence_items),
+            command=f'/last30days "{nomination.name.replace(chr(34), chr(39))}"',
+            evidence_urls=list(dict.fromkeys(item.url for item in evidence_items if item.url))[:5],
+            top_comment=_best_community_comment(evidence_items) if entry.report is not None else None,
+            corroboration_count=len(sources),
         ))
-        if len(topics) >= topic_limit:
-            break
+
+    outcome = "ok" if topics else "nothing-solid"
 
     warnings: list[str] = []
-    if len(topics) < 5:
-        warnings.append("Fewer than five topic clusters survived this domain sweep.")
+    if outcome == "nothing-solid":
+        warnings.append(
+            "No topic cleared the discovery confidence floor this window; "
+            "reporting nothing solid instead of ranked noise."
+        )
+    elif len(topics) < 5:
+        warnings.append("Fewer than five topic clusters cleared the confidence floor this window.")
     if topics and all(len(topic.sources) == 1 for topic in topics):
         warnings.append("Discovery evidence is single-source; configure Digg for broader confirmation.")
     failed = [
-        source for source, outcome in source_status.items()
-        if outcome.state not in {health.OK, schema.NO_RESULTS, schema.SKIPPED_UNCONFIGURED}
+        source for source, outcome_state in source_status.items()
+        if outcome_state.state not in {health.OK, schema.NO_RESULTS, schema.SKIPPED_UNCONFIGURED}
     ]
     if failed:
         warnings.append(f"Some discovery sources degraded: {', '.join(sorted(failed))}.")
@@ -700,6 +1036,8 @@ def run_discover(
         topics=topics,
         source_status=source_status,
         warnings=warnings,
+        outcome=outcome,
+        weak_signal=weak_signal[1] if weak_signal and not topics else None,
     )
 
 

skills/last30days/scripts/lib/planner.py

@@ -26,10 +26,32 @@ def build_discovery_plan(
     available_sources: list[str] | None = None,
     subreddits: list[str] | None = None,
 ) -> schema.DiscoveryPlan:
-    """Resolve a domain to the existing category-peer community feeds."""
+    """Resolve a domain to the existing category-peer community feeds.
+
+    An empty domain is global trending: sweep every river feed's own hot list
+    (r/all, HN front page, Digg) with no category scoping. Keyword-driven
+    sources (X, Techmeme, arXiv - none of which expose a river/front-page
+    lane) sit out of the global nominate stage and join per-topic at the
+    enrichment pass, where every nomination gets a full research run.
+    """
     normalized_domain = " ".join(domain.split())
     if not normalized_domain:
-        raise ValueError("Discovery domain cannot be empty")
+        resolved = [
+            subreddit.removeprefix("r/").strip()
+            for subreddit in (subreddits or ["all"])
+            if subreddit.strip()
+        ]
+        allowed = set(DISCOVERY_SOURCE_ORDER if available_sources is None else available_sources)
+        allowed.discard("x")
+        sources = [source for source in DISCOVERY_SOURCE_ORDER if source in allowed]
+        if not sources:
+            raise ValueError("No listing sources are available for global trending")
+        return schema.DiscoveryPlan(
+            domain="",
+            category=None,
+            subreddits=resolved or ["all"],
+            sources=sources,
+        )
 
     category = categories.detect_category(normalized_domain)
     candidate_subreddits = list(subreddits or categories.peer_subs_for(category))

skills/last30days/scripts/lib/render.py

@@ -79,9 +79,14 @@ def _format_discovery_engagement(
 
 def render_discovery(report: schema.DiscoveryReport) -> str:
     """Render a compact topic-per-section discovery brief."""
+    title = (
+        f"# Trending discovery: {report.domain}"
+        if report.domain
+        else "# Trending now"
+    )
     lines = [
         *_render_badge(),
-        f"# Trending discovery: {report.domain}",
+        title,
         "",
         f"Window: {report.range_from} to {report.range_to}",
         f"Feeds: {', '.join(report.plan.sources)}",
@@ -91,16 +96,42 @@ def render_discovery(report: schema.DiscoveryReport) -> str:
     lines.append("")
 
     if not report.topics:
-        lines.extend(["No trending topic clusters survived this sweep.", ""])
+        if report.outcome == "nothing-solid":
+            lines.extend([
+                "**Nothing solid this window.** No topic cleared the confidence "
+                "floor - not enough cross-source confirmation or engagement to "
+                "call anything a trend, and ranked noise would be worse than an "
+                "honest empty result.",
+                "",
+            ])
+            if report.weak_signal:
+                lines.extend([
+                    f"Closest weak signal: {report.weak_signal} (sub-floor; "
+                    "single-source or too little engagement).",
+                    "",
+                ])
+        else:
+            lines.extend(["No trending topic clusters survived this sweep.", ""])
     for topic in report.topics:
         momentum = "New this week" if topic.momentum == "new-this-week" else "Building"
+        confirmation = (
+            f" · confirmed across {topic.corroboration_count} sources"
+            if topic.corroboration_count >= 2 else ""
+        )
         lines.extend([
             f"## {topic.rank}. {topic.name}",
             "",
-            f"**Momentum:** {momentum} · velocity {topic.velocity_score:,.2f}",
+            f"**Momentum:** {momentum} · velocity {topic.velocity_score:,.2f}{confirmation}",
             "",
             topic.why_spiking,
             "",
+        ])
+        if topic.top_comment:
+            lines.extend([
+                f"**Community voice:** {topic.top_comment}",
+                "",
+            ])
+        lines.extend([
             f"**Evidence:** {_format_discovery_engagement(topic.engagement_by_source)}",
             "",
             f"**Research next:** `{topic.command}`",

skills/last30days/scripts/lib/rerank.py

@@ -84,6 +84,38 @@ def discovery_velocity_score(
     corroboration = 1.0 + (0.15 * max(0, source_count - 1))
     return round(raw * corroboration, 4)
 
+
+# Discovery confidence floor. The named 2026-07-12 failure mode: quiet feeds
+# left the sweep ranking noise against noise, and it dutifully emitted five
+# 1-like tweets as a "trend list". The floor makes "nothing solid this window"
+# a first-class outcome instead. Constants are deliberately tunable:
+# - FLOOR_MIN_ENGAGEMENT kills absolute junk (a 1-like tweet can never rank).
+# - A topic then clears via EITHER independent cross-source confirmation
+#   (>= FLOOR_MIN_SOURCES) OR a genuinely strong single-source spike
+#   (>= FLOOR_SINGLE_SOURCE_ENGAGEMENT) - a 1,600-point single-source HN
+#   thread is a real story, a 30-upvote single-source meme is not.
+FLOOR_MIN_ENGAGEMENT = 25.0
+FLOOR_MIN_SOURCES = 2
+FLOOR_SINGLE_SOURCE_ENGAGEMENT = 200.0
+
+
+def passes_discovery_floor(
+    *,
+    source_count: int,
+    engagement_total: float,
+    item_count: int,
+) -> bool:
+    """Whether a discovery topic's evidence is strong enough to show a user.
+
+    Below this floor the honest output is "nothing solid this window", not a
+    ranked list of whatever survived the sweep.
+    """
+    if item_count <= 0 or engagement_total < FLOOR_MIN_ENGAGEMENT:
+        return False
+    if source_count >= FLOOR_MIN_SOURCES:
+        return True
+    return engagement_total >= FLOOR_SINGLE_SOURCE_ENGAGEMENT
+
 # Engagement rescue: a high-engagement X post that is on-topic (entity-grounded
 # or first-party) cannot be fully zeroed by the other penalties. The floor is a
 # function of the post's engagement percentile *within the run's X pool* (so it

skills/last30days/scripts/lib/schema.py

@@ -278,7 +278,13 @@ class DiscoveryPlan:
 
 @dataclass(frozen=True)
 class DiscoveryTopic:
-    """One engagement-ranked topic produced by a discovery sweep."""
+    """One engagement-ranked topic produced by a discovery sweep.
+
+    ``top_comment`` is the strongest verbatim community comment from the
+    topic's enriched corpus (with attribution), present only on enriched runs.
+    ``corroboration_count`` is the number of distinct sources confirming the
+    topic - the floor's cross-source signal, surfaced for readers.
+    """
 
     rank: int
     name: str
@@ -289,11 +295,20 @@ class DiscoveryTopic:
     engagement_by_source: dict[str, dict[str, float | int]]
     command: str
     evidence_urls: list[str] = field(default_factory=list)
+    top_comment: str | None = None
+    corroboration_count: int = 0
 
 
 @dataclass
 class DiscoveryReport:
-    """Versioned result of a domain-level listing sweep."""
+    """Versioned result of a domain-level listing sweep.
+
+    ``outcome`` is "ok" when at least one topic cleared the confidence floor,
+    "nothing-solid" when the window's evidence was all sub-floor - an honest
+    empty result instead of ranked noise. ``weak_signal`` optionally names the
+    strongest sub-floor topic so a nothing-solid brief can still say what came
+    closest.
+    """
 
     domain: str
     range_from: str
@@ -303,6 +318,8 @@ class DiscoveryReport:
     topics: list[DiscoveryTopic]
     source_status: dict[str, SourceOutcome] = field(default_factory=dict)
     warnings: list[str] = field(default_factory=list)
+    outcome: str = "ok"
+    weak_signal: str | None = None
 
 
 @dataclass
@@ -841,8 +858,12 @@ def to_discovery_export(report: DiscoveryReport) -> dict[str, Any]:
                 "engagement": topic.engagement_by_source,
                 "command": topic.command,
                 "evidence_urls": list(topic.evidence_urls),
+                "top_comment": topic.top_comment,
+                "corroboration_count": topic.corroboration_count,
             }
             for topic in report.topics
         ],
         "warnings": list(report.warnings),
+        "outcome": report.outcome,
+        "weak_signal": report.weak_signal,
     }

tests/test_discover_enrich.py

@@ -0,0 +1,214 @@
+"""U3 - enrichment stage: full pipeline pass per nominated topic.
+
+The fault-tolerance contract is the point of these tests: one topic failing or
+running past the batch budget must never sink the others, and the batch never
+raises.
+"""
+
+import time
+from unittest import mock
+
+from lib import pipeline, schema
+
+
+def _nomination(name: str, score: float = 50.0) -> pipeline.Nomination:
+    return pipeline.Nomination(name=name, seed_score=score, items=[], summary=name)
+
+
+def _report(topic: str) -> schema.Report:
+    return schema.Report(
+        topic=topic,
+        range_from="2026-06-10",
+        range_to="2026-07-10",
+        generated_at="2026-07-10T00:00:00+00:00",
+        provider_runtime=schema.ProviderRuntime(
+            reasoning_provider="none",
+            planner_model="deterministic",
+            rerank_model="deterministic",
+        ),
+        query_plan=schema.QueryPlan(
+            intent="factual",
+            freshness_mode="balanced_recent",
+            cluster_mode="none",
+            raw_topic=topic,
+            subqueries=[],
+            source_weights={},
+        ),
+        clusters=[],
+        ranked_candidates=[],
+        items_by_source={},
+        errors_by_source={},
+    )
+
+
+def test_enrich_all_success_preserves_order():
+    nominations = [_nomination("Topic A"), _nomination("Topic B"), _nomination("Topic C")]
+
+    def fake_run(*, topic, **_kwargs):
+        return _report(topic)
+
+    with mock.patch.object(pipeline, "run", side_effect=fake_run):
+        enriched = pipeline.enrich_nominations(nominations, config={})
+
+    assert [entry.nomination.name for entry in enriched] == ["Topic A", "Topic B", "Topic C"]
+    assert all(entry.report is not None for entry in enriched)
+    assert all(entry.error is None for entry in enriched)
+
+
+def test_enrich_one_failure_does_not_sink_the_batch():
+    nominations = [_nomination("Good"), _nomination("Bad"), _nomination("Also good")]
+
+    def fake_run(*, topic, **_kwargs):
+        if topic == "Bad":
+            raise RuntimeError("upstream exploded")
+        return _report(topic)
+
+    with mock.patch.object(pipeline, "run", side_effect=fake_run):
+        enriched = pipeline.enrich_nominations(nominations, config={})
+
+    by_name = {entry.nomination.name: entry for entry in enriched}
+    assert by_name["Good"].report is not None
+    assert by_name["Also good"].report is not None
+    assert by_name["Bad"].report is None
+    assert "upstream exploded" in (by_name["Bad"].error or "")
+
+
+def test_enrich_budget_expiry_drops_slow_topic_to_nomination_only():
+    nominations = [_nomination("Fast"), _nomination("Slow")]
+
+    def fake_run(*, topic, **_kwargs):
+        if topic == "Slow":
+            time.sleep(5)
+        return _report(topic)
+
+    with mock.patch.object(pipeline, "run", side_effect=fake_run):
+        enriched = pipeline.enrich_nominations(
+            nominations, config={}, budget_seconds=1.0, max_workers=2,
+        )
+
+    by_name = {entry.nomination.name: entry for entry in enriched}
+    assert by_name["Fast"].report is not None
+    assert by_name["Slow"].report is None
+    assert "budget" in (by_name["Slow"].error or "")
+
+
+def test_enrich_runs_as_internal_subrun():
+    """Sub-runs must use the internal_subrun lane (no library context, capped
+    inner workers) exactly like comparison-mode entity passes."""
+    seen: dict[str, object] = {}
+
+    def fake_run(*, topic, **kwargs):
+        seen.update(kwargs)
+        return _report(topic)
+
+    with mock.patch.object(pipeline, "run", side_effect=fake_run):
+        pipeline.enrich_nominations([_nomination("One")], config={})
+
+    assert seen.get("internal_subrun") is True
+
+
+def test_enrich_empty_nominations_returns_empty():
+    assert pipeline.enrich_nominations([], config={}) == []
+
+
+def test_enrich_workers_are_daemon_threads():
+    """Stragglers must not block interpreter exit: every enrichment worker runs
+    as a daemon thread (the P1 from PR #816 review - a hung sub-run kept the
+    process alive past the wall-clock budget with non-daemon executor threads)."""
+    import threading
+
+    daemon_flags: list[bool] = []
+
+    def fake_run(*, topic, **_kwargs):
+        daemon_flags.append(threading.current_thread().daemon)
+        return _report(topic)
+
+    with mock.patch.object(pipeline, "run", side_effect=fake_run):
+        pipeline.enrich_nominations([_nomination("One"), _nomination("Two")], config={})
+
+    assert daemon_flags and all(daemon_flags)
+
+
+def test_enrich_concurrency_capped_by_semaphore():
+    """Never more than max_workers sub-runs in flight."""
+    import threading
+
+    lock = threading.Lock()
+    state = {"active": 0, "peak": 0}
+
+    def fake_run(*, topic, **_kwargs):
+        with lock:
+            state["active"] += 1
+            state["peak"] = max(state["peak"], state["active"])
+        time.sleep(0.05)
+        with lock:
+            state["active"] -= 1
+        return _report(topic)
+
+    nominations = [_nomination(f"T{i}") for i in range(6)]
+    with mock.patch.object(pipeline, "run", side_effect=fake_run):
+        enriched = pipeline.enrich_nominations(nominations, config={}, max_workers=2)
+
+    assert state["peak"] <= 2
+    assert all(entry.report is not None for entry in enriched)
+
+
+def test_enrichment_reaches_all_sources_by_default():
+    """No user source filter -> sub-runs get requested_sources=None, which is
+    what lets Techmeme, arXiv, YouTube, and Polymarket reach discovery despite
+    having no river feed of their own."""
+    seen: dict[str, object] = {}
+
+    def fake_run(*, topic, **kwargs):
+        seen.update(kwargs)
+        return _report(topic)
+
+    raw = {
+        "id": "seed1",
+        "title": "AI agents breakthrough sweeps the industry",
+        "url": "https://example.com/seed1",
+        "hn_url": "https://news.ycombinator.com/item?id=1",
+        "author": "example",
+        "date": "2026-07-09",
+        "engagement": {"points": 900, "comments": 400},
+        "relevance": 0.9,
+    }
+    with mock.patch.object(pipeline, "available_sources", return_value=["hackernews"]), \
+         mock.patch.object(pipeline, "_fetch_discovery_source", return_value=([raw], None)), \
+         mock.patch.object(pipeline, "run", side_effect=fake_run):
+        pipeline.run_discover(
+            domain="AI agents", config={}, as_of_date="2026-07-10", enrich=True,
+        )
+
+    assert seen.get("internal_subrun") is True
+    assert seen.get("requested_sources") is None
+
+
+def test_user_source_boundary_holds_through_enrichment():
+    """--search reddit must bound the sub-runs too, not just the sweep."""
+    seen: dict[str, object] = {}
+
+    def fake_run(*, topic, **kwargs):
+        seen.update(kwargs)
+        return _report(topic)
+
+    raw = {
+        "id": "seed1",
+        "title": "AI agents breakthrough sweeps the industry",
+        "url": "https://reddit.com/r/x/seed1",
+        "subreddit": "example",
+        "date": "2026-07-09",
+        "engagement": {"score": 900, "num_comments": 400},
+        "selftext": "AI agents breakthrough",
+        "relevance": 0.9,
+    }
+    with mock.patch.object(pipeline, "available_sources", return_value=["reddit"]), \
+         mock.patch.object(pipeline, "_fetch_discovery_source", return_value=([raw], None)), \
+         mock.patch.object(pipeline, "run", side_effect=fake_run):
+        pipeline.run_discover(
+            domain="AI agents", config={}, as_of_date="2026-07-10",
+            requested_sources=["reddit"], enrich=True,
+            enrich_requested_sources=["reddit"],
+        )
+
+    assert seen.get("requested_sources") == ["reddit"]

tests/test_discover_floor.py

@@ -0,0 +1,190 @@
+"""U4 - confidence floor: the fix for discovery's ranked-junk failure mode.
+
+The named 2026-07-12 regression: on quiet windows ("sports", "AI") the sweep
+ranked noise against noise and emitted five 1-like tweets as a trend list.
+These tests pin the new contract: sub-floor evidence never ranks, and the
+honest outcome is "nothing-solid" with the strongest weak signal named.
+"""
+
+from unittest import mock
+
+from lib import pipeline, rerank, schema
+
+
+def _x_item(item_id: str, text: str, likes: int, *, date: str = "2026-07-09") -> dict:
+    return {
+        "id": item_id,
+        "text": text,
+        "url": f"https://x.com/example/status/{item_id}",
+        "author_handle": "example",
+        "date": date,
+        "engagement": {"likes": likes, "reposts": 0, "replies": 0, "quotes": 0},
+        "relevance": 0.9,
+    }
+
+
+def _hn_item(item_id: str, title: str, points: int, comments: int, *, date: str = "2026-07-09") -> dict:
+    return {
+        "id": item_id,
+        "title": title,
+        "url": f"https://example.com/{item_id}",
+        "hn_url": f"https://news.ycombinator.com/item?id={item_id}",
+        "author": "example",
+        "date": date,
+        "engagement": {"points": points, "comments": comments},
+        "relevance": 0.9,
+    }
+
+
+def _reddit_item(item_id: str, title: str, score: int, comments: int, *, date: str = "2026-07-09") -> dict:
+    return {
+        "id": item_id,
+        "title": title,
+        "url": f"https://reddit.com/r/example/comments/{item_id}",
+        "subreddit": "example",
+        "date": date,
+        "engagement": {"score": score, "num_comments": comments},
+        "selftext": title,
+        "relevance": 0.9,
+    }
+
+
+def _run_discover_with(items_by_source: dict[str, list[dict]], **kwargs) -> schema.DiscoveryReport:
+    def fake_fetch(source, plan, *, from_date, to_date, depth, mock, config, keyword_gate=True):
+        return items_by_source.get(source, []), None
+
+    with mock.patch.object(
+        pipeline, "available_sources", return_value=list(items_by_source),
+    ), mock.patch.object(
+        pipeline, "_fetch_discovery_source", side_effect=fake_fetch,
+    ):
+        return pipeline.run_discover(
+            domain=kwargs.pop("domain", "sports"),
+            config={},
+            as_of_date="2026-07-10",
+            **kwargs,
+        )
+
+
+def test_junk_corpus_returns_nothing_solid_not_ranked_noise():
+    """THE regression: five single-source 1-like tweets (the 'sports' corpus)
+    must produce an honest empty result, never a ranked junk list."""
+    report = _run_discover_with({
+        "x": [
+            _x_item("junk1", "Wii Sports nostalgia thread about sports", 1),
+            _x_item("junk2", "kids travel sports burnout post", 1),
+            _x_item("junk3", "motorsports vs stick and ball sports", 1),
+            _x_item("junk4", "midjourney skateboarder sports prompt", 1),
+            _x_item("junk5", "manga review mentioning sports matches", 1),
+        ],
+    })
+
+    assert report.topics == []
+    assert report.outcome == "nothing-solid"
+    assert report.weak_signal is not None
+    assert any("confidence floor" in warning for warning in report.warnings)
+
+
+def test_strong_single_source_spike_clears_floor():
+    """A 1,084-point / 577-comment single-source HN thread (the '60% of US
+    consumers' case) is a real story and must rank."""
+    report = _run_discover_with(
+        {"hackernews": [_hn_item("big1", "Sixty percent of consumers say AI in sports ads is a turnoff", 1084, 577)]},
+        domain="sports",
+    )
+
+    assert report.outcome == "ok"
+    assert len(report.topics) == 1
+    assert "turnoff" in report.topics[0].name.lower() or report.topics[0].velocity_score > 0
+
+
+def test_weak_single_source_item_stays_buried():
+    """A 30-upvote single-source post is not a trend."""
+    report = _run_discover_with(
+        {"reddit": [_reddit_item("meh1", "Mildly interesting sports take", 25, 4)]},
+    )
+
+    assert report.topics == []
+    assert report.outcome == "nothing-solid"
+    assert report.weak_signal is not None
+
+
+def test_mixed_corpus_emits_only_floor_clearing_topics():
+    """Strong multi-source story ranks; 1-like junk is silently dropped."""
+    report = _run_discover_with({
+        "hackernews": [_hn_item("story1", "NBA finals collapse shocks sports world", 450, 200)],
+        "reddit": [_reddit_item("story1r", "NBA finals collapse shocks sports world", 900, 400)],
+        "x": [_x_item("junkA", "random sports meme", 1)],
+    })
+
+    assert report.outcome == "ok"
+    assert len(report.topics) >= 1
+    names = " ".join(topic.name.lower() for topic in report.topics)
+    assert "nba" in names or "finals" in names
+    assert all(topic.velocity_score > 0 for topic in report.topics)
+    # The 1-like junk never appears.
+    assert all("meme" not in topic.name.lower() for topic in report.topics)
+
+
+def test_enriched_evidence_is_judged_not_seed_evidence():
+    """With enrich=True, a topic whose seed was thin but whose full-pipeline
+    corpus is rich clears the floor on the enriched evidence."""
+    seed = {"x": [_x_item("seed1", "quiet sports story gathering steam", 40)]}
+
+    def fake_run(*, topic, **_kwargs):
+        items = {
+            "reddit": [
+                schema.SourceItem(
+                    item_id="e1", source="reddit", title=topic, body=topic,
+                    url="https://reddit.com/r/x/1", published_at="2026-07-09",
+                    engagement={"score": 800, "num_comments": 300}, snippet=topic,
+                ),
+            ],
+            "hackernews": [
+                schema.SourceItem(
+                    item_id="e2", source="hackernews", title=topic, body=topic,
+                    url="https://example.com/e2", published_at="2026-07-09",
+                    engagement={"points": 400, "comments": 150}, snippet=topic,
+                ),
+            ],
+        }
+        return schema.Report(
+            topic=topic,
+            range_from="2026-06-10", range_to="2026-07-10",
+            generated_at="2026-07-10T00:00:00+00:00",
+            provider_runtime=schema.ProviderRuntime(
+                reasoning_provider="none",
+                planner_model="deterministic",
+                rerank_model="deterministic",
+            ),
+            query_plan=schema.QueryPlan(
+                intent="factual", freshness_mode="balanced_recent",
+                cluster_mode="none", raw_topic=topic, subqueries=[],
+                source_weights={},
+            ),
+            clusters=[], ranked_candidates=[],
+            items_by_source=items, errors_by_source={},
+        )
+
+    with mock.patch.object(pipeline, "run", side_effect=fake_run):
+        report = _run_discover_with(seed, enrich=True)
+
+    assert report.outcome == "ok"
+    assert len(report.topics) == 1
+    topic = report.topics[0]
+    # Judged on the enriched corpus: multi-source, enriched engagement.
+    assert set(topic.sources) == {"hackernews", "reddit"}
+    assert "evidence item" in topic.why_spiking
+
+
+def test_passes_discovery_floor_policy():
+    floor = rerank.passes_discovery_floor
+    # Absolute junk gate.
+    assert not floor(source_count=1, engagement_total=1, item_count=1)
+    assert not floor(source_count=3, engagement_total=10, item_count=5)
+    assert not floor(source_count=2, engagement_total=500, item_count=0)
+    # Multi-source with modest engagement clears.
+    assert floor(source_count=2, engagement_total=30, item_count=2)
+    # Single-source needs a genuinely strong spike.
+    assert not floor(source_count=1, engagement_total=100, item_count=3)
+    assert floor(source_count=1, engagement_total=1600, item_count=1)

tests/test_discover_mode.py

@@ -77,6 +77,38 @@ def test_discovery_plan_keeps_keyless_reddit_for_unknown_domains():
     assert plan.sources == ["reddit", "hackernews"]
 
 
+def test_discovery_plan_empty_domain_is_global_trending():
+    """Bare --discover: sweep every river feed's hot list; X sits out of the
+    nominate stage because its search lane needs a keyword."""
+    plan = planner.build_discovery_plan(
+        "",
+        available_sources=["reddit", "hackernews", "digg", "x"],
+    )
+
+    assert plan.domain == ""
+    assert plan.category is None
+    assert plan.subreddits == ["all"]
+    assert plan.sources == ["reddit", "hackernews", "digg"]
+    assert "x" not in plan.sources
+
+
+def test_global_discovery_disables_keyword_gate():
+    """Global trending fetches with keyword_gate=False; domain runs keep it on."""
+    seen: dict[str, bool] = {}
+
+    def fake_fetch(source, plan, *, from_date, to_date, depth, mock, config, keyword_gate=True):
+        seen[plan.domain or "global"] = keyword_gate
+        return [], None
+
+    with mock.patch.object(pipeline, "available_sources", return_value=["hackernews"]), \
+         mock.patch.object(pipeline, "_fetch_discovery_source", side_effect=fake_fetch):
+        pipeline.run_discover(domain="", config={}, as_of_date="2026-07-10")
+        pipeline.run_discover(domain="AI agents", config={}, as_of_date="2026-07-10")
+
+    assert seen["global"] is False
+    assert seen["AI agents"] is True
+
+
 def test_uncategorized_discovery_uses_parseable_r_all_listing_paths():
     card = (
         '<shreddit-post permalink="/r/gardening/comments/abc123/urban_garden/" '
@@ -230,7 +262,7 @@ def test_discovery_renderer_snapshot():
 
 
 def test_keyless_discovery_degrades_without_digg():
-    def fake_fetch(source, plan, *, from_date, to_date, depth, mock, config):
+    def fake_fetch(source, plan, *, from_date, to_date, depth, mock, config, keyword_gate=True):
         return pipeline._mock_discovery_items(source, plan.domain, to_date), None
 
     with mock.patch.object(pipeline, "available_sources", return_value=["reddit", "hackernews"]), \
@@ -269,7 +301,8 @@ def test_discovery_drops_zero_velocity_clusters():
         )
 
     assert report.topics == []
-    assert "Fewer than five topic clusters survived this domain sweep." in report.warnings
+    assert report.outcome == "nothing-solid"
+    assert any("confidence floor" in warning for warning in report.warnings)
 
 
 def test_explicit_unavailable_discovery_source_does_not_widen_to_other_sources():
@@ -302,7 +335,7 @@ def test_discovery_reads_browser_credentials_and_does_not_schedule_pending_x():
         assert x_pending is False
         return ["reddit", "hackernews"] + (["x"] if x_pending is not False else [])
 
-    def fake_fetch(source, plan, *, from_date, to_date, depth, mock, config):
+    def fake_fetch(source, plan, *, from_date, to_date, depth, mock, config, keyword_gate=True):
         fetched_sources.append(source)
         return pipeline._mock_discovery_items(source, plan.domain, to_date), None
 
@@ -342,7 +375,7 @@ def test_authenticated_x_discovery_uses_available_backend():
 
 
 def test_listing_failure_is_not_reported_as_clean_no_results():
-    def fake_fetch(source, plan, *, from_date, to_date, depth, mock, config):
+    def fake_fetch(source, plan, *, from_date, to_date, depth, mock, config, keyword_gate=True):
         if source == "reddit":
             return [], "connection timed out"
         return pipeline._mock_discovery_items(source, plan.domain, to_date), None
@@ -435,6 +468,72 @@ def test_discovery_cli_json_contract_and_mutual_exclusion():
     assert "mutually exclusive" in drill_conflict.stderr
 
 
+def test_discovery_cli_bare_discover_is_global_trending():
+    """Bare --discover (no domain) must run global trending, not error."""
+    result = subprocess.run(
+        [
+            sys.executable,
+            "skills/last30days/scripts/last30days.py",
+            "--discover",
+            "--mock",
+            "--emit=json",
+        ],
+        cwd=REPO_ROOT,
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+    assert result.returncode == 0, result.stderr
+    payload = json.loads(result.stdout)
+    assert payload["kind"] == "discovery"
+    assert payload["domain"] == ""
+    assert payload["outcome"] in {"ok", "nothing-solid"}
+
+
+def test_discovery_cli_shallow_skips_enrichment():
+    """--discover-shallow ranks on listing evidence only (still floored)."""
+    result = subprocess.run(
+        [
+            sys.executable,
+            "skills/last30days/scripts/last30days.py",
+            "--discover", "AI agents",
+            "--discover-shallow",
+            "--mock",
+            "--emit=json",
+        ],
+        cwd=REPO_ROOT,
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+    assert result.returncode == 0, result.stderr
+    payload = json.loads(result.stdout)
+    assert payload["results"], "shallow mock sweep should still rank mock topics"
+    assert all(
+        "listing item" in topic["why_spiking"] for topic in payload["results"]
+    ), "shallow mode must be judged on listing evidence, not enriched corpora"
+
+
+def test_discovery_cli_rejects_shallow_without_discover():
+    """--discover-shallow on a normal topic run must error, not silently no-op
+    into a full research pass (P2 from PR #816 review)."""
+    result = subprocess.run(
+        [
+            sys.executable,
+            "skills/last30days/scripts/last30days.py",
+            "AI agents",
+            "--discover-shallow",
+            "--mock",
+        ],
+        cwd=REPO_ROOT,
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+    assert result.returncode == 2
+    assert "--discover-shallow only applies to --discover runs" in result.stderr
+
+
 def test_discovery_cli_rejects_historical_as_of():
     result = subprocess.run(
         [

tests/test_discover_nominate.py

@@ -0,0 +1,92 @@
+"""U1 - nominate stage: river/listing candidate discovery.
+
+Covers the two behaviors the nominate stage adds over the old inline sweep:
+a ``keyword_gate`` toggle (domain scoping vs global trending) and
+fault-tolerant per-source failure recording that never raises.
+"""
+
+from unittest import mock
+
+from lib import pipeline, reddit_listing, schema
+
+
+def _plan(domain: str, sources: list[str], subreddits: list[str] | None = None) -> schema.DiscoveryPlan:
+    return schema.DiscoveryPlan(
+        domain=domain,
+        category=None,
+        subreddits=subreddits or ["all"],
+        sources=sources,
+    )
+
+
+def test_fetch_discovery_source_reddit_gate_filters_off_domain():
+    """With the keyword gate on, off-domain listing items are dropped; with it
+    off (global trending), the feed's own hot ranking is kept verbatim."""
+    plan = _plan("AI agents", ["reddit"])
+    payload = {
+        "items": [
+            {"title": "New AI agents framework launched", "selftext": ""},
+            {"title": "Gardening tips for spring", "selftext": ""},
+        ],
+        "errors": [],
+    }
+    with mock.patch.object(reddit_listing, "fetch_discovery_listings", return_value=payload):
+        gated, _ = pipeline._fetch_discovery_source(
+            "reddit", plan,
+            from_date="2026-06-10", to_date="2026-07-10",
+            depth="default", mock=False, config={}, keyword_gate=True,
+        )
+        ungated, _ = pipeline._fetch_discovery_source(
+            "reddit", plan,
+            from_date="2026-06-10", to_date="2026-07-10",
+            depth="default", mock=False, config={}, keyword_gate=False,
+        )
+
+    assert [item["title"] for item in gated] == ["New AI agents framework launched"]
+    assert len(ungated) == 2
+
+
+def test_nominate_candidates_threads_keyword_gate():
+    """nominate_candidates forwards its keyword_gate to each source fetch, so a
+    global (no-domain) run really does disable the gate."""
+    plan = _plan("", ["reddit"])
+    seen: dict[str, bool] = {}
+
+    def fake_fetch(source, plan, *, from_date, to_date, depth, mock, config, keyword_gate=True):
+        seen["keyword_gate"] = keyword_gate
+        return [], None
+
+    with mock.patch.object(pipeline, "_fetch_discovery_source", side_effect=fake_fetch):
+        pipeline.nominate_candidates(
+            plan,
+            from_date="2026-06-10", to_date="2026-07-10",
+            depth="default", mock=False, config={}, lookback_days=30,
+            keyword_gate=False,
+        )
+
+    assert seen["keyword_gate"] is False
+
+
+def test_nominate_candidates_records_source_failure_without_raising():
+    """One dead feed is recorded on the bundle as a failure; the surviving feed
+    still yields candidates, and the call never raises."""
+    plan = _plan("AI agents", ["reddit", "hackernews"])
+
+    def fake_fetch(source, plan, *, from_date, to_date, depth, mock, config, keyword_gate=True):
+        if source == "hackernews":
+            raise TimeoutError("hn listing timed out")
+        return pipeline._mock_discovery_items(source, plan.domain, to_date), None
+
+    with mock.patch.object(pipeline, "_fetch_discovery_source", side_effect=fake_fetch):
+        bundle = pipeline.nominate_candidates(
+            plan,
+            from_date="2026-06-10", to_date="2026-07-10",
+            depth="default", mock=False, config={}, lookback_days=30,
+        )
+
+    # Surviving source produced candidates.
+    assert bundle.items_by_source.get("reddit")
+    # Dead source recorded as a failure, not silently dropped or raised.
+    hackernews = bundle.source_status["hackernews"]
+    assert hackernews.state not in (schema.NO_RESULTS,)
+    assert "ok" != hackernews.state

tests/test_discover_nominate_topics.py

@@ -0,0 +1,137 @@
+"""U2 - nomination ranking: cluster nominated items into named, seed-ranked
+candidate topics.
+
+nominate_topics() is the contract between the nominate stage and the
+enrichment fan-out: names ordered by seed velocity, deduped casefold, never
+padded past what the evidence supports.
+"""
+
+from lib import pipeline, schema
+
+
+def _item(
+    item_id: str,
+    source: str,
+    title: str,
+    *,
+    published_at: str = "2026-07-09",
+    engagement: dict[str, int | float] | None = None,
+) -> schema.SourceItem:
+    return schema.SourceItem(
+        item_id=item_id,
+        source=source,
+        title=title,
+        body=title,
+        url=f"https://{source}.example/{item_id}",
+        published_at=published_at,
+        engagement=engagement or {},
+        snippet=f"Evidence about {title}",
+    )
+
+
+def _bundle(items: list[schema.SourceItem]) -> schema.RetrievalBundle:
+    bundle = schema.RetrievalBundle()
+    by_source: dict[str, list[schema.SourceItem]] = {}
+    for item in items:
+        by_source.setdefault(item.source, []).append(item)
+    for source, source_items in by_source.items():
+        bundle.add_items("discovery-listings", source, source_items)
+    return bundle
+
+
+def _query_plan(domain: str, sources: list[str]) -> schema.QueryPlan:
+    return schema.QueryPlan(
+        intent="breaking_news",
+        freshness_mode="breaking",
+        cluster_mode="story",
+        raw_topic=domain,
+        subqueries=[schema.SubQuery(
+            label="discovery-listings",
+            search_query=domain,
+            ranking_query=f"What is accelerating in {domain}?",
+            sources=sources,
+        )],
+        source_weights={source: 1.0 for source in sources},
+        notes=["discover-mode", "listing-sweep"],
+    )
+
+
+def _plan(domain: str, sources: list[str]) -> schema.DiscoveryPlan:
+    return schema.DiscoveryPlan(
+        domain=domain, category=None, subreddits=["all"], sources=sources,
+    )
+
+
+def test_nominations_ranked_by_seed_velocity():
+    """A high-engagement recent story outranks a low-engagement one."""
+    items = [
+        _item("hot1", "hackernews", "GPT-6 rumors flood the valley",
+              engagement={"points": 900, "num_comments": 400}),
+        _item("cold1", "hackernews", "Minor framework patch notes released",
+              engagement={"points": 3, "num_comments": 1}),
+    ]
+    nominations = pipeline.nominate_topics(
+        _bundle(items), _query_plan("AI", ["hackernews"]), _plan("AI", ["hackernews"]),
+        to_date="2026-07-10", limit=10,
+    )
+    assert nominations, "expected at least one nomination"
+    assert "GPT-6" in nominations[0].name
+    assert nominations[0].seed_score >= (nominations[-1].seed_score)
+
+
+def test_nominations_dedupe_names_casefold():
+    """Two clusters resolving to the same casefolded name yield one nomination."""
+    items = [
+        _item("a1", "hackernews", "OpenAI Agent SDK",
+              engagement={"points": 500, "num_comments": 100}),
+        _item("a2", "reddit", "openai agent sdk",
+              engagement={"score": 300, "num_comments": 80}),
+    ]
+    nominations = pipeline.nominate_topics(
+        _bundle(items), _query_plan("AI agents", ["hackernews", "reddit"]),
+        _plan("AI agents", ["hackernews", "reddit"]),
+        to_date="2026-07-10", limit=10,
+    )
+    names = [nomination.name.casefold() for nomination in nominations]
+    assert len(names) == len(set(names))
+
+
+def test_fewer_clusters_than_limit_returns_all_without_padding():
+    items = [
+        _item("only1", "hackernews", "Quantum breakthrough announced",
+              engagement={"points": 250, "num_comments": 60}),
+    ]
+    nominations = pipeline.nominate_topics(
+        _bundle(items), _query_plan("quantum", ["hackernews"]),
+        _plan("quantum", ["hackernews"]),
+        to_date="2026-07-10", limit=8,
+    )
+    assert 1 <= len(nominations) < 8
+
+
+def test_zero_velocity_clusters_are_dropped():
+    """Items with no engagement produce no nomination at all."""
+    items = [
+        _item("dead1", "hackernews", "Silent post nobody engaged with",
+              engagement={"points": 0, "num_comments": 0}),
+    ]
+    nominations = pipeline.nominate_topics(
+        _bundle(items), _query_plan("AI", ["hackernews"]), _plan("AI", ["hackernews"]),
+        to_date="2026-07-10", limit=8,
+    )
+    assert nominations == []
+
+
+def test_nomination_carries_leader_summary_and_items():
+    items = [
+        _item("s1", "hackernews", "Rust rewrite of the Linux scheduler",
+              engagement={"points": 700, "num_comments": 250}),
+    ]
+    nominations = pipeline.nominate_topics(
+        _bundle(items), _query_plan("Linux", ["hackernews"]), _plan("Linux", ["hackernews"]),
+        to_date="2026-07-10", limit=8,
+    )
+    assert nominations
+    top = nominations[0]
+    assert top.items and top.items[0].item_id == "s1"
+    assert top.summary

tests/test_discover_render.py

@@ -0,0 +1,132 @@
+"""U6 - rich discovery rendering: enriched trend cards, the global-trending
+header, and the honest nothing-solid empty state."""
+
+from lib import pipeline, render, schema
+
+
+def _topic(rank: int, name: str, **overrides) -> schema.DiscoveryTopic:
+    fields = dict(
+        rank=rank,
+        name=name,
+        why_spiking=f"3 evidence items about {name}.",
+        momentum="building",
+        velocity_score=42.5,
+        sources=["hackernews", "reddit"],
+        engagement_by_source={"hackernews": {"points": 500}},
+        command=f'/last30days "{name}"',
+        evidence_urls=[f"https://example.com/{rank}"],
+    )
+    fields.update(overrides)
+    return schema.DiscoveryTopic(**fields)
+
+
+def _report(**overrides) -> schema.DiscoveryReport:
+    fields = dict(
+        domain="AI agents",
+        range_from="2026-06-10",
+        range_to="2026-07-10",
+        generated_at="2026-07-10T00:00:00+00:00",
+        plan=schema.DiscoveryPlan(
+            domain="AI agents", category=None, subreddits=["all"],
+            sources=["reddit", "hackernews"],
+        ),
+        topics=[_topic(1, "OpenAI Agent SDK")],
+        source_status={},
+        warnings=[],
+    )
+    fields.update(overrides)
+    return schema.DiscoveryReport(**fields)
+
+
+def test_global_trending_header():
+    report = _report(
+        domain="",
+        plan=schema.DiscoveryPlan(
+            domain="", category=None, subreddits=["all"],
+            sources=["reddit", "hackernews", "digg"],
+        ),
+    )
+    rendered = render.render_discovery(report)
+    assert "# Trending now" in rendered
+    assert "Trending discovery:" not in rendered
+
+
+def test_domain_header_unchanged():
+    rendered = render.render_discovery(_report())
+    assert "# Trending discovery: AI agents" in rendered
+
+
+def test_nothing_solid_renders_honest_empty_state():
+    report = _report(
+        topics=[],
+        outcome="nothing-solid",
+        weak_signal="Wii Sports nostalgia thread",
+        warnings=["No topic cleared the discovery confidence floor this window."],
+    )
+    rendered = render.render_discovery(report)
+    assert "Nothing solid this window." in rendered
+    assert "Wii Sports nostalgia thread" in rendered
+    assert "## 1." not in rendered  # no fabricated topic cards
+
+
+def test_community_voice_and_corroboration_render():
+    report = _report(topics=[_topic(
+        1, "OpenAI Agent SDK",
+        top_comment='"This changes everything about agent tooling" - dev_voice (1,200 votes)',
+        corroboration_count=3,
+    )])
+    rendered = render.render_discovery(report)
+    assert "**Community voice:**" in rendered
+    assert "dev_voice" in rendered
+    assert "confirmed across 3 sources" in rendered
+
+
+def test_no_voice_line_when_topic_has_no_comment():
+    rendered = render.render_discovery(_report())
+    assert "**Community voice:**" not in rendered
+
+
+def test_best_community_comment_prefers_platform_normalized_strength():
+    items = [
+        schema.SourceItem(
+            item_id="a", source="reddit", title="t", body="t",
+            url="https://reddit.com/a", metadata={"top_comments": [
+                {"text": "the sharpest take in the thread by far", "score": 4000, "author": "u/sharp"},
+                {"text": "short", "score": 9999},
+            ]},
+        ),
+        schema.SourceItem(
+            item_id="b", source="hackernews", title="t", body="t",
+            url="https://news.ycombinator.com/b", metadata={"top_comments": [
+                {"text": "a modest but thoughtful comment here", "score": 3, "author": "hn_user"},
+            ]},
+        ),
+    ]
+    comment = pipeline._best_community_comment(items)
+    assert comment is not None
+    assert "u/sharp" in comment
+    assert "4,000 votes" in comment
+    # Sub-12-char comment bodies never surface.
+    assert "short" not in comment
+
+
+def test_best_community_comment_strips_leading_quote_chars():
+    """A comment body that itself starts with a quote must not render as
+    doubled quotes inside the wrapping quotes."""
+    items = [schema.SourceItem(
+        item_id="a", source="reddit", title="t", body="t",
+        url="https://r.example/a", metadata={"top_comments": [
+            {"text": '"This much is clear: the quote should not double up', "score": 100, "author": "u/q"},
+        ]},
+    )]
+    comment = pipeline._best_community_comment(items)
+    assert comment is not None
+    assert '""' not in comment
+    assert comment.startswith('"This much is clear')
+
+
+def test_best_community_comment_none_when_no_comments():
+    items = [schema.SourceItem(
+        item_id="a", source="reddit", title="t", body="t", url="https://r.example/a",
+    )]
+    assert pipeline._best_community_comment(items) is None