[codex] docs: document native pipeline resumability (#2134)

* docs: document native pipeline resumability

Signed-off-by: Lawrence Lane <llane@nvidia.com>

* docs: link SLURM array operations

Signed-off-by: Lawrence Lane <llane@nvidia.com>

* docs: address resumability review feedback

Signed-off-by: Lawrence Lane <llane@nvidia.com>

* docs(pipeline): note checkpoint_path must be shared storage in multi-node runs

Address review feedback: clarify in the Pipeline.run() checkpoint_path
docstring that distributed runs must point every worker/node at the same
durable shared filesystem location, matching the Resumable Processing
reference guidance.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Signed-off-by: Lawrence Lane <llane@nvidia.com>

---------

Signed-off-by: Lawrence Lane <llane@nvidia.com>
Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: lbliii

fern/versions/main/pages/about/concepts/audio/audio-task.mdx

@@ -36,7 +36,6 @@ audio_task = AudioTask(
         "language": "en"
     },
     filepath_key="audio_filepath",
-    task_id="audio_task_001",
     dataset_name="my_speech_dataset"
 )
 ```
@@ -47,7 +46,7 @@ audio_task = AudioTask(
 |-----------|------|-------------|
 | `data` | `dict` | Audio manifest entry (single dict, exposed as `_AttrDict` for attribute-style access) |
 | `filepath_key` | `str \| None` | Key name for audio file paths in data (optional) |
-| `task_id` | `str` | Unique identifier for the task |
+| `task_id` | `str` | Framework-managed deterministic lineage ID. Treat as read-only; it is assigned at stage boundaries. |
 | `dataset_name` | `str` | Name of the source dataset |
 | `num_items` | `int` | Always returns `1` (read-only property) |
 

fern/versions/main/pages/about/release-notes/migration-faq.mdx

@@ -110,15 +110,15 @@ Each task is a data class. You can add whatever statistics you need (input or ou
 ## Fault Tolerance, Checkpointing, and Observability
 
 <Accordion title="What if a stage fails (for example, OOM or disk error)? Will the pipeline crash?">
-Ray will relaunch failed workers or actors, but robust error handling and resumption (for example, from last completed task) should be implemented in your stage logic as needed.
+Ray can relaunch failed workers or actors. For restart recovery, pass `checkpoint_path` to `Pipeline.run()` so completed source partitions are skipped on the next launch. A custom stage can return `FailedTask()` for a recoverable per-task failure that should remain pending. Refer to [Resumable Processing](/reference/infra/resumable-processing).
 </Accordion>
 
 <Accordion title="How is observability handled? Can I track pipeline performance, actor counts, and task durations?">
 All Ray pipelines expose resource and processing metrics via a built-in Grafana dashboard (with process time per task or actor, resource utilization, and so on). You can also summarize stats from task data at pipeline completion for custom reporting. For configuration options, refer to [Pipeline Execution Backends ](/reference/infra/execution-backends) and the [Ray Dashboard documentation](https://docs.ray.io/en/latest/ray-observability/getting-started.html).
 </Accordion>
 
 <Accordion title="Can I resume processing from mid-pipeline if interrupted?">
-For streaming pipelines, each completed task can be tracked (temporarily with final outputs), so resumption logic is "start from task N+1". If you rely on explicit intermediate checkpoints, you can extend the process logic to save and reload state as needed.
+Yes. Run the same pipeline with the same `checkpoint_path`. NeMo Curator records completed source partitions under `.nemo_curator_metadata` and skips them on a later launch. Incomplete sources restart from the source; stage payloads and in-memory state are not serialized. Refer to [Resumable Processing](/reference/infra/resumable-processing) for storage, replay, and multi-host constraints.
 </Accordion>
 
 ---

fern/versions/main/pages/api-reference/executors/experimental.mdx

@@ -82,7 +82,7 @@ class BaseExecutor(ABC):
 
         Args:
             stages: Processing stages to execute.
-            initial_tasks: Initial tasks (defaults to EmptyTask).
+            initial_tasks: Initial tasks (defaults to a fresh `EmptyTask()` instance).
 
         Returns:
             Output tasks from final stage.

fern/versions/main/pages/api-reference/executors/xenna-executor.mdx

@@ -128,7 +128,7 @@ def execute(
 
     Args:
         stages: List of processing stages to execute.
-        initial_tasks: Initial tasks (defaults to EmptyTask).
+        initial_tasks: Initial tasks (defaults to a fresh `EmptyTask()` instance).
 
     Returns:
         List of output tasks from the final stage.

fern/versions/main/pages/api-reference/pipeline.mdx

@@ -74,18 +74,26 @@ def run(
     self,
     executor: BaseExecutor | None = None,
     initial_tasks: list[Task] | None = None,
+    checkpoint_path: str | Path | None = None,
 ) -> list[Task] | None:
     """Execute the pipeline.
 
     Args:
         executor: Executor to use. Defaults to XennaExecutor.
         initial_tasks: Initial tasks to start the pipeline.
+        checkpoint_path: Optional resumability directory. Completed source
+            partitions are recorded under `.nemo_curator_metadata` and skipped
+            when the pipeline runs again with the same path. In multi-node or
+            distributed runs, this must resolve to the same durable shared
+            filesystem location on every worker and node.
 
     Returns:
         List of output tasks from the final stage, or None.
     """
 ```
 
