Schema rewrite to hinge on load strategy, not task

docs/add_new_schema.md

@@ -28,9 +28,12 @@ Create a file named `schema.yaml`. Start with the basic required fields:
 
 ```yaml
 dataset_id: "your-dataset-id"   # The unique ID of the dataset on MDC
-task: "ASR"                    # ASR, TTS, or OTH
+task: "ASR"                    # e.g. ASR, TTS, or OTH
 ```
 
+The `task` is used to validate that the loaded DataFrame contains 
+the task's required columns (e.g. ASR/TTS: `audio_path` + `transcription`).
+
 Then add the fields for your chosen strategy.
 
 #### Example: Index-based ASR

docs/api.md

@@ -23,7 +23,15 @@
 
 ::: datacollective.schema_loaders.cache_schema
 
-::: datacollective.schema_loaders.tasks.asr
+::: datacollective.schema_loaders.contracts
 
-::: datacollective.schema_loaders.tasks.tts
+::: datacollective.schema_loaders.strategies.index
+
+::: datacollective.schema_loaders.strategies.multi_split
+
+::: datacollective.schema_loaders.strategies.multi_sections
+
+::: datacollective.schema_loaders.strategies.paired_glob
+
+::: datacollective.schema_loaders.strategies.glob
 

docs/extend_schema_loading_logic.md

@@ -1,14 +1,16 @@
 # Extending Schema Loading Logic
 
-This document is for **developers** who want to add support for new tasks or implement new loading strategies within the MDC Python SDK.
+This document is for **developers** who want to add new loading strategies or task contracts within the MDC Python SDK.
 
-## 1. How to add a new task type
+Dispatch is **strategy-based**: the schema's `root_strategy` field (default: `"index"`) selects the loader class, and any strategy can be combined with any task. The optional `task` field only adds a validation step — the loaded DataFrame must contain the task's required logical columns.
 
-Supporting a new task (e.g., **MT** — Machine Translation) involves creating a new loader class and registering it.
+## 1. How to add a new strategy
+
+Supporting a new strategy (e.g. **manifest** — one JSON manifest per dataset) involves creating a new loader class and registering it.
 
 ### Step 1: Create the loader class
 
-Create a new file under `src/datacollective/schema_loaders/tasks/`, for example `mt.py`:
+Create a new file under `src/datacollective/schema_loaders/strategies/`, for example `manifest.py`:
 
 ```python
 from __future__ import annotations
@@ -17,20 +19,20 @@ import pandas as pd
 from datacollective.schema import DatasetSchema
 from datacollective.schema_loaders.base import BaseSchemaLoader
 
-class MTLoader(BaseSchemaLoader):
-    """Load a machine-translation dataset."""
+class ManifestLoader(BaseSchemaLoader):
+    """Load a dataset described by a single JSON manifest."""
 
     def __init__(self, schema: DatasetSchema, extract_dir: Path) -> None:
         super().__init__(schema, extract_dir)
-        # Validate required schema fields
+        # Validate required schema fields up front
         if not schema.index_file:
-            raise ValueError("MT schema must specify 'index_file'")
+            raise ValueError("manifest schema must specify 'index_file'")
 
     def load(self) -> pd.DataFrame:
         # BaseSchemaLoader provides shared helpers:
         # 1. Locate and read the index file
         raw_df = self._load_index_file()
-        
+
         # 2. Apply column mappings and dtypes
         return self._apply_column_mappings(raw_df)
 ```
@@ -47,39 +49,49 @@ When implementing `load()`, you can leverage these methods from the base class:
 
 ### Step 3: Register the loader
 
-Register your new class in `src/datacollective/schema_loaders/registry.py`:
+1. **Add to the Enum**: add your strategy to the `Strategy` enum in `src/datacollective/schema_loaders/base.py`.
+2. **Register the class** in `src/datacollective/schema_loaders/registry.py`:
 
 ```python
-from datacollective.schema_loaders.tasks.mt import MTLoader
+from datacollective.schema_loaders.strategies.manifest import ManifestLoader
 
-_TASK_REGISTRY: dict[str, Type[BaseSchemaLoader]] = {
-    "ASR": ASRLoader,
-    "TTS": TTSLoader,
-    "OTH": OTHLoader,
-    "MT":  MTLoader,  # Add your new task here
+_STRATEGY_REGISTRY: dict[Strategy, Type[BaseSchemaLoader]] = {
+    Strategy.INDEX: IndexLoader,
+    ...
+    Strategy.MANIFEST: ManifestLoader,  # Add your new strategy here
 }
 ```
 
