From 3db1f586a87814f3b3df59e3082de91d7ad99338 Mon Sep 17 00:00:00 2001 From: rob thijssen Date: Wed, 8 Jul 2026 14:29:32 +0300 Subject: [PATCH] feat(modelwatch): push producer for open-weight model releases MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 Claude-Session: https://claude.ai/code/session_016fKZzDpvjiJ9eYbPGgJvUP --- .gitea/workflows/deploy.yml | 14 +- CLAUDE.md | 16 +- Cargo.lock | 17 + asset/config/modelwatch.toml.tmpl | 35 ++ asset/systemd/newsfeed-modelwatch.service | 32 ++ asset/systemd/newsfeed-modelwatch.timer | 13 + crates/newsfeed-entities/src/item.rs | 5 +- crates/newsfeed-modelwatch/Cargo.toml | 32 ++ crates/newsfeed-modelwatch/src/config.rs | 93 ++++++ crates/newsfeed-modelwatch/src/lib.rs | 351 ++++++++++++++++++++ crates/newsfeed-modelwatch/src/main.rs | 217 ++++++++++++ script/infra-setup.sh | 22 +- web/src/api/bindings/CandidateSubmission.ts | 3 +- 13 files changed, 837 insertions(+), 13 deletions(-) create mode 100644 asset/config/modelwatch.toml.tmpl create mode 100644 asset/systemd/newsfeed-modelwatch.service create mode 100644 asset/systemd/newsfeed-modelwatch.timer create mode 100644 crates/newsfeed-modelwatch/Cargo.toml create mode 100644 crates/newsfeed-modelwatch/src/config.rs create mode 100644 crates/newsfeed-modelwatch/src/lib.rs create mode 100644 crates/newsfeed-modelwatch/src/main.rs diff --git a/.gitea/workflows/deploy.yml b/.gitea/workflows/deploy.yml index 8bd98f6..208700d 100644 --- a/.gitea/workflows/deploy.yml +++ b/.gitea/workflows/deploy.yml @@ -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 diff --git a/CLAUDE.md b/CLAUDE.md index ca51894..7e6abfb 100644 --- a/CLAUDE.md +++ b/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) diff --git a/Cargo.lock b/Cargo.lock index 3443c13..16d48e4 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -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" diff --git a/asset/config/modelwatch.toml.tmpl b/asset/config/modelwatch.toml.tmpl new file mode 100644 index 0000000..6e607c6 --- /dev/null +++ b/asset/config/modelwatch.toml.tmpl @@ -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"] diff --git a/asset/systemd/newsfeed-modelwatch.service b/asset/systemd/newsfeed-modelwatch.service new file mode 100644 index 0000000..04fa1c0 --- /dev/null +++ b/asset/systemd/newsfeed-modelwatch.service @@ -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 diff --git a/asset/systemd/newsfeed-modelwatch.timer b/asset/systemd/newsfeed-modelwatch.timer new file mode 100644 index 0000000..bf716d1 --- /dev/null +++ b/asset/systemd/newsfeed-modelwatch.timer @@ -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 diff --git a/crates/newsfeed-entities/src/item.rs b/crates/newsfeed-entities/src/item.rs index 92b0ed3..6dfa7f2 100644 --- a/crates/newsfeed-entities/src/item.rs +++ b/crates/newsfeed-entities/src/item.rs @@ -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. diff --git a/crates/newsfeed-modelwatch/Cargo.toml b/crates/newsfeed-modelwatch/Cargo.toml new file mode 100644 index 0000000..94f1e11 --- /dev/null +++ b/crates/newsfeed-modelwatch/Cargo.toml @@ -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 diff --git a/crates/newsfeed-modelwatch/src/config.rs b/crates/newsfeed-modelwatch/src/config.rs new file mode 100644 index 0000000..d231dca --- /dev/null +++ b/crates/newsfeed-modelwatch/src/config.rs @@ -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, + /// 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 { + 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()?) + } +} diff --git a/crates/newsfeed-modelwatch/src/lib.rs b/crates/newsfeed-modelwatch/src/lib.rs new file mode 100644 index 0000000..3346bc2 --- /dev/null +++ b/crates/newsfeed-modelwatch/src/lib.rs @@ -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, + #[serde(rename = "createdAt", default)] + pub created_at: Option>, + #[serde(default)] + pub downloads: Option, + #[serde(default)] + pub likes: Option, + #[serde(rename = "trendingScore", default)] + pub trending_score: Option, + #[serde(rename = "pipeline_tag", default)] + pub pipeline_tag: Option, + #[serde(default)] + pub gated: Gated, + #[serde(default)] + pub tags: Vec, + #[serde(default)] + pub safetensors: Option, + #[serde(rename = "cardData", default)] + pub card_data: Option, +} + +/// 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, +} + +/// The bits of `cardData` we care about. +#[derive(Debug, Clone, Default, Deserialize)] +pub struct CardData { + #[serde(default)] + pub license: Option, +} + +/// The license, lowercased — from `cardData.license`, else a `license:` tag. +pub fn license(m: &HfModel) -> Option { + 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, + /// 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, +} + +fn default_licenses() -> Vec { + 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 { + 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"); + } +} diff --git a/crates/newsfeed-modelwatch/src/main.rs b/crates/newsfeed-modelwatch/src/main.rs new file mode 100644 index 0000000..45354ec --- /dev/null +++ b/crates/newsfeed-modelwatch/src/main.rs @@ -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 = 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> { + let models = client + .get(url) + .send() + .await + .with_context(|| format!("GET {url}"))? + .error_for_status() + .with_context(|| format!("bad status from {url}"))? + .json::>() + .await + .with_context(|| format!("decoding list from {url}"))?; + Ok(models) +} + +async fn fetch_detail(client: &Client, base: &str, id: &str) -> anyhow::Result { + 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::() + .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(); + } +} diff --git a/script/infra-setup.sh b/script/infra-setup.sh index 4f322ec..674b3a2 100755 --- a/script/infra-setup.sh +++ b/script/infra-setup.sh @@ -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" } diff --git a/web/src/api/bindings/CandidateSubmission.ts b/web/src/api/bindings/CandidateSubmission.ts index f1e8603..4581959 100644 --- a/web/src/api/bindings/CandidateSubmission.ts +++ b/web/src/api/bindings/CandidateSubmission.ts @@ -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 = { /**