+Refer to [Resumable Processing](/reference/infra/resumable-processing) for checkpoint storage, replay semantics, task identity, and concurrency limitations.
+
 ### `describe()`
 
 Get a detailed description of the pipeline.
@@ -140,6 +148,14 @@ executor = XennaExecutor(config={"execution_mode": "streaming"})
 results = pipeline.run(executor=executor)
 ```
 
+### Resume an Interrupted Run
+
+```python
+results = pipeline.run(checkpoint_path="./checkpoints/text-curation")
+```
+
+Run the same pipeline with the same path after an interruption. Sources recorded as completed are skipped; incomplete sources restart from the source stage.
+
 ### Pipeline Configuration
 
 ```python

fern/versions/main/pages/api-reference/processing-stage.mdx

@@ -142,6 +142,10 @@ def process_batch(self, tasks: list[InputT]) -> list[OutputT | None]:
     """
 ```
 
+`task_id` is framework-owned. The executor adapter assigns it after either `process()` or `process_batch()` returns, so custom stages must not set or derive IDs. For deterministic lineage, preserve positional correspondence in batched code: return one task or `None` for every input. A batch that maps multiple inputs to a different number of outputs receives random `r`-prefixed IDs because parentage is ambiguous.
+
+For source-level checkpointing and the complete mapping rules, refer to [Resumable Processing](/reference/infra/resumable-processing#custom-stage-responsibilities).
+
 ## Creating Custom Stages
 
 ```python
@@ -176,7 +180,6 @@ class MyCustomStage(ProcessingStage[DocumentBatch, DocumentBatch]):
             return None
 
         return DocumentBatch(
-            task_id=f"{task.task_id}_{self.name}",
             dataset_name=task.dataset_name,
             data=df,
             _metadata=task._metadata,

fern/versions/main/pages/api-reference/tasks/audio-task.mdx

@@ -21,14 +21,14 @@ class AudioTask(Task[dict]):
     """Task containing a single audio manifest entry for processing.
 
     Attributes:
-        task_id: Unique identifier for this task.
+        task_id: Framework-managed deterministic lineage identifier.
         dataset_name: Name of the source dataset.
         data: Audio manifest entry (single dict, stored as _AttrDict).
     """
 
-    task_id: str
     dataset_name: str
     data: dict  # _AttrDict subclass — supports attribute-style access
+    # task_id is inherited from Task with init=False.
 ```
 
 ## Audio Manifest Format
@@ -67,7 +67,6 @@ from nemo_curator.tasks import AudioTask
 
 # Single manifest entry
 task = AudioTask(
-    task_id="audio_001",
     dataset_name="speech_dataset",
     data={
         "audio_filepath": "/data/audio/sample.wav",
@@ -81,6 +80,8 @@ task.data["audio_filepath"]   # "/data/audio/sample.wav"
 task.data.audio_filepath      # "/data/audio/sample.wav"
 ```
 
+Do not pass or derive `task_id`. Treat it as read-only even when working with a task subclass that exposes a compatibility default; the adapter overwrites it with framework lineage at the next stage boundary.
+
 ## Usage in Stages
 
 All audio stages subclass `ProcessingStage[AudioTask, AudioTask]` directly — there is no intermediate base class.

fern/versions/main/pages/api-reference/tasks/document-batch.mdx

@@ -23,14 +23,14 @@ class DocumentBatch(Task[pa.Table | pd.DataFrame]):
     """Task containing a batch of text documents.
 
     Attributes:
