# Filters and hydrators

> Pre-score and post-selection filters, query hydrators that fill ScoredPostsQuery, and candidate hydrators that must preserve order and length.

- Repository: xai-org/x-algorithm
- GitHub: https://github.com/xai-org/x-algorithm
- Human docs: https://grok-wiki.com/public/docs/xai-org-x-algorithm-23c09c39074c
- Complete Markdown: https://grok-wiki.com/public/docs/xai-org-x-algorithm-23c09c39074c/llms-full.txt

## Source Files

- `home-mixer/filters/mod.rs`
- `home-mixer/filters/age_filter.rs`
- `home-mixer/filters/vf_filter.rs`
- `home-mixer/query_hydrators/mod.rs`
- `home-mixer/query_hydrators/scoring_sequence_query_hydrator.rs`
- `home-mixer/candidate_hydrators/mod.rs`
- `home-mixer/candidate_hydrators/core_data_candidate_hydrator.rs`
- `candidate-pipeline/filter.rs`

---

---
title: "Filters and hydrators"
description: "Pre-score and post-selection filters, query hydrators that fill ScoredPostsQuery, and candidate hydrators that must preserve order and length."
---

`PhoenixCandidatePipeline` is the production Home Mixer surface that hydrates `ScoredPostsQuery`, enriches `PostCandidate`s, then partitions them with sequential filters. Query hydrators run in parallel and merge only the fields they own. Candidate hydrators also run in parallel and must return the same candidates in the same order — dropping a candidate belongs in a `Filter`, not a `Hydrator`. `ForYouCandidatePipeline` wraps scored posts plus ads and modules; it hydrates served history and request timestamps, then leaves candidate hydrators and filters empty.

These components live in `home-mixer/` and implement traits from `candidate-pipeline/`. Local Phoenix inference (`run_pipeline.py`) does not execute them. Production clients, `crate::params` feature switches, TES, VF, SocialGraph, and Strato are required at runtime.

<Info>
`CandidatePipeline::execute` already has a dedicated request-lifecycle page. This page is the component contract and the Phoenix / For You registries.
</Info>

## Contracts

Three traits own the behavior. `enable` defaults to `true`. Tracing spans are `query_hydrator`, `hydrator`, and `filter`, each recording `name`.

<ParamField body="QueryHydrator::hydrate" type="async fn(&Q) -> Result<Q, String>">
Fetch this hydrator's fields and return a new query. `update` copies only those fields onto the live query. A failed `hydrate` is logged (`Failed: …`) and skipped; the pipeline continues without that field set.
</ParamField>

<ParamField body="Hydrator::hydrate" type="async fn(&Q, &[C]) -> Vec<Result<C, String>>">
Return one result per input candidate, same order. Dropping candidates is not allowed. `Hydrator::run` replaces a length mismatch with `Err("Hydrator length_mismatch expected=N got=M")` for every slot and skips `update_all`. `update` copies only this hydrator's fields.
</ParamField>

<ParamField body="Filter::filter" type="fn(&Q, Vec<C>) -> FilterResult<C>">
Partition into `kept` (next stage) and `removed` (excluded). Filters run sequentially; each filter sees only the previous `kept` set. `run` records `kept_count`, `removed_count`, and `filter_rate`, and increments `{Name}.run` under `requests.kept` / `requests.removed`.
</ParamField>

`CachedHydrator` is a `Hydrator` adapter: per-candidate cache lookup, then `hydrate_from_client` for misses. A client length mismatch fails the **entire** candidate list. Cache hits and misses increment `{Name}.cache` under `requests.cache_hit` / `requests.cache_miss`.

```rust
// candidate-pipeline/hydrator.rs — length is a hard contract
async fn hydrate(&self, query: &Q, candidates: &[C]) -> Vec<Result<C, String>>;
// The returned vector must have the same candidates in the same order as the input.
// Dropping candidates in a hydrator is not allowed - use a filter stage instead.
```

