# Add a pipeline component

> Source, Hydrator, Filter, Scorer, Selector, QueryHydrator, and SideEffect contracts, including length-match and no-drop rules.

- 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

- `candidate-pipeline/candidate_pipeline.rs`
- `candidate-pipeline/source.rs`
- `candidate-pipeline/hydrator.rs`
- `candidate-pipeline/filter.rs`
- `candidate-pipeline/scorer.rs`
- `candidate-pipeline/selector.rs`
- `candidate-pipeline/query_hydrator.rs`
- `candidate-pipeline/side_effect.rs`

---

---
title: "Add a pipeline component"
description: "Source, Hydrator, Filter, Scorer, Selector, QueryHydrator, and SideEffect contracts, including length-match and no-drop rules."
---

`CandidatePipeline::execute` in `xai_candidate_pipeline` runs a typed `Q: PipelineQuery` / `C: PipelineCandidate` request through fixed stages. New behavior is a new trait object pushed into `PhoenixCandidatePipeline` or `ForYouCandidatePipeline`. Hydrators and scorers must return one result per input candidate in the same order; only a `Filter` or `Selector` may drop candidates.

<Warning>
Home Mixer and these traits depend on unpublished crates (`xai_candidate_pipeline::component_library`, feature-switch params, gRPC clients). This checkout can compile the trait sources and show registration, but it cannot run Home Mixer locally. The runnable surface in this repo is the Phoenix Python inference path.
</Warning>

## Choose a stage

| Stage | Trait | Parallelism | May drop candidates? | Failure behavior |
| --- | --- | --- | --- | --- |
| `QueryHydrator` | `QueryHydrator<Q>` | Parallel `hydrate`, then sequential `update` | n/a (query only) | `Err` is logged; that hydrator's `update` is skipped |
| `DependentQueryHydrator` | same trait, later wave | Same | n/a | Same |
| `Source` | `Source<Q, C>` | Parallel; results appended | n/a (creates candidates) | `Err` is logged and discarded by `flatten` |
| `Hydrator` / `PostSelectionHydrator` | `Hydrator<Q, C>` | Parallel `hydrate`, sequential `update_all` | **No** | Length mismatch skips the whole batch; per-item `Err` skips that `update` |
| `Filter` / `PostSelectionFilter` | `Filter<Q, C>` | Sequential | **Yes** — partition into `kept` / `removed` | Filters are sync and do not return `Result` |
| `Scorer` | `Scorer<Q, C>` | Sequential `run` + `update_all` | **No** | Same length-match skip as hydrators |
| `Selector` | `Selector<Q, C>` | Single | **Yes** — `selected` vs `non_selected` | If `enable` is false, all candidates stay selected |
| `SideEffect` | `SideEffect<Q, C>` | `tokio::spawn` + parallel | No effect on the returned feed | `join_all` result is discarded |

`PhoenixCandidatePipeline` fills every slot for `ScoredPostsQuery` / `PostCandidate`. `ForYouCandidatePipeline` uses `FeedItem` and returns empty slices for hydrators, filters, scorers, and post-selection stages; it only owns query hydrators, sources, `BlenderSelector`, and side effects.

```mermaid
flowchart TB
  subgraph queryLayer [Query layer - parallel hydrate, sequential update]
    QH["query_hydrators()"]
    DQH["dependent_query_hydrators()"]
  end
  subgraph fetchLayer [Fetch - parallel append]
    SRC["sources()"]
  end
  subgraph mutateLayer [Pre-score - hydrate cannot drop]
    HYD["hydrators() parallel"]
    FIL["filters() sequential"]
    SCO["scorers() sequential"]
  end
  subgraph selectLayer [Select and trim]
    SEL["selector()"]
    PSH["post_selection_hydrators()"]
    PSF["post_selection_filters()"]
    TRUNC["result_size() split_off"]
  end
  subgraph effectLayer [After return]
    SE["side_effects() spawned"]
  end
  QH --> DQH --> SRC --> HYD --> FIL --> SCO --> SEL --> PSH --> PSF --> TRUNC --> SE
```

## Shared surface

