From 586190e427a9090a3b719c6b559de6affe4793a3 Mon Sep 17 00:00:00 2001 From: "agent-kurodo[bot]" <268466204+agent-kurodo[bot]@users.noreply.github.com> Date: Tue, 21 Jul 2026 03:15:36 +0000 Subject: [PATCH 1/9] docs(specs): add design spec for ITL-531 PostRunPayload InvocationContext alignment --- ...ostrunpayload-invocation-context-design.md | 190 ++++++++++++++++++ 1 file changed, 190 insertions(+) create mode 100644 superpowers/specs/2026-07-21-postrunpayload-invocation-context-design.md diff --git a/superpowers/specs/2026-07-21-postrunpayload-invocation-context-design.md b/superpowers/specs/2026-07-21-postrunpayload-invocation-context-design.md new file mode 100644 index 00000000..2abc72ab --- /dev/null +++ b/superpowers/specs/2026-07-21-postrunpayload-invocation-context-design.md @@ -0,0 +1,190 @@ +# Design: Align PostRunPayload with SideEffectPod InvocationContext + +**Issue:** ITL-531 +**Date:** 2026-07-21 +**Status:** Approved + +--- + +## Overview + +`PostRunPayload` (in `hooks.py`) is the snapshot passed to every post-run hook +after a `FunctionPod` processes one row. It currently carries the output-keyed +`record_id_hash` (the output datagram UUID) but has no deterministic input-keyed +hash, no `pipeline_run_id`, and no formatting helper. + +`InvocationContext` (in `side_effects.py`) already provides all three — an +`invocation_hash` property, a `format_id(config)` method, and `pipeline_run_id` +— built from `pipeline_hash :: record_id_hash` over the input row. Post-run hook +authors currently cannot do equivalent provenance logging or external-system +stamping. + +This design adds `invocation_context: InvocationContext | None` to +`PostRunPayload`, making the same identification primitives available to all +function pod hooks. The change is purely additive — no existing consumers break. + +--- + +## Goals & Success Criteria + +- `PostRunPayload` exposes `invocation_context: InvocationContext | None`, + populated on every payload produced by the pod (never `None` in practice when + built via `_build_post_run_payload`). +- Hook authors can call `payload.invocation_context.invocation_hash`, + `.format_id(config)`, and `.pipeline_run_id` with identical semantics to + `SideEffectPod` / `FunctionPod` ctx-injection. +- `InvocationHashConfig` and `InvocationContext` live in a shared leaf module + (`src/orcapod/invocation.py`) with no orcapod-internal imports at runtime + (only `TYPE_CHECKING` guards), making them importable by both `hooks.py` and + `side_effects.py` without circular-import risk. +- All existing `PostRunPayload` consumers and current tests continue to pass + unmodified (additive field with `= None` default). + +--- + +## Architecture + +### New module: `src/orcapod/invocation.py` + +A new leaf module housing the invocation-identity types. These are extracted +verbatim from `side_effects.py` with no behavioral changes. + +**Contents:** +- `InvocationHashConfig` — encoding (`hex`/`base64`) and truncation config +- `_serialize_component(content_hash, config) -> str` — internal serialization helper +- `InvocationContext` — carries `pod_name`, `pipeline_run_id`, + `_pipeline_hash_ch`, `_record_id_hash_ch`, `_hash_config`, `_track_completion`; + exposes `invocation_hash` property and `format_id(config=None) -> str` + +**Runtime imports:** stdlib only (`base64`, `dataclasses`, typing). `ContentHash` +is import-guarded under `TYPE_CHECKING`. + +### Changes to `src/orcapod/side_effects.py` + +Remove the definitions of `InvocationHashConfig`, `_serialize_component`, and +`InvocationContext`. Replace with: + +```python +from orcapod.invocation import ( + InvocationContext, + InvocationHashConfig, + _serialize_component, +) +``` + +All internal uses remain unchanged. + +### Changes to `src/orcapod/hooks.py` + +Add import: +```python +from orcapod.invocation import InvocationContext +``` + +Add field to `PostRunPayload`: +```python +invocation_context: InvocationContext | None = None +``` + +Place after `pod: PodContext` so existing positional construction (if any) is +unaffected. Update class docstring. + +### Changes to `src/orcapod/core/function_pod.py` + +**`_build_post_run_payload`** gains `run_id: str | None = None` parameter. +It calls `self._build_invocation_context(tag, data, run_id=run_id)` and passes +the result as `invocation_context=` in the `PostRunPayload` constructor. + +**`_invoke_with_hooks`** and **`_async_invoke_with_hooks`** already carry +`run_id`; both calls to `_build_post_run_payload` are updated to pass `run_id`. + +**`_build_invocation_context`** local import switches from `orcapod.side_effects` +to `orcapod.invocation` for `InvocationContext` and `InvocationHashConfig`. +`_SIDE_EFFECT_RECOMPUTATION_INDEX_COL` stays imported from `orcapod.side_effects` +(it's a constant about the side-effect preimage format, not the invocation type). + +The `InvocationContext` is built with `_track_completion=True` and default +`InvocationHashConfig()`, matching the existing ctx-injection path. This produces +a 2-component hash (`pipeline_hash :: record_id_hash`). `pipeline_run_id` is +always stored and accessible regardless. + +### Changes to `src/orcapod/__init__.py` + +Update the import of `InvocationContext` and `InvocationHashConfig` from +`.side_effects` to `.invocation`. The exported symbols and their behavior are +identical. + +--- + +## Data Flow + +``` +_invoke_with_hooks(tag, data, run_id=run_id) + │ + ├─► process_data(tag, data, run_id=run_id) + │ └─► [if ctx_arg_name set] _build_invocation_context(tag, data, run_id) + │ └─► InvocationContext injected into user fn + │ + └─► _build_post_run_payload(tag, data, output, ..., run_id=run_id) + └─► _build_invocation_context(tag, data, run_id) + └─► InvocationContext attached to PostRunPayload.invocation_context +``` + +Note: `_build_invocation_context` is called twice for ctx-aware pods (once for +the function injection, once for the payload). The cost is small (one PyArrow +table hash) and only incurred when hooks are registered. + +--- + +## Import Graph (after change) + +``` +invocation.py ← stdlib only + ↑ ↑ +hooks.py side_effects.py + ↑ ↑ + └─ function_pod.py ──────────► invocation.py (direct, replaces local import) +``` + +No cycles. + +--- + +## Error Handling + +`_build_invocation_context` is called unconditionally in `_build_post_run_payload`, +including the error path in `_invoke_with_hooks`. The `tag` and `data` arguments +are always available even when the pod function raised, so the `InvocationContext` +is always constructible. If `_build_invocation_context` itself raises (e.g. due +to a corrupt hasher), the original exception semantics are unchanged — the new +exception would propagate as a hook-phase error. + +--- + +## Testing + +Add a new test class `TestInvocationContextOnPayload` in +`tests/test_core/function_pod/test_post_run_hooks.py`: + +| Test | What it checks | +|------|----------------| +| `test_invocation_context_always_present` | `invocation_context` is not `None` on COMPUTED payload | +| `test_invocation_hash_is_nonempty_string` | `invocation_hash` is a non-empty string | +| `test_format_id_matches_invocation_hash` | `format_id()` == `invocation_hash` | +| `test_format_id_base64_differs_from_hex` | `format_id(InvocationHashConfig(encoding="base64"))` ≠ `invocation_hash` | +| `test_pipeline_run_id_is_none_standalone` | `pipeline_run_id` is `None` in standalone mode | +| `test_invocation_hash_deterministic` | same input twice → same `invocation_hash` | +| `test_error_payload_has_invocation_context` | ERROR-status payload also has `invocation_context` set | + +Additionally verify that existing tests continue to pass without modification +(the `= None` default handles direct `PostRunPayload` construction in tests). + +--- + +## Out of Scope + +- Pre-run hooks or operator-pod hooks +- Async hook execution changes +- Changing the `SideEffectPod` / `InvocationContext` API +- Exposing a `pipeline_run_id` field *directly* on `PostRunPayload` (accessible + via `payload.invocation_context.pipeline_run_id`) From 29f7cef292bda65dd8a7f2641ae442e9183e5663 Mon Sep 17 00:00:00 2001 From: "agent-kurodo[bot]" <268466204+agent-kurodo[bot]@users.noreply.github.com> Date: Tue, 21 Jul 2026 03:24:12 +0000 Subject: [PATCH 2/9] docs(plans): add implementation plan for ITL-531 PostRunPayload InvocationContext --- ...07-21-postrunpayload-invocation-context.md | 810 ++++++++++++++++++ 1 file changed, 810 insertions(+) create mode 100644 superpowers/plans/2026-07-21-postrunpayload-invocation-context.md diff --git a/superpowers/plans/2026-07-21-postrunpayload-invocation-context.md b/superpowers/plans/2026-07-21-postrunpayload-invocation-context.md new file mode 100644 index 00000000..a76de822 --- /dev/null +++ b/superpowers/plans/2026-07-21-postrunpayload-invocation-context.md @@ -0,0 +1,810 @@ +# PostRunPayload InvocationContext Alignment Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use sensei:subagent-driven-development (recommended) or sensei:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Add `invocation_context: InvocationContext | None` to `PostRunPayload` so post-run hook authors have access to the same deterministic invocation hash, `format_id()`, and `pipeline_run_id` that `SideEffectPod` functions already receive. + +**Architecture:** Extract `InvocationContext`/`InvocationHashConfig` from `side_effects.py` into a new stdlib-only leaf module `invocation.py`. Add the field to `PostRunPayload` (with `= None` default for backward compat). Thread `run_id` into `_build_post_run_payload` and call `_build_invocation_context()` there so every hook payload carries a fully-populated context. + +**Tech Stack:** Python 3.11+, PyArrow (for preimage hashing in `_build_invocation_context`), pytest + uv. + +**Issue:** ITL-531 +**Spec:** `superpowers/specs/2026-07-21-postrunpayload-invocation-context-design.md` +**Branch:** `eywalker/itl-531-align-postrunpayload-with-sideeffectpod-invocationcontext` + +--- + +## File Map + +| Action | File | Purpose | +|--------|------|---------| +| **Create** | `src/orcapod/invocation.py` | New leaf module: `InvocationHashConfig`, `_serialize_component`, `InvocationContext` | +| **Modify** | `src/orcapod/side_effects.py` | Remove moved definitions; import from `invocation.py` | +| **Modify** | `src/orcapod/hooks.py` | Import `InvocationContext`; add `invocation_context` field to `PostRunPayload` | +| **Modify** | `src/orcapod/core/function_pod.py` | Thread `run_id` into `_build_post_run_payload`; build and attach `InvocationContext` | +| **Modify** | `src/orcapod/core/cached_function_pod.py` | Pass `run_id` to `_build_post_run_payload` in overridden hook methods | +| **Modify** | `src/orcapod/__init__.py` | Import `InvocationContext`/`InvocationHashConfig` from `.invocation` instead of `.side_effects` | +| **Modify** | `tests/test_core/function_pod/test_post_run_hooks.py` | Add `TestInvocationContextOnPayload` test class | + +--- + +## Task 1: Create `src/orcapod/invocation.py` leaf module + +**Files:** +- Create: `src/orcapod/invocation.py` + +This task extracts the three invocation-identity items from `side_effects.py` into a new module with no orcapod-internal runtime imports. No behavioral change — only relocation. + +- [ ] **Step 1: Create `src/orcapod/invocation.py`** + +```python +# src/orcapod/invocation.py +"""Invocation identity types for orcapod pipeline elements. + +Provides ``InvocationHashConfig``, ``InvocationContext``, and the internal +``_serialize_component`` helper. These types describe how a single pod +invocation is identified — independent of whether the pod is a side-effect +pod or a function pod. Extracted here so both ``hooks.py`` and +``side_effects.py`` can import them without circular-import risk. +""" + +from __future__ import annotations + +import base64 +import dataclasses +from typing import TYPE_CHECKING, Literal + +if TYPE_CHECKING: + from orcapod.types import ContentHash + + +# --------------------------------------------------------------------------- +# InvocationHashConfig +# --------------------------------------------------------------------------- + + +@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}" + + +# --------------------------------------------------------------------------- +# InvocationContext +# --------------------------------------------------------------------------- + + +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. + """ + + 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}" +``` + +- [ ] **Step 2: Update `side_effects.py` — remove moved definitions, import from `invocation.py`** + +In `src/orcapod/side_effects.py`, replace the three blocks (lines 50–185, which define `InvocationHashConfig`, `_serialize_component`, and `InvocationContext`) with a single import. The replacement goes right after the `_SIDE_EFFECT_RECOMPUTATION_INDEX_COL` constant definition. + +Remove these sections entirely from `side_effects.py`: +- The `InvocationHashConfig` dataclass (lines ~51–63) +- The `_serialize_component` function (lines ~65–86) +- The `InvocationContext` class (lines ~126–185) +- Their section comment headers (`# InvocationHashConfig`, `# InvocationContext`) + +Add this import block after the `_SIDE_EFFECT_RECOMPUTATION_INDEX_COL` line: + +```python +from orcapod.invocation import ( + InvocationContext, + InvocationHashConfig, + _serialize_component, +) +``` + +- [ ] **Step 3: Update `function_pod.py` — switch local import to `invocation.py`** + +In `src/orcapod/core/function_pod.py`, find the local import block inside `_build_invocation_context` (around line 185–189): + +```python + from orcapod.side_effects import ( + InvocationContext, + InvocationHashConfig, + _SIDE_EFFECT_RECOMPUTATION_INDEX_COL, + ) +``` + +Replace with: + +```python + from orcapod.invocation import InvocationContext, InvocationHashConfig + from orcapod.side_effects import _SIDE_EFFECT_RECOMPUTATION_INDEX_COL +``` + +- [ ] **Step 4: Update `__init__.py` — import from `.invocation` instead of `.side_effects`** + +In `src/orcapod/__init__.py`, change: + +```python +from .side_effects import ( + InvocationContext, + InvocationHashConfig, + SideEffectPod, + SideEffectPodConfig, + side_effect_pod, + sink_pod, + tap_pod, +) +``` + +To: + +```python +from .invocation import ( + InvocationContext, + InvocationHashConfig, +) +from .side_effects import ( + SideEffectPod, + SideEffectPodConfig, + side_effect_pod, + sink_pod, + tap_pod, +) +``` + +- [ ] **Step 5: Run existing tests to verify nothing broke** + +```bash +uv run pytest tests/ -x -q +``` + +Expected: all existing tests pass. If any fail, they indicate a missed import update — fix before proceeding. + +- [ ] **Step 6: Commit** + +```bash +git add src/orcapod/invocation.py src/orcapod/side_effects.py src/orcapod/core/function_pod.py src/orcapod/__init__.py +git commit -m "refactor(invocation): extract InvocationContext/InvocationHashConfig to leaf module (ITL-531)" +``` + +--- + +## Task 2: Write failing tests for `PostRunPayload.invocation_context` + +**Files:** +- Modify: `tests/test_core/function_pod/test_post_run_hooks.py` + +- [ ] **Step 1: Add `TestInvocationContextOnPayload` test class** + +Append this class to the end of `tests/test_core/function_pod/test_post_run_hooks.py`. Add `from orcapod import InvocationHashConfig` to the existing imports at the top of the file. + +```python +# --------------------------------------------------------------------------- +# InvocationContext on PostRunPayload (ITL-531) +# --------------------------------------------------------------------------- + + +class TestInvocationContextOnPayload: + def test_invocation_context_always_present(self): + pod = _make_double_pod() + payloads: list[PostRunPayload] = [] + pod.add_post_run_hook(payloads.append) + + stream = pod.process(_make_stream(n=1)) + list(stream.iter_data()) + + assert payloads[0].invocation_context is not None + + def test_invocation_hash_is_nonempty_string(self): + pod = _make_double_pod() + payloads: list[PostRunPayload] = [] + pod.add_post_run_hook(payloads.append) + + stream = pod.process(_make_stream(n=1)) + list(stream.iter_data()) + + ctx = payloads[0].invocation_context + assert isinstance(ctx.invocation_hash, str) + assert len(ctx.invocation_hash) > 0 + + def test_format_id_matches_invocation_hash(self): + pod = _make_double_pod() + payloads: list[PostRunPayload] = [] + pod.add_post_run_hook(payloads.append) + + stream = pod.process(_make_stream(n=1)) + list(stream.iter_data()) + + ctx = payloads[0].invocation_context + assert ctx.format_id() == ctx.invocation_hash + + def test_format_id_base64_differs_from_hex(self): + pod = _make_double_pod() + payloads: list[PostRunPayload] = [] + pod.add_post_run_hook(payloads.append) + + stream = pod.process(_make_stream(n=1)) + list(stream.iter_data()) + + ctx = payloads[0].invocation_context + hex_id = ctx.invocation_hash + b64_id = ctx.format_id(InvocationHashConfig(encoding="base64")) + assert hex_id != b64_id + + def test_pipeline_run_id_is_none_standalone(self): + pod = _make_double_pod() + payloads: list[PostRunPayload] = [] + pod.add_post_run_hook(payloads.append) + + stream = pod.process(_make_stream(n=1)) + list(stream.iter_data()) + + assert payloads[0].invocation_context.pipeline_run_id is None + + def test_invocation_hash_deterministic(self): + pod = _make_double_pod() + payloads: list[PostRunPayload] = [] + pod.add_post_run_hook(payloads.append) + + # Process the same input stream twice + stream1 = pod.process(_make_stream(n=1)) + list(stream1.iter_data()) + stream2 = pod.process(_make_stream(n=1)) + list(stream2.iter_data()) + + assert len(payloads) == 2 + assert ( + payloads[0].invocation_context.invocation_hash + == payloads[1].invocation_context.invocation_hash + ) + + def test_error_payload_has_invocation_context(self): + def explodes(x: int) -> int: + raise RuntimeError("boom") + + pf = PythonDataFunction(explodes, output_keys="result") + pod = FunctionPod(pf) + payloads: list[PostRunPayload] = [] + pod.add_post_run_hook(HookConfig(fn=payloads.append, on_error="log")) + + stream = pod.process(_make_stream(n=1)) + with pytest.raises(RuntimeError, match="boom"): + list(stream.iter_data()) + + assert len(payloads) == 1 + assert payloads[0].invocation_context is not None + assert isinstance(payloads[0].invocation_context.invocation_hash, str) +``` + +- [ ] **Step 2: Run tests to confirm they fail** + +```bash +uv run pytest tests/test_core/function_pod/test_post_run_hooks.py::TestInvocationContextOnPayload -v +``` + +Expected: All 7 tests FAIL. The first failure should be `AttributeError: 'PostRunPayload' object has no attribute 'invocation_context'` (or similar). If any pass unexpectedly, investigate before proceeding. + +--- + +## Task 3: Add `invocation_context` field to `PostRunPayload` + +**Files:** +- Modify: `src/orcapod/hooks.py` + +- [ ] **Step 1: Add import and field to `PostRunPayload`** + +In `src/orcapod/hooks.py`, add the import after the existing imports: + +```python +from __future__ import annotations + +import dataclasses +from collections.abc import Callable +from datetime import datetime +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 +``` + +Then update `PostRunPayload` to add the new field. The field goes after `pod` with a default of `None` so existing positional and keyword construction still works: + +```python +@dataclasses.dataclass(frozen=True) +class PostRunPayload: + """Payload passed to every post-run hook after a pod invocation. + + Attributes: + record_id_hash: String form of ``output.datagram_uuid`` — the same UUID + used as the primary key when the result is stored in a backing database. + ``None`` when ``output`` is ``None`` (filtered row or error). + tag: The input tag for this invocation. Treat as read-only. + input: The input data for this invocation. Treat as read-only. + 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 + tag: TagProtocol + input: DataProtocol + output: DataProtocol | None + stats: RunStats + pod: PodContext + invocation_context: InvocationContext | None = None +``` + +- [ ] **Step 2: Run the new tests — expect partial failure** + +```bash +uv run pytest tests/test_core/function_pod/test_post_run_hooks.py::TestInvocationContextOnPayload -v +``` + +Expected: Tests fail because `invocation_context` is always `None` (not yet populated by the pod). The assertion `assert payloads[0].invocation_context is not None` should now be the failure point rather than an `AttributeError`. + +- [ ] **Step 3: Run the full existing test suite to confirm no regressions** + +```bash +uv run pytest tests/ -x -q --ignore=tests/test_core/function_pod/test_post_run_hooks.py +``` + +Expected: All pass. `PostRunPayload`'s `= None` default means no existing construction sites break. + +--- + +## Task 4: Thread `run_id` into `_build_post_run_payload` and populate `invocation_context` + +**Files:** +- Modify: `src/orcapod/core/function_pod.py` + +- [ ] **Step 1: Update `_build_post_run_payload` signature and body** + +Find `_build_post_run_payload` in `_FunctionPodBase` (around line 352). Replace the entire method: + +```python + def _build_post_run_payload( + self, + tag: TagProtocol, + data: DataProtocol, + output_data: DataProtocol | None, + started_at: datetime, + finished_at: datetime, + status: InvocationStatus, + exc: Exception | None, + run_id: str | None = None, + ) -> PostRunPayload: + """Build a ``PostRunPayload`` from invocation results. + + Args: + tag: The input tag. + data: The input data. + output_data: The output data, or ``None`` if filtered or errored. + started_at: UTC timestamp when the invocation started. + 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, + input=data, + output=output_data, + stats=RunStats( + duration_ms=(finished_at - started_at).total_seconds() * 1000, + status=status, + started_at=started_at, + finished_at=finished_at, + error=exc, + ), + pod=PodContext( + label=self.label, + pod_hash=self.content_hash().to_string(), + ), + invocation_context=invocation_context, + ) +``` + +- [ ] **Step 2: Update `_invoke_with_hooks` to pass `run_id` to `_build_post_run_payload`** + +Find `_invoke_with_hooks` in `_FunctionPodBase` (around line 397). There are two calls to `_build_post_run_payload` inside it — the error path and the success path. Update both to pass `run_id=run_id`: + +```python + def _invoke_with_hooks( + self, + tag: TagProtocol, + data: DataProtocol, + *, + logger: DataExecutionLoggerProtocol | None = None, + run_id: str | None = None, + ) -> tuple[TagProtocol, DataProtocol | None]: + """Call ``process_data``, time it, and fire post-run hooks. + + When ``_post_run_hooks`` is empty, delegates directly to + ``process_data`` with zero overhead. Override in subclasses (e.g. + ``CachedFunctionPod``) to supply a different ``InvocationStatus``. + + Args: + 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`` and + ``PostRunPayload.invocation_context``. + + Returns: + A ``(tag, output_data)`` tuple. + """ + if not self._post_run_hooks: + return self.process_data(tag, data, logger=logger, run_id=run_id) + + started_at = datetime.now(timezone.utc) + out_tag = tag + output_data: DataProtocol | None = None + + try: + out_tag, output_data = self.process_data(tag, data, logger=logger, run_id=run_id) + except Exception as exc: + finished_at = datetime.now(timezone.utc) + self._fire_post_run_hooks( + self._build_post_run_payload( + tag, data, None, started_at, finished_at, + InvocationStatus.ERROR, exc, run_id=run_id, + ) + ) + raise # bare raise — preserves the original traceback exactly + + finished_at = datetime.now(timezone.utc) + self._fire_post_run_hooks( + self._build_post_run_payload( + tag, data, output_data, started_at, finished_at, + InvocationStatus.COMPUTED, None, run_id=run_id, + ) + ) + return out_tag, output_data +``` + +- [ ] **Step 3: Update `_async_invoke_with_hooks` to pass `run_id` to `_build_post_run_payload`** + +Find `_async_invoke_with_hooks` in `_FunctionPodBase` (around line 448). Update both `_build_post_run_payload` calls inside it to pass `run_id=run_id`: + +```python + async def _async_invoke_with_hooks( + self, + tag: TagProtocol, + data: DataProtocol, + *, + logger: DataExecutionLoggerProtocol | None = None, + run_id: str | None = None, + ) -> tuple[TagProtocol, DataProtocol | None]: + """Async counterpart of ``_invoke_with_hooks``. + + When ``_post_run_hooks`` is empty, delegates directly to + ``async_process_data`` with zero overhead. + + Args: + tag: The tag associated with the data. + 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`` + and ``PostRunPayload.invocation_context``. + + Returns: + A ``(tag, output_data)`` tuple. + """ + if not self._post_run_hooks: + return await self.async_process_data(tag, data, logger=logger, run_id=run_id) + + started_at = datetime.now(timezone.utc) + out_tag = tag + output_data: DataProtocol | None = None + + try: + out_tag, output_data = await self.async_process_data( + tag, data, logger=logger, run_id=run_id + ) + except Exception as exc: + finished_at = datetime.now(timezone.utc) + self._fire_post_run_hooks( + self._build_post_run_payload( + tag, data, None, started_at, finished_at, + InvocationStatus.ERROR, exc, run_id=run_id, + ) + ) + raise # bare raise — preserves the original traceback exactly + + finished_at = datetime.now(timezone.utc) + self._fire_post_run_hooks( + self._build_post_run_payload( + tag, data, output_data, started_at, finished_at, + InvocationStatus.COMPUTED, None, run_id=run_id, + ) + ) + return out_tag, output_data +``` + +- [ ] **Step 4: Run the new test class — all 7 tests should now pass** + +```bash +uv run pytest tests/test_core/function_pod/test_post_run_hooks.py::TestInvocationContextOnPayload -v +``` + +Expected output: +``` +tests/test_core/function_pod/test_post_run_hooks.py::TestInvocationContextOnPayload::test_invocation_context_always_present PASSED +tests/test_core/function_pod/test_post_run_hooks.py::TestInvocationContextOnPayload::test_invocation_hash_is_nonempty_string PASSED +tests/test_core/function_pod/test_post_run_hooks.py::TestInvocationContextOnPayload::test_format_id_matches_invocation_hash PASSED +tests/test_core/function_pod/test_post_run_hooks.py::TestInvocationContextOnPayload::test_format_id_base64_differs_from_hex PASSED +tests/test_core/function_pod/test_post_run_hooks.py::TestInvocationContextOnPayload::test_pipeline_run_id_is_none_standalone PASSED +tests/test_core/function_pod/test_post_run_hooks.py::TestInvocationContextOnPayload::test_invocation_hash_deterministic PASSED +tests/test_core/function_pod/test_post_run_hooks.py::TestInvocationContextOnPayload::test_error_payload_has_invocation_context PASSED + +7 passed +``` + +- [ ] **Step 5: Update `CachedFunctionPod._invoke_with_hooks` to pass `run_id`** + +`CachedFunctionPod` (in `src/orcapod/core/cached_function_pod.py`) overrides +`_invoke_with_hooks` and `_async_invoke_with_hooks`, calling `_build_post_run_payload` +without `run_id`. With the new `run_id` keyword argument added in Step 1, the code +still compiles (default `None`), but `pipeline_run_id` would silently be `None` +even when a run_id is in scope. Fix both calls in each override. + +In `CachedFunctionPod._invoke_with_hooks`, update the two `_build_post_run_payload` calls: + +```python + except Exception as exc: + finished_at = datetime.now(timezone.utc) + self._fire_post_run_hooks( + self._build_post_run_payload( + tag, data, None, started_at, finished_at, + InvocationStatus.ERROR, exc, run_id=run_id, + ) + ) + raise # bare raise — preserves the original traceback exactly + + finished_at = datetime.now(timezone.utc) + self._fire_post_run_hooks( + self._build_post_run_payload( + tag, data, output_data, started_at, finished_at, status, None, + run_id=run_id, + ) + ) +``` + +In `CachedFunctionPod._async_invoke_with_hooks`, make the same two changes: + +```python + except Exception as exc: + finished_at = datetime.now(timezone.utc) + self._fire_post_run_hooks( + self._build_post_run_payload( + tag, data, None, started_at, finished_at, + InvocationStatus.ERROR, exc, run_id=run_id, + ) + ) + raise # bare raise — preserves the original traceback exactly + + finished_at = datetime.now(timezone.utc) + self._fire_post_run_hooks( + self._build_post_run_payload( + tag, data, output_data, started_at, finished_at, status, None, + run_id=run_id, + ) + ) +``` + +- [ ] **Step 6: Run the full test suite** + +```bash +uv run pytest tests/ -x -q +``` + +Expected: All tests pass. Fix any failures before committing. + +- [ ] **Step 7: Commit** + +```bash +git add src/orcapod/hooks.py src/orcapod/core/function_pod.py src/orcapod/core/cached_function_pod.py tests/test_core/function_pod/test_post_run_hooks.py +git commit -m "feat(hooks): add invocation_context to PostRunPayload (ITL-531)" +``` + +--- + +## Task 5: Export `InvocationContext` from public API and finalize + +**Files:** +- Modify: `src/orcapod/__init__.py` (verify — already done in Task 1) +- Modify: `tests/test_core/function_pod/test_post_run_hooks.py` (add public API import check) + +- [ ] **Step 1: Verify `InvocationContext` is still exported from `orcapod`** + +```bash +uv run python -c "from orcapod import InvocationContext, InvocationHashConfig; print('OK')" +``` + +Expected: `OK` + +- [ ] **Step 2: Add public API import assertion to the existing `test_public_api_imports` test** + +In `tests/test_core/function_pod/test_post_run_hooks.py`, find `test_public_api_imports` in `TestDecoratorConvenience` and add two assertions: + +```python + def test_public_api_imports(self): + import orcapod + assert hasattr(orcapod, "PostRunPayload") + assert hasattr(orcapod, "HookConfig") + assert hasattr(orcapod, "InvocationStatus") + assert hasattr(orcapod, "RunStats") + assert hasattr(orcapod, "PodContext") + assert hasattr(orcapod, "InvocationContext") # ITL-531 + assert hasattr(orcapod, "InvocationHashConfig") # ITL-531 +``` + +- [ ] **Step 3: Run the updated test** + +```bash +uv run pytest tests/test_core/function_pod/test_post_run_hooks.py::TestDecoratorConvenience::test_public_api_imports -v +``` + +Expected: PASS + +- [ ] **Step 4: Run the complete test suite one final time** + +```bash +uv run pytest tests/ -q +``` + +Expected: All tests pass, no warnings about import paths. + +- [ ] **Step 5: Commit** + +```bash +git add tests/test_core/function_pod/test_post_run_hooks.py +git commit -m "test(hooks): verify InvocationContext/InvocationHashConfig in public API (ITL-531)" +``` + +--- + +## Task 6: Push branch and open PR + +- [ ] **Step 1: Push branch** + +```bash +git push -u origin eywalker/itl-531-align-postrunpayload-with-sideeffectpod-invocationcontext +``` + +- [ ] **Step 2: Open PR against `main`** + +```bash +gh pr create \ + --title "feat(hooks): align PostRunPayload with InvocationContext (ITL-531)" \ + --base main \ + --body "$(cat <<'EOF' +## Summary + +- Extracts `InvocationContext` and `InvocationHashConfig` out of `side_effects.py` into a new stdlib-only leaf module `src/orcapod/invocation.py` +- Adds `invocation_context: InvocationContext | None = None` to `PostRunPayload` — populated on every payload built by the pod +- Threads `run_id` through `_build_post_run_payload` so `pipeline_run_id` is accessible via the context + +Closes ITL-531 + +## Test plan + +- [ ] All existing tests pass unmodified (additive `= None` default) +- [ ] `TestInvocationContextOnPayload` (7 new tests) all pass +- [ ] `test_public_api_imports` verifies `InvocationContext` and `InvocationHashConfig` remain in the public API + +🤖 Generated with [Claude Code](https://claude.com/claude-code) +EOF +)" +``` From f2a26bf4c3ec065b87b52fcdbcb3006680e59d01 Mon Sep 17 00:00:00 2001 From: "agent-kurodo[bot]" <268466204+agent-kurodo[bot]@users.noreply.github.com> Date: Tue, 21 Jul 2026 03:40:31 +0000 Subject: [PATCH 3/9] refactor(invocation): extract InvocationContext/InvocationHashConfig to leaf module (ITL-531) Co-Authored-By: Claude Sonnet 4.6 --- src/orcapod/__init__.py | 4 +- src/orcapod/core/function_pod.py | 7 +- src/orcapod/invocation.py | 122 +++++++++++++++++++++++++++++++ src/orcapod/side_effects.py | 117 ++--------------------------- 4 files changed, 132 insertions(+), 118 deletions(-) create mode 100644 src/orcapod/invocation.py diff --git a/src/orcapod/__init__.py b/src/orcapod/__init__.py index ee6733a6..e5ea54e4 100644 --- a/src/orcapod/__init__.py +++ b/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, diff --git a/src/orcapod/core/function_pod.py b/src/orcapod/core/function_pod.py index 857484cc..5066d9c3 100644 --- a/src/orcapod/core/function_pod.py +++ b/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, diff --git a/src/orcapod/invocation.py b/src/orcapod/invocation.py new file mode 100644 index 00000000..eadcd4cb --- /dev/null +++ b/src/orcapod/invocation.py @@ -0,0 +1,122 @@ +# 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. + """ + + 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}" diff --git a/src/orcapod/side_effects.py b/src/orcapod/side_effects.py index 6ed2781e..6a7b2ea5 100644 --- a/src/orcapod/side_effects.py +++ b/src/orcapod/side_effects.py @@ -8,7 +8,6 @@ from __future__ import annotations import asyncio -import base64 import dataclasses import datetime import logging @@ -19,6 +18,11 @@ from orcapod.core.nodes.function_node import _build_record_id_preimage from orcapod.core.streams.base import StreamBase from orcapod.core.tracker import DEFAULT_TRACKER_MANAGER +from orcapod.invocation import ( + InvocationContext, + InvocationHashConfig, + _serialize_component, +) from orcapod.system_constants import TRACKING_DB_SCHEMA_VERSION from orcapod.utils.lazy_module import LazyModule @@ -44,50 +48,6 @@ _SIDE_EFFECT_RECOMPUTATION_INDEX_COL = "__pipeline_recomputation_index" -# --------------------------------------------------------------------------- -# InvocationHashConfig -# --------------------------------------------------------------------------- - - -@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}" - - # --------------------------------------------------------------------------- # SideEffectPodConfig # --------------------------------------------------------------------------- @@ -120,73 +80,6 @@ class SideEffectPodConfig: max_concurrency: int | None = 16 -# --------------------------------------------------------------------------- -# InvocationContext -# --------------------------------------------------------------------------- - - -class InvocationContext: - """Per-invocation context passed to every side-effect pod function. - - 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). - - Args: - pod_name: ``pod.label`` of the invoking pod. - pipeline_run_id: The current pipeline run identifier, or ``None`` - for standalone / lazy pipelines. - """ - - 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}" - - # --------------------------------------------------------------------------- # SideEffectPodStream # --------------------------------------------------------------------------- From 5f26134c2be2577b8128e6803786dccfe1250add Mon Sep 17 00:00:00 2001 From: "agent-kurodo[bot]" <268466204+agent-kurodo[bot]@users.noreply.github.com> Date: Tue, 21 Jul 2026 05:12:06 +0000 Subject: [PATCH 4/9] refactor(invocation): remove dead import, fix docstrings (ITL-531) Remove unused _serialize_component import from side_effects.py, update the module docstring to omit re-exported types, and add internal constructor parameters to InvocationContext.Args section. Co-Authored-By: Claude Sonnet 4.6 --- src/orcapod/invocation.py | 5 +++++ src/orcapod/side_effects.py | 5 ++--- 2 files changed, 7 insertions(+), 3 deletions(-) diff --git a/src/orcapod/invocation.py b/src/orcapod/invocation.py index eadcd4cb..c535ad31 100644 --- a/src/orcapod/invocation.py +++ b/src/orcapod/invocation.py @@ -75,6 +75,11 @@ class InvocationContext: 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__( diff --git a/src/orcapod/side_effects.py b/src/orcapod/side_effects.py index 6a7b2ea5..8b272cda 100644 --- a/src/orcapod/side_effects.py +++ b/src/orcapod/side_effects.py @@ -2,8 +2,8 @@ """SideEffectPod — pass-through pipeline node for side effects. Provides ``SideEffectPod``, ``SideEffectPodStream``, ``SideEffectJobNode``, -``InvocationContext``, ``InvocationHashConfig``, ``SideEffectPodConfig``, -and the ``side_effect_pod``, ``sink_pod``, ``tap_pod`` decorators. +``SideEffectPodConfig``, and the ``side_effect_pod``, ``sink_pod``, ``tap_pod`` +decorators. """ from __future__ import annotations @@ -21,7 +21,6 @@ from orcapod.invocation import ( InvocationContext, InvocationHashConfig, - _serialize_component, ) from orcapod.system_constants import TRACKING_DB_SCHEMA_VERSION from orcapod.utils.lazy_module import LazyModule From 30f8ac80842ba58c28c6b69e2ac1c6aac6949ca6 Mon Sep 17 00:00:00 2001 From: "agent-kurodo[bot]" <268466204+agent-kurodo[bot]@users.noreply.github.com> Date: Tue, 21 Jul 2026 05:13:43 +0000 Subject: [PATCH 5/9] test(hooks): add failing tests for PostRunPayload.invocation_context (ITL-531) Co-Authored-By: Claude Sonnet 4.6 --- .../function_pod/test_post_run_hooks.py | 98 +++++++++++++++++++ 1 file changed, 98 insertions(+) diff --git a/tests/test_core/function_pod/test_post_run_hooks.py b/tests/test_core/function_pod/test_post_run_hooks.py index b21c1b2e..558046b6 100644 --- a/tests/test_core/function_pod/test_post_run_hooks.py +++ b/tests/test_core/function_pod/test_post_run_hooks.py @@ -18,6 +18,7 @@ from orcapod.core.streams.arrow_table_stream import ArrowTableStream from orcapod.databases import InMemoryArrowDatabase from orcapod.hooks import HookConfig, InvocationStatus, PostRunPayload +from orcapod import InvocationHashConfig # --------------------------------------------------------------------------- @@ -435,3 +436,100 @@ def test_public_api_imports(self): assert hasattr(orcapod, "InvocationStatus") assert hasattr(orcapod, "RunStats") assert hasattr(orcapod, "PodContext") + + +# --------------------------------------------------------------------------- +# InvocationContext on PostRunPayload (ITL-531) +# --------------------------------------------------------------------------- + + +class TestInvocationContextOnPayload: + def test_invocation_context_always_present(self): + pod = _make_double_pod() + payloads: list[PostRunPayload] = [] + pod.add_post_run_hook(payloads.append) + + stream = pod.process(_make_stream(n=1)) + list(stream.iter_data()) + + assert payloads[0].invocation_context is not None + + def test_invocation_hash_is_nonempty_string(self): + pod = _make_double_pod() + payloads: list[PostRunPayload] = [] + pod.add_post_run_hook(payloads.append) + + stream = pod.process(_make_stream(n=1)) + list(stream.iter_data()) + + ctx = payloads[0].invocation_context + assert isinstance(ctx.invocation_hash, str) + assert len(ctx.invocation_hash) > 0 + + def test_format_id_matches_invocation_hash(self): + pod = _make_double_pod() + payloads: list[PostRunPayload] = [] + pod.add_post_run_hook(payloads.append) + + stream = pod.process(_make_stream(n=1)) + list(stream.iter_data()) + + ctx = payloads[0].invocation_context + assert ctx.format_id() == ctx.invocation_hash + + def test_format_id_base64_differs_from_hex(self): + pod = _make_double_pod() + payloads: list[PostRunPayload] = [] + pod.add_post_run_hook(payloads.append) + + stream = pod.process(_make_stream(n=1)) + list(stream.iter_data()) + + ctx = payloads[0].invocation_context + hex_id = ctx.invocation_hash + b64_id = ctx.format_id(InvocationHashConfig(encoding="base64")) + assert hex_id != b64_id + + def test_pipeline_run_id_is_none_standalone(self): + pod = _make_double_pod() + payloads: list[PostRunPayload] = [] + pod.add_post_run_hook(payloads.append) + + stream = pod.process(_make_stream(n=1)) + list(stream.iter_data()) + + assert payloads[0].invocation_context.pipeline_run_id is None + + def test_invocation_hash_deterministic(self): + pod = _make_double_pod() + payloads: list[PostRunPayload] = [] + pod.add_post_run_hook(payloads.append) + + # Process the same input stream twice + stream1 = pod.process(_make_stream(n=1)) + list(stream1.iter_data()) + stream2 = pod.process(_make_stream(n=1)) + list(stream2.iter_data()) + + assert len(payloads) == 2 + assert ( + payloads[0].invocation_context.invocation_hash + == payloads[1].invocation_context.invocation_hash + ) + + def test_error_payload_has_invocation_context(self): + def explodes(x: int) -> int: + raise RuntimeError("boom") + + pf = PythonDataFunction(explodes, output_keys="result") + pod = FunctionPod(pf) + payloads: list[PostRunPayload] = [] + pod.add_post_run_hook(HookConfig(fn=payloads.append, on_error="log")) + + stream = pod.process(_make_stream(n=1)) + with pytest.raises(RuntimeError, match="boom"): + list(stream.iter_data()) + + assert len(payloads) == 1 + assert payloads[0].invocation_context is not None + assert isinstance(payloads[0].invocation_context.invocation_hash, str) From 202fcbfc65e8eca090e5d8a2e6527adf257f66ee Mon Sep 17 00:00:00 2001 From: "agent-kurodo[bot]" <268466204+agent-kurodo[bot]@users.noreply.github.com> Date: Tue, 21 Jul 2026 05:22:32 +0000 Subject: [PATCH 6/9] feat(hooks): add invocation_context field to PostRunPayload (ITL-531) --- src/orcapod/hooks.py | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/src/orcapod/hooks.py b/src/orcapod/hooks.py index 5d88b35f..726e3ad6 100644 --- a/src/orcapod/hooks.py +++ b/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] From b4c6d20fabed884815bc51693e5ae2bc163e472c Mon Sep 17 00:00:00 2001 From: "agent-kurodo[bot]" <268466204+agent-kurodo[bot]@users.noreply.github.com> Date: Tue, 21 Jul 2026 05:29:55 +0000 Subject: [PATCH 7/9] feat(hooks): populate invocation_context on PostRunPayload (ITL-531) Thread run_id into _build_post_run_payload and call _build_invocation_context so every PostRunPayload carries a fully-populated InvocationContext with pipeline_run_id, invocation hash, and format_id support. --- src/orcapod/core/cached_function_pod.py | 6 ++++-- src/orcapod/core/function_pod.py | 13 +++++++++---- 2 files changed, 13 insertions(+), 6 deletions(-) diff --git a/src/orcapod/core/cached_function_pod.py b/src/orcapod/core/cached_function_pod.py index b59658ae..f61975ef 100644 --- a/src/orcapod/core/cached_function_pod.py +++ b/src/orcapod/core/cached_function_pod.py @@ -250,7 +250,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 +259,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 @@ -309,7 +310,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 +319,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 diff --git a/src/orcapod/core/function_pod.py b/src/orcapod/core/function_pod.py index 5066d9c3..bf82c2d5 100644 --- a/src/orcapod/core/function_pod.py +++ b/src/orcapod/core/function_pod.py @@ -347,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. @@ -358,6 +359,8 @@ 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. @@ -365,6 +368,7 @@ def _build_post_run_payload( 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, @@ -381,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( @@ -420,7 +425,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 @@ -429,7 +434,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 @@ -473,7 +478,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 @@ -482,7 +487,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 From 7df1d547d49f3bec399ff7e44b886a1ffc7d0566 Mon Sep 17 00:00:00 2001 From: "agent-kurodo[bot]" <268466204+agent-kurodo[bot]@users.noreply.github.com> Date: Tue, 21 Jul 2026 05:35:01 +0000 Subject: [PATCH 8/9] docs(hooks): correct run_id docstrings after invocation_context threading Update _invoke_with_hooks and _async_invoke_with_hooks docstrings in function_pod.py and cached_function_pod.py to accurately reflect that run_id is now forwarded to PostRunPayload.invocation_context, not just process_data / unused. Co-Authored-By: Claude Sonnet 4.6 --- src/orcapod/core/cached_function_pod.py | 6 ++++-- src/orcapod/core/function_pod.py | 6 ++++-- 2 files changed, 8 insertions(+), 4 deletions(-) diff --git a/src/orcapod/core/cached_function_pod.py b/src/orcapod/core/cached_function_pod.py index f61975ef..cdb5dc36 100644 --- a/src/orcapod/core/cached_function_pod.py +++ b/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. @@ -281,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. diff --git a/src/orcapod/core/function_pod.py b/src/orcapod/core/function_pod.py index bf82c2d5..f099d85e 100644 --- a/src/orcapod/core/function_pod.py +++ b/src/orcapod/core/function_pod.py @@ -406,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. @@ -457,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. From 2fda992cdebd36202a98abdd800b77d23288cee9 Mon Sep 17 00:00:00 2001 From: "agent-kurodo[bot]" <268466204+agent-kurodo[bot]@users.noreply.github.com> Date: Tue, 21 Jul 2026 05:40:23 +0000 Subject: [PATCH 9/9] test(hooks): assert InvocationContext and InvocationHashConfig in public API Extend test_public_api_imports to cover the two new exports added by ITL-531. Co-Authored-By: Claude Sonnet 4.6 --- tests/test_core/function_pod/test_post_run_hooks.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/tests/test_core/function_pod/test_post_run_hooks.py b/tests/test_core/function_pod/test_post_run_hooks.py index 558046b6..df94df33 100644 --- a/tests/test_core/function_pod/test_post_run_hooks.py +++ b/tests/test_core/function_pod/test_post_run_hooks.py @@ -436,6 +436,8 @@ def test_public_api_imports(self): assert hasattr(orcapod, "InvocationStatus") assert hasattr(orcapod, "RunStats") assert hasattr(orcapod, "PodContext") + assert hasattr(orcapod, "InvocationContext") + assert hasattr(orcapod, "InvocationHashConfig") # ---------------------------------------------------------------------------