From 1d8a2741eea12922975d4d860889dcff14f084af Mon Sep 17 00:00:00 2001 From: "agent-kurodo[bot]" <268466204+agent-kurodo[bot]@users.noreply.github.com> Date: Tue, 14 Jul 2026 01:14:25 +0000 Subject: [PATCH 01/10] docs(specs): add post-run hook design spec for ITL-523 --- ...2026-07-14-itl-523-post-run-hook-design.md | 288 ++++++++++++++++++ 1 file changed, 288 insertions(+) create mode 100644 superpowers/specs/2026-07-14-itl-523-post-run-hook-design.md diff --git a/superpowers/specs/2026-07-14-itl-523-post-run-hook-design.md b/superpowers/specs/2026-07-14-itl-523-post-run-hook-design.md new file mode 100644 index 00000000..d8836400 --- /dev/null +++ b/superpowers/specs/2026-07-14-itl-523-post-run-hook-design.md @@ -0,0 +1,288 @@ +# Post-Run Hook for Function Pods — Design Spec + +**Issue:** ITL-523 +**Date:** 2026-07-14 +**Status:** Approved + +--- + +## Overview + +Add a per-pod post-run hook mechanism to function pods. After each invocation of a +function pod — whether the result was freshly computed, served from a pod-level cache, or +resulted in an error — all hooks registered on that pod are called with a structured payload +describing the run. Hooks are the primitive for run-level logging, metrics collection, and +provenance stamping. + +--- + +## New Types — `src/orcapod/hooks.py` + +All public hook types live in a new top-level module `src/orcapod/hooks.py` and are +re-exported from `orcapod.__init__`. + +### `InvocationStatus` + +```python +class InvocationStatus(str, Enum): + COMPUTED = "computed" # function was invoked and produced a fresh result + HIT = "hit" # result served from pod-level cache (CachedFunctionPod) + ERROR = "error" # function raised an exception +``` + +### `RunStats` + +```python +@dataclass(frozen=True) +class RunStats: + duration_ms: float # wall-clock time in milliseconds + status: InvocationStatus + started_at: datetime # UTC + finished_at: datetime # UTC + error: Exception | None # populated when status == ERROR; None otherwise +``` + +### `PodContext` + +```python +@dataclass(frozen=True) +class PodContext: + label: str | None # pod.label — human-readable name + pod_hash: str # pod.content_hash().to_string() — identifies the function version +``` + +### `PostRunPayload` + +```python +@dataclass(frozen=True) +class PostRunPayload: + record_id_hash: str | None # str(output_data.datagram_uuid); None if filtered or error + tag: TagProtocol # input tag (immutable; do not mutate) + input: DataProtocol # input data (immutable; do not mutate) + output: DataProtocol | None # output data; None if filtered out or error + stats: RunStats + pod: PodContext +``` + +`record_id_hash` is the UUID of the output datagram — the same identifier used as the +primary key when the result is stored in a backing database. It uniquely identifies this +specific output record. It is `None` when `output` is `None` (i.e. the function filtered +the row out, or the invocation raised an error). Hooks must treat `tag`, `input`, and +`output` as read-only; the payload is frozen but the referenced datagram objects are not +deeply immutable. + +### Hook types + +```python +PostRunHookFn = Callable[[PostRunPayload], None] + +@dataclass(frozen=True) +class HookConfig: + fn: PostRunHookFn + on_error: Literal["raise", "log"] = "raise" + +PostRunHook = PostRunHookFn | HookConfig +``` + +A plain `PostRunHookFn` defaults to fail-loud (`on_error="raise"`). Wrap in `HookConfig` +to opt into resilient behaviour per hook. + +--- + +## Hook Registration API + +### `_FunctionPodBase.add_post_run_hook()` + +```python +def add_post_run_hook(self, hook: PostRunHook) -> None: + """Register a post-run hook on this pod. + + Hooks fire after every invocation (computed, cache hit, or error), in + registration order, before the result is emitted downstream. + + A plain callable defaults to fail-loud (exceptions propagate, stopping + the pod run). Wrap in ``HookConfig(fn=..., on_error="log")`` to log and + continue on hook failure. + + Args: + hook: A callable ``(PostRunPayload) -> None``, or a ``HookConfig`` + wrapping such a callable with explicit error handling. + """ + self._post_run_hooks.append(hook) +``` + +Hooks are stored in a plain `list[PostRunHook]` on `_FunctionPodBase`, initialised to `[]` +in `__init__`. This is intentionally mutable — hooks can be added after construction. + +### `@function_pod` decorator parameter + +```python +@function_pod( + output_keys="result", + post_run_hooks=[my_hook], # convenience parameter +) +def compute(x: int) -> int: + ... +``` + +The decorator accepts an optional `post_run_hooks: Sequence[PostRunHook] | None` +parameter. It calls `pod.add_post_run_hook(hook)` for each hook in order after the pod +is constructed (and after `CachedFunctionPod` wrapping, if applicable). This is strictly +a convenience — it is identical to calling `add_post_run_hook` manually after decoration. + +--- + +## Firing Semantics + +### When hooks fire + +Hooks fire **after** the pod's compute-or-lookup step completes (or raises), and +**before** the result is emitted downstream. They fire on: + +- Every freshly computed output (`InvocationStatus.COMPUTED`) +- Every pod-level cache hit (`InvocationStatus.HIT`) — i.e. when `CachedFunctionPod` + finds the result in its backing database +- Every invocation that raises an exception (`InvocationStatus.ERROR`) + +Hooks do **not** re-fire for `FunctionPodStream` in-memory re-iteration hits (when +`_cached_output_datas` already holds a result for an index within the same stream +object's lifetime). That is a low-level within-object optimisation, not a new pod +invocation — the hook already fired on the original computation. + +### Hook failure semantics + +Hooks run in registration order. Each hook runs to completion (or failure) before the +next. If a hook raises: + +- `on_error="raise"` (default for plain callables and `HookConfig` default): the + exception propagates immediately, stopping any remaining hooks and failing the pod + invocation. +- `on_error="log"`: the exception is logged at `WARNING` level and execution continues + with the next hook. The pod invocation result is still returned normally. + +When the pod function itself raises (`InvocationStatus.ERROR`), hooks fire first with +`status=ERROR` and `stats.error` set to the exception. After all hooks complete, the +original exception is re-raised. This means a hook with `on_error="raise"` that also +raises will replace the original exception in the traceback. + +### Ordering guarantees + +Hooks fire in registration order within a single invocation. There is no ordering +guarantee across concurrent invocations (each invocation fires its own hooks +independently). + +--- + +## Implementation — `_invoke_with_hooks()` + +A new internal method `_invoke_with_hooks()` (and async counterpart +`_async_invoke_with_hooks()`) is added to `_FunctionPodBase`. All call sites that +previously called `process_data()` or `async_process_data()` directly are updated to +call `_invoke_with_hooks()` instead. `process_data()` and `async_process_data()` are +unchanged. + +``` +Call sites updated: + FunctionPodStream._iter_data_sequential() → _invoke_with_hooks() + FunctionPodStream._iter_data_concurrent() → _invoke_with_hooks() / _async_invoke_with_hooks() + _FunctionPodBase.async_execute() → _async_invoke_with_hooks() +``` + +The base class `_invoke_with_hooks()`: + +1. Records `started_at`. +2. Calls `self.process_data(tag, data, logger=logger)`. +3. Records `finished_at`. +4. Determines `InvocationStatus` (always `COMPUTED` at the base class level). +5. If `self._post_run_hooks` is non-empty, constructs `PostRunPayload` and calls + `_fire_post_run_hooks(payload)`. +6. Re-raises any exception after hooks fire. +7. Returns `(out_tag, output_data)`. + +When `self._post_run_hooks` is empty, step 5 is skipped entirely — zero overhead for +pods with no hooks. + +### `CachedFunctionPod` override + +`CachedFunctionPod` overrides `_invoke_with_hooks()` (and its async counterpart) to +supply the correct `InvocationStatus`. It calls `self.process_data()` (which already +owns the cache lookup and store logic), then reads the `RESULT_COMPUTED_FLAG` meta field +on the output data to determine whether the result was a cache hit or fresh computation: + +- `RESULT_COMPUTED_FLAG == False` → `InvocationStatus.HIT` +- `RESULT_COMPUTED_FLAG == True` or absent → `InvocationStatus.COMPUTED` +- exception raised → `InvocationStatus.ERROR` + +This keeps `process_data()` as the single source of truth for caching behaviour. + +--- + +## Attachment Point + +**Per-pod only (v1).** There are no pipeline-level hooks in this release. Pipeline-level +hooks — where a single hook registration applies to every pod in a pipeline — are a +follow-up. They can be built on top of per-pod hooks without any API changes. + +--- + +## Sync vs Async + +**Synchronous hooks only (v1).** Hook callables are `(PostRunPayload) -> None` — plain +synchronous functions. They run on the critical path; slow hooks stall the pipeline. +Documentation must make this explicit. + +Async hooks are out of scope for v1. The `async_execute()` path calls +`_async_invoke_with_hooks()` but the hooks themselves are still invoked synchronously +within it. + +--- + +## Scope + +In scope: + +- `src/orcapod/hooks.py` — all new public types +- `src/orcapod/core/function_pod.py` — `_post_run_hooks` list, `add_post_run_hook()`, + `_invoke_with_hooks()`, `_async_invoke_with_hooks()`, `_fire_post_run_hooks()`; + updated call sites in `FunctionPodStream` and `async_execute()` +- `src/orcapod/core/cached_function_pod.py` — override `_invoke_with_hooks()` and + `_async_invoke_with_hooks()` +- `src/orcapod/__init__.py` — re-export new public types +- `tests/test_core/function_pod/test_post_run_hooks.py` — full test suite +- Docstrings on all new public API + +Out of scope: + +- Pre-run hooks +- Operator pod hooks +- Pipeline-level hooks +- Async hooks +- Built-in hook implementations (logging sink, metrics, EDI provenance) +- Remote / cross-process hook execution + +--- + +## Tests + +All tests in `tests/test_core/function_pod/test_post_run_hooks.py`: + +1. **Single hook fires with correct payload** — verify `record_id_hash`, `tag`, `input`, + `output`, `stats.status == COMPUTED`, `stats.duration_ms > 0`, `pod.label` +2. **Multiple hooks fire in registration order** — two hooks append to a shared list; + verify order matches registration order +3. **Fail-loud hook error** — plain callable hook raises; exception propagates, pod run + fails +4. **Resilient hook error** — `HookConfig(on_error="log")` raises; logged and suppressed, + computation result still returned, next hook still fires +5. **Cache hit status** — `CachedFunctionPod`, run same input twice; second call fires + hook with `InvocationStatus.HIT` +6. **Error status** — function raises; hook fires with `InvocationStatus.ERROR`, + `stats.error` is the exception, original exception re-raised after hooks +7. **Filtered output** — function returns `None` (filtered); hook fires with + `output=None`, `record_id_hash=None`, `stats.status == COMPUTED` +8. **Parallel execution** — concurrent pod with multiple inputs; hooks fire for all + inputs, no dropped calls +9. **Decorator convenience** — `@function_pod(post_run_hooks=[fn])` registers hook; + fires identically to `pod.add_post_run_hook(fn)` +10. **Empty hooks — no overhead path** — pod with no hooks registered; no payload + constructed (verified by confirming `_post_run_hooks` is empty and no side effects) From 0861f9e66bb090ba8417e13f150e56e72aad48fe Mon Sep 17 00:00:00 2001 From: "agent-kurodo[bot]" <268466204+agent-kurodo[bot]@users.noreply.github.com> Date: Tue, 14 Jul 2026 01:29:24 +0000 Subject: [PATCH 02/10] docs(plans): add post-run hook implementation plan (ITL-523) --- .../plans/2026-07-14-itl-523-post-run-hook.md | 1374 +++++++++++++++++ 1 file changed, 1374 insertions(+) create mode 100644 superpowers/plans/2026-07-14-itl-523-post-run-hook.md diff --git a/superpowers/plans/2026-07-14-itl-523-post-run-hook.md b/superpowers/plans/2026-07-14-itl-523-post-run-hook.md new file mode 100644 index 00000000..3a0ba5be --- /dev/null +++ b/superpowers/plans/2026-07-14-itl-523-post-run-hook.md @@ -0,0 +1,1374 @@ +# Post-Run Hook for Function Pods — 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 a per-pod post-run hook mechanism to `FunctionPod` so registered callables fire after every invocation with a typed payload describing what ran. + +**Architecture:** New types live in `src/orcapod/hooks.py`. `_FunctionPodBase` gains a mutable hook list plus an `_invoke_with_hooks()` wrapper method; all call sites that previously called `process_data()` directly now call `_invoke_with_hooks()` instead. `CachedFunctionPod` overrides `_invoke_with_hooks()` to supply the correct `InvocationStatus.HIT` when a database cache hit occurs. + +**Tech Stack:** Python 3.11+, PyArrow, pytest, uv (all commands via `uv run`) + +**Spec:** `superpowers/specs/2026-07-14-itl-523-post-run-hook-design.md` + +--- + +## File Map + +| Action | Path | Responsibility | +|--------|------|----------------| +| **Create** | `src/orcapod/hooks.py` | All public hook types: `InvocationStatus`, `RunStats`, `PodContext`, `PostRunPayload`, `PostRunHookFn`, `HookConfig`, `PostRunHook` | +| **Modify** | `src/orcapod/core/function_pod.py` | Add `_post_run_hooks`, `add_post_run_hook()`, `_fire_post_run_hooks()`, `_invoke_with_hooks()`, `_async_invoke_with_hooks()` to `_FunctionPodBase`; update call sites in `FunctionPodStream._iter_data_sequential`, `_iter_data_concurrent`, and `async_execute` | +| **Modify** | `src/orcapod/core/cached_function_pod.py` | Override `_invoke_with_hooks()` and `_async_invoke_with_hooks()` to detect `InvocationStatus.HIT` via `RESULT_COMPUTED_FLAG` | +| **Modify** | `src/orcapod/__init__.py` | Re-export all public hook types | +| **Create** | `tests/test_core/function_pod/test_post_run_hooks.py` | Full test suite (10 test classes) | + +--- + +## Task 1: Create `src/orcapod/hooks.py` + +**Files:** +- Create: `src/orcapod/hooks.py` + +- [ ] **Step 1: Create the hooks module** + +```python +# src/orcapod/hooks.py +"""Post-run hook types for function pods. + +Defines the payload, status, and hook configuration types used by the +post-run hook mechanism on function pods. Import these when writing or +registering hooks. +""" + +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 + +if TYPE_CHECKING: + from orcapod.protocols.core_protocols import DataProtocol, TagProtocol + + +class InvocationStatus(str, Enum): + """Status of a single function pod invocation. + + Attributes: + COMPUTED: The function was invoked and produced a fresh result. + HIT: The result was served from a pod-level database cache + (``CachedFunctionPod``). + ERROR: The function raised an exception. + """ + + COMPUTED = "computed" + HIT = "hit" + ERROR = "error" + + +@dataclasses.dataclass(frozen=True) +class RunStats: + """Timing and status information for a single pod invocation. + + Attributes: + duration_ms: Wall-clock time in milliseconds. + status: Whether the result was freshly computed, a cache hit, or an error. + started_at: UTC timestamp when the invocation started. + finished_at: UTC timestamp when the invocation finished (after hooks fire). + error: The exception raised, if ``status == ERROR``; ``None`` otherwise. + """ + + duration_ms: float + status: InvocationStatus + started_at: datetime + finished_at: datetime + error: Exception | None = None + + +@dataclasses.dataclass(frozen=True) +class PodContext: + """Identity information about the pod that produced a result. + + Attributes: + label: Human-readable pod label (``pod.label``); ``None`` if not set. + pod_hash: Hex-string content hash of the pod (``pod.content_hash().to_string()``). + Changes when the underlying function code or version changes. + """ + + label: str | None + pod_hash: str + + +@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. + """ + + record_id_hash: str | None + tag: TagProtocol + input: DataProtocol + output: DataProtocol | None + stats: RunStats + pod: PodContext + + +PostRunHookFn = Callable[["PostRunPayload"], None] +"""A plain hook callable: ``(PostRunPayload) -> None``. + +Defaults to fail-loud on error (exceptions propagate). +""" + + +@dataclasses.dataclass(frozen=True) +class HookConfig: + """Hook callable with explicit error-handling behaviour. + + Use this instead of a plain callable when you want the hook to log and + continue on failure rather than propagating the exception. + + Attributes: + fn: The hook callable. + on_error: ``"raise"`` (default) propagates exceptions; ``"log"`` logs at + WARNING level and continues. + + Example: + pod.add_post_run_hook(HookConfig(fn=my_hook, on_error="log")) + """ + + fn: PostRunHookFn + on_error: Literal["raise", "log"] = "raise" + + +PostRunHook = PostRunHookFn | HookConfig +"""A hook is either a plain callable (fail-loud) or a ``HookConfig`` wrapper.""" +``` + +- [ ] **Step 2: Verify module imports cleanly** + +```bash +uv run python -c "from orcapod.hooks import PostRunPayload, HookConfig, InvocationStatus, RunStats, PodContext, PostRunHook, PostRunHookFn; print('OK')" +``` + +Expected: `OK` + +- [ ] **Step 3: Commit** + +```bash +cd /home/kurouto/kurouto-jobs/09c4ea39-6353-4e02-ba3a-a9b86b0e157b/orcapod-python +git add src/orcapod/hooks.py +git commit -m "feat(hooks): add PostRunPayload and hook types module (ITL-523)" +``` + +--- + +## Task 2: Hook infrastructure on `_FunctionPodBase` + sequential call site + +Add `_post_run_hooks`, `add_post_run_hook`, `_fire_post_run_hooks`, and `_invoke_with_hooks` to `_FunctionPodBase`. Update `FunctionPodStream._iter_data_sequential` to call `_invoke_with_hooks`. This covers tests 1–4, 6, and 7. + +**Files:** +- Create: `tests/test_core/function_pod/test_post_run_hooks.py` +- Modify: `src/orcapod/core/function_pod.py` + +- [ ] **Step 1: Create the test file with failing tests** + +```python +# tests/test_core/function_pod/test_post_run_hooks.py +"""Tests for FunctionPod post-run hooks (ITL-523). + +Covers: registration, firing order, failure semantics, payload correctness, +filtered output, error status, cache hit status, parallel execution, +decorator convenience, and empty-hooks no-overhead path. +""" + +from __future__ import annotations + +import pyarrow as pa +import pytest + +from orcapod.core.cached_function_pod import CachedFunctionPod +from orcapod.core.data_function import PythonDataFunction +from orcapod.core.function_pod import FunctionPod, function_pod +from orcapod.core.streams.arrow_table_stream import ArrowTableStream +from orcapod.databases import InMemoryArrowDatabase +from orcapod.hooks import HookConfig, InvocationStatus, PostRunPayload + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def _make_stream(n: int = 2) -> ArrowTableStream: + schema = pa.schema([ + pa.field("id", pa.int64(), nullable=False), + pa.field("x", pa.int64(), nullable=False), + ]) + table = pa.table( + { + "id": pa.array(list(range(n)), type=pa.int64()), + "x": pa.array(list(range(n)), type=pa.int64()), + }, + schema=schema, + ) + return ArrowTableStream(table, tag_columns=["id"]) + + +def double(x: int) -> int: + return x * 2 + + +def _make_double_pod() -> FunctionPod: + pf = PythonDataFunction(double, output_keys="result") + return FunctionPod(pf) + + +# --------------------------------------------------------------------------- +# 1. Single hook fires with correct payload +# --------------------------------------------------------------------------- + + +class TestSingleHookPayload: + def test_hook_fires_with_correct_payload(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 len(payloads) == 1 + p = payloads[0] + assert p.stats.status == InvocationStatus.COMPUTED + assert p.stats.duration_ms >= 0 + assert p.stats.error is None + assert p.output is not None + assert p.record_id_hash == str(p.output.datagram_uuid) + assert p.pod.label == pod.label + assert p.pod.pod_hash == pod.content_hash().to_string() + assert p.input is not None + assert p.tag is not None + + def test_hook_fires_for_each_row(self): + pod = _make_double_pod() + payloads: list[PostRunPayload] = [] + pod.add_post_run_hook(payloads.append) + + stream = pod.process(_make_stream(n=3)) + list(stream.iter_data()) + + assert len(payloads) == 3 + + +# --------------------------------------------------------------------------- +# 2. Multiple hooks fire in registration order +# --------------------------------------------------------------------------- + + +class TestHookOrdering: + def test_multiple_hooks_fire_in_order(self): + pod = _make_double_pod() + fired: list[str] = [] + pod.add_post_run_hook(lambda p: fired.append("first")) + pod.add_post_run_hook(lambda p: fired.append("second")) + + stream = pod.process(_make_stream(n=1)) + list(stream.iter_data()) + + assert fired == ["first", "second"] + + def test_hooks_fire_in_order_for_every_row(self): + pod = _make_double_pod() + fired: list[str] = [] + pod.add_post_run_hook(lambda p: fired.append("first")) + pod.add_post_run_hook(lambda p: fired.append("second")) + + stream = pod.process(_make_stream(n=2)) + list(stream.iter_data()) + + assert fired == ["first", "second", "first", "second"] + + +# --------------------------------------------------------------------------- +# 3. Fail-loud hook error +# --------------------------------------------------------------------------- + + +class TestHookFailureFast: + def test_failing_hook_propagates_exception(self): + pod = _make_double_pod() + + def bad_hook(p: PostRunPayload) -> None: + raise ValueError("hook exploded") + + pod.add_post_run_hook(bad_hook) + + stream = pod.process(_make_stream(n=1)) + with pytest.raises(ValueError, match="hook exploded"): + list(stream.iter_data()) + + def test_failing_hook_stops_remaining_hooks(self): + pod = _make_double_pod() + second_fired: list[bool] = [] + + def bad_hook(p: PostRunPayload) -> None: + raise ValueError("stops here") + + pod.add_post_run_hook(bad_hook) + pod.add_post_run_hook(lambda p: second_fired.append(True)) + + stream = pod.process(_make_stream(n=1)) + with pytest.raises(ValueError): + list(stream.iter_data()) + + assert second_fired == [] + + +# --------------------------------------------------------------------------- +# 4. Resilient hook error +# --------------------------------------------------------------------------- + + +class TestHookFailureResilient: + def test_resilient_hook_suppresses_exception(self): + pod = _make_double_pod() + second_fired: list[bool] = [] + + def bad_hook(p: PostRunPayload) -> None: + raise ValueError("suppressed") + + def second_hook(p: PostRunPayload) -> None: + second_fired.append(True) + + pod.add_post_run_hook(HookConfig(fn=bad_hook, on_error="log")) + pod.add_post_run_hook(second_hook) + + stream = pod.process(_make_stream(n=1)) + results = list(stream.iter_data()) + + assert results # computation result returned + assert second_fired == [True] # next hook still fired + + def test_hookconfig_raise_is_same_as_plain_callable(self): + pod = _make_double_pod() + + def bad_hook(p: PostRunPayload) -> None: + raise ValueError("still loud") + + pod.add_post_run_hook(HookConfig(fn=bad_hook, on_error="raise")) + + stream = pod.process(_make_stream(n=1)) + with pytest.raises(ValueError, match="still loud"): + list(stream.iter_data()) + + +# --------------------------------------------------------------------------- +# 6. Error status (pod function raises) +# --------------------------------------------------------------------------- + + +class TestErrorStatus: + def test_pod_error_fires_hook_with_error_status(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 + p = payloads[0] + assert p.stats.status == InvocationStatus.ERROR + assert isinstance(p.stats.error, RuntimeError) + assert str(p.stats.error) == "boom" + assert p.output is None + assert p.record_id_hash is None + + def test_original_exception_reraises_after_hooks(self): + def explodes(x: int) -> int: + raise RuntimeError("original") + + pf = PythonDataFunction(explodes, output_keys="result") + pod = FunctionPod(pf) + pod.add_post_run_hook(HookConfig(fn=lambda p: None, on_error="log")) + + stream = pod.process(_make_stream(n=1)) + with pytest.raises(RuntimeError, match="original"): + list(stream.iter_data()) + + +# --------------------------------------------------------------------------- +# 7. Filtered output +# --------------------------------------------------------------------------- + + +class TestFilteredOutput: + def test_filtered_row_fires_hook_with_none_output(self): + def filter_all(x: int) -> int | None: + return None + + pf = PythonDataFunction(filter_all, output_keys="result") + pod = FunctionPod(pf) + + payloads: list[PostRunPayload] = [] + pod.add_post_run_hook(payloads.append) + + stream = pod.process(_make_stream(n=1)) + results = list(stream.iter_data()) + + assert results == [] + assert len(payloads) == 1 + p = payloads[0] + assert p.output is None + assert p.record_id_hash is None + assert p.stats.status == InvocationStatus.COMPUTED + assert p.stats.error is None + + +# --------------------------------------------------------------------------- +# 10. Empty hooks — no overhead path +# --------------------------------------------------------------------------- + + +class TestEmptyHooks: + def test_pod_with_no_hooks_has_empty_list(self): + pod = _make_double_pod() + assert pod._post_run_hooks == [] + + def test_pod_with_no_hooks_processes_normally(self): + pod = _make_double_pod() + stream = pod.process(_make_stream(n=2)) + results = list(stream.iter_data()) + assert len(results) == 2 +``` + +- [ ] **Step 2: Run the tests to confirm they fail with AttributeError** + +```bash +uv run pytest tests/test_core/function_pod/test_post_run_hooks.py -x -q 2>&1 | head -20 +``` + +Expected: `AttributeError: '_FunctionPodBase' object has no attribute 'add_post_run_hook'` (or similar) + +- [ ] **Step 3: Add imports to `function_pod.py`** + +At the top of `src/orcapod/core/function_pod.py`, add these imports after the existing import block (after line ~39): + +```python +from datetime import datetime, timezone + +from orcapod.hooks import ( + HookConfig, + InvocationStatus, + PodContext, + PostRunPayload, + PostRunHook, + RunStats, +) +``` + +- [ ] **Step 4: Add `_post_run_hooks` initialisation to `_FunctionPodBase.__init__`** + +In `_FunctionPodBase.__init__` (around line 76, after `self._data_function = data_function`), add: + +```python + self._post_run_hooks: list[PostRunHook] = [] +``` + +- [ ] **Step 5: Add `add_post_run_hook`, `_fire_post_run_hooks`, and `_invoke_with_hooks` to `_FunctionPodBase`** + +After the `process_data` method (around line 173 in `function_pod.py`), insert these three methods: + +```python + def add_post_run_hook(self, hook: PostRunHook) -> None: + """Register a post-run hook on this pod. + + Hooks fire after every invocation (computed, cache hit, or error), in + registration order, before the result is emitted downstream. + + A plain callable defaults to fail-loud (exceptions propagate, stopping + the pod run). Wrap in ``HookConfig(fn=..., on_error="log")`` to log and + continue on hook failure. + + Args: + hook: A callable ``(PostRunPayload) -> None``, or a ``HookConfig`` + wrapping such a callable with explicit error handling. + """ + self._post_run_hooks.append(hook) + + def _fire_post_run_hooks(self, payload: PostRunPayload) -> None: + """Fire all registered hooks with payload in registration order. + + Args: + payload: The post-run payload to pass to each hook. + """ + for hook in self._post_run_hooks: + fn = hook.fn if isinstance(hook, HookConfig) else hook + on_error = hook.on_error if isinstance(hook, HookConfig) else "raise" + try: + fn(payload) + except Exception as exc: + if on_error == "raise": + raise + logger.warning( + "Post-run hook %r raised and was suppressed: %s", + fn, + exc, + exc_info=True, + ) + + def _invoke_with_hooks( + self, + tag: TagProtocol, + data: DataProtocol, + *, + logger: DataExecutionLoggerProtocol | None = None, + ) -> tuple[TagProtocol, DataProtocol | None]: + """Call ``process_data``, time it, and fire post-run hooks. + + This is the call site used by ``FunctionPodStream`` and + ``async_execute``; ``process_data`` itself is unchanged. 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``. + + Returns: + A ``(tag, output_data)`` tuple. + """ + started_at = datetime.now(timezone.utc) + exc: Exception | None = None + out_tag = tag + output_data: DataProtocol | None = None + + try: + out_tag, output_data = self.process_data(tag, data, logger=logger) + status = InvocationStatus.COMPUTED + except Exception as e: + exc = e + status = InvocationStatus.ERROR + + finished_at = datetime.now(timezone.utc) + + if self._post_run_hooks: + record_id = ( + str(output_data.datagram_uuid) if output_data is not None else None + ) + payload = 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(), + ), + ) + self._fire_post_run_hooks(payload) + + if exc is not None: + raise exc + return out_tag, output_data + + async def _async_invoke_with_hooks( + self, + tag: TagProtocol, + data: DataProtocol, + *, + logger: DataExecutionLoggerProtocol | None = None, + ) -> tuple[TagProtocol, DataProtocol | None]: + """Async counterpart of ``_invoke_with_hooks``. + + Args: + tag: The tag associated with the data. + data: The input data to process. + logger: Optional data execution logger forwarded to ``async_process_data``. + + Returns: + A ``(tag, output_data)`` tuple. + """ + started_at = datetime.now(timezone.utc) + exc: Exception | None = None + out_tag = tag + output_data: DataProtocol | None = None + + try: + out_tag, output_data = await self.async_process_data( + tag, data, logger=logger + ) + status = InvocationStatus.COMPUTED + except Exception as e: + exc = e + status = InvocationStatus.ERROR + + finished_at = datetime.now(timezone.utc) + + if self._post_run_hooks: + record_id = ( + str(output_data.datagram_uuid) if output_data is not None else None + ) + payload = 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(), + ), + ) + self._fire_post_run_hooks(payload) + + if exc is not None: + raise exc + return out_tag, output_data +``` + +- [ ] **Step 6: Update `FunctionPodStream._iter_data_sequential` to call `_invoke_with_hooks`** + +In `FunctionPodStream._iter_data_sequential` (around line 530), replace: + +```python + tag, output_data = self._function_pod.process_data(tag, data) +``` + +with: + +```python + tag, output_data = self._function_pod._invoke_with_hooks(tag, data) +``` + +- [ ] **Step 7: Run the tests — most should pass now** + +```bash +uv run pytest tests/test_core/function_pod/test_post_run_hooks.py -x -q 2>&1 | head -30 +``` + +Expected: `TestSingleHookPayload`, `TestHookOrdering`, `TestHookFailureFast`, `TestHookFailureResilient`, `TestErrorStatus`, `TestFilteredOutput`, `TestEmptyHooks` all pass. `TestCacheHitStatus`, `TestParallelExecution`, `TestDecoratorConvenience` will be added in later tasks. + +- [ ] **Step 8: Run the existing function pod test suite to confirm no regressions** + +```bash +uv run pytest tests/test_core/function_pod/ -q 2>&1 | tail -10 +``` + +Expected: all existing tests pass. + +- [ ] **Step 9: Commit** + +```bash +git add src/orcapod/core/function_pod.py tests/test_core/function_pod/test_post_run_hooks.py +git commit -m "feat(function_pod): add post-run hook infrastructure and sequential call site (ITL-523)" +``` + +--- + +## Task 3: Update `FunctionPodStream._iter_data_concurrent` + parallel test + +**Files:** +- Modify: `src/orcapod/core/function_pod.py` (two lines in `_iter_data_concurrent`) +- Modify: `tests/test_core/function_pod/test_post_run_hooks.py` (add `TestParallelExecution`) + +- [ ] **Step 1: Add `TestParallelExecution` to the test file** + +Append to `tests/test_core/function_pod/test_post_run_hooks.py`: + +```python +# --------------------------------------------------------------------------- +# 8. Parallel execution (concurrent path via _iter_data_concurrent) +# --------------------------------------------------------------------------- + + +class _ConcurrentExecutor: + """Minimal executor that marks supports_concurrent_execution=True.""" + + @property + def executor_type_id(self) -> str: + return "test-concurrent" + + def supported_function_type_ids(self) -> frozenset[str]: + return frozenset() + + @property + def supports_concurrent_execution(self) -> bool: + return True + + def execute(self, data_function, data): + return data_function.direct_call(data) + + async def async_execute(self, data_function, data): + return data_function.direct_call(data) + + def execute_callable(self, fn, kwargs, executor_options=None, **kw): + return fn(**kwargs) + + async def async_execute_callable(self, fn, kwargs, executor_options=None, **kw): + return fn(**kwargs) + + def with_options(self, **kwargs): + return self + + +class TestParallelExecution: + def test_hooks_fire_for_all_inputs_under_concurrent_executor(self): + executor = _ConcurrentExecutor() + pf = PythonDataFunction(double, output_keys="result", executor=executor) + pod = FunctionPod(pf) + + payloads: list[PostRunPayload] = [] + pod.add_post_run_hook(payloads.append) + + stream = pod.process(_make_stream(n=4)) + results = list(stream.iter_data()) + + assert len(results) == 4 + assert len(payloads) == 4 + assert all( + p.stats.status == InvocationStatus.COMPUTED for p in payloads + ) +``` + +- [ ] **Step 2: Run the new test to confirm it fails** + +```bash +uv run pytest tests/test_core/function_pod/test_post_run_hooks.py::TestParallelExecution -x -q 2>&1 | head -15 +``` + +Expected: FAIL — hooks fire 0 times (concurrent path still calls `process_data` / `async_process_data` directly). + +- [ ] **Step 3: Update `_iter_data_concurrent` — sync fallback path** + +In `FunctionPodStream._iter_data_concurrent` (around line 564), replace: + +```python + results = [ + self._function_pod.process_data(tag, pkt) + for _, tag, pkt in to_compute + ] +``` + +with: + +```python + results = [ + self._function_pod._invoke_with_hooks(tag, pkt) + for _, tag, pkt in to_compute + ] +``` + +- [ ] **Step 4: Update `_iter_data_concurrent` — async gather path** + +In `FunctionPodStream._iter_data_concurrent` (around line 570), replace: + +```python + async def _gather() -> list[tuple[TagProtocol, DataProtocol | None]]: + return list( + await asyncio.gather( + *[ + self._function_pod.async_process_data(tag, pkt) + for _, tag, pkt in to_compute + ] + ) + ) +``` + +with: + +```python + async def _gather() -> list[tuple[TagProtocol, DataProtocol | None]]: + return list( + await asyncio.gather( + *[ + self._function_pod._async_invoke_with_hooks(tag, pkt) + for _, tag, pkt in to_compute + ] + ) + ) +``` + +- [ ] **Step 5: Run the parallel test — should pass now** + +```bash +uv run pytest tests/test_core/function_pod/test_post_run_hooks.py::TestParallelExecution -x -q +``` + +Expected: PASS + +- [ ] **Step 6: Run full function pod suite for regressions** + +```bash +uv run pytest tests/test_core/function_pod/ -q 2>&1 | tail -10 +``` + +Expected: all pass. + +- [ ] **Step 7: Commit** + +```bash +git add src/orcapod/core/function_pod.py tests/test_core/function_pod/test_post_run_hooks.py +git commit -m "feat(function_pod): update concurrent call sites to use _invoke_with_hooks (ITL-523)" +``` + +--- + +## Task 4: Update `async_execute` call site + +**Files:** +- Modify: `src/orcapod/core/function_pod.py` (`async_execute`) +- Modify: `tests/test_core/function_pod/test_post_run_hooks.py` (add `TestAsyncExecuteHooks`) + +- [ ] **Step 1: Add `TestAsyncExecuteHooks` to the test file** + +Append to `tests/test_core/function_pod/test_post_run_hooks.py`: + +```python +# --------------------------------------------------------------------------- +# async_execute path +# --------------------------------------------------------------------------- + + +class TestAsyncExecuteHooks: + @pytest.mark.asyncio + async def test_hooks_fire_through_async_execute(self): + from orcapod.channels import Channel + + pod = _make_double_pod() + payloads: list[PostRunPayload] = [] + pod.add_post_run_hook(payloads.append) + + stream = _make_stream(n=3) + input_ch: Channel = Channel() + output_ch: Channel = Channel() + + async def feed() -> None: + for tag, data in stream.iter_data(): + await input_ch.writer.send((tag, data)) + await input_ch.writer.close() + + import asyncio + await asyncio.gather( + feed(), + pod.async_execute([input_ch.reader], output_ch.writer), + ) + + results = [] + async for item in output_ch.reader: + results.append(item) + + assert len(results) == 3 + assert len(payloads) == 3 + assert all(p.stats.status == InvocationStatus.COMPUTED for p in payloads) +``` + +- [ ] **Step 2: Run to confirm it fails** + +```bash +uv run pytest tests/test_core/function_pod/test_post_run_hooks.py::TestAsyncExecuteHooks -x -q 2>&1 | head -15 +``` + +Expected: FAIL — hooks fire 0 times (async_execute still calls `async_process_data` directly). + +- [ ] **Step 3: Update `async_execute` to call `_async_invoke_with_hooks`** + +In `_FunctionPodBase.async_execute`, inside the `process_one` inner function (around line 241), replace: + +```python + out_tag, result_data = await self.async_process_data( + tag, data, logger=pkt_logger + ) +``` + +with: + +```python + out_tag, result_data = await self._async_invoke_with_hooks( + tag, data, logger=pkt_logger + ) +``` + +- [ ] **Step 4: Run the async_execute test — should pass** + +```bash +uv run pytest tests/test_core/function_pod/test_post_run_hooks.py::TestAsyncExecuteHooks -x -q +``` + +Expected: PASS + +- [ ] **Step 5: Run full function pod suite for regressions** + +```bash +uv run pytest tests/test_core/function_pod/ -q 2>&1 | tail -10 +``` + +Expected: all pass. + +- [ ] **Step 6: Commit** + +```bash +git add src/orcapod/core/function_pod.py tests/test_core/function_pod/test_post_run_hooks.py +git commit -m "feat(function_pod): update async_execute to use _async_invoke_with_hooks (ITL-523)" +``` + +--- + +## Task 5: `CachedFunctionPod` override for cache hit status + +**Files:** +- Modify: `src/orcapod/core/cached_function_pod.py` +- Modify: `tests/test_core/function_pod/test_post_run_hooks.py` (add `TestCacheHitStatus`) + +- [ ] **Step 1: Add `TestCacheHitStatus` to the test file** + +Append to `tests/test_core/function_pod/test_post_run_hooks.py`: + +```python +# --------------------------------------------------------------------------- +# 5. Cache hit status +# --------------------------------------------------------------------------- + + +class TestCacheHitStatus: + def test_second_call_fires_hook_with_hit_status(self): + inner = _make_double_pod() + db = InMemoryArrowDatabase() + pod = CachedFunctionPod(function_pod=inner, result_database=db) + + payloads: list[PostRunPayload] = [] + pod.add_post_run_hook(payloads.append) + + 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].stats.status == InvocationStatus.COMPUTED + assert payloads[1].stats.status == InvocationStatus.HIT + + def test_cache_hit_payload_has_record_id(self): + inner = _make_double_pod() + db = InMemoryArrowDatabase() + pod = CachedFunctionPod(function_pod=inner, result_database=db) + + payloads: list[PostRunPayload] = [] + pod.add_post_run_hook(payloads.append) + + stream1 = pod.process(_make_stream(n=1)) + list(stream1.iter_data()) + stream2 = pod.process(_make_stream(n=1)) + list(stream2.iter_data()) + + assert payloads[1].record_id_hash is not None + assert payloads[1].output is not None +``` + +- [ ] **Step 2: Run to confirm it fails** + +```bash +uv run pytest tests/test_core/function_pod/test_post_run_hooks.py::TestCacheHitStatus -x -q 2>&1 | head -15 +``` + +Expected: FAIL — `payloads[1].stats.status == InvocationStatus.COMPUTED` instead of `HIT` (base class always returns `COMPUTED`). + +- [ ] **Step 3: Add imports to `cached_function_pod.py`** + +At the top of `src/orcapod/core/cached_function_pod.py`, add after existing imports: + +```python +from datetime import datetime, timezone + +from orcapod.hooks import ( + HookConfig, + InvocationStatus, + PodContext, + PostRunPayload, + RunStats, +) +from orcapod.types import ColumnConfig +``` + +- [ ] **Step 4: Add `_invoke_with_hooks` override to `CachedFunctionPod`** + +After the `async_process_data` method in `CachedFunctionPod` (around line 149), add: + +```python + def _invoke_with_hooks( + self, + tag: TagProtocol, + data: DataProtocol, + *, + logger: DataExecutionLoggerProtocol | None = None, + ) -> tuple[TagProtocol, DataProtocol | None]: + """Override to detect cache hit status from ``RESULT_COMPUTED_FLAG`` meta. + + Calls ``self.process_data()`` (which owns all cache lookup and store + logic), then reads ``RESULT_COMPUTED_FLAG`` from the output data meta to + determine ``InvocationStatus.HIT`` vs ``InvocationStatus.COMPUTED``. + + Args: + tag: The tag associated with the data. + data: The input data to process. + logger: Optional data execution logger. + + Returns: + A ``(tag, output_data)`` tuple. + """ + started_at = datetime.now(timezone.utc) + exc: Exception | None = None + out_tag = tag + output_data: DataProtocol | None = None + status = InvocationStatus.COMPUTED + + try: + out_tag, output_data = self.process_data(tag, data, logger=logger) + if output_data is not None: + meta = output_data.as_dict(columns=ColumnConfig(meta=True)) + if meta.get(self.RESULT_COMPUTED_FLAG) is False: + status = InvocationStatus.HIT + except Exception as e: + exc = e + status = InvocationStatus.ERROR + + finished_at = datetime.now(timezone.utc) + + if self._post_run_hooks: + record_id = ( + str(output_data.datagram_uuid) if output_data is not None else None + ) + payload = 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(), + ), + ) + self._fire_post_run_hooks(payload) + + if exc is not None: + raise exc + return out_tag, output_data + + async def _async_invoke_with_hooks( + self, + tag: TagProtocol, + data: DataProtocol, + *, + logger: DataExecutionLoggerProtocol | None = None, + ) -> tuple[TagProtocol, DataProtocol | None]: + """Async counterpart of ``_invoke_with_hooks`` for ``CachedFunctionPod``. + + Args: + tag: The tag associated with the data. + data: The input data to process. + logger: Optional data execution logger. + + Returns: + A ``(tag, output_data)`` tuple. + """ + started_at = datetime.now(timezone.utc) + exc: Exception | None = None + out_tag = tag + output_data: DataProtocol | None = None + status = InvocationStatus.COMPUTED + + try: + out_tag, output_data = await self.async_process_data( + tag, data, logger=logger + ) + if output_data is not None: + meta = output_data.as_dict(columns=ColumnConfig(meta=True)) + if meta.get(self.RESULT_COMPUTED_FLAG) is False: + status = InvocationStatus.HIT + except Exception as e: + exc = e + status = InvocationStatus.ERROR + + finished_at = datetime.now(timezone.utc) + + if self._post_run_hooks: + record_id = ( + str(output_data.datagram_uuid) if output_data is not None else None + ) + payload = 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(), + ), + ) + self._fire_post_run_hooks(payload) + + if exc is not None: + raise exc + return out_tag, output_data +``` + +You also need to add the missing import at the top of `cached_function_pod.py` — `DataExecutionLoggerProtocol` is already imported, but verify it is present: + +```python +from orcapod.protocols.observability_protocols import DataExecutionLoggerProtocol +``` + +- [ ] **Step 5: Run the cache hit tests — should pass** + +```bash +uv run pytest tests/test_core/function_pod/test_post_run_hooks.py::TestCacheHitStatus -x -q +``` + +Expected: PASS + +- [ ] **Step 6: Run full test suite for regressions** + +```bash +uv run pytest tests/test_core/function_pod/ -q 2>&1 | tail -10 +``` + +Expected: all pass. + +- [ ] **Step 7: Commit** + +```bash +git add src/orcapod/core/cached_function_pod.py tests/test_core/function_pod/test_post_run_hooks.py +git commit -m "feat(cached_function_pod): override _invoke_with_hooks for cache hit status (ITL-523)" +``` + +--- + +## Task 6: `@function_pod` decorator + `orcapod.__init__` exports + +**Files:** +- Modify: `src/orcapod/core/function_pod.py` (`function_pod` decorator signature) +- Modify: `src/orcapod/__init__.py` +- Modify: `tests/test_core/function_pod/test_post_run_hooks.py` (add `TestDecoratorConvenience`) + +- [ ] **Step 1: Add `TestDecoratorConvenience` to the test file** + +Append to `tests/test_core/function_pod/test_post_run_hooks.py`: + +```python +# --------------------------------------------------------------------------- +# 9. Decorator convenience +# --------------------------------------------------------------------------- + + +class TestDecoratorConvenience: + def test_decorator_post_run_hooks_fires_hook(self): + payloads: list[PostRunPayload] = [] + + @function_pod(output_keys="result", post_run_hooks=[payloads.append]) + def compute(x: int) -> int: + return x * 3 + + stream = compute.pod.process(_make_stream(n=2)) + list(stream.iter_data()) + + assert len(payloads) == 2 + assert all(p.stats.status == InvocationStatus.COMPUTED for p in payloads) + + def test_decorator_hookconfig_works(self): + payloads: list[PostRunPayload] = [] + + @function_pod( + output_keys="result", + post_run_hooks=[HookConfig(fn=payloads.append, on_error="log")], + ) + def compute2(x: int) -> int: + return x + 1 + + stream = compute2.pod.process(_make_stream(n=1)) + list(stream.iter_data()) + + assert len(payloads) == 1 + + 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") +``` + +- [ ] **Step 2: Run to confirm decorator test fails** + +```bash +uv run pytest tests/test_core/function_pod/test_post_run_hooks.py::TestDecoratorConvenience -x -q 2>&1 | head -15 +``` + +Expected: `TypeError: function_pod() got an unexpected keyword argument 'post_run_hooks'` + +- [ ] **Step 3: Update `function_pod` decorator signature** + +In `src/orcapod/core/function_pod.py`, find the `function_pod` function definition (around line 700). Add `post_run_hooks` parameter before `**kwargs`: + +```python +def function_pod( + output_keys: str | Sequence[str] | None = None, + function_name: str | None = None, + version: str = "v0.0", + label: str | None = None, + result_database: ArrowDatabaseProtocol | None = None, + pod_cache_database: ArrowDatabaseProtocol | None = None, + executor: DataFunctionExecutorProtocol | None = None, + post_run_hooks: Sequence[PostRunHook] | None = None, + **kwargs, +) -> Callable[..., CallableWithPodProtocol]: +``` + +Also add the `Sequence` import if not already present — it's already in the imports at the top of `function_pod.py` (`from collections.abc import Callable, Collection, Iterator, Sequence`), so no change needed there. + +- [ ] **Step 4: Register hooks in the decorator body** + +Inside the `decorator` function, after the `CachedFunctionPod` wrapping block (after the `if pod_cache_database is not None:` block, around line 760), add: + +```python + if post_run_hooks: + for hook in post_run_hooks: + pod.add_post_run_hook(hook) +``` + +- [ ] **Step 5: Update `src/orcapod/__init__.py` to export hook types** + +In `src/orcapod/__init__.py`, add after the existing imports: + +```python +from .hooks import ( + HookConfig, + InvocationStatus, + PodContext, + PostRunHook, + PostRunHookFn, + PostRunPayload, + RunStats, +) +``` + +And add these names to `__all__`: + +```python + # Post-run hook types (ITL-523) + "HookConfig", + "InvocationStatus", + "PodContext", + "PostRunHook", + "PostRunHookFn", + "PostRunPayload", + "RunStats", +``` + +- [ ] **Step 6: Run decorator and public API tests — should pass** + +```bash +uv run pytest tests/test_core/function_pod/test_post_run_hooks.py::TestDecoratorConvenience -x -q +``` + +Expected: PASS + +- [ ] **Step 7: Run the complete new test file** + +```bash +uv run pytest tests/test_core/function_pod/test_post_run_hooks.py -v 2>&1 | tail -30 +``` + +Expected: all 10 test classes pass. + +- [ ] **Step 8: Commit** + +```bash +git add src/orcapod/core/function_pod.py src/orcapod/__init__.py tests/test_core/function_pod/test_post_run_hooks.py +git commit -m "feat(function_pod): add post_run_hooks param to decorator and export public API (ITL-523)" +``` + +--- + +## Task 7: Full test suite + final commit + +- [ ] **Step 1: Run the entire test suite** + +```bash +uv run pytest tests/ -q 2>&1 | tail -20 +``` + +Expected: all tests pass, no regressions. + +- [ ] **Step 2: If any test fails, investigate and fix before proceeding** + +Common failure modes: +- Import errors: check that all new imports in `function_pod.py` and `cached_function_pod.py` are at the top of the file, not inside methods. +- `AttributeError: '_post_run_hooks'`: the `__init__` assignment in `_FunctionPodBase` may not have been saved — re-check step 4 of Task 2. +- `TypeError` in `_iter_data_concurrent`: verify both sync fallback and async gather paths were updated in Task 3. + +- [ ] **Step 3: Checkout the feature branch (if not already on it)** + +```bash +git branch --show-current +``` + +Expected: `eywalker/itl-523-function-pods-configurable-post-run-hook-invoked-after-every` + +If not on the branch: +```bash +git checkout -b eywalker/itl-523-function-pods-configurable-post-run-hook-invoked-after-every +``` + +- [ ] **Step 4: Push to remote** + +```bash +git push -u origin eywalker/itl-523-function-pods-configurable-post-run-hook-invoked-after-every +``` + +Expected: branch pushed, ready for PR. From d9213c4958ebb8cbadf8d182e9804801199829f6 Mon Sep 17 00:00:00 2001 From: "agent-kurodo[bot]" <268466204+agent-kurodo[bot]@users.noreply.github.com> Date: Tue, 14 Jul 2026 01:33:39 +0000 Subject: [PATCH 03/10] feat(hooks): add PostRunPayload and hook types module (ITL-523) --- src/orcapod/hooks.py | 120 +++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 120 insertions(+) create mode 100644 src/orcapod/hooks.py diff --git a/src/orcapod/hooks.py b/src/orcapod/hooks.py new file mode 100644 index 00000000..1d0d40ec --- /dev/null +++ b/src/orcapod/hooks.py @@ -0,0 +1,120 @@ +# src/orcapod/hooks.py +"""Post-run hook types for function pods. + +Defines the payload, status, and hook configuration types used by the +post-run hook mechanism on function pods. Import these when writing or +registering hooks. +""" + +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 + +if TYPE_CHECKING: + from orcapod.protocols.core_protocols import DataProtocol, TagProtocol + + +class InvocationStatus(str, Enum): + """Status of a single function pod invocation. + + Attributes: + COMPUTED: The function was invoked and produced a fresh result. + HIT: The result was served from a pod-level database cache + (``CachedFunctionPod``). + ERROR: The function raised an exception. + """ + + COMPUTED = "computed" + HIT = "hit" + ERROR = "error" + + +@dataclasses.dataclass(frozen=True) +class RunStats: + """Timing and status information for a single pod invocation. + + Attributes: + duration_ms: Wall-clock time in milliseconds. + status: Whether the result was freshly computed, a cache hit, or an error. + started_at: UTC timestamp when the invocation started. + finished_at: UTC timestamp when the invocation finished (after hooks fire). + error: The exception raised, if ``status == ERROR``; ``None`` otherwise. + """ + + duration_ms: float + status: InvocationStatus + started_at: datetime + finished_at: datetime + error: Exception | None = None + + +@dataclasses.dataclass(frozen=True) +class PodContext: + """Identity information about the pod that produced a result. + + Attributes: + label: Human-readable pod label (``pod.label``); ``None`` if not set. + pod_hash: Hex-string content hash of the pod (``pod.content_hash().to_string()``). + Changes when the underlying function code or version changes. + """ + + label: str | None + pod_hash: str + + +@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. + """ + + record_id_hash: str | None + tag: TagProtocol + input: DataProtocol + output: DataProtocol | None + stats: RunStats + pod: PodContext + + +PostRunHookFn = Callable[["PostRunPayload"], None] +"""A plain hook callable: ``(PostRunPayload) -> None``. + +Defaults to fail-loud on error (exceptions propagate). +""" + + +@dataclasses.dataclass(frozen=True) +class HookConfig: + """Hook callable with explicit error-handling behaviour. + + Use this instead of a plain callable when you want the hook to log and + continue on failure rather than propagating the exception. + + Attributes: + fn: The hook callable. + on_error: ``"raise"`` (default) propagates exceptions; ``"log"`` logs at + WARNING level and continues. + + Example: + pod.add_post_run_hook(HookConfig(fn=my_hook, on_error="log")) + """ + + fn: PostRunHookFn + on_error: Literal["raise", "log"] = "raise" + + +PostRunHook = PostRunHookFn | HookConfig +"""A hook is either a plain callable (fail-loud) or a ``HookConfig`` wrapper.""" From 71a1764b171ec3b0a4e905364d1c420cc98ea9ab Mon Sep 17 00:00:00 2001 From: "agent-kurodo[bot]" <268466204+agent-kurodo[bot]@users.noreply.github.com> Date: Tue, 14 Jul 2026 01:42:58 +0000 Subject: [PATCH 04/10] feat(function_pod): add post-run hook infrastructure and sequential call site (ITL-523) Co-Authored-By: Claude Sonnet 4.6 --- src/orcapod/core/function_pod.py | 174 ++++++++++- .../function_pod/test_post_run_hooks.py | 271 ++++++++++++++++++ 2 files changed, 444 insertions(+), 1 deletion(-) create mode 100644 tests/test_core/function_pod/test_post_run_hooks.py diff --git a/src/orcapod/core/function_pod.py b/src/orcapod/core/function_pod.py index 2be8c7bb..b77fb9c1 100644 --- a/src/orcapod/core/function_pod.py +++ b/src/orcapod/core/function_pod.py @@ -38,6 +38,17 @@ from orcapod.utils import arrow_utils, schema_utils from orcapod.utils.lazy_module import LazyModule +from datetime import datetime, timezone + +from orcapod.hooks import ( + HookConfig, + InvocationStatus, + PodContext, + PostRunHook, + PostRunPayload, + RunStats, +) + logger = logging.getLogger(__name__) if TYPE_CHECKING: @@ -76,6 +87,7 @@ def __init__( ) self.tracker_manager = tracker_manager or DEFAULT_TRACKER_MANAGER self._data_function = data_function + self._post_run_hooks: list[PostRunHook] = [] # Union-typed input args (e.g. x: str | Path) are deliberately accepted # at construction time. ensure_types_registered_for_schemas registers # each non-None branch individually; the union is only resolved to a @@ -183,6 +195,166 @@ async def async_process_data( result = await self.data_function.async_call(data, logger=logger) return tag, result + def add_post_run_hook(self, hook: PostRunHook) -> None: + """Register a post-run hook on this pod. + + Hooks fire after every invocation (computed, cache hit, or error), in + registration order, before the result is emitted downstream. + + A plain callable defaults to fail-loud (exceptions propagate, stopping + the pod run). Wrap in ``HookConfig(fn=..., on_error="log")`` to log and + continue on hook failure. + + Args: + hook: A callable ``(PostRunPayload) -> None``, or a ``HookConfig`` + wrapping such a callable with explicit error handling. + """ + self._post_run_hooks.append(hook) + + def _fire_post_run_hooks(self, payload: PostRunPayload) -> None: + """Fire all registered hooks with payload in registration order. + + Args: + payload: The post-run payload to pass to each hook. + """ + for hook in self._post_run_hooks: + fn = hook.fn if isinstance(hook, HookConfig) else hook + on_error = hook.on_error if isinstance(hook, HookConfig) else "raise" + try: + fn(payload) + except Exception as exc: + if on_error == "raise": + raise + logger.warning( + "Post-run hook %r raised and was suppressed: %s", + fn, + exc, + exc_info=True, + ) + + def _invoke_with_hooks( + self, + tag: TagProtocol, + data: DataProtocol, + *, + logger: DataExecutionLoggerProtocol | None = None, + ) -> tuple[TagProtocol, DataProtocol | None]: + """Call ``process_data``, time it, and fire post-run hooks. + + This is the call site used by ``FunctionPodStream`` and + ``async_execute``; ``process_data`` itself is unchanged. 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``. + + Returns: + A ``(tag, output_data)`` tuple. + """ + started_at = datetime.now(timezone.utc) + exc: Exception | None = None + out_tag = tag + output_data: DataProtocol | None = None + + try: + out_tag, output_data = self.process_data(tag, data, logger=logger) + status = InvocationStatus.COMPUTED + except Exception as e: + exc = e + status = InvocationStatus.ERROR + + finished_at = datetime.now(timezone.utc) + + if self._post_run_hooks: + record_id = ( + str(output_data.datagram_uuid) if output_data is not None else None + ) + payload = 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(), + ), + ) + self._fire_post_run_hooks(payload) + + if exc is not None: + raise exc + return out_tag, output_data + + async def _async_invoke_with_hooks( + self, + tag: TagProtocol, + data: DataProtocol, + *, + logger: DataExecutionLoggerProtocol | None = None, + ) -> tuple[TagProtocol, DataProtocol | None]: + """Async counterpart of ``_invoke_with_hooks``. + + Args: + tag: The tag associated with the data. + data: The input data to process. + logger: Optional data execution logger forwarded to ``async_process_data``. + + Returns: + A ``(tag, output_data)`` tuple. + """ + started_at = datetime.now(timezone.utc) + exc: Exception | None = None + out_tag = tag + output_data: DataProtocol | None = None + + try: + out_tag, output_data = await self.async_process_data( + tag, data, logger=logger + ) + status = InvocationStatus.COMPUTED + except Exception as e: + exc = e + status = InvocationStatus.ERROR + + finished_at = datetime.now(timezone.utc) + + if self._post_run_hooks: + record_id = ( + str(output_data.datagram_uuid) if output_data is not None else None + ) + payload = 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(), + ), + ) + self._fire_post_run_hooks(payload) + + if exc is not None: + raise exc + return out_tag, output_data + def handle_input_streams(self, *streams: StreamProtocol) -> StreamProtocol: """Handle multiple input streams by joining them if necessary. @@ -528,7 +700,7 @@ def _iter_data_sequential( yield tag, data else: # Process data - tag, output_data = self._function_pod.process_data(tag, data) + tag, output_data = self._function_pod._invoke_with_hooks(tag, data) self._cached_output_datas[i] = (tag, output_data) if output_data is not None: yield tag, output_data diff --git a/tests/test_core/function_pod/test_post_run_hooks.py b/tests/test_core/function_pod/test_post_run_hooks.py new file mode 100644 index 00000000..36ef5477 --- /dev/null +++ b/tests/test_core/function_pod/test_post_run_hooks.py @@ -0,0 +1,271 @@ +# tests/test_core/function_pod/test_post_run_hooks.py +"""Tests for FunctionPod post-run hooks (ITL-523). + +Covers: registration, firing order, failure semantics, payload correctness, +filtered output, error status, cache hit status, parallel execution, +decorator convenience, and empty-hooks no-overhead path. +""" + +from __future__ import annotations + +import pyarrow as pa +import pytest + +from orcapod.core.cached_function_pod import CachedFunctionPod +from orcapod.core.data_function import PythonDataFunction +from orcapod.core.function_pod import FunctionPod, function_pod +from orcapod.core.streams.arrow_table_stream import ArrowTableStream +from orcapod.databases import InMemoryArrowDatabase +from orcapod.hooks import HookConfig, InvocationStatus, PostRunPayload + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def _make_stream(n: int = 2) -> ArrowTableStream: + schema = pa.schema([ + pa.field("id", pa.int64(), nullable=False), + pa.field("x", pa.int64(), nullable=False), + ]) + table = pa.table( + { + "id": pa.array(list(range(n)), type=pa.int64()), + "x": pa.array(list(range(n)), type=pa.int64()), + }, + schema=schema, + ) + return ArrowTableStream(table, tag_columns=["id"]) + + +def double(x: int) -> int: + return x * 2 + + +def _make_double_pod() -> FunctionPod: + pf = PythonDataFunction(double, output_keys="result") + return FunctionPod(pf) + + +# --------------------------------------------------------------------------- +# 1. Single hook fires with correct payload +# --------------------------------------------------------------------------- + + +class TestSingleHookPayload: + def test_hook_fires_with_correct_payload(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 len(payloads) == 1 + p = payloads[0] + assert p.stats.status == InvocationStatus.COMPUTED + assert p.stats.duration_ms >= 0 + assert p.stats.error is None + assert p.output is not None + assert p.record_id_hash == str(p.output.datagram_uuid) + assert p.pod.label == pod.label + assert p.pod.pod_hash == pod.content_hash().to_string() + assert p.input is not None + assert p.tag is not None + + def test_hook_fires_for_each_row(self): + pod = _make_double_pod() + payloads: list[PostRunPayload] = [] + pod.add_post_run_hook(payloads.append) + + stream = pod.process(_make_stream(n=3)) + list(stream.iter_data()) + + assert len(payloads) == 3 + + +# --------------------------------------------------------------------------- +# 2. Multiple hooks fire in registration order +# --------------------------------------------------------------------------- + + +class TestHookOrdering: + def test_multiple_hooks_fire_in_order(self): + pod = _make_double_pod() + fired: list[str] = [] + pod.add_post_run_hook(lambda p: fired.append("first")) + pod.add_post_run_hook(lambda p: fired.append("second")) + + stream = pod.process(_make_stream(n=1)) + list(stream.iter_data()) + + assert fired == ["first", "second"] + + def test_hooks_fire_in_order_for_every_row(self): + pod = _make_double_pod() + fired: list[str] = [] + pod.add_post_run_hook(lambda p: fired.append("first")) + pod.add_post_run_hook(lambda p: fired.append("second")) + + stream = pod.process(_make_stream(n=2)) + list(stream.iter_data()) + + assert fired == ["first", "second", "first", "second"] + + +# --------------------------------------------------------------------------- +# 3. Fail-loud hook error +# --------------------------------------------------------------------------- + + +class TestHookFailureFast: + def test_failing_hook_propagates_exception(self): + pod = _make_double_pod() + + def bad_hook(p: PostRunPayload) -> None: + raise ValueError("hook exploded") + + pod.add_post_run_hook(bad_hook) + + stream = pod.process(_make_stream(n=1)) + with pytest.raises(ValueError, match="hook exploded"): + list(stream.iter_data()) + + def test_failing_hook_stops_remaining_hooks(self): + pod = _make_double_pod() + second_fired: list[bool] = [] + + def bad_hook(p: PostRunPayload) -> None: + raise ValueError("stops here") + + pod.add_post_run_hook(bad_hook) + pod.add_post_run_hook(lambda p: second_fired.append(True)) + + stream = pod.process(_make_stream(n=1)) + with pytest.raises(ValueError): + list(stream.iter_data()) + + assert second_fired == [] + + +# --------------------------------------------------------------------------- +# 4. Resilient hook error +# --------------------------------------------------------------------------- + + +class TestHookFailureResilient: + def test_resilient_hook_suppresses_exception(self): + pod = _make_double_pod() + second_fired: list[bool] = [] + + def bad_hook(p: PostRunPayload) -> None: + raise ValueError("suppressed") + + def second_hook(p: PostRunPayload) -> None: + second_fired.append(True) + + pod.add_post_run_hook(HookConfig(fn=bad_hook, on_error="log")) + pod.add_post_run_hook(second_hook) + + stream = pod.process(_make_stream(n=1)) + results = list(stream.iter_data()) + + assert results # computation result returned + assert second_fired == [True] # next hook still fired + + def test_hookconfig_raise_is_same_as_plain_callable(self): + pod = _make_double_pod() + + def bad_hook(p: PostRunPayload) -> None: + raise ValueError("still loud") + + pod.add_post_run_hook(HookConfig(fn=bad_hook, on_error="raise")) + + stream = pod.process(_make_stream(n=1)) + with pytest.raises(ValueError, match="still loud"): + list(stream.iter_data()) + + +# --------------------------------------------------------------------------- +# 6. Error status (pod function raises) +# --------------------------------------------------------------------------- + + +class TestErrorStatus: + def test_pod_error_fires_hook_with_error_status(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 + p = payloads[0] + assert p.stats.status == InvocationStatus.ERROR + assert isinstance(p.stats.error, RuntimeError) + assert str(p.stats.error) == "boom" + assert p.output is None + assert p.record_id_hash is None + + def test_original_exception_reraises_after_hooks(self): + def explodes(x: int) -> int: + raise RuntimeError("original") + + pf = PythonDataFunction(explodes, output_keys="result") + pod = FunctionPod(pf) + pod.add_post_run_hook(HookConfig(fn=lambda p: None, on_error="log")) + + stream = pod.process(_make_stream(n=1)) + with pytest.raises(RuntimeError, match="original"): + list(stream.iter_data()) + + +# --------------------------------------------------------------------------- +# 7. Filtered output +# --------------------------------------------------------------------------- + + +class TestFilteredOutput: + def test_filtered_row_fires_hook_with_none_output(self): + pf = PythonDataFunction(double, output_keys="result") + pf.set_active(False) # set_active(False) causes process_data to return None output + pod = FunctionPod(pf) + + payloads: list[PostRunPayload] = [] + pod.add_post_run_hook(payloads.append) + + stream = pod.process(_make_stream(n=1)) + results = list(stream.iter_data()) + + assert results == [] + assert len(payloads) == 1 + p = payloads[0] + assert p.output is None + assert p.record_id_hash is None + assert p.stats.status == InvocationStatus.COMPUTED + assert p.stats.error is None + + +# --------------------------------------------------------------------------- +# 10. Empty hooks — no overhead path +# --------------------------------------------------------------------------- + + +class TestEmptyHooks: + def test_pod_with_no_hooks_has_empty_list(self): + pod = _make_double_pod() + assert pod._post_run_hooks == [] + + def test_pod_with_no_hooks_processes_normally(self): + pod = _make_double_pod() + stream = pod.process(_make_stream(n=2)) + results = list(stream.iter_data()) + assert len(results) == 2 From ce23d95274e154a31d5547c5a403e86174f51203 Mon Sep 17 00:00:00 2001 From: "agent-kurodo[bot]" <268466204+agent-kurodo[bot]@users.noreply.github.com> Date: Tue, 14 Jul 2026 01:49:13 +0000 Subject: [PATCH 05/10] feat(function_pod): update concurrent call sites to use _invoke_with_hooks (ITL-523) Co-Authored-By: Claude Sonnet 4.6 --- src/orcapod/core/function_pod.py | 4 +- .../function_pod/test_post_run_hooks.py | 46 +++++++++++++++++++ 2 files changed, 48 insertions(+), 2 deletions(-) diff --git a/src/orcapod/core/function_pod.py b/src/orcapod/core/function_pod.py index b77fb9c1..377655c5 100644 --- a/src/orcapod/core/function_pod.py +++ b/src/orcapod/core/function_pod.py @@ -734,7 +734,7 @@ def _iter_data_concurrent( if loop is not None: # Already in event loop — fall back to sequential sync results = [ - self._function_pod.process_data(tag, pkt) + self._function_pod._invoke_with_hooks(tag, pkt) for _, tag, pkt in to_compute ] else: @@ -743,7 +743,7 @@ async def _gather() -> list[tuple[TagProtocol, DataProtocol | None]]: return list( await asyncio.gather( *[ - self._function_pod.async_process_data(tag, pkt) + self._function_pod._async_invoke_with_hooks(tag, pkt) for _, tag, pkt in to_compute ] ) 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 36ef5477..e558e476 100644 --- a/tests/test_core/function_pod/test_post_run_hooks.py +++ b/tests/test_core/function_pod/test_post_run_hooks.py @@ -13,6 +13,7 @@ from orcapod.core.cached_function_pod import CachedFunctionPod from orcapod.core.data_function import PythonDataFunction +from orcapod.core.executors.base import PythonFunctionExecutorBase from orcapod.core.function_pod import FunctionPod, function_pod from orcapod.core.streams.arrow_table_stream import ArrowTableStream from orcapod.databases import InMemoryArrowDatabase @@ -269,3 +270,48 @@ def test_pod_with_no_hooks_processes_normally(self): stream = pod.process(_make_stream(n=2)) results = list(stream.iter_data()) assert len(results) == 2 + + +# --------------------------------------------------------------------------- +# 8. Parallel execution (concurrent path via _iter_data_concurrent) +# --------------------------------------------------------------------------- + + +class _ConcurrentExecutor(PythonFunctionExecutorBase): + """Minimal executor that marks supports_concurrent_execution=True.""" + + @property + def executor_type_id(self) -> str: + return "test-concurrent" + + def supported_function_type_ids(self) -> frozenset[str]: + return frozenset() + + @property + def supports_concurrent_execution(self) -> bool: + return True + + def execute_callable(self, fn, kwargs, executor_options=None, **kw): + return fn(**kwargs) + + async def async_execute_callable(self, fn, kwargs, executor_options=None, **kw): + return fn(**kwargs) + + +class TestParallelExecution: + def test_hooks_fire_for_all_inputs_under_concurrent_executor(self): + executor = _ConcurrentExecutor() + pf = PythonDataFunction(double, output_keys="result", executor=executor) + pod = FunctionPod(pf) + + payloads: list[PostRunPayload] = [] + pod.add_post_run_hook(payloads.append) + + stream = pod.process(_make_stream(n=4)) + results = list(stream.iter_data()) + + assert len(results) == 4 + assert len(payloads) == 4 + assert all( + p.stats.status == InvocationStatus.COMPUTED for p in payloads + ) From 417deddbe0cf9929cfb7595514f45728bb62eb2a Mon Sep 17 00:00:00 2001 From: "agent-kurodo[bot]" <268466204+agent-kurodo[bot]@users.noreply.github.com> Date: Tue, 14 Jul 2026 01:52:11 +0000 Subject: [PATCH 06/10] feat(function_pod): update async_execute to use _async_invoke_with_hooks (ITL-523) Co-Authored-By: Claude Sonnet 4.6 --- src/orcapod/core/function_pod.py | 2 +- .../function_pod/test_post_run_hooks.py | 36 +++++++++++++++++++ 2 files changed, 37 insertions(+), 1 deletion(-) diff --git a/src/orcapod/core/function_pod.py b/src/orcapod/core/function_pod.py index 377655c5..6537422b 100644 --- a/src/orcapod/core/function_pod.py +++ b/src/orcapod/core/function_pod.py @@ -410,7 +410,7 @@ async def process_one(tag: TagProtocol, data: DataProtocol) -> None: obs.on_data_start(pod_label, tag, data) pkt_logger = obs.create_data_logger(tag, data) try: - out_tag, result_data = await self.async_process_data( + out_tag, result_data = await self._async_invoke_with_hooks( tag, data, logger=pkt_logger ) except Exception as exc: 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 e558e476..b0705c1d 100644 --- a/tests/test_core/function_pod/test_post_run_hooks.py +++ b/tests/test_core/function_pod/test_post_run_hooks.py @@ -315,3 +315,39 @@ def test_hooks_fire_for_all_inputs_under_concurrent_executor(self): assert all( p.stats.status == InvocationStatus.COMPUTED for p in payloads ) + + +# --------------------------------------------------------------------------- +# async_execute path +# --------------------------------------------------------------------------- + + +class TestAsyncExecuteHooks: + @pytest.mark.asyncio + async def test_hooks_fire_through_async_execute(self): + from orcapod.channels import Channel + + pod = _make_double_pod() + payloads: list[PostRunPayload] = [] + pod.add_post_run_hook(payloads.append) + + stream = _make_stream(n=3) + input_ch: Channel = Channel(buffer_size=16) + output_ch: Channel = Channel(buffer_size=16) + + async def feed() -> None: + for tag, data in stream.iter_data(): + await input_ch.writer.send((tag, data)) + await input_ch.writer.close() + + import asyncio + await asyncio.gather( + feed(), + pod.async_execute([input_ch.reader], output_ch.writer), + ) + + results = await output_ch.reader.collect() + + assert len(results) == 3 + assert len(payloads) == 3 + assert all(p.stats.status == InvocationStatus.COMPUTED for p in payloads) From 3f1a45b912268df2fa0b55bc5b944fea545ecf1d Mon Sep 17 00:00:00 2001 From: "agent-kurodo[bot]" <268466204+agent-kurodo[bot]@users.noreply.github.com> Date: Tue, 14 Jul 2026 01:55:46 +0000 Subject: [PATCH 07/10] feat(cached_function_pod): override _invoke_with_hooks for cache hit status (ITL-523) Co-Authored-By: Claude Sonnet 4.6 --- src/orcapod/core/cached_function_pod.py | 138 ++++++++++++++++++ .../function_pod/test_post_run_hooks.py | 41 ++++++ 2 files changed, 179 insertions(+) diff --git a/src/orcapod/core/cached_function_pod.py b/src/orcapod/core/cached_function_pod.py index e8c5129f..9c4116b4 100644 --- a/src/orcapod/core/cached_function_pod.py +++ b/src/orcapod/core/cached_function_pod.py @@ -3,11 +3,18 @@ from __future__ import annotations import logging +from datetime import datetime, timezone from typing import TYPE_CHECKING, Any from orcapod.core.datagrams import Datagram from orcapod.core.function_pod import WrappedFunctionPod from orcapod.core.result_cache import ResultCache +from orcapod.hooks import ( + InvocationStatus, + PodContext, + PostRunPayload, + RunStats, +) from orcapod.protocols.core_protocols import ( FunctionPodProtocol, DataProtocol, @@ -16,6 +23,7 @@ ) from orcapod.protocols.database_protocols import ArrowDatabaseProtocol from orcapod.protocols.observability_protocols import DataExecutionLoggerProtocol +from orcapod.types import ColumnConfig if TYPE_CHECKING: import pyarrow as pa @@ -148,6 +156,136 @@ async def async_process_data( output = output.with_meta_columns(**{self.RESULT_COMPUTED_FLAG: True}) return tag, output + def _invoke_with_hooks( + self, + tag: TagProtocol, + data: DataProtocol, + *, + logger: DataExecutionLoggerProtocol | None = None, + ) -> tuple[TagProtocol, DataProtocol | None]: + """Override to detect cache hit status from ``RESULT_COMPUTED_FLAG`` meta. + + Calls ``self.process_data()`` (which owns all cache lookup and store + logic), then reads ``RESULT_COMPUTED_FLAG`` from the output data meta to + determine ``InvocationStatus.HIT`` vs ``InvocationStatus.COMPUTED``. + + Args: + tag: The tag associated with the data. + data: The input data to process. + logger: Optional data execution logger. + + Returns: + A ``(tag, output_data)`` tuple. + """ + started_at = datetime.now(timezone.utc) + exc: Exception | None = None + out_tag = tag + output_data: DataProtocol | None = None + status = InvocationStatus.COMPUTED + + try: + out_tag, output_data = self.process_data(tag, data, logger=logger) + if output_data is not None: + meta = output_data.as_dict(columns=ColumnConfig(meta=True)) + if meta.get(self.RESULT_COMPUTED_FLAG) is False: + status = InvocationStatus.HIT + except Exception as e: + exc = e + status = InvocationStatus.ERROR + + finished_at = datetime.now(timezone.utc) + + if self._post_run_hooks: + record_id = ( + str(output_data.datagram_uuid) if output_data is not None else None + ) + payload = 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(), + ), + ) + self._fire_post_run_hooks(payload) + + if exc is not None: + raise exc + return out_tag, output_data + + async def _async_invoke_with_hooks( + self, + tag: TagProtocol, + data: DataProtocol, + *, + logger: DataExecutionLoggerProtocol | None = None, + ) -> tuple[TagProtocol, DataProtocol | None]: + """Async counterpart of ``_invoke_with_hooks`` for ``CachedFunctionPod``. + + Args: + tag: The tag associated with the data. + data: The input data to process. + logger: Optional data execution logger. + + Returns: + A ``(tag, output_data)`` tuple. + """ + started_at = datetime.now(timezone.utc) + exc: Exception | None = None + out_tag = tag + output_data: DataProtocol | None = None + status = InvocationStatus.COMPUTED + + try: + out_tag, output_data = await self.async_process_data( + tag, data, logger=logger + ) + if output_data is not None: + meta = output_data.as_dict(columns=ColumnConfig(meta=True)) + if meta.get(self.RESULT_COMPUTED_FLAG) is False: + status = InvocationStatus.HIT + except Exception as e: + exc = e + status = InvocationStatus.ERROR + + finished_at = datetime.now(timezone.utc) + + if self._post_run_hooks: + record_id = ( + str(output_data.datagram_uuid) if output_data is not None else None + ) + payload = 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(), + ), + ) + self._fire_post_run_hooks(payload) + + if exc is not None: + raise exc + return out_tag, output_data + def get_all_cached_outputs( self, include_system_columns: bool = False ) -> "pa.Table | None": 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 b0705c1d..71ac16b2 100644 --- a/tests/test_core/function_pod/test_post_run_hooks.py +++ b/tests/test_core/function_pod/test_post_run_hooks.py @@ -351,3 +351,44 @@ async def feed() -> None: assert len(results) == 3 assert len(payloads) == 3 assert all(p.stats.status == InvocationStatus.COMPUTED for p in payloads) + + +# --------------------------------------------------------------------------- +# 5. Cache hit status +# --------------------------------------------------------------------------- + + +class TestCacheHitStatus: + def test_second_call_fires_hook_with_hit_status(self): + inner = _make_double_pod() + db = InMemoryArrowDatabase() + pod = CachedFunctionPod(function_pod=inner, result_database=db) + + payloads: list[PostRunPayload] = [] + pod.add_post_run_hook(payloads.append) + + 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].stats.status == InvocationStatus.COMPUTED + assert payloads[1].stats.status == InvocationStatus.HIT + + def test_cache_hit_payload_has_record_id(self): + inner = _make_double_pod() + db = InMemoryArrowDatabase() + pod = CachedFunctionPod(function_pod=inner, result_database=db) + + payloads: list[PostRunPayload] = [] + pod.add_post_run_hook(payloads.append) + + stream1 = pod.process(_make_stream(n=1)) + list(stream1.iter_data()) + stream2 = pod.process(_make_stream(n=1)) + list(stream2.iter_data()) + + assert payloads[1].record_id_hash is not None + assert payloads[1].output is not None From cfed32a56164cdea983f4deb9fd144da234bc3d0 Mon Sep 17 00:00:00 2001 From: "agent-kurodo[bot]" <268466204+agent-kurodo[bot]@users.noreply.github.com> Date: Tue, 14 Jul 2026 01:58:07 +0000 Subject: [PATCH 08/10] feat(function_pod): add post_run_hooks param to decorator and export public API (ITL-523) Co-Authored-By: Claude Sonnet 4.6 --- src/orcapod/__init__.py | 19 ++++++++ src/orcapod/core/function_pod.py | 8 ++++ .../function_pod/test_post_run_hooks.py | 43 +++++++++++++++++++ 3 files changed, 70 insertions(+) diff --git a/src/orcapod/__init__.py b/src/orcapod/__init__.py index f2ce587d..e24d83a5 100644 --- a/src/orcapod/__init__.py +++ b/src/orcapod/__init__.py @@ -37,6 +37,17 @@ # Extension registration API (ITL-473) from .extensions import OrcapodExtension, register_extension +# Post-run hook types (ITL-523) +from .hooks import ( + HookConfig, + InvocationStatus, + PodContext, + PostRunHook, + PostRunHookFn, + PostRunPayload, + RunStats, +) + __all__ = [ "DEFAULT_CONFIG", "DisplayConfig", @@ -63,6 +74,14 @@ # Extension registration API (ITL-473) "OrcapodExtension", "register_extension", + # Post-run hook types (ITL-523) + "HookConfig", + "InvocationStatus", + "PodContext", + "PostRunHook", + "PostRunHookFn", + "PostRunPayload", + "RunStats", ] diff --git a/src/orcapod/core/function_pod.py b/src/orcapod/core/function_pod.py index 6537422b..09f0a5b8 100644 --- a/src/orcapod/core/function_pod.py +++ b/src/orcapod/core/function_pod.py @@ -877,6 +877,7 @@ def function_pod( result_database: ArrowDatabaseProtocol | None = None, pod_cache_database: ArrowDatabaseProtocol | None = None, executor: DataFunctionExecutorProtocol | None = None, + post_run_hooks: Sequence[PostRunHook] | None = None, **kwargs, ) -> Callable[..., CallableWithPodProtocol]: """Decorator that attaches a ``FunctionPod`` as a ``pod`` attribute. @@ -892,6 +893,9 @@ def function_pod( (wraps the pod in ``CachedFunctionPod``, which caches at the ``process_data`` level using input data content hash). executor: Optional executor for running the data function. + post_run_hooks: Optional list of post-run hooks to register on the + pod after construction. Each entry is either a plain callable + ``(PostRunPayload) -> None`` or a ``HookConfig``. **kwargs: Forwarded to ``PythonDataFunction``. Returns: @@ -933,6 +937,10 @@ def decorator(func: Callable) -> CallableWithPodProtocol: result_database=pod_cache_database, ) + if post_run_hooks: + for hook in post_run_hooks: + pod.add_post_run_hook(hook) + @wraps(func) def wrapper(*args, **kwargs): return func(*args, **kwargs) 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 71ac16b2..b21c1b2e 100644 --- a/tests/test_core/function_pod/test_post_run_hooks.py +++ b/tests/test_core/function_pod/test_post_run_hooks.py @@ -392,3 +392,46 @@ def test_cache_hit_payload_has_record_id(self): assert payloads[1].record_id_hash is not None assert payloads[1].output is not None + + +# --------------------------------------------------------------------------- +# 9. Decorator convenience +# --------------------------------------------------------------------------- + + +class TestDecoratorConvenience: + def test_decorator_post_run_hooks_fires_hook(self): + payloads: list[PostRunPayload] = [] + + @function_pod(output_keys="result", post_run_hooks=[payloads.append]) + def compute(x: int) -> int: + return x * 3 + + stream = compute.pod.process(_make_stream(n=2)) + list(stream.iter_data()) + + assert len(payloads) == 2 + assert all(p.stats.status == InvocationStatus.COMPUTED for p in payloads) + + def test_decorator_hookconfig_works(self): + payloads: list[PostRunPayload] = [] + + @function_pod( + output_keys="result", + post_run_hooks=[HookConfig(fn=payloads.append, on_error="log")], + ) + def compute2(x: int) -> int: + return x + 1 + + stream = compute2.pod.process(_make_stream(n=1)) + list(stream.iter_data()) + + assert len(payloads) == 1 + + 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") From 78bcaf67977fb8b1cf657ae4487340c4940458c6 Mon Sep 17 00:00:00 2001 From: "agent-kurodo[bot]" <268466204+agent-kurodo[bot]@users.noreply.github.com> Date: Tue, 14 Jul 2026 02:30:06 +0000 Subject: [PATCH 09/10] =?UTF-8?q?fix(hooks):=20address=20PR=20review=20?= =?UTF-8?q?=E2=80=94=20traceback=20preservation,=20empty-hooks=20fast-path?= =?UTF-8?q?,=20docstring=20(ITL-523)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Add fast-path early return to all four _invoke_with_hooks / _async_invoke_with_hooks methods (base + CachedFunctionPod) so pods without hooks pay zero overhead and retain exactly the same exception behaviour as before - Restructure error path to fire hooks inside the except block and use bare raise, so the original traceback is preserved exactly instead of adding a wrapper frame - Extract _build_post_run_payload() helper on _FunctionPodBase to eliminate the duplicated payload-construction code across sync/async/cached variants - Fix RunStats docstring: finished_at and duration_ms now correctly describe compute-or-lookup completion time, before hooks fire Co-Authored-By: Claude Sonnet 4.6 --- src/orcapod/core/cached_function_pod.py | 129 +++++++++----------- src/orcapod/core/function_pod.py | 152 ++++++++++++++---------- src/orcapod/hooks.py | 7 +- 3 files changed, 148 insertions(+), 140 deletions(-) diff --git a/src/orcapod/core/cached_function_pod.py b/src/orcapod/core/cached_function_pod.py index 9c4116b4..3f43575a 100644 --- a/src/orcapod/core/cached_function_pod.py +++ b/src/orcapod/core/cached_function_pod.py @@ -9,12 +9,7 @@ from orcapod.core.datagrams import Datagram from orcapod.core.function_pod import WrappedFunctionPod from orcapod.core.result_cache import ResultCache -from orcapod.hooks import ( - InvocationStatus, - PodContext, - PostRunPayload, - RunStats, -) +from orcapod.hooks import InvocationStatus from orcapod.protocols.core_protocols import ( FunctionPodProtocol, DataProtocol, @@ -165,9 +160,12 @@ def _invoke_with_hooks( ) -> tuple[TagProtocol, DataProtocol | None]: """Override to detect cache hit status from ``RESULT_COMPUTED_FLAG`` meta. - Calls ``self.process_data()`` (which owns all cache lookup and store - logic), then reads ``RESULT_COMPUTED_FLAG`` from the output data meta to - determine ``InvocationStatus.HIT`` vs ``InvocationStatus.COMPUTED``. + When ``_post_run_hooks`` is empty, delegates directly to + ``process_data`` with zero overhead. Otherwise calls + ``self.process_data()`` (which owns all cache lookup and store logic), + reads ``RESULT_COMPUTED_FLAG`` from the output data meta to determine + ``InvocationStatus.HIT`` vs ``InvocationStatus.COMPUTED``, and fires + registered hooks. Args: tag: The tag associated with the data. @@ -177,49 +175,40 @@ def _invoke_with_hooks( Returns: A ``(tag, output_data)`` tuple. """ + if not self._post_run_hooks: + return self.process_data(tag, data, logger=logger) + started_at = datetime.now(timezone.utc) - exc: Exception | None = None out_tag = tag output_data: DataProtocol | None = None - status = InvocationStatus.COMPUTED try: out_tag, output_data = self.process_data(tag, data, logger=logger) if output_data is not None: meta = output_data.as_dict(columns=ColumnConfig(meta=True)) - if meta.get(self.RESULT_COMPUTED_FLAG) is False: - status = InvocationStatus.HIT - except Exception as e: - exc = e - status = InvocationStatus.ERROR + status = ( + InvocationStatus.HIT + if meta.get(self.RESULT_COMPUTED_FLAG) is False + else InvocationStatus.COMPUTED + ) + else: + status = InvocationStatus.COMPUTED + 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, + ) + ) + raise # bare raise — preserves the original traceback exactly finished_at = datetime.now(timezone.utc) - - if self._post_run_hooks: - record_id = ( - str(output_data.datagram_uuid) if output_data is not None else None + self._fire_post_run_hooks( + self._build_post_run_payload( + tag, data, output_data, started_at, finished_at, status, None, ) - payload = 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(), - ), - ) - self._fire_post_run_hooks(payload) - - if exc is not None: - raise exc + ) return out_tag, output_data async def _async_invoke_with_hooks( @@ -231,6 +220,9 @@ async def _async_invoke_with_hooks( ) -> tuple[TagProtocol, DataProtocol | None]: """Async counterpart of ``_invoke_with_hooks`` for ``CachedFunctionPod``. + 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. @@ -239,11 +231,12 @@ async def _async_invoke_with_hooks( Returns: A ``(tag, output_data)`` tuple. """ + if not self._post_run_hooks: + return await self.async_process_data(tag, data, logger=logger) + started_at = datetime.now(timezone.utc) - exc: Exception | None = None out_tag = tag output_data: DataProtocol | None = None - status = InvocationStatus.COMPUTED try: out_tag, output_data = await self.async_process_data( @@ -251,39 +244,29 @@ async def _async_invoke_with_hooks( ) if output_data is not None: meta = output_data.as_dict(columns=ColumnConfig(meta=True)) - if meta.get(self.RESULT_COMPUTED_FLAG) is False: - status = InvocationStatus.HIT - except Exception as e: - exc = e - status = InvocationStatus.ERROR + status = ( + InvocationStatus.HIT + if meta.get(self.RESULT_COMPUTED_FLAG) is False + else InvocationStatus.COMPUTED + ) + else: + status = InvocationStatus.COMPUTED + 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, + ) + ) + raise # bare raise — preserves the original traceback exactly finished_at = datetime.now(timezone.utc) - - if self._post_run_hooks: - record_id = ( - str(output_data.datagram_uuid) if output_data is not None else None + self._fire_post_run_hooks( + self._build_post_run_payload( + tag, data, output_data, started_at, finished_at, status, None, ) - payload = 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(), - ), - ) - self._fire_post_run_hooks(payload) - - if exc is not None: - raise exc + ) return out_tag, output_data def get_all_cached_outputs( diff --git a/src/orcapod/core/function_pod.py b/src/orcapod/core/function_pod.py index 09f0a5b8..aa0204c1 100644 --- a/src/orcapod/core/function_pod.py +++ b/src/orcapod/core/function_pod.py @@ -232,6 +232,51 @@ def _fire_post_run_hooks(self, payload: PostRunPayload) -> None: exc_info=True, ) + 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, + ) -> 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. + + Returns: + A ``PostRunPayload`` ready to pass to registered hooks. + """ + record_id = ( + str(output_data.datagram_uuid) if output_data is not None else None + ) + 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(), + ), + ) + def _invoke_with_hooks( self, tag: TagProtocol, @@ -241,10 +286,9 @@ def _invoke_with_hooks( ) -> tuple[TagProtocol, DataProtocol | None]: """Call ``process_data``, time it, and fire post-run hooks. - This is the call site used by ``FunctionPodStream`` and - ``async_execute``; ``process_data`` itself is unchanged. Override in - subclasses (e.g. ``CachedFunctionPod``) to supply a different - ``InvocationStatus``. + 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. @@ -254,45 +298,32 @@ def _invoke_with_hooks( Returns: A ``(tag, output_data)`` tuple. """ + if not self._post_run_hooks: + return self.process_data(tag, data, logger=logger) + started_at = datetime.now(timezone.utc) - exc: Exception | None = None out_tag = tag output_data: DataProtocol | None = None try: out_tag, output_data = self.process_data(tag, data, logger=logger) - status = InvocationStatus.COMPUTED - except Exception as e: - exc = e - status = InvocationStatus.ERROR + 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, + ) + ) + raise # bare raise — preserves the original traceback exactly finished_at = datetime.now(timezone.utc) - - if self._post_run_hooks: - record_id = ( - str(output_data.datagram_uuid) if output_data is not None else None + self._fire_post_run_hooks( + self._build_post_run_payload( + tag, data, output_data, started_at, finished_at, + InvocationStatus.COMPUTED, None, ) - payload = 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(), - ), - ) - self._fire_post_run_hooks(payload) - - if exc is not None: - raise exc + ) return out_tag, output_data async def _async_invoke_with_hooks( @@ -304,16 +335,22 @@ async def _async_invoke_with_hooks( ) -> 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``. + logger: Optional data execution logger forwarded to + ``async_process_data``. Returns: A ``(tag, output_data)`` tuple. """ + if not self._post_run_hooks: + return await self.async_process_data(tag, data, logger=logger) + started_at = datetime.now(timezone.utc) - exc: Exception | None = None out_tag = tag output_data: DataProtocol | None = None @@ -321,38 +358,23 @@ async def _async_invoke_with_hooks( out_tag, output_data = await self.async_process_data( tag, data, logger=logger ) - status = InvocationStatus.COMPUTED - except Exception as e: - exc = e - status = InvocationStatus.ERROR + 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, + ) + ) + raise # bare raise — preserves the original traceback exactly finished_at = datetime.now(timezone.utc) - - if self._post_run_hooks: - record_id = ( - str(output_data.datagram_uuid) if output_data is not None else None + self._fire_post_run_hooks( + self._build_post_run_payload( + tag, data, output_data, started_at, finished_at, + InvocationStatus.COMPUTED, None, ) - payload = 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(), - ), - ) - self._fire_post_run_hooks(payload) - - if exc is not None: - raise exc + ) return out_tag, output_data def handle_input_streams(self, *streams: StreamProtocol) -> StreamProtocol: diff --git a/src/orcapod/hooks.py b/src/orcapod/hooks.py index 1d0d40ec..5d88b35f 100644 --- a/src/orcapod/hooks.py +++ b/src/orcapod/hooks.py @@ -38,10 +38,13 @@ class RunStats: """Timing and status information for a single pod invocation. Attributes: - duration_ms: Wall-clock time in milliseconds. + duration_ms: Wall-clock milliseconds elapsed during the compute-or-lookup + step only. Hook execution time is not included. status: Whether the result was freshly computed, a cache hit, or an error. started_at: UTC timestamp when the invocation started. - finished_at: UTC timestamp when the invocation finished (after hooks fire). + finished_at: UTC timestamp when the compute-or-lookup step completed, + before hooks fire. Together with ``started_at`` it gives the raw + compute time, independent of hook overhead. error: The exception raised, if ``status == ERROR``; ``None`` otherwise. """ From 6b5f1a29b681ced3fc87e14ba8d4f81cb46090bf Mon Sep 17 00:00:00 2001 From: "agent-kurodo[bot]" <268466204+agent-kurodo[bot]@users.noreply.github.com> Date: Tue, 14 Jul 2026 04:56:39 +0000 Subject: [PATCH 10/10] perf(cached_function_pod): use get_meta_value to read RESULT_COMPUTED_FLAG (ITL-523) Replace output_data.as_dict(columns=ColumnConfig(meta=True)) with output_data.get_meta_value(RESULT_COMPUTED_FLAG) in both sync and async _invoke_with_hooks overrides. The old approach materialized the full output datagram dict (data + meta) just to read a single boolean flag; get_meta_value reads directly from the internal _meta dict in O(1) without any allocation. Also removes the now-unused ColumnConfig import from the file. Co-Authored-By: Claude Sonnet 4.6 --- src/orcapod/core/cached_function_pod.py | 7 ++----- 1 file changed, 2 insertions(+), 5 deletions(-) diff --git a/src/orcapod/core/cached_function_pod.py b/src/orcapod/core/cached_function_pod.py index 3f43575a..65e24f5c 100644 --- a/src/orcapod/core/cached_function_pod.py +++ b/src/orcapod/core/cached_function_pod.py @@ -18,7 +18,6 @@ ) from orcapod.protocols.database_protocols import ArrowDatabaseProtocol from orcapod.protocols.observability_protocols import DataExecutionLoggerProtocol -from orcapod.types import ColumnConfig if TYPE_CHECKING: import pyarrow as pa @@ -185,10 +184,9 @@ def _invoke_with_hooks( try: out_tag, output_data = self.process_data(tag, data, logger=logger) if output_data is not None: - meta = output_data.as_dict(columns=ColumnConfig(meta=True)) status = ( InvocationStatus.HIT - if meta.get(self.RESULT_COMPUTED_FLAG) is False + if output_data.get_meta_value(self.RESULT_COMPUTED_FLAG) is False else InvocationStatus.COMPUTED ) else: @@ -243,10 +241,9 @@ async def _async_invoke_with_hooks( tag, data, logger=logger ) if output_data is not None: - meta = output_data.as_dict(columns=ColumnConfig(meta=True)) status = ( InvocationStatus.HIT - if meta.get(self.RESULT_COMPUTED_FLAG) is False + if output_data.get_meta_value(self.RESULT_COMPUTED_FLAG) is False else InvocationStatus.COMPUTED ) else: