diff --git a/nemo_curator/stages/audio/tagging/inference/nemo_asr_align.py b/nemo_curator/stages/audio/tagging/inference/nemo_asr_align.py
index 5f3908f075..7c4a32b3e5 100644
--- a/nemo_curator/stages/audio/tagging/inference/nemo_asr_align.py
+++ b/nemo_curator/stages/audio/tagging/inference/nemo_asr_align.py
@@ -35,13 +35,20 @@
 from nemo.collections.asr.parts.submodules.rnnt_decoding import RNNTDecodingConfig
 
 from nemo_curator.backends.base import NodeInfo, WorkerMetadata
+from nemo_curator.stages.audio._agent._agent_ready import (
+    AgentReady,
+    ConditionalWrite,
+    Gates,
+    IOSpec,
+    StageContract,
+)
 from nemo_curator.stages.base import ProcessingStage
 from nemo_curator.stages.resources import Resources
 from nemo_curator.tasks import AudioTask
 
 
 @dataclass
-class BaseASRProcessorStage(ProcessingStage[AudioTask, AudioTask]):
+class BaseASRProcessorStage(AgentReady, ProcessingStage[AudioTask, AudioTask]):
     """Base class for ASR stages with shared config and segment preparation.
 
     Provides common fields and _prepare_segment_batch_with_metadata for
@@ -72,9 +79,14 @@ class BaseASRProcessorStage(ProcessingStage[AudioTask, AudioTask]):
     # Output keys
     text_key: str = "text"
     words_key: str = "words"
+    alignment_key: str = "alignment"
 
     compute_timestamps: bool = True
     segments_key: str = "segments"
+    audio_filepath_key: str = "audio_filepath"
+    resampled_audio_filepath_key: str = "resampled_audio_filepath"
+    split_filepaths_key: str = "split_filepaths"
+    split_metadata_key: str = "split_metadata"
 
     # Stage metadata (subclasses can override)
     name: str = "BaseASRProcessor"
@@ -110,7 +122,7 @@ def _prepare_segment_batch_with_metadata(
 
         if cut_audio_segments:
             for metadata_idx, metadata in enumerate(metadata_batch):
-                audio_path = metadata.get("resampled_audio_filepath", metadata.get("audio_filepath"))
+                audio_path = metadata.get(self.resampled_audio_filepath_key, metadata.get(self.audio_filepath_key))
                 if not audio_path:
                     continue
                 audio, sr = torchaudio.load(audio_path)
@@ -131,10 +143,10 @@ def _prepare_segment_batch_with_metadata(
         else:
             for metadata_idx, metadata in enumerate(metadata_batch):
                 for segment_idx, segment in enumerate(metadata.get(segments_key, [])):
-                    if "resampled_audio_filepath" in segment:
+                    if self.resampled_audio_filepath_key in segment:
                         segment_metadata_list.append(
                             {
-                                "resampled_audio_filepath": segment["resampled_audio_filepath"],
+                                self.resampled_audio_filepath_key: segment[self.resampled_audio_filepath_key],
                                 "metadata_idx": metadata_idx,
                                 "segment_idx": segment_idx,
                             }
@@ -190,6 +202,9 @@ class NeMoASRAlignerStage(BaseASRProcessorStage):
 
     # input keys
     segments_key: str = "segments"
+    split_filepaths_key: str = "split_filepaths"
+    split_metadata_key: str = "split_metadata"
+    alignment_key: str = "alignment"
 
     # Output keys
     text_key: str = "text"
@@ -268,13 +283,48 @@ def setup(self, _: WorkerMetadata | None = None) -> None:
 
     def inputs(self) -> tuple[list[str], list[str]]:
         if self.infer_segment_only:
-            return ["data"], ["resampled_audio_filepath", self.segments_key]
-        return ["data"], ["duration", self.segments_key, "split_filepaths", "split_metadata"]
+            return ["data"], [self.resampled_audio_filepath_key, self.segments_key]
+        return ["data"], ["duration", self.segments_key, self.split_filepaths_key, self.split_metadata_key]
 
     def outputs(self) -> tuple[list[str], list[str]]:
         if self.infer_segment_only:
-            return ["data"], ["resampled_audio_filepath", self.segments_key]
-        return ["data"], ["duration", self.segments_key, "split_filepaths", "split_metadata"]
+            return ["data"], [self.resampled_audio_filepath_key, self.segments_key]
+        return ["data"], ["duration", self.segments_key, self.split_filepaths_key, self.split_metadata_key]
+
+    def describe(self) -> StageContract:
+        if self.infer_segment_only:
+            reads = IOSpec(data_keys=[self.resampled_audio_filepath_key, self.segments_key])
+            writes = IOSpec(segment_data_keys=[self.text_key, self.words_key])
+            iteration_key = self.segments_key
+            conditional_writes = []
+        else:
+            reads = IOSpec(
+                data_keys=["duration", self.segments_key, self.split_filepaths_key, self.split_metadata_key]
+            )
+            writes = IOSpec(segment_data_keys=[self.text_key, self.alignment_key])
+            iteration_key = self.split_metadata_key
+            conditional_writes = [
+                ConditionalWrite(
+                    writes=IOSpec(data_keys=[self.text_key, self.alignment_key]),
+                    condition=(
+                        f"'{self.split_filepaths_key}' is an empty list, or a transcribed split "
+                        f"has no corresponding item in '{self.split_metadata_key}'"
+                    ),
+                )
+            ]
+        return StageContract(
+            reads=reads,
+            writes=writes,
+            cardinality="1:1 nested-list",
+            iteration_key=iteration_key,
+            gates=Gates(
+                requires_gpu=self.resources.requires_gpu,
+                requires_internet_first_run=self.model_path is None,
+                # Batched for GPU throughput, but each row is transcribed and aligned on its own.
+                per_row_independent=True,
+            ),
+            conditional_writes=conditional_writes,
+        )
 
     def get_alignments_text(self, hypotheses: Any) -> tuple[list, str]:  # noqa: ANN401
         """Extract word alignments and text from model hypotheses."""
@@ -346,7 +396,7 @@ def process_full_audio(self, tasks: list[AudioTask]) -> list[AudioTask]:  # noqa
         skip_indices = []
         meta_indices = []
         for i, data in enumerate(entries):
-            split_filepaths = data.get("split_filepaths")
+            split_filepaths = data.get(self.split_filepaths_key)
             has_splits = isinstance(split_filepaths, list) and len(split_filepaths) > 0
             if has_splits or split_filepaths is None:
                 meta_indices.append(i)
@@ -355,14 +405,14 @@ def process_full_audio(self, tasks: list[AudioTask]) -> list[AudioTask]:  # noqa
 
         for i in skip_indices:
             entries[i][self.text_key] = ""
-            entries[i]["alignment"] = []
+            entries[i][self.alignment_key] = []
 
         # collect all split paths of all entries in the batch
         all_paths = []
         path_to_entry_and_split = []
         for entry_idx in meta_indices:
             meta_entry = entries[entry_idx]
-            split_filepaths = meta_entry.get("split_filepaths")
+            split_filepaths = meta_entry.get(self.split_filepaths_key)
             if not split_filepaths:
                 logger.warning(f"[{self.name}] Entry at index {entry_idx} has no split_filepaths, skipping.")
                 continue
@@ -404,13 +454,13 @@ def process_full_audio(self, tasks: list[AudioTask]) -> list[AudioTask]:  # noqa
             else:
                 alignments, text = [], ""
 
-            split_metadata = meta_entry.get("split_metadata")
+            split_metadata = meta_entry.get(self.split_metadata_key)
             if split_metadata and split_idx < len(split_metadata):
                 split_metadata[split_idx][self.text_key] = text
-                split_metadata[split_idx]["alignment"] = alignments
+                split_metadata[split_idx][self.alignment_key] = alignments
             else:
                 meta_entry[self.text_key] = text
-                meta_entry["alignment"] = alignments
+                meta_entry[self.alignment_key] = alignments
 
         return tasks
 
@@ -434,7 +484,7 @@ def process_segments(self, tasks: list[AudioTask]) -> list[AudioTask]:
             with torch.no_grad():
                 hypotheses_list = self._asr_model.transcribe(all_segments, override_config=self._override_cfg)
         except Exception as e:
-            files_list = [x.get("resampled_audio_filepath", x.get("audio_filepath")) for x in entries]
+            files_list = [x.get(self.resampled_audio_filepath_key, x.get(self.audio_filepath_key)) for x in entries]
             msg = f"[{self.name}] Exception for audio list: {files_list}, error: {e}"
             raise ValueError(msg) from e
 
diff --git a/nemo_curator/stages/audio/tagging/merge_alignment_diarization.py b/nemo_curator/stages/audio/tagging/merge_alignment_diarization.py
index 986c7fba6f..57c67fadf0 100644
--- a/nemo_curator/stages/audio/tagging/merge_alignment_diarization.py
+++ b/nemo_curator/stages/audio/tagging/merge_alignment_diarization.py
@@ -21,12 +21,19 @@
 
 from loguru import logger
 
+from nemo_curator.stages.audio._agent._agent_ready import (
+    AgentReady,
+    ConditionalWrite,
+    Gates,
+    IOSpec,
+    StageContract,
+)
 from nemo_curator.stages.base import ProcessingStage
 from nemo_curator.tasks import AudioTask
 
 
 @dataclass
-class MergeAlignmentDiarizationStage(ProcessingStage[AudioTask, AudioTask]):
+class MergeAlignmentDiarizationStage(AgentReady, ProcessingStage[AudioTask, AudioTask]):
     """
     Stage that merges alignment and diarization information.
 
@@ -51,15 +58,32 @@ class MergeAlignmentDiarizationStage(ProcessingStage[AudioTask, AudioTask]):
     # Output keys
     text_key: str = "text"
     words_key: str = "words"
+    alignment_key: str = "alignment"
+    segments_key: str = "segments"
 
     # Stage metadata
     name: str = "MergeAlignmentDiarization"
 
     def inputs(self) -> tuple[list[str], list[str]]:
-        return [], ["alignment", "segments"]
+        return [], [self.alignment_key, self.segments_key]
 
     def outputs(self) -> tuple[list[str], list[str]]:
