fix(cache): make task fingerprints deterministic and collision-free

docs/api/tasks.rst

@@ -184,6 +184,49 @@ PyHealth detects the existing cache and skips reprocessing. During
 development it is useful to set ``dev=True`` on the dataset, which limits
 processing to 1 000 patients so iterations are fast.
 
+What counts as the same configuration?
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+Each task configuration gets its own cache directory, named after a
+fingerprint of everything that determines the generated samples:
+
+- the ``__init__`` arguments of your task, with defaults applied --
+  ``MyTask()`` and ``MyTask(window=timedelta(days=15))`` share a cache when
+  15 days is the default;
+- class-level attributes, the input schema and the output schema;
+- ``BaseTask.version`` (see below).
+
+Arguments that do not affect the output are ignored, so changing
+``num_workers`` reuses the cache. Add your own via ``fingerprint_exclude``::
+
+    class MyTask(BaseTask):
+        fingerprint_exclude = frozenset({"debug_dir"})
+
+The fingerprint is a digest, so the directory name alone does not tell you
+which parameter changed. Each cache directory therefore contains a
+``task_meta.json`` recording the full configuration in readable form, and
+``set_task()`` logs its path at INFO level.
+
+.. warning::
+
+   **Editing** ``__call__`` **does not invalidate the cache.** The fingerprint
+   covers configuration, not code, so changing your labelling logic without
+   changing any argument will silently reuse the samples built by the previous
+   version. Bump ``version`` on the task when its logic changes::
+
+       class MyTask(BaseTask):
+           version = "2"   # was "1"; excluded ICU stays under 24h
+
+   Alternatively, set ``PYHEALTH_FINGERPRINT_SOURCE=1`` to fold a structural
+   hash of ``__call__`` and ``pre_filter`` into the fingerprint. This is off by
+   default because it invalidates the cache on cosmetic edits too.
+
+If a task argument cannot be fingerprinted deterministically -- an object
+whose ``repr()`` embeds a memory address, for instance -- ``set_task()``
+raises ``UnfingerprintableError`` rather than risk a colliding or unstable
+cache key. Give the class a ``__pyhealth_fingerprint__()`` method returning a
+plain description of its configuration, or exclude the argument.
+
 .. note::
 
    **A note on multiprocessing.** ``set_task()`` can spawn worker processes
@@ -206,6 +249,7 @@ Available Tasks
     :maxdepth: 3
 
     Base Task <tasks/pyhealth.tasks.BaseTask>
+    Task Fingerprint <tasks/pyhealth.tasks.fingerprint>
     In-Hospital Mortality (MIMIC-IV) <tasks/pyhealth.tasks.InHospitalMortalityMIMIC4>
     In-Hospital Mortality (MEDS) <tasks/pyhealth.tasks.InHospitalMortalityMEDS>
     MIMIC-III ICD-9 Coding <tasks/pyhealth.tasks.MIMIC3ICD9Coding>

docs/api/tasks/pyhealth.tasks.fingerprint.rst

@@ -0,0 +1,11 @@
+pyhealth.tasks.fingerprint
+=======================================
+
+Deterministic cache keys for ``set_task``. Replaces the previous
+``json.dumps(..., default=str)`` hash, which was neither stable across
+processes nor injective for large arrays.
+
+.. automodule:: pyhealth.tasks.fingerprint
+    :members: UnfingerprintableError, task_spec, task_fingerprint, task_cache_name, processors_fingerprint, write_task_metadata, slugify, record_init_args
+    :undoc-members:
+    :show-inheritance:

examples/task_fingerprint.py