Every component is `Send + Sync`. `enable` defaults to `true`. `name()` defaults to the last `::` segment of `type_name_of_val(self)` via `util::short_type_name`. Tracing spans record `total_count`, `enabled_count`, and a comma-joined `disabled` list.

<ParamField body="enable" type="fn(&Q) -> bool" default="true">
Skip the component for this request. `SideEffect::enable` takes `Arc<Q>`, not `&Q`.
</ParamField>

<ParamField body="name" type="fn() -> &'static str" default="short type name">
Used in tracing, stats (`{name}.run`, `{name}.cache`, `{pipeline}.execute`), and `CandidatePipeline::components()`.
</ParamField>

`Q` must implement `PipelineQuery` (`Clone + Send + Sync + 'static`, plus `params()` and `decider()`). `ScoredPostsQuery` is the published query. `C` is any `Clone + Send + Sync + 'static` type; Home Mixer uses `PostCandidate` in Phoenix and `FeedItem` in For You.

`CandidatePipeline` also requires:

<ParamField body="result_size" type="fn() -> usize" required>
Hard cap after post-selection filters. Excess candidates move to `SideEffectInput.non_selected_candidates`. Phoenix uses `params::RESULT_SIZE`; For You uses `params::FOR_YOU_MAX_RESULT_SIZE`.
</ParamField>

<ParamField body="finalize" type="fn(&Q, &mut Vec&lt;C&gt;)" default="no-op">
Runs after truncation, before side effects. Neither published pipeline overrides it.
</ParamField>

<ParamField body="dependent_query_hydrators" type="&[Box&lt;dyn QueryHydrator&lt;Q&gt;&gt;]" default="&[]">
Second query-hydration wave. It sees fields written by the first wave. Neither published pipeline overrides this accessor.
</ParamField>

## Length-match and no-drop

`Hydrator::hydrate` and `Scorer::score` take `&[C]` and must return `Vec<Result<C, String>>` with **the same length and order**. The trait comments state dropping is not allowed — use a filter.

`Hydrator::run` / `Scorer::run` enforce that:

1. If `result.len() == candidates.len()`, the vector is passed to `update_all`.
2. Otherwise the stage logs `Skipped: length_mismatch expected=N got=M`, replaces the vector with `N` copies of `Err("… length_mismatch expected=N got=M")`, and `update_all` copies **no** fields.

`update_all` zips by index and calls `update` only on `Ok`. A per-candidate `Err` leaves that candidate unchanged.

`CachedHydrator` has a second check: `hydrate_from_client` must return one result per cache miss. A miss-list mismatch returns `Err("CachedHydrator length_mismatch …")` for **every** input candidate, including cache hits.

Use a defaulted sibling struct in `hydrate` / `score` and copy only owned fields in `update`. Sibling hydrators run in parallel on the pre-hydration slice, so they cannot read each other's new fields. Scorers run sequentially, so later scorers can read earlier scores.

<Info>
`CoreDataHydrationFilter` drops `author_id == 0` after `CoreDataCandidateHydrator`. Missing TES data is `Ok(PostCandidate::default())` (length preserved); the filter, not the hydrator, removes the candidate.
</Info>

## QueryHydrator

Fills `ScoredPostsQuery` fields before sources run. All enabled hydrators receive the **same original query** in parallel. `update` then applies each `Ok` result in vec order.

```rust
#[async_trait]
pub trait QueryHydrator<Q>: Any + Send + Sync
where
    Q: PipelineQuery,
{
    fn enable(&self, _query: &Q) -> bool { true }
    async fn hydrate(&self, query: &Q) -> Result<Q, String>;
    fn update(&self, query: &mut Q, hydrated: Q);
}
```

`update` must copy only this hydrator's fields. `FollowedUserIdsQueryHydrator` writes `user_features.followed_user_ids`. `RetrievalSequenceQueryHydrator` writes `retrieval_sequence` and `columnar_retrieval_sequence`.

Query hydrators cannot see sibling results in the first wave. A hydrator that needs `followed_user_ids` or `retrieval_sequence` belongs in `dependent_query_hydrators()`, or it must not depend on another hydrator's output.

`PhoenixCandidatePipeline::build_with_clients` constructs `ImpressedPostsQueryHydrator` into `_impressed_posts_hydrator` and never inserts it. Construction is not registration.

