From 1c231a46e99954f4179c60d9d9e1266d6e127257 Mon Sep 17 00:00:00 2001 From: Jon Bailey <297513015+Pitchfork-and-Torch@users.noreply.github.com> Date: Fri, 14 Aug 2026 07:43:42 -0400 Subject: [PATCH 1/3] Meritocratic For You: rank by value, not audience size (#1) * Add meritocratic author-size IPS to RankingScorer. * Make For You rank by merit, not reach. Add size-aware OON relief for small creators, origin-author diversity so viral originals cannot flood via many retweeters, and mute/block symmetry for quotes and reposts. Complements author-size IPS; docs in FEED_FAIRNESS. Co-authored-by: Jon Bailey --------- Co-authored-by: Cursor Agent Co-authored-by: Jon Bailey --- README.md | 13 +- docs/FEED_FAIRNESS.md | 65 ++++ docs/MERITOCRATIC_AUTHOR_SIZE_IPS.md | 61 ++++ .../filters/author_socialgraph_filter.rs | 82 ++++++ home-mixer/params/param.rs | 48 +++ home-mixer/scorers/author_size_ips.rs | 244 +++++++++++++++ home-mixer/scorers/mod.rs | 1 + home-mixer/scorers/ranking_scorer.rs | 278 +++++++++++++++++- 8 files changed, 773 insertions(+), 19 deletions(-) create mode 100644 docs/FEED_FAIRNESS.md create mode 100644 docs/MERITOCRATIC_AUTHOR_SIZE_IPS.md create mode 100644 home-mixer/scorers/author_size_ips.rs diff --git a/README.md b/README.md index ca947bf9..bef2de17 100644 --- a/README.md +++ b/README.md @@ -109,8 +109,8 @@ Ranking sets the order. Whether a post can be shown at all is decided separately │ ┌────────────────────────────────────────────────────────────────────────────────────┐ │ │ │ 5. SCORING │ │ │ │ PhoenixScorer a probability for each action the viewer might take │ │ -│ │ RankingScorer weighted sum, then repeated-author decay, an │ │ -│ │ out-of-network discount, a new-author boost │ │ +│ │ RankingScorer weighted sum, then author-size IPS, origin-author │ │ +│ │ diversity decay, size-aware OON discount, new-author boost │ │ │ │ VMRanker calls the reranking service in vm-ranker/ │ │ │ └────────────────────────────────────────────────────────────────────────────────────┘ │ │ ▼ │ @@ -329,10 +329,11 @@ Final Score = Σ (weight_i × P(action_i)) Positive actions carry positive weights, negative actions negative ones. The weights are in [`home-mixer/params/param.rs`](home-mixer/params/param.rs); the arithmetic is in [`home-mixer/scorers/ranking_scorer.rs`](home-mixer/scorers/ranking_scorer.rs). -Three adjustments follow: +Four adjustments follow: -- **Author Diversity**: each post after an author's first is multiplied by a decaying factor, down to a floor. -- **Out-of-Network Discount**: posts from accounts the viewer does not follow are multiplied by a factor below 1, as are replies and reposts from accounts the viewer does follow. +- **Author-size IPS**: a batch-mean-normalized inverse-propensity multiplier on `ln(1 + followers)`. Equal Phoenix scores are not ranked by audience size. Quality still wins: the default clamp is 2x, so a large-author post that is 3x more relevant still ranks higher. Math: [`docs/MERITOCRATIC_AUTHOR_SIZE_IPS.md`](docs/MERITOCRATIC_AUTHOR_SIZE_IPS.md). +- **Author Diversity**: each post after an author's first is multiplied by a decaying factor, down to a floor. By default the decay key is the *content origin* (`retweeted_user_id` when present), so a viral original cannot flood the slate through many distinct retweeters. +- **Out-of-Network Discount**: posts from accounts the viewer does not follow are multiplied by a factor below 1, as are replies and reposts from accounts the viewer does follow. Size-aware relief softens that tax for small authors (default: half relief below 1k followers, tapering to none by 100k) so discovery is not double-penalized by audience size and the follow graph. See [`docs/FEED_FAIRNESS.md`](docs/FEED_FAIRNESS.md). - **New-Author Boost**: posts from authors whose impressions are below a threshold are lifted toward a target position. `VMRanker` then calls [`vm-ranker/`](vm-ranker/), a separate service that reorders the result. @@ -356,7 +357,7 @@ Three adjustments follow: | `PreviouslySeenPostsBackupFilter` | The same, from a second record of impressions | | `PreviouslyServedPostsFilter` | Posts already served earlier in the session | | `MutedKeywordFilter` | Posts matching the viewer's muted keywords | -| `AuthorSocialgraphFilter` | Posts from accounts the viewer blocks or mutes | +| `AuthorSocialgraphFilter` | Posts from accounts the viewer blocks or mutes, including quotes/reposts of muted or blocked authors | | `VideoFilter` | Video posts, when the request excludes video | | `TopicIdsFilter` | Posts outside the requested topics, and posts in excluded topics | | `NewUserMinEngagementFilter` | For new accounts, out-of-network posts below an engagement threshold | diff --git a/docs/FEED_FAIRNESS.md b/docs/FEED_FAIRNESS.md new file mode 100644 index 00000000..2571760d --- /dev/null +++ b/docs/FEED_FAIRNESS.md @@ -0,0 +1,65 @@ +# Feed fairness: merit over reach + +This package makes For You rank by predicted value for the viewer, not by how large an author's existing audience already is — while keeping user mutes/blocks consistent and stopping viral originals from flooding a slate through many retweeters. + +It is **not** demographic parity, identity quotas, or equal impressions for every account. Low Phoenix quality stays low. Spam and predicted block/mute/report stay negative and are not lifted. + +## Changes + +### 1. Author-size IPS (ranking residual) + +Equal Phoenix scores should not be ordered by follower count. Horvitz–Thompson residual on `ln(1 + followers)`, mean-normalized inside the scored batch. + +Full math, knobs, and worked example: [`MERITOCRATIC_AUTHOR_SIZE_IPS.md`](MERITOCRATIC_AUTHOR_SIZE_IPS.md). + +Code: `home-mixer/scorers/author_size_ips.rs`. + +### 2. Size-aware out-of-network relief + +Discovery for accounts a viewer does not follow is almost entirely OON. The flat `OonWeightFactor` (default 0.75) stacked on audience-size leakage double-penalizes small creators. + +``` +oon' = base + (1 - base) * relief * t +``` + +`t` is 1 at or below `SizeAwareOonFollowerFloor` (default 1k), 0 at or above `SizeAwareOonFollowerCeiling` (default 100k), linear between. Default `relief = 0.5` → a 500-follower OON post uses 0.875 instead of 0.75. Missing follower counts keep the base tax. Large OON accounts keep the full tax. + +| Param | Default | Meaning | +|---|---|---| +| `rust_home_mixer_enable_size_aware_oon_relief` | true | Master switch | +| `rust_home_mixer_size_aware_oon_follower_floor` | 1000 | Full relief at/below | +| `rust_home_mixer_size_aware_oon_follower_ceiling` | 100000 | No relief at/above | +| `rust_home_mixer_size_aware_oon_relief` | 0.5 | Share of the gap to 1.0 | + +Code: `RankingScorer::oon_weight_for` in `home-mixer/scorers/ranking_scorer.rs`. + +### 3. Origin-author diversity + +Author diversity used `candidate.author_id`. A viral original could occupy many slate slots via distinct retweeters, each counted as a first appearance. + +With `EnableOriginAuthorDiversity` (default true), diversity counts `get_original_author_id()` (`retweeted_user_id` when present). The second and later appearances of the same original author decay. + +Quotes still diversity-key on the quoter: a quote is new content by that author. + +| Param | Default | Meaning | +|---|---|---| +| `rust_home_mixer_enable_origin_author_diversity` | true | Decay on content origin for retweets | + +### 4. Mute/block symmetry for quotes and reposts + +`AuthorSocialgraphFilter` already dropped quotes and reposts of **blocked** authors. Mute only checked the candidate author, so a quote or repost of a muted user could still serve. + +Mute now mirrors block for `quoted_user_id` and `retweeted_user_id`. Viewer preference is respected whether the muted account posts directly or is amplified by someone else. + +Code: `home-mixer/filters/author_socialgraph_filter.rs`. + +## What this does not do + +- Does not replace retrieval. A post that never enters the candidate set cannot be residualized. SidTail / `post_creation` retrieval remain the natural next step. +- Does not use UserCred / PageRank as a For You weight. +- Does not soften safety or spam drops. Visibility filtering stays separate from ranking. + +## References + +- Ashudeep Singh and Thorsten Joachims. Fairness of Exposure in Rankings. KDD 2018. https://arxiv.org/abs/1802.07281 +- Asia J. Biega, Krishna P. Gummadi, and Gerhard Weikum. Equity of Attention: Amortizing Individual Fairness in Rankings. SIGIR 2018. https://arxiv.org/abs/1805.01788 diff --git a/docs/MERITOCRATIC_AUTHOR_SIZE_IPS.md b/docs/MERITOCRATIC_AUTHOR_SIZE_IPS.md new file mode 100644 index 00000000..3b7fc6a8 --- /dev/null +++ b/docs/MERITOCRATIC_AUTHOR_SIZE_IPS.md @@ -0,0 +1,61 @@ +# Meritocratic author-size IPS + +This is a ranking residual, not a quota. + +Phoenix predicts `P(action | viewer, post)` per impression. That is the quality signal. Author follower count is not a Phoenix feature. Audience size still leaks into For You through the follow graph (Thunder), hashed author IDs, SimClusters log-favorite retrieval, and the flat 0.75 out-of-network tax. Two posts with the same predicted value for this viewer should not be ordered by how many people already follow the author. + +## What this is + +Singh and Joachims (KDD 2018), Example 3: speakers get access to willing listeners in proportion to relevance, not in proportion to prior reach. Biega, Gummadi, and Weikum (SIGIR 2018) call the same idea equity of attention: exposure tracks merit. + +Take groups of size one (every author is their own group). Disparate treatment says: + +``` +exposure(i) / merit(i) ~= exposure(j) / merit(j) +``` + +Merit here is the Phoenix weighted score. Historical show probability grows with `ln(1 + followers)`. The Horvitz-Thompson correction is `1 / ln(1 + followers)`, then mean-normalized inside the scored batch so the adjustment reallocates score instead of inflating it: + +``` +p_i = ln(1 + max(followers_i, 1)) +ips_i = 1 / p_i +m_i = 1 + alpha * (ips_i / mean(ips) - 1) +m_i = clamp(m_i, 1 / max_boost, max_boost) +score'_i = score_i * m_i if score_i > 0 + = score_i otherwise +``` + +Missing follower counts are identity (`m_i = 1`). Negative scores are not lifted. `alpha = 0` is a no-op. Default `alpha = 0.5`, `max_boost = 2.0`. + +Worked example at the defaults, authors with 100 / 10k / 1M followers: + +``` +p ~= 4.62 / 9.21 / 13.82 +ips ~= 0.217 / 0.109 / 0.072 +m ~= 1.32 / 0.91 / 0.77 +``` + +A 100-follower post and a 1M-follower post that Phoenix scored equally (pairwise) rank 1.25 / 0.75 = 1.67 apart. A 1M-follower post that Phoenix scored 3x higher still wins (3 * 0.75 > 1 * 1.25). + +## What this is not + +- Not demographic parity. There are no identity groups, no protected-class buckets, no guaranteed impression share by category. +- Not "every account gets the same impressions." Low-quality posts stay low. Spam and predicted block/mute/report stay negative and are not boosted. +- Not a replacement for retrieval. A post that never enters the candidate set cannot be residualized. SidTail (authors under 1k followers) and a SID `post_creation` window are the retrieval-side counterparts. They are not wired in this change. Ranking-side companions (size-aware OON relief, origin-author diversity, mute symmetry) are in [`FEED_FAIRNESS.md`](FEED_FAIRNESS.md). +- Not UserCred. PageRank is an enforcement prestige graph, not a For You weight. Do not add it as a rank feature. + +## Knobs + +| Param | Default | Meaning | +|---|---|---| +| `rust_home_mixer_enable_author_size_ips` | true | Master switch | +| `rust_home_mixer_author_size_ips_alpha` | 0.5 | 0 = off, 1 = full batch IPS | +| `rust_home_mixer_author_size_ips_max_boost` | 2.0 | Clamp on the multiplier | + +Code: `home-mixer/scorers/author_size_ips.rs`, applied in `RankingScorer` after the Phoenix weighted sum and before author diversity, the OON tax, and cold start. + +## References + +- Ashudeep Singh and Thorsten Joachims. Fairness of Exposure in Rankings. KDD 2018. https://arxiv.org/abs/1802.07281 +- Asia J. Biega, Krishna P. Gummadi, and Gerhard Weikum. Equity of Attention: Amortizing Individual Fairness in Rankings. SIGIR 2018. https://arxiv.org/abs/1805.01788 +- D. G. Horvitz and D. J. Thompson. A Generalization of Sampling Without Replacement From a Finite Universe. JASA 1952. diff --git a/home-mixer/filters/author_socialgraph_filter.rs b/home-mixer/filters/author_socialgraph_filter.rs index 9b04ad34..1aa37ed3 100644 --- a/home-mixer/filters/author_socialgraph_filter.rs +++ b/home-mixer/filters/author_socialgraph_filter.rs @@ -36,18 +36,28 @@ impl Filter for AuthorSocialgraphFilter { .quoted_user_id .map(|uid| viewer_blocked_user_ids.contains(&(uid as i64))) .unwrap_or(false); + let viewer_mutes_quoted_author = candidate + .quoted_user_id + .map(|uid| viewer_muted_user_ids.contains(&(uid as i64))) + .unwrap_or(false); let viewer_blocks_retweeted_user = candidate .retweeted_user_id .map(|uid| viewer_blocked_user_ids.contains(&(uid as i64))) .unwrap_or(false); + let viewer_mutes_retweeted_user = candidate + .retweeted_user_id + .map(|uid| viewer_muted_user_ids.contains(&(uid as i64))) + .unwrap_or(false); if muted || blocked || author_blocks_viewer || quoted_author_blocks_viewer || viewer_blocks_quoted_author + || viewer_mutes_quoted_author || viewer_blocks_retweeted_user + || viewer_mutes_retweeted_user { removed.push(candidate); } else { @@ -284,6 +294,78 @@ mod tests { assert_eq!(result.removed.len(), 3); } + #[tokio::test] + async fn test_muted_quoted_author_is_removed() { + let filter = AuthorSocialgraphFilter; + let user_features = UserFeatures { + muted_user_ids: vec![200], + ..Default::default() + }; + let query = make_query_with_features(user_features); + + let mut quote = make_candidate(1, 100); + quote.quoted_user_id = Some(200); + + let candidates = vec![quote, make_candidate(3, 300)]; + let result = filter.filter(&query, candidates); + + assert_eq!(result.kept.len(), 1); + assert_eq!(result.kept[0].author_id, 300); + assert_eq!(result.removed.len(), 1); + assert_eq!(result.removed[0].quoted_user_id, Some(200)); + } + + #[tokio::test] + async fn test_unmuted_quoted_author_is_kept() { + let filter = AuthorSocialgraphFilter; + let query = make_query_with_features(UserFeatures::default()); + + let mut quote = make_candidate(1, 100); + quote.quoted_user_id = Some(200); + + let result = filter.filter(&query, vec![quote]); + + assert_eq!(result.kept.len(), 1); + assert_eq!(result.kept[0].quoted_user_id, Some(200)); + assert!(result.removed.is_empty()); + } + + #[tokio::test] + async fn test_muted_retweeted_author_is_removed() { + let filter = AuthorSocialgraphFilter; + let user_features = UserFeatures { + muted_user_ids: vec![200], + ..Default::default() + }; + let query = make_query_with_features(user_features); + + let mut retweet = make_candidate(1, 100); + retweet.retweeted_user_id = Some(200); + + let candidates = vec![retweet, make_candidate(3, 300)]; + let result = filter.filter(&query, candidates); + + assert_eq!(result.kept.len(), 1); + assert_eq!(result.kept[0].author_id, 300); + assert_eq!(result.removed.len(), 1); + assert_eq!(result.removed[0].retweeted_user_id, Some(200)); + } + + #[tokio::test] + async fn test_unmuted_retweeted_author_is_kept() { + let filter = AuthorSocialgraphFilter; + let query = make_query_with_features(UserFeatures::default()); + + let mut retweet = make_candidate(1, 100); + retweet.retweeted_user_id = Some(200); + + let result = filter.filter(&query, vec![retweet]); + + assert_eq!(result.kept.len(), 1); + assert_eq!(result.kept[0].retweeted_user_id, Some(200)); + assert!(result.removed.is_empty()); + } + #[tokio::test] async fn test_author_blocks_viewer() { let filter = AuthorSocialgraphFilter; diff --git a/home-mixer/params/param.rs b/home-mixer/params/param.rs index dc3a12e4..3852c72f 100644 --- a/home-mixer/params/param.rs +++ b/home-mixer/params/param.rs @@ -237,6 +237,30 @@ param!( "rust_home_mixer_author_diversity_floor", 0.25 ); +param!( + EnableAuthorSizeIps, + bool, + "rust_home_mixer_enable_author_size_ips", + true +); +param!( + AuthorSizeIpsAlpha, + f64, + "rust_home_mixer_author_size_ips_alpha", + 0.5 +); +param!( + AuthorSizeIpsMaxBoost, + f64, + "rust_home_mixer_author_size_ips_max_boost", + 2.0 +); +param!( + EnableOriginAuthorDiversity, + bool, + "rust_home_mixer_enable_origin_author_diversity", + true +); param!( LogSlateContext, bool, @@ -249,6 +273,30 @@ param!( "rust_home_mixer_oon_weight_factor", 0.75 ); +param!( + EnableSizeAwareOonRelief, + bool, + "rust_home_mixer_enable_size_aware_oon_relief", + true +); +param!( + SizeAwareOonFollowerFloor, + i64, + "rust_home_mixer_size_aware_oon_follower_floor", + 1000 +); +param!( + SizeAwareOonFollowerCeiling, + i64, + "rust_home_mixer_size_aware_oon_follower_ceiling", + 100000 +); +param!( + SizeAwareOonRelief, + f64, + "rust_home_mixer_size_aware_oon_relief", + 0.5 +); param!( EnableMpnScoring, diff --git a/home-mixer/scorers/author_size_ips.rs b/home-mixer/scorers/author_size_ips.rs new file mode 100644 index 00000000..37c3cc47 --- /dev/null +++ b/home-mixer/scorers/author_size_ips.rs @@ -0,0 +1,244 @@ +use crate::models::candidate::PostCandidate; +use crate::models::query::ScoredPostsQuery; +use crate::params::{AuthorSizeIpsAlpha, AuthorSizeIpsMaxBoost, EnableAuthorSizeIps}; + +/// Horvitz-Thompson residual of author audience size. +/// +/// Phoenix already estimates P(action | viewer, post). Follower count is not a +/// Phoenix feature, but size still leaks into the slate via the follow graph, +/// hashed author IDs, SimClusters log-fav retrieval, and the flat OON tax. +/// This multiplier residualizes ln(1 + followers) inside the scored batch so +/// two posts with the same Phoenix score are not ranked by author size. +/// +/// p_i = ln(1 + max(followers_i, 1)) +/// ips_i = 1 / p_i +/// m_i = 1 + alpha * (ips_i / mean(ips) - 1) +/// m_i = clamp(m_i, 1 / max_boost, max_boost) +/// +/// Mean-normalization keeps the batch arithmetic mean at 1: this reallocates +/// score, it does not inflate it. Quality still wins: a 3x Phoenix gap beats +/// the default 2x clamp. Missing follower counts are identity (multiplier 1). +/// Negative scores are not boosted. +/// +/// This is individual meritocratic fairness (Singh and Joachims 2018, groups +/// of size one; Biega et al. equity of attention). It is not a demographic +/// quota. +pub(crate) fn multipliers_for(query: &ScoredPostsQuery, candidates: &[PostCandidate]) -> Vec { + if !query.params.get(EnableAuthorSizeIps) { + return vec![1.0; candidates.len()]; + } + multipliers( + candidates.iter().map(|c| c.author_followers_count), + query.params.get(AuthorSizeIpsAlpha), + query.params.get(AuthorSizeIpsMaxBoost), + ) +} + +pub(crate) fn apply( + query: &ScoredPostsQuery, + candidates: &[PostCandidate], + scores: &[f64], +) -> Vec { + let multipliers = multipliers_for(query, candidates); + scores + .iter() + .zip(multipliers) + .map(|(&score, multiplier)| { + if score > 0.0 && multiplier.is_finite() { + score * multiplier + } else { + score + } + }) + .collect() +} + +pub(crate) fn multipliers( + followers: impl IntoIterator>, + alpha: f64, + max_boost: f64, +) -> Vec { + let followers: Vec> = followers.into_iter().collect(); + let n = followers.len(); + if n == 0 || !alpha.is_finite() || alpha <= 0.0 { + return vec![1.0; n]; + } + + let cap = if max_boost.is_finite() && max_boost >= 1.0 { + max_boost + } else { + 1.0 + }; + let floor = 1.0 / cap; + + let mut ips = vec![None; n]; + let mut ips_sum = 0.0; + let mut known = 0usize; + for (i, follower_count) in followers.iter().enumerate() { + let Some(raw) = follower_count else { + continue; + }; + let propensity = (1.0 + f64::from((*raw).max(1))).ln(); + if !propensity.is_finite() || propensity <= 0.0 { + continue; + } + let weight = 1.0 / propensity; + if !weight.is_finite() || weight <= 0.0 { + continue; + } + ips[i] = Some(weight); + ips_sum += weight; + known += 1; + } + + if known == 0 { + return vec![1.0; n]; + } + + let mean_ips = ips_sum / known as f64; + if !mean_ips.is_finite() || mean_ips <= 0.0 { + return vec![1.0; n]; + } + + ips.into_iter() + .map(|weight| match weight { + Some(weight) => { + let raw = 1.0 + alpha * (weight / mean_ips - 1.0); + if raw.is_finite() { + raw.clamp(floor, cap) + } else { + 1.0 + } + } + None => 1.0, + }) + .collect() +} + +#[cfg(test)] +mod tests { + use super::*; + + fn approx(a: f64, b: f64) { + assert!((a - b).abs() < 1e-9, "{a} != {b}"); + } + + #[test] + fn identity_when_alpha_is_zero() { + let m = multipliers([Some(100), Some(1_000_000)], 0.0, 2.0); + assert_eq!(m, vec![1.0, 1.0]); + } + + #[test] + fn identity_when_followers_missing() { + let m = multipliers([None, None], 0.5, 2.0); + assert_eq!(m, vec![1.0, 1.0]); + } + + #[test] + fn identity_when_all_authors_same_size() { + let m = multipliers([Some(12_000), Some(12_000), Some(12_000)], 0.5, 2.0); + for value in m { + approx(value, 1.0); + } + } + + #[test] + fn missing_followers_stay_identity_among_sized_authors() { + let m = multipliers([Some(100), None, Some(1_000_000)], 0.5, 2.0); + assert!(m[0] > 1.0, "small author should lift: {}", m[0]); + approx(m[1], 1.0); + assert!(m[2] < 1.0, "large author should recede: {}", m[2]); + } + + #[test] + fn smaller_author_gets_higher_multiplier() { + let m = multipliers([Some(100), Some(10_000), Some(1_000_000)], 0.5, 2.0); + assert!(m[0] > m[1] && m[1] > m[2], "{m:?}"); + assert!(m[0] > 1.0 && m[2] < 1.0, "{m:?}"); + } + + #[test] + fn mean_of_known_multipliers_is_one() { + let m = multipliers([Some(50), Some(5_000), Some(500_000)], 0.5, 8.0); + let mean = m.iter().sum::() / m.len() as f64; + approx(mean, 1.0); + } + + #[test] + fn clamp_respects_max_boost() { + let m = multipliers([Some(1), Some(2_000_000_000)], 8.0, 1.5); + assert!(m[0] <= 1.5 + 1e-12, "{}", m[0]); + assert!(m[1] >= 1.0 / 1.5 - 1e-12, "{}", m[1]); + } + + #[test] + fn zero_followers_treated_as_one() { + let a = multipliers([Some(0), Some(10_000)], 0.5, 2.0); + let b = multipliers([Some(1), Some(10_000)], 0.5, 2.0); + approx(a[0], b[0]); + approx(a[1], b[1]); + } + + #[test] + fn apply_does_not_boost_non_positive_scores() { + let mut query = ScoredPostsQuery::default(); + let fs = xai_feature_switches::FeatureSwitches::new(vec![]).unwrap(); + let mut results = + fs.match_recipient(&xai_feature_switches::RecipientBuilder::new().build()); + results.override_fs("rust_home_mixer_enable_author_size_ips".into(), "true"); + results.override_fs("rust_home_mixer_author_size_ips_alpha".into(), "1.0"); + query.params = results.into(); + + let candidates = vec![ + PostCandidate { + author_followers_count: Some(10), + ..Default::default() + }, + PostCandidate { + author_followers_count: Some(1_000_000), + ..Default::default() + }, + ]; + let out = apply(&query, &candidates, &[-2.0, 0.0]); + assert_eq!(out, vec![-2.0, 0.0]); + } + + #[test] + fn apply_lifts_small_author_on_positive_score() { + let mut query = ScoredPostsQuery::default(); + let fs = xai_feature_switches::FeatureSwitches::new(vec![]).unwrap(); + let mut results = + fs.match_recipient(&xai_feature_switches::RecipientBuilder::new().build()); + results.override_fs("rust_home_mixer_enable_author_size_ips".into(), "true"); + results.override_fs("rust_home_mixer_author_size_ips_alpha".into(), "0.5"); + results.override_fs("rust_home_mixer_author_size_ips_max_boost".into(), "2.0"); + query.params = results.into(); + + let candidates = vec![ + PostCandidate { + author_followers_count: Some(100), + ..Default::default() + }, + PostCandidate { + author_followers_count: Some(1_000_000), + ..Default::default() + }, + ]; + let out = apply(&query, &candidates, &[10.0, 10.0]); + assert!(out[0] > out[1], "{out:?}"); + assert!(out[0] > 10.0 && out[1] < 10.0, "{out:?}"); + } + + #[test] + fn equal_phoenix_gap_still_loses_to_large_quality() { + // Default clamp is 2x. A 3x Phoenix advantage must still win. + let m = multipliers([Some(100), Some(1_000_000)], 0.5, 2.0); + let small = 1.0 * m[0]; + let large = 3.0 * m[1]; + assert!( + large > small, + "quality must still dominate: {small} vs {large}" + ); + } +} diff --git a/home-mixer/scorers/mod.rs b/home-mixer/scorers/mod.rs index ed0eed1c..60c86157 100644 --- a/home-mixer/scorers/mod.rs +++ b/home-mixer/scorers/mod.rs @@ -1,4 +1,5 @@ pub mod author_cold_start; +pub mod author_size_ips; pub mod phoenix_scorer; pub mod phoenix_scores_ranking_scorer; pub mod ranking_scorer; diff --git a/home-mixer/scorers/ranking_scorer.rs b/home-mixer/scorers/ranking_scorer.rs index 6dfcb7aa..e411e1d0 100644 --- a/home-mixer/scorers/ranking_scorer.rs +++ b/home-mixer/scorers/ranking_scorer.rs @@ -1,7 +1,10 @@ -use crate::models::candidate::{MpnParts, PhoenixScores, PostCandidate, SlateContext}; +use crate::models::candidate::{ + CandidateHelpers, MpnParts, PhoenixScores, PostCandidate, SlateContext, +}; use crate::models::query::ScoredPostsQuery; use crate::params::*; use crate::scorers::author_cold_start::AuthorColdStart; +use crate::scorers::author_size_ips; use crate::scorers::value_model_gate::GateModel; use rustc_hash::FxHashMap; use std::cmp::Ordering; @@ -616,6 +619,7 @@ impl RankingScorer { } fn compute_slate_contexts( + query: &ScoredPostsQuery, candidates: &[PostCandidate], pre_diversity_scores: &[f64], ) -> Vec { @@ -626,12 +630,19 @@ impl RankingScorer { .collect(); indexed.sort_by(|(_, a), (_, b)| b.partial_cmp(a).unwrap_or(Ordering::Equal)); + let use_origin = query.params.get(EnableOriginAuthorDiversity); let mut contexts = vec![SlateContext::default(); candidates.len()]; let mut author_counts: FxHashMap = FxHashMap::default(); let mut last_author_rank: FxHashMap = FxHashMap::default(); for (rank, (idx, score)) in indexed.into_iter().enumerate() { let rank = rank as u32; - let author_id = candidates[idx].author_id; + // Count diversity against the content origin so a viral original + // cannot flood the slate via many distinct retweeters. + let author_id = if use_origin { + candidates[idx].get_original_author_id() + } else { + candidates[idx].author_id + }; let k = author_counts.get(&author_id).copied().unwrap_or(0); let rank_gap = last_author_rank.get(&author_id).map(|last| rank - last); contexts[idx] = SlateContext { @@ -698,6 +709,44 @@ impl RankingScorer { oon_weight_factor } } + + /// Soften the flat OON tax for small authors. + /// + /// Nearly all discovery for accounts outside a viewer's follow graph is + /// OON. Stacking a flat 0.75 tax on top of audience-size leakage makes + /// equal Phoenix quality lose to large accounts twice. Relief interpolates + /// from `SizeAwareOonRelief` (followers <= floor) down to 0 (followers >= + /// ceiling): `oon' = base + (1 - base) * relief * t`. + fn oon_weight_for(query: &ScoredPostsQuery, candidate: &PostCandidate, base: f64) -> f64 { + if !query.params.get(EnableSizeAwareOonRelief) { + return base; + } + let Some(followers) = candidate.author_followers_count else { + return base; + }; + let floor = query.params.get(SizeAwareOonFollowerFloor).max(0); + let ceiling = query.params.get(SizeAwareOonFollowerCeiling).max(floor + 1); + let relief = query.params.get(SizeAwareOonRelief).clamp(0.0, 1.0); + if !relief.is_finite() || relief <= 0.0 { + return base; + } + + let followers = i64::from(followers.max(0)); + let t = if followers <= floor { + 1.0 + } else if followers >= ceiling { + 0.0 + } else { + let span = (ceiling - floor) as f64; + 1.0 - (followers - floor) as f64 / span + }; + let adjusted = base + (1.0 - base) * relief * t; + if adjusted.is_finite() { + adjusted.clamp(base.min(1.0), 1.0) + } else { + base + } + } } #[async_trait] @@ -738,6 +787,9 @@ impl Scorer for RankingScorer { .collect() }; + let ips_multipliers = author_size_ips::multipliers_for(query, candidates); + let size_adjusted_scores = author_size_ips::apply(query, candidates, &weighted_scores); + let mpn_scoring = query.params.get(EnableMpnScoring) && !use_dwell_regret; let effective_oon = Self::effective_oon_weight(query); @@ -757,7 +809,11 @@ impl Scorer for RankingScorer { let persisted_contexts: Option> = if query.has_cached_posts { Self::stored_slate_contexts(candidates) } else { - Some(Self::compute_slate_contexts(candidates, &weighted_scores)) + Some(Self::compute_slate_contexts( + query, + candidates, + &size_adjusted_scores, + )) }; let diversity_multipliers: Vec = if enable_author_diversity { @@ -765,8 +821,11 @@ impl Scorer for RankingScorer { let scoring_contexts: &[SlateContext] = match &persisted_contexts { Some(contexts) if !query.has_cached_posts => contexts, _ => { - recomputed_contexts = - Self::compute_slate_contexts(candidates, &weighted_scores); + recomputed_contexts = Self::compute_slate_contexts( + query, + candidates, + &size_adjusted_scores, + ); &recomputed_contexts } }; @@ -779,9 +838,9 @@ impl Scorer for RankingScorer { .iter() .enumerate() .map(|(i, c)| { - let mut m = diversity_multipliers[i]; + let mut m = diversity_multipliers[i] * ips_multipliers[i]; if oon_applies(c) { - m *= effective_oon; + m *= Self::oon_weight_for(query, c, effective_oon); } m }) @@ -819,14 +878,18 @@ impl Scorer for RankingScorer { .collect(); } - let adjusted_scores = self - .author_cold_start - .apply(query, candidates, &weighted_scores); + let adjusted_scores = + self.author_cold_start + .apply(query, candidates, &size_adjusted_scores); let persisted_contexts: Option> = if query.has_cached_posts { Self::stored_slate_contexts(candidates) } else { - Some(Self::compute_slate_contexts(candidates, &adjusted_scores)) + Some(Self::compute_slate_contexts( + query, + candidates, + &adjusted_scores, + )) }; let diversity_adjusted = if enable_author_diversity { @@ -835,7 +898,7 @@ impl Scorer for RankingScorer { Some(contexts) if !query.has_cached_posts => contexts, _ => { recomputed_contexts = - Self::compute_slate_contexts(candidates, &adjusted_scores); + Self::compute_slate_contexts(query, candidates, &adjusted_scores); &recomputed_contexts } }; @@ -850,7 +913,7 @@ impl Scorer for RankingScorer { .map(|(i, c)| { let after_diversity = diversity_adjusted[i]; if oon_applies(c) { - after_diversity * effective_oon + after_diversity * Self::oon_weight_for(query, c, effective_oon) } else { after_diversity } @@ -1044,6 +1107,195 @@ mod tests { assert!((oon_score - in_network_score * 0.75).abs() < 1e-9); } + fn candidate_with_followers( + author_id: u64, + in_network: Option, + followers: i32, + ) -> PostCandidate { + PostCandidate { + author_id, + in_network, + author_followers_count: Some(followers), + ..Default::default() + } + } + + #[tokio::test] + async fn applies_author_size_ips_to_equal_phoenix_scores() { + let scorer = test_scorer(); + let candidates = vec![ + candidate_with_followers(1, Some(true), 100), + candidate_with_followers(2, Some(true), 1_000_000), + ]; + let query = query_with_flags(&[ + ("rust_home_mixer_enable_author_size_ips", "true"), + ("rust_home_mixer_author_size_ips_alpha", "0.5"), + ("rust_home_mixer_enable_author_diversity", "false"), + ("rust_home_mixer_value_model_mode", "weighted"), + ("rust_home_mixer_enable_mpn_scoring", "false"), + ]); + let scored = scorer.score(&query, &candidates).await; + let small = scored[0].as_ref().unwrap().score.unwrap(); + let large = scored[1].as_ref().unwrap().score.unwrap(); + assert!(small > large, "small={small} large={large}"); + let weighted_small = scored[0].as_ref().unwrap().weighted_score.unwrap(); + let weighted_large = scored[1].as_ref().unwrap().weighted_score.unwrap(); + assert!((weighted_small - weighted_large).abs() < 1e-9); + } + + #[tokio::test] + async fn author_size_ips_can_be_disabled() { + let scorer = test_scorer(); + let candidates = vec![ + candidate_with_followers(1, Some(true), 100), + candidate_with_followers(2, Some(true), 1_000_000), + ]; + let query = query_with_flags(&[ + ("rust_home_mixer_enable_author_size_ips", "false"), + ("rust_home_mixer_enable_author_diversity", "false"), + ("rust_home_mixer_value_model_mode", "weighted"), + ("rust_home_mixer_enable_mpn_scoring", "false"), + ]); + let scored = scorer.score(&query, &candidates).await; + let small = scored[0].as_ref().unwrap().score.unwrap(); + let large = scored[1].as_ref().unwrap().score.unwrap(); + assert!((small - large).abs() < 1e-9, "small={small} large={large}"); + } + + #[test] + fn size_aware_oon_softens_tax_for_small_authors() { + let query = query_with_flags(&[ + ("rust_home_mixer_enable_size_aware_oon_relief", "true"), + ("rust_home_mixer_size_aware_oon_follower_floor", "1000"), + ("rust_home_mixer_size_aware_oon_follower_ceiling", "100000"), + ("rust_home_mixer_size_aware_oon_relief", "0.5"), + ("rust_home_mixer_oon_weight_factor", "0.75"), + ]); + let base = 0.75; + let small = candidate_with_followers(1, Some(false), 500); + let mid = candidate_with_followers(2, Some(false), 50_500); + let large = candidate_with_followers(3, Some(false), 500_000); + let missing = PostCandidate { + author_id: 4, + in_network: Some(false), + author_followers_count: None, + ..Default::default() + }; + + let small_w = RankingScorer::oon_weight_for(&query, &small, base); + let mid_w = RankingScorer::oon_weight_for(&query, &mid, base); + let large_w = RankingScorer::oon_weight_for(&query, &large, base); + let missing_w = RankingScorer::oon_weight_for(&query, &missing, base); + + // floor: 0.75 + 0.25 * 0.5 * 1.0 = 0.875 + assert!((small_w - 0.875).abs() < 1e-9, "{small_w}"); + // midpoint of [1000, 100000]: t = 0.5 → 0.75 + 0.25 * 0.5 * 0.5 = 0.8125 + assert!((mid_w - 0.8125).abs() < 1e-9, "{mid_w}"); + assert!((large_w - base).abs() < 1e-9, "{large_w}"); + assert!((missing_w - base).abs() < 1e-9, "{missing_w}"); + assert!(small_w > mid_w && mid_w > large_w); + } + + #[tokio::test] + async fn size_aware_oon_lifts_small_oon_relative_to_large_oon() { + let scorer = test_scorer(); + let candidates = vec![ + candidate_with_followers(1, Some(false), 100), + candidate_with_followers(2, Some(false), 1_000_000), + ]; + let query = query_with_flags(&[ + ("rust_home_mixer_enable_author_size_ips", "false"), + ("rust_home_mixer_enable_author_diversity", "false"), + ("rust_home_mixer_enable_size_aware_oon_relief", "true"), + ("rust_home_mixer_size_aware_oon_relief", "0.5"), + ("rust_home_mixer_oon_weight_factor", "0.75"), + ("rust_home_mixer_value_model_mode", "weighted"), + ("rust_home_mixer_enable_mpn_scoring", "false"), + ]); + let scored = scorer.score(&query, &candidates).await; + let small = scored[0].as_ref().unwrap().score.unwrap(); + let large = scored[1].as_ref().unwrap().score.unwrap(); + assert!(small > large, "small={small} large={large}"); + } + + #[tokio::test] + async fn origin_author_diversity_decays_retweets_of_same_original() { + let scorer = test_scorer(); + let first = PostCandidate { + tweet_id: 10, + author_id: 100, + retweeted_user_id: Some(999), + retweeted_tweet_id: Some(1), + in_network: Some(true), + ..Default::default() + }; + let second = PostCandidate { + tweet_id: 11, + author_id: 200, + retweeted_user_id: Some(999), + retweeted_tweet_id: Some(1), + in_network: Some(true), + ..Default::default() + }; + let other = candidate(300, Some(true)); + + let query = query_with_flags(&[ + ("rust_home_mixer_enable_author_diversity", "true"), + ("rust_home_mixer_enable_origin_author_diversity", "true"), + ("rust_home_mixer_author_diversity_decay", "0.5"), + ("rust_home_mixer_author_diversity_floor", "0.25"), + ("rust_home_mixer_enable_author_size_ips", "false"), + ("rust_home_mixer_enable_oon_rescore_for_in_network_replies_retweets", "false"), + ("rust_home_mixer_value_model_mode", "weighted"), + ("rust_home_mixer_enable_mpn_scoring", "false"), + ]); + let scored = scorer.score(&query, &[first, second, other]).await; + let a = scored[0].as_ref().unwrap().score.unwrap(); + let b = scored[1].as_ref().unwrap().score.unwrap(); + let c = scored[2].as_ref().unwrap().score.unwrap(); + let expected = RankingScorer::diversity_multiplier(0.5, 0.25, 1.0); + // first RT and unrelated author share top score; second RT of same origin decays + assert!((a - c).abs() < 1e-9, "a={a} c={c}"); + assert!((b - a * expected).abs() < 1e-9, "b={b} expected={}", a * expected); + } + + #[tokio::test] + async fn origin_author_diversity_can_be_disabled() { + let scorer = test_scorer(); + let first = PostCandidate { + tweet_id: 10, + author_id: 100, + retweeted_user_id: Some(999), + retweeted_tweet_id: Some(1), + in_network: Some(true), + ..Default::default() + }; + let second = PostCandidate { + tweet_id: 11, + author_id: 200, + retweeted_user_id: Some(999), + retweeted_tweet_id: Some(1), + in_network: Some(true), + ..Default::default() + }; + + let query = query_with_flags(&[ + ("rust_home_mixer_enable_author_diversity", "true"), + ("rust_home_mixer_enable_origin_author_diversity", "false"), + ("rust_home_mixer_author_diversity_decay", "0.5"), + ("rust_home_mixer_author_diversity_floor", "0.25"), + ("rust_home_mixer_enable_author_size_ips", "false"), + ("rust_home_mixer_enable_oon_rescore_for_in_network_replies_retweets", "false"), + ("rust_home_mixer_value_model_mode", "weighted"), + ("rust_home_mixer_enable_mpn_scoring", "false"), + ]); + let scored = scorer.score(&query, &[first, second]).await; + let a = scored[0].as_ref().unwrap().score.unwrap(); + let b = scored[1].as_ref().unwrap().score.unwrap(); + // Distinct retweeters: without origin diversity both keep full score. + assert!((a - b).abs() < 1e-9, "a={a} b={b}"); + } + #[test] fn video_open_head_is_weighted_into_score() { let zero_query = query_with_flags(&[("rust_home_mixer_video_open_weight", "0.0")]); From 34015e90077b02ee32ae42c96c5bd9fe17f64c95 Mon Sep 17 00:00:00 2001 From: Jon Bailey <297513015+Pitchfork-and-Torch@users.noreply.github.com> Date: Thu, 3 Sep 2026 23:32:03 -0700 Subject: [PATCH 2/3] home-mixer: keep VF verdicts separate per safety level in VFCandidateHydrator (#6) VFCandidateHydrator asks visibility filtering twice per request: once at TimelineHome for in-network candidates (plus repost sources) and once at TimelineHomeRecommendations for out-of-network candidates (plus ancestors and quoted posts). It then merged both answers into one HashMap keyed by tweet id, with the recommendations map applied last. A tweet id can be in both sets. The common case is a followed author's own thread: the root post is an in-network candidate, and the reply in the same thread lists the root as an ancestor. The same happens whenever another selected candidate quotes or replies to an in-network post. In every such case the recommendations verdict overwrote the in-network verdict, so the in-network post was judged under the rules that are meant to apply only to recommendations from accounts the viewer does not follow (SpamHighRecall, NsfwHighRecall, DoNotAmplify, NsfwText, FosnrAbuseInsults, the NSFW author/tweet flags, DMCA and geo-restricted media, and the OON-only user labels in visibility-filtering/rules/registry.rs). VFFilter then removed the post from the viewer's For You feed even though the viewer follows the author and README.md states that "the same post is allowed to a follower". The same collision runs the other way for an out-of-network candidate that is also the source of a followed account's repost: the merge order decides which verdict wins, and neither order is right for both cases. Keep the two result maps separate and route every lookup to the map matching how the id was requested: a candidate's own verdict comes from the map for its in_network flag; ancestors and quoted posts read the recommendations map; repost sources read the in-network map. No VF rule changes and no extra RPCs. Tests cover both collision directions, the ancillary routing, tombstoned ancestors, interstitials, error propagation, and an end-to-end hydrate() run with a client that answers Allow at TimelineHome and Drop at TimelineHomeRecommendations. The end-to-end test fails on the previous code with the root post carrying the recommendations-only drop reason. Co-authored-by: Cursor Agent Co-authored-by: Jon Bailey --- .../vf_candidate_hydrator.rs | 342 ++++++++++++++++-- 1 file changed, 312 insertions(+), 30 deletions(-) diff --git a/home-mixer/candidate_hydrators/vf_candidate_hydrator.rs b/home-mixer/candidate_hydrators/vf_candidate_hydrator.rs index 24cca235..1ccaa360 100644 --- a/home-mixer/candidate_hydrators/vf_candidate_hydrator.rs +++ b/home-mixer/candidate_hydrators/vf_candidate_hydrator.rs @@ -103,31 +103,15 @@ impl Hydrator for VFCandidateHydrator { ); let (in_network_result, oon_result) = join(in_network_future, oon_future).await; - let mut all_results: HashMap>> = HashMap::new(); - all_results.extend(in_network_result); - all_results.extend(oon_result); - - let mut hydrated_candidates = Vec::with_capacity(candidates.len()); - for candidate in candidates { - let primary_result = all_results.get(&candidate.tweet_id); - let visibility_reason = match primary_result { - Some(Ok(Some(reason))) => Some(reason.clone()), - _ => None, - }; - - let drop_ancillary = should_drop_ancillary(candidate, &all_results); + let verdicts = VfVerdicts { + in_network: in_network_result, + oon: oon_result, + }; - let hydrated = match primary_result { - Some(Err(err)) => Err(err.to_string()), - _ => Ok(PostCandidate { - visibility_reason, - drop_ancillary_posts: Some(drop_ancillary), - ..Default::default() - }), - }; - hydrated_candidates.push(hydrated); - } - hydrated_candidates + candidates + .iter() + .map(|candidate| resolve_visibility(candidate, &verdicts)) + .collect() } fn update(&self, candidate: &mut PostCandidate, hydrated: PostCandidate) { @@ -136,15 +120,60 @@ impl Hydrator for VFCandidateHydrator { } } -fn should_drop_ancillary( +type VfResults = HashMap>>; + +/// VF verdicts kept separate by the safety level each id was evaluated under. +/// +/// The same tweet id can legitimately be requested at both levels: an +/// in-network candidate may also be an ancestor or quoted post of another +/// candidate (evaluated as a recommendation), and an out-of-network candidate +/// may also be the source of a followed account's repost (evaluated as +/// in-network). Collapsing the two maps into one keyed by id alone lets one +/// verdict overwrite the other, so every lookup must go to the map that +/// matches how the id was bucketed above. +struct VfVerdicts { + in_network: VfResults, + oon: VfResults, +} + +impl VfVerdicts { + fn primary(&self, candidate: &PostCandidate) -> Option<&Result>> { + if candidate.in_network.unwrap_or(false) { + self.in_network.get(&candidate.tweet_id) + } else { + self.oon.get(&candidate.tweet_id) + } + } +} + +fn resolve_visibility( candidate: &PostCandidate, - vf_results: &HashMap>>, -) -> bool { + verdicts: &VfVerdicts, +) -> Result { + let primary_result = verdicts.primary(candidate); + let visibility_reason = match primary_result { + Some(Ok(Some(reason))) => Some(reason.clone()), + _ => None, + }; + + let drop_ancillary = should_drop_ancillary(candidate, verdicts); + + match primary_result { + Some(Err(err)) => Err(err.to_string()), + _ => Ok(PostCandidate { + visibility_reason, + drop_ancillary_posts: Some(drop_ancillary), + ..Default::default() + }), + } +} + +fn should_drop_ancillary(candidate: &PostCandidate, verdicts: &VfVerdicts) -> bool { for &ancestor_id in &candidate.ancestors { if candidate.tombstone_ancestor_ids.contains(&ancestor_id) { continue; } - if let Some(Ok(Some(reason))) = vf_results.get(&ancestor_id) + if let Some(Ok(Some(reason))) = verdicts.oon.get(&ancestor_id) && should_drop_reason(reason) { return true; @@ -152,14 +181,14 @@ fn should_drop_ancillary( } if let Some(quoted_id) = candidate.quoted_tweet_id - && let Some(Ok(Some(reason))) = vf_results.get("ed_id) + && let Some(Ok(Some(reason))) = verdicts.oon.get("ed_id) && should_drop_reason(reason) { return true; } if let Some(retweeted_id) = candidate.retweeted_tweet_id - && let Some(Ok(Some(reason))) = vf_results.get(&retweeted_id) + && let Some(Ok(Some(reason))) = verdicts.in_network.get(&retweeted_id) && should_drop_reason(reason) { return true; @@ -176,3 +205,256 @@ fn should_drop_reason(reason: &FilteredReason) -> bool { _ => true, } } + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::Mutex; + use xai_visibility_filtering::models::SafetyResult; + + fn oon_only_drop() -> FilteredReason { + FilteredReason::PossiblyUndesirable + } + + fn interstitial() -> FilteredReason { + FilteredReason::SafetyResult(SafetyResult { + reason: None, + action: Action::Interstitial, + }) + } + + fn verdicts( + in_network: Vec<(u64, Result>)>, + oon: Vec<(u64, Result>)>, + ) -> VfVerdicts { + VfVerdicts { + in_network: in_network.into_iter().collect(), + oon: oon.into_iter().collect(), + } + } + + #[test] + fn in_network_candidate_reads_timeline_home_verdict_when_id_is_also_an_oon_ancillary() { + let verdicts = verdicts( + vec![(1, Ok(None))], + vec![(1, Ok(Some(oon_only_drop()))), (2, Ok(None))], + ); + let in_network_post = PostCandidate { + tweet_id: 1, + in_network: Some(true), + ..Default::default() + }; + + let hydrated = resolve_visibility(&in_network_post, &verdicts).unwrap(); + + assert_eq!(hydrated.visibility_reason, None); + assert_eq!(hydrated.drop_ancillary_posts, Some(false)); + } + + #[test] + fn oon_candidate_reads_recommendations_verdict_when_id_is_also_a_repost_source() { + let verdicts = verdicts(vec![(1, Ok(None))], vec![(1, Ok(Some(oon_only_drop())))]); + let oon_post = PostCandidate { + tweet_id: 1, + in_network: Some(false), + ..Default::default() + }; + + let hydrated = resolve_visibility(&oon_post, &verdicts).unwrap(); + + assert_eq!(hydrated.visibility_reason, Some(oon_only_drop())); + } + + #[test] + fn missing_in_network_flag_is_treated_as_oon() { + let verdicts = verdicts(vec![(1, Ok(None))], vec![(1, Ok(Some(oon_only_drop())))]); + let post = PostCandidate { + tweet_id: 1, + in_network: None, + ..Default::default() + }; + + let hydrated = resolve_visibility(&post, &verdicts).unwrap(); + + assert_eq!(hydrated.visibility_reason, Some(oon_only_drop())); + } + + #[test] + fn ancestors_and_quoted_posts_use_recommendations_verdict() { + let verdicts = verdicts( + vec![(10, Ok(None)), (20, Ok(None))], + vec![ + (10, Ok(Some(oon_only_drop()))), + (20, Ok(Some(oon_only_drop()))), + ], + ); + + let reply = PostCandidate { + tweet_id: 1, + in_network: Some(true), + ancestors: vec![10], + ..Default::default() + }; + assert!(should_drop_ancillary(&reply, &verdicts)); + + let quote = PostCandidate { + tweet_id: 2, + in_network: Some(true), + quoted_tweet_id: Some(20), + ..Default::default() + }; + assert!(should_drop_ancillary("e, &verdicts)); + } + + #[test] + fn repost_source_uses_timeline_home_verdict() { + let verdicts = verdicts(vec![(10, Ok(None))], vec![(10, Ok(Some(oon_only_drop())))]); + let repost = PostCandidate { + tweet_id: 1, + in_network: Some(true), + retweeted_tweet_id: Some(10), + ..Default::default() + }; + + assert!(!should_drop_ancillary(&repost, &verdicts)); + + let source_dropped_in_network = VfVerdicts { + in_network: HashMap::from([(10, Ok(Some(oon_only_drop())))]), + oon: HashMap::new(), + }; + assert!(should_drop_ancillary(&repost, &source_dropped_in_network)); + } + + #[test] + fn tombstoned_ancestors_are_skipped() { + let verdicts = verdicts(vec![], vec![(10, Ok(Some(oon_only_drop())))]); + let reply = PostCandidate { + tweet_id: 1, + ancestors: vec![10], + tombstone_ancestor_ids: vec![10], + ..Default::default() + }; + + assert!(!should_drop_ancillary(&reply, &verdicts)); + } + + #[test] + fn interstitial_on_ancillary_does_not_drop() { + let verdicts = verdicts(vec![], vec![(10, Ok(Some(interstitial())))]); + let quote = PostCandidate { + tweet_id: 1, + quoted_tweet_id: Some(10), + ..Default::default() + }; + + assert!(!should_drop_ancillary("e, &verdicts)); + } + + #[test] + fn primary_lookup_error_is_surfaced() { + let verdicts = verdicts(vec![(1, Err(anyhow::anyhow!("vf unavailable")))], vec![]); + let post = PostCandidate { + tweet_id: 1, + in_network: Some(true), + ..Default::default() + }; + + let err = resolve_visibility(&post, &verdicts).unwrap_err(); + + assert!(err.contains("vf unavailable")); + } + + /// Answers Allow at TimelineHome and Drop at TimelineHomeRecommendations for + /// every id, mimicking a post whose author carries an OON-only label. + struct LevelSensitiveVfClient { + calls: Mutex)>>, + } + + #[async_trait] + impl VfClient for LevelSensitiveVfClient { + async fn get_result( + &self, + post_ids: Vec, + safety_level: SafetyLevel, + _for_user_id: u64, + _context: Option, + ) -> HashMap>> { + self.calls + .lock() + .unwrap() + .push((safety_level.clone(), post_ids.clone())); + let reason = match safety_level { + TimelineHome => None, + _ => Some(oon_only_drop()), + }; + post_ids + .into_iter() + .map(|id| (id, Ok(reason.clone()))) + .collect() + } + } + + #[tokio::test] + async fn followed_author_thread_root_is_not_dropped_because_reply_lists_it_as_ancestor() { + let client = Arc::new(LevelSensitiveVfClient { + calls: Mutex::new(Vec::new()), + }); + let hydrator = VFCandidateHydrator::new(client.clone(), client.clone()).await; + let root = PostCandidate { + tweet_id: 1, + in_network: Some(true), + ..Default::default() + }; + let reply_in_thread = PostCandidate { + tweet_id: 2, + in_network: Some(true), + ancestors: vec![1], + ..Default::default() + }; + let quote_of_root = PostCandidate { + tweet_id: 3, + in_network: Some(false), + quoted_tweet_id: Some(1), + ..Default::default() + }; + + let results = hydrator + .hydrate( + &ScoredPostsQuery::default(), + &[root, reply_in_thread, quote_of_root], + ) + .await; + + let root = results[0].as_ref().unwrap(); + assert_eq!( + root.visibility_reason, None, + "in-network root must keep its TimelineHome verdict" + ); + assert_eq!(root.drop_ancillary_posts, Some(false)); + + let reply = results[1].as_ref().unwrap(); + assert_eq!(reply.visibility_reason, None); + assert_eq!( + reply.drop_ancillary_posts, + Some(true), + "ancestor is still judged as a recommendation" + ); + + let quote = results[2].as_ref().unwrap(); + assert_eq!(quote.visibility_reason, Some(oon_only_drop())); + assert_eq!(quote.drop_ancillary_posts, Some(true)); + + let calls = client.calls.lock().unwrap(); + let in_network_ids: Vec = calls + .iter() + .filter(|(level, _)| *level == TimelineHome) + .flat_map(|(_, ids)| ids.iter().copied()) + .collect(); + let oon_ids: Vec = calls + .iter() + .filter(|(level, _)| *level == TimelineHomeRecommendations) + .flat_map(|(_, ids)| ids.iter().copied()) + .collect(); + assert!(in_network_ids.contains(&1) && oon_ids.contains(&1)); + } +} From b4a3553059e821efdd9ab69a82e0c325bd8122eb Mon Sep 17 00:00:00 2001 From: Jon Bailey <297513015+Pitchfork-and-Torch@users.noreply.github.com> Date: Fri, 4 Sep 2026 04:07:09 -0700 Subject: [PATCH 3/3] abuse-enforcement-service: fail closed when the allowlist store errors (#7) The allowlist is an exemption check that runs before any rule can fire. Until now a Manhattan GET failure during that check was swallowed inside ManhattanAllowlist::get_entity and returned as None, which the fetch helpers read as is_allowlisted: false. Enforcement then continued into the rule pipeline for an account or post that may have been exempt, while the neighbouring Gizmoduck and credibility fetches on the same path abort with `?` and are retried. Make the allowlist lookup behave like those fetches: - ManhattanAllowlist::get / get_entity return anyhow::Result>. Ok(None) means the store confirmed the key is absent. A GET error or an undecodable stored entry is returned as Err instead of None. - fetch_user_allowlist / fetch_entity_allowlist return Result and propagate the error. Only a confirmed absence maps to "not allowlisted". - run_enforcement_inner uses `?` on the allowlist lookups, so a store error aborts the attempt and the score lands in the existing retry queue (backoff, then dropped without enforcing) rather than proceeding to rules. - Admin handlers: GET /allowlist/{id} and GET /allowlist/{type}/{id} return 500 on a read error instead of 404; bulk upsert reports a failed pre-read as a per-row error; the DELETE audit snapshot stays best-effort. Adds unit tests for the lookup-to-facts conversion, including one that asserts a store error is not turned into is_allowlisted=false. Co-authored-by: Cursor Agent Co-authored-by: Jon Bailey --- .../service-lib/src/allowlist.rs | 43 +++--- .../service-lib/src/lib.rs | 3 +- .../service-lib/src/service.rs | 46 ++++++- .../service-lib/src/strato.rs | 126 +++++++++++------- 4 files changed, 151 insertions(+), 67 deletions(-) diff --git a/abuse-enforcement-service/service-lib/src/allowlist.rs b/abuse-enforcement-service/service-lib/src/allowlist.rs index bef3e83a..62b9bf81 100644 --- a/abuse-enforcement-service/service-lib/src/allowlist.rs +++ b/abuse-enforcement-service/service-lib/src/allowlist.rs @@ -29,15 +29,17 @@ impl ManhattanAllowlist { } } - pub async fn get(&self, user_id: i64) -> Option { - let (entry, ttl_secs) = self.get_entity(EntityType::User, user_id).await?; - Some(AllowlistRecord { - user_id, - added_by: entry.added_by, - reason: entry.reason, - added_at: entry.added_at, - ttl_secs, - }) + pub async fn get(&self, user_id: i64) -> anyhow::Result> { + Ok(self + .get_entity(EntityType::User, user_id) + .await? + .map(|(entry, ttl_secs)| AllowlistRecord { + user_id, + added_by: entry.added_by, + reason: entry.reason, + added_at: entry.added_at, + ttl_secs, + })) } pub async fn add( @@ -54,30 +56,41 @@ impl ManhattanAllowlist { self.remove_entity(EntityType::User, user_id).await } + /// Looks up an allowlist entry. `Ok(None)` means the store confirmed the + /// entity is absent. A store or decode failure is returned as `Err` so + /// callers can distinguish "not allowlisted" from "could not check", and + /// never treat an unreadable entry as not allowlisted. pub async fn get_entity( &self, entity_type: EntityType, entity_id: i64, - ) -> Option<(AllowlistEntry, i64)> { + ) -> anyhow::Result> { let lkey = Self::entity_lkey(entity_type, entity_id); let item = match self .client .get(self.tenant.clone(), [MH_PKEY], [lkey.as_str()]) .await { - Ok(v) => v?, + Ok(Some(item)) => item, + Ok(None) => return Ok(None), Err(e) => { - warn!("Manhattan allowlist GET failed: {e}"); + warn!("Manhattan allowlist GET failed for {lkey}: {e}"); crate::metrics::MANHATTAN_ERRORS_TOTAL.inc(); - return None; + return Err(anyhow::anyhow!( + "Manhattan allowlist GET failed for {lkey}: {e}" + )); } }; - let entry: AllowlistEntry = serde_json::from_slice(item.value().as_bytes()).ok()?; + let entry: AllowlistEntry = + serde_json::from_slice(item.value().as_bytes()).map_err(|e| { + warn!("Manhattan allowlist entry decode failed for {lkey}: {e}"); + anyhow::anyhow!("Manhattan allowlist entry decode failed for {lkey}: {e}") + })?; let ttl_secs = crate::manhattan::remaining_ttl_secs( item.expires_at() .map(|d| d.timestamp_nanos_opt().unwrap_or(0) as u64), ); - Some((entry, ttl_secs)) + Ok(Some((entry, ttl_secs))) } pub async fn add_entity( diff --git a/abuse-enforcement-service/service-lib/src/lib.rs b/abuse-enforcement-service/service-lib/src/lib.rs index 244ab59e..084ae8dc 100644 --- a/abuse-enforcement-service/service-lib/src/lib.rs +++ b/abuse-enforcement-service/service-lib/src/lib.rs @@ -218,7 +218,7 @@ async fn run_enforcement_inner( let mut facts = match entity_type { EntityType::User => { let user_id = entity_id; - let allowlist = fetch_user_allowlist(ctx.allowlist.as_ref(), user_id).await; + let allowlist = fetch_user_allowlist(ctx.allowlist.as_ref(), user_id).await?; if allowlist.is_allowlisted { let partial = Facts { entity_type, @@ -265,6 +265,7 @@ async fn run_enforcement_inner( fetch_entity_allowlist(ctx.allowlist.as_ref(), EntityType::Post, entity_id), fetch_user_allowlist(ctx.allowlist.as_ref(), author_id), ); + let (post_allowlist, author_allowlist) = (post_allowlist?, author_allowlist?); if post_allowlist.is_allowlisted || author_allowlist.is_allowlisted { let reason = if post_allowlist.is_allowlisted { "post_in_allowlist" diff --git a/abuse-enforcement-service/service-lib/src/service.rs b/abuse-enforcement-service/service-lib/src/service.rs index fd63db9d..2d7ceee7 100644 --- a/abuse-enforcement-service/service-lib/src/service.rs +++ b/abuse-enforcement-service/service-lib/src/service.rs @@ -1549,9 +1549,12 @@ pub async fn handle_allowlist_delete( ); }; info!("DELETE /api/allowlist/{user_id}"); + // Audit snapshot only; a failed read must not block the delete. let before_json = al .get(user_id) .await + .ok() + .flatten() .map(|r| serde_json::to_string(&r).unwrap_or_default()) .unwrap_or_default(); let result = al.remove(user_id).await; @@ -1619,7 +1622,21 @@ pub async fn bulk_allowlist_impl( continue; } - let existing_ttl = al.get(entry.user_id).await.map(|r| r.ttl_secs); + let existing_ttl = match al.get(entry.user_id).await { + Ok(existing) => existing.map(|r| r.ttl_secs), + Err(e) => { + errors += 1; + results.push(BulkRowResult { + user_id: entry.user_id, + ok: false, + mode: String::new(), + error: Some(format!("allowlist lookup failed: {e}")), + existing_ttl_secs: None, + ttl_secs: Some(entry.ttl_secs), + }); + continue; + } + }; let mode = if existing_ttl.is_some() { would_update += 1; "update" @@ -1783,6 +1800,7 @@ pub async fn handle_allowlist_list(State(state): State>) -> impl I responses( (status = 200, description = "Allowlist record for this user_id", body = AllowlistRecord), (status = 404, description = "No allowlist entry for this user_id"), + (status = 500, description = "Manhattan read failed"), (status = 503, description = "Allowlist not configured"), ), )] @@ -1798,11 +1816,18 @@ pub async fn handle_allowlist_get( ); }; match al.get(user_id).await { - Some(record) => ( + Ok(Some(record)) => ( StatusCode::OK, Json(serde_json::to_value(&record).unwrap_or_default()), ), - None => (StatusCode::NOT_FOUND, Json(json!({"error": "not found"}))), + Ok(None) => (StatusCode::NOT_FOUND, Json(json!({"error": "not found"}))), + Err(e) => { + error!("GET /api/allowlist/{user_id} failed: {e}"); + ( + StatusCode::INTERNAL_SERVER_ERROR, + Json(json!({"error": e.to_string()})), + ) + } } } @@ -1820,6 +1845,7 @@ pub async fn handle_allowlist_get( (status = 200, description = "Allowlist record for this entity", body = Object), (status = 400, description = "Unknown entity_type"), (status = 404, description = "No allowlist entry for this entity"), + (status = 500, description = "Manhattan read failed"), (status = 503, description = "Allowlist not configured"), ), )] @@ -1843,7 +1869,7 @@ pub async fn handle_allowlist_get_entity( ); }; match al.get_entity(et, entity_id).await { - Some((entry, ttl_secs)) => ( + Ok(Some((entry, ttl_secs))) => ( StatusCode::OK, Json(json!({ "entity_type": et.as_str(), @@ -1854,7 +1880,14 @@ pub async fn handle_allowlist_get_entity( "ttl_secs": ttl_secs, })), ), - None => (StatusCode::NOT_FOUND, Json(json!({"error": "not found"}))), + Ok(None) => (StatusCode::NOT_FOUND, Json(json!({"error": "not found"}))), + Err(e) => { + error!("GET /api/allowlist/{}/{entity_id} failed: {e}", et.as_str()); + ( + StatusCode::INTERNAL_SERVER_ERROR, + Json(json!({"error": e.to_string()})), + ) + } } } @@ -2009,9 +2042,12 @@ pub async fn handle_allowlist_delete_entity( ); }; info!("DELETE /api/allowlist/{}/{entity_id}", et.as_str()); + // Audit snapshot only; a failed read must not block the delete. let before_json = al .get_entity(et, entity_id) .await + .ok() + .flatten() .map(|(entry, ttl_secs)| { json!({ "entity_type": et.as_str(), diff --git a/abuse-enforcement-service/service-lib/src/strato.rs b/abuse-enforcement-service/service-lib/src/strato.rs index 6b21b2ad..687f0702 100644 --- a/abuse-enforcement-service/service-lib/src/strato.rs +++ b/abuse-enforcement-service/service-lib/src/strato.rs @@ -8,7 +8,7 @@ use serde::Serialize; use tracing::{info, warn}; use xai_strato::Strato; -use crate::allowlist::ManhattanAllowlist; +use crate::allowlist::{AllowlistEntry, ManhattanAllowlist}; use crate::entities::{self, HighPageRankUser}; use crate::facts::{AllowlistFacts, CredFacts, EntityType, GizmoduckFacts}; use crate::gizmoduck::GizmoduckCoreClient; @@ -56,37 +56,45 @@ where } } +/// Turns the result of an allowlist store lookup into facts. +/// +/// A store error is propagated rather than mapped to `is_allowlisted: false`: +/// the allowlist is an exemption, so "could not check" must stop enforcement +/// (the caller retries) instead of silently reading as "not exempt". Only a +/// confirmed `Ok(None)` from the store means the entity is not allowlisted. +fn allowlist_facts(lookup: Result>) -> Result { + Ok(match lookup? { + Some((entry, ttl_secs)) => AllowlistFacts { + is_allowlisted: true, + added_by: Some(entry.added_by), + reason: Some(entry.reason), + ttl_secs: Some(ttl_secs), + }, + None => AllowlistFacts::default(), + }) +} + #[tracing::instrument(skip_all, fields(is_allowlisted))] pub async fn fetch_user_allowlist( allowlist: Option<&ManhattanAllowlist>, user_id: i64, -) -> AllowlistFacts { +) -> Result { let span = tracing::Span::current(); let Some(al) = allowlist else { span.record("is_allowlisted", false); - return AllowlistFacts::default(); + return Ok(AllowlistFacts::default()); }; - match al.get(user_id).await { - Some(record) => { - span.record("is_allowlisted", true); - warn!( - added_by = record.added_by, - ttl_secs = record.ttl_secs, - reason = record.reason, - "user is in allowlist; will skip", - ); - AllowlistFacts { - is_allowlisted: true, - added_by: Some(record.added_by), - reason: Some(record.reason), - ttl_secs: Some(record.ttl_secs), - } - } - None => { - span.record("is_allowlisted", false); - AllowlistFacts::default() - } + let facts = allowlist_facts(al.get_entity(EntityType::User, user_id).await)?; + span.record("is_allowlisted", facts.is_allowlisted); + if facts.is_allowlisted { + warn!( + added_by = facts.added_by.as_deref(), + ttl_secs = facts.ttl_secs, + reason = facts.reason.as_deref(), + "user is in allowlist; will skip", + ); } + Ok(facts) } #[tracing::instrument(skip_all, fields(entity_type = entity_type.as_str(), is_allowlisted))] @@ -94,34 +102,24 @@ pub async fn fetch_entity_allowlist( allowlist: Option<&ManhattanAllowlist>, entity_type: EntityType, entity_id: i64, -) -> AllowlistFacts { +) -> Result { let span = tracing::Span::current(); let Some(al) = allowlist else { span.record("is_allowlisted", false); - return AllowlistFacts::default(); + return Ok(AllowlistFacts::default()); }; - match al.get_entity(entity_type, entity_id).await { - Some((entry, ttl_secs)) => { - span.record("is_allowlisted", true); - warn!( - entity_id, - added_by = entry.added_by, - ttl_secs, - reason = entry.reason, - "entity is in allowlist; will skip", - ); - AllowlistFacts { - is_allowlisted: true, - added_by: Some(entry.added_by), - reason: Some(entry.reason), - ttl_secs: Some(ttl_secs), - } - } - None => { - span.record("is_allowlisted", false); - AllowlistFacts::default() - } + let facts = allowlist_facts(al.get_entity(entity_type, entity_id).await)?; + span.record("is_allowlisted", facts.is_allowlisted); + if facts.is_allowlisted { + warn!( + entity_id, + added_by = facts.added_by.as_deref(), + ttl_secs = facts.ttl_secs, + reason = facts.reason.as_deref(), + "entity is in allowlist; will skip", + ); } + Ok(facts) } pub async fn fetch_user(gd: &GizmoduckCoreClient, user_id: i64) -> Result { @@ -280,4 +278,40 @@ mod tests { let r: entities::StratoResponse = serde_json::from_str(json).unwrap(); assert_eq!(r.v, Some(true)); } + + #[test] + fn allowlist_facts_absent_is_not_allowlisted() { + let f = allowlist_facts(Ok(None)).unwrap(); + assert!(!f.is_allowlisted); + assert_eq!(f.added_by, None); + assert_eq!(f.reason, None); + assert_eq!(f.ttl_secs, None); + } + + #[test] + fn allowlist_facts_present_is_allowlisted() { + let entry = AllowlistEntry { + added_by: "oncall".into(), + reason: "false positive".into(), + added_at: 1_700_000_000, + }; + let f = allowlist_facts(Ok(Some((entry, 3600)))).unwrap(); + assert!(f.is_allowlisted); + assert_eq!(f.added_by.as_deref(), Some("oncall")); + assert_eq!(f.reason.as_deref(), Some("false positive")); + assert_eq!(f.ttl_secs, Some(3600)); + } + + #[test] + fn allowlist_facts_store_error_is_not_read_as_not_allowlisted() { + let err = allowlist_facts(Err(anyhow::anyhow!("manhattan unavailable"))); + assert!( + err.is_err(), + "a failed allowlist lookup must propagate as an error, not as is_allowlisted=false" + ); + assert!(err + .unwrap_err() + .to_string() + .contains("manhattan unavailable")); + } }