@@ -0,0 +1,24 @@
+"""Show that task cache keys include init args (issue #916).
+
+Two ReadmissionPredictionMIMIC3 configurations that differ only by
+``window`` must produce different cache directory names. A slash in
+``task_name`` (as used by BenchmarkEHRShot) is slugified to a single
+path component.
+"""
+
+from datetime import timedelta
+
+from pyhealth.tasks.benchmark_ehrshot import BenchmarkEHRShot
+from pyhealth.tasks.fingerprint import task_cache_name, task_fingerprint
+from pyhealth.tasks.readmission_prediction import ReadmissionPredictionMIMIC3
+
+if __name__ == "__main__":
+    t15 = ReadmissionPredictionMIMIC3()
+    t30 = ReadmissionPredictionMIMIC3(window=timedelta(days=30))
+    print("15-day window:", task_cache_name(t15))
+    print("30-day window:", task_cache_name(t30))
+    assert task_fingerprint(t15) != task_fingerprint(t30)
+
+    ehrshot = task_cache_name(BenchmarkEHRShot(task="guo_los"))
+    print("EHRShot guo_los:", ehrshot)
+    assert "/" not in ehrshot

pyhealth/datasets/base_dataset.py

@@ -43,6 +43,11 @@
 from ..processors.base_processor import FeatureProcessor
 from .configs import load_yaml_config
 from .sample_dataset import SampleDataset, SampleBuilder
+from ..tasks.fingerprint import (
+    processors_fingerprint,
+    task_cache_name,
+    write_task_metadata,
+)
 from ..utils import set_env
 
 # Set logging level for distributed to ERROR to reduce verbosity
@@ -388,9 +393,10 @@ def _init_cache_dir(self, cache_dir: str | Path | None) -> Path:
                 tmp/                                # Temporary files during processing
                 global_event_df.parquet/            # Cached global event dataframe
                 tasks/                              # Cached task-specific data
-                    {task_name}_{task_uuid}/        # Cached data for specific task based on task name, schema, and args
+                    {task_name}_{task_fingerprint}/ # One task configuration (see set_task)
+                        task_meta.json              # Readable spec behind the fingerprint
                         task_df.ld/                 # Intermediate task dataframe based on schema
-                        samples_{proc_uuid}.ld/     # Final processed samples after applying processors
+                        samples_{proc_fingerprint}.ld/  # Final processed samples after applying processors
 
         Returns:
             Path: The resolved cache directory path.
