feat: show who writes the wire in a voices panel

backend/app/catalog/facets.py

@@ -1,4 +1,4 @@
-"""Facet counts for the feed's filters and rails: how many articles each publisher, tag and origin has."""
+"""Facet counts for the feed's filters and rails: how many articles each publisher, tag and origin has, and who wrote them."""
 
 from __future__ import annotations
 
@@ -22,6 +22,9 @@
     PublisherKind,
     Tag,
     TagFacet,
+    Voice,
+    VoiceArticle,
+    VoiceKind,
 )
 
 
@@ -101,3 +104,67 @@ def count(*conditions: ColumnElement[bool]) -> int:
         unlabelled=unlabelled,
         found_via=[FoundViaCount(name=n, count=c) for n, c in found_via_rows],
     )
+
+
+# A bar per article; past this many the row would outgrow the rail.
+VOICE_SCORES = 15
+
+
+def voices(
+    session: Session, kind: VoiceKind, since: datetime, limit: int = 8
+) -> list[Voice]:
+    """The publishers of one kind with the most articles since `since`, busiest
+    first and then by their best score: each with its scores oldest first and
+    its best article."""
+    conditions: list[ColumnElement[bool]] = [
+        col(Article.published_at) >= since,
+        col(Publisher.kind) == PublisherKind(kind.value),
+    ]
+    # On GitHub and Hugging Face the account is the publisher (ADR 13), so a
+    # writer's repos and models would fill the tab with repo owners.
+    if kind == VoiceKind.individual:
+        conditions.append(col(Article.kind) == ArticleKind.post)
+
+    count_expr = func.count(col(Article.id))
+    best_expr = func.max(col(Article.score))
+    top = session.exec(
+        select(Publisher.id, Publisher.slug, Publisher.name, count_expr)
+        .join(Article, col(Article.publisher_id) == col(Publisher.id))
+        .where(*conditions)
+        .group_by(col(Publisher.id))
+        .order_by(count_expr.desc(), best_expr.desc(), col(Publisher.name))
+        .limit(limit)
+    ).all()
+    if not top:
+        return []
+
+    articles = session.exec(
+        select(Article)
+        .join(Publisher, col(Publisher.id) == col(Article.publisher_id))
+        .where(*conditions, col(Article.publisher_id).in_([t[0] for t in top]))
+        .order_by(col(Article.published_at), col(Article.id))
+    ).all()
+    by_publisher: dict[int | None, list[Article]] = {}
+    for a in articles:
+        by_publisher.setdefault(a.publisher_id, []).append(a)
+
+    result = []
+    for publisher_id, slug, name, count in top:
+        rows = by_publisher[publisher_id]
+        # Ties go to the newest: rows are oldest first and max keeps the first.
+        best = max(reversed(rows), key=lambda a: a.score)
+        result.append(
+            Voice(
+                slug=slug,
+                name=name,
+                count=count,
+                scores=[a.score for a in rows[-VOICE_SCORES:]],
+                best=VoiceArticle(
+                    title=best.title,
+                    url=best.url,
+                    score=best.score,
+                    published_at=best.published_at,
+                ),
+            )
+        )
+    return result

backend/app/catalog/models.py

@@ -353,6 +353,33 @@ class OriginCounts(SQLModel):
     found_via: list[FoundViaCount]
 
 
+# ─── Voices (the left rail's "Voices") ─────────────────────────────────────
+
+
+class VoiceKind(StrEnum):
+    """The Voices panel's tabs: Writers, Labs, Companies."""
+
+    individual = "individual"
+    lab = "lab"
+    company = "company"
+
+
+class VoiceArticle(SQLModel):
+    title: str
+    url: str
+    score: int
+    published_at: datetime | None = None
+
+
+class Voice(SQLModel):
+    slug: str
+    name: str
+    count: int
+    # Oldest first, the last `facets.VOICE_SCORES` of `count`.
+    scores: list[int]
+    best: VoiceArticle
+
+
 # ─── Stories (right rail) ──────────────────────────────────────────────────
 
 

backend/app/catalog/routes.py

