refactor(ENG-379): make iter_packets() read-only, consolidate DB join sites

[kurodo3]

Summary

• iter_packets() strictly read-only: computation now only via run()/execute()/async_execute(); callers must explicitly trigger computation before iterating
• _load_cached_entries() helper: consolidates 4+ duplicated DB join sites into a single reusable method
• execute() rewrite: fires on_packet_start before each packet (fixes StatusObserver RUNNING records), removes incorrect concurrent asyncio.gather path (always sequential — async belongs in async_execute()), calls _update_modified_time() after completion so subsequent iter_packets() serves from in-memory cache without stale DB reload

Test Plan

• All 3179 tests pass (uv run pytest)
• test_fail_fast_aborts_and_records_status — RUNNING record written before crash
• test_explicit_sync_config_overrides_async_default — sync path used when config=SYNCHRONOUS
• test_cache_only_function_node_serves_cached_results — in-memory cache served after execute()
• test_definition_save_load_run_roundtrip — loaded pipeline calls run() before iter_packets()

🤖 Generated with Claude Code

[codecov]

Codecov Report

❌ Patch coverage is 96.25000% with 3 lines in your changes missing coverage. Please review.

Files with missing lines	Patch %	Lines
src/orcapod/core/nodes/function_node.py	96.25%	3 Missing ⚠️

📢 Thoughts on this report? Let us know!

[Copilot]

Pull request overview

This PR refactors FunctionNode iteration and caching semantics to make iter_packets() strictly read-only, moves all computation behind explicit run() / execute() / async_execute(), and consolidates repeated pipeline+result DB join logic into a single _load_cached_entries() helper.

Changes:

• Make FunctionNode.iter_packets() yield only already-computed results (no implicit computation); callers must explicitly compute via run() / execute() / async_execute().
• Consolidate duplicated DB join logic into _load_cached_entries() and switch _cached_output_packets to be keyed by pipeline entry_id (string).
• Update/extend tests and add design docs to reflect the new read-only iteration contract.

Reviewed changes

Copilot reviewed 15 out of 15 changed files in this pull request and generated 5 comments.

Show a summary per file

File	Description
src/orcapod/core/nodes/function_node.py	Core refactor: read-only iter_packets(), _load_cached_entries(), entry-id keyed cache, updated execute()/run()/as_table() behavior.
tests/test_core/nodes/test_function_node_iteration.py	New test suite for ENG-379 iteration semantics and cache behavior.
tests/test_pipeline/test_serialization.py	Ensure loaded pipeline explicitly calls run() before iterating packets.
tests/test_data/test_polars_nullability/test_function_node_nullability.py	Update nullability test to use _load_cached_entries() instead of removed DB-iteration helper.
tests/test_core/sources/test_derived_source.py	Update downstream derived-source test to explicitly compute before iterating.
tests/test_core/packet_function/test_executor.py	Adjust executor-path expectations now that run() triggers sync execute() semantics.
tests/test_core/function_pod/test_pipeline_hash_integration.py	Ensure nodes compute via run() before iter_packets().
tests/test_core/function_pod/test_function_pod_node.py	Update fixture to compute via run() before iterating.
tests/test_core/function_pod/test_function_pod_node_stream.py	Update many tests to call run() before iteration / as_table(), aligning with read-only iteration.
tests/test_core/function_pod/test_function_node_caching.py	Ensure nodes compute via run() prior to iter_packets() assertions.
tests/test_core/function_pod/test_function_node_attach_db.py	Ensure tests explicitly compute via run() before iterating after DB attach.
tests/test_channels/test_node_async_execute.py	Update patched internal signature to match _process_packet_internal(..., logger=...).
docs/superpowers/specs/2026-04-07-function-node-refactor-design.md	Design spec documenting the refactor rationale and intended behavior.
docs/superpowers/plans/2026-04-07-eng379-function-node-refactor.md	Implementation plan capturing tasks/steps and test migration guidance.

---

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

[Copilot]

In the cached-hit branch, the code appends result to output unconditionally. Since _cached_output_packets stores PacketProtocol | None (and get_cached_results() can return values from that store), result can be None, which violates the method contract (“excluding None outputs”) and can break callers expecting a PacketProtocol. Mirror the non-cached branch by only appending when result is not None (while still firing observer hooks as desired).

