fix(side-effects): add SideEffectNode blueprint class and fix record_id uniqueness

Introduce a lightweight SideEffectNode (blueprint) class so that Pipeline
uses schema-only nodes without DB state, matching the FunctionNode /
FunctionJobNode split. SideEffectJobNode now extends SideEffectNode and
inherits node_uri, making Pipeline.save() and PipelineJob.as_pipeline() /
to_invocations() work correctly on pipelines containing SideEffectPod.

Also append uuid4() to the _write_invocation_row SHA-256 input to eliminate
same-microsecond record_id collisions.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: kurodo3[bot]

src/orcapod/pipeline/base.py

@@ -15,6 +15,7 @@
     PodInvocation,
     SideEffectInvocation,
 )
+from orcapod.side_effects import SideEffectNode
 from orcapod.protocols import core_protocols as cp
 from orcapod.utils.lazy_module import LazyModule
 
@@ -536,6 +537,12 @@ def to_invocations(self) -> InvocationGraph:
                     input_streams=(node.upstreams[0],),
                     label=node._label,
                 )
+            elif isinstance(node, SideEffectNode):
+                inv_by_node_hash[node_hash] = SideEffectInvocation(
+                    pod=node._pod,
+                    input_streams=(node.upstreams[0],),
+                    label=node._label,
+                )
             else:
                 if node._operator is None:
                     raise RuntimeError(

src/orcapod/pipeline/graph.py

@@ -12,7 +12,7 @@
     OperatorNode,
     SourceNode,
 )
-from orcapod.side_effects import SideEffectJobNode
+from orcapod.side_effects import SideEffectNode
 from orcapod.core.tracker import AutoRegisteringContextBasedTracker
 from orcapod.pipeline.base import AbstractPipelineBase
 from orcapod.pipeline.dag import OrcaDAG
@@ -65,7 +65,7 @@ class Pipeline(AbstractPipelineBase[GraphNode]):
     source_node_class = SourceNode
     function_node_class = FunctionNode
     operator_node_class = OperatorNode
-    side_effect_node_class = SideEffectJobNode
+    side_effect_node_class = SideEffectNode
 
     def __init__(
         self,

src/orcapod/pipeline/job.py

@@ -468,6 +468,7 @@ def as_pipeline(self) -> "Pipeline":
         from orcapod.core.nodes.function_node import FunctionNode
         from orcapod.core.nodes.operator_node import OperatorNode
         from orcapod.pipeline.graph import Pipeline
+        from orcapod.side_effects import SideEffectNode
 
         if not self._compiled:
             raise RuntimeError(
@@ -519,6 +520,14 @@ def as_pipeline(self) -> "Pipeline":
                     tracker_manager=job_node.tracker_manager,
                 )
 
+            elif isinstance(job_node, SideEffectJobNode):
+                upstream_bp_hash = job_id_to_bp_hash[id(job_node._input_stream)]
+                node_map[node_hash] = SideEffectNode(
+                    side_effect_pod=job_node._pod,
+                    input_stream=node_map[upstream_bp_hash],
+                    label=job_node._label,
+                )
+
             else:
                 node_map[node_hash] = job_node.as_node()
 

src/orcapod/side_effects.py

@@ -13,6 +13,7 @@
 import datetime
 import hashlib
 import logging
+import uuid
 from collections.abc import Callable, Collection, Iterator, Sequence
 from typing import TYPE_CHECKING, Any, Literal
 
@@ -418,7 +419,7 @@ def _write_invocation_row(
     executed_at = datetime.datetime.now(datetime.timezone.utc)
     executed_at_str = executed_at.isoformat()
     record_id_src = (
-        f"{fip_hash_str}::{pod_content_hash_str}::{run_id}::{status}::{executed_at_str}"
+        f"{fip_hash_str}::{pod_content_hash_str}::{run_id}::{status}::{executed_at_str}::{uuid.uuid4()}"
     ).encode("utf-8")
     record_id = hashlib.sha256(record_id_src).digest()
 
@@ -711,20 +712,16 @@ def _merge_config(
 
 
 # ---------------------------------------------------------------------------
-# SideEffectJobNode
+# SideEffectNode — lightweight blueprint node (no DB)
 # ---------------------------------------------------------------------------
 
 
-class SideEffectJobNode(StreamBase):
-    """DB-backed execution node for side-effect pods.
-
-    Created at pipeline compile time by ``PipelineJob``. Receives a
-    ``pipeline_database`` via ``attach_databases()``. ``run_id`` is passed
-    as a call-time keyword argument from the orchestrator.
+class SideEffectNode(StreamBase):
+    """Lightweight blueprint node for side-effect pods.
 
-    Inherits from ``StreamBase`` for identity infrastructure and to satisfy
-    the ``producer`` / ``upstreams`` / ``output_schema`` contract required
-    by ``SyncPipelineOrchestrator._materialize_as_stream``.
+    Used by ``Pipeline`` (the blueprint) to represent a side-effect pod
+    invocation without any DB attachment or execution logic. Analogous to
+    ``FunctionNode`` in the function pod hierarchy.
 
     Args:
         side_effect_pod: The ``SideEffectPod`` this node wraps.
@@ -743,8 +740,6 @@ def __init__(
         self._pod = side_effect_pod
         self._input_stream = input_stream
         super().__init__(label=label)
-        self._pipeline_database: ArrowDatabaseProtocol | None = None
-        self._table_path: tuple[str, ...] | None = None
 
     # ------------------------------------------------------------------
     # StreamBase interface
@@ -808,14 +803,7 @@ def as_table(
         columns: ColumnConfig | dict[str, Any] | None = None,
         all_info: bool = False,
     ) -> pa.Table:
-        """Collect all rows from ``iter_data()`` into an Arrow table.
-
-        Warning:
-            Calling ``as_table()`` on a ``SideEffectJobNode`` iterates via
-            ``iter_data()``, which re-invokes the side-effect function for each
-            row with no DB logging and no ``run_id``. Use ``execute()`` for
-            orchestrated execution.
-        """
+        """Collect all rows from ``iter_data()`` into an Arrow table."""
         from orcapod.types import ColumnConfig as _ColumnConfig
         from orcapod.utils import arrow_utils
 
@@ -832,6 +820,47 @@ def as_table(
             pa.concat_tables(data_tables),
         )
 
+    @property
+    def node_uri(self) -> tuple[str, ...]:
+        """Canonical URI tuple identifying this side-effect node.
+
+        Returns:
+            A tuple of the form ``("side_effect", label, content_hash_string)``.
+        """
+        return ("side_effect", self._pod.label, self._pod.content_hash().to_string())
+
+
+# ---------------------------------------------------------------------------
+# SideEffectJobNode — DB-backed execution node
+# ---------------------------------------------------------------------------
+
+
+class SideEffectJobNode(SideEffectNode):
+    """DB-backed execution node for side-effect pods.
+
+    Created at pipeline compile time by ``PipelineJob``. Receives a
+    ``pipeline_database`` via ``attach_databases()``. ``run_id`` is passed
+    as a call-time keyword argument from the orchestrator.
+
+    Extends ``SideEffectNode`` with DB attachment and orchestrated execution
+    methods. Analogous to ``FunctionJobNode`` in the function pod hierarchy.
+
+    Args:
+        side_effect_pod: The ``SideEffectPod`` this node wraps.
+        input_stream: The upstream stream at compile time.
+        label: Optional display label.
+    """
+
+    def __init__(
+        self,
+        side_effect_pod: SideEffectPod,
+        input_stream: StreamProtocol,
+        label: str | None = None,
+    ) -> None:
+        super().__init__(side_effect_pod=side_effect_pod, input_stream=input_stream, label=label)
+        self._pipeline_database: ArrowDatabaseProtocol | None = None
+        self._table_path: tuple[str, ...] | None = None
+
     # ------------------------------------------------------------------
     # DB attachment
     # ------------------------------------------------------------------