-## 2. How to extend or update strategies
-
-Strategies define the high-level approach to locating data (e.g., using an index file vs. globbing).
+3. **Update Schema**: if the strategy requires new YAML fields, add them to `DatasetSchema` in `src/datacollective/schema.py`.
 
 ### Loading Strategies (`Strategy` enum)
 
-Strategies are defined in the `Strategy` enum in `src/datacollective/schema_loaders/base.py`:
-
 | Enum Member | YAML Value | Description |
 |---|---|---|
-| `Strategy.MULTI_SPLIT` | `"multi_split"` | Loads multiple split files matching a pattern (ASR). |
-| `Strategy.MULTI_SECTIONS` | `"multi_sections"` | Loads one index file per section directory, adding a `section` column (TTS). |
-| `Strategy.PAIRED_GLOB` | `"paired_glob"` | Pairs audio files with sidecar files: `.txt` transcriptions (TTS) or JSON metadata/utterance files (ASR, with `format: "json"` and optional `record_path`). |
-| `Strategy.GLOB` | `"glob"` | Walks directory-structured datasets, deriving metadata from the path hierarchy (OTH). |
+| `Strategy.INDEX` | `"index"` (default) | Loads a single delimited index file, optionally applying column mappings. |
+| `Strategy.MULTI_SPLIT` | `"multi_split"` | Loads multiple split files matching a pattern, adding a `split` column. |
+| `Strategy.MULTI_SECTIONS` | `"multi_sections"` | Loads one index file per section directory, adding a `section` column. |
+| `Strategy.PAIRED_GLOB` | `"paired_glob"` | Pairs audio files with sidecar files: `.txt` transcriptions, or JSON metadata/utterance files (with `format: "json"` and optional `record_path`). |
+| `Strategy.GLOB` | `"glob"` | Walks directory-structured datasets, deriving metadata from the path hierarchy. |
+
+## 2. How to add a task contract
+
+The optional `task` field validates the loaded DataFrame against the task's required logical columns. Contracts live in `src/datacollective/schema_loaders/contracts.py`:
+
+```python
+TASK_CONTRACTS: dict[str, frozenset[str]] = {
+    "ASR": frozenset({"audio_path", "transcription"}),
+    "TTS": frozenset({"audio_path", "transcription"}),
+    "LLM": frozenset({"text"}),
+    # Add your new task contract here, e.g.
+    # "MT": frozenset({"source_text", "target_text"}),
+}
+```
 
-### Adding a new strategy
+A contract violation raises `TaskValidationError`. Schemas whose task has no contract (e.g. `OTH`), or with no task at all, load without validation.
 
-1. **Add to the Enum**: Add your new strategy to the `Strategy` class in `base.py`.
-2. **Implement Logic**: Add a branch in the relevant loader's `load()` method to handle the new strategy.
-3. **Update Schema**: If the strategy requires new YAML fields, add them to `DatasetSchema` in `src/datacollective/schema.py`.
+> **Note for registry schemas:** keep the `task` field in existing `schema.yaml`
+> files — older SDK versions still require it.
 
 ## 3. Architecture Overview
 
@@ -91,18 +103,21 @@ When a user calls `load_dataset("id")`:
 2. **`_extract_archive()`**: Extracts it to a local directory. (Skipped if already extracted)
 3. **`_resolve_schema()`**: Locates or downloads `schema.yaml`.
 4. **`_parse_schema()`**: Validates YAML into a `DatasetSchema` object.
-5. **`_load_dataset_from_schema()`**: 
+5. **`_load_dataset_from_schema()`**:
     - If the schema specifies `extract_files`, extracts inner archives (skipped when already extracted).
-    - Finds the correct loader in the **Registry**.
+    - Resolves the strategy loader from the **Registry** (`root_strategy`, default `"index"`).
+    - If the task has a contract and the schema declares column mappings, fails fast when the declared logical columns cannot satisfy the contract.
     - Calls `loader.load()`.
+    - Validates the loaded DataFrame against the task contract (when one exists).
     - Returns the final **pandas DataFrame**.
 
 ### Module Map
 
 | Module | Responsibility |
 |---|---|
 | `datacollective.schema` | Pydantic models and YAML parsing. |