-        task_id: Unique identifier for this batch.
+        task_id: Framework-managed deterministic lineage identifier.
         dataset_name: Name of the source dataset.
         data: DataFrame or PyArrow Table containing documents.
     """
 
-    task_id: str
     dataset_name: str
     data: pa.Table | pd.DataFrame
+    # task_id is inherited from Task with init=False.
 ```
 
 ## Expected Data Schema
@@ -112,14 +112,15 @@ df = pd.DataFrame({
 })
 
 batch = DocumentBatch(
-    task_id="batch_001",
     dataset_name="my_dataset",
     data=df,
 )
 
 print(f"Batch contains {batch.num_items} documents")
 ```
 
+Do not pass `task_id` to the constructor. It remains empty until the framework assigns lineage at a stage boundary or when the task is supplied through `initial_tasks`.
+
 ## Usage in Stages
 
 ```python
@@ -151,7 +152,6 @@ class TextFilterStage(ProcessingStage[DocumentBatch, DocumentBatch]):
             return None
 
         return DocumentBatch(
-            task_id=f"{task.task_id}_filtered",
             dataset_name=task.dataset_name,
             data=filtered_df,
             _metadata=task._metadata,
@@ -169,7 +169,6 @@ def process(self, task: DocumentBatch) -> DocumentBatch:
     df["word_count"] = df["text"].str.split().str.len()
 
     return DocumentBatch(
-        task_id=f"{task.task_id}_{self.name}",
         dataset_name=task.dataset_name,
         data=df,
         _metadata=task._metadata,
@@ -188,7 +187,6 @@ def process(self, task: DocumentBatch) -> DocumentBatch | None:
         return None  # Filter out entire batch
 
     return DocumentBatch(
-        task_id=f"{task.task_id}_{self.name}",
         dataset_name=task.dataset_name,
         data=filtered,
         _metadata=task._metadata,

fern/versions/main/pages/api-reference/tasks/image-batch.mdx

@@ -22,14 +22,14 @@ class ImageBatch(Task[list[ImageObject]]):
     """Task containing a batch of images.
 
     Attributes:
-        task_id: Unique identifier for this batch.
+        task_id: Framework-managed deterministic lineage identifier.
         dataset_name: Name of the source dataset.
         data: List of ImageObject instances.
     """
 
-    task_id: str
     dataset_name: str
     data: list[ImageObject]
+    # task_id is inherited from Task with init=False.
 ```
 
 ## ImageObject
@@ -88,12 +88,13 @@ images = [
 
 # Create batch
 batch = ImageBatch(
-    task_id="img_batch_001",
     dataset_name="image_dataset",
     data=images,
 )
 ```
 
+Do not pass `task_id` to the constructor. The framework assigns it after the task crosses a stage boundary.
+
 ## Usage in Stages
 
 ```python
@@ -125,7 +126,6 @@ class ImageFilterStage(ProcessingStage[ImageBatch, ImageBatch]):
             return None
 
         return ImageBatch(
-            task_id=f"{task.task_id}_filtered",
             dataset_name=task.dataset_name,
             data=filtered,
             _metadata=task._metadata,
@@ -143,7 +143,6 @@ def process(self, task: ImageBatch) -> ImageBatch:
         img.embeddings = self.model.encode(img.path)
 
     return ImageBatch(
-        task_id=f"{task.task_id}_{self.name}",
         dataset_name=task.dataset_name,
         data=task.data,
         _metadata=task._metadata,
@@ -164,7 +163,6 @@ def process(self, task: ImageBatch) -> ImageBatch | None:
         return None
 
     return ImageBatch(
-        task_id=f"{task.task_id}_{self.name}",
         dataset_name=task.dataset_name,
         data=filtered,
         _metadata=task._metadata,

fern/versions/main/pages/api-reference/tasks/video-task.mdx

@@ -22,14 +22,14 @@ class VideoTask(Task[Video]):
     """Task containing a single video for processing.
 
     Attributes:
-        task_id: Unique identifier for this task.
+        task_id: Framework-managed deterministic lineage identifier.
         dataset_name: Name of the source dataset.
         data: Video object with path and metadata.
     """
 
-    task_id: str
     dataset_name: str
     data: Video
+    # task_id is inherited from Task with init=False.
 ```
 
 ## Video Class
@@ -88,12 +88,13 @@ video = Video(
 
 # Create task
 task = VideoTask(
-    task_id="video_001",
     dataset_name="video_dataset",
     data=video,
 )
 ```
 
+Do not pass `task_id` to the constructor. The framework assigns it after the task crosses a stage boundary.
+
 ## Usage in Stages
 
 ```python
@@ -123,7 +124,6 @@ class VideoFilterStage(ProcessingStage[VideoTask, VideoTask]):
             return None
 
         return VideoTask(
-            task_id=f"{task.task_id}_filtered",
             dataset_name=task.dataset_name,
             data=video,
             _metadata=task._metadata,
@@ -150,7 +150,6 @@ def process(self, task: VideoTask) -> list[VideoTask]:
             metadata=video.metadata.copy(),
         )
         clips.append(VideoTask(
-            task_id=f"{task.task_id}_clip_{i}",
             dataset_name=task.dataset_name,
             data=clip_video,
             _metadata=task._metadata,
@@ -168,7 +167,6 @@ def process(self, task: VideoTask) -> VideoTask:
     video.embeddings = self.encoder.encode(video.path)
 
     return VideoTask(
-        task_id=f"{task.task_id}_{self.name}",
         dataset_name=task.dataset_name,
         data=video,
         _metadata=task._metadata,