feat(config): add missing_cache_policy to NodeConfig (ITL-604)

src/orcapod/errors.py

@@ -195,3 +195,14 @@ def __init__(
             f"node={node_identity_path!r} "
             f"cached_content_hash={cached_content_hash!r}"
         )
+
+
+class CacheMissError(Exception):
+    """Raised when a persistent (non-ephemeral) result-store entry is absent
+    and ``NodeConfig.missing_cache_policy`` is ``"strict"``.
+
+    A missing durable result indicates data loss or corruption. Set
+    ``missing_cache_policy="recompute"`` (the default) to fall back to
+    recomputation, or ``"as_empty"`` to propagate an ``EmptyData`` token
+    downstream instead of raising.
+    """

src/orcapod/types.py

@@ -19,7 +19,7 @@
 from datetime import date, datetime
 from enum import Enum
 from types import UnionType
-from typing import TYPE_CHECKING, Any, Generic, Self, TypeAlias, TypeVar
+from typing import TYPE_CHECKING, Any, Generic, Literal, Self, TypeAlias, TypeVar
 
 if TYPE_CHECKING:
     import pyarrow as pa
@@ -360,10 +360,30 @@ class NodeConfig:
             means no old schema is tolerated — any detected v0 table raises
             ``SchemaVersionError``. Pass ``("v0",)`` to suppress the error
             and allow the node to recompute all results from scratch.
+        missing_cache_policy: Controls how the node reacts when the pipeline
+            table has an entry for an input but the result store does not.
+            ``None`` inherits the default (``"recompute"``).
+
+            * ``"recompute"`` *(default)* — WARNING logged; the entry falls
+              through to recomputation from the original upstream data.
+              For ephemeral stores, ``EmptyData`` acts as a recompute sentinel
+              (same as current behaviour).
+            * ``"as_empty"`` — WARNING logged (non-ephemeral) or INFO logged
+              (ephemeral); an ``EmptyData`` token is emitted directly. The
+              downstream node attempts to serve the result from its own cache.
+              Only use when partial gaps are semantically expected (e.g.
+              shared read-only stores, exploratory pipelines).
+            * ``"strict"`` — ERROR logged and ``CacheMissError`` raised for
+              non-ephemeral misses. Ephemeral misses still degrade gracefully
+              to ``EmptyData`` (raising on an ephemeral miss would contradict
+              the semantics of ephemeral storage). Use in production pipelines
+              where a missing durable result always indicates a bug or data
+              loss.
     """
 
     is_result_ephemeral: bool | None = None
     ignore_schema: tuple[str, ...] | None = None
+    missing_cache_policy: Literal["recompute", "as_empty", "strict"] | None = None
 
     def merge(self, other: "NodeConfig") -> "NodeConfig":
         """Return a new ``NodeConfig`` with ``other``'s non-``None`` fields overriding self.
@@ -395,6 +415,11 @@ def merge(self, other: "NodeConfig") -> "NodeConfig":
                 if other.ignore_schema is not None
                 else self.ignore_schema
             ),
+            missing_cache_policy=(
+                other.missing_cache_policy
+                if other.missing_cache_policy is not None
+                else self.missing_cache_policy
+            ),
         )