feat(modelwatch): push producer for open-weight model releases
Add newsfeed-modelwatch: a one-shot, systemd-timer-driven producer that watches
the Hugging Face Hub for open-weight releases and pushes filtered candidates to
the ingest endpoint. Answers "how do I automate the HF firehose into my feed" —
the RSS half already works on the pull rail; this is the JSON/API half.
How it works: each run polls the HF models API (a watchlist of orgs +
trendingScore), applies an admission predicate (license allowlist, total-params
cap, safetensors present, not gated, pipeline_tag), and POSTs the survivors to
POST /v1/ingest/candidates under a bearer token, tagged source="huggingface".
Two properties of the existing ingest rail shape the design:
- Ingest is idempotent on (user, external_id). Using the repo id as external_id
makes the producer STATELESS — no dedup table; re-runs/overlapping timers just
re-submit and the server drops repeats.
- HF `tags` are copied onto the candidate, so the user's per-interest weights do
the ranking. The predicate is only an ADMISSION filter (what's worth surfacing
at all) and stays subordinate to explicit weights, per the house rule.
Layout: newsfeed-modelwatch is a client of the API, not an internal component —
it holds no core/data deps. Pure logic (HF types, predicate, mapping to
CandidateSubmission, param formatting) lives in the lib with unit tests; the bin
does the HTTP polling/posting. CandidateSubmission gains Serialize so the
in-workspace producer can build and post one.
Ops:
- asset/systemd/newsfeed-modelwatch.{service,timer}: oneshot + 3-hourly timer.
The ingest token is a secret, kept in /etc/newsfeed/modelwatch.env
(NEWSFEED_TOKEN, mapped to `token` by figment) so a redeploy of the config
never clobbers it.
- asset/config/modelwatch.toml.tmpl: watchlist + predicate, no secret.
- infra-setup.sh installs the units, creates the env placeholder (never
overwritten), enables the timer, and prints the token step.
- deploy.yml builds + ships the binary and the (secret-free) config each deploy.
Verified: unit tests (gated union, license precedence, admit accept/reject,
mapping); dry-run against live HF; full loop against a local API — minted token,
submitted a real release, confirmed it landed in the feed with the huggingface
source linked and HF tags carried through. fmt/clippy/test all green.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_016fKZzDpvjiJ9eYbPGgJvUP
This commit is contained in:
@@ -44,9 +44,10 @@ jobs:
|
||||
- run: cargo test --workspace
|
||||
- name: build binaries
|
||||
run: |
|
||||
cargo build --release -p newsfeed-api -p newsfeed-worker
|
||||
cargo build --release -p newsfeed-api -p newsfeed-worker -p newsfeed-modelwatch
|
||||
mkdir -p out
|
||||
cp target/release/newsfeed-api target/release/newsfeed-worker out/
|
||||
cp target/release/newsfeed-api target/release/newsfeed-worker \
|
||||
target/release/newsfeed-modelwatch out/
|
||||
- uses: actions/upload-artifact@v3
|
||||
with:
|
||||
name: binaries
|
||||
@@ -88,7 +89,7 @@ jobs:
|
||||
- name: rsync binaries
|
||||
run: |
|
||||
chmod +x bin/*
|
||||
for b in newsfeed-api newsfeed-worker; do
|
||||
for b in newsfeed-api newsfeed-worker newsfeed-modelwatch; do
|
||||
rsync -az -e "ssh $SSH_OPTS -i ~/.ssh/id_deploy" \
|
||||
--rsync-path='sudo rsync' --chown=root:root --chmod=0755 --mkpath \
|
||||
"bin/$b" gitea_ci@"$API_HOST":/usr/local/bin/"$b"
|
||||
@@ -102,11 +103,16 @@ jobs:
|
||||
rsync -az -e "ssh $SSH_OPTS -i ~/.ssh/id_deploy" \
|
||||
--rsync-path='sudo rsync' --chown=root:newsfeed --chmod=0640 --mkpath \
|
||||
asset/config/worker.toml.tmpl gitea_ci@"$API_HOST":/etc/newsfeed/worker.toml
|
||||
# modelwatch config carries no secret (the token lives in modelwatch.env), so it
|
||||
# is safe to overwrite on every deploy.
|
||||
rsync -az -e "ssh $SSH_OPTS -i ~/.ssh/id_deploy" \
|
||||
--rsync-path='sudo rsync' --chown=root:newsfeed --chmod=0640 --mkpath \
|
||||
asset/config/modelwatch.toml.tmpl gitea_ci@"$API_HOST":/etc/newsfeed/modelwatch.toml
|
||||
|
||||
- name: apply system state
|
||||
run: |
|
||||
ssh $SSH_OPTS -i ~/.ssh/id_deploy gitea_ci@"$API_HOST" '
|
||||
sudo restorecon -R /usr/local/bin/newsfeed-api /usr/local/bin/newsfeed-worker /etc/newsfeed /var/lib/newsfeed
|
||||
sudo restorecon -R /usr/local/bin/newsfeed-api /usr/local/bin/newsfeed-worker /usr/local/bin/newsfeed-modelwatch /etc/newsfeed /var/lib/newsfeed
|
||||
sudo systemctl daemon-reload
|
||||
sudo systemctl restart newsfeed-api.service
|
||||
sudo systemctl restart newsfeed-worker.service
|
||||
|
||||
16
CLAUDE.md
16
CLAUDE.md
@@ -19,11 +19,19 @@ per-source and per-interest weights the user sets.
|
||||
- `newsfeed-core` — business logic + data-access **ports** (traits). Pure where possible
|
||||
(ranking, auth hashing). No DB or network calls.
|
||||
- `newsfeed-data` — SQLite **adapters** implementing the core ports.
|
||||
- `newsfeed-fetch` — the outbound-feed adapter (RSS fetch, feed discovery, OPML parse);
|
||||
implements the `FeedProbe` port. The one crate that talks HTTP to the wider web for
|
||||
feeds — used by the api (discovery on add) and the worker (polling). Kept separate from
|
||||
`data` (SQLite-only) on purpose.
|
||||
- `newsfeed-api` / `newsfeed-worker` — thin binaries; wire config/logging/signals and the
|
||||
concrete store. No business logic that could live in a library crate.
|
||||
- `newsfeed-modelwatch` — a **push producer** (lib + one-shot bin): watches the Hugging
|
||||
Face Hub for open-weight releases, applies an admission predicate, and POSTs candidates
|
||||
to `/v1/ingest/candidates` under a token. A *client* of the API, not an internal
|
||||
component — it holds no DB/core deps; its pure logic (predicate, mapping) is in the lib.
|
||||
|
||||
New types → entities. New logic → core. New I/O → data. Add a port to
|
||||
`core::ports`, implement it in `data::store`.
|
||||
New types → entities. New logic → core. New feed I/O → `fetch`. New DB I/O → `data`. Add a
|
||||
port to `core::ports`, implement it in the matching adapter crate.
|
||||
|
||||
## Deliberate deviations from `generic.md`
|
||||
|
||||
@@ -55,7 +63,9 @@ Crypto/token logic lives in `core::auth`; the flow orchestration in `core::servi
|
||||
|
||||
## Deploy topology
|
||||
|
||||
api + worker → `slartibartfast.kosherinata.internal`; SPA → `oolon.kosherinata.internal`
|
||||
api + worker (+ the `newsfeed-modelwatch` systemd timer, 3-hourly, whose ingest token lives
|
||||
in `/etc/newsfeed/modelwatch.env` — a secret kept out of the deployable config so redeploys
|
||||
never clobber it) → `slartibartfast.kosherinata.internal`; SPA → `oolon.kosherinata.internal`
|
||||
(nginx serves + proxies `/v1` to the API over the mesh; TLS terminates at oolon). API
|
||||
port 22672 (registered in architecture `port-allocations.md`; deliberately not the crowded
|
||||
`8081`), plain HTTP behind firewalld. See `.gitea/workflows/deploy.yml` (infra truth)
|
||||
|
||||
17
Cargo.lock
generated
17
Cargo.lock
generated
@@ -1371,6 +1371,23 @@ dependencies = [
|
||||
"url",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "newsfeed-modelwatch"
|
||||
version = "0.1.0"
|
||||
dependencies = [
|
||||
"anyhow",
|
||||
"chrono",
|
||||
"clap",
|
||||
"figment",
|
||||
"newsfeed-entities",
|
||||
"reqwest",
|
||||
"serde",
|
||||
"serde_json",
|
||||
"tokio",
|
||||
"tracing",
|
||||
"tracing-subscriber",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "newsfeed-worker"
|
||||
version = "0.1.0"
|
||||
|
||||
35
asset/config/modelwatch.toml.tmpl
Normal file
35
asset/config/modelwatch.toml.tmpl
Normal file
@@ -0,0 +1,35 @@
|
||||
# newsfeed-modelwatch configuration (production). Rendered to /etc/newsfeed/modelwatch.toml
|
||||
# by the deploy workflow.
|
||||
#
|
||||
# The ingest token is NOT here — it is a secret, supplied via NEWSFEED_TOKEN in
|
||||
# /etc/newsfeed/modelwatch.env (created once by infra-setup.sh). That keeps redeploys of
|
||||
# this file from ever clobbering the token.
|
||||
|
||||
ingest_url = "http://127.0.0.1:22672/v1/ingest/candidates"
|
||||
source_name = "huggingface"
|
||||
|
||||
# HF orgs to watch for new open-weight repos.
|
||||
authors = [
|
||||
"tencent", "Qwen", "deepseek-ai", "mistralai", "zai-org", "moonshotai",
|
||||
"openbmb", "allenai", "nvidia", "ibm-granite", "google", "meta-llama", "microsoft",
|
||||
]
|
||||
|
||||
# The trending list is the best built-in noise filter; a notable release surfaces fast.
|
||||
include_trending = true
|
||||
trending_limit = 25
|
||||
per_author_limit = 15
|
||||
# Ignore repos older than this so the first run doesn't backfill old "newest" repos.
|
||||
max_age_days = 30
|
||||
|
||||
http_timeout_secs = 20
|
||||
user_agent = "newsfeed-modelwatch (+https://git.lair.cafe/grenade/newsfeed)"
|
||||
|
||||
# Admission filter: what's worth putting in front of you at all. Ranking is NOT decided
|
||||
# here — the HF tags ride onto each candidate so your per-interest weights do the ranking.
|
||||
[predicate]
|
||||
licenses = ["apache-2.0", "mit"]
|
||||
# Total safetensors params. NB: this is total, not MoE-active (that needs config.json math).
|
||||
max_params = 40000000000
|
||||
require_safetensors = true
|
||||
skip_gated = true
|
||||
pipeline_tags = ["text-generation"]
|
||||
32
asset/systemd/newsfeed-modelwatch.service
Normal file
32
asset/systemd/newsfeed-modelwatch.service
Normal file
@@ -0,0 +1,32 @@
|
||||
[Unit]
|
||||
Description=newsfeed model-release watcher (Hugging Face Hub -> ingest)
|
||||
Documentation=https://git.lair.cafe/grenade/newsfeed
|
||||
After=network-online.target newsfeed-api.service
|
||||
Wants=network-online.target
|
||||
# Timer-activated (see newsfeed-modelwatch.timer); no [Install] here.
|
||||
|
||||
[Service]
|
||||
Type=oneshot
|
||||
User=newsfeed
|
||||
Group=newsfeed
|
||||
# The ingest token (NEWSFEED_TOKEN) lives here, kept out of the deployable config so a
|
||||
# redeploy of modelwatch.toml never clobbers it.
|
||||
EnvironmentFile=/etc/newsfeed/modelwatch.env
|
||||
ExecStart=/usr/local/bin/newsfeed-modelwatch --config /etc/newsfeed/modelwatch.toml
|
||||
|
||||
# Hardening (architecture generic.md §8). No DB access — this talks HTTP only, so no
|
||||
# ReadWritePaths are granted.
|
||||
NoNewPrivileges=true
|
||||
ProtectSystem=strict
|
||||
ProtectHome=true
|
||||
PrivateTmp=true
|
||||
PrivateDevices=true
|
||||
ProtectKernelTunables=true
|
||||
ProtectKernelModules=true
|
||||
ProtectControlGroups=true
|
||||
RestrictRealtime=true
|
||||
RestrictSUIDSGID=true
|
||||
LockPersonality=true
|
||||
MemoryDenyWriteExecute=true
|
||||
SystemCallArchitectures=native
|
||||
RestrictAddressFamilies=AF_UNIX AF_INET AF_INET6
|
||||
13
asset/systemd/newsfeed-modelwatch.timer
Normal file
13
asset/systemd/newsfeed-modelwatch.timer
Normal file
@@ -0,0 +1,13 @@
|
||||
[Unit]
|
||||
Description=Run newsfeed-modelwatch periodically (open-weight release sweep)
|
||||
Documentation=https://git.lair.cafe/grenade/newsfeed
|
||||
|
||||
[Timer]
|
||||
# Every 3 hours, jittered, so releases through the day are picked up without hammering HF.
|
||||
OnCalendar=*-*-* 00/3:00:00
|
||||
RandomizedDelaySec=600
|
||||
# Fire a missed run once the host is back up.
|
||||
Persistent=true
|
||||
|
||||
[Install]
|
||||
WantedBy=timers.target
|
||||
@@ -68,8 +68,9 @@ pub struct ContentItem {
|
||||
|
||||
/// A candidate pushed to `POST /v1/ingest/candidates` by an agentic or algorithmic
|
||||
/// source. The authenticated API token determines the owning user; `source` names the
|
||||
/// producer for attribution and per-source weighting.
|
||||
#[derive(Debug, Clone, Deserialize, TS)]
|
||||
/// producer for attribution and per-source weighting. Derives `Serialize` too so
|
||||
/// in-workspace producers (e.g. `newsfeed-modelwatch`) can build and post one.
|
||||
#[derive(Debug, Clone, Serialize, Deserialize, TS)]
|
||||
#[ts(export)]
|
||||
pub struct CandidateSubmission {
|
||||
/// Stable de-dup key from the producer. Re-submitting the same key is a no-op.
|
||||
|
||||
32
crates/newsfeed-modelwatch/Cargo.toml
Normal file
32
crates/newsfeed-modelwatch/Cargo.toml
Normal file
@@ -0,0 +1,32 @@
|
||||
[package]
|
||||
name = "newsfeed-modelwatch"
|
||||
version.workspace = true
|
||||
edition.workspace = true
|
||||
rust-version.workspace = true
|
||||
license.workspace = true
|
||||
authors.workspace = true
|
||||
description = "Push producer: watches Hugging Face Hub for open-weight model releases and submits filtered candidates to the newsfeed ingest endpoint."
|
||||
|
||||
[lib]
|
||||
name = "newsfeed_modelwatch"
|
||||
path = "src/lib.rs"
|
||||
|
||||
[[bin]]
|
||||
name = "newsfeed-modelwatch"
|
||||
path = "src/main.rs"
|
||||
|
||||
[dependencies]
|
||||
newsfeed-entities.workspace = true
|
||||
|
||||
tokio.workspace = true
|
||||
reqwest.workspace = true
|
||||
|
||||
serde.workspace = true
|
||||
serde_json.workspace = true
|
||||
chrono.workspace = true
|
||||
|
||||
anyhow.workspace = true
|
||||
tracing.workspace = true
|
||||
tracing-subscriber.workspace = true
|
||||
figment.workspace = true
|
||||
clap.workspace = true
|
||||
93
crates/newsfeed-modelwatch/src/config.rs
Normal file
93
crates/newsfeed-modelwatch/src/config.rs
Normal file
@@ -0,0 +1,93 @@
|
||||
//! Layered config: built-in defaults → optional TOML file → `NEWSFEED_*` env, matching
|
||||
//! the api/worker convention.
|
||||
|
||||
use std::path::Path;
|
||||
|
||||
use figment::Figment;
|
||||
use figment::providers::{Env, Format, Serialized, Toml};
|
||||
use serde::{Deserialize, Serialize};
|
||||
|
||||
use newsfeed_modelwatch::Predicate;
|
||||
|
||||
/// modelwatch producer configuration.
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub struct Config {
|
||||
/// Full URL of the newsfeed ingest endpoint.
|
||||
pub ingest_url: String,
|
||||
/// Bearer token minted in the newsfeed UI (owns the ingested items). Required unless
|
||||
/// `--dry-run`.
|
||||
pub token: String,
|
||||
/// Source name attached to every candidate; link an agentic source with this name to
|
||||
/// give the whole firehose one weight you control.
|
||||
pub source_name: String,
|
||||
/// HF orgs to watch for new repos.
|
||||
pub authors: Vec<String>,
|
||||
/// Also poll the global trending list (best built-in noise filter).
|
||||
pub include_trending: bool,
|
||||
/// How many trending models to consider.
|
||||
pub trending_limit: u32,
|
||||
/// How many newest repos to consider per org.
|
||||
pub per_author_limit: u32,
|
||||
/// Ignore repos created more than this many days ago (keeps the first run from
|
||||
/// backfilling old "newest" repos). 0 = no age limit.
|
||||
pub max_age_days: i64,
|
||||
/// HF Hub models API base (overridable for a mirror or a test server).
|
||||
pub hub_api: String,
|
||||
/// Outbound HTTP timeout.
|
||||
pub http_timeout_secs: u64,
|
||||
pub user_agent: String,
|
||||
/// Log what would be submitted without posting.
|
||||
pub dry_run: bool,
|
||||
/// The admission predicate.
|
||||
#[serde(default)]
|
||||
pub predicate: Predicate,
|
||||
}
|
||||
|
||||
impl Default for Config {
|
||||
fn default() -> Self {
|
||||
Self {
|
||||
ingest_url: "http://127.0.0.1:22672/v1/ingest/candidates".to_string(),
|
||||
token: String::new(),
|
||||
source_name: "huggingface".to_string(),
|
||||
authors: [
|
||||
"tencent",
|
||||
"Qwen",
|
||||
"deepseek-ai",
|
||||
"mistralai",
|
||||
"zai-org",
|
||||
"moonshotai",
|
||||
"openbmb",
|
||||
"allenai",
|
||||
"nvidia",
|
||||
"ibm-granite",
|
||||
"google",
|
||||
"meta-llama",
|
||||
"microsoft",
|
||||
]
|
||||
.iter()
|
||||
.map(|s| s.to_string())
|
||||
.collect(),
|
||||
include_trending: true,
|
||||
trending_limit: 25,
|
||||
per_author_limit: 15,
|
||||
max_age_days: 30,
|
||||
hub_api: "https://huggingface.co/api/models".to_string(),
|
||||
http_timeout_secs: 20,
|
||||
user_agent: concat!("newsfeed-modelwatch/", env!("CARGO_PKG_VERSION")).to_string(),
|
||||
dry_run: false,
|
||||
predicate: Predicate::default(),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl Config {
|
||||
/// Load configuration, layering an optional TOML file and `NEWSFEED_`-prefixed env over
|
||||
/// the defaults.
|
||||
pub fn load(file: Option<&Path>) -> anyhow::Result<Self> {
|
||||
let mut fig = Figment::from(Serialized::defaults(Config::default()));
|
||||
if let Some(path) = file {
|
||||
fig = fig.merge(Toml::file(path));
|
||||
}
|
||||
Ok(fig.merge(Env::prefixed("NEWSFEED_")).extract()?)
|
||||
}
|
||||
}
|
||||
351
crates/newsfeed-modelwatch/src/lib.rs
Normal file
351
crates/newsfeed-modelwatch/src/lib.rs
Normal file
@@ -0,0 +1,351 @@
|
||||
//! Pure logic for the Hugging Face Hub release watcher: the API response shapes, the
|
||||
//! admission predicate, and the mapping to a [`CandidateSubmission`]. No I/O — the binary
|
||||
//! ([`main`](../main.rs)) does the HTTP polling and posting; everything here is testable
|
||||
//! without a network.
|
||||
//!
|
||||
//! Design note: the predicate is an **admission filter** (is this release worth putting in
|
||||
//! front of the user at all?), not a ranking. Ranking stays with the user's per-source and
|
||||
//! per-interest weights — the HF `tags` are copied onto the candidate precisely so those
|
||||
//! weights bite. See the newsfeed CLAUDE.md house rule.
|
||||
|
||||
use std::collections::HashMap;
|
||||
|
||||
use chrono::{DateTime, Utc};
|
||||
use serde::{Deserialize, Serialize};
|
||||
|
||||
use newsfeed_entities::item::CandidateSubmission;
|
||||
|
||||
/// A model record from the HF Hub API. The list and detail endpoints share this shape;
|
||||
/// detail-only fields (`safetensors`, `card_data`) are optional so one struct serves both.
|
||||
/// Unknown fields are ignored, so the API can grow without breaking us.
|
||||
#[derive(Debug, Clone, Deserialize)]
|
||||
pub struct HfModel {
|
||||
/// `"org/name"`.
|
||||
pub id: String,
|
||||
#[serde(default)]
|
||||
pub author: Option<String>,
|
||||
#[serde(rename = "createdAt", default)]
|
||||
pub created_at: Option<DateTime<Utc>>,
|
||||
#[serde(default)]
|
||||
pub downloads: Option<u64>,
|
||||
#[serde(default)]
|
||||
pub likes: Option<u64>,
|
||||
#[serde(rename = "trendingScore", default)]
|
||||
pub trending_score: Option<f64>,
|
||||
#[serde(rename = "pipeline_tag", default)]
|
||||
pub pipeline_tag: Option<String>,
|
||||
#[serde(default)]
|
||||
pub gated: Gated,
|
||||
#[serde(default)]
|
||||
pub tags: Vec<String>,
|
||||
#[serde(default)]
|
||||
pub safetensors: Option<Safetensors>,
|
||||
#[serde(rename = "cardData", default)]
|
||||
pub card_data: Option<CardData>,
|
||||
}
|
||||
|
||||
/// HF's `gated` is a union: `false`, or a string mode (`"auto"`/`"manual"`).
|
||||
#[derive(Debug, Clone, PartialEq, Eq, Deserialize)]
|
||||
#[serde(untagged)]
|
||||
pub enum Gated {
|
||||
Flag(bool),
|
||||
Mode(String),
|
||||
}
|
||||
|
||||
impl Default for Gated {
|
||||
fn default() -> Self {
|
||||
Gated::Flag(false)
|
||||
}
|
||||
}
|
||||
|
||||
impl Gated {
|
||||
pub fn is_gated(&self) -> bool {
|
||||
match self {
|
||||
Gated::Flag(b) => *b,
|
||||
Gated::Mode(_) => true,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// The `safetensors` block: total parameter count plus a per-dtype breakdown.
|
||||
#[derive(Debug, Clone, Deserialize)]
|
||||
pub struct Safetensors {
|
||||
#[serde(default)]
|
||||
pub total: u64,
|
||||
#[serde(default)]
|
||||
pub parameters: HashMap<String, u64>,
|
||||
}
|
||||
|
||||
/// The bits of `cardData` we care about.
|
||||
#[derive(Debug, Clone, Default, Deserialize)]
|
||||
pub struct CardData {
|
||||
#[serde(default)]
|
||||
pub license: Option<String>,
|
||||
}
|
||||
|
||||
/// The license, lowercased — from `cardData.license`, else a `license:<x>` tag.
|
||||
pub fn license(m: &HfModel) -> Option<String> {
|
||||
if let Some(cd) = &m.card_data {
|
||||
if let Some(l) = &cd.license {
|
||||
return Some(l.to_lowercase());
|
||||
}
|
||||
}
|
||||
m.tags
|
||||
.iter()
|
||||
.find_map(|t| t.strip_prefix("license:").map(|s| s.to_lowercase()))
|
||||
}
|
||||
|
||||
/// The admission predicate: which releases are worth submitting. Configurable; an empty
|
||||
/// allowlist means "don't filter on this axis".
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub struct Predicate {
|
||||
/// Allowed licenses, lowercased. Empty = any license.
|
||||
#[serde(default = "default_licenses")]
|
||||
pub licenses: Vec<String>,
|
||||
/// Reject models whose total safetensors parameter count exceeds this. 0 = no cap.
|
||||
/// Note: this is *total* params; MoE active-param filtering needs config.json math and
|
||||
/// is a future refinement.
|
||||
#[serde(default = "default_max_params")]
|
||||
pub max_params: u64,
|
||||
/// Require a `safetensors` block (i.e. real, inspectable weights present).
|
||||
#[serde(default = "default_true")]
|
||||
pub require_safetensors: bool,
|
||||
/// Skip gated repos (access-request walls).
|
||||
#[serde(default = "default_true")]
|
||||
pub skip_gated: bool,
|
||||
/// Allowed `pipeline_tag`s. Empty = any.
|
||||
#[serde(default = "default_pipeline_tags")]
|
||||
pub pipeline_tags: Vec<String>,
|
||||
}
|
||||
|
||||
fn default_licenses() -> Vec<String> {
|
||||
vec!["apache-2.0".into(), "mit".into()]
|
||||
}
|
||||
fn default_max_params() -> u64 {
|
||||
40_000_000_000
|
||||
}
|
||||
fn default_true() -> bool {
|
||||
true
|
||||
}
|
||||
fn default_pipeline_tags() -> Vec<String> {
|
||||
vec!["text-generation".into()]
|
||||
}
|
||||
|
||||
impl Default for Predicate {
|
||||
fn default() -> Self {
|
||||
Self {
|
||||
licenses: default_licenses(),
|
||||
max_params: default_max_params(),
|
||||
require_safetensors: default_true(),
|
||||
skip_gated: default_true(),
|
||||
pipeline_tags: default_pipeline_tags(),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl Predicate {
|
||||
/// Decide whether to submit `m`. `Ok(())` admits; `Err(reason)` rejects (reason is for
|
||||
/// logging).
|
||||
pub fn admit(&self, m: &HfModel) -> Result<(), &'static str> {
|
||||
if self.skip_gated && m.gated.is_gated() {
|
||||
return Err("gated");
|
||||
}
|
||||
if !self.pipeline_tags.is_empty()
|
||||
&& !m
|
||||
.pipeline_tag
|
||||
.as_ref()
|
||||
.is_some_and(|p| self.pipeline_tags.iter().any(|a| a == p))
|
||||
{
|
||||
return Err("pipeline_tag not allowed");
|
||||
}
|
||||
match &m.safetensors {
|
||||
None if self.require_safetensors => return Err("no safetensors"),
|
||||
Some(st) if self.max_params > 0 && st.total > self.max_params => {
|
||||
return Err("too large");
|
||||
}
|
||||
_ => {}
|
||||
}
|
||||
if !self.licenses.is_empty()
|
||||
&& !license(m).is_some_and(|l| self.licenses.iter().any(|a| a == &l))
|
||||
{
|
||||
return Err("license not allowed");
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
/// Cap on how many HF tags to copy onto a candidate, keeping items tidy.
|
||||
const MAX_TAGS: usize = 32;
|
||||
|
||||
/// Map an admitted model to a [`CandidateSubmission`]. `external_id` is the repo id so the
|
||||
/// idempotent ingest endpoint drops repeats — the producer needs no local dedup state.
|
||||
pub fn to_submission(m: &HfModel, source: &str) -> CandidateSubmission {
|
||||
let author = m
|
||||
.author
|
||||
.clone()
|
||||
.or_else(|| m.id.split('/').next().map(str::to_string));
|
||||
|
||||
let mut parts = Vec::new();
|
||||
if let Some(total) = m.safetensors.as_ref().map(|s| s.total).filter(|&t| t > 0) {
|
||||
parts.push(human_params(total));
|
||||
}
|
||||
if let Some(l) = license(m) {
|
||||
parts.push(l);
|
||||
}
|
||||
if let Some(pt) = &m.pipeline_tag {
|
||||
parts.push(pt.clone());
|
||||
}
|
||||
if let Some(d) = m.downloads {
|
||||
parts.push(format!("↓{d}"));
|
||||
}
|
||||
if let Some(k) = m.likes {
|
||||
parts.push(format!("♥{k}"));
|
||||
}
|
||||
let summary = (!parts.is_empty()).then(|| parts.join(" · "));
|
||||
|
||||
let tags = m.tags.iter().take(MAX_TAGS).cloned().collect();
|
||||
|
||||
CandidateSubmission {
|
||||
external_id: m.id.clone(),
|
||||
url: Some(format!("https://huggingface.co/{}", m.id)),
|
||||
title: m.id.clone(),
|
||||
summary,
|
||||
author,
|
||||
tags,
|
||||
media: Vec::new(),
|
||||
published_at: m.created_at,
|
||||
source: Some(source.to_string()),
|
||||
}
|
||||
}
|
||||
|
||||
/// Human-readable parameter count, e.g. `12.3B params` / `770M params`.
|
||||
pub fn human_params(total: u64) -> String {
|
||||
let b = total as f64 / 1e9;
|
||||
if b >= 1.0 {
|
||||
format!("{b:.1}B params")
|
||||
} else {
|
||||
format!("{:.0}M params", total as f64 / 1e6)
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
fn model(json: &str) -> HfModel {
|
||||
serde_json::from_str(json).expect("parse HfModel")
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn gated_union_parses_both_forms() {
|
||||
assert_eq!(
|
||||
model(r#"{"id":"a/b","gated":false}"#).gated,
|
||||
Gated::Flag(false)
|
||||
);
|
||||
assert!(model(r#"{"id":"a/b","gated":"manual"}"#).gated.is_gated());
|
||||
assert!(!model(r#"{"id":"a/b"}"#).gated.is_gated()); // default
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn license_from_card_then_tags() {
|
||||
let m = model(r#"{"id":"a/b","cardData":{"license":"Apache-2.0"}}"#);
|
||||
assert_eq!(license(&m).as_deref(), Some("apache-2.0"));
|
||||
let m = model(r#"{"id":"a/b","tags":["license:mit","text-generation"]}"#);
|
||||
assert_eq!(license(&m).as_deref(), Some("mit"));
|
||||
assert_eq!(license(&model(r#"{"id":"a/b"}"#)), None);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn admit_happy_path() {
|
||||
let m = model(
|
||||
r#"{"id":"deepseek-ai/x","pipeline_tag":"text-generation","gated":false,
|
||||
"cardData":{"license":"mit"},"safetensors":{"total":7000000000}}"#,
|
||||
);
|
||||
assert_eq!(Predicate::default().admit(&m), Ok(()));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn admit_rejects_gated_large_wrong_license_and_missing_weights() {
|
||||
let p = Predicate::default();
|
||||
// gated
|
||||
assert_eq!(
|
||||
p.admit(&model(
|
||||
r#"{"id":"a/b","pipeline_tag":"text-generation","gated":"auto",
|
||||
"cardData":{"license":"mit"},"safetensors":{"total":1}}"#
|
||||
)),
|
||||
Err("gated")
|
||||
);
|
||||
// too large
|
||||
assert_eq!(
|
||||
p.admit(&model(
|
||||
r#"{"id":"a/b","pipeline_tag":"text-generation","cardData":{"license":"mit"},
|
||||
"safetensors":{"total":70000000000}}"#
|
||||
)),
|
||||
Err("too large")
|
||||
);
|
||||
// wrong license
|
||||
assert_eq!(
|
||||
p.admit(&model(
|
||||
r#"{"id":"a/b","pipeline_tag":"text-generation",
|
||||
"cardData":{"license":"cc-by-nc-4.0"},"safetensors":{"total":1}}"#
|
||||
)),
|
||||
Err("license not allowed")
|
||||
);
|
||||
// no safetensors
|
||||
assert_eq!(
|
||||
p.admit(&model(
|
||||
r#"{"id":"a/b","pipeline_tag":"text-generation","cardData":{"license":"mit"}}"#
|
||||
)),
|
||||
Err("no safetensors")
|
||||
);
|
||||
// wrong pipeline
|
||||
assert_eq!(
|
||||
p.admit(&model(
|
||||
r#"{"id":"a/b","pipeline_tag":"image-classification",
|
||||
"cardData":{"license":"mit"},"safetensors":{"total":1}}"#
|
||||
)),
|
||||
Err("pipeline_tag not allowed")
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn empty_allowlists_disable_those_axes() {
|
||||
let p = Predicate {
|
||||
licenses: vec![],
|
||||
max_params: 0,
|
||||
require_safetensors: false,
|
||||
skip_gated: false,
|
||||
pipeline_tags: vec![],
|
||||
};
|
||||
assert_eq!(p.admit(&model(r#"{"id":"a/b"}"#)), Ok(()));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn submission_maps_id_url_tags_and_summary() {
|
||||
let m = model(
|
||||
r#"{"id":"qwen/Qwen9","author":"qwen","createdAt":"2026-01-02T03:04:05Z",
|
||||
"pipeline_tag":"text-generation","downloads":1234,"likes":56,
|
||||
"cardData":{"license":"apache-2.0"},
|
||||
"safetensors":{"total":14000000000},
|
||||
"tags":["text-generation","moe","license:apache-2.0"]}"#,
|
||||
);
|
||||
let s = to_submission(&m, "huggingface");
|
||||
assert_eq!(s.external_id, "qwen/Qwen9");
|
||||
assert_eq!(s.url.as_deref(), Some("https://huggingface.co/qwen/Qwen9"));
|
||||
assert_eq!(s.author.as_deref(), Some("qwen"));
|
||||
assert_eq!(s.source.as_deref(), Some("huggingface"));
|
||||
assert!(s.tags.contains(&"moe".to_string()));
|
||||
let summary = s.summary.unwrap();
|
||||
assert!(summary.contains("14.0B params"), "{summary}");
|
||||
assert!(
|
||||
summary.contains("apache-2.0") && summary.contains("↓1234"),
|
||||
"{summary}"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn human_params_scales() {
|
||||
assert_eq!(human_params(14_000_000_000), "14.0B params");
|
||||
assert_eq!(human_params(770_000_000), "770M params");
|
||||
}
|
||||
}
|
||||
217
crates/newsfeed-modelwatch/src/main.rs
Normal file
217
crates/newsfeed-modelwatch/src/main.rs
Normal file
@@ -0,0 +1,217 @@
|
||||
//! newsfeed-modelwatch — a one-shot push producer for open-weight model releases.
|
||||
//!
|
||||
//! Driven by a systemd timer. Each run polls the Hugging Face Hub API (a watchlist of orgs
|
||||
//! plus the trending list), applies an admission predicate (license / size / weights
|
||||
//! present / not gated), and posts the survivors to the newsfeed ingest endpoint under a
|
||||
//! bearer token. Ingest is idempotent on `external_id` (the repo id), so this process holds
|
||||
//! no state — re-runs and overlapping timers simply re-submit and the server drops repeats.
|
||||
|
||||
mod config;
|
||||
|
||||
use std::collections::HashSet;
|
||||
use std::path::PathBuf;
|
||||
use std::time::Duration;
|
||||
|
||||
use anyhow::{Context, bail};
|
||||
use clap::Parser;
|
||||
use reqwest::Client;
|
||||
|
||||
use newsfeed_entities::item::CandidateSubmission;
|
||||
use newsfeed_modelwatch::{HfModel, to_submission};
|
||||
|
||||
use crate::config::Config;
|
||||
|
||||
#[derive(Parser)]
|
||||
#[command(
|
||||
name = "newsfeed-modelwatch",
|
||||
version,
|
||||
about = "Watch Hugging Face for open-weight releases and push candidates to newsfeed"
|
||||
)]
|
||||
struct Cli {
|
||||
/// Path to a TOML config file. Env (`NEWSFEED_*`) overrides file values.
|
||||
#[arg(long, default_value = "/etc/newsfeed/modelwatch.toml")]
|
||||
config: PathBuf,
|
||||
/// Poll and log what would be submitted, but post nothing.
|
||||
#[arg(long)]
|
||||
dry_run: bool,
|
||||
}
|
||||
|
||||
#[tokio::main]
|
||||
async fn main() -> anyhow::Result<()> {
|
||||
init_tracing();
|
||||
|
||||
let cli = Cli::parse();
|
||||
let config_file = cli.config.exists().then_some(cli.config.as_path());
|
||||
let cfg = Config::load(config_file).context("loading configuration")?;
|
||||
let dry_run = cfg.dry_run || cli.dry_run;
|
||||
|
||||
if cfg.token.is_empty() && !dry_run {
|
||||
bail!(
|
||||
"no ingest token configured (set `token` in the config or NEWSFEED_TOKEN); or pass --dry-run"
|
||||
);
|
||||
}
|
||||
|
||||
let client = Client::builder()
|
||||
.user_agent(&cfg.user_agent)
|
||||
.timeout(Duration::from_secs(cfg.http_timeout_secs))
|
||||
.build()
|
||||
.context("building HTTP client")?;
|
||||
|
||||
tracing::info!(
|
||||
authors = cfg.authors.len(),
|
||||
trending = cfg.include_trending,
|
||||
dry_run,
|
||||
"starting modelwatch cycle"
|
||||
);
|
||||
|
||||
// 1. Collect candidate summaries from each org plus (optionally) the trending list.
|
||||
let mut summaries: Vec<HfModel> = Vec::new();
|
||||
for author in &cfg.authors {
|
||||
let url = format!(
|
||||
"{}?author={}&sort=createdAt&direction=-1&limit={}",
|
||||
cfg.hub_api, author, cfg.per_author_limit
|
||||
);
|
||||
match fetch_list(&client, &url).await {
|
||||
Ok(mut v) => summaries.append(&mut v),
|
||||
Err(e) => tracing::warn!(%author, error = %e, "listing org failed"),
|
||||
}
|
||||
}
|
||||
if cfg.include_trending {
|
||||
let url = format!(
|
||||
"{}?sort=trendingScore&direction=-1&limit={}&pipeline_tag=text-generation",
|
||||
cfg.hub_api, cfg.trending_limit
|
||||
);
|
||||
match fetch_list(&client, &url).await {
|
||||
Ok(mut v) => summaries.append(&mut v),
|
||||
Err(e) => tracing::warn!(error = %e, "listing trending failed"),
|
||||
}
|
||||
}
|
||||
|
||||
// 2. Dedup, cheaply pre-filter on summary fields, detail-fetch the rest, admit, submit.
|
||||
let mut seen = HashSet::new();
|
||||
let (mut scanned, mut admitted, mut submitted) = (0u32, 0u32, 0u32);
|
||||
let cutoff = (cfg.max_age_days > 0)
|
||||
.then(|| chrono::Utc::now() - chrono::Duration::days(cfg.max_age_days));
|
||||
|
||||
for summary in summaries {
|
||||
if !seen.insert(summary.id.clone()) {
|
||||
continue;
|
||||
}
|
||||
// Cheap rejects using fields already in the list response — avoids a detail fetch.
|
||||
if let (Some(cutoff), Some(created)) = (cutoff, summary.created_at) {
|
||||
if created < cutoff {
|
||||
continue;
|
||||
}
|
||||
}
|
||||
if cfg.predicate.skip_gated && summary.gated.is_gated() {
|
||||
continue;
|
||||
}
|
||||
if !cfg.predicate.pipeline_tags.is_empty()
|
||||
&& summary
|
||||
.pipeline_tag
|
||||
.as_ref()
|
||||
.is_some_and(|p| !cfg.predicate.pipeline_tags.iter().any(|a| a == p))
|
||||
{
|
||||
continue;
|
||||
}
|
||||
|
||||
scanned += 1;
|
||||
let detail = match fetch_detail(&client, &cfg.hub_api, &summary.id).await {
|
||||
Ok(d) => d,
|
||||
Err(e) => {
|
||||
tracing::warn!(id = %summary.id, error = %e, "detail fetch failed");
|
||||
continue;
|
||||
}
|
||||
};
|
||||
|
||||
match cfg.predicate.admit(&detail) {
|
||||
Ok(()) => {
|
||||
admitted += 1;
|
||||
let submission = to_submission(&detail, &cfg.source_name);
|
||||
if dry_run {
|
||||
tracing::info!(id = %detail.id, summary = ?submission.summary, "would submit");
|
||||
} else {
|
||||
match submit(&client, &cfg, &submission).await {
|
||||
Ok(()) => {
|
||||
submitted += 1;
|
||||
tracing::info!(id = %detail.id, "submitted candidate");
|
||||
}
|
||||
Err(e) => tracing::warn!(id = %detail.id, error = %e, "submit failed"),
|
||||
}
|
||||
}
|
||||
}
|
||||
Err(reason) => tracing::debug!(id = %summary.id, reason, "rejected"),
|
||||
}
|
||||
}
|
||||
|
||||
tracing::info!(
|
||||
scanned,
|
||||
admitted,
|
||||
submitted,
|
||||
dry_run,
|
||||
"modelwatch cycle complete"
|
||||
);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn fetch_list(client: &Client, url: &str) -> anyhow::Result<Vec<HfModel>> {
|
||||
let models = client
|
||||
.get(url)
|
||||
.send()
|
||||
.await
|
||||
.with_context(|| format!("GET {url}"))?
|
||||
.error_for_status()
|
||||
.with_context(|| format!("bad status from {url}"))?
|
||||
.json::<Vec<HfModel>>()
|
||||
.await
|
||||
.with_context(|| format!("decoding list from {url}"))?;
|
||||
Ok(models)
|
||||
}
|
||||
|
||||
async fn fetch_detail(client: &Client, base: &str, id: &str) -> anyhow::Result<HfModel> {
|
||||
let url = format!("{base}/{id}");
|
||||
let model = client
|
||||
.get(&url)
|
||||
.send()
|
||||
.await
|
||||
.with_context(|| format!("GET {url}"))?
|
||||
.error_for_status()
|
||||
.with_context(|| format!("bad status from {url}"))?
|
||||
.json::<HfModel>()
|
||||
.await
|
||||
.with_context(|| format!("decoding detail from {url}"))?;
|
||||
Ok(model)
|
||||
}
|
||||
|
||||
/// POST a candidate to the ingest endpoint. The endpoint replies `202 ACCEPTED` whether the
|
||||
/// item was newly created or a dedup no-op, so any 2xx counts as delivered.
|
||||
async fn submit(client: &Client, cfg: &Config, sub: &CandidateSubmission) -> anyhow::Result<()> {
|
||||
let resp = client
|
||||
.post(&cfg.ingest_url)
|
||||
.bearer_auth(&cfg.token)
|
||||
.json(sub)
|
||||
.send()
|
||||
.await
|
||||
.context("posting candidate")?;
|
||||
let status = resp.status();
|
||||
if status.is_success() {
|
||||
Ok(())
|
||||
} else {
|
||||
let body = resp.text().await.unwrap_or_default();
|
||||
bail!("ingest returned {status}: {body}")
|
||||
}
|
||||
}
|
||||
|
||||
/// JSON logs under journald (`JOURNAL_STREAM` set), pretty logs on a TTY.
|
||||
fn init_tracing() {
|
||||
use tracing_subscriber::{EnvFilter, fmt, prelude::*};
|
||||
let filter = EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info"));
|
||||
let registry = tracing_subscriber::registry().with(filter);
|
||||
if std::env::var_os("JOURNAL_STREAM").is_some() {
|
||||
registry
|
||||
.with(fmt::layer().json().flatten_event(true))
|
||||
.init();
|
||||
} else {
|
||||
registry.with(fmt::layer()).init();
|
||||
}
|
||||
}
|
||||
@@ -113,6 +113,14 @@ SYSU
|
||||
install -d -m 0750 -o root -g newsfeed /etc/newsfeed
|
||||
install -d -m 0750 -o newsfeed -g newsfeed /var/lib/newsfeed
|
||||
install -d -m 0755 -o root -g root /usr/local/bin
|
||||
|
||||
# modelwatch ingest token holder — a secret, so it lives outside the deployable
|
||||
# config and is created empty here, never clobbered by a redeploy.
|
||||
if [ ! -f /etc/newsfeed/modelwatch.env ]; then
|
||||
printf '# Ingest token for newsfeed-modelwatch. Mint one in the newsfeed UI\n# (Settings -> API tokens), paste below, then: systemctl start newsfeed-modelwatch\nNEWSFEED_TOKEN=\n' > /etc/newsfeed/modelwatch.env
|
||||
chown root:newsfeed /etc/newsfeed/modelwatch.env
|
||||
chmod 0640 /etc/newsfeed/modelwatch.env
|
||||
fi
|
||||
restorecon -R /etc/newsfeed /var/lib/newsfeed
|
||||
|
||||
# SELinux: label the API port so the daemon may bind it
|
||||
@@ -129,14 +137,19 @@ FWD
|
||||
"
|
||||
|
||||
info "[$host] installing systemd units"
|
||||
for unit in newsfeed-api.service newsfeed-worker.service; do
|
||||
for unit in newsfeed-api.service newsfeed-worker.service \
|
||||
newsfeed-modelwatch.service newsfeed-modelwatch.timer; do
|
||||
local content; content="$(cat "$REPO_ROOT/asset/systemd/$unit")"
|
||||
run_remote "$host" "cat > /etc/systemd/system/$unit <<'UNIT'
|
||||
${content}
|
||||
UNIT
|
||||
systemctl daemon-reload
|
||||
systemctl enable $unit"
|
||||
"
|
||||
done
|
||||
run_remote "$host" "
|
||||
systemctl daemon-reload
|
||||
systemctl enable newsfeed-api.service newsfeed-worker.service
|
||||
# modelwatch is a oneshot driven by its timer — enable the timer, not the service.
|
||||
systemctl enable --now newsfeed-modelwatch.timer"
|
||||
}
|
||||
|
||||
# --- WEB host (oolon): webroot, internal cert, nginx vhost --------------------------
|
||||
@@ -284,6 +297,9 @@ main() {
|
||||
provision_api_host
|
||||
provision_web_host
|
||||
dns_reminders
|
||||
info "modelwatch: mint an ingest token in the newsfeed UI (Settings -> API tokens),"
|
||||
info " put it in ${API_HOST}:/etc/newsfeed/modelwatch.env (NEWSFEED_TOKEN=...), then"
|
||||
info " 'systemctl start newsfeed-modelwatch' to seed immediately (the timer runs 3-hourly)."
|
||||
info "done"
|
||||
}
|
||||
|
||||
|
||||
@@ -4,7 +4,8 @@ import type { Media } from "./Media";
|
||||
/**
|
||||
* A candidate pushed to `POST /v1/ingest/candidates` by an agentic or algorithmic
|
||||
* source. The authenticated API token determines the owning user; `source` names the
|
||||
* producer for attribution and per-source weighting.
|
||||
* producer for attribution and per-source weighting. Derives `Serialize` too so
|
||||
* in-workspace producers (e.g. `newsfeed-modelwatch`) can build and post one.
|
||||
*/
|
||||
export type CandidateSubmission = {
|
||||
/**
|
||||
|
||||
Reference in New Issue
Block a user