feat(audio-advanced): agent-ready AudioDataFilter composite pipeline

[shubhamNvidia]

Summary

Makes the composite AudioDataFilterStage agent-ready. Additive only.

• advanced_pipelines/audio_data_filter/audio_data_filter.py — the composite (which decomposes into the underlying filter stages) now subclasses AgentReady and exposes describe() / StageContract. Decomposition and runtime behavior unchanged.

Notes for reviewers

• Additive, backward compatible.

Test plan

• pytest tests/stages/audio/advanced_pipelines -q

[copy-pr-bot]

This pull request requires additional validation before any workflows can run on NVIDIA's runners.

Pull request vetters can view their responsibilities here.

Contributors can view more details about this message here.

[greptile-apps]

Greptile Summary

This PR makes AudioDataFilterStage and five common stages (GetAudioDurationStage, PreserveByValueStage, ManifestReaderStage, ManifestReader, ManifestWriterStage) agent-ready by adding the AgentReady mixin and describe() implementations, and introduces two new supporting modules (_agent_ready.py and _residency.py) that define the contract dataclasses and audio-residency utilities.

• _agent_ready.py defines the StageContract, IOSpec, Gates, StaticHints, and AgentReady mixin; all defaults are safe and the JSON-serialization helpers are robust.
• _residency.py provides resolve_audio, resolve_audio_path, and related helpers for handling file-vs-waveform residency in stages.
• common.py stages each receive accurate describe() implementations; audio_data_filter.py adds describe() but the returned contract uses default cardinality=\"1:1\" and empty Gates, which does not reflect the actual fan-out topology or GPU requirements of the composite pipeline.

Confidence Score: 4/5

Safe to merge with the understanding that AudioDataFilterStage's describe() misleads agents about its cardinality and resource requirements.

The new AudioDataFilterStage.describe() returns a contract with default cardinality='1:1' and empty Gates, but the pipeline fans out through VAD and SpeakerSeparation in the default config and is GPU-bound throughout. An agent that consults this contract for scheduling or resource allocation will work from incorrect information. The rest of the contracts added in common.py are accurate, and the two new modules are well-designed.

audio_data_filter.py — the describe() contract needs cardinality and gates corrected to match the actual pipeline behaviour.

Important Files Changed

Filename	Overview
nemo_curator/stages/audio/_agent_ready.py	New file introducing AgentReady mixin, StageContract, IOSpec, Gates, and helpers; well-structured, all defaults are safe, JSON serialization is robust.
nemo_curator/stages/audio/_residency.py	New residency helpers; temp-file leak on sf.write failure and edge-case channels>=samples fallthrough were flagged in existing threads and remain unaddressed in this diff.
nemo_curator/stages/audio/advanced_pipelines/audio_data_filter/audio_data_filter.py	AgentReady added; describe() returns a sparse contract with incorrect default cardinality ('1:1') and empty gates, misrepresenting the fan-out pipeline and GPU requirements to agents.
nemo_curator/stages/audio/common.py	Five stages upgraded to AgentReady with describe() implementations; PreserveByValueStage.describe() still omits batch_only=True (existing thread); other contracts are accurate.

Class Diagram

%%{init: {'theme': 'neutral'}}%%
classDiagram
    class AgentReady {
        +AGENT_STATIC: ClassVar[StaticHints | None]
        +BATCH_ONLY: ClassVar[bool]
        +KEY_ROLE_OVERRIDES: ClassVar[Mapping]
        +describe() StageContract
        +describe_static() StageContract
    }
    class StageContract {
        +reads: IOSpec
        +writes: IOSpec
        +cardinality: Cardinality
        +gates: Gates
        +wrappable: bool
        +batch_only: bool
        +to_dict() dict
    }
    class IOSpec {
        +data_keys: list[str]
        +accepts: list[AudioForm]
        +produces: list[ProducedForm]
    }
    class Gates {
        +writes_to_disk: bool
        +requires_gpu: bool
        +lifecycle_side_effects: bool
        +requires_serializable_input: bool
    }
    class AudioDataFilterStage {
        +describe() StageContract
        +decompose() list
    }
    class GetAudioDurationStage {
        +describe() StageContract
    }
    class PreserveByValueStage {
        +BATCH_ONLY = True
        +describe() StageContract
    }
    class ManifestReaderStage {
        +describe() StageContract
    }
    class ManifestReader {
        +describe() StageContract
    }
    class ManifestWriterStage {
        +describe() StageContract
    }
    AgentReady <|-- AudioDataFilterStage
    AgentReady <|-- GetAudioDurationStage
    AgentReady <|-- PreserveByValueStage
    AgentReady <|-- ManifestReaderStage
    AgentReady <|-- ManifestReader
    AgentReady <|-- ManifestWriterStage
    StageContract *-- IOSpec
    StageContract *-- Gates
    AgentReady ..> StageContract : describe()

Loading