## Source

Creates candidates. Enabled sources run in parallel; `Ok` vectors are appended. Source order in the vec is the append order.

```rust
#[async_trait]
pub trait Source<Q, C>: Any + Send + Sync
where
    Q: PipelineQuery,
    C: PipelineCandidate,
{
    fn enable(&self, _query: &Q) -> bool { true }
    async fn source(&self, query: &Q) -> Result<Vec<C>, String>;
}
```

`Source::run` logs `Fetched N candidates` or `Failed: …`. A failed source contributes nothing; other sources still run.

Typical `enable` predicates:

- `ThunderSource`: `!query.has_cached_posts`
- `PhoenixSource`: not a non-bulk topic request, not new-user topic retrieval, `!query.in_network_only`, `!query.has_cached_posts`

`PhoenixSource` errors with `PhoenixSource: missing retrieval_sequence` when the query hydrator left that field empty. That is a source `Err`, not a hydrator length mismatch.

## Hydrator

Enriches candidates. Same trait is used for `hydrators()` and `post_selection_hydrators()`.

```rust
#[async_trait]
pub trait Hydrator<Q, C>: Any + Send + Sync {
    async fn hydrate(&self, query: &Q, candidates: &[C]) -> Vec<Result<C, String>>;
    fn update(&self, candidate: &mut C, hydrated: C);
}
```

`InNetworkCandidateHydrator` is the minimal pattern: map one output per input, set only `in_network`, copy that field in `update`.

TES-backed hydrators (`CoreDataCandidateHydrator`, `HasMediaHydrator`, `LanguageCodeHydrator`, …) implement `CachedHydrator` instead. The blanket `impl Hydrator for T: CachedHydrator` handles cache lookup. Implement:

| Method | Role |
| --- | --- |
| `cache_store` | `CacheStore<CacheKey, CacheValue>` (Home Mixer uses `MokaCache`) |
| `cache_key` | Usually `candidate.tweet_id` |
| `cache_value` / `hydrate_from_cache` | Round-trip the owned fields |
| `hydrate_from_client` | One `Result` per **miss**, same order as the miss slice |
| `update` | Copy owned fields onto the live candidate |

Successful client hydrations are inserted into the cache. `Err` results are not cached.

Post-selection hydrators (`VFCandidateHydrator`, brand-safety, `TweetTypeMetricsHydrator`, …) run only on the selected set.

## Filter

The only pre-score stage that may remove candidates. Filters run **sequentially**; each sees the previous `kept` list.

```rust
pub struct FilterResult<C> {
    pub kept: Vec<C>,
    pub removed: Vec<C>,
}

pub trait Filter<Q, C>: Any + Send + Sync {
    fn filter(&self, query: &Q, candidates: Vec<C>) -> FilterResult<C>;
}
```

`kept` continues. `removed` accumulates into `PipelineResult.filtered_candidates` (pre-score removals plus post-selection removals). Post-selection removals are **not** added to `SideEffectInput.non_selected_candidates`.

`DropDuplicatesFilter` keeps first `tweet_id`. `SelfTweetFilter` drops `author_id == query.user_id`. `TopicIdsFilter` overrides `enable` to `query.is_topic_request() || query.has_excluded_topics()`.

`Filter::run` records `kept_count`, `removed_count`, `filter_rate`, and increments `{name}.run` kept/removed stats.

Phoenix pre-score filters: `DropDuplicatesFilter`, `CoreDataHydrationFilter`, `AgeFilter`, `SelfTweetFilter`, `RetweetDeduplicationFilter`, `IneligibleSubscriptionFilter`, `PreviouslySeenPostsFilter`, `PreviouslySeenPostsBackupFilter`, `PreviouslyServedPostsFilter`, `MutedKeywordFilter`, `AuthorSocialgraphFilter`, `VideoFilter`, `TopicIdsFilter`, `NewUserTopicIdsFilter`.

Phoenix post-selection filters: `VFFilter`, `AncillaryVFFilter`, `DedupConversationFilter`.

## Scorer

Writes score fields. Scorers run **one at a time**: `run` then `update_all`, then the next scorer. Later scorers may read earlier fields.