Suggested change

	output.append((tag_out, result))
	if result is not None:
	output.append((tag_out, result))

[kurodo3]

Fixed. Added if result is not None: guard in the cached-hit branch of execute() to match the non-cached branch, ensuring None outputs are never appended to the return list regardless of whether the result came from cache or fresh computation.

[Copilot]

get_cached_results() is annotated to return dict[str, tuple[TagProtocol, PacketProtocol]], but it returns values directly from _cached_output_packets, whose packet value is PacketProtocol | None. This can surface None packets to callers (and currently feeds into execute()), contradicting the docstring (“excluding None outputs”). Consider filtering out None packets in the returned mapping (and optionally treating cached None values as cache misses if recomputation/DB load should be allowed).

[kurodo3]

Fixed. get_cached_results() now filters out entries where the packet is None before returning, so callers always receive (TagProtocol, PacketProtocol) pairs. Stored None values are still retained in _cached_output_packets to prevent unnecessary DB re-fetches — they just won't surface to callers.

[Copilot]

run() doesn’t handle LoadStatus.READ_ONLY (PacketFunctionProxy): it will call execute() and then fail with PacketFunctionUnavailableError. It also doesn’t clear in-memory caches when self.is_stale, which can leave stale entries in the add-only _cached_output_packets and cause iter_packets() to yield a mix of old+new results after a rerun. Recommend: (1) guard READ_ONLY with a clear RuntimeError/no-op (similar to CACHE_ONLY), and (2) if is_stale then clear_cache() before computing to avoid retaining stale entries.

[kurodo3]

Fixed both issues. run() now treats READ_ONLY as a no-op (same as CACHE_ONLY) — proxy function pods cannot compute, so silently returning is the right behaviour. Also added if self.is_stale: self.clear_cache() before calling execute() so any stale in-memory entries are discarded before a fresh computation run.

[Copilot]

Test name is misleading: test_execute_concurrent_error_policy_continue uses LocalExecutor with supports_concurrent_execution false, so it exercises the sequential execute() path. Rename the test to avoid “concurrent” in the name, or switch to an executor that actually enables the concurrent path (if that behavior is intended to be covered here).

[kurodo3]

Fixed. Renamed to test_execute_error_policy_continue_skips_failures and added a clarifying note to the docstring that the test uses LocalExecutor (non-concurrent) and exercises the sequential execute() path.

[Copilot]

test_derived_source_matches_node_output in this file still collects node_results via node.iter_packets() without calling node.run() first. With ENG-379 making iter_packets() read-only, this will now yield an empty result set and the assertion will no longer validate the intended behavior. Update that test to call node.run() (or node.execute(...)) before comparing node vs DerivedSource outputs.

[kurodo3]

Fixed. Added node.run() before node.iter_packets() in test_derived_source_matches_node_output so the test properly exercises the read-only iteration contract (compute via run(), then iterate).

[kurodo3]

Review round addressed — commit d67aca1

All 5 Copilot comments resolved in a single follow-up commit:

#	Comment	Fix
1	Misleading test name (test_execute_concurrent_error_policy_continue used a non-concurrent executor)	Renamed to test_execute_error_policy_continue_skips_failures; added docstring note clarifying it exercises the sequential path via LocalExecutor
2	Cached-hit branch appends None unconditionally in execute()	Added if result is not None: guard to mirror the non-cached branch
3	test_derived_source_matches_node_output missing node.run()	Added node.run() before node.iter_packets()
4	run() missing READ_ONLY guard + stale cache not cleared	READ_ONLY now treated as no-op (like CACHE_ONLY); added if self.is_stale: self.clear_cache() before execute()
5	get_cached_results() could return None packets despite PacketProtocol annotation	Return dict now filters out entries where the packet is None; None values remain in _cached_output_packets as cache sentinels to avoid unnecessary DB re-fetches

All 3724 tests pass (uv run pytest tests/ test-objective/).

[Copilot]

Pull request overview

Copilot reviewed 15 out of 15 changed files in this pull request and generated 3 comments.

---

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

[Copilot]