-| `datacollective.schema_loaders.base` | Abstract base class and strategy definitions. |
-| `datacollective.schema_loaders.registry` | Task-to-loader mapping. |
+| `datacollective.schema_loaders.base` | Abstract base class, shared helpers, and strategy definitions. |
+| `datacollective.schema_loaders.registry` | Strategy-to-loader mapping and load orchestration. |
+| `datacollective.schema_loaders.contracts` | Task contracts and their validation. |
 | `datacollective.schema_loaders.cache_schema` | Local schema caching and checksum validation. |
-| `datacollective.schema_loaders.tasks.*` | Implementation of task-specific logic (ASR, TTS, OTH). |
+| `datacollective.schema_loaders.strategies.*` | One loader per strategy (index, multi_split, multi_sections, paired_glob, glob). |

docs/loaders/asr.md

@@ -1,5 +1,10 @@
 # ASR Loader
 
+> **Note:** since v0.6 loading strategies are **task-agnostic** — the schema's
+> `root_strategy` selects the loader, and any strategy can be used with any
+> task. This page describes the strategies commonly used for ASR datasets.
+> The `task` field now only validates that the loaded DataFrame contains the
+> task's required columns (`audio_path` + `transcription`).
 Loader for **ASR** (Automatic Speech Recognition) datasets.
 
 There are three parsing strategies for ASR datasets, controlled by the `root_strategy` field in the schema.

docs/loaders/oth.md

@@ -1,5 +1,10 @@
 # OTH Loader
 
+> **Note:** since v0.6 loading strategies are **task-agnostic** — the schema's
+> `root_strategy` selects the loader, and any strategy can be used with any
+> task. This page describes the strategies commonly used for OTH datasets.
+> The `task` field now only validates that the loaded DataFrame contains the
+> task's required columns (no contract — OTH loads without validation).
 Loader for tasks classified as **OTH** (other).
 
 There are two parsing strategies, controlled by the `root_strategy` field in the schema.

docs/loaders/tts.md

@@ -1,5 +1,10 @@
 # TTS Loader
 
+> **Note:** since v0.6 loading strategies are **task-agnostic** — the schema's
+> `root_strategy` selects the loader, and any strategy can be used with any
+> task. This page describes the strategies commonly used for TTS datasets.
+> The `task` field now only validates that the loaded DataFrame contains the
+> task's required columns (`audio_path` + `transcription`).
 Loader for **TTS** (Text-to-Speech) datasets.
 
 There are three parsing strategies for TTS datasets, controlled by the `root_strategy` field in the schema.

docs/schema_documentation.md

@@ -19,13 +19,13 @@ Under the hood, `load_dataset()` performs the following steps automatically:
 1. **Resolve the schema**: check local cache or the schema registry for `schema.yaml`. If the dataset is not registered this step raises a warning, so we never download an unsupported archive.
 2. **Download** the archive (with resume support). The schema we fetched in step 1 tells the loader how the files are structured.
 3. **Extract** the `.tar.gz` / `.zip` to a local directory.
-4. **Parse** the YAML into a validated `DatasetSchema` (Pydantic model) and dispatch to the task-specific loader (ASR, TTS, OTH, …), which returns the final **DataFrame**.
+4. **Parse** the YAML into a validated `DatasetSchema` (Pydantic model) and dispatch to the loader for the schema's `root_strategy` (index, glob, …), which returns the final **DataFrame**. When the schema declares a `task` with a known contract (ASR, TTS, LLM), the loaded DataFrame is validated to contain the task's required logical columns.
 
 The schema file describes:
 
-- **What task** the dataset is for (ASR, TTS, …).
 - **How to find** the data files (index file path, glob pattern, etc.).
 - **How to map** raw columns / files into a clean DataFrame.
+- Optionally, **what task** the dataset is for (ASR, TTS, …), which the loaded DataFrame is validated against.
 
 ### Minimal example
 
@@ -56,7 +56,8 @@ Every schema **must** have:
 | Field | Type | Required | Description |
 |---|---|---|---|
 | `dataset_id` | `str` | ✓ | Unique dataset identifier on MDC. |
-| `task` | `str` | ✓ | Task type: determines which loader is used (`"ASR"`, `"TTS"`, `"OTH"`). |
+| `task` | `str` | ✗ | *(optional)* Task type as defined on the MDC Platform (`"ASR"`, `"TTS"`, …). When set to a task with a known contract, the loaded DataFrame is validated to contain the task's required logical columns (e.g. ASR/TTS: `audio_path` + `transcription`; LLM: `text`). Tasks without a contract (e.g. `"OTH"`) load without validation. |
+
 
 ### Loading strategies
 

src/datacollective/datasets.py

