@@ -14,11 +14,13 @@
 
 import os
 import shutil
-from dataclasses import dataclass
+from dataclasses import dataclass, field
+from typing import ClassVar
 
 from huggingface_hub import hf_hub_download
 from loguru import logger
 
+from nemo_curator.stages.audio._agent._agent_ready import AgentReady, Gates, IOSpec, StageContract, StaticHints
 from nemo_curator.stages.audio.datasets.file_utils import extract_archive
 from nemo_curator.stages.base import ProcessingStage
 from nemo_curator.tasks import AudioTask, EmptyTask
@@ -40,7 +42,7 @@ def get_fleurs_filenames(lang: str, split: str) -> tuple[str, str]:
 
 
 @dataclass
-class CreateInitialManifestFleursStage(ProcessingStage[EmptyTask, AudioTask]):
+class CreateInitialManifestFleursStage(AgentReady, ProcessingStage[EmptyTask, AudioTask]):
     """Create initial manifest for the FLEURS dataset.
 
     Dataset link: https://huggingface.co/datasets/google/fleurs
@@ -70,10 +72,19 @@ class CreateInitialManifestFleursStage(ProcessingStage[EmptyTask, AudioTask]):
             ``benchmarking/data_prep/prepare_fleurs_data.py``).
     """
 
+    AGENT_STATIC: ClassVar[StaticHints] = StaticHints(
+        gates=Gates(
+            writes_to_disk=True,
+            output_path_params=["raw_data_dir", "cache_dir"],
+            requires_internet_first_run=True,
+            per_row_independent=True,
+        )
+    )
+
     name: str = "CreateInitialManifestFleurs"
-    lang: str = ""
-    split: str = ""
-    raw_data_dir: str = ""
+    lang: str = field(default="", metadata={"agent_required": True})
+    split: str = field(default="", metadata={"agent_required": True})
+    raw_data_dir: str = field(default="", metadata={"agent_required": True})
     filepath_key: str = "audio_filepath"
     text_key: str = "text"
     batch_size: int = 1
@@ -85,13 +96,42 @@ def __post_init__(self) -> None:
             if not getattr(self, attr):
                 msg = f"{attr} is required for CreateInitialManifestFleursStage"
                 raise ValueError(msg)
+        output_key_fields = {
+            "filepath_key": self.filepath_key,
+            "text_key": self.text_key,
+        }
+        for field_name, key in output_key_fields.items():
+            if not isinstance(key, str) or not key.strip():
+                msg = f"{field_name} must be a non-empty string"
+                raise ValueError(msg)
+        if self.filepath_key == self.text_key:
+            msg = f"filepath_key={self.filepath_key!r} conflicts with text_key"
+            raise ValueError(msg)
 
     def inputs(self) -> tuple[list[str], list[str]]:
         return [], []
 
     def outputs(self) -> tuple[list[str], list[str]]:
         return [], [self.filepath_key, self.text_key]
 
+    def describe(self) -> StageContract:
+        return StageContract(
+            writes=IOSpec(
+                data_keys=[self.filepath_key, self.text_key],
+                produces=["disk"] if self.auto_download else [],
+            ),
+            cardinality="1:N fan-out",
+            # The download stages the split once; after that every valid transcript line
+            # emits its own path and text in input order. Malformed lines with fewer than
+            # three tab-separated columns are skipped.
+            gates=Gates(
+                writes_to_disk=self.auto_download,
+                output_path_params=["raw_data_dir", "cache_dir"] if self.auto_download else [],
+                requires_internet_first_run=self.auto_download,
+                per_row_independent=True,
+            ),
+        )
+
     def language_data_dir(self) -> str:
         """Return the per-language download directory under ``raw_data_dir``.
 
 
@@ -16,9 +16,11 @@
 import os
 import subprocess
 from dataclasses import dataclass
+from typing import ClassVar
 
 from loguru import logger
 
+from nemo_curator.stages.audio._agent._agent_ready import AgentReady, Gates, IOSpec, StageContract, StaticHints
 from nemo_curator.stages.audio.datasets.file_utils import download_file
 from nemo_curator.stages.base import ProcessingStage
 from nemo_curator.tasks import AudioTask, EmptyTask
@@ -33,7 +35,7 @@
 
 
 @dataclass
-class CreateInitialManifestReadSpeechStage(ProcessingStage[EmptyTask, AudioTask]):
+class CreateInitialManifestReadSpeechStage(AgentReady, ProcessingStage[EmptyTask, AudioTask]):
     """
     Stage to create initial manifest for the DNS Challenge Read Speech dataset.
 
@@ -47,24 +49,90 @@ class CreateInitialManifestReadSpeechStage(ProcessingStage[EmptyTask, AudioTask]
         raw_data_dir: Directory where data will be downloaded/extracted to.
         max_samples: Maximum number of samples to include (-1 for all).
         auto_download: If True, automatically download and extract dataset.
+        filepath_key: Key name used for each emitted audio path.
+        text_key: Key name used for each emitted transcript.
+        sample_rate_key: Key name used for the constant 48 kHz sample rate.
+        book_id_key: Key name used for the parsed book identifier.
+        reader_id_key: Key name used for the parsed reader identifier.
     """
 
+    AGENT_STATIC: ClassVar[StaticHints] = StaticHints(
+        gates=Gates(
+            writes_to_disk=True,
+            output_path_params=["raw_data_dir"],
+            requires_internet_first_run=True,
+            per_row_independent=False,
+        )
+    )
+    INTERNAL_KEY_FIELDS: ClassVar[frozenset[str]] = frozenset({"book_id_key", "reader_id_key"})
+
     raw_data_dir: str
     max_samples: int = 5000
     auto_download: bool = True
     filepath_key: str = "audio_filepath"
     text_key: str = "text"
     name: str = "CreateInitialManifestReadSpeech"
     batch_size: int = 1
+    sample_rate_key: str = "sample_rate"
+    book_id_key: str = "book_id"
+    reader_id_key: str = "reader_id"
 
     def __post_init__(self):
         super().__init__()
+        output_key_fields = {
+            "filepath_key": self.filepath_key,
+            "text_key": self.text_key,
+            "sample_rate_key": self.sample_rate_key,
+            "book_id_key": self.book_id_key,
+            "reader_id_key": self.reader_id_key,
+        }
+        for field_name, key in output_key_fields.items():
+            if not isinstance(key, str) or not key.strip():
+                msg = f"{field_name} must be a non-empty string"
+                raise ValueError(msg)
+            conflicting_fields = sorted(
+                other_name
+                for other_name, other_key in output_key_fields.items()
+                if other_name != field_name and other_key == key
+            )
+            if conflicting_fields:
+                msg = f"{field_name}={key!r} conflicts with {', '.join(conflicting_fields)}"
+                raise ValueError(msg)
 
     def inputs(self) -> tuple[list[str], list[str]]:
         return [], []
 
     def outputs(self) -> tuple[list[str], list[str]]:
-        return [], [self.filepath_key, self.text_key]
+        return [], [
+            self.filepath_key,
+            self.text_key,
+            self.sample_rate_key,
+            self.book_id_key,
+            self.reader_id_key,
+        ]
+
+    def describe(self) -> StageContract:
+        return StageContract(
+            writes=IOSpec(
+                data_keys=[
+                    self.filepath_key,
+                    self.text_key,
+                    self.sample_rate_key,
+                    self.book_id_key,
+                    self.reader_id_key,
+                ],
+                produces=["disk"] if self.auto_download else [],
+            ),
+            cardinality="1:N fan-out",
+            gates=Gates(
+                writes_to_disk=self.auto_download,
+                requires_internet_first_run=self.auto_download,
+                output_path_params=["raw_data_dir"] if self.auto_download else [],
+                # A positive max_samples selects a prefix of the globally sorted corpus, so a
+                # delta containing only changed files can admit rows a full run would exclude.
+                per_row_independent=self.max_samples <= 0,
+            ),
+        )
 
     def download_and_extract(self) -> str:
         """Download and extract DNS Challenge Read Speech dataset (~4.88 GB)."""
@@ -249,9 +317,9 @@ def collect_audio_files(self, search_dir: str) -> list[dict]:
             entry = {
                 self.filepath_key: os.path.abspath(wav_path),
                 self.text_key: "",
-                "sample_rate": SAMPLE_RATE_48KHZ,
-                "book_id": metadata.get("book_id", ""),
-                "reader_id": metadata.get("reader_id", ""),
+                self.sample_rate_key: SAMPLE_RATE_48KHZ,
+                self.book_id_key: metadata.get("book_id", ""),
+                self.reader_id_key: metadata.get("reader_id", ""),
             }
             entries.append(entry)
 
@@ -286,10 +354,10 @@ def verify_dataset_structure(self, entries: list[dict]) -> None:
         unique_readers = set()
         unique_books = set()
         for entry in entries:
-            if entry.get("reader_id"):
-                unique_readers.add(entry["reader_id"])
-            if entry.get("book_id"):
-                unique_books.add(entry["book_id"])
+            if entry.get(self.reader_id_key):
+                unique_readers.add(entry[self.reader_id_key])
+            if entry.get(self.book_id_key):
+                unique_books.add(entry[self.book_id_key])
 
         logger.info(f"Unique readers: {len(unique_readers)}")
         logger.info(f"Unique books: {len(unique_books)}")
 
@@ -20,6 +20,11 @@
 
 import pytest
 
+from nemo_curator.stages.audio._agent._agent_ready import to_json_schema
+from nemo_curator.stages.audio._agent._agent_registry import build_contract, stage_params, static_contract
+from nemo_curator.stages.audio._agent._catalog import role_index
+from nemo_curator.stages.audio._agent._conformance import assert_agent_ready
+
 
 def _import_stage_module() -> tuple[Any, Any]:
     # Inject a stub for optional dependency 'wget' to avoid import errors
@@ -74,6 +79,41 @@ def test_inputs_outputs(tmp_path: Path) -> None:
     assert stage.outputs() == ([], ["audio_filepath", "text"])
 
 
+def test_agent_schema_marks_runtime_validated_params_required() -> None:
+    stage_cls, _ = _import_stage_module()
+    params = stage_params(stage_cls)
+
+    assert to_json_schema(params)["required"] == ["lang", "split", "raw_data_dir"]
+
+    index = role_index()
+    assert "CreateInitialManifestFleursStage" not in index["unresolved_stages"]
+    assert "CreateInitialManifestFleursStage" in index["producers"]["audio_filepath"]
+    assert "CreateInitialManifestFleursStage" in index["producers"]["text"]
+
+
+def test_contract_has_conditional_download_gate_and_static_hints(tmp_path: Path) -> None:
+    stage_cls, _ = _import_stage_module()
+    local = stage_cls(lang="en_us", split="dev", raw_data_dir=str(tmp_path), auto_download=False)
+    downloading = stage_cls(lang="en_us", split="dev", raw_data_dir=str(tmp_path), auto_download=True)
+
+    local_contract = build_contract(local)
+    assert local_contract.writes.produces == []
+    assert local_contract.gates.writes_to_disk is False
+    assert local_contract.gates.requires_internet_first_run is False
+    assert local_contract.gates.output_path_params == []
+
+    download_contract = build_contract(downloading)
+    assert download_contract.writes.produces == ["disk"]
+    assert download_contract.gates.writes_to_disk is True
+    assert download_contract.gates.requires_internet_first_run is True
+    assert download_contract.gates.output_path_params == ["raw_data_dir", "cache_dir"]
+
+    static = static_contract(stage_cls)
+    assert static.gates.writes_to_disk is True
+    assert static.gates.requires_internet_first_run is True
+    assert static.gates.output_path_params == ["raw_data_dir", "cache_dir"]
+
+
 def test_language_data_dir_is_namespaced_per_language(tmp_path: Path) -> None:
     stage_cls, _ = _import_stage_module()
     raw = str(tmp_path / "fleurs")
@@ -208,6 +248,63 @@ def test_process_no_download_reads_prestaged(tmp_path: Path) -> None:
     assert results[1].data["text"] == "world"
 
 
+def test_process_uses_renamed_output_keys_declared_by_contract(tmp_path: Path) -> None:
+    stage_cls, _ = _import_stage_module()
+    raw_dir = _stage_prestaged_layout(tmp_path)
+    stage = stage_cls(
+        lang="hy_am",
+        split="train",
+        raw_data_dir=str(raw_dir),
+        auto_download=False,
+        filepath_key="path",
+        text_key="transcript",
+    )
+
+    from nemo_curator.tasks import EmptyTask
+
+    results = stage.process(EmptyTask(dataset_name="test", data=None))
+
+    assert results
+    assert set(results[0].data) == {"path", "transcript"}
+    assert build_contract(stage).writes.data_keys == ["path", "transcript"]
+
+
+@pytest.mark.parametrize(
+    "kwargs",
+    [
+        {"filepath_key": ""},
+        {"filepath_key": "   "},
+        {"text_key": ""},
+        {"filepath_key": "value", "text_key": "value"},
+    ],
+)
+def test_output_keys_reject_empty_names_and_collisions(tmp_path: Path, kwargs: dict) -> None:
+    stage_cls, _ = _import_stage_module()
+
+    with pytest.raises(ValueError, match=r"must be a non-empty string|conflicts with"):
+        stage_cls(
+            lang="hy_am",
+            split="train",
+            raw_data_dir=str(tmp_path),
+            auto_download=False,
+            **kwargs,
+        )
+
+
+def test_agent_conformance(tmp_path: Path) -> None:
+    stage_cls, _ = _import_stage_module()
+    raw_dir = _stage_prestaged_layout(tmp_path)
+    stage = stage_cls(lang="hy_am", split="train", raw_data_dir=str(raw_dir), auto_download=False)
+    from nemo_curator.tasks import EmptyTask
+
+    assert_agent_ready(
+        stage,
+        lambda: EmptyTask(dataset_name="test", data=None),
+        expected_cardinality="1:N fan-out",
+        available_keys=set(),
+    )
+
+
 def _import_prep_module() -> types.ModuleType:
     """Import benchmarking/data_prep/prepare_fleurs_data.py (not on the default path)."""
     prep_dir = Path(__file__).resolve().parents[4] / "benchmarking" / "data_prep"