Signals: index the judge's query, six calls at once, one call per item on the server's key, tokens counted, a week's look-back an admin sets, edits that ask again only what was picked out (#274, #275, #276)

The judge's query read all of entries (605 MB, 566 ms) on every wake; an
index on each feed's items by date, as the query asks, and the rechecks
in a half of their own. Calls ran one at a time, about 150 ms each, so a
new Signal's month of about 8,000 items took twenty minutes: six in flight
now. On the server's key everyone's Signals on an item share one call.
TypeSafe's token counts go to /metrics and each jev span.

An item's text is cut at 1,000 characters instead of 2,000: on 55 real
items of eight podcast feeds and six Signals, on the development key,
1,000 changed none of 330 verdicts (nor did the same request twice) and
saved 12% of the input tokens, 1,266 to 1,112 a call; 500 changed one.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

crates/ipx/CHANGELOG.md

@@ -14,12 +14,20 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
-- `/metrics` counts Signals: items judged and fitting, calls to Jev by how they ended and the time they took, time since the judge's last pass, and keys refused; a failed call to Jev shows as a failed trace (#270).
+- `/metrics` counts Signals: items judged and fitting, calls to Jev by how they ended, the time they took and the tokens TypeSafe counted, time since the judge's last pass, and keys refused; a failed call to Jev shows as a failed trace (#270, #275).
+- An admin setting for how many days back a new Signal reads (#276).
 - An admin setting, Signals for everyone on the server's TypeSafe key, so the server pays for Jev instead of each person (#271).
 
 ### Changed
 
 - A TypeSafe key is checked with TypeSafe before it is saved, and one it refuses is not saved (#271).
+- A new Signal reads the last week instead of the last 30 days, unless an admin sets otherwise (#276).
+- Editing a Signal's name, words or description asks again only about what it had picked out, and about new items as they come, instead of the whole window again (#276).
+- Signals fill faster and cost less: six calls to TypeSafe at once instead of one, everyone's Signals on an item in one call on the server's key, and an item's text cut at 1,000 characters instead of 2,000 (#275).
+
+### Fixed
+
+- Looking for items to read for Signals read every item on the server on each scan; it now reads only recent ones (#274).
 
 ## [1.3.0] - 2026-10-16
 

crates/ipx/src/config.rs

@@ -53,6 +53,9 @@ pub struct General {
     /// Signals are judged on the server's TypeSafe key (`TYPESAFE_KEY`), for everyone, rather than
     /// on each person's own (#271). Without that key it changes nothing.
     pub signals_on_server: bool,
+    /// How far back a new Signal looks, in days (#276): each item is a call to TypeSafe, and a
+    /// month of production's feeds was about 8,000.
+    pub signal_days: u64,
 }
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Deserialize, Serialize)]
@@ -211,6 +214,7 @@ impl Default for General {
             directory_token: String::new(),
             directory_share: false,
             signals_on_server: false,
+            signal_days: 7,
         }
     }
 }