```rust
#[async_trait]
pub trait Scorer<Q, C>: Send + Sync {
    async fn score(&self, query: &Q, candidates: &[C]) -> Vec<Result<C, String>>;
    fn update(&self, candidate: &mut C, scored: C);
}
```

Phoenix registers `PhoenixScorer`, `RankingScorer`, then `VMRanker`. `PhoenixScorer` is disabled when `query.has_cached_posts`. If `scoring_sequence` is missing it still returns `vec![Ok(PostCandidate::default()); candidates.len()]` so length-match holds.

`RankingScorer::update` copies `weighted_score` and `score`. `TopKScoreSelector` reads `candidate.score.unwrap_or(f64::NEG_INFINITY)`.

`home-mixer/scorers/weighted_scorer.rs` and `oon_scorer.rs` exist on disk but are not in `scorers/mod.rs` and are not in the Phoenix scorer vec. Registering a scorer requires both the module and the vec.

## Selector

Exactly one selector per pipeline. Default `select` sorts by `score()` descending and truncates to `size()` when `Some`.

```rust
pub struct SelectResult<C> {
    pub selected: Vec<C>,
    pub non_selected: Vec<C>,
}

pub trait Selector<Q, C>: Send + Sync {
    fn score(&self, candidate: &C) -> f64;
    fn size(&self) -> Option<usize> { None }
    fn select(&self, query: &Q, candidates: Vec<C>) -> SelectResult<C> { /* sort + split_off */ }
}
```

`TopKScoreSelector` (Phoenix) uses `params::TOP_K_CANDIDATES_TO_SELECT`. `BlenderSelector` (For You) overrides `select` to blend ads / prompts / who-to-follow / push-to-home and does not use the default sort path (`score()` returns `0.0`).

If `selector.enable(query)` is false, every candidate is selected and `non_selected` is empty.

After selection, `execute` still truncates `final_candidates` to `result_size()` and appends the tail to `non_selected`.

## SideEffect

Runs after `finalize`. Cannot change `PipelineResult`. `execute` returns while the spawned task is still running.

```rust
pub struct SideEffectInput<Q, C> {
    pub query: Arc<Q>,
    pub selected_candidates: Vec<C>,
    pub non_selected_candidates: Vec<C>,
}

#[async_trait]
pub trait SideEffect<Q, C>: Send + Sync {
    fn enable(&self, _query: Arc<Q>) -> bool { true }
    async fn side_effect(&self, input: Arc<SideEffectInput<Q, C>>) -> Result<(), String>;
}
```

`non_selected_candidates` is selector leftovers plus `result_size` overflow. It does not include filter removals.

`RedisPostCandidateCacheSideEffect::enable` is `is_prod() && !query.has_cached_posts`. `ScoredStatsSideEffect` always enables and samples stats from selected (and sometimes non-selected) candidates.

## Register the component

<Steps>
<Step title="Add the module">
Create the file next to siblings:

- Phoenix query hydrators: `home-mixer/query_hydrators/`
- Phoenix candidate hydrators: `home-mixer/candidate_hydrators/`
- Phoenix filters: `home-mixer/filters/`
- Phoenix scorers: `home-mixer/scorers/`
- Phoenix sources: `home-mixer/sources/`
- Phoenix side effects: `home-mixer/side_effects/`
- For You extras: same trees; `ForYouCandidatePipeline::build` owns the For You vecs

Declare `pub mod …;` in that folder's `mod.rs`.
</Step>

<Step title="Implement the trait">
Use `ScoredPostsQuery` plus `PostCandidate` (Phoenix) or `FeedItem` (For You). Override `enable` when the component is request-conditional. Keep `hydrate` / `score` length-matched. Copy only owned fields in `update`.
</Step>

<Step title="Push into the pipeline vec">
Phoenix: `PhoenixCandidatePipeline::build_with_clients`. For You: `ForYouCandidatePipeline::build`.

```rust
// Phoenix hydrators vec — same pattern for every stage
hydrators.push(Box::new(YourHydrator { /* clients */ }));
```

For You hydrators, filters, scorers, and post-selection stages currently return `&[]`. Adding one of those types to For You also requires storing a `Vec` on the struct and changing the `CandidatePipeline` accessor.
</Step>

