diff --git a/.github/workflows/convergence-tests.yml b/.github/workflows/convergence-tests.yml
index dc4f2305a1..dd0fd0b45b 100644
--- a/.github/workflows/convergence-tests.yml
+++ b/.github/workflows/convergence-tests.yml
@@ -22,6 +22,7 @@ on:
           - esm2_native_te_3b
           - esm2_native_te_15b
           - codonfm_ptl_te
+          - evo2_sae_smoke
       branch:
         description: "Branch to use (ignored if commit SHA is provided)"
         required: true
diff --git a/.github/workflows/unit-tests-interpretability-recipes.yaml b/.github/workflows/unit-tests-interpretability-recipes.yaml
new file mode 100644
index 0000000000..c14f5b9f53
--- /dev/null
+++ b/.github/workflows/unit-tests-interpretability-recipes.yaml
@@ -0,0 +1,149 @@
+name: "BioNeMo Interpretability Recipes CI"
+
+# CI for the SAE interpretability recipes under interpretability/sparse_autoencoders/recipes/*.
+# Generic matrix, modeled on the repo-wide unit-tests-recipes.yml but scoped to the interp subtree:
+#   * `changed-dirs` (cheap ubuntu) discovers which interp recipes to test and emits a matrix.
+#   * `unit-tests` runs each selected recipe on the L4: its own `.ci_build.sh` + `pytest tests/`.
+#
+# Eligible recipes = any interp recipe with its own `.ci_build.sh`. Today that's `evo2`; codonfm/esm2
+# have no `.ci_build.sh`/tests yet, so they are skipped until they add them. This also makes the lane
+# a green no-op before the evo2 SAE recipe lands (#1622) — the presence guard is "has a .ci_build.sh".
+#
+# What runs when:
+#   * change under a recipe's own dir            -> that recipe.
+#   * change to the shared `sae` lib, or to this workflow file, or the nightly schedule
+#                                                -> ALL eligible recipes (they all depend on sae).
+#   * change to an unrelated recipe / elsewhere  -> nothing (empty matrix, green no-op).
+# Each recipe's `.ci_build.sh` owns its own build (evo2 -> mbridge bionemo.evo2; esm2 -> HF; etc.),
+# so a codonfm/esm2 change never triggers the Evo2 megatron build, and vice-versa.
+
+on:
+  push:
+    branches:
+      - "pull-request/[0-9]+"
+      - "dependabot/**"
+  merge_group:
+    types: [checks_requested]
+  schedule:
+    - cron: "0 9 * * *" # Runs at 9 AM UTC daily (2 AM MST)
+
+defaults:
+  run:
+    shell: bash -x -e -u -o pipefail {0}
+
+concurrency:
+  group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }}
+  cancel-in-progress: true
+
+jobs:
+  changed-dirs:
+    # Cheap ubuntu pre-job: decide which interp recipes to test and emit the matrix.
+    runs-on: ubuntu-latest
+    permissions:
+      contents: read
+    outputs:
+      dirs: ${{ steps.set-dirs.outputs.dirs }}
+    steps:
+      - name: Checkout repository
+        uses: actions/checkout@34e114876b0b11c390a56381ad16ebd13914f8d5
+        with:
+          fetch-depth: 0
+          persist-credentials: false
+
+      - name: Determine which interp recipes to run
+        id: set-dirs
+        env:
+          EVENT_NAME: ${{ github.event_name }}
+        run: |
+          RROOT="interpretability/sparse_autoencoders/recipes"
+          SAE="interpretability/sparse_autoencoders/sae"
+          WF=".github/workflows/unit-tests-interpretability-recipes.yaml"
+
+          # Eligible recipes = those with their own .ci_build.sh (others are green no-ops until
+          # they add one; empty before #1622 lands -> whole lane is a no-op).
+          # Use `if` (not `[ -f ] && echo`): the short-circuit leaves exit 1 when the last dir
+          # lacks a .ci_build.sh, which `set -e` would turn into a job failure instead of an
+          # empty (green no-op) matrix.
+          ALL=$(for d in "$RROOT"/*/; do if [ -f "${d}.ci_build.sh" ]; then echo "${d%/}"; fi; done \
+                | jq -R -s -c 'split("\n") | map(select(length > 0))')
+          echo "Eligible recipes (have .ci_build.sh): $ALL"
+
+          MERGE_BASE=$(git merge-base HEAD origin/main)
+          CHANGED=$(git diff --name-only "$MERGE_BASE" HEAD)
+          echo "Changed files:"; echo "$CHANGED" | sed 's/^/  - /'
+
+          # sae lib / this workflow / nightly -> run EVERY eligible recipe (all depend on sae).
+          if [ "$EVENT_NAME" = "schedule" ] || echo "$CHANGED" | grep -qE "^(${SAE}/|${WF}$)"; then
+            DIRS="$ALL"
+          else
+            # otherwise -> only eligible recipes that have a changed file under them.
+            DIRS=$(echo "$ALL" | jq -c --arg changed "$CHANGED" '
+              ($changed | split("\n")) as $cf
+              | map(select(. as $d | $cf | any(startswith($d + "/"))))
+            ')
+          fi
+          echo "Recipes to run: $DIRS"
+
+          # Attach the runner image (the dockerhub-cached squashed pytorch, like the repo-wide lane).
+          IMG="svcbionemo023/bionemo-framework:pytorch26.04-py3-squashed"
+          MATRIX=$(echo "$DIRS" | jq -c --arg img "$IMG" \
+            'map({dir: ., name: (. | split("/") | last), image: $img})')
+          echo "dirs=$MATRIX" >> "$GITHUB_OUTPUT"
+          echo "matrix: $MATRIX"
+
+  unit-tests:
+    needs: changed-dirs
+    if: ${{ needs.changed-dirs.outputs.dirs != '' && needs.changed-dirs.outputs.dirs != '[]' }}
+    runs-on: linux-amd64-gpu-l4-latest-1
+    name: "interp-unit-tests (${{ matrix.recipe.name }})"
+    permissions:
+      contents: read
+    container:
+      image: ${{ matrix.recipe.image }}
+      options: --shm-size=16G
+      env:
+        CI: true
+        HF_TOKEN: ${{ secrets.HF_TOKEN }}
+        HF_HOME: /cache/huggingface
+    strategy:
+      fail-fast: false
+      matrix:
+        recipe: ${{ fromJson(needs.changed-dirs.outputs.dirs) }}
+
+    steps:
+      - name: Show GPU info
+        run: nvidia-smi
+
+      - name: Setup proxy cache
+        uses: nv-gha-runners/setup-proxy-cache@14229018fe157c83e03c008f27d183d8e99bc67c
+
+      - name: Checkout repository
+        uses: actions/checkout@34e114876b0b11c390a56381ad16ebd13914f8d5
+        with:
+          # Interp recipes live under interpretability/sparse_autoencoders (recipe + shared sae);
+          # evo2 additionally builds on recipes/evo2_megatron. Check out both so any recipe's
+          # .ci_build.sh has what it needs.
+          sparse-checkout: |
+            interpretability/sparse_autoencoders
+            recipes/evo2_megatron
+          sparse-checkout-cone-mode: false
+          persist-credentials: false
+
+      - name: Cache Hugging Face models
+        uses: actions/cache@0057852bfaa89a56745cba8c7296529d2fc39830
+        with:
+          path: /cache/huggingface
+          key: ${{ runner.os }}-huggingface-interp-${{ matrix.recipe.name }}-${{ github.sha }}
+          restore-keys: |
+            ${{ runner.os }}-huggingface-interp-${{ matrix.recipe.name }}-
+            ${{ runner.os }}-huggingface-
+
+      - name: Install dependencies
+        working-directory: ${{ matrix.recipe.dir }}
+        run: bash .ci_build.sh
+
+      - name: Run tests
+        working-directory: ${{ matrix.recipe.dir }}
+        run: |
+          [ -f .ci_test_env.sh ] && source .ci_test_env.sh
+          pytest -v tests/
diff --git a/ci/lepton/model_convergence/configs/recipes/evo2_sae_smoke.yaml b/ci/lepton/model_convergence/configs/recipes/evo2_sae_smoke.yaml
new file mode 100644
index 0000000000..4e6d0e9166
--- /dev/null
+++ b/ci/lepton/model_convergence/configs/recipes/evo2_sae_smoke.yaml
@@ -0,0 +1,194 @@
+# @package _global_
+defaults:
+  - /base
+  - _self_
+
+############################################################
+# lepton job info  (node_group + mount_from are inherited from /base)
+############################################################
+num_nodes: 1
+device_type: gpu
+# Two GPUs as a producer/consumer pipeline (NOT data parallel): the Evo2 producer runs
+# forwards on cuda:0 and the SAE consumer trains on cuda:1 (train_streaming.py auto-places
+# the consumer on the 2nd GPU). Still one predict rank: --nproc_per_node 1, --dp-size 1.
+num_devices: 2
+gpu_type: h100-sxm
+resource_shape: "${device_type}.${num_devices}x${gpu_type}"
+
+job_name: "evo2-sae-1b-smoke"
+
+############################################################
+# Telemetry
+############################################################
+# Disabled by default because this is an example smoke run. Set both
+# log_to_wandb=true and train_wandb_enabled=true to collect W&B metrics.
+log_to_wandb: false
+log_to_kratos: false
+kratos_subject: "interpretability_sae_examples_v0.0.1"
+wandb_dir: /workspace/bionemo-framework/interpretability/sparse_autoencoders/recipes/evo2/wandb
+
+############################################################
+# recipe identifiers
+############################################################
+recipe_subdir: interpretability/sparse_autoencoders/recipes/evo2
+model_type: evo2_sae
+variant: smoke
+framework: sae
+precision: bf16
+
+total_gpus: ${multiply:${num_devices},${num_nodes}}
+
+wandb_init_args:
+  project: "interpretability_sae_examples__${sanitize:${branch}}"
+  group: "${model_type}__${variant}__${total_gpus}gpus__${sanitize:${gpu_type}}"
+  job_type: "${recipe_subdir}"
+  name: null
+  entity: clara-discovery
+  mode: online
+
+############################################################
+# Evo2 SAE streaming smoke settings
+############################################################
+run_name: lepton_evo2_1b_sae_smoke
+
+# Fetched by identifier and converted to MBridge in-job; nothing is pre-staged.
+model_tag: evo2/1b-8k-bf16:1.0
+model_size: evo2_1b_base
+seq_length: 8192
+bionemo_data_source: ""  # set to "pbss" if NGC is unreachable from the job
+ckpt_dir: ${output_base}/evo2_1b_mbridge
+layer: 12
+input_dim: 1920  # 1B residual-stream width (Hyena1bModelProvider.hidden_size)
+micro_batch_size: 4
+max_tokens: 100000
+chunk_bp: 8192
+extract_dtype: fp32
+
+dp_size: 1
+train_n_epochs: 1
+max_steps: 20
+train_batch_size: 256
+expansion_factor: 8
+top_k: 32
+auxk: 256
+learning_rate: 1e-4
+log_interval: 10
+checkpoint_steps: 999999
+init_pre_bias: false
+train_wandb_enabled: false
+queue_size: 4
+shuffle_buffer_size: 8192
+
+data_dir: /data/interpretability/sae/evo2_smoke/data
+output_base: /data/interpretability/sae/evo2_smoke/outputs
+
+products:
+  - config: evo2_1b_smoke
+
+############################################################
+# Checkout Script
+############################################################
+checkout_script: |
+  set -euo pipefail
+  cd /workspace
+  git clone https://github.com/NVIDIA/bionemo-framework.git
+  cd bionemo-framework
+  if [ -n "${commit_sha}" ]; then
+    echo "Checking out commit: ${commit_sha}"
+    git checkout "${commit_sha}"
+  elif [ "${branch}" != "main" ]; then
+    echo "Checking out branch: ${branch}"
+    git checkout "${branch}"
+  fi
+  # Builds evo2_megatron's venv (bionemo.evo2 / predict) + installs sae + evo2_sae on top.
+  cd interpretability/sparse_autoencoders/recipes/evo2
+  bash .ci_build.sh
+
+############################################################
+# run script
+############################################################
+run_script: |
+  cd /workspace/bionemo-framework/interpretability/sparse_autoencoders/recipes/evo2
+  # The recipe shares evo2_megatron's venv (see .ci_build.sh / .ci_test_env.sh).
+  source ../../../../recipes/evo2_megatron/.venv/bin/activate
+
+  WANDB_FLAG="--no-wandb"
+  if [ "${train_wandb_enabled}" = "true" ]; then
+    WANDB_FLAG="--wandb"
+  fi
+
+  INIT_PRE_BIAS_FLAG="--no-init-pre-bias"
+  if [ "${init_pre_bias}" = "true" ]; then
+    INIT_PRE_BIAS_FLAG="--init-pre-bias"
+  fi
+
+  # Read EVO2_CKPT_DIR via printenv so Hydra does not treat it as an interpolation.
+  CKPT_DIR="$(printenv EVO2_CKPT_DIR || true)"
+  if [ -z "$CKPT_DIR" ]; then
+    if [ -n "${bionemo_data_source}" ]; then
+      export BIONEMO_DATA_SOURCE="${bionemo_data_source}"
+    fi
+    python scripts/prepare_1b_checkpoint.py \
+      --mbridge-ckpt-dir ${ckpt_dir} \
+      --model-tag ${model_tag} \
+      --model-size ${model_size} \
+      --seq-length ${seq_length}
+    CKPT_DIR="${ckpt_dir}"
+  fi
+
+  # Synthesize a tiny FASTA and chunk it to <= chunk_bp bp.
+  mkdir -p "${data_dir}"
+  RAW_FASTA="${data_dir}/smoke_raw.fasta"
+  CHUNKED_FASTA="${data_dir}/smoke_chunked${chunk_bp}.fasta"
+  python - "$RAW_FASTA" <<'PY'
+  import random, sys
+  random.seed(42)
+  with open(sys.argv[1], "w") as f:
+      for i in range(4):
+          seq = "".join(random.choice("ACGT") for _ in range(4096))
+          f.write(f">smoke_{i}\n{seq}\n")
+  PY
+  python scripts/chunk_fasta.py --input "$RAW_FASTA" --output "$CHUNKED_FASTA" --window ${chunk_bp}
+
+  # Producer/consumer streaming: predict feeds activations into a queue; the SAE
+  # Trainer consumes them. Under torchrun (predict needs the distributed env).
+  torchrun --nproc_per_node 1 scripts/train_streaming.py \
+    --ckpt-dir "$CKPT_DIR" \
+    --fasta "$CHUNKED_FASTA" \
+    --embedding-layer ${layer} \
+    --input-dim ${input_dim} \
+    --micro-batch-size ${micro_batch_size} \
+    --max-tokens ${max_tokens} \
+    --dtype ${extract_dtype} \
+    --model-type topk \
+    --expansion-factor ${expansion_factor} \
+    --top-k ${top_k} \
+    --normalize-input \
+    --auxk ${auxk} \
+    --auxk-coef 0.03125 \
+    $INIT_PRE_BIAS_FLAG \
+    --n-epochs ${train_n_epochs} \
+    --max-steps ${max_steps} \
+    --batch-size ${train_batch_size} \
+    --lr ${learning_rate} \
+    --log-interval ${log_interval} \
+    --queue-size ${queue_size} \
+    --shuffle-buffer-size ${shuffle_buffer_size} \
+    $WANDB_FLAG \
+    --wandb-project ${wandb_init_args.project} \
+    --wandb-run-name ${run_name} \
+    --wandb-group ${wandb_init_args.group} \
+    --wandb-job-type ${wandb_init_args.job_type} \
+    --dp-size ${dp_size} \
+    --seed 42 \
+    --output-dir ${output_base}/${run_name} \
+    --checkpoint-dir ${output_base}/${run_name}/checkpoints \
+    --checkpoint-steps ${checkpoint_steps}
+
+  # Fail loudly on a silent no-op (streaming produced nothing / training skipped).
+  CKPT="${output_base}/${run_name}/checkpoints/checkpoint_final.pt"
+  if [ ! -f "$CKPT" ]; then
+    echo "ERROR: expected SAE checkpoint not written: $CKPT" >&2
+    exit 1
+  fi
+  echo "SMOKE OK: $CKPT"
diff --git a/interpretability/sparse_autoencoders/recipes/evo2/scripts/prepare_1b_checkpoint.py b/interpretability/sparse_autoencoders/recipes/evo2/scripts/prepare_1b_checkpoint.py
new file mode 100644
index 0000000000..e3ad9e8091
--- /dev/null
+++ b/interpretability/sparse_autoencoders/recipes/evo2/scripts/prepare_1b_checkpoint.py
@@ -0,0 +1,80 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: LicenseRef-Apache2
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Fetch an Evo2 NeMo2 checkpoint by identifier and convert it to MBridge.
+
+``bionemo_load(tag)`` (NGC/PBSS) followed by ``run_nemo2_to_mbridge(...)``, so no
+checkpoint needs to be pre-staged. Idempotent: if ``<mbridge-ckpt-dir>/iter_0000001``
+already exists it is reused. The resulting parent dir is what ``predict --ckpt-dir``
+expects.
+
+    python prepare_1b_checkpoint.py --mbridge-ckpt-dir /data/.../evo2_1b_mbridge
+"""
+
+from __future__ import annotations
+
+import argparse
+from pathlib import Path
+
+
+def parse_args() -> argparse.Namespace:
+    """Parse command-line arguments."""
+    p = argparse.ArgumentParser(
+        description="Fetch + convert an Evo2 NeMo2 checkpoint to MBridge",
+        formatter_class=argparse.ArgumentDefaultsHelpFormatter,
+    )
+    p.add_argument("--mbridge-ckpt-dir", type=Path, required=True, help="Output MBridge checkpoint parent dir")
+    p.add_argument("--model-tag", type=str, default="evo2/1b-8k-bf16:1.0", help="bionemo_load identifier (NGC/PBSS)")
+    p.add_argument("--model-size", type=str, default="evo2_1b_base", help="MODEL_OPTIONS key for the converter")
+    p.add_argument("--seq-length", type=int, default=8192)
+    p.add_argument("--mixed-precision-recipe", type=str, default="bf16_mixed")
+    p.add_argument("--vortex-style-fp8", action=argparse.BooleanOptionalAction, default=False)
+    return p.parse_args()
+
+
+def main() -> None:
+    """Fetch the NeMo2 checkpoint and convert it to MBridge (idempotent)."""
+    args = parse_args()
+    iter_dir = args.mbridge_ckpt_dir / "iter_0000001"
+    if iter_dir.exists():
+        print(f"Reusing existing MBridge checkpoint: {iter_dir}")
+        return
+
+    # bionemo.common on the migrated recipes layout; bionemo.core on older builds.
+    try:
+        from bionemo.common.data.load import load as bionemo_load
+    except ImportError:
+        from bionemo.core.data.load import load as bionemo_load
+    from bionemo.evo2.data.dataset_tokenizer import DEFAULT_HF_TOKENIZER_MODEL_PATH_512
+    from bionemo.evo2.utils.checkpoint.nemo2_to_mbridge import run_nemo2_to_mbridge
+
+    print(f"Fetching {args.model_tag} (set BIONEMO_DATA_SOURCE=pbss if NGC is unavailable) ...")
+    nemo2_ckpt_path = bionemo_load(args.model_tag)
+
+    args.mbridge_ckpt_dir.parent.mkdir(parents=True, exist_ok=True)
+    res_dir = run_nemo2_to_mbridge(
+        nemo2_ckpt_dir=nemo2_ckpt_path,
+        tokenizer_path=DEFAULT_HF_TOKENIZER_MODEL_PATH_512,
+        mbridge_ckpt_dir=args.mbridge_ckpt_dir,
+        model_size=args.model_size,
+        seq_length=args.seq_length,
+        mixed_precision_recipe=args.mixed_precision_recipe,
+        vortex_style_fp8=args.vortex_style_fp8,
+    )
+    print(f"MBridge checkpoint ready: {res_dir}/iter_0000001")
+
+
+if __name__ == "__main__":
+    main()
diff --git a/interpretability/sparse_autoencoders/recipes/evo2/scripts/train_streaming.py b/interpretability/sparse_autoencoders/recipes/evo2/scripts/train_streaming.py
new file mode 100644
index 0000000000..fc6542c45c
--- /dev/null
+++ b/interpretability/sparse_autoencoders/recipes/evo2/scripts/train_streaming.py
@@ -0,0 +1,360 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: LicenseRef-Apache2
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+r"""Train an Evo2 SAE with producer-consumer activation streaming.
+
+This is the Evo2 analogue of the ESM2 ``train_streaming.py``. It does NOT
+materialize an ActivationStore on disk: a background producer thread runs Evo2
+(Megatron) forward passes and pushes flattened layer activations into a bounded
+queue; the SAE ``Trainer`` consumes re-batched activation tensors from that
+queue in the main process.
+
+Bridge to Megatron
+------------------
+Evo2's forward pass is ``bionemo.evo2.run.predict``, which owns its own batch
+loop and exposes no generator API -- it only calls a module-level
+``_write_predictions_batch(...)`` callback per batch. We reuse the same hook
+``extract.py`` uses: monkeypatch ``predict._write_predictions_batch`` with a
+writer that, instead of appending to a parquet store, pushes the flattened
+``[n_tokens, hidden_dim]`` activations onto a queue. ``predict.main()`` runs in
+a daemon thread; this script's producer generator drains the queue and yields
+chunks to ``sae.streaming``. Converting predict's push-callback into a
+pull-generator REQUIRES a thread + queue (predict cannot be paused mid-loop).
+
+Launch
+------
+Run under ``torchrun --nproc_per_node 1`` (NOT bare ``python``): ``predict``
+calls ``initialize_inference_distributed`` and needs ``RANK``/``WORLD_SIZE``/
+``MASTER_*``. It is one predict rank (``--dp-size 1``); this is a producer/consumer
+pipeline, NOT data parallelism.
+
+Two GPUs: because producer and consumer are separate stages, a second GPU is used to
+run them concurrently rather than to data-parallel. When >1 GPU is visible the SAE
+consumer trains on ``cuda:1`` while the Evo2 producer runs its forwards on ``cuda:0``
+(LOCAL_RANK 0), so the two stages stop contending for one device. Activations cross
+the boundary as CPU tensors via the queue, so no NCCL/DDP is involved. ``--dp-size``
+stays 1; pass ``--device`` to override the consumer's GPU.
+
+    torchrun --nproc_per_node 1 train_streaming.py \
+        --ckpt-dir CKPT --fasta SEQS.fasta --embedding-layer 12 --input-dim 1920 \
+        --max-tokens 100000 --micro-batch-size 4 \
+        --model-type topk --expansion-factor 8 --top-k 32 --normalize-input \
+        --n-epochs 1 --batch-size 256 --lr 1e-4 --no-init-pre-bias --no-wandb \
+        --output-dir OUT --checkpoint-dir OUT/checkpoints
+"""
+
+from __future__ import annotations
+
+import argparse
+import queue
+import sys
+import tempfile
+import threading
+
+import torch
+from sae.architectures import ReLUSAE, TopKSAE
+from sae.perf_logger import PerfLogger
+from sae.streaming import StreamingConfig, make_streaming_dataloader
+from sae.training import ParallelConfig, Trainer, TrainingConfig, WandbConfig
+from sae.utils import get_device, set_seed
+
+
+# Stored precision for activations. bf16 is excluded: NumPy/Arrow (and the SAE's
+# float math downstream) want a numpy-representable dtype, so we cast Evo2's bf16
+# residual stream to fp32/fp16 on the device->host copy (mirrors extract.py).
+_DTYPES = {"fp32": torch.float32, "fp16": torch.float16}
+
+# Sentinel marking the producer thread finished (unique object; never a chunk).
+_DONE = object()
+
+
+def parse_args() -> argparse.Namespace:
+    """Parse command-line arguments."""
+    p = argparse.ArgumentParser(
+        description="Train an SAE directly from streamed Evo2 activations (no disk store)",
+        formatter_class=argparse.ArgumentDefaultsHelpFormatter,
+    )
+
+    src = p.add_argument_group("Evo2 producer")
+    src.add_argument("--ckpt-dir", type=str, required=True, help="Evo2 MBridge checkpoint directory")
+    src.add_argument("--fasta", type=str, required=True, help="Input FASTA (chunk long sequences first)")
+    src.add_argument("--embedding-layer", "--layer", dest="layer", type=int, required=True, help="Layer to extract")
+    src.add_argument("--input-dim", type=int, required=True, help="Residual-stream width of --embedding-layer")
+    src.add_argument("--micro-batch-size", type=int, default=4, help="Sequences per Evo2 forward pass")
+    src.add_argument("--max-tokens", type=int, default=0, help="Cap activation rows produced (0 = no cap)")
+    src.add_argument("--dtype", choices=list(_DTYPES), default="fp32", help="Activation cast (bf16 not storable)")
+
+    sae_group = p.add_argument_group("SAE model")
+    sae_group.add_argument("--model-type", type=str, default="topk", choices=["topk", "relu"])
+    sae_group.add_argument("--expansion-factor", type=int, default=8)
+    sae_group.add_argument("--top-k", type=int, default=32)
+    sae_group.add_argument("--normalize-input", action=argparse.BooleanOptionalAction, default=False)
+    sae_group.add_argument("--auxk", type=int, default=None)
+    sae_group.add_argument("--auxk-coef", type=float, default=1 / 32)
+    sae_group.add_argument("--dead-tokens-threshold", type=int, default=10_000_000)
+    sae_group.add_argument("--aggregate-loss", action=argparse.BooleanOptionalAction, default=False)
+    sae_group.add_argument("--dead-count-global", action=argparse.BooleanOptionalAction, default=False)
+    sae_group.add_argument("--init-pre-bias", action=argparse.BooleanOptionalAction, default=False)
+    sae_group.add_argument("--pre-bias-sample-size", type=int, default=32768)
+    sae_group.add_argument("--l1-coeff", type=float, default=1e-2, help="L1 coefficient (relu only)")
+
+    train_group = p.add_argument_group("SAE training")
+    train_group.add_argument("--lr", type=float, default=1e-4)
+    train_group.add_argument("--n-epochs", type=int, default=1)
+    train_group.add_argument("--max-steps", type=int, default=None, help="Exact optimizer-step budget")
+    train_group.add_argument("--batch-size", type=int, default=1024, help="Activation rows per SAE training batch")
+    train_group.add_argument("--log-interval", type=int, default=50)
+    train_group.add_argument("--max-grad-norm", type=float, default=None)
+    train_group.add_argument("--grad-accumulation-steps", type=int, default=1)
+    train_group.add_argument("--warmup-steps", type=int, default=0)
+    train_group.add_argument("--lr-schedule", choices=["constant", "cosine", "linear"], default="constant")
+    train_group.add_argument("--lr-min", type=float, default=0.0)
+    train_group.add_argument("--lr-decay-steps", type=int, default=None)
+
+    stream_group = p.add_argument_group("Producer-consumer streaming")
+    stream_group.add_argument("--queue-size", type=int, default=4, help="Activation chunks buffered (backpressure)")
+    stream_group.add_argument("--shuffle-buffer-size", type=int, default=65536)
+    stream_group.add_argument("--drop-last", action=argparse.BooleanOptionalAction, default=False)
+
+    wb_group = p.add_argument_group("Weights & Biases")
+    wb_group.add_argument("--wandb", action=argparse.BooleanOptionalAction, default=False, dest="wandb_enabled")
+    wb_group.add_argument("--wandb-project", type=str, default="evo2-sae-v2-diverse")
+    wb_group.add_argument("--wandb-run-name", type=str, default=None)
+    wb_group.add_argument("--wandb-group", type=str, default=None)
+    wb_group.add_argument("--wandb-job-type", type=str, default=None)
+
+    p.add_argument("--checkpoint-dir", type=str, default=None)
+    p.add_argument("--checkpoint-steps", type=int, default=None)
+    p.add_argument("--resume-from", type=str, default=None)
+    p.add_argument("--output-dir", type=str, default="./outputs")
+    p.add_argument("--seed", type=int, default=42)
+    p.add_argument("--device", type=str, default=None, help="Device for SAE training")
+    p.add_argument("--dp-size", type=int, default=1, help="Only dp-size=1 is supported by this streaming script")
+    return p.parse_args()
+
+
+def build_sae(args: argparse.Namespace, input_dim: int) -> torch.nn.Module:
+    """Build an SAE model (mirrors evo2 train.py:build_sae)."""
+    hidden_dim = input_dim * args.expansion_factor
+    if args.model_type == "topk":
+        return TopKSAE(
+            input_dim=input_dim,
+            hidden_dim=hidden_dim,
+            top_k=args.top_k,
+            normalize_input=args.normalize_input,
+            auxk=args.auxk,
+            auxk_coef=args.auxk_coef,
+            dead_tokens_threshold=args.dead_tokens_threshold,
+            aggregate_loss=args.aggregate_loss,
+            dead_count_global=args.dead_count_global,
+        )
+    return ReLUSAE(input_dim=input_dim, hidden_dim=hidden_dim, l1_coeff=args.l1_coeff)
+
+
+class Evo2ActivationProducer:
+    """Producer factory bridging Evo2 ``predict`` (push) to a pull-generator.
+
+    Calling an instance returns a fresh generator that runs ``predict.main()`` in
+    a daemon thread (with the per-batch writer monkeypatched to enqueue flattened
+    activations) and yields ``[n_tokens, hidden_dim]`` fp32 CPU chunks until the
+    thread finishes. Exceptions from the predict thread are re-raised in the
+    consumer so a dead producer fails loudly instead of hanging.
+    """
+
+    def __init__(self, args: argparse.Namespace):
+        """Store the parsed args; the activation generator is built lazily on ``__call__``."""
+        self.args = args
+
+    def _make_writer(self, q: "queue.Queue", state: dict):
+        """Return a drop-in replacement for predict._write_predictions_batch."""
+        cast = _DTYPES[self.args.dtype]
+        budget = self.args.max_tokens
+
+        def writer(
+            predictions,
+            output_dir,
+            batch_idx,
+            global_rank,
+            dp_rank,
+            files_per_subdir=None,
+            num_files_written=0,
+            data_parallel_world_size=1,
+        ):
+            if not predictions:
+                return output_dir, num_files_written, 0
+            # Past the row budget: keep running cheap forwards but stop enqueuing.
+            if budget and state["n_tokens"] >= budget:
+                return output_dir, num_files_written, 0
+            hidden = predictions["hidden_embeddings"]  # [B, S, H]
+            mask = predictions["pad_mask"].bool()
+            flat = hidden[mask].to(cast).cpu()  # [N_unpadded_tokens, H]
+            q.put(flat)
+            state["n_tokens"] += flat.shape[0]
+            return output_dir, num_files_written + 1, 0
+
+        return writer
+
+    def __call__(self):
+        """Return a fresh generator yielding ``[n_tokens, hidden_dim]`` CPU activation chunks."""
+        args = self.args
+        q: "queue.Queue" = queue.Queue(maxsize=args.queue_size)
+        state = {"n_tokens": 0}
+
+        from bionemo.evo2.run import predict as predict_mod  # lazy: heavy Megatron import
+
+        def run_predict() -> None:
+            # predict requires --output-dir even though our writer ignores it.
+            scratch = tempfile.mkdtemp(prefix="evo2_stream_predict_unused_")
+            predict_mod._write_predictions_batch = self._make_writer(q, state)
+            sys.argv = [
+                "predict_evo2",
+                "--ckpt-dir",
+                args.ckpt_dir,
+                "--fasta",
+                args.fasta,
+                "--embedding-layer",
+                str(args.layer),
+                "--micro-batch-size",
+                str(args.micro_batch_size),
+                "--write-interval",
+                "batch",
+                "--output-dir",
+                scratch,
+            ]
+            try:
+                # Evo2's params are bf16; TransformerEngine asserts param/input dtypes match
+                # unless inside an autocast region, and predict's minimal-arg path sets none.
+                # Wrap the forward loop (mirrors core.Evo2SAE._forward_hidden) so the streamed
+                # forwards run under bf16 autocast instead of tripping that assertion.
+                with torch.autocast(device_type="cuda", dtype=torch.bfloat16):
+                    predict_mod.main()
+                q.put(_DONE)
+            except BaseException as exc:  # surface producer failure to the consumer
+                q.put(exc)
+
+        thread = threading.Thread(target=run_predict, name="evo2-predict-producer", daemon=True)
+        thread.start()
+        checked = False
+        try:
+            while True:
+                item = q.get()
+                if item is _DONE:
+                    break
+                if isinstance(item, BaseException):
+                    raise item
+                if not checked:
+                    # Fail clearly if --input-dim disagrees with the layer's true residual
+                    # width, instead of an opaque matmul shape error inside the SAE encoder.
+                    if item.shape[1] != args.input_dim:
+                        raise ValueError(
+                            f"--input-dim={args.input_dim} but streamed Evo2 activations are width "
+                            f"{item.shape[1]} (embedding-layer {args.layer}); set --input-dim to match."
+                        )
+                    checked = True
+                yield item
+        finally:
+            thread.join(timeout=30.0)
+
+
+def main() -> None:
+    """Run streaming SAE training."""
+    args = parse_args()
+    if args.dp_size != 1:
+        raise ValueError("train_streaming.py supports only --dp-size 1; use one GPU for this streaming path.")
+
+    set_seed(args.seed)
+    # Producer/consumer split across GPUs (this is a pipeline, not data-parallel): the Evo2
+    # producer (predict) pins the first visible GPU (LOCAL_RANK 0 -> cuda:0); put the SAE
+    # consumer on a second GPU when one is available so the two stages run concurrently
+    # instead of sharing one device. --device overrides; single-GPU falls back to that device.
+    if args.device:
+        device = args.device
+    elif torch.cuda.is_available() and torch.cuda.device_count() > 1:
+        device = "cuda:1"
+    else:
+        device = get_device()
+    print(f"SAE consumer device: {device}  (Evo2 producer runs on cuda:0)")
+
+    input_dim = args.input_dim
+    sae = build_sae(args, input_dim)
+    print(f"SAE: {args.model_type}, input_dim={input_dim}, hidden_dim={sae.hidden_dim}")
+
+    producer = Evo2ActivationProducer(args)
+    if args.init_pre_bias:
+        # Streaming pre-bias init would need a *second* Evo2 `predict` pass to sample
+        # activations before training. `predict` initializes Megatron global state
+        # (num-microbatches calculator, model-parallel groups) that it never tears down,
+        # so a second pass crashes with "num microbatches calculator is already
+        # initialized". Fail loudly instead of dying cryptically mid-run. Tracked for a
+        # single-pass fix (sample the first rows off the one training stream).
+        raise NotImplementedError(
+            "--init-pre-bias is not supported by the streaming path (a second Evo2 predict "
+            "pass collides with Megatron global state). Re-run with --no-init-pre-bias."
+        )
+
+    dataloader = make_streaming_dataloader(
+        producer,
+        batch_size=args.batch_size,
+        config=StreamingConfig(
+            enabled=True,
+            queue_size=args.queue_size,
+            shuffle_buffer_size=args.shuffle_buffer_size,
+            seed=args.seed,
+            drop_last=args.drop_last,
+        ),
+    )
+
+    training_config = TrainingConfig(
+        lr=args.lr,
+        n_epochs=args.n_epochs,
+        max_steps=args.max_steps,
+        batch_size=args.batch_size,
+        device=device,
+        log_interval=args.log_interval,
+        checkpoint_dir=args.checkpoint_dir,
+        checkpoint_steps=args.checkpoint_steps,
+        grad_accumulation_steps=args.grad_accumulation_steps,
+        warmup_steps=args.warmup_steps,
+        max_grad_norm=args.max_grad_norm,
+        lr_schedule=args.lr_schedule,
+        lr_min=args.lr_min,
+        lr_decay_steps=args.lr_decay_steps,
+    )
+    wandb_config = WandbConfig(
+        enabled=args.wandb_enabled,
+        project=args.wandb_project,
+        run_name=args.wandb_run_name,
+        group=args.wandb_group,
+        job_type=args.wandb_job_type,
+        config=vars(args),
+    )
+    perf_logger = PerfLogger(
+        log_interval=args.log_interval,
+        use_wandb=args.wandb_enabled,
+        print_logs=True,
+        device=device,
+    )
+    trainer = Trainer(
+        sae,
+        training_config,
+        wandb_config=wandb_config,
+        perf_logger=perf_logger,
+        parallel_config=ParallelConfig(dp_size=args.dp_size),
+    )
+    trainer.fit(dataloader, resume_from=args.resume_from, data_sharded=True)
+
+
+if __name__ == "__main__":
+    main()
diff --git a/interpretability/sparse_autoencoders/recipes/evo2/tests/test_streaming.py b/interpretability/sparse_autoencoders/recipes/evo2/tests/test_streaming.py
new file mode 100644
index 0000000000..06d3d61f97
--- /dev/null
+++ b/interpretability/sparse_autoencoders/recipes/evo2/tests/test_streaming.py
@@ -0,0 +1,123 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: LicenseRef-Apache2
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""CPU guards for scripts/train_streaming.py.
+
+These run without CUDA or bionemo.evo2: the heavy Evo2 ``predict`` import is lazy
+(inside ``Evo2ActivationProducer.__call__``), so we inject a fake ``predict`` module
+that drives the same per-batch writer hook the real producer monkeypatches. This
+exercises the producer/queue plumbing, the --input-dim guard, and the
+--init-pre-bias rejection on CPU.
+"""
+
+import argparse
+import importlib.util
+import sys
+import types
+from pathlib import Path
+
+import pytest
+import torch
+
+_SCRIPT = Path(__file__).resolve().parents[1] / "scripts" / "train_streaming.py"
+
+
+def _load_train_streaming():
+    """Import scripts/train_streaming.py as a module (it is not an installed package)."""
+    spec = importlib.util.spec_from_file_location("train_streaming", _SCRIPT)
+    mod = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(mod)
+    return mod
+
+
+def _install_fake_predict(monkeypatch, width: int, n_batches: int = 1):
+    """Inject a fake bionemo.evo2.run.predict whose main() drives the writer hook.
+
+    The producer sets ``predict_mod._write_predictions_batch`` then calls ``main()``;
+    our fake main() invokes that writer with ``n_batches`` of ``[1, 4, width]`` activations
+    (pad_mask all ones), mimicking Evo2 emitting a residual stream of the given width.
+    """
+    fake = types.ModuleType("bionemo.evo2.run.predict")
+    fake._write_predictions_batch = None  # replaced by the producer before main() runs
+
+    def main():
+        for b in range(n_batches):
+            hidden = torch.zeros(1, 4, width)
+            preds = {"hidden_embeddings": hidden, "pad_mask": torch.ones(1, 4)}
+            fake._write_predictions_batch(preds, "unused", b, 0, 0)
+
+    fake.main = main
+
+    # Provide the parent packages so `from bionemo.evo2.run import predict` resolves to the fake.
+    for name in ("bionemo", "bionemo.evo2", "bionemo.evo2.run"):
+        monkeypatch.setitem(sys.modules, name, types.ModuleType(name))
+    sys.modules["bionemo.evo2.run"].predict = fake
+    monkeypatch.setitem(sys.modules, "bionemo.evo2.run.predict", fake)
+    return fake
+
+
+def _producer_args(ts, **overrides):
+    """Minimal argparse.Namespace for Evo2ActivationProducer."""
+    base = dict(
+        ckpt_dir="unused", fasta="unused", layer=12, input_dim=1920, micro_batch_size=4,
+        max_tokens=0, dtype="fp32", queue_size=4,
+    )
+    base.update(overrides)
+    return argparse.Namespace(**base)
+
+
+def test_input_dim_mismatch_raises_clear_error(monkeypatch):
+    """#4: a width != --input-dim fails with a clear message, not an opaque matmul error."""
+    ts = _load_train_streaming()
+    _install_fake_predict(monkeypatch, width=1920)
+    producer = ts.Evo2ActivationProducer(_producer_args(ts, input_dim=1024))  # deliberately wrong
+    with pytest.raises(ValueError, match=r"--input-dim=1024.*width 1920"):
+        list(producer())
+
+
+def test_input_dim_match_streams_chunks(monkeypatch):
+    """Matching width streams chunks through to the consumer and terminates on the sentinel."""
+    ts = _load_train_streaming()
+    _install_fake_predict(monkeypatch, width=1920, n_batches=2)
+    producer = ts.Evo2ActivationProducer(_producer_args(ts, input_dim=1920))
+    chunks = list(producer())
+    assert len(chunks) == 2
+    assert all(c.shape[1] == 1920 for c in chunks)
+
+
+def test_producer_propagates_predict_failure(monkeypatch):
+    """A crash in the predict thread surfaces in the consumer instead of hanging."""
+    ts = _load_train_streaming()
+    fake = _install_fake_predict(monkeypatch, width=1920)
+
+    def boom():
+        raise RuntimeError("predict exploded")
+
+    fake.main = boom
+    producer = ts.Evo2ActivationProducer(_producer_args(ts))
+    with pytest.raises(RuntimeError, match="predict exploded"):
+        list(producer())
+
+
+def test_init_pre_bias_is_rejected(monkeypatch):
+    """#1: --init-pre-bias is unsupported on the streaming path and fails fast (no GPU needed)."""
+    ts = _load_train_streaming()
+    monkeypatch.setattr(
+        sys, "argv",
+        ["train_streaming.py", "--ckpt-dir", "x", "--fasta", "y", "--embedding-layer", "12",
+         "--input-dim", "8", "--init-pre-bias", "--no-wandb"],
+    )
+    with pytest.raises(NotImplementedError, match="init-pre-bias"):
+        ts.main()