In execute(), cache-hit detection relies on the dict returned by get_cached_results(). That method intentionally filters out entries whose cached packet is None, so any entry_id already present in _cached_output_packets with a None output will be treated as a cache miss and recomputed on subsequent execute() calls (potentially re-triggering side effects). Consider using _cached_output_packets membership for the cache-hit check (including None) and only filtering out None when appending to the returned output list; keep get_cached_results() primarily for DB hot-loading of missing entry_ids.

[kurodo3]

Fixed. execute() now uses _cached_output_packets directly for the cache-hit membership check (which includes None-output entries) and calls get_cached_results() only for its DB hot-load side effect (the returned dict is discarded). A None packet stored in the cache correctly prevents recomputation on subsequent execute() calls while still being excluded from the output list.

[Copilot]

This design doc describes an execute() “concurrent gather” path keyed off _executor_supports_concurrent(...). In the current implementation of FunctionNode.execute(), computation is always sequential and the async path is only in async_execute(). Please update this section to match the actual behavior (or clearly mark the concurrent path as out-of-scope/future work) so readers don’t implement against an outdated contract.

Suggested change

	if _executor_supports_concurrent(self._packet_function) and to_compute:
	inline concurrent gather (see below)
	else:
	for tag, pkt, eid in to_compute:
	try:
	_process_packet_internal(tag, pkt)
	except Exception as exc:
	log warning; on_packet_crash; re-raise if error_policy=="fail_fast"
	# Current behavior: execute() is always synchronous/sequential.
	# Any async/concurrent gather path is handled by async_execute()
	# and is out of scope for execute() in this design.
	for tag, pkt, eid in to_compute:
	try:
	_process_packet_internal(tag, pkt)
	except Exception as exc:
	log warning; on_packet_crash; re-raise if error_policy=="fail_fast"

[kurodo3]

Fixed. Replaced the stale concurrent-gather pseudocode in execute() section of the design doc with accurate sequential-only pseudocode that matches the implementation. Added an explicit note that execute() is always synchronous and that async/concurrent execution is exclusively in async_execute(). Also updated testing item 10 and the migration-notes section accordingly.

[Copilot]

This plan snippet still says pf.executor = LocalExecutor() # enables concurrent path and uses the old test name test_execute_concurrent_error_policy_continue. In the current code/tests, LocalExecutor is non-concurrent and the execute() coverage here is sequential (the test was renamed accordingly). Please update the plan so it matches the implemented behavior.

[kurodo3]

Fixed. Updated the plan doc in three places: (1) module docstring changed from "concurrent path via asyncio.gather" to "always sequential; async/concurrent path is only in async_execute()"; (2) test renamed from test_execute_concurrent_error_policy_continue to test_execute_error_policy_continue_skips_failures; (3) LocalExecutor comment changed from "enables concurrent path" to "non-concurrent; tests sequential execute() path". Also updated Task 9 heading, the pytest invocation, and the git commit message in the plan.

[kurodo3]

Review round 2 addressed — commit 289573f

All 3 Copilot comments resolved in a single follow-up commit:

#	Comment	Fix
1	execute() cache-hit check relies on filtered get_cached_results() return — None-output entries would be treated as cache misses and recomputed	execute() now uses self._cached_output_packets directly for membership check (includes None); get_cached_results() is still called but its return value is discarded — the side effect (DB hot-load into _cached_output_packets) is all that's needed
2	Design doc describes stale concurrent asyncio.gather path in execute()	Replaced pseudocode with accurate sequential-only pseudocode; added explicit note that execute() is always synchronous and that async/concurrent execution is exclusively in async_execute(); updated test item 10 and migration notes
3	Plan doc still says LocalExecutor() # enables concurrent path and uses old test name	Fixed module docstring, test name, LocalExecutor comment, Task 9 heading, pytest invocation, and git commit message in the plan

All 3724 tests pass (uv run pytest tests/ test-objective/).

[eywalker]

Shouldn't you be able to specify nullability on the arrays and skip having to specify it on the schema?

[kurodo3]

PyArrow doesn't expose nullability at the array level — nullable is a property of pa.field only, not pa.array. There's no pa.array(..., nullable=False) API. The explicit schema=pa.schema([...]) block is the standard PyArrow idiom for enforcing non-nullability on a raw table, so I've kept it as-is.