@@ -1005,11 +1011,28 @@ def set_task(
         """Processes the base dataset to generate the task-specific sample dataset.
         The cache structure is as follows::
 
-            {task_name}_{task_uuid}/        # Cached data for specific task based on task name, schema, and args
-                task_df.ld/                 # Intermediate task dataframe based on schema
-                samples_{proc_uuid}.ld/     # Final processed samples after applying processors
-                    schema.pkl              # Saved SampleBuilder schema
-                    *.bin                   # Processed sample files
+            {task_name}_{task_fingerprint}/     # One task configuration
+                task_meta.json                  # Readable spec behind the fingerprint
+                build.lock                      # Guards concurrent builds
+                task_df.ld/                     # Intermediate task dataframe based on schema
+                samples_{proc_fingerprint}.ld/  # Final processed samples after applying processors
+                    schema.pkl                  # Saved SampleBuilder schema
+                    *.bin                       # Processed sample files
+
+        ``task_fingerprint`` is a digest of everything that determines the
+        generated samples: the ``__init__`` arguments of the task (with defaults
+        applied, so passing a default explicitly is not a different config),
+        class-level attributes, the input and output schemas, and
+        ``BaseTask.version``. Changing any of them yields a new directory, so
+        two configurations can never share a cache. Arguments that do not affect
+        the output -- ``num_workers``, ``verbose`` -- are excluded, and a task
+        can exclude more of its own via ``fingerprint_exclude``.
+
+        Because the digest is opaque, ``task_meta.json`` records the full spec
+        in readable form: consult it to see which parameter produced a given
+        directory. Note that editing ``__call__`` does *not* change the
+        fingerprint by itself -- bump ``version`` on the task when its logic
+        changes, or existing caches will be silently reused.
 
         Args:
             task (Optional[BaseTask]): The task to set. Uses default task if None.
@@ -1040,52 +1063,25 @@ def set_task(
             f"Setting task {task.task_name} for {self.dataset_name} base dataset..."
         )
 
-        task_params = json.dumps(
-            {
-                **vars(task),
-                "input_schema": task.input_schema,
-                "output_schema": task.output_schema,
-            },
-            sort_keys=True,
-            default=str,
-        )
-
-        cache_dir = (
-            self.cache_dir
-            / "tasks"
-            / f"{task.task_name}_{uuid.uuid5(uuid.NAMESPACE_DNS, task_params)}"
-        )
+        cache_dir = self.cache_dir / "tasks" / task_cache_name(task)
         cache_dir.mkdir(parents=True, exist_ok=True)
 
-        proc_params = json.dumps(
-            {
-                "input_processors": (
-                    {
-                        f"{k}_{v.__class__.__name__}": vars(v)
-                        for k, v in input_processors.items()
-                    }
-                    if input_processors
-                    else None
-                ),
-                "output_processors": (
-                    {
-                        f"{k}_{v.__class__.__name__}": vars(v)
-                        for k, v in output_processors.items()
-                    }
-                    if output_processors
-                    else None
-                ),
-            },
-            sort_keys=True,
-            default=str,
-        )
-
         task_df_path = Path(cache_dir) / "task_df.ld"
         samples_path = (
             Path(cache_dir)
-            / f"samples_{uuid.uuid5(uuid.NAMESPACE_DNS, proc_params)}.ld"
+            / f"samples_{processors_fingerprint(input_processors, output_processors)[:16]}.ld"
         )
 
+        # The cache key is opaque by construction; the sidecar makes it
+        # auditable. Written before the build so a crashed build still leaves
+        # a directory that explains what it was trying to produce.
+        meta_path = write_task_metadata(cache_dir, task)
+        logger.info(
+            "Task cache key %s -- derived from init args, schemas and task "
+            "version. Full spec: %s",
+            task_cache_name(task),
+            meta_path,
+        )
         logger.info(f"Task cache paths: task_df={task_df_path}, samples={samples_path}")
 
         task_df_path.mkdir(parents=True, exist_ok=True)

pyhealth/tasks/base_task.py

@@ -3,12 +3,47 @@
 
 import polars as pl
 
+from .fingerprint import record_init_args
+
 
 class BaseTask(ABC):
+    """Base class for PyHealth predictive tasks.
+
+    Init arguments, class-level configuration, and ``version`` are part of
+    the ``set_task`` cache key. Bump ``version`` when ``__call__`` or
+    ``pre_filter`` changes in a way that alters generated samples.
+
+    Example:
+        >>> from pyhealth.tasks.base_task import BaseTask
+        >>> class ToyTask(BaseTask):
+        ...     task_name = "toy"
+        ...     input_schema = {"x": "sequence"}
+        ...     output_schema = {"y": "binary"}
+        ...     def __call__(self, patient):
+        ...         return []
+        >>> ToyTask().task_name
+        'toy'
+    """
+
     task_name: str
     input_schema: Dict[str, Union[str, Type]]
     output_schema: Dict[str, Union[str, Type]]
 
+    #: Bump when ``__call__`` or ``pre_filter`` logic changes in a way that
+    #: alters the generated samples. Init args alone cannot detect this.
+    version: str = "1"
+
+    #: Attribute names that do not affect the generated samples and must not
+    #: invalidate the cache (e.g. ``num_workers``, ``verbose``). Denylist, not
+    #: allowlist: forgetting an entry here costs a spurious rebuild, whereas
+    #: forgetting to allowlist a semantic arg silently reuses a stale cache.
+    fingerprint_exclude: frozenset[str] = frozenset()
+
+    def __init_subclass__(cls, **kwargs) -> None:
+        """Record the effective ``__init__`` arguments of every task instance."""
+        super().__init_subclass__(**kwargs)
+        record_init_args(cls)
+
     def __init__(
         self,
         code_mapping: Optional[Dict[str, Tuple[str, str]]] = None,