@@ -16,6 +16,8 @@
     PublisherFacet,
     StoryPublic,
     TagFacet,
+    Voice,
+    VoiceKind,
 )
 from app.deps import CurrentUserOptional, SessionDep
 
@@ -93,6 +95,20 @@ def article_origins(session: SessionDep, since: str) -> OriginCounts:
     return facets.origin_counts(session, since_dt)
 
 
+@router.get("/voices", response_model=list[Voice])
+def article_voices(
+    session: SessionDep,
+    kind: VoiceKind,
+    since: str,
+    limit: int = Query(default=8, ge=1, le=20),
+) -> list[Voice]:
+    try:
+        since_dt = datetime.fromisoformat(since)
+    except ValueError:
+        raise HTTPException(status_code=422, detail="Invalid 'since' datetime")
+    return facets.voices(session, kind, since_dt, limit)
+
+
 @router.get("/publishers", response_model=list[PublisherFacet])
 def search_publishers(
     session: SessionDep,

backend/app/catalog/tests/test_routes.py

@@ -6,6 +6,7 @@
 from collections.abc import Generator
 from dataclasses import dataclass
 from datetime import UTC, datetime, timedelta
+from typing import Any
 
 import pytest
 from fastapi.testclient import TestClient
@@ -66,6 +67,7 @@ def auth_client(
     "/tags",
     "/stats",
     "/origins?since=2026-01-01",
+    "/voices?kind=individual&since=2026-01-01",
 ]
 
 
@@ -617,3 +619,120 @@ def test_origin_counts_group_found_via(client: TestClient, db: Session) -> None:
 def test_origin_counts_malformed_since_returns_422(client: TestClient) -> None:
     r = client.get(f"{ARTICLES_URL}/origins", params={"since": "not-a-date"})
     assert r.status_code == 422
+
+
+# --- voices ------------------------------------------------------------------
+
+VOICES_URL = f"{ARTICLES_URL}/voices"
+
+
+def _voices(
+    client: TestClient, kind: str, slugs: set[str], since: str = RECENT_SINCE
+) -> list[dict[str, Any]]:
+    """The voices of `kind`, narrowed to the publishers a test made: other
+    tests' articles share the window."""
+    r = client.get(VOICES_URL, params={"kind": kind, "since": since, "limit": 20})
+    assert r.status_code == 200
+    return [v for v in r.json() if v["slug"] in slugs]
+
+
+def _ago(hours: float) -> datetime:
+    return datetime.now(UTC) - timedelta(hours=hours)
+
+
+def test_voices_filter_by_kind(client: TestClient, db: Session) -> None:
+    lab = create_random_publisher(db, kind=PublisherKind.lab)
+    company = create_random_publisher(db, kind=PublisherKind.company)
+    for p in (lab, company):
+        create_random_article(db, publisher_id=p.id, published_at=_ago(1))
+    slugs = {lab.slug, company.slug}
+
+    assert [v["slug"] for v in _voices(client, "lab", slugs)] == [lab.slug]
+    assert [v["slug"] for v in _voices(client, "company", slugs)] == [company.slug]
+    assert _voices(client, "individual", slugs) == []
+
+
+def test_voices_writers_count_posts_only(client: TestClient, db: Session) -> None:
+    writer = create_random_publisher(db, kind=PublisherKind.individual)
+    owner = create_random_publisher(db, kind=PublisherKind.individual)
+    create_random_article(db, publisher_id=writer.id, published_at=_ago(1))
+    create_random_article(
+        db, publisher_id=writer.id, kind=ArticleKind.repo, published_at=_ago(1)
+    )
+    create_random_article(
+        db, publisher_id=owner.id, kind=ArticleKind.model, published_at=_ago(1)
+    )
+
+    voices = _voices(client, "individual", {writer.slug, owner.slug})
+    assert [(v["slug"], v["count"]) for v in voices] == [(writer.slug, 1)]
+
+
+def test_voices_labs_count_every_kind(client: TestClient, db: Session) -> None:
+    lab = create_random_publisher(db, kind=PublisherKind.lab)
+    for kind in (ArticleKind.post, ArticleKind.model, ArticleKind.paper):
+        create_random_article(db, publisher_id=lab.id, kind=kind, published_at=_ago(1))
+    assert _voices(client, "lab", {lab.slug})[0]["count"] == 3
+
+
+def test_voices_are_bounded_by_since(client: TestClient, db: Session) -> None:
+    lab = create_random_publisher(db, kind=PublisherKind.lab)
+    create_random_article(db, publisher_id=lab.id, published_at=_ago(1))
+    create_random_article(db, publisher_id=lab.id, published_at=_ago(24 * 10))
+    assert _voices(client, "lab", {lab.slug})[0]["count"] == 1
+
+
+def test_voices_scores_oldest_first_capped(client: TestClient, db: Session) -> None:
+    lab = create_random_publisher(db, kind=PublisherKind.lab)
+    # 17 articles, oldest first; scores cycle 1..10 so order is visible.
+    scores = [i % 10 + 1 for i in range(17)]
+    for i, score in enumerate(scores):
+        create_random_article(
+            db, publisher_id=lab.id, score=score, published_at=_ago(20 - i)
+        )
+    voice = _voices(client, "lab", {lab.slug})[0]
+    assert voice["count"] == 17
+    assert voice["scores"] == scores[-15:]
+
+
+def test_voices_best_is_highest_score(client: TestClient, db: Session) -> None:
+    lab = create_random_publisher(db, kind=PublisherKind.lab)
+    create_random_article(db, publisher_id=lab.id, score=4, published_at=_ago(3))
+    best = create_random_article(db, publisher_id=lab.id, score=9, published_at=_ago(2))
+    create_random_article(db, publisher_id=lab.id, score=6, published_at=_ago(1))
+
+    voice = _voices(client, "lab", {lab.slug})[0]
+    assert voice["best"]["title"] == best.title
+    assert voice["best"]["url"] == best.url
+    assert voice["best"]["score"] == 9
+
+
+def test_voices_order_by_count_then_best(client: TestClient, db: Session) -> None:
+    busy = create_random_publisher(db, kind=PublisherKind.company)
+    strong = create_random_publisher(db, kind=PublisherKind.company)
+    weak = create_random_publisher(db, kind=PublisherKind.company)
+    for score in (2, 3, 2):
+        create_random_article(
+            db, publisher_id=busy.id, score=score, published_at=_ago(1)
+        )
+    for score in (9, 1):
+        create_random_article(
+            db, publisher_id=strong.id, score=score, published_at=_ago(1)
+        )
+    for score in (5, 5):
+        create_random_article(
+            db, publisher_id=weak.id, score=score, published_at=_ago(1)
+        )
+
+    voices = _voices(client, "company", {busy.slug, strong.slug, weak.slug})
+    assert [v["slug"] for v in voices] == [busy.slug, strong.slug, weak.slug]
+
+
+def test_voices_unknown_kind_returns_422(client: TestClient) -> None:
+    for kind in ("media", "nope"):
+        r = client.get(VOICES_URL, params={"kind": kind, "since": RECENT_SINCE})
+        assert r.status_code == 422
+
+
+def test_voices_malformed_since_returns_422(client: TestClient) -> None:
+    r = client.get(VOICES_URL, params={"kind": "lab", "since": "not-a-date"})
+    assert r.status_code == 422

docs/product.md

@@ -17,6 +17,7 @@ Companies, Writers, Media, Repos, Papers, Models), by topic (a hand-curated set
 tags and kinds, e.g. "Open weights" or "Harnesses"), and by time window
 (last 3 days / week / month).
 A left rail beside the feed shows how that window splits by origin and how its articles were found (publisher feeds, Hacker News, newsletters); a row sets the From filter.
+A Voices panel there lists the writers, labs and companies with the most articles in the last 30 days, each with a bar per article and its best one; a name filters the feed to that publisher.
 
 **A topic map beside the feed.** A panel in the left rail draws the week's articles
 (the month's, when the week is thin) as dots grouped by meaning into labelled