@@ -337,6 +341,12 @@ pub struct Stored {
     pub directory_share: bool,
     #[serde(default)]
     pub signals_on_server: bool,
+    #[serde(default = "signal_days")]
+    pub signal_days: u64,
+}
+
+fn signal_days() -> u64 {
+    General::default().signal_days
 }
 
 fn art_cache_mb() -> u64 {
@@ -344,7 +354,7 @@ fn art_cache_mb() -> u64 {
 }
 
 /// `[general]` keys that live in the database once it holds the configuration.
-const STORED_KEYS: [&str; 10] = [
+const STORED_KEYS: [&str; 11] = [
     "schedule",
     "max_total_gb",
     "max_age_days",
@@ -355,6 +365,7 @@ const STORED_KEYS: [&str; 10] = [
     "directory_token",
     "directory_share",
     "signals_on_server",
+    "signal_days",
 ];
 
 impl Stored {
@@ -371,6 +382,7 @@ impl Stored {
             directory_token: g.directory_token.clone(),
             directory_share: g.directory_share,
             signals_on_server: g.signals_on_server,
+            signal_days: g.signal_days,
         }
     }
 
@@ -386,6 +398,7 @@ impl Stored {
         g.directory_token = self.directory_token;
         g.directory_share = self.directory_share;
         g.signals_on_server = self.signals_on_server;
+        g.signal_days = self.signal_days;
     }
 }
 

crates/ipx/src/db.rs

@@ -79,6 +79,10 @@ async fn create_missing(orm: &sea_orm::DatabaseConnection) -> Result<()> {
         // and visits the table only for the rows it shows. Sorting All Subscriptions from the
         // table read all 13,700 of its rows, show notes and all, half a second to two cold (#188).
         "CREATE INDEX IF NOT EXISTS entries_by_time ON entries (feed_id, guid, published, first_seen)",
+        // The Signals judge's question on every wake, what is recent in each subscribed feed: as
+        // the expression it is asked by. Without it, each wake read all of entries, 605 MB, 566 ms
+        // to find nothing (#274).
+        "CREATE INDEX IF NOT EXISTS entries_recent ON entries (feed_id, (coalesce(published, first_seen)))",
     ] {
         orm.execute_unprepared(sql).await.with_context(|| sql.to_owned())?;
     }
@@ -113,6 +117,10 @@ async fn create_missing(orm: &sea_orm::DatabaseConnection) -> Result<()> {
             orm.execute_unprepared(sql).await.with_context(|| sql.to_owned())?;
         }
     }
+    if orm.execute_unprepared("SELECT recheck FROM signal_verdicts LIMIT 0").await.is_err() {
+        let sql = "ALTER TABLE signal_verdicts ADD COLUMN recheck BOOLEAN NOT NULL DEFAULT false";
+        orm.execute_unprepared(sql).await.with_context(|| sql.to_owned())?;
+    }
     if orm.execute_unprepared("SELECT icon FROM signals LIMIT 0").await.is_err() {
         let sql = "ALTER TABLE signals ADD COLUMN icon TEXT";
         orm.execute_unprepared(sql).await.with_context(|| sql.to_owned())?;
@@ -2589,8 +2597,6 @@ pub struct Unjudged {
     pub signals: Vec<Ask>,
 }
 
-/// How far back a Signal looks when it is made: a month of items, not every back catalogue.
-pub const SIGNAL_DAYS: i64 = 30;
 
 impl Db {
     pub async fn signals(&self, user_id: i64) -> Result<Vec<SignalRow>> {
@@ -2654,9 +2660,10 @@ impl Db {
         Ok(m.id)
     }
 
-    /// Changes a Signal. A new name, words or sentence puts every item to Jev again, since what it
-    /// said was about the old question (#259); a new icon does not. False for a Signal that is not
-    /// this person's.
+    /// Changes a Signal. A new name, words or sentence puts what it picked out to Jev again, since
+    /// that was the old question's answer; what it passed over stays passed over, and new items
+    /// are asked the new question as they come (#276). A new icon asks nothing. False for a Signal
+    /// that is not this person's.
     pub async fn set_signal(&self, user_id: i64, id: i64, change: SignalChange<'_>) -> Result<bool> {
         let SignalChange { name, words, description, icon } = change;
         if self.signal(user_id, id).await?.is_none() {
@@ -2676,7 +2683,12 @@ impl Db {
             m.icon = Set(i.map(str::to_owned));
         }
         if name.is_some() || words.is_some() || description.is_some() {
-            signal_verdicts::Entity::delete_many().filter(signal_verdicts::Column::SignalId.eq(id)).exec(&self.orm).await?;
+            signal_verdicts::Entity::update_many()
+                .col_expr(signal_verdicts::Column::Recheck, Expr::val(true).into())
+                .filter(signal_verdicts::Column::SignalId.eq(id))
+                .filter(signal_verdicts::Column::Hit.eq(true))
+                .exec(&self.orm)
+                .await?;
         }
         if name.is_some() || words.is_some() || description.is_some() || icon.is_some() {
             m.update(&self.orm).await?;
@@ -2703,13 +2715,14 @@ impl Db {
             .await
     }
 
-    /// Up to `limit` items, newest first, that someone with a key has a Signal not yet asked
-    /// about: from the feeds they subscribe to, the last `SIGNAL_DAYS`, and not hidden from them.
-    /// `skip` are people whose key TypeSafe has refused. With `everyone`, the server's key is
-    /// used, and people without a key of their own are asked for too (#271).
-    pub async fn unjudged(&self, skip: &[i64], limit: i64, everyone: bool) -> Result<Vec<Unjudged>> {
+    /// Up to `limit` items, newest first, that someone with a key has a Signal to ask about: not
+    /// yet asked, from the last `days` (#276), or picked out before the Signal changed, however
+    /// old; from the feeds they subscribe to, and not hidden from them. `skip` are people whose
+    /// key TypeSafe has refused. With `everyone`, the server's key is used, and people without a
+    /// key of their own are asked for too (#271).
+    pub async fn unjudged(&self, skip: &[i64], limit: i64, everyone: bool, days: u64) -> Result<Vec<Unjudged>> {
         let mut a = Args::default();
-        let since = a.p(now() - SIGNAL_DAYS * 86_400);
+        let since = a.p(now() - days as i64 * 86_400);
         let skip = if skip.is_empty() {
             String::new()
         } else {
@@ -2717,23 +2730,32 @@ impl Db {
         };
         let limit = a.p(limit);
         let keyed = if everyone { "true" } else { "u.typesafe_key IS NOT NULL" };
-        let sql = format!(
-            "SELECT g.id AS signal_id, g.name, g.keywords, g.description AS signal_description, g.user_id, u.typesafe_key, e.feed_id, e.guid, e.title,
-                    e.description, f.title AS feed_title
-               FROM signals g
-               JOIN users u ON u.id = g.user_id
-               JOIN subscriptions s ON s.user_id = g.user_id
-               JOIN entries e ON e.feed_id = s.feed_id
-               LEFT JOIN feeds f ON f.id = e.feed_id
-              WHERE {keyed} {skip}
-                AND coalesce(e.published, e.first_seen) >= {since}
-                AND NOT EXISTS (SELECT 1 FROM signal_verdicts v
-                                 WHERE v.signal_id = g.id AND v.feed_id = e.feed_id AND v.guid = e.guid)
-                AND NOT EXISTS (SELECT 1 FROM hidden h
-                                 WHERE h.user_id = g.user_id AND h.feed_id = e.feed_id AND h.guid = e.guid)
-              ORDER BY coalesce(e.published, e.first_seen) DESC, e.feed_id, e.guid, g.user_id
-              LIMIT {limit}"
+        // Two halves, not one OR: the recent half reads entries_recent, and the rechecks are a few
+        // rows of signal_verdicts. The same placeholders serve both.
+        let half = |items: &str| {
+            format!(
+                "SELECT g.id AS signal_id, g.name, g.keywords, g.description AS signal_description, g.user_id, u.typesafe_key,
+                        e.feed_id, e.guid, e.title, e.description, f.title AS feed_title, coalesce(e.published, e.first_seen) AS at
+                   FROM signals g
+                   JOIN users u ON u.id = g.user_id
+                   JOIN subscriptions s ON s.user_id = g.user_id
+                   {items}
+                   LEFT JOIN feeds f ON f.id = e.feed_id
+                  WHERE {keyed} {skip}
+                    AND NOT EXISTS (SELECT 1 FROM hidden h
+                                     WHERE h.user_id = g.user_id AND h.feed_id = e.feed_id AND h.guid = e.guid)"
+            )
+        };
+        let recent = half(&format!(
+            "JOIN entries e ON e.feed_id = s.feed_id AND coalesce(e.published, e.first_seen) >= {since}
+              AND NOT EXISTS (SELECT 1 FROM signal_verdicts v
+                               WHERE v.signal_id = g.id AND v.feed_id = e.feed_id AND v.guid = e.guid)"
+        ));
+        let again = half(
+            "JOIN signal_verdicts v ON v.signal_id = g.id AND v.feed_id = s.feed_id AND v.recheck
+             JOIN entries e ON e.feed_id = v.feed_id AND e.guid = v.guid",
         );
+        let sql = format!("SELECT * FROM ({recent} UNION ALL {again}) q ORDER BY at DESC, feed_id, guid, user_id LIMIT {limit}");
         let mut out: Vec<Unjudged> = Vec::new();
         for r in self.rows(&sql, a.0).await? {
             let (user_id, feed_id, guid): (i64, String, String) =
@@ -2769,9 +2791,9 @@ impl Db {
             // would refuse and SQLite would keep for nothing.
             // shortcut: keywords changed mid-call keep a verdict on the old ones for that item.
             self.exec(
-                "INSERT INTO signal_verdicts (signal_id, feed_id, guid, hit)
-                 SELECT $1, $2, $3, $4 WHERE EXISTS (SELECT 1 FROM signals WHERE id = $1)
-                 ON CONFLICT (signal_id, feed_id, guid) DO UPDATE SET hit = excluded.hit",
+                "INSERT INTO signal_verdicts (signal_id, feed_id, guid, hit, recheck)
+                 SELECT $1, $2, $3, $4, false WHERE EXISTS (SELECT 1 FROM signals WHERE id = $1)
+                 ON CONFLICT (signal_id, feed_id, guid) DO UPDATE SET hit = excluded.hit, recheck = false",
                 vec![(*signal).into(), feed_id.into(), guid.into(), (*hit).into()],
             )
             .await?;
@@ -3397,23 +3419,23 @@ mod tests {
         let food = db.add_signal(1, "Food", &words(&["baking"]), Some("Bread and cakes")).await.unwrap();
         db.add_signal(2, "Theirs", &words(&["x"]), None).await.unwrap();
 
-        assert!(db.unjudged(&[], 50, false).await.unwrap().is_empty(), "nobody has a key yet");
-        let everyone = db.unjudged(&[], 50, true).await.unwrap();
+        assert!(db.unjudged(&[], 50, false, 30).await.unwrap().is_empty(), "nobody has a key yet");
+        let everyone = db.unjudged(&[], 50, true, 30).await.unwrap();
         let mut seen: Vec<_> = everyone.iter().map(|u| (u.user_id, u.guid.as_str(), u.key.is_none())).collect();
         seen.sort();
         assert_eq!(seen, [(1, "x", true), (1, "y", true), (2, "x", true)], "on the server's key, everyone's Signals, no key of their own needed");
         db.set_typesafe_key(1, Some("k")).await.unwrap();
-        let queue = db.unjudged(&[], 50, false).await.unwrap();
+        let queue = db.unjudged(&[], 50, false, 30).await.unwrap();
         let mut seen: Vec<_> = queue.iter().map(|u| (u.guid.as_str(), u.signals.len())).collect();
         seen.sort();
         assert_eq!(seen, [("x", 2), ("y", 2)], "each recent subscribed item once, with both Signals; old and unsubscribed left out");
         let ask = queue[0].signals.iter().find(|a| a.id == food).unwrap();
         assert_eq!((ask.name.as_str(), ask.description.as_deref()), ("Food", Some("Bread and cakes")), "Jev is asked about the name and the sentence");
-        assert!(db.unjudged(&[1], 50, false).await.unwrap().is_empty(), "a refused key is passed over");
+        assert!(db.unjudged(&[1], 50, false, 30).await.unwrap().is_empty(), "a refused key is passed over");
 
         db.record_verdicts("f", "x", &[(f1, true), (food, false)]).await.unwrap();
         db.record_verdicts("g", "y", &[(f1, false), (food, true)]).await.unwrap();
-        assert!(db.unjudged(&[], 50, false).await.unwrap().is_empty(), "nothing is asked twice");
+        assert!(db.unjudged(&[], 50, false, 30).await.unwrap().is_empty(), "nothing is asked twice");
 
         let order = order_sql("published", "desc");
         let page = db.signal_entries(1, f1, Filter::All, None, 0, 50, &order).await.unwrap();
@@ -3430,15 +3452,25 @@ mod tests {
         assert!(!db.set_signal(2, f1, SignalChange { name: Some("mine"), ..Default::default() }).await.unwrap());
         assert!(!db.remove_signal(2, f1).await.unwrap());
 
-        db.set_signal(1, f1, SignalChange { words: Some(&words(&["f1"])), ..Default::default() }).await.unwrap();
-        assert_eq!(db.signal_count(1, f1, Filter::All, None).await.unwrap(), 0, "new keywords are judged afresh");
-        db.record_verdicts("f", "x", &[(f1, true)]).await.unwrap();
+        // An edit asks again only what was picked out, however old, and leaves what was passed
+        // over (#276). Until answered, what was picked out stays in the list.
+        db.record_verdicts("g", "old", &[(f1, true)]).await.unwrap();
         db.set_signal(1, f1, SignalChange { description: Some(Some("Grand prix racing")), ..Default::default() }).await.unwrap();
-        assert_eq!(db.signal_count(1, f1, Filter::All, None).await.unwrap(), 0, "so is a new sentence");
         assert_eq!(db.signal(1, f1).await.unwrap().unwrap().description.as_deref(), Some("Grand prix racing"));
+        let again: Vec<_> = db.unjudged(&[], 50, false, 30).await.unwrap().iter()
+            .map(|u| (u.guid.clone(), u.signals.iter().map(|a| a.id).collect::<Vec<_>>())).collect();
+        assert_eq!(again, [("x".to_string(), vec![f1]), ("old".to_string(), vec![f1])], "y was passed over, and stays so");
+        assert_eq!(db.signal_count(1, f1, Filter::All, None).await.unwrap(), 2);
+        db.record_verdicts("f", "x", &[(f1, false)]).await.unwrap();
+        db.record_verdicts("g", "old", &[(f1, true)]).await.unwrap();
+        assert!(db.unjudged(&[], 50, false, 30).await.unwrap().is_empty(), "answered, not asked again");
+        assert_eq!(db.signal_count(1, f1, Filter::All, None).await.unwrap(), 1);
+        assert!(db.unjudged(&[], 50, false, 1).await.unwrap().is_empty());
+
         db.record_verdicts("f", "x", &[(f1, true)]).await.unwrap();
         db.set_signal(1, f1, SignalChange { icon: Some(Some("solid/flag-checkered")), ..Default::default() }).await.unwrap();
-        assert_eq!(db.signal_count(1, f1, Filter::All, None).await.unwrap(), 1, "a new icon asks nothing again");
+        assert_eq!(db.signal_count(1, f1, Filter::All, None).await.unwrap(), 2, "x and old, kept");
+        assert!(db.unjudged(&[], 50, false, 30).await.unwrap().is_empty(), "a new icon asks nothing again");
         assert_eq!(db.signal(1, f1).await.unwrap().unwrap().icon.as_deref(), Some("solid/flag-checkered"));
         assert!(db.remove_signal(1, food).await.unwrap());
         db.record_verdicts("f", "x", &[(food, true)]).await.unwrap();

crates/ipx/src/entity.rs

@@ -333,6 +333,10 @@ pub mod signal_verdicts {
         #[sea_orm(primary_key, auto_increment = false, column_type = "Text")]
         pub guid: String,
         pub hit: bool,
+        /// A picked-out item to ask about again, since its Signal changed (#276). It stays in the
+        /// Signal's list until Jev has answered the new question.
+        #[sea_orm(default_value = false)]
+        pub recheck: bool,
     }
 
     #[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)]

crates/ipx/src/metrics.rs

@@ -48,6 +48,8 @@ pub struct Stats {
     /// Calls to Jev by how they ended, in `JEV_RESULTS`' order, and their microseconds all told.
     pub jev_calls: [AtomicU64; 4],
     pub jev_micros: AtomicU64,
+    /// Tokens TypeSafe counted, input and output, as its answers say.
+    pub jev_tokens: [AtomicU64; 2],
     /// The worker's job now, and since when.
     job: std::sync::Mutex<Option<(&'static str, std::time::Instant)>>,
     /// Jobs finished, by kind: how many, and their seconds all told.
@@ -77,6 +79,7 @@ impl Stats {
             signals_last_pass: AtomicI64::new(0),
             jev_calls: [const { AtomicU64::new(0) }; 4],
             jev_micros: AtomicU64::new(0),
+            jev_tokens: [const { AtomicU64::new(0) }; 2],
             job: std::sync::Mutex::new(None),
             jobs: std::sync::Mutex::new(std::collections::BTreeMap::new()),
         }
@@ -250,6 +253,8 @@ fn work(out: &mut String, st: &Stats, event_queue: usize, retries: usize, art_li
     let calls: Vec<(&str, f64)> = labels.iter().zip(&st.jev_calls).map(|(l, c)| (l.as_str(), c.load(Relaxed) as f64)).collect();
     counter(out, "ipx_jev_requests_total", "Calls to TypeSafe's Jev, Signals' and their icons', by how they ended.", &calls);
     counter(out, "ipx_jev_seconds_total", "Seconds spent waiting on Jev.", &[("", st.jev_micros.load(Relaxed) as f64 / 1e6)]);
+    counter(out, "ipx_jev_tokens_total", "Tokens TypeSafe counted for calls to Jev, what they cost.",
+        &[("kind=\"input\"", st.jev_tokens[0].load(Relaxed) as f64), ("kind=\"output\"", st.jev_tokens[1].load(Relaxed) as f64)]);
     for (name, help, v) in [
         ("ipx_feeds", "Feeds in the catalogue, as last counted.", &st.feeds),
         ("ipx_downloads_pending", "Files a scan will download on its own, as last counted.", &st.pending),

crates/ipx/src/signals.rs

@@ -5,6 +5,9 @@
 
 use std::collections::HashSet;
 use std::sync::Arc;
+use std::sync::atomic::AtomicBool;
+
+use futures_util::StreamExt as _;
 
 use anyhow::Result;
 use tracing::Instrument;
@@ -21,10 +24,15 @@ use crate::metrics::STATS;
 const BATCH: i64 = 50;
 /// Between passes when nothing wakes it: a scan or a new Signal does, so this is a backstop.
 const IDLE: std::time::Duration = std::time::Duration::from_secs(300);
+/// Calls to Jev in flight at once (#275). One at a time, about 150 ms each, a new Signal's month
+/// of items took twenty minutes; few enough at once to stay a polite client of TypeSafe.
+const IN_FLIGHT: usize = 6;
 /// How sure Jev has to be that an item is about the keywords.
 const THRESHOLD: f64 = 0.5;
-/// Of an item's text, which is a podcast's whole show notes at times.
-const TEXT_CHARS: usize = 2000;
+/// Of an item's text, which is a podcast's whole show notes at times. 2000 until #275: on 55 real
+/// items and six Signals, 1000 changed none of 330 verdicts and saved an eighth of the input
+/// tokens; 500 changed one, an Apple podcast's episode no longer about Apple.
+const TEXT_CHARS: usize = 1000;
 
 pub async fn run(ctx: Arc<Ctx>, mut stop: tokio::sync::watch::Receiver<bool>) {
     loop {
@@ -66,56 +74,78 @@ async fn pass(ctx: &Ctx) -> Result<bool> {
     let server = server_key(ctx);
     // People's refused keys are not what is asked with when the server's is.
     let skip: Vec<i64> = if server.is_some() { Vec::new() } else { ctx.signals_refused.lock().unwrap().iter().copied().collect() };
-    let items = ctx.db.unjudged(&skip, BATCH, server.is_some()).await?;
+    let days = ctx.cfg().general.signal_days;
+    let items = ctx.db.unjudged(&skip, BATCH, server.is_some(), days).await?;
     if items.is_empty() {
         return Ok(false);
     }
-    let span = tracing::info_span!("signals", entries = items.len(), judged = tracing::field::Empty, hits = tracing::field::Empty);
+    let span = tracing::info_span!("signals", entries = items.len(), calls = tracing::field::Empty, judged = tracing::field::Empty, hits = tracing::field::Empty);
     async {
-        let mut refused = HashSet::new();
+        let groups = calls(&items, server.is_some());
+        tracing::Span::current().record("calls", groups.len());
+        let refused = std::sync::Mutex::new(HashSet::new());
+        let stopped = AtomicBool::new(false);
         let mut hits_for = std::collections::BTreeMap::<i64, usize>::new();
-        let mut stopped = false;
         let mut judged = 0;
-        for item in &items {
-            if refused.contains(&item.user_id) {
-                continue;
-            }
-            let Some(key) = server.as_deref().or(item.key.as_deref()) else { continue };
-            match judge(ctx, item, key).await {
-                Ok(verdicts) => {
-                    judged += 1;
-                    let hits = verdicts.iter().filter(|(_, h)| *h).count();
-                    STATS.signals_judged.fetch_add(1, Relaxed);
-                    STATS.signals_hits.fetch_add(hits as u64, Relaxed);
-                    if hits > 0 {
-                        *hits_for.entry(item.user_id).or_default() += hits;
-                        tracing::info!(ev = "signal", user = item.user_id, feed = %item.feed_id, new = hits, "an item fits a Signal");
+        // Each call starts only when a place is free, so one that sees a key refused or the pass
+        // stopped is not sent; those already in flight are let finish, and their answers kept.
+        // By index: a closure taking a reference here is not Send for every lifetime, which
+        // tokio::spawn asks of the judge's future.
+        let mut answers = futures_util::stream::iter(0..groups.len())
+            .map(|n| {
+                let (group, refused, stopped, server) = (&groups[n], &refused, &stopped, &server);
+                async move {
+                    let first = group[0];
+                    if stopped.load(Relaxed) || refused.lock().unwrap().contains(&first.user_id) {
+                        return None;
+                    }
+                    let key = server.as_deref().or(first.key.as_deref())?;
+                    Some((group, judge(ctx, group, key).await))
+                }
+            })
+            .buffer_unordered(IN_FLIGHT);
+        while let Some(answer) = answers.next().await {
+            let Some((group, answer)) = answer else { continue };
+            let first = group[0];
+            match answer {
+                Ok(verdicts) => {
+                    for (item, verdicts) in group.iter().zip(verdicts) {
+                        judged += 1;
+                        let hits = verdicts.iter().filter(|(_, h)| *h).count();
+                        STATS.signals_judged.fetch_add(1, Relaxed);
+                        STATS.signals_hits.fetch_add(hits as u64, Relaxed);
+                        if hits > 0 {
+                            *hits_for.entry(item.user_id).or_default() += hits;
+                            tracing::info!(ev = "signal", user = item.user_id, feed = %item.feed_id, new = hits, "an item fits a Signal");
+                        }
+                        ctx.db.record_verdicts(&item.feed_id, &item.guid, &verdicts).instrument(step("record_verdicts")).await?;
+                    }
+                }
+                // The server's key refused is everyone's: the queue waits for the next pass.
+                Err(e) if refuses_key(&e) && server.is_some() => {
+                    if !stopped.swap(true, Relaxed) {
+                        tracing::warn!(error = %format!("{e:#}"), "TypeSafe refused the server's key for Signals");
                     }
-                    ctx.db.record_verdicts(&item.feed_id, &item.guid, &verdicts).instrument(step("record_verdicts")).await?;
                 }
                 // A key TypeSafe will not take is not sent the rest of the queue. It is tried
                 // again once its owner saves one (`web::patch_me`) or the daemon restarts.
-                // The server's key refused is everyone's: the queue waits for the next pass.
-                Err(e) if refuses_key(&e) && server.is_some() => {
-                    tracing::warn!(error = %format!("{e:#}"), "TypeSafe refused the server's key for Signals");
-                    stopped = true;
-                    break;
-                }
                 Err(e) if refuses_key(&e) => {
-                    tracing::warn!(user = item.user_id, error = %format!("{e:#}"), "TypeSafe refused a Signal key");
-                    refused.insert(item.user_id);
+                    if refused.lock().unwrap().insert(first.user_id) {
+                        tracing::warn!(user = first.user_id, error = %format!("{e:#}"), "TypeSafe refused a Signal key");
+                    }
                 }
                 // TypeSafe says the request itself is wrong, so every other item's would be too:
                 // the rest wait for the next pass rather than each failing the same way (#257).
                 Err(e) if malformed(&e) => {
-                    tracing::warn!(user = item.user_id, error = %format!("{e:#}"), "TypeSafe refused a Signals request as malformed");
-                    stopped = true;
-                    break;
+                    if !stopped.swap(true, Relaxed) {
+                        tracing::warn!(user = first.user_id, error = %format!("{e:#}"), "TypeSafe refused a Signals request as malformed");
+                    }
                 }
                 // Left unjudged, so it is asked again next pass.
-                Err(e) => tracing::warn!(user = item.user_id, feed = %item.feed_id, error = %format!("{e:#}"), "Jev could not judge an item"),
+                Err(e) => tracing::warn!(user = first.user_id, feed = %first.feed_id, error = %format!("{e:#}"), "Jev could not judge an item"),
             }
         }
+        drop(answers);
         let span = tracing::Span::current();
         span.record("judged", judged);
         span.record("hits", hits_for.values().sum::<usize>());
@@ -123,16 +153,31 @@ async fn pass(ctx: &Ctx) -> Result<bool> {
         for (user, hits) in hits_for {
             ctx.out.emit(crate::ipc::Event::SignalHits { user, hits });
         }
+        let refused = refused.into_inner().unwrap();
         let some_refused = !refused.is_empty();
         ctx.signals_refused.lock().unwrap().extend(refused);
         // A pass that only failed would otherwise go straight round again on the same items, and
         // one TypeSafe refused as malformed would ask again at once, however full.
-        anyhow::Ok(!stopped && (full(&items) || some_refused))
+        anyhow::Ok(!stopped.into_inner() && (full(&items) || some_refused))
     }
     .instrument(span)
     .await
 }
 
+/// The calls a batch makes, each one item and the people asking about it. On the server's key
+/// everyone's Signals on an item go in one call, its text sent once (#275); on people's own keys
+/// each pays for their own. `unjudged` puts one item's rows together.
+fn calls(items: &[Unjudged], share: bool) -> Vec<Vec<&Unjudged>> {
+    let mut out: Vec<Vec<&Unjudged>> = Vec::new();
+    for item in items {
+        match out.last_mut() {
+            Some(g) if share && (g[0].feed_id.as_str(), g[0].guid.as_str()) == (item.feed_id.as_str(), item.guid.as_str()) => g.push(item),
+            _ => out.push(vec![item]),
+        }
+    }
+    out
+}
+
 /// A call TypeSafe answered with an error, and what it said: its body says why, where the status
 /// alone said only 422 (#257).
 #[derive(Debug)]
@@ -168,23 +213,28 @@ fn malformed(e: &anyhow::Error) -> bool {
     matches!(status(e), Some(400 | 422))
 }
 
-async fn judge(ctx: &Ctx, item: &Unjudged, key: &str) -> Result<Vec<(i64, bool)>> {
+/// One call about one item, for everyone in `group`: their verdicts, in the group's order.
+async fn judge(ctx: &Ctx, group: &[&Unjudged], key: &str) -> Result<Vec<Vec<(i64, bool)>>> {
+    let asks: Vec<&Ask> = group.iter().flat_map(|i| &i.signals).collect();
     let span = tracing::info_span!(
         "jev",
-        user = item.user_id,
-        signal_count = item.signals.len(),
+        user = group[0].user_id,
+        people = group.len(),
+        signal_count = asks.len(),
         "http.response.status_code" = tracing::field::Empty,
         error = tracing::field::Empty,
         "otel.status_code" = tracing::field::Empty,
+        input_tokens = tracing::field::Empty,
+        output_tokens = tracing::field::Empty,
     );
-    let answer = called(span, ask(ctx, key, &request(item))).await?;
+    let answer = called(span, ask(ctx, key, &request(group[0], &asks))).await?;
     // A question Jev left unanswered stays unjudged and is asked again.
-    Ok(item.signals.iter().filter_map(|a| hit(&answer, a.id).map(|h| (a.id, h))).collect())
+    Ok(group.iter().map(|i| i.signals.iter().filter_map(|a| hit(&answer, a.id).map(|h| (a.id, h))).collect()).collect())
 }
 
 /// The item as Jev sees it, as text, which is what its docs show: an object with an empty feed
 /// title in it sent a null. One yes-or-no question for each Signal, keyed by its id.
-fn request(item: &Unjudged) -> serde_json::Value {
+fn request(item: &Unjudged, asks: &[&Ask]) -> serde_json::Value {
     let text: String = text_of(item.description.as_deref().unwrap_or("")).chars().take(TEXT_CHARS).collect();
     let state = format!(
         "Feed: {}\nTitle: {}\n\n{text}",
@@ -192,7 +242,7 @@ fn request(item: &Unjudged) -> serde_json::Value {
         item.title.as_deref().unwrap_or("")
     );
     let questions: serde_json::Map<String, serde_json::Value> =
-        item.signals.iter().map(|a| (key(a.id), question(a))).collect();
+        asks.iter().map(|a| (key(a.id), question(a))).collect();
     serde_json::json!({ "model": "jev-latest", "state": state, "questions": questions })
 }
 
@@ -236,6 +286,8 @@ pub async fn pick_icon(ctx: &Ctx, key: &str, name: &str, words: &[String], descr
         "http.response.status_code" = tracing::field::Empty,
         error = tracing::field::Empty,
         "otel.status_code" = tracing::field::Empty,
+        input_tokens = tracing::field::Empty,
+        output_tokens = tracing::field::Empty,
     );
     called(span, async {
         let answer = ask(ctx, key, &icon_request(name, words, description, candidates)).await?;
@@ -252,6 +304,8 @@ pub async fn check_key(ctx: &Ctx, key: &str) -> Result<()> {
         "http.response.status_code" = tracing::field::Empty,
         error = tracing::field::Empty,
         "otel.status_code" = tracing::field::Empty,
+        input_tokens = tracing::field::Empty,
+        output_tokens = tracing::field::Empty,
     );
     let body = serde_json::json!({
         "model": "jev-latest",
@@ -277,7 +331,15 @@ async fn ask(ctx: &Ctx, key: &str, body: &serde_json::Value) -> Result<serde_jso
         let body: String = r.text().await.unwrap_or_default().chars().take(500).collect();
         return Err(Refused { status, body }.into());
     }
-    Ok(r.json().await?)
+    let answer: serde_json::Value = r.json().await?;
+    // What a call cost, which TypeSafe says and nothing else would (#275).
+    for (n, kind) in ["input_tokens", "output_tokens"].iter().enumerate() {
+        if let Some(t) = answer["usage"][kind].as_u64() {
+            STATS.jev_tokens[n].fetch_add(t, Relaxed);
+            tracing::Span::current().record(*kind, t);
+        }
+    }
+    Ok(answer)
 }
 
 /// Runs a call to Jev in `span`, counting it in /metrics by how it ended, and marking the span
@@ -418,6 +480,21 @@ mod tests {
         assert_eq!(named(Err(anyhow::anyhow!("timed out"))), "error");
     }
 
+    #[test]
+    fn on_the_servers_key_one_item_is_one_call_for_everyone() {
+        let row = |user_id, guid: &str| Unjudged {
+            user_id, key: None, feed_id: "f".into(), guid: guid.into(), feed_title: None, title: None, description: None,
+            signals: vec![Ask { id: user_id * 10, name: "s".into(), keywords: vec![], description: None }],
+        };
+        let items = [row(1, "x"), row(2, "x"), row(1, "y")];
+        let shape = |share| calls(&items, share).iter().map(|g| g.iter().map(|i| i.user_id).collect::<Vec<_>>()).collect::<Vec<_>>();
+        assert_eq!(shape(true), [vec![1, 2], vec![1]], "x asked once, for both");
+        assert_eq!(shape(false), [vec![1], vec![2], vec![1]], "each on their own key");
+        let both: Vec<&Ask> = calls(&items, true)[0].iter().flat_map(|i| &i.signals).collect();
+        let r = request(&items[0], &both);
+        assert!(r["questions"]["signal_10"].is_object() && r["questions"]["signal_20"].is_object());
+    }
+
     #[test]
     fn a_batch_is_full_by_its_rows_not_its_items() {
         let item = |n: usize| Unjudged {
@@ -443,7 +520,7 @@ mod tests {
                 Ask { id: 4, name: "Food".into(), keywords: vec!["cooking".into()], description: Some("Recipes, not restaurants.".into()) },
             ],
         };
-        let r = request(&item);
+        let r = request(&item, &item.signals.iter().collect::<Vec<_>>());
         assert_eq!(r["state"], "Feed: Pit Lane\nTitle: Monaco\n\nQualifying report");
         let q = &r["questions"]["signal_3"];
         assert_eq!(q["type"], "noul");

crates/ipx/src/web.rs

@@ -389,6 +389,7 @@ async fn me(
         "typesafe_key_set": state.ctx.db.typesafe_key_set(user.id).await.unwrap_or(false),
         // Whose key Signals are judged on (#271): "server", for everyone, or "own", each person's.
         "signals": if crate::signals::server_key(&state.ctx).is_some() { "server" } else { "own" },
+        "signal_days": state.ctx.cfg().general.signal_days,
     }))
 }
 
@@ -2332,6 +2333,7 @@ struct Settings {
     directory_token_set: bool,
     directory_share: bool,
     signals_on_server: bool,
+    signal_days: u64,
     /// Whether the server has a TypeSafe key (`TYPESAFE_KEY`) for `signals_on_server` to use.
     typesafe_server_key: bool,
 }
@@ -2351,6 +2353,7 @@ async fn get_settings(State(state): State<WebState>) -> Json<Settings> {
         directory_token_set: !cfg.general.directory_token.trim().is_empty(),
         directory_share: cfg.general.directory_share,
         signals_on_server: cfg.general.signals_on_server,
+        signal_days: cfg.general.signal_days,
         typesafe_server_key: crate::signals::typesafe_key().is_some(),
     })
 }
@@ -2369,6 +2372,8 @@ struct SettingsPatch {
     directory_token: Option<String>,
     directory_share: Option<bool>,
     signals_on_server: Option<bool>,
+    /// 1 to 365.
+    signal_days: Option<u64>,
 }
 
 async fn patch_settings(
@@ -2421,6 +2426,14 @@ async fn patch_settings(
     if let Some(v) = body.directory_share {
         cfg.general.directory_share = v;
     }
+    // A longer window has older items to ask about; a shorter one leaves what was asked.
+    let wider = body.signal_days.is_some_and(|d| d > cfg.general.signal_days);
+    if let Some(d) = body.signal_days {
+        if !(1..=365).contains(&d) {
+            return Err(ApiError::bad_request("Signals look back between 1 and 365 days"));
+        }
+        cfg.general.signal_days = d;
+    }
     let signals_changed = body.signals_on_server.is_some_and(|v| v != cfg.general.signals_on_server);
     if let Some(v) = body.signals_on_server {
         cfg.general.signals_on_server = v;
@@ -2429,6 +2442,8 @@ async fn patch_settings(
     if signals_changed {
         // Whoever's key was refused is tried again on whichever key is now used.
         state.ctx.signals_refused.lock().unwrap().clear();
+    }
+    if signals_changed || wider {
         state.ctx.signals.notify_one();
     }
     Ok(StatusCode::NO_CONTENT)