-        return [], ["alignment", "segments"]
+        return [], [self.alignment_key, self.segments_key]
+
+    def describe(self) -> StageContract:
+        return StageContract(
+            reads=IOSpec(data_keys=[self.alignment_key, self.segments_key]),
+            conditional_writes=[
+                ConditionalWrite(
+                    writes=IOSpec(segment_data_keys=[self.text_key, self.words_key]),
+                    condition=(
+                        f"both '{self.alignment_key}' and '{self.segments_key}' are non-empty"
+                    ),
+                )
+            ],
+            # Merges this row's own alignment with this row's own segments.
+            gates=Gates(per_row_independent=True),
+        )
 
     @staticmethod
     def align_words_to_segments(
@@ -183,8 +207,8 @@ def process(self, task: AudioTask) -> AudioTask:
         """Process entry to merge alignment and diarization."""
         t0 = time.perf_counter()
         data_entry = task.data
-        alignment = data_entry.get("alignment", [])
-        segments = data_entry.get("segments", [])
+        alignment = data_entry.get(self.alignment_key, [])
+        segments = data_entry.get(self.segments_key, [])
 
         if alignment and segments:
             self.align_words_to_segments(alignment, segments, self.text_key, self.words_key)
diff --git a/nemo_curator/stages/audio/tagging/prepare_module_segments.py b/nemo_curator/stages/audio/tagging/prepare_module_segments.py
index 1a7fd64084..0a5e27c438 100644
--- a/nemo_curator/stages/audio/tagging/prepare_module_segments.py
+++ b/nemo_curator/stages/audio/tagging/prepare_module_segments.py
@@ -24,6 +24,7 @@
 
 from loguru import logger
 
+from nemo_curator.stages.audio._agent._agent_ready import AgentReady, Gates, IOSpec, StageContract
 from nemo_curator.stages.base import ProcessingStage
 from nemo_curator.tasks import AudioTask
 
@@ -32,7 +33,7 @@
 
 
 @dataclass
-class PrepareModuleSegmentsStage(ProcessingStage[AudioTask, AudioTask]):
+class PrepareModuleSegmentsStage(AgentReady, ProcessingStage[AudioTask, AudioTask]):
     """
     Stage that prepares segments for TTS or ASR by merging and splitting based on
     duration, punctuation, and bandwidth.
@@ -65,11 +66,28 @@ class PrepareModuleSegmentsStage(ProcessingStage[AudioTask, AudioTask]):
     terminal_punct_marks: str = ".!?。？？！。"  # noqa: RUF001
     full_utterance_ratio: float = 1.0
     punctuation_split_only: bool = False
+    segments_key: str = "segments"
+    duration_key: str = "duration"
+    metrics_key: str = "metrics"
 
     name: str = "PrepareModuleSegments"
 
     def inputs(self) -> tuple[list[str], list[str]]:
-        return [], ["segments", "duration"]
+        return [], [self.segments_key, self.duration_key]
+
+    def outputs(self) -> tuple[list[str], list[str]]:
+        return [], [self.segments_key]
+
+    def describe(self) -> StageContract:
+        return StageContract(
+            reads=IOSpec(data_keys=[self.segments_key, self.duration_key]),
+            writes=IOSpec(data_keys=[self.segments_key]),
+            # The ``asr`` module draws its per-segment length limit from ``self._rng``, which
+            # ``process`` reseeds from a hash of the row's own id before touching it. So the draws
+            # a row gets depend on that row alone, unlike PyAnnoteDiarizationStage's unseeded
+            # generator, whose sequence position is decided by how many rows preceded it.
+            gates=Gates(per_row_independent=True),
+        )
 
     def __post_init__(self):
         if self.module not in ("tts", "asr"):
@@ -94,8 +112,8 @@ def get_words_list_from_all_segments(self, metadata: dict[str, Any]) -> list[dic
                 - sisdr_squim: The SI-SDR score of the word if available
                 - bandwidth: The bandwidth of the word if available
         """
-        segments = metadata["segments"]
-        audio_duration = metadata.get("duration", 0.0)
+        segments = metadata[self.segments_key]
+        audio_duration = metadata.get(self.duration_key, 0.0)
 
         if "overlap_segments" not in metadata:
             add_non_speaker_segments(segments, audio_duration)
@@ -111,8 +129,8 @@ def get_words_list_from_all_segments(self, metadata: dict[str, Any]) -> list[dic
                 for word in segment[self.words_key]:
                     new_word = dict(word)
                     new_word["speaker"] = segment["speaker"]
-                    if "metrics" in segment:
-                        m = segment["metrics"]
+                    if self.metrics_key in segment:
+                        m = segment[self.metrics_key]
                         new_word["stoi_squim"] = m.get("stoi_squim") if isinstance(m, dict) else None
                         new_word["sisdr_squim"] = m.get("sisdr_squim") if isinstance(m, dict) else None
                         new_word["pesq_squim"] = m.get("pesq_squim") if isinstance(m, dict) else None
@@ -345,7 +363,7 @@ def add_new_segments_to_metadata(self, metadata: dict[str, Any], new_segments: l
                     {"word": w.get("word", ""), "start": w.get("start", 0.0), "end": w.get("end", 0.0)}
                     for w in new_segment["words"]
                 ],
-                "metrics": {
+                self.metrics_key: {
                     "pesq_squim": [w.get("pesq_squim") for w in new_segment["words"]],
                     "stoi_squim": [w.get("stoi_squim") for w in new_segment["words"]],
                     "sisdr_squim": [w.get("sisdr_squim") for w in new_segment["words"]],
@@ -354,7 +372,7 @@ def add_new_segments_to_metadata(self, metadata: dict[str, Any], new_segments: l
             }
             segments.append(seg)
 
-        metadata["segments"] = segments
+        metadata[self.segments_key] = segments
 
     def prepare_asr_segments(self, words: list[dict[str, Any]], metadata: dict[str, Any]) -> None:
         """Prepare ASR segments (multi-speaker per segment allowed)."""
@@ -412,7 +430,7 @@ def process(self, task: AudioTask) -> AudioTask:
         seed = int(hashlib.md5(entry_id.encode()).hexdigest()[:8], 16)  # noqa: S324
         self._rng.seed(seed)
         try:
-            if "segments" not in data_entry:
+            if self.segments_key not in data_entry:
                 logger.info(f"[{self.name}] No segments in metadata for: {data_entry.get('audio_filepath', '')}")
                 return task
 
diff --git a/nemo_curator/stages/audio/tagging/resample_audio.py b/nemo_curator/stages/audio/tagging/resample_audio.py
index 2b4772eb82..41ee32aa08 100644
--- a/nemo_curator/stages/audio/tagging/resample_audio.py
+++ b/nemo_curator/stages/audio/tagging/resample_audio.py
@@ -25,20 +25,40 @@
 import os
 import shutil
 import subprocess
+import tempfile
 import time
 import uuid
 from dataclasses import dataclass
+from typing import ClassVar
 
+import soundfile
 from fsspec.core import url_to_fs
 
 from nemo_curator.backends.base import NodeInfo, WorkerMetadata
-from nemo_curator.stages.audio.common import get_audio_duration
+from nemo_curator.stages.audio._agent._agent_ready import (
+    AgentReady,
+    ConditionalWrite,
+    Gates,
+    IOSpec,
+    StageContract,
+    StaticHints,
+)
+from nemo_curator.stages.audio._agent._residency import (
+    InputResidency,
+    cleanup_temp_files,
+    drop_resident_audio,
+    produce_audio_filepath,
+    reject_sinkless_conversion,
+    residency_read_specs,
+    resolve_audio_path,
+)
+from nemo_curator.stages.audio.common import get_audio_duration, load_audio_file
 from nemo_curator.stages.base import ProcessingStage
 from nemo_curator.tasks import AudioTask
 
 
 @dataclass
-class ResampleAudioStage(ProcessingStage[AudioTask, AudioTask]):
+class ResampleAudioStage(AgentReady, ProcessingStage[AudioTask, AudioTask]):
     """
     Stage for resampling audio files in a TTS/ALM dataset.
 
@@ -48,6 +68,15 @@ class ResampleAudioStage(ProcessingStage[AudioTask, AudioTask]):
 
     """
 
+    AGENT_STATIC: ClassVar[StaticHints] = StaticHints(
+        gates=Gates(
+            writes_to_disk=True,
+            requires_ffmpeg=True,
+            output_path_params=["resampled_audio_dir"],
+            per_row_independent=True,
+        )
+    )
+
     # Processing parameters
     resampled_audio_dir: str
     input_format: str = "wav"
@@ -58,12 +87,28 @@ class ResampleAudioStage(ProcessingStage[AudioTask, AudioTask]):
     # Key names
     audio_filepath_key: str = "audio_filepath"
     resampled_audio_filepath_key: str = "resampled_audio_filepath"
+    waveform_key: str = "waveform"
+    sample_rate_key: str = "sample_rate"
     duration_key: str = "duration"
     audio_item_id_key: str = "audio_item_id"
+    original_audio_filepath_key: str = "original_audio_filepath"
+
+    input_residency: InputResidency = "file"
+    keep_waveform_in_task: bool = False
+    write_to_disk: bool = True
+    update_audio_filepath: bool = False
 
     # Stage metadata
     name: str = "ResampleAudio"
 
+    def __post_init__(self) -> None:
+        reject_sinkless_conversion(
+            stage=type(self).__name__,
+            keep_waveform_in_task=self.keep_waveform_in_task,
+            write_to_disk=self.write_to_disk,
+            update_audio_filepath=self.update_audio_filepath,
+        )
+
     def setup_on_node(
         self, _node_info: NodeInfo | None = None, _worker_metadata: WorkerMetadata | None = None
     ) -> None:
@@ -76,13 +121,111 @@ def setup_on_node(
     def inputs(self) -> tuple[list[str], list[str]]:
         return [], [self.audio_filepath_key]
 
+    def validate_input(self, task: AudioTask) -> bool:
+        """Validate the configured file/waveform residency alternative."""
+        data = task.data
+        has_file = bool(data.get(self.audio_filepath_key))
+        has_waveform = data.get(self.waveform_key) is not None and data.get(self.sample_rate_key) is not None
+        if self.input_residency == "file":
+            return has_file
+        if self.input_residency == "waveform":
+            return has_waveform
+        return has_file or has_waveform
+
     def outputs(self) -> tuple[list[str], list[str]]:
-        return [], [
-            self.audio_filepath_key,
-            self.audio_item_id_key,
-            self.resampled_audio_filepath_key,
-            self.duration_key,
-        ]
+        outputs = [self.audio_item_id_key, self.duration_key]
+        if self.write_to_disk:
+            outputs.append(self.resampled_audio_filepath_key)
+        if self.keep_waveform_in_task:
+            outputs.extend([self.waveform_key, self.sample_rate_key])
+        if self.update_audio_filepath:
+            outputs.extend([self.audio_filepath_key, self.original_audio_filepath_key])
+        return [], outputs
+
+    def describe(self) -> StageContract:
+        writes = [self.audio_item_id_key, self.duration_key]
+        produces = []
+        conditional_writes = []
+        if self.write_to_disk:
+            writes.append(self.resampled_audio_filepath_key)
+            produces.append("disk")
+        if self.keep_waveform_in_task:
+            writes.extend([self.waveform_key, self.sample_rate_key])
+            produces.append("tensor")
+        if self.update_audio_filepath:
+            writes.append(self.audio_filepath_key)
+            conditional_writes.append(
+                ConditionalWrite(
+                    writes=IOSpec(data_keys=[self.original_audio_filepath_key]),
+                    condition=(
+                        f"'{self.audio_filepath_key}' exists and "
+                        f"'{self.original_audio_filepath_key}' is not already present"
+                    ),
+                )
+            )
+        return StageContract(
+            reads_one_of=residency_read_specs(
+                self.input_residency,
+                audio_filepath_key=self.audio_filepath_key,
+                waveform_key=self.waveform_key,
+                sample_rate_key=self.sample_rate_key,
+            ),
+            writes=IOSpec(data_keys=writes, produces=produces),
+            removes_keys=(
+                [self.waveform_key, self.sample_rate_key]
+                if self.write_to_disk and not self.keep_waveform_in_task
+                else []
+            ),
+            conditional_writes=conditional_writes,
+            gates=Gates(
+                writes_to_disk=self.write_to_disk,
+                requires_ffmpeg=True,
+                output_path_params=["resampled_audio_dir"],
+                per_row_independent=True,
+            ),
+        )
+
+    def _audio_digest(self, local_audio_path: str) -> str:
+        """A short digest of this audio and the settings about to be applied to it."""
+        digest = hashlib.sha256()
+        with open(local_audio_path, "rb") as handle:
+            for chunk in iter(lambda: handle.read(1 << 20), b""):
+                digest.update(chunk)
+        digest.update(f"|{self.target_sample_rate}|{self.target_nchannels}|{self.target_format}".encode())
+        return digest.hexdigest()[:16]
+
+    def _item_id(self, local_audio_path: str, *, from_scratch_file: bool, source: str | None) -> str:
+        """The output filename for a row that does not already carry an id.
+
+        A real input path is stable across runs, so it can name the output. A scratch path
+        materialised from a waveform is not: naming a persistent output after it made every run
+        write a fresh set of files (measured: 129 on disk against 65 manifest rows). So hash the
+        audio instead, which is the identity the path was standing in for -- and folding in the
+        settings makes the "already converted, skip it" check below correct, not merely fast.
+
+        The path branch is unchanged, so pipelines reading real files keep their output names.
+        """
+        if not from_scratch_file:
+            stem = os.path.splitext(os.path.basename(local_audio_path))[0]
+            return f"{stem}_{hashlib.sha256(local_audio_path.encode()).hexdigest()[:8]}"
+        # Keep the source name on the front so a clip stays traceable by eye.
+        stem = os.path.splitext(os.path.basename(str(source)))[0] if source else "clip"
+        return f"{stem}_{self._audio_digest(local_audio_path)}"
+
+    def _matches_target(self, path: str) -> bool:
+        """Whether the file at the output path really holds the conversion asked for.
+
+        The file-route name carries the source path, never the settings, so a name hit is not
+        evidence the work is done: a second run at a different ``target_sample_rate`` used to skip
+        and serve the old rate, with the duration measured off the stale file. Reading the header
+        is free beside spawning ffmpeg. A header-valid but truncated file still passes this, which
+        is why the conversion below writes to a temp name and renames.
+        """
+        try:
+            info = soundfile.info(path)
+        except Exception:  # noqa: BLE001 - unreadable or not-audio -> convert it again
+            return False
+        return info.samplerate == self.target_sample_rate and info.channels == self.target_nchannels
 
     def process(self, task: AudioTask) -> AudioTask:
         """
@@ -97,31 +240,89 @@ def process(self, task: AudioTask) -> AudioTask:
         t0 = time.perf_counter()
         data_entry = task.data
 
-        if self.audio_filepath_key not in data_entry:
-            msg = "Absolute audio filepath is required"
+        temp_paths: list[str] = []
+        input_audio_path = resolve_audio_path(
+            data_entry,
+            residency=self.input_residency,  # type: ignore[arg-type]
+            audio_filepath_key=self.audio_filepath_key,
+            waveform_key=self.waveform_key,
+            sample_rate_key=self.sample_rate_key,
+            register_temp=temp_paths,
+        )
+        if input_audio_path is None:
+            msg = "Audio file path or waveform/sample_rate is required"
             raise ValueError(msg)
 
-        original_audio_filepath = data_entry[self.audio_filepath_key]
-        _, local_audio_path = url_to_fs(original_audio_filepath)
-        if self.audio_item_id_key not in data_entry:
-            stem = os.path.splitext(os.path.basename(local_audio_path))[0]
-            path_hash = hashlib.sha256(local_audio_path.encode()).hexdigest()[:8]
-            data_entry[self.audio_item_id_key] = f"{stem}_{path_hash}"
+        original_audio_filepath = data_entry.get(self.audio_filepath_key)
+        _, local_audio_path = url_to_fs(input_audio_path)
+        inherited_id = self.audio_item_id_key in data_entry
+        if not inherited_id:
+            data_entry[self.audio_item_id_key] = self._item_id(
+                local_audio_path,
+                from_scratch_file=bool(temp_paths),
+                source=original_audio_filepath,
+            )
+        output_stem = data_entry[self.audio_item_id_key]
+        if inherited_id and temp_paths:
+            # A fan-out gives every child the parent's id, so an inherited id is not a filename:
+            # 26 VAD segments of utt1.wav once collapsed onto one, with all 26 rows pointing at
+            # the survivor. The FILE takes a digest; the row keeps the id its producer gave it,
+            # which downstream stages read as the shared ``item_id`` role.
+            output_stem = f"{output_stem}_{self._audio_digest(local_audio_path)}"
 
-        input_audio_path = local_audio_path
-        output_audio_path = os.path.join(
-            self.resampled_audio_dir,
-            data_entry[self.audio_item_id_key] + "." + self.target_format,
-        )
+        if self.write_to_disk:
+            output_audio_path = os.path.join(
+                self.resampled_audio_dir,
+                output_stem + "." + self.target_format,
+            )
+        else:
+            fd, output_audio_path = tempfile.mkstemp(suffix=f".{self.target_format}")
+            os.close(fd)
+
+        try:
+            return self._convert_and_update(
+                task,
+                input_audio_path=input_audio_path,
+                output_audio_path=output_audio_path,
+                original_audio_filepath=original_audio_filepath,
+                started_at=t0,
+            )
+        finally:
+            cleanup_temp_files(temp_paths)
+            if not self.write_to_disk:
+                cleanup_temp_files([output_audio_path])
+
+    def _convert_and_update(
+        self,
+        task: AudioTask,
+        *,
+        input_audio_path: str,
+        output_audio_path: str,
+        original_audio_filepath: str | None,
+        started_at: float,
+    ) -> AudioTask:
+        """Convert one resolved input and update its task metadata."""
+        data_entry = task.data
 
         # Convert audio file if not already done
         fs, output_path = url_to_fs(output_audio_path)
-        skipped_conversion = fs.exists(output_path)
+        skipped_conversion = self.write_to_disk and fs.exists(output_path) and self._matches_target(output_path)
         if not skipped_conversion:
+            # ffmpeg used to write straight to the deliverable. That was survivable while every
+            # run picked a new output name, but the name is stable now, so a run killed mid-write
+            # leaves a stump the NEXT run finds -- and a truncated WAV keeps a valid header, so
+            # the skip above waves it through and a fragment's duration lands in the manifest.
+            # Convert to a sibling temp name and rename, which is atomic on POSIX. Upstream
+            # landed the same fix independently; this keeps its naming so the two do not drift.
+            staging_dir = os.path.dirname(output_audio_path)
+            if staging_dir:
+                # setup_on_node makes this, but process() must not depend on having been through it.
+                os.makedirs(staging_dir, exist_ok=True)
             output_stem, output_extension = os.path.splitext(output_audio_path)
             temporary_audio_path = f"{output_stem}.{uuid.uuid4().hex}.tmp{output_extension}"
             cmd = [
                 "ffmpeg",
+                "-y",
                 "-v",
                 "error",
                 "-i",
@@ -142,18 +343,36 @@ def process(self, task: AudioTask) -> AudioTask:
                 msg = f"Error converting {input_audio_path}: {e}"
                 raise RuntimeError(msg) from e
             finally:
-                if os.path.exists(temporary_audio_path):
-                    os.remove(temporary_audio_path)
+                cleanup_temp_files([temporary_audio_path])
 
-        # Update metadata — preserve original URL for cloud paths
-        data_entry[self.audio_filepath_key] = original_audio_filepath
-        data_entry[self.resampled_audio_filepath_key] = output_audio_path
+        # Update metadata — preserve original URL for cloud paths.
+        if original_audio_filepath is not None:
+            data_entry[self.audio_filepath_key] = original_audio_filepath
+        if self.write_to_disk:
+            data_entry[self.resampled_audio_filepath_key] = output_audio_path
+            if self.update_audio_filepath:
+                produce_audio_filepath(
+                    data_entry,
+                    output_audio_path,
+                    key=self.audio_filepath_key,
+                    original_key=self.original_audio_filepath_key,
+                )
+        if self.keep_waveform_in_task:
+            waveform, sample_rate = load_audio_file(output_audio_path, mono=False)
+            data_entry[self.waveform_key] = waveform
+            data_entry[self.sample_rate_key] = sample_rate
+        elif self.write_to_disk:
+            drop_resident_audio(
+                data_entry,
+                waveform_key=self.waveform_key,
+                sample_rate_key=self.sample_rate_key,
+            )
         duration = get_audio_duration(output_audio_path)
         data_entry[self.duration_key] = duration
 
         self._log_metrics(
             {
-                "process_time": time.perf_counter() - t0,
+                "process_time": time.perf_counter() - started_at,
                 "duration": max(duration, 0.0),
                 "skipped_conversion": float(skipped_conversion),
             }
diff --git a/nemo_curator/stages/audio/tagging/split.py b/nemo_curator/stages/audio/tagging/split.py
index 27a9b0ff68..be50e29aaf 100644
--- a/nemo_curator/stages/audio/tagging/split.py
+++ b/nemo_curator/stages/audio/tagging/split.py
@@ -17,21 +17,31 @@
 
 """
 
+import contextlib
+import hashlib
 import math
+import os
+import posixpath
+import shutil
+import tempfile
 import time
 from dataclasses import dataclass
+from typing import Any, ClassVar
 
 import torchaudio
 from fsspec.core import url_to_fs
+from fsspec.implementations.local import LocalFileSystem
+from fsspec.spec import AbstractFileSystem
 from loguru import logger
 
+from nemo_curator.stages.audio._agent._agent_ready import AgentReady, Gates, IOSpec, StageContract, StaticHints
 from nemo_curator.stages.audio.tagging.inference.nemo_asr_align import NeMoASRAlignerStage
 from nemo_curator.stages.base import CompositeStage, ProcessingStage
 from nemo_curator.tasks import AudioTask
 
 
 @dataclass
-class SplitLongAudioStage(ProcessingStage[AudioTask, AudioTask]):
+class SplitLongAudioStage(AgentReady, ProcessingStage[AudioTask, AudioTask]):
     """
     Stage that splits long audio files into smaller segments.
 
@@ -41,36 +51,82 @@ class SplitLongAudioStage(ProcessingStage[AudioTask, AudioTask]):
     Args:
         suggested_max_len: Target maximum length for audio segments in seconds
         min_len: Minimum length for any split segment
+        output_dir: Optional directory for written split audio. When unset,
+            split files remain beside the source audio for backward compatibility.
     """
 
+    AGENT_STATIC: ClassVar[StaticHints] = StaticHints(
+        gates=Gates(
+            writes_to_disk=True,
+            output_path_params=["output_dir"],
+            per_row_independent=False,
+        )
+    )
+
     # Split parameters
     suggested_max_len: float = 3600.0
     min_len: float = 1.0
 
-    # Stage metadata
+    # Stage metadata. Keep the legacy positional fields before additive keys.
     name: str = "SplitLongAudio"
 
+    duration_key: str = "duration"
+    segments_key: str = "segments"
+    audio_filepath_key: str = "resampled_audio_filepath"
+    audio_item_id_key: str = "audio_item_id"
+    split_filepaths_key: str = "split_filepaths"
+    split_metadata_key: str = "split_metadata"
+    split_offsets_key: str = "split_offsets"
+    split_timestamps_key: str = "split_timestamps"
+
+    # Additive agent-only routing knob.
+    output_dir: str | None = None
+
     def inputs(self) -> tuple[list[str], list[str]]:
-        return [], ["duration", "segments", "resampled_audio_filepath"]
+        return [], [self.duration_key, self.segments_key, self.audio_filepath_key]
 
     def outputs(self) -> tuple[list[str], list[str]]:
         return [], [
-            "duration",
-            "segments",
-            "resampled_audio_filepath",
-            "split_filepaths",
-            "split_metadata",
-            "split_offsets",
-            "split_timestamps",
+            self.duration_key,
+            self.segments_key,
+            self.audio_filepath_key,
+            self.split_filepaths_key,
+            self.split_metadata_key,
+            self.split_offsets_key,
+            self.split_timestamps_key,
         ]
 
+    def describe(self) -> StageContract:
+        return StageContract(
+            reads=IOSpec(data_keys=[self.duration_key, self.segments_key, self.audio_filepath_key]),
+            writes=IOSpec(
+                data_keys=[
+                    self.split_filepaths_key,
+                    self.split_metadata_key,
+                    self.split_offsets_key,
+                    self.split_timestamps_key,
+                ],
+                produces=["disk"],
+            ),
+            cardinality="1:1 nested-list",
+            iteration_key=self.split_metadata_key,
+            # A row splits by its own duration and segments. With an ``output_dir`` every file
+            # shares one flat namespace, so the stem carries the source and effective split-plan
+            # identity. Without one, splits land beside their source.
+            gates=Gates(
+                writes_to_disk=True,
+                output_path_params=["output_dir"],
+                per_row_independent=self.output_dir is None,
+            ),
+        )
+
     def get_split_points(self, metadata: dict) -> list[float]:
         """Get the split points for the audio file based on segments."""
         splits = []
         split_start = 0
         prev_end = 0
 
-        segments = sorted(metadata.get("segments", []), key=lambda s: s.get("start", 0))
+        segments = sorted(metadata.get(self.segments_key, []), key=lambda s: s.get("start", 0))
         for segment in segments:
             end = segment.get("end", 0)
 
@@ -82,6 +138,79 @@ def get_split_points(self, metadata: dict) -> list[float]:
 
         return splits
 
+    def _prepare_output_dir(self) -> tuple[AbstractFileSystem | None, str]:
+        """Create and resolve an explicit output directory."""
+        if self.output_dir is None:
+            return None, ""
+        output_fs, resolved_output_dir = url_to_fs(self.output_dir)
+        output_fs.makedirs(resolved_output_dir, exist_ok=True)
+        return output_fs, resolved_output_dir
+
+    def _split_paths(
+        self,
+        split_name: str,
+        parent_url: str,
+        resolved_parent: str,
+        resolved_output_dir: str,
+    ) -> tuple[str, str]:
+        """Build the stored and resolved paths for one split."""
+        if self.output_dir is None:
+            split_filepath = f"{parent_url}/{split_name}" if parent_url else split_name
+            split_resolved = f"{resolved_parent}/{split_name}" if resolved_parent else split_name
+            return split_filepath, split_resolved
+        return posixpath.join(self.output_dir, split_name), posixpath.join(resolved_output_dir, split_name)
+
+    def _shared_output_stem(
+        self,
+        stem: str,
+        audio_path: str,
+        splits: list[float],
+        sample_rate: int,
+        total_frames: int,
+    ) -> str:
+        """Return a corpus-flat name keyed by source and effective split plan."""
+        accepted_boundaries = []
+        split_start = 0
+        for split in splits:
+            split_end = math.ceil(split * sample_rate)
+            if split_end - split_start > self.min_len * sample_rate:
+                accepted_boundaries.append(split_end)
+                split_start = split_end
+        if total_frames - split_start > self.min_len * sample_rate:
+            accepted_boundaries.append(total_frames)
+
+        identity = hashlib.sha256()
+        identity.update(audio_path.encode())
+        identity.update(
+            (
+                f"|{sample_rate}|{total_frames}|{self.suggested_max_len}|{self.min_len}|"
+                + ",".join(str(boundary) for boundary in accepted_boundaries)
+            ).encode()
+        )
+        return f"{stem}_{identity.hexdigest()[:32]}"
+
+    def _save_split(
+        self,
+        path: str,
+        waveform: Any,  # noqa: ANN401 - torchaudio accepts tensor-like waveforms
+        sample_rate: int,
+        output_fs: AbstractFileSystem | None,
+    ) -> None:
+        """Write directly locally or upload a temporary WAV through fsspec."""
+        if output_fs is None or isinstance(output_fs, LocalFileSystem):
+            torchaudio.save(path, waveform, sample_rate)
+            return
+
+        fd, local_temp = tempfile.mkstemp(suffix=".wav")
+        os.close(fd)
+        try:
+            torchaudio.save(local_temp, waveform, sample_rate)
+            with open(local_temp, "rb") as source, output_fs.open(path, "wb") as target:
+                shutil.copyfileobj(source, target)
+        finally:
+            with contextlib.suppress(OSError):
+                os.remove(local_temp)
+
     def process(self, task: AudioTask) -> AudioTask:
         """Process entry to split long audio files."""
         with self._time_metric("process_time"):
@@ -90,25 +219,25 @@ def process(self, task: AudioTask) -> AudioTask:
     def _do_split(self, task: AudioTask) -> AudioTask:
         """Core splitting logic, separated to keep statement count within limits."""
         data_entry = task.data
-        duration = data_entry["duration"]
+        duration = data_entry[self.duration_key]
 
         if duration < self.suggested_max_len:
-            data_entry["split_filepaths"] = [data_entry["resampled_audio_filepath"]]
-            data_entry["split_metadata"] = [
+            data_entry[self.split_filepaths_key] = [data_entry[self.audio_filepath_key]]
+            data_entry[self.split_metadata_key] = [
                 {
-                    "audio_item_id": data_entry.get("audio_item_id", "unknown"),
-                    "resampled_audio_filepath": data_entry["resampled_audio_filepath"],
-                    "duration": duration,
+                    self.audio_item_id_key: data_entry.get(self.audio_item_id_key, "unknown"),
+                    self.audio_filepath_key: data_entry[self.audio_filepath_key],
+                    self.duration_key: duration,
                 }
             ]
-            data_entry["split_offsets"] = [0.0]
-            data_entry["split_timestamps"] = [0.0]
+            data_entry[self.split_offsets_key] = [0.0]
+            data_entry[self.split_timestamps_key] = [0.0]
             self._log_metrics({"input_duration": duration, "splits_produced": 1})
             return task
 
         splits = self.get_split_points(data_entry)
 
-        audio_path = data_entry["resampled_audio_filepath"]
+        audio_path = data_entry[self.audio_filepath_key]
         _fs, resolved_path = url_to_fs(audio_path)
 
         # parent_url preserves protocol prefix (e.g. "s3://bucket/dir") for stored paths;
@@ -116,38 +245,54 @@ def _do_split(self, task: AudioTask) -> AudioTask:
         parent_url, filename = audio_path.rsplit("/", 1) if "/" in audio_path else ("", audio_path)
         resolved_parent = resolved_path.rsplit("/", 1)[0] if "/" in resolved_path else ""
         stem = filename.rsplit(".", 1)[0] if "." in filename else filename
+        output_fs, resolved_output_dir = self._prepare_output_dir()
 
         audio, sr = torchaudio.load(resolved_path)
+        if self.output_dir is not None:
+            # The default remains byte-for-byte source-adjacent. A shared directory needs the
+            # source and effective split plan because either can otherwise overwrite another row.
+            stem = self._shared_output_stem(stem, audio_path, splits, sr, len(audio[0]))
 
         split_start = 0
         split_filepaths, actual_splits, split_durations = [], [], []
 
         for k, split in enumerate(splits):
             split_name = f"{stem}.{k + 1}_of_{1 + len(splits)}.wav"
-            split_filepath = f"{parent_url}/{split_name}" if parent_url else split_name
-            split_resolved = f"{resolved_parent}/{split_name}" if resolved_parent else split_name
+            split_filepath, split_resolved = self._split_paths(
+                split_name,
+                parent_url,
+                resolved_parent,
+                resolved_output_dir,
+            )
             split_end = math.ceil(split * sr)
 
             if split_end - split_start > self.min_len * sr:
-                torchaudio.save(split_resolved, audio[:, split_start:split_end], sr)
+                self._save_split(split_resolved, audio[:, split_start:split_end], sr, output_fs)
                 split_filepaths.append(split_filepath)
                 actual_splits.append(split_start / sr)
                 split_durations.append((split_end - split_start) / sr)
                 split_start = split_end
 
         split_name = f"{stem}.{1 + len(splits)}_of_{1 + len(splits)}.wav"
-        split_filepath = f"{parent_url}/{split_name}" if parent_url else split_name
-        split_resolved = f"{resolved_parent}/{split_name}" if resolved_parent else split_name
+        split_filepath, split_resolved = self._split_paths(
+            split_name,
+            parent_url,
+            resolved_parent,
+            resolved_output_dir,
+        )
         last_frame = len(audio[0])
         remaining_frames = last_frame - split_start
 
         if remaining_frames > self.min_len * sr:
-            torchaudio.save(split_resolved, audio[:, split_start:], sr)
+            self._save_split(split_resolved, audio[:, split_start:], sr, output_fs)
             split_filepaths.append(split_filepath)
             split_durations.append(remaining_frames / sr)
             actual_splits.append(split_start / sr)
 
-        audio_item_id, split_filepaths_before = data_entry.get("audio_item_id", "unknown"), bool(split_filepaths)
+        audio_item_id, split_filepaths_before = (
+            data_entry.get(self.audio_item_id_key, "unknown"),
+            bool(split_filepaths),
+        )
 
         if not split_filepaths:
             logger.warning(
@@ -159,20 +304,20 @@ def _do_split(self, task: AudioTask) -> AudioTask:
             split_durations = [duration]
             actual_splits = [0.0]
 
-        data_entry["split_metadata"] = self._build_split_metadata(
+        data_entry[self.split_metadata_key] = self._build_split_metadata(
             audio_item_id,
             split_filepaths,
             split_durations,
             fallback=not split_filepaths_before,
         )
-        data_entry["split_filepaths"] = split_filepaths
-        data_entry["split_offsets"] = actual_splits
-        data_entry["split_timestamps"] = splits
+        data_entry[self.split_filepaths_key] = split_filepaths
+        data_entry[self.split_offsets_key] = actual_splits
+        data_entry[self.split_timestamps_key] = splits
         self._log_metrics({"input_duration": duration, "splits_produced": len(split_filepaths)})
         return task
 
-    @staticmethod
     def _build_split_metadata(
+        self,
         audio_item_id: str,
         split_filepaths: list[str],
         split_durations: list[float],
@@ -183,23 +328,23 @@ def _build_split_metadata(
         if fallback:
             return [
                 {
-                    "audio_item_id": audio_item_id,
-                    "resampled_audio_filepath": split_filepaths[0],
-                    "duration": split_durations[0],
+                    self.audio_item_id_key: audio_item_id,
+                    self.audio_filepath_key: split_filepaths[0],
+                    self.duration_key: split_durations[0],
                 }
             ]
         return [
             {
-                "audio_item_id": f"{audio_item_id}_{idx}",
-                "resampled_audio_filepath": path,
-                "duration": split_durations[idx],
+                self.audio_item_id_key: f"{audio_item_id}_{idx}",
+                self.audio_filepath_key: path,
+                self.duration_key: split_durations[idx],
             }
             for idx, path in enumerate(split_filepaths)
         ]
 
 
 @dataclass
-class JoinSplitAudioMetadataStage(ProcessingStage[AudioTask, AudioTask]):
+class JoinSplitAudioMetadataStage(AgentReady, ProcessingStage[AudioTask, AudioTask]):
     """
     Stage for joining metadata of previously split audio files.
 
@@ -213,13 +358,42 @@ class JoinSplitAudioMetadataStage(ProcessingStage[AudioTask, AudioTask]):
     """
 
     text_key: str = "text"
+
+    # Stage metadata. Keep the legacy positional fields before additive keys.
     name: str = "JoinSplitAudioMetadata"
 
+    split_filepaths_key: str = "split_filepaths"
+    split_metadata_key: str = "split_metadata"
+    split_offsets_key: str = "split_offsets"
+    split_timestamps_key: str = "split_timestamps"
+    alignment_key: str = "alignment"
+
     def inputs(self) -> tuple[list[str], list[str]]:
-        return [], ["split_filepaths", "split_metadata", "split_offsets", "split_timestamps"]
+        return [], [
+            self.split_filepaths_key,
+            self.split_metadata_key,
+            self.split_offsets_key,
+            self.split_timestamps_key,
+        ]
 
     def outputs(self) -> tuple[list[str], list[str]]:
-        return [], [self.text_key, "alignment"]
+        return [], [self.text_key, self.alignment_key]
+
+    def describe(self) -> StageContract:
+        return StageContract(
+            reads=IOSpec(
+                data_keys=[
+                    self.split_filepaths_key,
+                    self.split_metadata_key,
+                    self.split_offsets_key,
+                    self.split_timestamps_key,
+                ]
+            ),
+            writes=IOSpec(data_keys=[self.text_key, self.alignment_key]),
+            removes_keys=[self.split_filepaths_key, self.split_metadata_key],
+            # Rejoins the chunks THIS row was split into, all of which came from its own file.
+            gates=Gates(per_row_independent=True),
+        )
 
     def process(self, task: AudioTask) -> AudioTask:
         """
@@ -234,13 +408,16 @@ def process(self, task: AudioTask) -> AudioTask:
         words_aligned = 0
 
         # Check if this is a meta-entry with split information
-        if "split_filepaths" in data_entry:
-            if data_entry["split_filepaths"] is None:
-                del data_entry["split_filepaths"]
+        if self.split_filepaths_key in data_entry:
+            if data_entry[self.split_filepaths_key] is None:
+                data_entry.setdefault(self.text_key, "")
+                data_entry.setdefault(self.alignment_key, [])
+                for key in [self.split_filepaths_key, self.split_metadata_key]:
+                    data_entry.pop(key, None)
             else:
-                splits_joined = len(data_entry.get("split_metadata", []))
+                splits_joined = len(data_entry.get(self.split_metadata_key, []) or [])
                 self._join_split_metadata(data_entry)
-                words_aligned = len(data_entry.get("alignment", []))
+            words_aligned = len(data_entry.get(self.alignment_key, []) or [])
 
         self._log_metrics(
             {
@@ -253,11 +430,14 @@ def process(self, task: AudioTask) -> AudioTask:
 
     def _join_split_metadata(self, meta_entry: dict) -> None:
         """Join metadata from split audio files."""
-        split_metadata = meta_entry.get("split_metadata", [])
-        split_offsets = meta_entry.get("split_offsets", [])
+        split_metadata = meta_entry.get(self.split_metadata_key, [])
+        split_offsets = meta_entry.get(self.split_offsets_key, [])
 
         if not split_metadata:
-            del meta_entry["split_filepaths"]
+            meta_entry.setdefault(self.text_key, "")
+            meta_entry.setdefault(self.alignment_key, [])
+            for key in [self.split_filepaths_key, self.split_metadata_key]:
+                meta_entry.pop(key, None)
             return
 
         transcripts = []
@@ -269,7 +449,7 @@ def _join_split_metadata(self, meta_entry: dict) -> None:
             if text:
                 transcripts.append(text)
 
-            alignment = split_entry.get("alignment", [])
+            alignment = split_entry.get(self.alignment_key, [])
             offset = split_offsets[idx] if idx < len(split_offsets) else 0
 
             for word in alignment:
@@ -280,15 +460,15 @@ def _join_split_metadata(self, meta_entry: dict) -> None:
 
         # Create joined entry
         meta_entry[self.text_key] = " ".join(transcripts)
-        meta_entry["alignment"] = alignments
+        meta_entry[self.alignment_key] = alignments
 
         # Remove split-related fields
-        for key in ["split_filepaths", "split_metadata"]:
+        for key in [self.split_filepaths_key, self.split_metadata_key]:
             meta_entry.pop(key, None)
 
 
 @dataclass
-class SplitASRAlignJoinStage(CompositeStage[AudioTask, AudioTask]):
+class SplitASRAlignJoinStage(AgentReady, CompositeStage[AudioTask, AudioTask]):
     """Composite stage: Split long audio -> ASR align -> Join results.
 
     Decomposes into three sequential stages that always run together:
@@ -299,6 +479,8 @@ class SplitASRAlignJoinStage(CompositeStage[AudioTask, AudioTask]):
     Args:
         suggested_max_len: Target max length for audio segments (seconds).
         min_len: Minimum length for any split segment (also used by ASR).
+        output_dir: Optional directory for split audio chunks. When unset,
+            chunks are written beside their source audio.
         max_len: Maximum length of audio segments for ASR processing (seconds).
         model_name: Pretrained NeMo ASR model name.
         model_path: Local model file path (overrides ``model_name`` if set).
@@ -349,15 +531,32 @@ class SplitASRAlignJoinStage(CompositeStage[AudioTask, AudioTask]):
     segments_key: str = "segments"
 
     name: str = "SplitASRAlignJoin"
+    # Additive agent-only routing knob. Keep it after every legacy field so
+    # positional construction retains its historical argument order.
+    output_dir: str | None = None
 
     def __post_init__(self) -> None:
         super().__init__()
 
+    def describe(self) -> StageContract:
+        return StageContract(
+            wrappable=False,
+            # Mirrors the delegate that decides it: the aligner and the join are per-row, so the
+            # composite is independent exactly when its ``SplitLongAudioStage`` is -- which is
+            # when no ``output_dir`` flattens every source's splits into one namespace.
+            gates=Gates(per_row_independent=self.output_dir is None),
+        )
+
     def decompose(self) -> list[ProcessingStage]:
         return [
             SplitLongAudioStage(
                 suggested_max_len=self.suggested_max_len,
                 min_len=self.min_len,
+                output_dir=self.output_dir,
+                # Forwarded, or configuring the composite would silently not reach the splitter:
+                # it would keep reading "segments" while the aligner read the configured key,
+                # which is what blocks feeding diarization segments into ASR.
+                segments_key=self.segments_key,
             ),
             NeMoASRAlignerStage(
                 model_name=self.model_name,
diff --git a/nemo_curator/stages/audio/tagging/text/chinese_conversion.py b/nemo_curator/stages/audio/tagging/text/chinese_conversion.py
index ea07008bad..9e9007f235 100644
--- a/nemo_curator/stages/audio/tagging/text/chinese_conversion.py
+++ b/nemo_curator/stages/audio/tagging/text/chinese_conversion.py
@@ -21,12 +21,19 @@
 from opencc import OpenCC
 
 from nemo_curator.backends.base import WorkerMetadata
+from nemo_curator.stages.audio._agent._agent_ready import (
+    AgentReady,
+    ConditionalWrite,
+    Gates,
+    IOSpec,
+    StageContract,
+)
 from nemo_curator.stages.base import ProcessingStage
 from nemo_curator.tasks import AudioTask
 
 
 @dataclass
-class ChineseConversionStage(ProcessingStage[AudioTask, AudioTask]):
+class ChineseConversionStage(AgentReady, ProcessingStage[AudioTask, AudioTask]):
     """Convert Traditional Chinese text to Simplified Chinese (or other OpenCC conversions).
 
     Iterates over the ``segments`` list of each entry and writes the converted
@@ -40,6 +47,8 @@ class ChineseConversionStage(ProcessingStage[AudioTask, AudioTask]):
 
     text_key: str = "text"
     convert_type: str = "t2s"
+    segments_key: str = "segments"
+    output_suffix: str = "_simplified"
 
     # Stage metadata
     name: str = "ChineseConversion"
@@ -48,10 +57,23 @@ class ChineseConversionStage(ProcessingStage[AudioTask, AudioTask]):
     _converter: Any = field(default=None, repr=False)
 
     def inputs(self) -> tuple[list[str], list[str]]:
-        return [], ["segments"]
+        return [], [self.segments_key]
 
     def outputs(self) -> tuple[list[str], list[str]]:
-        return [], ["segments"]
+        return [], [self.segments_key]
+
+    def describe(self) -> StageContract:
+        return StageContract(
+            reads=IOSpec(data_keys=[self.segments_key]),
+            conditional_writes=[
+                ConditionalWrite(
+                    writes=IOSpec(segment_data_keys=[f"{self.text_key}{self.output_suffix}"]),
+                    condition=f"an item in '{self.segments_key}' contains '{self.text_key}'",
+                )
+            ],
+            # The OpenCC mapping is fixed by ``convert_type``; each segment's text converts alone.
+            gates=Gates(per_row_independent=True),
+        )
 
     def setup(self, _worker_metadata: WorkerMetadata | None = None) -> None:
         """Setup stage."""
@@ -61,8 +83,8 @@ def setup(self, _worker_metadata: WorkerMetadata | None = None) -> None:
 
     def process(self, task: AudioTask) -> AudioTask:
         data_entry = task.data
-        output_key = f"{self.text_key}_simplified"
-        for segment in data_entry.get("segments", []):
+        output_key = f"{self.text_key}{self.output_suffix}"
+        for segment in data_entry.get(self.segments_key, []):
             if self.text_key in segment:
                 try:
                     segment[output_key] = self._converter.convert(segment[self.text_key])
diff --git a/nemo_curator/stages/audio/tagging/text/itn.py b/nemo_curator/stages/audio/tagging/text/itn.py
index 1b0863b5fc..f81c38bc36 100644
--- a/nemo_curator/stages/audio/tagging/text/itn.py
+++ b/nemo_curator/stages/audio/tagging/text/itn.py
@@ -23,12 +23,19 @@
 )
 
 from nemo_curator.backends.base import WorkerMetadata
+from nemo_curator.stages.audio._agent._agent_ready import (
+    AgentReady,
+    ConditionalWrite,
+    Gates,
+    IOSpec,
+    StageContract,
+)
 from nemo_curator.stages.base import ProcessingStage
 from nemo_curator.tasks import AudioTask
 
 
 @dataclass
-class InverseTextNormalizationStage(ProcessingStage[AudioTask, AudioTask]):
+class InverseTextNormalizationStage(AgentReady, ProcessingStage[AudioTask, AudioTask]):
     """
     Stage that performs inverse text normalization on text data.
 
@@ -45,6 +52,8 @@ class InverseTextNormalizationStage(ProcessingStage[AudioTask, AudioTask]):
 
     # Text key
     text_key: str = "text"
+    segments_key: str = "segments"
+    output_suffix: str = "_ITN"
 
     # Stage metadata
     name: str = "InverseTextNormalization"
@@ -52,10 +61,25 @@ class InverseTextNormalizationStage(ProcessingStage[AudioTask, AudioTask]):
     _normalizer: Any = field(default=None, repr=False)
 
     def inputs(self) -> tuple[list[str], list[str]]:
-        return [], ["segments"]
+        return [], [self.segments_key]
 
     def outputs(self) -> tuple[list[str], list[str]]:
-        return [], ["segments"]
+        return [], [self.segments_key]
+
+    def describe(self) -> StageContract:
+        return StageContract(
+            reads=IOSpec(data_keys=[self.segments_key]),
+            conditional_writes=[
+                ConditionalWrite(
+                    writes=IOSpec(segment_data_keys=[f"{self.text_key}{self.output_suffix}"]),
+                    condition=(
+                        f"an item in '{self.segments_key}' contains a truthy '{self.text_key}' value"
+                    ),
+                )
+            ],
+            # The normalizer's grammars come from ``language``; each segment's text normalizes alone.
+            gates=Gates(per_row_independent=True),
+        )
 
     def setup(self, _worker_metadata: WorkerMetadata | None = None) -> None:
         """Load the inverse normalizer once per worker."""
@@ -66,13 +90,13 @@ def setup(self, _worker_metadata: WorkerMetadata | None = None) -> None:
     def process(self, task: AudioTask) -> AudioTask:
         """Process entry for inverse text normalization."""
         data_entry = task.data
-        segments = data_entry.get("segments", [])
+        segments = data_entry.get(self.segments_key, [])
         for segment in segments:
             if self.text_key in segment:
                 text = segment[self.text_key]
                 if text:
                     sentences = self._normalizer.split_text_into_sentences(text)
                     text_itn = " ".join(self._normalizer.normalize_list(sentences))
-                    segment[f"{self.text_key}_ITN"] = text_itn
+                    segment[f"{self.text_key}{self.output_suffix}"] = text_itn
 
         return task
diff --git a/tests/stages/audio/tagging/inference/test_nemo_asr_align.py b/tests/stages/audio/tagging/inference/test_nemo_asr_align.py
index 9d758be381..19278625bb 100644
--- a/tests/stages/audio/tagging/inference/test_nemo_asr_align.py
+++ b/tests/stages/audio/tagging/inference/test_nemo_asr_align.py
@@ -14,13 +14,34 @@
 
 from pathlib import Path
 from typing import Any
-from unittest.mock import MagicMock
+from unittest.mock import MagicMock, patch
+
+import pytest
+import torch
+from nemo_curator.stages.audio._agent._agent_registry import build_contract
+from nemo_curator.stages.audio._agent._conformance import assert_agent_ready
+from nemo_curator.stages.audio._agent._planning import validate_pipeline
 
 from nemo_curator.stages.audio.tagging.inference.nemo_asr_align import NeMoASRAlignerStage
+from nemo_curator.stages.audio.tagging.merge_alignment_diarization import (
+    MergeAlignmentDiarizationStage,
+)
+from nemo_curator.stages.audio.tagging.split import JoinSplitAudioMetadataStage
 from nemo_curator.stages.resources import Resources
 from nemo_curator.tasks import AudioTask
 
 
+def _stub_asr(stage: NeMoASRAlignerStage) -> MagicMock:
+    model = MagicMock()
+    model.transcribe.return_value = [MagicMock()]
+    stage._asr_model = model
+    stage._override_cfg = MagicMock()
+    stage.get_alignments_text = MagicMock(
+        return_value=([{"word": "hello", "start": 0.0, "end": 0.5, "confidence": 0.9}], "hello")
+    )
+    return model
+
+
 class TestNeMoASRAlignerStage:
     def test_io_contracts(self) -> None:
         full_audio_stage = NeMoASRAlignerStage(infer_segment_only=False)
@@ -37,6 +58,220 @@ def test_io_contracts(self) -> None:
         assert segment_stage.inputs() == (["data"], ["resampled_audio_filepath", "segments"])
         assert segment_stage.outputs() == (["data"], ["resampled_audio_filepath", "segments"])
 
+    def test_describe_scopes_configured_full_and_segment_writes(self) -> None:
+        full = build_contract(
+            NeMoASRAlignerStage(
+                infer_segment_only=False,
+                split_metadata_key="chunks",
+                split_filepaths_key="chunk_paths",
+                text_key="transcript",
+                alignment_key="token_alignment",
+                words_key="must_not_be_claimed",
+            )
+        )
+        assert full.cardinality == "1:1 nested-list"
+        assert full.iteration_key == "chunks"
+        assert full.writes.data_keys == []
+        assert full.writes.segment_data_keys == ["transcript", "token_alignment"]
+        assert "must_not_be_claimed" not in full.writes.segment_data_keys
+        assert len(full.conditional_writes) == 1
+        assert full.conditional_writes[0].writes.data_keys == ["transcript", "token_alignment"]
+        assert full.conditional_writes[0].writes.segment_data_keys == []
+
+        segment_only = build_contract(
+            NeMoASRAlignerStage(
+                infer_segment_only=True,
+                segments_key="utterances",
+                text_key="transcript",
+                words_key="tokens",
+            )
+        )
+        assert segment_only.cardinality == "1:1 nested-list"
+        assert segment_only.iteration_key == "utterances"
+        assert segment_only.writes.data_keys == []
+        assert segment_only.writes.segment_data_keys == ["transcript", "tokens"]
+        assert segment_only.conditional_writes == []
+
+    def test_agent_ready_full_normal_split_writes_nested_metadata(self) -> None:
+        stage = NeMoASRAlignerStage(
+            resources=Resources(cpus=1.0),
+            split_filepaths_key="chunk_paths",
+            split_metadata_key="chunks",
+            text_key="transcript",
+            alignment_key="token_alignment",
+        )
+        _stub_asr(stage)
+        task = AudioTask(
+            dataset_name="test",
+            data={
+                "duration": 2.0,
+                "segments": [],
+                "chunk_paths": ["chunk.wav"],
+                "chunks": [{"start": 0.0, "end": 2.0}],
+            },
+        )
+
+        assert_agent_ready(
+            stage,
+            lambda: task,
+            expected_cardinality="1:1 nested-list",
+            available_keys={"duration", "segments", "chunk_paths", "chunks"},
+        )
+
+        assert task.data["chunks"][0]["transcript"] == "hello"
+        assert task.data["chunks"][0]["token_alignment"][0]["word"] == "hello"
+        assert "transcript" not in task.data
+        assert "token_alignment" not in task.data
+
+    def test_agent_ready_full_no_split_and_missing_metadata_write_top_level(self) -> None:
+        no_split_stage = NeMoASRAlignerStage(resources=Resources(cpus=1.0))
+        no_split_model = _stub_asr(no_split_stage)
+        no_split_task = AudioTask(
+            dataset_name="test",
+            data={"duration": 0.0, "segments": [], "split_filepaths": [], "split_metadata": []},
+        )
+        assert_agent_ready(
+            no_split_stage,
+            lambda: no_split_task,
+            expected_cardinality="1:1 nested-list",
+            available_keys={"duration", "segments", "split_filepaths", "split_metadata"},
+        )
+        assert no_split_task.data["text"] == ""
+        assert no_split_task.data["alignment"] == []
+        no_split_model.transcribe.assert_not_called()
+
+        fallback_stage = NeMoASRAlignerStage(resources=Resources(cpus=1.0))
+        _stub_asr(fallback_stage)
+        fallback_task = AudioTask(
+            dataset_name="test",
+            data={
+                "duration": 2.0,
+                "segments": [],
+                "split_filepaths": ["chunk.wav"],
+                "split_metadata": [],
+            },
+        )
+        assert_agent_ready(
+            fallback_stage,
+            lambda: fallback_task,
+            expected_cardinality="1:1 nested-list",
+            available_keys={"duration", "segments", "split_filepaths", "split_metadata"},
+        )
+        assert fallback_task.data["text"] == "hello"
+        assert fallback_task.data["alignment"][0]["word"] == "hello"
+
+    def test_agent_ready_segment_only_writes_configured_segments(self) -> None:
+        stage = NeMoASRAlignerStage(
+            infer_segment_only=True,
+            resources=Resources(cpus=1.0),
+            segments_key="utterances",
+            text_key="transcript",
+            words_key="tokens",
+        )
+        _stub_asr(stage)
+        task = AudioTask(
+            dataset_name="test",
+            data={
+                "resampled_audio_filepath": "audio.wav",
+                "utterances": [{"start": 1.0, "end": 3.0}],
+            },
+        )
+
+        with patch(
+            "nemo_curator.stages.audio.tagging.inference.nemo_asr_align.torchaudio.load",
+            return_value=(torch.zeros(1, 32000), 16000),
+        ):
+            assert_agent_ready(
+                stage,
+                lambda: task,
+                expected_cardinality="1:1 nested-list",
+                available_keys={"resampled_audio_filepath", "utterances"},
+            )
+
+        assert task.data["utterances"][0]["transcript"] == "hello"
+        assert task.data["utterances"][0]["tokens"][0]["start"] == 1.0
+
+    def test_nested_full_alignment_does_not_plan_as_top_level_merge_input(self) -> None:
+        aligner = NeMoASRAlignerStage(resources=Resources(cpus=1.0))
+        merger = MergeAlignmentDiarizationStage()
+        report = validate_pipeline(
+            [aligner, merger],
+            initial_keys={"duration", "segments", "split_filepaths", "split_metadata"},
+            initial_task_type="AudioTask",
+        )
+        assert not report.ok
+        assert any(
+            issue.stage_index == 1 and issue.code == "unsatisfied_reads" for issue in report.issues
+        )
+
+        _stub_asr(aligner)
+        task = AudioTask(
+            dataset_name="test",
+            data={
+                "duration": 2.0,
+                "segments": [{"speaker": "s1", "start": 0.0, "end": 2.0}],
+                "split_filepaths": ["chunk.wav"],
+                "split_metadata": [{"start": 0.0, "end": 2.0}],
+            },
+        )
+        aligner.process(task)
+        assert task.data["split_metadata"][0]["alignment"][0]["word"] == "hello"
+        assert "alignment" not in task.data
+
+        with pytest.raises(ValueError, match="failed validation"):
+            merger.process_batch([task])
+
+    def test_tutorial_full_asr_join_merge_chain(self) -> None:
+        aligner = NeMoASRAlignerStage(resources=Resources(cpus=1.0))
+        joiner = JoinSplitAudioMetadataStage()
+        merger = MergeAlignmentDiarizationStage()
+        stages = [aligner, joiner, merger]
+        initial_keys = {
+            "duration",
+            "segments",
+            "split_filepaths",
+            "split_metadata",
+            "split_offsets",
+            "split_timestamps",
+        }
+
+        report = validate_pipeline(
+            stages,
+            initial_roles={"duration", "segments"},
+            initial_keys=initial_keys,
+            initial_task_type="AudioTask",
+        )
+
+        assert report.ok
+        assert report.keys_ok
+        assert "text" in report.produced_keys
+        assert "alignment" in report.produced_keys
+        assert "split_filepaths" not in report.produced_keys
+        assert "split_metadata" not in report.produced_keys
+
+        _stub_asr(aligner)
+        task = AudioTask(
+            dataset_name="test",
+            data={
+                "duration": 2.0,
+                "segments": [{"speaker": "s1", "start": 0.0, "end": 2.0}],
+                "split_filepaths": ["chunk.wav"],
+                "split_metadata": [{"start": 0.0, "end": 2.0}],
+                "split_offsets": [0.0],
+                "split_timestamps": [],
+            },
+        )
+
+        for stage in stages:
+            stage.process(task)
+
+        assert task.data["text"] == "hello"
+        assert task.data["alignment"][0]["word"] == "hello"
+        assert task.data["segments"][0]["text"] == "hello"
+        assert task.data["segments"][0]["words"] == task.data["alignment"]
+        assert "split_filepaths" not in task.data
+        assert "split_metadata" not in task.data
+
     def test_setup_configures_rnnt_cuda_graphs(self) -> None:
         model = MagicMock()
         stage = NeMoASRAlignerStage(
diff --git a/tests/stages/audio/tagging/test_merge_alignment_diarization.py b/tests/stages/audio/tagging/test_merge_alignment_diarization.py
index dc728770da..3d614e0282 100644
--- a/tests/stages/audio/tagging/test_merge_alignment_diarization.py
+++ b/tests/stages/audio/tagging/test_merge_alignment_diarization.py
@@ -14,6 +14,10 @@
 
 from collections.abc import Callable
 
+import pytest
+from nemo_curator.stages.audio._agent._conformance import assert_agent_ready
+from nemo_curator.stages.audio._agent._planning import validate_pipeline
+
 from nemo_curator.stages.audio.tagging.merge_alignment_diarization import (
     MergeAlignmentDiarizationStage,
 )
@@ -110,3 +114,69 @@ def test_process_no_segments_passthrough(self, audio_task: Callable[..., AudioTa
         )
         result = stage.process(task)
         assert result.data["segments"] == []
+
+    @pytest.mark.parametrize(
+        ("data", "expected"),
+        [
+            (
+                {
+                    "alignment": [],
+                    "segments": [{"speaker": "s1", "start": 0.0, "end": 1.0}],
+                },
+                None,
+            ),
+            (
+                {
+                    "alignment": [{"word": "hello", "start": 0.0, "end": 0.5}],
+                    "segments": [],
+                },
+                None,
+            ),
+            (
+                {
+                    "alignment": [{"word": "hello", "start": 0.0, "end": 0.5}],
+                    "segments": [{"speaker": "s1", "start": 0.0, "end": 1.0}],
+                },
+                "hello",
+            ),
+        ],
+        ids=["empty-alignment", "empty-segments", "populated"],
+    )
+    def test_agent_ready_conditional_nested_outputs(
+        self,
+        data: dict,
+        expected: str | None,
+    ) -> None:
+        stage = MergeAlignmentDiarizationStage()
+        task = AudioTask(dataset_name="test", data=data)
+
+        contract = assert_agent_ready(
+            stage,
+            lambda: task,
+            segments_key="segments",
+        )
+
+        assert contract.reads.data_keys == ["alignment", "segments"]
+        assert contract.writes.segment_data_keys == []
+        assert len(contract.conditional_writes) == 1
+        assert contract.conditional_writes[0].writes.segment_data_keys == ["text", "words"]
+        if expected is None:
+            assert all(
+                "text" not in segment and "words" not in segment
+                for segment in task.data.get("segments", [])
+            )
+        else:
+            assert task.data["segments"][0]["text"] == expected
+            assert task.data["segments"][0]["words"] == task.data["alignment"]
+
+    def test_planner_does_not_guarantee_conditional_nested_outputs(self) -> None:
+        report = validate_pipeline(
+            [MergeAlignmentDiarizationStage()],
+            initial_roles={"alignment", "segments"},
+            initial_keys={"alignment", "segments"},
+            initial_task_type="AudioTask",
+        )
+
+        assert report.ok
+        assert "text" not in report.produced_keys
+        assert "words" not in report.produced_keys
diff --git a/tests/stages/audio/tagging/test_prepare_module_segments.py b/tests/stages/audio/tagging/test_prepare_module_segments.py
index e66162e6e8..78e678c238 100644
--- a/tests/stages/audio/tagging/test_prepare_module_segments.py
+++ b/tests/stages/audio/tagging/test_prepare_module_segments.py
@@ -190,7 +190,7 @@ def test_prepare_module_segments_stage_sdp_style_input(
         max_pause=2,
         text_key="text",
         words_key="words",
-        terminal_punct_marks=".!?。？？！。",  # noqa: RUF001
+        terminal_punct_marks=".!?。？？！。",  # noqa: RUF001 - fullwidth CJK punctuation is the test input
         full_utterance_ratio=1.0,
         punctuation_split_only=False,
     )
diff --git a/tests/stages/audio/tagging/test_resample_audio.py b/tests/stages/audio/tagging/test_resample_audio.py
index 48c8a27a82..a0284b4d69 100644
--- a/tests/stages/audio/tagging/test_resample_audio.py
+++ b/tests/stages/audio/tagging/test_resample_audio.py
@@ -12,12 +12,23 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import hashlib
+import os
+import shutil
 import subprocess
 import tempfile
 from collections.abc import Callable
 from pathlib import Path
+from types import SimpleNamespace
+from typing import Any
 
+import numpy as np
 import pytest
+import soundfile as sf
+import torch
+from nemo_curator.stages.audio._agent._agent_registry import build_contract, static_contract
+from nemo_curator.stages.audio._agent._conformance import assert_agent_ready
+from nemo_curator.stages.audio._agent._residency import resolve_audio
 
 import nemo_curator.stages.audio.tagging.resample_audio as resample_audio_module
 from nemo_curator.stages.audio.tagging.resample_audio import ResampleAudioStage
@@ -41,6 +52,337 @@ def test_process(self, audio_task: Callable[..., AudioTask], audio_filepath: Pat
             assert out.get("resampled_audio_filepath") == f"{tmpdir}/id_1.wav"
             assert out.get("duration") == 60.0
 
+    def test_a_file_input_keeps_the_name_it_has_always_had(self, audio_filepath: Path) -> None:
+        """Every tutorial reads real files off disk; their output names must not move."""
+        with tempfile.TemporaryDirectory() as tmpdir:
+            stage = ResampleAudioStage(resampled_audio_dir=tmpdir)
+            stage.setup()
+            stage.process(AudioTask(task_id="t", dataset_name="d", data={"audio_filepath": str(audio_filepath)}))
+
+            path_hash = hashlib.sha256(str(audio_filepath).encode()).hexdigest()[:8]
+            assert os.listdir(tmpdir) == [f"{audio_filepath.stem}_{path_hash}.wav"]
+
+    def test_a_waveform_input_writes_one_file_however_often_it_is_rerun(self) -> None:
+        waveform = torch.sin(torch.arange(0, 16000 * 2) * 0.01).unsqueeze(0)
+        with tempfile.TemporaryDirectory() as tmpdir:
+            for _ in range(3):
+                stage = ResampleAudioStage(resampled_audio_dir=tmpdir, input_residency="waveform")
+                stage.setup()
+                stage.process(
+                    AudioTask(
+                        task_id="t",
+                        dataset_name="d",
+                        data={"waveform": waveform.clone(), "sample_rate": 16000},
+                    )
+                )
+
+            assert len(os.listdir(tmpdir)) == 1, "the same audio must not pile up a file per run"
+
+    def test_changing_the_target_rate_does_not_reuse_the_old_conversion(self) -> None:
+        """The name carries the settings, so 'it already exists, skip it' cannot serve 48 kHz for 16."""
+        waveform = torch.sin(torch.arange(0, 16000 * 2) * 0.01).unsqueeze(0)
+        with tempfile.TemporaryDirectory() as tmpdir:
+            for rate in (16000, 8000):
+                stage = ResampleAudioStage(
+                    resampled_audio_dir=tmpdir, input_residency="waveform", target_sample_rate=rate
+                )
+                stage.setup()
+                stage.process(
+                    AudioTask(
+                        task_id="t",
+                        dataset_name="d",
+                        data={"waveform": waveform.clone(), "sample_rate": 16000},
+                    )
+                )
+
+            assert len(os.listdir(tmpdir)) == 2, "a different target rate must not answer from the old file"
+
+    def test_a_second_run_at_a_new_rate_does_not_serve_the_old_file(self) -> None:
+        import numpy as np
+        import soundfile
+
+        with tempfile.TemporaryDirectory() as srcdir, tempfile.TemporaryDirectory() as out:
+            src = os.path.join(srcdir, "a.wav")
+            soundfile.write(src, np.sin(np.arange(48000) * 0.01).astype("float32"), 48000)
+
+            for rate in (16000, 8000):
+                stage = ResampleAudioStage(resampled_audio_dir=out, write_to_disk=True, target_sample_rate=rate)
+                stage.setup()
+                stage.process(AudioTask(task_id="t", dataset_name="d", data={"audio_filepath": src}))
+
+                written = os.path.join(out, os.listdir(out)[0])
+                assert soundfile.info(written).samplerate == rate, "served audio at the previous run's rate"
+
+    def test_segments_sharing_a_parent_id_each_get_their_own_file(self) -> None:
+        waveform = torch.sin(torch.arange(0, 16000) * 0.01).unsqueeze(0)
+        with tempfile.TemporaryDirectory() as tmpdir:
+            stage = ResampleAudioStage(resampled_audio_dir=tmpdir, input_residency="waveform")
+            stage.setup()
+            for segment in range(3):
+                stage.process(
+                    AudioTask(
+                        task_id="t",
+                        dataset_name="d",
+                        data={
+                            # What VAD hands every child: the parent's id, identical across siblings.
+                            "audio_item_id": "utt1",
+                            "waveform": (waveform * (segment + 1)).clone(),
+                            "sample_rate": 16000,
+                        },
+                    )
+                )
+
+            assert len(os.listdir(tmpdir)) == 3, "sibling segments collapsed onto one filename"
+
+    def test_process_batch_accepts_every_advertised_residency(
+        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        monkeypatch.setattr(resample_audio_module.subprocess, "run", _fake_ffmpeg_copy)
+        source = tmp_path / "source.wav"
+        sf.write(source, np.zeros(16000, dtype=np.float32), 16000)
+        waveform = torch.ones(1, 16000)
+        cases = [
+            ("file", {"audio_filepath": str(source)}),
+            ("waveform", {"waveform": waveform, "sample_rate": 16000}),
+            ("auto", {"audio_filepath": str(source)}),
+            ("auto", {"waveform": waveform, "sample_rate": 16000}),
+        ]
+
+        for index, (residency, data) in enumerate(cases):
+            stage = ResampleAudioStage(
+                resampled_audio_dir=str(tmp_path / f"unused-{index}"),
+                input_residency=residency,
+                write_to_disk=False,
+                keep_waveform_in_task=True,
+            )
+            result = stage.process_batch([AudioTask(dataset_name="d", data=dict(data))])
+            assert len(result) == 1
+            assert result[0].data["sample_rate"] == 16000
+
+    def test_process_batch_rejects_incomplete_residencies(self, tmp_path: Path) -> None:
+        waveform = torch.ones(1, 16)
+        cases = [
+            ("file", {}),
+            ("file", {"audio_filepath": None}),
+            ("waveform", {"waveform": waveform}),
+            ("waveform", {"sample_rate": 16000}),
+            ("waveform", {"waveform": None, "sample_rate": 16000}),
+            ("auto", {}),
+            ("auto", {"waveform": waveform}),
+        ]
+
+        for residency, data in cases:
+            stage = ResampleAudioStage(
+                resampled_audio_dir=str(tmp_path / "unused"),
+                input_residency=residency,
+                write_to_disk=False,
+                keep_waveform_in_task=True,
+            )
+            with pytest.raises(ValueError, match="failed validation"):
+                stage.process_batch([AudioTask(dataset_name="d", data=data)])
+
+    @pytest.mark.parametrize("write_to_disk", [False, True])
+    def test_output_cleanup_when_loading_converted_audio_fails(
+        self,
+        tmp_path: Path,
+        monkeypatch: pytest.MonkeyPatch,
+        write_to_disk: bool,
+    ) -> None:
+        monkeypatch.setattr(resample_audio_module.subprocess, "run", _fake_ffmpeg_copy)
+        source = tmp_path / "source.wav"
+        sf.write(source, np.zeros(16000, dtype=np.float32), 16000)
+        attempted_outputs: list[str] = []
+
+        def fail_load(path: str, *, mono: bool) -> tuple[torch.Tensor, int]:
+            assert mono is False
+            attempted_outputs.append(path)
+            message = "cannot load converted output"
+            raise OSError(message)
+
+        monkeypatch.setattr(resample_audio_module, "load_audio_file", fail_load)
+        stage = ResampleAudioStage(
+            resampled_audio_dir=str(tmp_path / "out"),
+            write_to_disk=write_to_disk,
+            keep_waveform_in_task=True,
+        )
+
+        with pytest.raises(OSError, match="cannot load"):
+            stage.process(
+                AudioTask(
+                    dataset_name="d",
+                    data={"audio_filepath": str(source), "audio_item_id": "failure"},
+                )
+            )
+
+        assert len(attempted_outputs) == 1
+        assert os.path.exists(attempted_outputs[0]) is write_to_disk
+
+    def test_disk_only_conversion_removes_stale_resident_audio(
+        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        monkeypatch.setattr(resample_audio_module.subprocess, "run", _fake_ffmpeg_convert)
+        source = tmp_path / "source.wav"
+        sf.write(source, np.zeros((8000, 2), dtype=np.float32), 8000)
+        task = AudioTask(
+            dataset_name="d",
+            data={
+                "audio_filepath": str(source),
+                "waveform": torch.stack([torch.zeros(8000), torch.ones(8000)]),
+                "sample_rate": 8000,
+            },
+        )
+        stage = ResampleAudioStage(
+            resampled_audio_dir=str(tmp_path / "out"),
+            input_residency="waveform",
+            target_sample_rate=16000,
+            target_nchannels=1,
+            write_to_disk=True,
+            keep_waveform_in_task=False,
+            update_audio_filepath=True,
+        )
+
+        result = stage.process(task)
+
+        assert "waveform" not in result.data
+        assert "sample_rate" not in result.data
+        assert set(build_contract(stage).removes_keys) == {"waveform", "sample_rate"}
+        consumed = resolve_audio(result.data, residency="auto", mono=False)
+        assert consumed is not None
+        converted, sample_rate = consumed
+        assert sample_rate == 16000
+        assert tuple(converted.shape[:1]) == (1,)
+
+    def test_sink_contracts_and_static_gates(
+        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        monkeypatch.setattr(resample_audio_module.subprocess, "run", _fake_ffmpeg_copy)
+        source = tmp_path / "source.wav"
+        sf.write(source, np.zeros(16000, dtype=np.float32), 16000)
+
+        with pytest.raises(ValueError, match="keep_waveform_in_task or write_to_disk"):
+            ResampleAudioStage(resampled_audio_dir=str(tmp_path), write_to_disk=False)
+        with pytest.raises(ValueError, match="update_audio_filepath"):
+            ResampleAudioStage(
+                resampled_audio_dir=str(tmp_path),
+                write_to_disk=False,
+                keep_waveform_in_task=True,
+                update_audio_filepath=True,
+            )
+
+        supported = [
+            {"write_to_disk": True, "keep_waveform_in_task": False},
+            {"write_to_disk": False, "keep_waveform_in_task": True},
+            {"write_to_disk": True, "keep_waveform_in_task": True},
+            {
+                "write_to_disk": True,
+                "keep_waveform_in_task": False,
+                "update_audio_filepath": True,
+            },
+        ]
+        for index, config in enumerate(supported):
+            stage = ResampleAudioStage(resampled_audio_dir=str(tmp_path / f"out-{index}"), **config)
+            assert_agent_ready(
+                stage,
+                lambda: AudioTask(dataset_name="d", data={"audio_filepath": str(source)}),
+                available_keys={"audio_filepath"},
+            )
+
+        replacement = build_contract(
+            ResampleAudioStage(
+                resampled_audio_dir=str(tmp_path / "replacement"),
+                update_audio_filepath=True,
+            )
+        )
+        assert replacement.writes.data_keys.count("audio_filepath") == 1
+        assert [write.writes.data_keys for write in replacement.conditional_writes] == [
+            ["original_audio_filepath"]
+        ]
+
+        static = static_contract(ResampleAudioStage)
+        configured = build_contract(ResampleAudioStage(resampled_audio_dir=str(tmp_path / "configured")))
+        assert static.gates == configured.gates
+
+
+def _fake_ffmpeg_copy(cmd: list[str], **_: Any) -> SimpleNamespace:  # noqa: ANN401
+    """Stand in for the ffmpeg call by copying the source to the requested output."""
+    src = cmd[cmd.index("-i") + 1]
+    dst = cmd[-1]
+    os.makedirs(os.path.dirname(dst), exist_ok=True)
+    shutil.copyfile(src, dst)
+    return SimpleNamespace(returncode=0)
+
+
+def _fake_ffmpeg_convert(cmd: list[str], **_: Any) -> SimpleNamespace:  # noqa: ANN401
+    """Apply the requested rate/channel header changes without invoking FFmpeg."""
+    src = cmd[cmd.index("-i") + 1]
+    dst = cmd[-1]
+    target_rate = int(cmd[cmd.index("-ar") + 1])
+    target_channels = int(cmd[cmd.index("-ac") + 1])
+    samples, source_rate = sf.read(src, always_2d=True)
+    if target_channels == 1:
+        samples = samples.mean(axis=1, keepdims=True)
+    output_frames = round(len(samples) * target_rate / source_rate)
+    old_positions = np.linspace(0.0, 1.0, len(samples), endpoint=False)
+    new_positions = np.linspace(0.0, 1.0, output_frames, endpoint=False)
+    converted = np.stack(
+        [np.interp(new_positions, old_positions, samples[:, channel]) for channel in range(target_channels)],
+        axis=1,
+    )
+    os.makedirs(os.path.dirname(dst), exist_ok=True)
+    sf.write(dst, converted, target_rate)
+    return SimpleNamespace(returncode=0)
+
+
+class TestSkippingExistingOutput:
+    """Re-running is idempotent on disk, but an in-memory run must never wrongly skip.
+
+    Lifted from tests/stages/audio/test_agent_simulation_pipelines.py: it drives only
+    ResampleAudioStage, and counts ffmpeg invocations -- a property the naming tests above
+    do not cover, since they count output FILES rather than conversions.
+    """
+
+    def test_disk_output_is_converted_once_but_memory_output_every_time(
+        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        from nemo_curator.stages.audio.tagging import resample_audio as resample_module
+
+        calls = {"n": 0}
+
+        def counting_ffmpeg(cmd: list[str], **kwargs: Any) -> SimpleNamespace:  # noqa: ANN401
+            calls["n"] += 1
+            return _fake_ffmpeg_copy(cmd, **kwargs)
+
+        monkeypatch.setattr(resample_module.subprocess, "run", counting_ffmpeg)
+        source = tmp_path / "src.wav"
+        sf.write(source, torch.linspace(-0.25, 0.25, 16000).numpy(), 16000)
+
+        disk_stage = ResampleAudioStage(
+            resampled_audio_dir=str(tmp_path / "out"),
+            input_residency="file",
+            write_to_disk=True,
+            keep_waveform_in_task=False,
+        )
+
+        def disk_task() -> AudioTask:
+            return AudioTask(dataset_name="t", data={"audio_filepath": str(source), "audio_item_id": "fixed_id"})
+
+        disk_stage.process(disk_task())
+        assert calls["n"] == 1
+        disk_stage.process(disk_task())
+        assert calls["n"] == 1, "the output already exists on disk, so it must be skipped"
+
+        # write_to_disk=False writes to a fresh temp path each run, so it must always convert.
+        calls["n"] = 0
+        mem_stage = ResampleAudioStage(
+            resampled_audio_dir=str(tmp_path / "unused"),
+            input_residency="file",
+            write_to_disk=False,
+            keep_waveform_in_task=True,
+        )
+        for _ in range(2):
+            mem_stage.process(AudioTask(dataset_name="t", data={"audio_filepath": str(source), "audio_item_id": "m"}))
+        assert calls["n"] == 2, "an in-memory run has no durable output to skip"
+
     def test_process_removes_partial_output_after_ffmpeg_failure(
         self,
         tmp_path: Path,
diff --git a/tests/stages/audio/tagging/test_split.py b/tests/stages/audio/tagging/test_split.py
index 48f349719b..f457e11007 100644
--- a/tests/stages/audio/tagging/test_split.py
+++ b/tests/stages/audio/tagging/test_split.py
@@ -12,19 +12,56 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import hashlib
 from collections.abc import Callable
 from pathlib import Path
 
 import numpy as np
+import pytest
 import soundfile as sf
+from fsspec.core import url_to_fs
+from nemo_curator.stages.audio._agent._agent_registry import build_contract, static_contract
+from nemo_curator.stages.audio._agent._conformance import assert_agent_ready
+from nemo_curator.stages.audio._agent._planning import validate_pipeline
 
+from nemo_curator.stages.audio.tagging.merge_alignment_diarization import (
+    MergeAlignmentDiarizationStage,
+)
 from nemo_curator.stages.audio.tagging.split import (
     JoinSplitAudioMetadataStage,
+    SplitASRAlignJoinStage,
     SplitLongAudioStage,
 )
 from nemo_curator.tasks import AudioTask
 
 
+def _patch_audio_io(monkeypatch: pytest.MonkeyPatch, saved_paths: list[str]) -> None:
+    def fake_load(_path: str) -> tuple[np.ndarray, int]:
+        return np.linspace(-0.5, 0.5, 80, dtype=np.float32)[None, :], 10
+
+    def fake_save(path: str, waveform: np.ndarray, sample_rate: int) -> None:
+        saved_paths.append(path)
+        sf.write(path, waveform.T, sample_rate)
+
+    monkeypatch.setattr("nemo_curator.stages.audio.tagging.split.torchaudio.load", fake_load)
+    monkeypatch.setattr("nemo_curator.stages.audio.tagging.split.torchaudio.save", fake_save)
+
+
+def test_additive_fields_preserve_legacy_positional_arguments() -> None:
+    splitter = SplitLongAudioStage(120.0, 2.0, "custom-split")
+    joiner = JoinSplitAudioMetadataStage("transcript", "custom-join")
+    composite = SplitASRAlignJoinStage(120.0, 2.0, "legacy/model")
+
+    assert splitter.name == "custom-split"
+    assert splitter.duration_key == "duration"
+    assert splitter.output_dir is None
+    assert joiner.text_key == "transcript"
+    assert joiner.name == "custom-join"
+    assert joiner.split_filepaths_key == "split_filepaths"
+    assert composite.model_name == "legacy/model"
+    assert composite.output_dir is None
+
+
 class TestSplitLongAudioStageGetSplitPoints:
     """Tests for SplitLongAudioStage.get_split_points."""
 
@@ -105,22 +142,325 @@ def test_long_audio_round_trip_with_torchaudio(
         assert result.data["split_offsets"] == [0.0, 1.0, 2.0]
         assert all(sf.info(path).frames == sample_rate for path in result.data["split_filepaths"])
 
+    def test_default_output_paths_remain_source_adjacent(
+        self,
+        tmp_path: Path,
+        monkeypatch: pytest.MonkeyPatch,
+        audio_task: Callable[..., AudioTask],
+    ) -> None:
+        """The default keeps the exact sibling path format used before output_dir existed."""
+        saved_paths: list[str] = []
+        _patch_audio_io(monkeypatch, saved_paths)
+        source_path = tmp_path / "recording.flac"
+        stage = SplitLongAudioStage(suggested_max_len=5.0, min_len=0.5)
+        task = audio_task(
+            duration=8.0,
+            audio_item_id="sample",
+            resampled_audio_filepath=str(source_path),
+            segments=[{"start": 0.0, "end": 4.0}, {"start": 4.0, "end": 8.0}],
+        )
+
+        result = stage.process(task)
+
+        expected_paths = [
+            str(tmp_path / "recording.1_of_2.wav"),
+            str(tmp_path / "recording.2_of_2.wav"),
+        ]
+        assert saved_paths == expected_paths
+        assert result.data["split_filepaths"] == expected_paths
+        assert [entry["resampled_audio_filepath"] for entry in result.data["split_metadata"]] == expected_paths
+
+    def test_output_dir_redirects_written_and_returned_split_paths(
+        self,
+        tmp_path: Path,
+        monkeypatch: pytest.MonkeyPatch,
+        audio_task: Callable[..., AudioTask],
+    ) -> None:
+        """An explicit output_dir contains every written and returned chunk."""
+        saved_paths: list[str] = []
+        _patch_audio_io(monkeypatch, saved_paths)
+        source_dir = tmp_path / "input"
+        source_dir.mkdir()
+        source_path = source_dir / "recording.flac"
+        output_dir = tmp_path / "smoke-chunks"
+        stage = SplitLongAudioStage(
+            suggested_max_len=5.0,
+            min_len=0.5,
+            output_dir=str(output_dir),
+        )
+        task = audio_task(
+            duration=8.0,
+            audio_item_id="sample",
+            resampled_audio_filepath=str(source_path),
+            segments=[{"start": 0.0, "end": 4.0}, {"start": 4.0, "end": 8.0}],
+        )
+
+        result = stage.process(task)
+
+        stem = stage._shared_output_stem("recording", str(source_path), [4.0], 10, 80)
+        expected_paths = [
+            str(output_dir / f"{stem}.1_of_2.wav"),
+            str(output_dir / f"{stem}.2_of_2.wav"),
+        ]
+        assert output_dir.is_dir()
+        assert saved_paths == expected_paths
+        assert result.data["split_filepaths"] == expected_paths
+        assert [entry["resampled_audio_filepath"] for entry in result.data["split_metadata"]] == expected_paths
+        assert not list(source_dir.glob("recording.*_of_2.wav"))
+
+    def test_two_recordings_sharing_a_basename_do_not_overwrite_each_other(
+        self,
+        tmp_path: Path,
+        monkeypatch: pytest.MonkeyPatch,
+        audio_task: Callable[..., AudioTask],
+    ) -> None:
+        saved_paths: list[str] = []
+        _patch_audio_io(monkeypatch, saved_paths)
+        output_dir = tmp_path / "chunks"
+
+        for speaker in ("spk1", "spk2"):
+            source_dir = tmp_path / speaker
+            source_dir.mkdir()
+            stage = SplitLongAudioStage(suggested_max_len=5.0, min_len=0.5, output_dir=str(output_dir))
+            stage.process(
+                audio_task(
+                    duration=8.0,
+                    audio_item_id="utt1",
+                    resampled_audio_filepath=str(source_dir / "utt1.wav"),
+                    segments=[{"start": 0.0, "end": 4.0}, {"start": 4.0, "end": 8.0}],
+                )
+            )
+
+        assert len(saved_paths) == len(set(saved_paths)), f"one speaker overwrote the other: {saved_paths}"
+
+    def test_same_source_with_different_split_plans_gets_distinct_files(
+        self,
+        tmp_path: Path,
+        monkeypatch: pytest.MonkeyPatch,
+        audio_task: Callable[..., AudioTask],
+    ) -> None:
+        saved_paths: list[str] = []
+        _patch_audio_io(monkeypatch, saved_paths)
+        source_path = tmp_path / "same.wav"
+        output_dir = tmp_path / "chunks"
+        plans = [
+            [{"start": 0.0, "end": 4.0}, {"start": 4.0, "end": 8.0}],
+            [{"start": 0.0, "end": 3.0}, {"start": 3.0, "end": 8.0}],
+        ]
+        emitted: list[list[str]] = []
+
+        for segments in plans:
+            result = SplitLongAudioStage(
+                suggested_max_len=5.0,
+                min_len=0.5,
+                output_dir=str(output_dir),
+            ).process(
+                audio_task(
+                    duration=8.0,
+                    audio_item_id="same",
+                    resampled_audio_filepath=str(source_path),
+                    segments=segments,
+                )
+            )
+            emitted.append(result.data["split_filepaths"])
+
+        assert set(emitted[0]).isdisjoint(emitted[1])
+        assert len(saved_paths) == len(set(saved_paths)) == 4
+        assert sorted(sf.info(path).frames for path in saved_paths) == [30, 40, 40, 50]
+
+    def test_known_short_source_hash_collision_gets_distinct_shared_paths(
+        self,
+        tmp_path: Path,
+        monkeypatch: pytest.MonkeyPatch,
+        audio_task: Callable[..., AudioTask],
+    ) -> None:
+        first = "/dataset/spk25433/utt.wav"
+        second = "/dataset/spk158142/utt.wav"
+        assert hashlib.sha256(first.encode()).hexdigest()[:8] == hashlib.sha256(second.encode()).hexdigest()[:8]
+        saved_paths: list[str] = []
+        _patch_audio_io(monkeypatch, saved_paths)
+
+        for source in (first, second):
+            SplitLongAudioStage(
+                suggested_max_len=5.0,
+                min_len=0.5,
+                output_dir=str(tmp_path / "chunks"),
+            ).process(
+                audio_task(
+                    duration=8.0,
+                    audio_item_id="utt",
+                    resampled_audio_filepath=source,
+                    segments=[{"start": 0.0, "end": 4.0}, {"start": 4.0, "end": 8.0}],
+                )
+            )
+
+        assert len(saved_paths) == len(set(saved_paths)) == 4
+
+    def test_remote_output_paths_are_written_through_fsspec(
+        self,
+        tmp_path: Path,
+        monkeypatch: pytest.MonkeyPatch,
+        audio_task: Callable[..., AudioTask],
+    ) -> None:
+        local_writes: list[str] = []
+        _patch_audio_io(monkeypatch, local_writes)
+        output_dir = f"memory://split-{tmp_path.name}"
+        result = SplitLongAudioStage(
+            suggested_max_len=5.0,
+            min_len=0.5,
+            output_dir=output_dir,
+        ).process(
+            audio_task(
+                duration=8.0,
+                audio_item_id="remote",
+                resampled_audio_filepath=str(tmp_path / "source.wav"),
+                segments=[{"start": 0.0, "end": 4.0}, {"start": 4.0, "end": 8.0}],
+            )
+        )
+
+        assert all(path.startswith(f"{output_dir}/") for path in result.data["split_filepaths"])
+        for advertised in result.data["split_filepaths"]:
+            fs, path = url_to_fs(advertised)
+            assert fs.exists(path)
+        assert local_writes
+        assert all(not Path(path).exists() for path in local_writes)
+
+    def test_remote_output_cleans_local_temp_when_save_fails(
+        self,
+        tmp_path: Path,
+        monkeypatch: pytest.MonkeyPatch,
+        audio_task: Callable[..., AudioTask],
+    ) -> None:
+        local_writes: list[str] = []
+        _patch_audio_io(monkeypatch, local_writes)
+
+        def fail_save(path: str, waveform: np.ndarray, sample_rate: int) -> None:
+            local_writes.append(path)
+            sf.write(path, waveform.T, sample_rate)
+            message = "local encoding failed"
+            raise OSError(message)
+
+        monkeypatch.setattr("nemo_curator.stages.audio.tagging.split.torchaudio.save", fail_save)
+        stage = SplitLongAudioStage(
+            suggested_max_len=5.0,
+            min_len=0.5,
+            output_dir=f"memory://split-failure-{tmp_path.name}",
+        )
+
+        with pytest.raises(OSError, match="local encoding failed"):
+            stage.process(
+                audio_task(
+                    duration=8.0,
+                    audio_item_id="remote",
+                    resampled_audio_filepath=str(tmp_path / "source.wav"),
+                    segments=[{"start": 0.0, "end": 4.0}, {"start": 4.0, "end": 8.0}],
+                )
+            )
+
+        assert len(local_writes) == 1
+        assert not Path(local_writes[0]).exists()
+
+    def test_static_contract_exposes_conservative_split_gates(self, tmp_path: Path) -> None:
+        static = static_contract(SplitLongAudioStage)
+        configured_default = build_contract(SplitLongAudioStage())
+        configured_shared = build_contract(SplitLongAudioStage(output_dir=str(tmp_path)))
+
+        assert static.gates.writes_to_disk is True
+        assert static.gates.output_path_params == ["output_dir"]
+        assert static.gates.per_row_independent is False
+        assert configured_default.gates.per_row_independent is True
+        assert configured_shared.gates.per_row_independent is False
+
+
+def test_split_asr_align_join_forwards_output_dir(tmp_path: Path) -> None:
+    """Composite construction forwards both redirected and legacy defaults."""
+    output_dir = str(tmp_path / "smoke-chunks")
+    redirected_splitter = SplitASRAlignJoinStage(output_dir=output_dir).decompose()[0]
+    default_splitter = SplitASRAlignJoinStage().decompose()[0]
+
+    assert isinstance(redirected_splitter, SplitLongAudioStage)
+    assert redirected_splitter.output_dir == output_dir
+    assert isinstance(default_splitter, SplitLongAudioStage)
+    assert default_splitter.output_dir is None
+
 
 class TestJoinSplitAudioMetadataStage:
     """Tests for JoinSplitAudioMetadataStage."""
 
+    def test_contract_guarantees_outputs_and_removes_temporary_keys(self) -> None:
+        stage = JoinSplitAudioMetadataStage(
+            text_key="transcript",
+            alignment_key="word_alignment",
+            split_filepaths_key="chunk_paths",
+            split_metadata_key="chunks",
+        )
+
+        contract = build_contract(stage)
+
+        assert contract.writes.data_keys == ["transcript", "word_alignment"]
+        assert contract.removes_keys == ["chunk_paths", "chunks"]
+
     def test_no_split_passthrough(self, audio_task: Callable[..., AudioTask]) -> None:
-        """Entry with split_filepaths=None (no split occurred) returns entry without key."""
+        """No-split preserves existing outputs while dropping both temporary keys."""
         stage = JoinSplitAudioMetadataStage()
+        original_alignment = [{"word": "hello", "start": 0.0, "end": 0.5}]
         task = audio_task(
             audio_item_id="x",
             split_filepaths=None,
+            split_metadata=[{"text": "must not replace the top-level value"}],
+            split_offsets=[1.25],
+            split_timestamps=[2.5],
             text="hello",
+            alignment=original_alignment,
         )
-        result = stage.process(task)
-        out = result.data
+
+        assert_agent_ready(
+            stage,
+            lambda: task,
+            available_keys={
+                "split_filepaths",
+                "split_metadata",
+                "split_offsets",
+                "split_timestamps",
+            },
+        )
+
+        out = task.data
         assert "split_filepaths" not in out
+        assert "split_metadata" not in out
         assert out["text"] == "hello"
+        assert out["alignment"] is original_alignment
+        assert out["split_offsets"] == [1.25]
+        assert out["split_timestamps"] == [2.5]
+
+    def test_empty_split_supplies_safe_defaults(self, audio_task: Callable[..., AudioTask]) -> None:
+        """Empty split metadata supplies outputs without removing unrelated split timing."""
+        stage = JoinSplitAudioMetadataStage()
+        task = audio_task(
+            audio_item_id="empty",
+            split_filepaths=[],
+            split_metadata=[],
+            split_offsets=[],
+            split_timestamps=[],
+        )
+
+        assert_agent_ready(
+            stage,
+            lambda: task,
+            available_keys={
+                "split_filepaths",
+                "split_metadata",
+                "split_offsets",
+                "split_timestamps",
+            },
+        )
+
+        assert task.data["text"] == ""
+        assert task.data["alignment"] == []
+        assert "split_filepaths" not in task.data
+        assert "split_metadata" not in task.data
+        assert task.data["split_offsets"] == []
+        assert task.data["split_timestamps"] == []
 
     def test_join_split_metadata_concatenates_text_and_alignments(self, audio_task: Callable[..., AudioTask]) -> None:
         """Meta-entry with split_metadata joins text and adjusts alignment timestamps."""
@@ -145,9 +485,21 @@ def test_join_split_metadata_concatenates_text_and_alignments(self, audio_task:
                 },
             ],
             split_offsets=[0.0, 5.0],
+            split_timestamps=[5.0],
         )
-        result = stage.process(task)
-        out = result.data
+
+        assert_agent_ready(
+            stage,
+            lambda: task,
+            available_keys={
+                "split_filepaths",
+                "split_metadata",
+                "split_offsets",
+                "split_timestamps",
+            },
+        )
+
+        out = task.data
         assert out["text"] == "first part second part"
         assert "split_filepaths" not in out
         assert "split_metadata" not in out
@@ -159,3 +511,46 @@ def test_join_split_metadata_concatenates_text_and_alignments(self, audio_task:
         assert align[2]["word"] == "second"
         assert align[2]["start"] == 5.0
         assert align[2]["end"] == 5.5
+
+    @pytest.mark.parametrize(
+        ("removed_key", "consumer"),
+        [
+            (
+                "split_filepaths",
+                MergeAlignmentDiarizationStage(alignment_key="split_filepaths"),
+            ),
+            (
+                "split_metadata",
+                MergeAlignmentDiarizationStage(segments_key="split_metadata"),
+            ),
+        ],
+    )
+    def test_planner_does_not_carry_removed_temporary_key(
+        self,
+        removed_key: str,
+        consumer: MergeAlignmentDiarizationStage,
+    ) -> None:
+        report = validate_pipeline(
+            [JoinSplitAudioMetadataStage(), consumer],
+            initial_roles={"alignment", "segments", "text"},
+            initial_keys={
+                "alignment",
+                "segments",
+                "split_filepaths",
+                "split_metadata",
+                "split_offsets",
+                "split_timestamps",
+                "text",
+            },
+            initial_task_type="AudioTask",
+        )
+
+        assert report.ok
+        assert not report.keys_ok
+        assert removed_key not in report.produced_keys
+        assert any(
+            issue.stage_index == 1
+            and issue.code == "dangling_key"
+            and removed_key in issue.message
+            for issue in report.issues
+        )
diff --git a/tests/stages/audio/tagging/text/test_itn.py b/tests/stages/audio/tagging/text/test_itn.py
index ea4b7285cd..9bb7247ff1 100644
--- a/tests/stages/audio/tagging/text/test_itn.py
+++ b/tests/stages/audio/tagging/text/test_itn.py
@@ -13,6 +13,11 @@
 # limitations under the License.
 
 from collections.abc import Callable
+from unittest.mock import MagicMock
+
+import pytest
+from nemo_curator.stages.audio._agent._conformance import assert_agent_ready
+from nemo_curator.stages.audio._agent._planning import validate_pipeline
 
 from nemo_curator.stages.audio.tagging.text.itn import InverseTextNormalizationStage
 from nemo_curator.tasks import AudioTask
@@ -36,3 +41,50 @@ def test_process(self, audio_task: Callable[..., AudioTask]) -> None:
         assert len(out["segments"]) == 2
         assert out["segments"][0]["text_ITN"] == "hello"
         assert out["segments"][1]["text_ITN"] == "the answer is 42"
+
+    @pytest.mark.parametrize(
+        ("data", "expected"),
+        [
+            ({"segments": [{}]}, None),
+            ({"segments": [{"text": ""}]}, None),
+            ({"segments": [{"text": "forty two"}]}, "42"),
+        ],
+        ids=["missing-text", "empty-text", "populated"],
+    )
+    def test_agent_ready_conditional_nested_output(
+        self,
+        data: dict,
+        expected: str | None,
+    ) -> None:
+        stage = InverseTextNormalizationStage()
+        normalizer = MagicMock()
+        normalizer.split_text_into_sentences.side_effect = lambda text: [text]
+        normalizer.normalize_list.return_value = ["42"]
+        stage._normalizer = normalizer
+        task = AudioTask(dataset_name="test", data=data)
+
+        contract = assert_agent_ready(
+            stage,
+            lambda: task,
+            segments_key="segments",
+        )
+
+        assert contract.reads.data_keys == ["segments"]
+        assert contract.writes.segment_data_keys == []
+        assert len(contract.conditional_writes) == 1
+        assert contract.conditional_writes[0].writes.segment_data_keys == ["text_ITN"]
+        if expected is None:
+            assert all("text_ITN" not in segment for segment in task.data.get("segments", []))
+        else:
+            assert task.data["segments"][0]["text_ITN"] == expected
+
+    def test_planner_does_not_guarantee_conditional_nested_output(self) -> None:
+        report = validate_pipeline(
+            [InverseTextNormalizationStage()],
+            initial_roles={"segments"},
+            initial_keys={"segments"},
+            initial_task_type="AudioTask",
+        )
+
+        assert report.ok
+        assert "text_ITN" not in report.produced_keys
diff --git a/tests/stages/audio/tagging/text/test_text.py b/tests/stages/audio/tagging/text/test_text.py
index 8d807199b7..b863a9ab28 100644
--- a/tests/stages/audio/tagging/text/test_text.py
+++ b/tests/stages/audio/tagging/text/test_text.py
@@ -13,6 +13,11 @@
 # limitations under the License.
 
 from collections.abc import Callable
+from unittest.mock import MagicMock
+
+import pytest
+from nemo_curator.stages.audio._agent._conformance import assert_agent_ready
+from nemo_curator.stages.audio._agent._planning import validate_pipeline
 
 from nemo_curator.stages.audio.tagging.text.chinese_conversion import ChineseConversionStage
 from nemo_curator.tasks import AudioTask
@@ -43,3 +48,51 @@ def test_segment_without_text_key_is_skipped(self, audio_task: Callable[..., Aud
         result = stage.process(task)
         out = result.data
         assert "text_simplified" not in out["segments"][0]
+
+    @pytest.mark.parametrize(
+        ("data", "expected"),
+        [
+            ({"segments": [{}]}, None),
+            ({"segments": []}, None),
+            ({"segments": [{"text": "漢字"}]}, "汉字"),
+        ],
+        ids=["missing-text", "empty-segments", "populated"],
+    )
+    def test_agent_ready_conditional_nested_output(
+        self,
+        data: dict,
+        expected: str | None,
+    ) -> None:
+        stage = ChineseConversionStage()
+        converter = MagicMock()
+        converter.convert.return_value = "汉字"
+        stage._converter = converter
+        task = AudioTask(dataset_name="test", data=data)
+
+        contract = assert_agent_ready(
+            stage,
+            lambda: task,
+            segments_key="segments",
+        )
+
+        assert contract.reads.data_keys == ["segments"]
+        assert contract.writes.segment_data_keys == []
+        assert len(contract.conditional_writes) == 1
+        assert contract.conditional_writes[0].writes.segment_data_keys == ["text_simplified"]
+        if expected is None:
+            assert all(
+                "text_simplified" not in segment for segment in task.data.get("segments", [])
+            )
+        else:
+            assert task.data["segments"][0]["text_simplified"] == expected
+
+    def test_planner_does_not_guarantee_conditional_nested_output(self) -> None:
+        report = validate_pipeline(
+            [ChineseConversionStage()],
+            initial_roles={"segments"},
+            initial_keys={"segments"},
+            initial_task_type="AudioTask",
+        )
+
+        assert report.ok
+        assert "text_simplified" not in report.produced_keys