feat(function_pod): unify async_execute() concurrency paths (ITL-516)

[kurodo3]

Summary

• Moves async_execute() from FunctionPod to _FunctionPodBase so all pod types (including CachedFunctionPod) inherit a single authoritative dispatch loop with observer hooks (on_data_start, on_data_end(cached=), on_data_crash)
• Redesigns FunctionJobNode.async_execute() DB path as a 3-stage concurrent pipeline (route_inputs → execution_pod.async_execute() → record_and_forward) using bounded intermediate channels for backpressure; no-DB path now delegates entirely to self._function_pod.async_execute()
• Removes the orphan _async_execute_one_data method and duplicate semaphore+TaskGroup loop; adds defensive ignore_missing=True on all drop_meta_columns(_TAG_NODE_INPUT_REF) calls to guard against future call-site changes

Test plan

• tests/test_core/nodes/test_function_job_node_async_execute.py — 11 new tests covering no-DB delegation, DB cache hits/misses, correlation key absence from output tags, pipeline record writes, observer cached flag, ephemeral path, and semaphore-not-leaked on crash
• tests/test_core/function_pod/test_function_pod_async_execute.py — observer hooks and backpressure via _FunctionPodBase.async_execute()
• Full test suite: 4288 passed, 62 skipped, 6 xfailed (zero failures)

Linear

Fixes ITL-516
ITL-512

[codecov]

Codecov Report

❌ Patch coverage is 94.59459% with 6 lines in your changes missing coverage. Please review.

Files with missing lines	Patch %	Lines
src/orcapod/core/nodes/function_node.py	93.58%	5 Missing ⚠️
src/orcapod/core/function_pod.py	96.96%	1 Missing ⚠️

📢 Thoughts on this report? Let us know!

[Copilot]

Pull request overview

This PR refactors async execution so there is a single authoritative concurrency/backpressure loop for function pods, and FunctionJobNode.async_execute() becomes a DB-aware orchestrator that routes cache hits/misses through a bounded 3-stage pipeline.

Changes:

• Moved async_execute() to _FunctionPodBase and added per-item observer hooks (on_data_start, on_data_end(cached=...), on_data_crash) so all pod types share one dispatch loop.
• Reworked FunctionJobNode.async_execute() into a concurrent route → compute → record pipeline with bounded intermediate channels and correlation-key stamping/stripping.
• Updated and added tests to cover no-DB delegation, DB cache hit/miss behavior, observer semantics, correlation-key stripping, and backpressure.

Reviewed changes

Copilot reviewed 9 out of 9 changed files in this pull request and generated 2 comments.

Show a summary per file

File	Description
src/orcapod/core/function_pod.py	Centralizes async dispatch logic in _FunctionPodBase.async_execute() and introduces pod_config plumbing + observer hooks.
src/orcapod/core/nodes/function_node.py	Replaces duplicated async dispatch with delegation (no-DB) and a DB-aware 3-stage async pipeline with correlation stamping and stripping.
src/orcapod/protocols/core_protocols/function_pod.py	Extends the protocol with a pod_config property for consistent concurrency configuration.
tests/test_core/test_regression_fixes.py	Removes the now-orphaned backpressure regression test and renumbers the regression list.
tests/test_core/nodes/test_function_job_node_async_execute.py	New comprehensive tests for FunctionJobNode.async_execute() across no-DB, DB, cache hit/miss, observer, and ephemeral modes.
tests/test_core/function_pod/test_function_pod_async_execute.py	New tests for _FunctionPodBase.async_execute() observer hooks and backpressure/concurrency limits.
tests/test_channels/test_node_async_execute.py	Updates an assertion test to verify no-DB delegation to the pod’s async processing path.
superpowers/specs/2026-07-09-itl-516-unify-async-execute-design.md	Design spec documenting the unified async execution model and invariants.
superpowers/plans/2026-07-10-itl-516-unify-async-execute.md	Implementation plan for the refactor and associated tests/cleanup.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[kurodo3]