## Execute order

`CandidatePipeline::execute` applies these stages in order. Query hydrators are two waves: `query_hydrators` then `dependent_query_hydrators`. Neither Phoenix nor For You overrides the second wave (default empty).

```mermaid
flowchart TB
  subgraph queryLayer [Query layer — parallel]
    QH[query_hydrators]
    DQH[dependent_query_hydrators]
    QH --> DQH
  end
  subgraph sourceLayer [Sources — parallel]
    SRC[Thunder / Phoenix / TweetMixer / cache]
  end
  subgraph candLayer [Candidate layer — parallel]
    CH[hydrators]
  end
  subgraph filterLayer [Pre-score filters — sequential]
    F[filters]
  end
  subgraph scoreLayer [Score and select]
    SC[scorers]
    SEL[selector]
    SC --> SEL
  end
  subgraph postLayer [Post-selection]
    PSH[post_selection_hydrators — parallel]
    PSF[post_selection_filters — sequential]
    PSH --> PSF
  end
  queryLayer --> sourceLayer --> candLayer --> filterLayer --> scoreLayer --> postLayer
```

Removed candidates from both filter stages accumulate in `PipelineResult.filtered_candidates`. After post-selection filters, `execute` truncates to `result_size()` (`params::RESULT_SIZE` on Phoenix, `params::FOR_YOU_MAX_RESULT_SIZE` on For You).

## Query hydrators

Query hydrators fill `ScoredPostsQuery` before sources run. Each `hydrate` returns `ScoredPostsQuery { …owned fields, ..Default::default() }` and `update` copies only those fields so parallel merges do not clobber siblings.

### Phoenix registry

Wired in `PhoenixCandidatePipeline::build_with_clients`, in this order (execution is still parallel):

| Hydrator | Writes | Enable |
|---|---|---|
| `ScoringSequenceQueryHydrator` | `scoring_sequence`, `columnar_scoring_sequence` | always |
| `RetrievalSequenceQueryHydrator` | `retrieval_sequence`, `columnar_retrieval_sequence` | always |
| `BlockedUserIdsQueryHydrator` | `user_features.blocked_user_ids` | always |
| `MutedUserIdsQueryHydrator` | `user_features.muted_user_ids` | always |
| `FollowedUserIdsQueryHydrator` | `user_features.followed_user_ids` | always |
| `SubscribedUserIdsQueryHydrator` | `user_features.subscribed_user_ids` | always |
| `CachedPostsQueryHydrator` | `cached_posts`, `has_cached_posts` | `EnableCachedPosts` |
| `MutualFollowQueryHydrator` | `viewer_minhash` | `EnableMutualFollowJaccardHydration` |
| `UserDemographicsQueryHydrator` | `user_demographics` | `EnableContextFeatures` or `is_shadow_traffic` |
| `FollowedGrokTopicsQueryHydrator` | `followed_grok_topics`, maybe `new_user_topic_ids` | `EnableContextFeatures`, shadow traffic, `EnableNewUserTopicRetrieval`, or `EnableNewUserTopicFiltering` |
| `FollowedStarterPacksQueryHydrator` | `followed_starter_packs` | `EnableContextFeatures` or `is_shadow_traffic` |
| `InferredGrokTopicsQueryHydrator` | `inferred_grok_topics` | `EnableGrokTopicsHydration` |
| `ImpressionBloomFilterQueryHydrator` | `bloom_filter_entries` | always |
| `IpQueryHydrator` | `ip_location` | `EnableIpFeature` and non-empty `ip_address` |
| `UserInferredGenderQueryHydrator` | `user_inferred_gender`, `user_inferred_gender_score` | `EnableInferredGenderHydration` or `is_shadow_traffic` |

Social-graph hydrators call `SocialGraphClientOps` (`get_blocked_user_ids`, `get_muted_user_ids`, `get_followed_user_ids`, `get_subscribed_user_ids`) and write into `UserFeatures`. Downstream filters read those lists; they are not request-body fields.

### Sequence hydrators

Both sequence hydrators call `UserActionAggregationClient::fetch_aggregated_sequence` with `UAS_WINDOW_TIME_MS`, `UasSourceDataType` (default `Arrow`), `UseXdsForUas`, and optional realtime actions.

| | Scoring | Retrieval |
|---|---|---|
| Length param | `MaxSeqLengthScoring` | `MaxSeqLengthRetrieval` |
| Aggregation default | `DenseWithNotInterestedIn` (`PhoenixAggregationType`) | `Dense` (`PhoenixRetrievalAggregationType`) |
| `prediction_id` | `Some(query.prediction_id as i64)` | `None` |
| Response | `ResponseFormat::Arrow` | `ResponseFormat::Arrow` |

A missing `retrieval_sequence` later fails Phoenix retrieval. Sequence fields are `#[serde(skip)]` on `ScoredPostsQuery`.

### Cached posts

`CachedPostsQueryHydrator` GETs Redis key `cached_posts_key(user_id, topic_ids, in_network_only, exclude_videos)` with a 300 ms timeout, zstd-decompresses, and JSON-decodes `Vec<PostCandidate>`. `has_cached_posts` is `cached_posts.len() >= 500`. An empty payload leaves defaults. When `has_cached_posts` is true, most TES / SocialGraph candidate hydrators disable themselves so cached fields are not overwritten.

<Warning>
`ImpressedPostsQueryHydrator` is constructed in `build_with_clients` as `_impressed_posts_hydrator` and is **not** pushed into `query_hydrators`. `PreviouslySeenPostsBackupFilter` therefore only sees `impressed_post_ids` if some other path populated them. The hydrator itself writes `impressed_post_ids` from `ImpressedPostsClient::get(user_id)`.
</Warning>

### For You registry

`ForYouCandidatePipeline` registers two query hydrators, both gated on `EnableUrtMigrationComponents`:

| Hydrator | Writes |
|---|---|
| `ServedHistoryQueryHydrator` | `served_history`, `served_ids` (recent tweet / source IDs within `ExcludeServedTweetIdsDuration`, capped by `ExcludeServedTweetIdsNumber`), `who_to_follow_eligible` (`WhoToFollowFatigueHours`) |
| `PastRequestTimestampsQueryHydrator` | `non_polling_timestamps` |

### Present on disk, not on Phoenix

`UserActionSeqQueryHydrator` and `UserFeaturesQueryHydrator` implement `QueryHydrator` against the older `home-mixer/candidate_pipeline/query.rs` `ScoredPostsQuery` (`user_action_sequence`, `user_id: i64`). They are not in `query_hydrators/mod.rs` and are not registered on `PhoenixCandidatePipeline`. The live query type is `home-mixer/models/query.rs`.

## Candidate hydrators

Candidate hydrators run after sources and before pre-score filters. Most TES-backed hydrators implement `CachedHydrator` with a Moka cache and disable when `query.has_cached_posts`.

### Pre-score (`hydrators`)

| Hydrator | Writes | Enable |
|---|---|---|
| `InNetworkCandidateHydrator` | `in_network` | `!has_cached_posts` |
| `CoreDataCandidateHydrator` | `retweeted_user_id`, `retweeted_tweet_id`, `in_reply_to_tweet_id`, `tweet_text` | `!has_cached_posts` |
| `QuoteHydrator` | `quoted_tweet_id`, `quoted_user_id`, `quoted_author_blocks_viewer`, `quoted_video_duration_ms` | `!has_cached_posts` |
| `VideoDurationCandidateHydrator` | `min_video_duration_ms` | `!has_cached_posts` |
| `HasMediaHydrator` | `has_media` | (`EnableHasMediaHydration` or shadow traffic) and `!has_cached_posts` |
| `SubscriptionHydrator` | `subscription_author_id` | `!has_cached_posts` |
| `GizmoduckCandidateHydrator` | `author_followers_count`, `author_screen_name`, `retweeted_screen_name` | `!has_cached_posts` |
| `BlockedByHydrator` | `author_blocks_viewer` | `!has_cached_posts` |
| `FilteredTopicsHydrator` | `filtered_topic_ids`, `unfiltered_topic_ids` | topic request, excluded topics, or (`EnableNewUserTopicFiltering` and `has_new_user_topic_ids`) |
| `LanguageCodeHydrator` | `language_code` | `!has_cached_posts` |

`InNetworkCandidateHydrator` sets `in_network` when `author_id == query.user_id` or `author_id` is in `user_features.followed_user_ids`.

`CoreDataCandidateHydrator` loads TES `get_tweet_core_datas`. A miss still returns `Ok(PostCandidate::default())` so length is preserved; found/missing counts increment `{Name}.hydrate` under `hydration.found` / `hydration.missing`. `update` does **not** copy `author_id`. Sources (`ThunderSource`, `PhoenixSource`, `PhoenixTopicsSource`, `PhoenixMOESource`, `TweetMixerSource`) already set `author_id`. `CoreDataHydrationFilter` then drops `author_id == 0`.

`EngagementCountsHydrator` exists under `candidate_hydrators/` and writes `fav_count` / `reply_count` / `repost_count` / `quote_count`, but it is not in `mod.rs` and is not registered.

### Post-selection (`post_selection_hydrators`)

These run on the selector's `selected` set only.

| Hydrator | Writes | Enable |
|---|---|---|
| `VFCandidateHydrator` | `visibility_reason`, `drop_ancillary_posts` | always |
| `AdsBrandSafetyHydrator` | `brand_safety_verdict`, `safety_labels` | `EnableAdsBrandSafetyHydrator` and decider `vf_brand_safety_dark_traffic` is **off** |
| `AdsBrandSafetyVfHydrator` | same fields | `EnableAdsBrandSafetyHydrator` and that decider is **on** |
| `TweetTypeMetricsHydrator` | `tweet_type_metrics` | always |
| `FollowingRepliedUsersHydrator` | `following_replied_user_ids` | `EnableFollowingRepliedUsersFacepile` and `follower_count >= 1000` |
| `MutualFollowJaccardHydrator` | `mutual_follow_jaccard` | `EnableMutualFollowJaccardHydration` and `viewer_minhash` is `Some` |

`VFCandidateHydrator` splits IDs by network:

- In-network tweet IDs and retweet source IDs → VF safety level `TimelineHome`
- Out-of-network tweet IDs, ancestors, and quoted IDs → `TimelineHomeRecommendations`

`drop_ancillary_posts` is true when an ancestor, quoted tweet, or retweeted tweet has a drop-class VF reason. Primary tweet VF is stored in `visibility_reason`.

## Pre-score filters

`PhoenixCandidatePipeline` runs these sequentially after candidate hydration and **before** scorers. Order is the drop-priority order.

| Filter | Drops when | Enable |
|---|---|---|
| `DropDuplicatesFilter` | duplicate `tweet_id` (keeps first) | always |
| `CoreDataHydrationFilter` | `author_id == 0` | always |
| `AgeFilter` | snowflake age of `tweet_id` &gt; `Duration::from_secs(params::MAX_POST_AGE)`, or age unknown | always |
| `SelfTweetFilter` | `author_id == query.user_id` | always |
| `RetweetDeduplicationFilter` | duplicate `retweeted_tweet_id.unwrap_or(tweet_id)` (keeps first) | always |
| `IneligibleSubscriptionFilter` | `subscription_author_id` is `Some` and not in `user_features.subscribed_user_ids` | always |
| `PreviouslySeenPostsFilter` | any related post ID is in `seen_ids` or any impression bloom filter | always |
| `PreviouslySeenPostsBackupFilter` | any related post ID is in `impressed_post_ids`; no-op if that vec is empty | always |
| `PreviouslyServedPostsFilter` | any related post ID is in `served_ids` | `EnableServedFilterAllRequests`, or (`is_bottom_request` and request context is not `ForegroundTruncate`) |
| `MutedKeywordFilter` | tokenized `tweet_text` matches `user_features.muted_keywords`; no-op if keywords empty | always |
| `AuthorSocialgraphFilter` | author muted/blocked, author blocks viewer, quoted author blocks viewer, viewer blocks quoted author, or viewer blocks retweeted user | always |
| `VideoFilter` | `min_video_duration_ms` is `Some` | `query.exclude_videos` |
| `TopicIdsFilter` | topic-request / excluded-topic mismatch (see below) | `is_topic_request()` or `has_excluded_topics()` |
| `NewUserTopicIdsFilter` | not in-network and no expanded `new_user_topic_ids` overlap | `EnableNewUserTopicFiltering` and `has_new_user_topic_ids()` and not a topic request |

Related-post checks use `get_related_post_ids` (`crate::util::candidates_util`, unpublished helper used by the seen / served filters).

`MutedKeywordFilter` tokenizes with `TweetTokenizer` / `UserMutes` / `MatchTweetGroup` inside `tokio::task::block_in_place`.

### Topic filtering

`TopicIdsFilter` uses `TopicIdExpansion` (static category / supertopic maps). `ScoredPostsQuery::is_topic_request` is `!topic_ids.is_empty()`. `is_bulk_topic_request` is `topic_ids.len() > 6`.

- **Single / small topic request:** keep if expanded or unfiltered topic IDs match the requested IDs (including supertopic expansion).
- **Bulk topic request:** keep unless every `filtered_topic_id` is in the complement of the expanded request set; empty topic lists are kept.
- **Excluded topics:** after the keep pass, drop any candidate whose `filtered_topic_ids` intersect the expanded excluded set. Candidates with empty `filtered_topic_ids` are dropped when exclusions are present. Each excluded ID increments `TopicIdsFilter.excluded_topic_id`.

`FollowedGrokTopicsQueryHydrator` fills `new_user_topic_ids` only when new-user topic params are on, the request is not a topic request, `in_network_only` is false, and snowflake age of `user_id` is under `NewUserTopicAgeThresholdSecs`. `update` copies `new_user_topic_ids` only when the hydrated vec is non-empty.

## Post-selection filters

These run after `TopKScoreSelector` and the post-selection hydrators.

| Filter | Drops when |
|---|---|
| `VFFilter` | `visibility_reason` is `SafetyResult` with `Action::Drop(_)`, or any non-`SafetyResult` reason. `None` is kept. |
| `AncillaryVFFilter` | `drop_ancillary_posts == Some(true)` |
| `DedupConversationFilter` | not the highest `score` in the conversation. Conversation id is `ancestors.iter().min()` or `tweet_id`. |

`ForYouCandidatePipeline` sets `filters()`, `hydrators()`, `post_selection_hydrators()`, and `post_selection_filters()` to empty slices. Organic visibility filtering happens inside Phoenix before `ScoredPostsSource` returns.

## Layout

:::files
home-mixer/
  query_hydrators/          # QueryHydrator&lt;ScoredPostsQuery&gt;
  candidate_hydrators/      # Hydrator / CachedHydrator&lt;ScoredPostsQuery, PostCandidate&gt;
  filters/                  # Filter&lt;ScoredPostsQuery, PostCandidate&gt;
  candidate_pipeline/
    phoenix_candidate_pipeline.rs   # registries
    for_you_candidate_pipeline.rs   # served-history query hydrators only
  models/query.rs           # live ScoredPostsQuery
candidate-pipeline/
  query_hydrator.rs
  hydrator.rs
  filter.rs
  candidate_pipeline.rs     # execute + run_hydrators / run_filters
:::

## Failure modes

| Signal | Meaning | What continues |
|---|---|---|
| `query_hydrator` log `Failed: …` | `hydrate` returned `Err` | Query keeps prior / default field; other hydrators still merge |
| `Skipped: length_mismatch expected=N got=M` on a hydrator | Returned vec length ≠ input | That hydrator's fields are not applied; candidates stay in place |
| `CachedHydrator length_mismatch` | Miss-path client returned the wrong length | Every candidate gets `Err` for that hydrator |
| `CoreDataHydrationFilter` removals | `author_id == 0` after sources + TES | Candidate never reaches scoring |
| `has_cached_posts == true` | Redis cache ≥ 500 posts | TES / SocialGraph / in-network hydrators skip; cached fields stand |
| `PreviouslySeenPostsBackupFilter` removes nothing | `impressed_post_ids` empty | Expected unless another writer filled the field |
| `Aggregation service call failed: …` | UAS fetch failed | Scoring or retrieval sequence stays `None` |
| `CachedPostsQueryHydrator redis GET timed out` / `GET error` | 300 ms timeout or Redis error | `cached_posts` stays empty; live TES hydration runs |
| `FollowedGrokTopics MH get timed out` | 300 ms Manhattan timeout | Followed-topic / new-user topic fields stay unset |
| Empty For You / Scored Posts feed | Test-user short-circuit or failed query build | See troubleshooting; not a filter miss |

`params::*` identifiers (`MAX_POST_AGE`, `EnableCachedPosts`, `RESULT_SIZE`, …) come from unpublished `crate::params`. This checkout imports them but does not ship the module; do not assume numeric defaults from this tree.

## Add a component

<Steps>
<Step title="Pick the stage">
Query-level data → `QueryHydrator`. Per-candidate enrichment that must not drop rows → `Hydrator` / `CachedHydrator`. Drop / keep → `Filter`. Post-score VF-style work → `post_selection_hydrators` then `post_selection_filters`.
</Step>
<Step title="Implement update-only fields">
`hydrate` returns a default struct plus owned fields. `update` copies only those fields. Do not assign `user_features = hydrated.user_features` unless you own the whole struct (the live Phoenix hydrators copy one list at a time).
</Step>
<Step title="Preserve length and order">
Candidate hydrators: one `Result` per input, same index. Use `Ok(Default::default())` for misses. Use a filter to drop. Length mismatches are skipped, not retried.
</Step>
<Step title="Register on the pipeline">
Push into the matching `vec!` in `PhoenixCandidatePipeline::build_with_clients` (or For You's `build`). Declare `pub mod` in the crate `mod.rs`. `enable` should read `query.params` / `query.decider` the same way siblings do.
</Step>
<Step title="Verify">
`CandidatePipeline::components()` lists every registered name by `PipelineStage`. Filter spans expose `removed_per_filter [Name=N,…]`. Hydrator skips show `length_mismatch` warnings.
</Step>
</Steps>

## Next

<CardGroup>
<Card title="For You request lifecycle" href="/request-lifecycle">
`CandidatePipeline::execute` stages from query hydration through side effects.
</Card>
<Card title="Add a pipeline component" href="/add-pipeline-component">
Source, Hydrator, Filter, Scorer, Selector, QueryHydrator, and SideEffect contracts.
</Card>
<Card title="ScoredPostsQuery and gRPC" href="/scored-posts-query">
Query fields, QueryBuilder defaults, and response mapping.
</Card>
<Card title="Candidate sources" href="/candidate-sources">
What runs after query hydration and before these hydrators.
</Card>
<Card title="Scorers and weights" href="/scorers-and-weights">
What scores the `kept` set after pre-score filters.
</Card>
<Card title="Runtime boundaries" href="/runtime-boundaries">
What this checkout can run locally versus production Home Mixer.
</Card>
</CardGroup>