<Step title="Verify">
Confirm `CandidatePipeline::components()` lists the new `name()`. Hydrator/scorer length bugs show up as `Skipped: length_mismatch` and unchanged candidate fields. Source `Err` only appears as a missing batch, not a pipeline abort. Side effects must not be required for a correct `PipelineResult`.
</Step>
</Steps>

### Phoenix registration order

`PhoenixCandidatePipeline::build_with_clients` currently wires:

| Accessor | Types |
| --- | --- |
| `query_hydrators` | Scoring + retrieval sequences, blocked/muted/followed/subscribed IDs, cached posts, mutual follow, demographics, Grok topics, starter packs, inferred topics, impression bloom, IP, inferred gender |
| `sources` | `ThunderSource`, `TweetMixerSource`, `PhoenixSource`, `PhoenixTopicsSource`, `PhoenixMOESource`, `CachedPostsSource` |
| `hydrators` | In-network, TES core/quote/video/media/subscription/language, Gizmoduck, blocked-by, filtered topics |
| `filters` | Dedup through topic filters listed above |
| `scorers` | `PhoenixScorer`, `RankingScorer`, `VMRanker` |
| `selector` | `TopKScoreSelector` |
| `post_selection_hydrators` | VF, ads brand safety, tweet-type metrics, following-replied, mutual-follow Jaccard |
| `post_selection_filters` | VF, ancillary VF, conversation dedup |
| `side_effects` | Phoenix experiments Kafka, reranking Kafka, Redis cache, scored stats, mutual-follow stats, Phoenix request cache |

### For You registration order

| Accessor | Types |
| --- | --- |
| `query_hydrators` | `ServedHistoryQueryHydrator`, `PastRequestTimestampsQueryHydrator` |
| `sources` | `ScoredPostsSource`, `AdsSource`, `WhoToFollowSource`, `PromptsSource`, `PushToHomeSource` |
| `selector` | `BlenderSelector` |
| `side_effects` | Ads injection log, seen-ids Kafka, served-candidates Kafka, client events, For You stats, past-request timestamps, served-history update + truncate |

## Error and skip matrix

| Symptom | Likely cause |
| --- | --- |
| Component never appears in traces | Not in the pipeline vec, or `mod.rs` omitted |
| `disabled=` in the stage span | `enable` returned false |
| Hydrator/scorer fields stay `None` / default | Length mismatch, or `update` does not copy the field, or per-item `Err` |
| Query field stays default | `hydrate` returned `Err`, or two hydrators overwrite the same field, or the hydrator is first-wave but needs a sibling's output |
| Missing candidate batch, pipeline still returns | Source `Err` flattened away |
| Feed unchanged by a "post-process" | Implemented as `SideEffect` (fire-and-forget) instead of hydrator/filter/scorer |
| For You hydrator never runs | `ForYouCandidatePipeline::hydrators()` returns `&[]` |

`PipelineResult` after `execute`:

<ResponseField name="retrieved_candidates" type="Vec<C>">
Candidates after the first hydrate wave, before pre-score filters.
</ResponseField>

<ResponseField name="filtered_candidates" type="Vec<C>">
Union of pre-score and post-selection `removed` lists.
</ResponseField>

<ResponseField name="selected_candidates" type="Vec<C>">
Post-selection kept set after `result_size` truncation and `finalize`.
</ResponseField>

<ResponseField name="query" type="Arc<Q>">
Query after both hydration waves.
</ResponseField>

## Next

<CardGroup>
<Card title="For You request lifecycle" href="/request-lifecycle">
Stage order inside `CandidatePipeline::execute` and how For You wraps Phoenix.
</Card>
<Card title="Candidate sources" href="/candidate-sources">
Enable predicates, cluster resolution, and `served_type` assignment.
</Card>
<Card title="Filters and hydrators" href="/filters-and-hydrators">
Pre-score vs post-selection lists and query-hydrator field ownership.
</Card>
<Card title="Scorers and weights" href="/scorers-and-weights">
`PhoenixScorer`, `RankingScorer`, diversity, and `TopKScoreSelector`.
</Card>
<Card title="Runtime boundaries" href="/runtime-boundaries">
What this checkout can run versus unpublished Home Mixer crates.
</Card>
</CardGroup>