feat(hooks): align PostRunPayload with InvocationContext (ITL-531)

src/orcapod/__init__.py

@@ -10,9 +10,11 @@
     function_pod,
     side_effect_function_pod,
 )
-from .side_effects import (
+from .invocation import (
     InvocationContext,
     InvocationHashConfig,
+)
+from .side_effects import (
     SideEffectPod,
     SideEffectPodConfig,
     side_effect_pod,

src/orcapod/core/cached_function_pod.py

@@ -223,7 +223,8 @@ def _invoke_with_hooks(
             tag: The tag associated with the data.
             data: The input data to process.
             logger: Optional data execution logger.
-            run_id: Pipeline run identifier (unused in cached path).
+            run_id: Pipeline run identifier forwarded to
+                ``PostRunPayload.invocation_context``.
 
         Returns:
             A ``(tag, output_data)`` tuple.
@@ -250,7 +251,7 @@ def _invoke_with_hooks(
             self._fire_post_run_hooks(
                 self._build_post_run_payload(
                     tag, data, None, started_at, finished_at,
-                    InvocationStatus.ERROR, exc,
+                    InvocationStatus.ERROR, exc, run_id=run_id,
                 )
             )
             raise  # bare raise — preserves the original traceback exactly
@@ -259,6 +260,7 @@ def _invoke_with_hooks(
         self._fire_post_run_hooks(
             self._build_post_run_payload(
                 tag, data, output_data, started_at, finished_at, status, None,
+                run_id=run_id,
             )
         )
         return out_tag, output_data
@@ -280,7 +282,8 @@ async def _async_invoke_with_hooks(
             tag: The tag associated with the data.
             data: The input data to process.
             logger: Optional data execution logger.
-            run_id: Pipeline run identifier (unused in cached path).
+            run_id: Pipeline run identifier forwarded to
+                ``PostRunPayload.invocation_context``.
 
         Returns:
             A ``(tag, output_data)`` tuple.
@@ -309,7 +312,7 @@ async def _async_invoke_with_hooks(
             self._fire_post_run_hooks(
                 self._build_post_run_payload(
                     tag, data, None, started_at, finished_at,
-                    InvocationStatus.ERROR, exc,
+                    InvocationStatus.ERROR, exc, run_id=run_id,
                 )
             )
             raise  # bare raise — preserves the original traceback exactly
@@ -318,6 +321,7 @@ async def _async_invoke_with_hooks(
         self._fire_post_run_hooks(
             self._build_post_run_payload(
                 tag, data, output_data, started_at, finished_at, status, None,
+                run_id=run_id,
             )
         )
         return out_tag, output_data

src/orcapod/core/function_pod.py

@@ -185,11 +185,8 @@ def _build_invocation_context(
         """
         import pyarrow as pa
         from orcapod.core.nodes.function_node import _build_record_id_preimage
-        from orcapod.side_effects import (
-            InvocationContext,
-            InvocationHashConfig,
-            _SIDE_EFFECT_RECOMPUTATION_INDEX_COL,
-        )
+        from orcapod.invocation import InvocationContext, InvocationHashConfig
+        from orcapod.side_effects import _SIDE_EFFECT_RECOMPUTATION_INDEX_COL
 
         preimage = _build_record_id_preimage(tag, data).append_column(
             _SIDE_EFFECT_RECOMPUTATION_INDEX_COL,
@@ -350,6 +347,7 @@ def _build_post_run_payload(
         finished_at: datetime,
         status: InvocationStatus,
         exc: Exception | None,
+        run_id: str | None = None,
     ) -> PostRunPayload:
         """Build a ``PostRunPayload`` from invocation results.
 
@@ -361,13 +359,16 @@ def _build_post_run_payload(
             finished_at: UTC timestamp when compute-or-lookup completed.
             status: Invocation status (``COMPUTED``, ``HIT``, or ``ERROR``).
             exc: The exception raised, if ``status == ERROR``; ``None`` otherwise.
+            run_id: Pipeline run identifier, or ``None`` in standalone mode.
+                Forwarded to ``InvocationContext.pipeline_run_id``.
 
         Returns:
             A ``PostRunPayload`` ready to pass to registered hooks.
         """
         record_id = (
             str(output_data.datagram_uuid) if output_data is not None else None
         )
+        invocation_context = self._build_invocation_context(tag, data, run_id=run_id)
         return PostRunPayload(
             record_id_hash=record_id,
             tag=tag,
@@ -384,6 +385,7 @@ def _build_post_run_payload(
                 label=self.label,
                 pod_hash=self.content_hash().to_string(),
             ),
+            invocation_context=invocation_context,
         )
 
     def _invoke_with_hooks(
@@ -404,7 +406,8 @@ def _invoke_with_hooks(
             tag: The tag associated with the data.
             data: The input data to process.
             logger: Optional data execution logger forwarded to ``process_data``.
-            run_id: Pipeline run identifier forwarded to ``process_data``.
+            run_id: Pipeline run identifier forwarded to ``process_data`` and
+                ``PostRunPayload.invocation_context``.
 
         Returns:
             A ``(tag, output_data)`` tuple.
@@ -423,7 +426,7 @@ def _invoke_with_hooks(
             self._fire_post_run_hooks(
                 self._build_post_run_payload(
                     tag, data, None, started_at, finished_at,
-                    InvocationStatus.ERROR, exc,
+                    InvocationStatus.ERROR, exc, run_id=run_id,
                 )
             )
             raise  # bare raise — preserves the original traceback exactly
@@ -432,7 +435,7 @@ def _invoke_with_hooks(
         self._fire_post_run_hooks(
             self._build_post_run_payload(
                 tag, data, output_data, started_at, finished_at,
-                InvocationStatus.COMPUTED, None,
+                InvocationStatus.COMPUTED, None, run_id=run_id,
             )
         )
         return out_tag, output_data
@@ -455,7 +458,8 @@ async def _async_invoke_with_hooks(
             data: The input data to process.
             logger: Optional data execution logger forwarded to
                 ``async_process_data``.
-            run_id: Pipeline run identifier forwarded to ``async_process_data``.
+            run_id: Pipeline run identifier forwarded to ``async_process_data``
+                and ``PostRunPayload.invocation_context``.
 
         Returns:
             A ``(tag, output_data)`` tuple.
@@ -476,7 +480,7 @@ async def _async_invoke_with_hooks(
             self._fire_post_run_hooks(
                 self._build_post_run_payload(
                     tag, data, None, started_at, finished_at,
-                    InvocationStatus.ERROR, exc,
+                    InvocationStatus.ERROR, exc, run_id=run_id,
                 )
             )
             raise  # bare raise — preserves the original traceback exactly
@@ -485,7 +489,7 @@ async def _async_invoke_with_hooks(
         self._fire_post_run_hooks(
             self._build_post_run_payload(
                 tag, data, output_data, started_at, finished_at,
-                InvocationStatus.COMPUTED, None,
+                InvocationStatus.COMPUTED, None, run_id=run_id,
             )
         )
         return out_tag, output_data

src/orcapod/hooks.py

@@ -14,6 +14,8 @@
 from enum import Enum
 from typing import TYPE_CHECKING, Literal
 
+from orcapod.invocation import InvocationContext
+
 if TYPE_CHECKING:
     from orcapod.protocols.core_protocols import DataProtocol, TagProtocol
 
@@ -82,6 +84,12 @@ class PostRunPayload:
         output: The output data; ``None`` if the function filtered the row out or raised.
         stats: Timing and status bundle.
         pod: Identity of the pod that produced this result.
+        invocation_context: Deterministic invocation identity for this row.
+            Carries ``invocation_hash`` (input-keyed, encoding-configurable),
+            ``format_id(config)`` for custom serialization, and
+            ``pipeline_run_id``. ``None`` only when ``PostRunPayload`` is
+            constructed directly without providing this argument (e.g. in
+            tests); always populated when built by the pod itself.
     """
 
     record_id_hash: str | None
@@ -90,6 +98,7 @@ class PostRunPayload:
     output: DataProtocol | None
     stats: RunStats
     pod: PodContext
+    invocation_context: InvocationContext | None = None
 
 
 PostRunHookFn = Callable[["PostRunPayload"], None]

src/orcapod/invocation.py

@@ -0,0 +1,127 @@
+# src/orcapod/invocation.py
+"""Invocation identity types for orcapod pipeline elements.
+
+Provides ``InvocationHashConfig``, ``InvocationContext``, and the internal
+``_serialize_component`` helper. These types are shared between
+``side_effects.py`` and ``hooks.py`` and are extracted here to avoid
+circular imports.
+"""
+
+from __future__ import annotations
+
+import base64
+import dataclasses
+from typing import TYPE_CHECKING, Literal
+
+if TYPE_CHECKING:
+    from orcapod.types import ContentHash
+
+
+@dataclasses.dataclass(frozen=True)
+class InvocationHashConfig:
+    """Controls how ``InvocationContext.invocation_hash`` is serialized.
+
+    Args:
+        encoding: Output encoding — ``"hex"`` (default) or ``"base64"``.
+        component_length: Bytes of raw digest to use per component. ``None``
+            means full digest length. Applied identically to every
+            ``::``-separated component.
+    """
+
+    encoding: Literal["hex", "base64"] = "hex"
+    component_length: int | None = None
+
+
+def _serialize_component(content_hash: ContentHash, config: InvocationHashConfig) -> str:
+    """Serialize one ``ContentHash`` component per ``InvocationHashConfig``.
+
+    The method name is always included as a prefix (e.g. ``"arrow_v2.1:abcd1234"``).
+    Only the digest bytes are subject to truncation via ``component_length``.
+
+    Args:
+        content_hash: The hash to serialize.
+        config: Encoding and truncation config.
+
+    Returns:
+        A string of the form ``"{method}:{encoded_digest}"`` where the digest
+        is optionally truncated then encoded as hex or base64.
+    """
+    raw = content_hash.digest
+    if config.component_length is not None:
+        raw = raw[:config.component_length]
+    if config.encoding == "base64":
+        encoded = base64.b64encode(raw).decode("ascii")
+    else:
+        encoded = raw.hex()
+    return f"{content_hash.method}:{encoded}"
+
+
+class InvocationContext:
+    """Per-invocation context describing a single pod call.
+
+    Carries a deterministic ``invocation_hash`` and metadata about the
+    current delivery. ``invocation_hash`` is a computed property that
+    delegates to ``format_id()`` with the pod's default
+    ``InvocationHashConfig``. ``format_id()`` can be called with a custom
+    config to re-serialize without recomputation.
+
+    Public fields are read-only by convention (no public setters).
+
+    Available on:
+    - Side-effect pod functions (injected as ``ctx`` argument).
+    - Function pod post-run hooks (via ``PostRunPayload.invocation_context``).
+
+    Args:
+        pod_name: ``pod.label`` of the invoking pod.
+        pipeline_run_id: The current pipeline run identifier, or ``None``
+            for standalone / lazy pipelines.
+        _pipeline_hash_ch: Internal. Pipeline-level ``ContentHash`` component.
+        _record_id_hash_ch: Internal. Record-level ``ContentHash`` component.
+        _hash_config: Internal. Default encoding config for ``format_id``.
+        _track_completion: Internal. When ``True`` omits ``pipeline_run_id``
+            from the hash string.
+    """
+
+    def __init__(
+        self,
+        pod_name: str,
+        pipeline_run_id: str | None,
+        _pipeline_hash_ch: ContentHash,
+        _record_id_hash_ch: ContentHash,
+        _hash_config: InvocationHashConfig,
+        _track_completion: bool,
+    ) -> None:
+        self.pod_name = pod_name
+        self.pipeline_run_id = pipeline_run_id
+        self._pipeline_hash_ch = _pipeline_hash_ch
+        self._record_id_hash_ch = _record_id_hash_ch
+        self._hash_config = _hash_config
+        self._track_completion = _track_completion
+
+    @property
+    def invocation_hash(self) -> str:
+        """Serialized invocation hash — delegates to ``format_id()``."""
+        return self.format_id()
+
+    def format_id(self, config: InvocationHashConfig | None = None) -> str:
+        """Return the invocation hash string with an optional format override.
+
+        Serializes the stored ``ContentHash`` components. Uses ``config``
+        if supplied, otherwise the pod's own ``InvocationHashConfig``.
+
+        Args:
+            config: Optional encoding/truncation override.
+
+        Returns:
+            A string of the form ``"{component1}::{component2}"``
+            (two components when ``track_completion=True``) or
+            ``"{c1}::{c2}::{run_id}"`` (three components when
+            ``track_completion=False`` and ``pipeline_run_id`` is not ``None``).
+            Each component is ``"{method}:{encoded_digest}"``.
+        """
+        cfg = config or self._hash_config
+        c1 = _serialize_component(self._pipeline_hash_ch, cfg)
+        c2 = _serialize_component(self._record_id_hash_ch, cfg)
+        if not self._track_completion and self.pipeline_run_id is not None:
+            return f"{c1}::{c2}::{self.pipeline_run_id}"
+        return f"{c1}::{c2}"