Fixed in commit 20a2c99.

_NodeLabelObserver.on_data_crash now pops the correlation key from input_store after forwarding the event to the external observer:

correlation_key = tag.get_meta_value(_TAG_NODE_INPUT_REF, None)
if correlation_key is not None:
    input_store.pop(correlation_key, None)

A regression test test_crashed_items_cleaned_from_input_store confirms that a node with intermittent crashes completes correctly and produces the right results on a second run (where non-crashed items are cache hits).

[kurodo3]

Fixed in commit 20a2c99. Two issues addressed:

input_store accumulation: _NodeLabelObserver.on_data_end now pops the correlation key from input_store when out is None.

Missing _cached_output_datas update: The old _async_process_data_internal always wrote self._cached_output_datas[base_entry_id] = (tag_out, output_data) even for filtered items, so the sync iter_data() path saw the cached None result. The new pipeline path never called record_and_forward() for filtered items, so this update was lost. Now on_data_end does it directly when out is None:

if out is None:
    correlation_key = tag.get_meta_value(_TAG_NODE_INPUT_REF, None)
    if correlation_key is not None:
        original = input_store.pop(correlation_key, None)
        if original is not None:
            orig_tag, orig_data = original
            base_entry_id = fn_node.compute_base_entry_id(orig_tag, orig_data)
            fn_node._cached_output_datas[base_entry_id] = (clean, None)
            fn_node._cached_output_table = None
            fn_node._cached_content_hash_column = None

Note: filtered items are still not written to the pipeline DB (consistent with the old code — add_pipeline_record was also not called for None outputs in _async_process_data_internal). The async route_inputs() path therefore still recomputes filtered items on subsequent async_execute calls, same as before.

A regression test test_filtered_items_update_in_memory_cache verifies both the zero-output behaviour and the _cached_output_datas state.

[kurodo3]

Review response — commit 20a2c99

Addressed both Copilot comments from the automated review:

Comment 1: Stale input_store entries on computation crash

_NodeLabelObserver.on_data_crash now pops the correlation key from input_store after forwarding to the external observer. This prevents the dict from accumulating entries for crashed items over a long-running call.

New regression test: test_crashed_items_cleaned_from_input_store

Comment 2: Filtered items leave dangling input_store entries and skip _cached_output_datas update

_NodeLabelObserver.on_data_end now handles the out is None case by:

1. Popping the correlation key from input_store
2. Writing (clean_tag, None) into _cached_output_datas — restoring the behaviour of the old _async_process_data_internal which always stored the result even when None

One nuance: filtered items are intentionally not written to the pipeline DB (no add_pipeline_record call), matching the old code's behaviour. The async route_inputs() path therefore still recomputes filtered inputs on subsequent calls (they aren't in cached_by_base_entry_id which comes from the pipeline DB). The _cached_output_datas update is a correctness restoration for the sync iter_data() path.

New regression test: test_filtered_items_update_in_memory_cache

Both new tests are in tests/test_core/nodes/test_function_job_node_async_execute.py. Full suite: 2724 passed, 0 failures.

[eywalker]

why is this necessary? If it's not clear function pod would contain pod_config, this should be FunctionPodProtocol should be amended with pod_config

[kurodo3]

Fixed in commit 29000d3. FunctionPodProtocol already has pod_config (added in this PR), so the getattr fallback was unnecessary. Changed to a direct call:

@property
def pod_config(self) -> PodConfig:
    return self._function_pod.pod_config

[Copilot]

Pull request overview

Copilot reviewed 9 out of 9 changed files in this pull request and generated 2 comments.

[kurodo3]

Fixed in commit e82a062. The test now patches async_call instead of call, so any recomputation via the async path would increment call_count and fail the assertion:

original_async_call = node._function_pod.data_function.async_call

async def counted_async_call(data, **kw):
    nonlocal call_count
    call_count += 1
    return await original_async_call(data, **kw)

node._function_pod.data_function.async_call = counted_async_call

[kurodo3]

Fixed in commit e82a062. The comment now correctly describes the scope of the update:

Update the in-memory cache so iter_data() (the sync path) sees a cached None result — restoring the old _async_process_data_internal behaviour of always writing (tag, None) to _cached_output_datas. Note: route_inputs() checks cached_by_base_entry_id (pipeline DB) and filtered items are never written to the pipeline DB, so subsequent async_execute() calls still re-send filtered inputs to the pod.

[kurodo3]

Review response — commit 29000d3

WrappedFunctionPod.pod_config — removed getattr fallback

The defensive getattr(self._function_pod, "pod_config", PodConfig()) was left over from before pod_config was added to FunctionPodProtocol. Since the protocol now declares the property, self._function_pod.pod_config is a direct, type-safe call with no need for a fallback.

[kurodo3]

Review response — commit e82a062

Addressed both Copilot comments from the second automated review:

Comment 1: Test patches .call but async path uses .async_call

test_cache_hits_emitted_without_recomputation previously patched data_function.call. Since async_execute() dispatches through data_function.async_call(), any spurious recomputation via the async path would have gone undetected. Fixed by patching async_call instead.

Comment 2: Comment overstated the effect of the _cached_output_datas update

The comment in _NodeLabelObserver.on_data_end claimed the update prevented recomputation on the next call. That's only true for the sync iter_data() path. The async route_inputs() checks cached_by_base_entry_id (populated from the pipeline DB), and filtered items are never written to the pipeline DB, so the async path still re-sends them to the pod on subsequent calls. Comment updated to accurately describe both behaviors.

[Copilot]

Pull request overview

Copilot reviewed 9 out of 9 changed files in this pull request and generated 2 comments.

Comments suppressed due to low confidence (1)

src/orcapod/protocols/core_protocols/function_pod.py:34

• FunctionJobNode.async_execute() now calls self._function_pod.async_execute(...), but FunctionPodProtocol does not declare an async_execute method. This makes the new delegation path invisible to static type-checking and weakens the pod/node contract. Consider adding async_execute to FunctionPodProtocol (even if typed as Any/forward refs) so the protocol matches actual usage.

    @property
    def pod_config(self) -> PodConfig:
        """Per-pod executor configuration."""
        ...

    def process_data(
        self, tag: TagProtocol, data: DataProtocol
    ) -> tuple[TagProtocol, DataProtocol | None]: ...

    async def async_process_data(
        self, tag: TagProtocol, data: DataProtocol
    ) -> tuple[TagProtocol, DataProtocol | None]: ...

[kurodo3]

Fixed in commit 603443f — same fix as the pod-level test: decrement moved into a finally block so the counter spans the full coroutine duration including the underlying original_async_call.

[kurodo3]

Fixed in commit 603443f. The decrement is now in a finally block wrapping both the sleep and the underlying call, so concurrent_count stays elevated for the full task duration:

async def patched(data, **kw):
    nonlocal concurrent_count, max_observed
    concurrent_count += 1
    max_observed = max(max_observed, concurrent_count)
    try:
        await asyncio.sleep(0.01)
        return await original_async_call(data, **kw)
    finally:
        concurrent_count -= 1

[kurodo3]

Review response — commit 603443f

Concurrent-count decrement before await (both test files)

Both concurrency tests previously decremented concurrent_count before await original_async_call(...), so any overlap that occurred during the underlying async call was invisible to the measurement. Fixed in both files by wrapping the sleep + underlying call in a try/finally so the counter stays elevated for the full task duration:

• tests/test_core/function_pod/test_function_pod_async_execute.py — TestFunctionPodBaseAsyncExecuteBackpressure
• tests/test_core/nodes/test_function_job_node_async_execute.py — TestFunctionJobNodeAsyncExecuteNoDB::test_max_concurrency_limits_tasks