%%{init: {'theme': 'base', 'themeVariables': {"darkMode": true, "background": "#0d1117", "primaryColor": "#21262d", "primaryTextColor": "#e6edf3", "primaryBorderColor": "#8b949e", "lineColor": "#8b949e", "textColor": "#e6edf3", "edgeLabelBackground": "#161b22", "actorBkg": "#21262d", "actorBorder": "#8b949e", "actorTextColor": "#e6edf3", "actorLineColor": "#8b949e", "signalColor": "#8b949e", "signalTextColor": "#e6edf3", "noteBkgColor": "#373320", "noteBorderColor": "#d4a72c", "noteTextColor": "#f0e6c0", "labelBoxBkgColor": "#21262d", "labelBoxBorderColor": "#8b949e", "labelTextColor": "#e6edf3", "loopTextColor": "#e6edf3", "activationBkgColor": "#30363d", "activationBorderColor": "#8b949e"}}}%%
classDiagram
    class AgentReady {
        +AGENT_STATIC: ClassVar[StaticHints | None]
        +BATCH_ONLY: ClassVar[bool]
        +KEY_ROLE_OVERRIDES: ClassVar[Mapping]
        +describe() StageContract
        +describe_static() StageContract
    }
    class StageContract {
        +reads: IOSpec
        +writes: IOSpec
        +cardinality: Cardinality
        +gates: Gates
        +wrappable: bool
        +batch_only: bool
        +to_dict() dict
    }
    class IOSpec {
        +data_keys: list[str]
        +accepts: list[AudioForm]
        +produces: list[ProducedForm]
    }
    class Gates {
        +writes_to_disk: bool
        +requires_gpu: bool
        +lifecycle_side_effects: bool
        +requires_serializable_input: bool
    }
    class AudioDataFilterStage {
        +describe() StageContract
        +decompose() list
    }
    class GetAudioDurationStage {
        +describe() StageContract
    }
    class PreserveByValueStage {
        +BATCH_ONLY = True
        +describe() StageContract
    }
    class ManifestReaderStage {
        +describe() StageContract
    }
    class ManifestReader {
        +describe() StageContract
    }
    class ManifestWriterStage {
        +describe() StageContract
    }
    AgentReady <|-- AudioDataFilterStage
    AgentReady <|-- GetAudioDurationStage
    AgentReady <|-- PreserveByValueStage
    AgentReady <|-- ManifestReaderStage
    AgentReady <|-- ManifestReader
    AgentReady <|-- ManifestWriterStage
    StageContract *-- IOSpec
    StageContract *-- Gates
    AgentReady ..> StageContract : describe()

Loading

_{Reviews (2): Last reviewed commit: "fix(audio): match _residency.py to found..." | Re-trigger Greptile}

[greptile-apps]

PreserveByValueStage.describe() returns StageContract with batch_only at its default of False, yet BATCH_ONLY = True is set on the class and process() immediately raises NotImplementedError. Any agent or tool that calls stage.describe() directly (instead of going through the registry's build_contract) will receive a contract that incorrectly advertises process() as callable, which would raise at runtime. The fix is to propagate the class flag into the returned contract.

Suggested change

	def describe(self) -> StageContract:
	return StageContract(
	reads=IOSpec(data_keys=[self.input_value_key]),
	writes=IOSpec(data_keys=[self.input_value_key]),
	cardinality="filter",
	)
	def describe(self) -> StageContract:
	return StageContract(
	reads=IOSpec(data_keys=[self.input_value_key]),
	writes=IOSpec(data_keys=[self.input_value_key]),
	cardinality="filter",
	batch_only=True,
	)

[greptile-apps]

If sf.write() raises (e.g. the waveform dtype is unsupported, or the array shape is wrong after _as_soundfile_array), the temp file created by mkstemp is left on disk but never appended to register_temp. The caller therefore never learns about it and cleanup_temp_files will not remove it. Wrapping the write in a try/except and cleaning up on failure closes the leak.

Suggested change

	fd, tmp = tempfile.mkstemp(suffix=".wav", dir=temp_dir)
	os.close(fd)
	sf.write(tmp, _as_soundfile_array(waveform), int(sample_rate))
	if register_temp is not None:
	register_temp.append(tmp)
	return tmp
	fd, tmp = tempfile.mkstemp(suffix=".wav", dir=temp_dir)
	os.close(fd)
	try:
	sf.write(tmp, _as_soundfile_array(waveform), int(sample_rate))
	except Exception:
	with contextlib.suppress(OSError):
	os.remove(tmp)
	raise
	if register_temp is not None:
	register_temp.append(tmp)
	return tmp

[greptile-apps]

_as_soundfile_array falls through to return waveform when channels >= samples (i.e. a 2-D array whose first dimension is not smaller than the second). SoundFile expects audio in (samples, channels) shape for multi-channel data, so returning the un-transposed (channels, samples) array in that branch would write audio with channels and samples swapped, producing a corrupt WAV. In practice audio files have far more samples than channels, but a very short clip (e.g. a 1-sample test fixture) would silently trigger this.