@@ -161,7 +161,7 @@ def load_dataset(
     If there is a directory with the same name as the archive file without the suffix extension, we assume
     it has already been extracted, and it will not be re-extracted unless `overwrite_extracted=True`.
 
-    Uses the dataset schema to determine task-specific loading logic.
+    Uses the dataset schema to determine the loading strategy.
 
     Automatically resumes interrupted downloads if a .checksum file exists from a
     previous attempt.

src/datacollective/errors.py

@@ -42,6 +42,10 @@ def __str__(self) -> str:
         return "Download failed. Unfortunately this dataset does not support resuming downloads — please try again."
 
 
+class TaskValidationError(ValueError):
+    """Raised when a loaded dataset does not satisfy its task's column contract."""
+
+
 class MissingDependencyError(ImportError):
     """Raised when an optional dependency required for a feature is not installed."""
 

src/datacollective/schema.py

@@ -67,22 +67,27 @@ class DatasetSchema(BaseModel):
     """
     Task-agnostic representation of a dataset schema, as defined by a ``schema.yaml`` file.
 
-    Every schema **must** have ``dataset_id`` and ``task``.  The remaining
-    fields depend on the task type and the ``root_strategy``
-    (``"index"`` vs ``"glob"``).
+    Every schema **must** have ``dataset_id``.  The remaining fields depend on
+    the ``root_strategy`` (``"index"`` by default); the loader registered for
+    that strategy decides which fields are required at load time.
 
-    New task types only need to populate the fields they care about;
-    the loader registered for that task will decide which fields are
-    required at load time.
+    ``task`` is optional.  When set to a task with a known contract (e.g. ASR,
+    TTS), the loaded DataFrame is validated to contain the task's required
+    logical columns.
     """
 
     model_config = ConfigDict(frozen=False)
 
     dataset_id: str = Field(
         description="Unique identifier for the dataset in the registry"
     )
-    task: str = Field(
-        description="A task as defined in the MDC Platform e.g. ASR, TTS etc"
+    task: str | None = Field(
+        default=None,
+        description=(
+            "Optional task as defined in the MDC Platform e.g. ASR, TTS etc. "
+            "When set to a task with a known contract, the loaded dataset is "
+            "validated against it."
+        ),
     )
 
     # --- Index-based strategy (ASR / TTS) ---
@@ -114,9 +119,13 @@ class DatasetSchema(BaseModel):
         default="utf-8", description='file encoding (e.g. "utf-8-sig" for BOM)'
     )
 
-    # --- Glob-based strategy (LM, paired-file TTS) ---
+    # --- Loading strategy ---
     root_strategy: str | None = Field(
-        default=None, description='"glob" | "paired_glob" | "multi_split"'
+        default=None,
+        description=(
+            '"index" (default) | "glob" | "paired_glob" | "multi_split" | '
+            '"multi_sections"'
+        ),
     )
     file_pattern: str | None = Field(default=None, description='e.g. "**/*.txt"')
     audio_extension: str | None = Field(
@@ -230,9 +239,9 @@ def _parse_schema(raw: str | dict[str, Any] | Path) -> DatasetSchema:
     data: dict[str, Any] = raw
 
     dataset_id = data.get("dataset_id")
+    if not dataset_id:
+        raise ValueError("schema.yaml must contain 'dataset_id'")
     task = data.get("task")
-    if not dataset_id or not task:
-        raise ValueError("schema.yaml must contain 'dataset_id' and 'task'")
 
     # Columns (index-based)
     columns: dict[str, ColumnMapping] = {}
@@ -286,7 +295,7 @@ def _parse_schema(raw: str | dict[str, Any] | Path) -> DatasetSchema:
 
     return DatasetSchema(
         dataset_id=str(dataset_id),
-        task=str(task).upper(),
+        task=str(task).upper() if task else None,
         format=data.get("format"),
         index_file=data.get("index_file"),
         base_audio_path=data.get("base_audio_path"),

src/datacollective/schema_loaders/__init__.py

@@ -3,15 +3,17 @@
     FORMAT_SEP,
     Strategy,
 )
+from datacollective.schema_loaders.contracts import TASK_CONTRACTS
 from datacollective.schema_loaders.registry import (
-    _get_task_loader,
+    _get_strategy_loader,
     _load_dataset_from_schema,
 )
 
 __all__ = [
     "BaseSchemaLoader",
     "FORMAT_SEP",
     "Strategy",
-    "_get_task_loader",
+    "TASK_CONTRACTS",
+    "_get_strategy_loader",
     "_load_dataset_from_schema",
 ]