refactor: decouple pipeline_path from nodes + observer contextualization (ENG-340 / ENG-349)

[kurodo3]

Summary

• ENG-340: Removes pipeline_path from FunctionNode / OperatorNode entirely — the pipeline now pre-scopes each database via db.at(*pipeline_name) at compile() time and passes scoped views to nodes
• ENG-349: Adds contextualize(*identity_path) to ExecutionObserverProtocol; StatusObserver / LoggingObserver / CompositeObserver are rewritten to receive a pre-scoped DB at construction and produce per-node wrappers via contextualize() — eliminating pipeline_path from all observer hooks
• Orchestrators (SyncOrchestrator, AsyncOrchestrator) are now fully stateless w.r.t. observer — observer is passed only to run(), not stored at __init__
• Full to_config / from_config round-trip for all observers; Pipeline.save(level='full') now serializes the default observer
• ObservabilityReader._discover_tables() updated to use _status / _log path keys
• CachedFunctionPod: removed record_path_prefix parameter

Test plan

• pytest tests/ -q --tb=short — 3020 passing, 23 pre-existing optional-dep failures (graphviz, ray)
• pytest tests/test_pipeline/test_integration_smoke.py -v — all 3 smoke tests pass
• pytest tests/test_databases/ — scoped DB tracking + serialization tests pass
• pytest tests/test_observability/ — observer protocol tests pass
• Confirm no pipeline_path= keyword arguments remain in src/
• Confirm save(level='full') round-trip includes observer config in JSON output

Linear

• Closes ENG-340
• Closes ENG-349

🤖 Generated with Claude Code

[codecov]

Codecov Report

❌ Patch coverage is 90.66059% with 41 lines in your changes missing coverage. Please review.

Files with missing lines	Patch %	Lines
src/orcapod/pipeline/status_observer.py	81.15%	13 Missing ⚠️
src/orcapod/pipeline/logging_observer.py	81.48%	10 Missing ⚠️
src/orcapod/pipeline/graph.py	88.88%	9 Missing ⚠️
src/orcapod/core/nodes/operator_node.py	88.37%	5 Missing ⚠️
src/orcapod/core/nodes/function_node.py	97.87%	1 Missing ⚠️
src/orcapod/databases/delta_lake_databases.py	97.05%	1 Missing ⚠️
src/orcapod/databases/in_memory_databases.py	95.23%	1 Missing ⚠️
src/orcapod/pipeline/serialization.py	95.83%	1 Missing ⚠️

📢 Thoughts on this report? Let us know!

[Copilot]

Pull request overview

This PR refactors pipeline execution to remove pipeline_path threading from nodes and observers by pre-scoping databases at Pipeline.compile() time, and by changing the observer API to contextualize via a variadic identity_path.

Changes:

• Replace node/observer pipeline_path scoping with pre-scoped DB views (db.at(*pipeline_name).at("_result"/"_status"/"_log")) and node_identity_path.
• Move orchestrator observer injection from __init__ to run() and update protocol signatures (contextualize(*identity_path); remove pipeline_path from hooks).
• Add scoped DB tracking/serialization and observer (de)serialization; update tests/docs accordingly.

Reviewed changes

Copilot reviewed 42 out of 43 changed files in this pull request and generated 12 comments.

Show a summary per file

File	Description
src/orcapod/pipeline/graph.py	Create scoped DB views + default observer at compile; add observer= to Pipeline.run; serialize observer at save(level="full").
src/orcapod/pipeline/sync_orchestrator.py	Make orchestrator stateless w.r.t. observer; accept observer= in run().
src/orcapod/pipeline/async_orchestrator.py	Same observer statelessness; accept observer= in run() / run_async().
src/orcapod/protocols/observability_protocols.py	Update observer protocol (contextualize(*identity_path); remove pipeline_path from hook signatures).
src/orcapod/pipeline/observer.py	Update NoOpObserver to match new protocol signatures.
src/orcapod/pipeline/status_observer.py	Rewrite status observer to use contextualized wrappers + flat storage + config round-trip.
src/orcapod/pipeline/logging_observer.py	Rewrite logging observer to use contextualized wrappers + flat storage + config round-trip.
src/orcapod/pipeline/composite_observer.py	Update composite observer contextualization and add observer config round-trip.
src/orcapod/pipeline/serialization.py	Add observer registry + scoped DB resolution support in resolve_database_from_config.
src/orcapod/pipeline/observability_reader.py	Update discovery to look for _status / _log path keys.
src/orcapod/core/nodes/function_node.py	Rename pipeline_path → node_identity_path; remove prefix plumbing; contextualize observer per node.
src/orcapod/core/nodes/operator_node.py	Same identity path + contextualized observer usage; remove prefix plumbing.
src/orcapod/core/cached_function_pod.py	Remove record_path_prefix; store results directly under pod.uri.
src/orcapod/databases/in_memory_databases.py	Add _root / _scoped_path tracking for at() and scoped to_config(db_registry=...).
src/orcapod/databases/delta_lake_databases.py	Add scoped tracking and shared pending state; adjust flush to support shared pending keys.
src/orcapod/databases/connector_arrow_database.py	Add scoped tracking for at().
src/orcapod/databases/noop_database.py	Add scoped tracking for at().
tests/test_pipeline/test_sync_orchestrator.py	Update orchestrator usage + observer contextualize signature changes.
tests/test_pipeline/test_orchestrator.py	Update async orchestrator observer injection and signatures; add signature assertion test.
tests/test_pipeline/test_pipeline.py	Update pipeline DB naming + add tests for scoped DBs and default observer behavior.
tests/test_pipeline/test_serialization.py	Rename function_database → result_database assertions and round-trip tests.
tests/test_pipeline/test_status_observer_integration.py	Update status observer tests for flat storage + serialization + contextualize guards.
tests/test_pipeline/test_logging_observer_integration.py	Update logging observer tests for flat storage + serialization + contextualize guards.
tests/test_pipeline/test_composite_observer.py	Update composite observer integration + serialization tests.
tests/test_pipeline/test_observability_reader.py	Update expected _status / _log directory layout; add discovery test.
tests/test_pipeline/test_node_protocols.py	Update test observers to contextualize(*identity_path).
tests/test_pipeline/test_node_descriptors.py	Update descriptor construction to use pre-scoped pipeline DB views.
tests/test_pipeline/test_integration_smoke.py	Add smoke tests for scoped DBs + save(level="full") observer serialization + default observer run.
tests/test_observability/test_observer_protocol.py	Add protocol compliance tests for NoOpObserver and new hook signatures.
tests/test_observability/init.py	Package marker for new observability tests.
tests/test_databases/test_scoped_database_tracking.py	New tests for _root / _scoped_path tracking across DBs.
tests/test_databases/test_scoped_database_serialization.py	New tests for scoped DB config format + registry-based resolution.
tests/test_core/nodes/test_node_execute.py	Update pipeline DB record reads to use node_identity_path.
tests/test_core/operators/test_operator_node.py	Update identity-path tests for operator nodes.
tests/test_core/operators/test_operator_node_non_active.py	Update DB record reads to use node_identity_path.
tests/test_core/operators/test_operator_node_attach_db.py	Update attach-db tests to assert node_identity_path.
tests/test_core/function_pod/test_pipeline_hash_integration.py	Update hash/identity-path assertions.
tests/test_core/function_pod/test_function_pod_node.py	Update identity-path assertions and docs.
tests/test_core/function_pod/test_function_node_caching.py	Update DB record reads to use node_identity_path.
tests/test_core/function_pod/test_function_node_attach_db.py	Update attach-db tests to assert node_identity_path.
tests/test_core/function_pod/test_cached_function_pod.py	Update cached pod record path expectations + assert removed param behavior.
docs/superpowers/specs/2026-04-01-eng340-eng349-design.md	Add/record the design spec for ENG-340/349.
docs/superpowers/plans/2026-04-01-eng340-eng349-decouple-pipeline-path.md	Add/record the implementation plan for ENG-340/349.

Comments suppressed due to low confidence (1)

src/orcapod/pipeline/observability_reader.py:68

• ObservabilityReader assumes the directory immediately after '_status' / '_log' is a node name. However, StatusObserver/LoggingObserver now write to DEFAULT_STATUS_PATH/DEFAULT_LOG_PATH (e.g., 'execution_status'/'execution_logs'), so this discovery will likely surface table names instead of actual node labels. If the new design is flat tables, consider updating discovery (and nodes) to derive node names from the _status_node_label/_log_node_label columns instead of path segments.

            if "_status" in parts:
                idx = parts.index("_status")
                if idx + 1 < len(parts):
                    node_name = parts[idx + 1]
                    self._status_tables.setdefault(node_name, []).append(
                        table_dir
                    )
            elif "_log" in parts:
                idx = parts.index("_log")
                if idx + 1 < len(parts):
                    node_name = parts[idx + 1]
                    self._log_tables.setdefault(node_name, []).append(
                        table_dir

---

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

[Copilot]

Pipeline.compile() sets pipeline_db=None when no pipeline_database is provided, but still unconditionally calls node.attach_databases(pipeline_database=pipeline_db, ...). This will raise at runtime for pipelines that should run compute-only without persistence. Guard attach_databases calls (and any DB-dependent wiring) behind if pipeline_db is not None, leaving nodes unattached when no DB is configured.

[kurodo3]

Fixed. The attach_databases() call in compile() is now guarded with if pipeline_db is not None, so pipelines without a database run compute-only with no attempt to wire storage.

[Copilot]

FunctionNode.node_identity_path currently returns an empty tuple when no pipeline database is attached. Identity paths are needed for observer contextualization and should be computable from the function URI + hashes even in DB-less runs; returning () will cause contextualize() guards to raise and makes DerivedSource IDs less informative. Consider always computing node_identity_path regardless of DB attachment (only pipeline_path should depend on DB prefix).

[kurodo3]

Fixed. The DB-attachment guard was removed — node_identity_path now always computes pf.uri + (schema_hash, instance_hash) from the live pod, even when no pipeline database is set. This also fixes DerivedSource IDs for DB-less pipelines.

[Copilot]

attach_databases() still accepts result_path_prefix and computes computed_result_path_prefix, but the value is unused after removing record_path_prefix scoping. This is a silent no-op parameter now; either remove result_path_prefix from the API or reintroduce explicit scoping behavior (e.g., by scoping the result_database view) so callers aren’t misled.

[kurodo3]

Fixed. result_path_prefix was removed from both attach_databases() and __init__(). The default scoping is now pipeline_database.at("_result"), handled at compile time rather than via a prefix parameter.

[Copilot]

OperatorNode.node_identity_path returns () when no pipeline database is attached. Like FunctionNode, the identity path should be computable without a DB (it’s a computation identity, not a storage prefix). Returning () can break observer contextualization (Composite/Status/Logging contextualize guards) for DB-less pipelines.

[kurodo3]

Fixed. Same removal applied to OperatorNode.node_identity_path — the DB-attachment guard is gone and the path is always computable from operator.uri + (schema_hash, instance_hash).

[Copilot]

LoggingObserver.get_logs() reads logs by introspecting private in-memory fields (_tables/_pending_batches). For non-in-memory databases (e.g., DeltaTableDatabase, ConnectorArrowDatabase) these fields won’t exist, so get_logs() will incorrectly return None even when logs were written. Since logs are written to DEFAULT_LOG_PATH, prefer self._db.get_all_records(DEFAULT_LOG_PATH) (and drop internal __record_id if present) so this works across all ArrowDatabaseProtocol implementations.

[kurodo3]

Fixed. get_logs() now uses self._db.list_sources() and iterates over matching paths with self._db.get_all_records(src) — fully protocol-level, no private field access. Falls back to get_all_records(DEFAULT_LOG_PATH) for DBs that don't implement list_sources.

[Copilot]

DeltaTableDatabase._flush_by_key() constructs local table paths via Path.joinpath(*abs_path_components) without applying _sanitize_path_component(). This reintroduces invalid filenames on Windows for identity-path components containing ':' (e.g., 'schema:...'). Consider reusing _get_table_uri() logic or sanitizing each component when building table_uri for local paths.

[kurodo3]

Fixed. _flush_by_key() now applies _sanitize_path_component() to each component when building the local table_uri, reusing the same sanitization logic that _get_table_uri() uses.

[Copilot]

_flush_by_key() invalidates the existing-IDs cache via _invalidate_cache(abs_path_components). Because _invalidate_cache calls _get_record_key(record_path) which prepends self._path_prefix, calling flush() on a scoped instance will double-prefix the cache key and leave the real entry marked clean. Consider invalidating by record_key directly (e.g., self._cache_dirty[record_key]=True) or delegating cache invalidation to the root instance.

Suggested change

	# invalidate record id cache (uses abs_path_components as the record_path)
	self._invalidate_cache(abs_path_components)
	# Invalidate record ID cache for this absolute record_key directly to
	# avoid double-prefixing when called from scoped instances.
	self._cache_dirty[record_key] = True

[kurodo3]

Fixed. Changed to self._cache_dirty[record_key] = True to invalidate by the already-computed absolute record_key directly, avoiding the double-prefix that _invalidate_cache() would introduce on scoped instances.

[Copilot]

DeltaTableDatabase.at() docstring says scoped instances do NOT share pending state, but the implementation now explicitly shares _pending_batches/_pending_record_ids with the root. Update the docstring to match the new behavior (or reconsider whether sharing is intended), since it affects correctness expectations around flush behavior.

[kurodo3]

Fixed. The at() docstring now correctly states that scoped instances share _pending_batches and _pending_record_ids with the root, making flushed writes visible across all scopes of the same root database.

[Copilot]

The create_packet_logger() docstring example still uses the old contextualize(node_hash, node_label) signature; the new protocol uses contextualize(*identity_path). Updating the example avoids confusion for API consumers.

Suggested change

	ctx_obs = observer.contextualize(node_hash, node_label)
	ctx_obs = observer.contextualize(*identity_path)

[kurodo3]

Already addressed in the current code — create_packet_logger() docstring uses contextualize(*identity_path) in both the module docstring and inline comments. The old contextualize(node_hash, node_label) signature no longer appears anywhere.

[Copilot]

tempfile is imported but never used in this test module (and ObservabilityReader is imported twice: module-level and inside the test). Removing unused/redundant imports will keep lint/static analysis clean.

[kurodo3]

Fixed. Removed the unused import tempfile and the redundant in-function from orcapod.pipeline.observability_reader import ObservabilityReader import (the module-level import is sufficient).

[eywalker]

now that we established that we'll make use of "at" logic for contextualizing database (if not explicitly provided), do we still need result_path_prefxi? Is it to offer a choice other than the default _result that'd be used if no result_database is provided? If so, this is not obvious from the call signature.

[kurodo3]

Fixed — has been removed entirely from both and . Since we now pre-scope the result database at compile time via , the prefix parameter serves no purpose and its removal eliminates the confusion.

[eywalker]

I don't think this is correct -- this would lead to results being dumped directly into pipeline database at the same level as the pipeline results. If result database is to be based on the pipeline database, we must provide subpath (e.g. _result) as was done before. Also, this should have been caught by the tests -- check and explain why this was not something tests had caught and add tests to catch this scenario.

[kurodo3]

Fixed. The default in attach_databases() is now result_database = pipeline_database.at("_result"), so results land in a dedicated sub-path rather than at the same level as pipeline records. On why tests didn't catch it: the relevant tests read back via get_all_records(node.node_identity_path) using the node's result DB directly, so they passed regardless of whether the result DB was a sub-scope or not. A dedicated test has been added in test_function_node_attach_db.py to assert that the default result DB is a scoped view of the pipeline DB.

[eywalker]

it makes no sense for an empty path to be returned when no database is set -- node identity path in itself is NOT persistent layer dependent whatsoever.

[kurodo3]

Fixed. node_identity_path now always computes from pod.uri + (schema_hash, instance_hash) regardless of whether a pipeline database is attached. The DB-attachment guard was removed entirely — the identity path is a pure function of the computation graph, not of the storage layer.

[eywalker]

who uses this? The name of pipeline_path is obviously confusing against the old definition and I also just don't see why such a property reaching into its pipeline_database's prefix properties (which is completely private) should even exist?

[kurodo3]

Agreed — the property was encapsulation-violating and the name was confusing against the old definition. pipeline_path has been removed entirely from FunctionNode. The pipeline name prefix now lives in the database's own path (via db.at(*pipeline_name) at compile time), so no node property needs to reach into _path_prefix.

[eywalker]

Again what's the point of creating this confusing and in principle encapsulation violating method?

[kurodo3]

Same removal applied to OperatorNode. The pipeline_path property is gone from both node types.

[eywalker]

remove quotes around type hint

[kurodo3]

Fixed. Added from __future__ import annotations to noop_database.py and removed the quoted 'NoOpArrowDatabase' self-reference.

[eywalker]

Do not make this Any -- be more type specific

[kurodo3]

Fixed. _default_observer is now typed as ExecutionObserverProtocol | None with a TYPE_CHECKING import to avoid the circular import.

[eywalker]

provider stronger typing based on observer protocol

[kurodo3]

Fixed. The observer parameter in both Pipeline.run() and the orchestrator call sites is typed as ExecutionObserverProtocol | None = None.

[eywalker]

attach database should behave similarly to how things would work on at the init

[kurodo3]

Fixed. compile() now guards the attach_databases() call behind if pipeline_db is not None, matching the behaviour you'd expect from __init__ — nodes are left unattached when no database is configured.

[eywalker]

not entirely comfortable absorbing all exceptions here -- should at least include TODO to re-evaluate the strategy

[kurodo3]

Done — added a # TODO: re-evaluate broad exception absorption comment to _write_event() noting that ideally the catch should be narrowed to DB-specific exceptions (IOError, ArrowInvalid) once the full set of failure modes is known.

[brian-arnold]

Code Review

Critical

1. LoggingObserver.get_logs() breaks database encapsulation — InMemoryArrowDatabase-only
The new get_logs() accesses db._tables and db._pending_batches directly via getattr instead of going through the database protocol (get_all_records). This will silently return empty results for DeltaTableDatabase or any other non-in-memory implementation. The old implementation used the protocol-level API. This is a regression.

2. Spec/implementation mismatch on identity_path usage in StatusObserver
Both StatusObserver._write_event and _ContextualizedStatusObserver._write_event write to DEFAULT_STATUS_PATH (a fixed constant), never using self._identity_path for DB path scoping. The node identity is only stamped as column values. The spec says writes should go to identity_path-scoped locations, but the implementation uses a flat table. This isn't necessarily wrong (column-based filtering works), but it means contextualize() doesn't actually scope the storage — only the metadata. This should be an intentional, documented decision.

3. Dead result_path_prefix parameter in FunctionNode.attach_databases()
The parameter is still accepted and computed_result_path_prefix is computed, but it's never passed anywhere since CachedFunctionPod no longer accepts record_path_prefix. Callers setting it get silent no-ops. Should be removed.

Moderate

4. LoggingObserver.create_packet_logger raises NotImplementedError on root observer
Unlike StatusObserver which returns a no-op logger from the root, LoggingObserver raises if create_packet_logger is called before contextualize(). This asymmetry is a runtime landmine if the observer is used standalone.

5. DeltaTableDatabase._flush_by_key uses bare except Exception
The except Exception: delta_table = None catch during table existence checks is overly broad — it treats network errors, permission errors, etc. as "table doesn't exist." This could mask real failures in production.

6. NoOpArrowDatabase.to_config() signature not updated for db_registry
Other DB classes were updated to accept db_registry=None, but NoOpArrowDatabase (and possibly ConnectorArrowDatabase/DeltaTableDatabase) weren't. Scoped NoOp instances won't serialize correctly through the observer pipeline.

Minor

7. pipeline_path property on nodes accesses private _path_prefix
Both FunctionNode and OperatorNode reconstruct absolute paths via getattr(self._pipeline_database, "_path_prefix", ()) — coupling to an implementation detail.

8. OBSERVER_REGISTRY triggers eager imports
Unlike the LazyModule pattern used elsewhere, serialization.py eagerly imports all observer classes at module load. Potential circular import risk.

9. Inconsistent empty identity_path validation
NoOpObserver.contextualize() accepts empty args, but all real observers raise ValueError. Code that works with NoOpObserver could break when swapped to a real observer.

---

Overall: The architectural direction is sound — pre-scoping databases eliminates a major source of parameter threading and makes the node/observer APIs cleaner. The critical items (especially #1) should be addressed before merge. The dead result_path_prefix (#3) is easy cleanup. The spec/implementation mismatch (#2) should at minimum be documented as an intentional design choice.

[kurodo3]

Update: All review comments addressed

Thanks for the thorough reviews (@eywalker, @brian-arnold, @copilot-pull-request-reviewer). All issues have been addressed across the subsequent commits. Here is a full accounting:

Critical / Major items

LoggingObserver.get_logs() DB encapsulation — Fixed. Uses list_sources() + get_all_records() protocol-level API with fallback. No more private field access.

Spec/implementation mismatch — observer path scoping — Fixed. contextualize(*identity_path) now calls self._db.at(*identity_path), so the database is path-scoped per node. _status_node_label/hash and _log_node_label/hash columns removed entirely.

Dead result_path_prefix parameter — Fixed. Removed from FunctionNode.attach_databases() and __init__().

LoggingObserver.create_packet_logger raises NotImplementedError — Fixed. Root observer now returns a working PacketLogger.

DeltaTableDatabase._flush_by_key broad exception — Narrowed to TableNotFoundError | DeltaProtocolError. TODO comment added to StatusObserver._write_event for remaining broad catch.

to_config() db_registry kwarg missing from some DBs — Fixed. All DB to_config() methods now accept db_registry=None.

pipeline_path accessing private _path_prefix — Fixed. Property removed entirely from FunctionNode and OperatorNode. Pipeline prefix lives in the pre-scoped DB path.

OBSERVER_REGISTRY eager imports — Fixed. Replaced with lazy _get_observer_registry() accessor.

result_database defaulting wrong — Fixed. attach_databases() defaults to pipeline_database.at("_result").

node_identity_path returns () without DB — Fixed. Always computable from pod.uri + (schema_hash, instance_hash) regardless of DB attachment.

Typing / style items

Type annotations in graph.py — Fixed. _default_observer: ExecutionObserverProtocol | None, observer: ExecutionObserverProtocol | None = None in run().

String quotes around type hints — Fixed across connector_arrow_database.py, in_memory_databases.py, noop_database.py via from __future__ import annotations.

db_registry type hint — Fixed with TYPE_CHECKING import of DatabaseRegistry.

Copilot items

attach_databases unconditional call in compile() — Fixed. Guarded with if pipeline_db is not None.

_flush_by_key path sanitization — Fixed. _sanitize_path_component() applied per component.

_flush_by_key cache invalidation — Fixed. Changed to self._cache_dirty[record_key] = True.

at() docstring — Fixed. States scoped instances share _pending_batches/_pending_record_ids.

Unused tempfile import — Fixed.

---

Final commit (d690da7) updates the 12 test references from pipeline_path → node_identity_path. Full suite: 3046 passed, 56 skipped.

[Copilot]

Pull request overview

Copilot reviewed 44 out of 45 changed files in this pull request and generated 8 comments.

---

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

[Copilot]

In compile(), an explicitly provided result_database is used unscoped (result_db = self._result_database ...). Since CachedFunctionPod now stores results under pod.uri with no prefix, sharing a result_database across multiple pipelines can cause cross-pipeline cache collisions (same function URI → same record_path). Consider scoping the provided result_database to the pipeline namespace (e.g. result_database.at(*self._name).at("_result")), or enforcing/validating that callers pass an already-scoped DB view.

[kurodo3]

No change here — intentional design. When a caller explicitly provides a result_database to compile(), it is used as-is; the caller is responsible for scoping it appropriately (e.g. db.at(pipeline_name, "_result")). Silently re-scoping an explicitly provided database would be surprising and could break callers who have already pre-scoped it. The default path (when no result_database is given) already scopes correctly to pipeline_db.at("_result"). The Pipeline docstring has been updated to document this expectation.

[Copilot]

Pipeline.load() sets pipeline._compiled = True but does not initialize the scoped database view fields (_scoped_pipeline_database, _result_database_scoped, _status_database, _log_database). As a result, properties like result_database, status_database, and log_database will return None on loaded pipelines even when databases are available. Load should populate these fields consistently (e.g. pipeline_db, pipeline_db.at("_result"), pipeline_db.at("_status"), pipeline_db.at("_log")) so the public properties work after deserialization.

Suggested change

	# Ensure scoped database views are restored on deserialized pipelines
	if "databases" in data:
	# `pipeline_db` is the pipeline-level database view constructed earlier
	pipeline._scoped_pipeline_database = pipeline_db
	if pipeline_db is not None:
	pipeline._result_database_scoped = pipeline_db.at("_result")
	pipeline._status_database = pipeline_db.at("_status")
	pipeline._log_database = pipeline_db.at("_log")

[kurodo3]

Fixed. Pipeline.load() now populates _scoped_pipeline_database, _result_database_scoped, _status_database, and _log_database from the resolved pipeline_db before setting _compiled = True — mirroring exactly what compile() does at runtime. The result_database, status_database, and log_database properties now return meaningful values on loaded pipelines.

[Copilot]

get_status() concatenates status tables across node identity subpaths but drops all node identity columns, so the returned combined Table cannot be attributed back to a node (especially since _status_node_label/_status_node_hash were removed). Consider adding an explicit node identity column while aggregating (e.g. derive identity_prefix = src[:-len(DEFAULT_STATUS_PATH)] and add it as a column before concat) or returning a mapping keyed by identity_path so consumers can filter per-node.

[kurodo3]

No code change — by design. StatusObserver.get_status() is a convenience aggregator for quick inspection of all status events across a run; it is intentionally flat. Per-node breakdown is the responsibility of ObservabilityReader, which discovers status tables by path and derives node identity from the path structure. Adding a _node_identity column to the flat aggregate would force us to define a serialization format for the path, which ObservabilityReader already handles properly. The get_status() docstring has been updated to make this distinction explicit.

[Copilot]

The exception handling in get_status() catches all Exceptions from list_sources() and silently falls back to a single-table read. This can mask real bugs (e.g. unexpected runtime errors in a DB implementation) and make failures harder to diagnose. Consider narrowing this to AttributeError/NotImplementedError (and possibly known DB capability errors), and letting other exceptions propagate or at least logging them before falling back.

Suggested change

	except (NotImplementedError, Exception):
	except (AttributeError, NotImplementedError):

[kurodo3]

Fixed. Narrowed except (NotImplementedError, Exception) to except (AttributeError, NotImplementedError) in both get_status() and get_logs(). The original clause with Exception made the NotImplementedError narrowing meaningless and could silently swallow real runtime bugs.

[Copilot]

get_logs() concatenates log tables across node identity subpaths but drops node identity columns, so the returned combined Table cannot be filtered per-node (and _log_node_label/_log_node_hash were removed). Consider adding a derived node identity column while aggregating (similar to ObservabilityReader adding node_label from the directory), or returning per-node/per-identity tables to avoid losing attribution.

[kurodo3]

No code change — by design, mirroring the same decision made for StatusObserver.get_status(). LoggingObserver.get_logs() is a flat convenience aggregator for quick inspection of all log rows across a run. Per-node breakdown is the responsibility of ObservabilityReader, which discovers log tables by path and derives node identity from the path structure. Adding a _node_identity column here would duplicate logic that ObservabilityReader already handles correctly. The get_logs() docstring has been updated to make this explicit.

[Copilot]

The exception handling in get_logs() catches all Exceptions from list_sources() and falls back to reading DEFAULT_LOG_PATH. Catching broad Exception here can hide real runtime errors in the DB implementation. Consider narrowing to capability-related exceptions (e.g. AttributeError/NotImplementedError) and logging unexpected exceptions rather than silently switching behavior.

Suggested change

	except (NotImplementedError, Exception):
	return self._db.get_all_records(DEFAULT_LOG_PATH)
	except (AttributeError, NotImplementedError):
	# Databases that do not support listing sources fall back to a direct read.
	return self._db.get_all_records(DEFAULT_LOG_PATH)
	except Exception:
	# Unexpected errors from the database should not be silently hidden.
	logger.exception("Unexpected error while listing log sources in LoggingObserver.get_logs")
	raise

[kurodo3]

Fixed. Narrowed except (NotImplementedError, Exception) to except (AttributeError, NotImplementedError) in get_logs(). The original clause with bare Exception made the NotImplementedError narrowing meaningless and could silently swallow real runtime bugs in DB implementations.

[Copilot]

This section header mentions filtering by _log_node_label, but the refactor removes _log_node_label/_log_node_hash columns and LoggingObserver.get_logs() currently returns a combined table without any node-identity column. The comment (and possibly the test name) should be updated to reflect the new way to identify per-node logs (e.g. include a derived node_label column during aggregation, or clarify that per-node filtering isn't supported by get_logs()).

[kurodo3]

Fixed. The section header has been updated to 'get_logs() returns all logs aggregated across all node sub-paths', and the test method test_get_logs_filters_by_node_label_column has been renamed to test_get_logs_returns_rows_for_all_nodes to reflect that node identity is path-based, not column-based.

[Copilot]

This comment says get_status() results can be filtered by a node_label column, but _status_node_label/_status_node_hash were removed and StatusObserver.get_status() currently returns an aggregated table without any node-identity column. Update the comment to match the actual API (or consider adding a node identity column during aggregation if per-node filtering is intended).

[kurodo3]

Fixed. The section header has been updated to 'get_status() returns combined status across all node sub-paths'. The _status_node_label / _status_node_hash columns no longer exist; node identity is encoded in the database path, not in column values.

[eywalker]

generally utilizing possibly non-existing property is not an ideal design. Can we not make this a field always present for this concrete implementation?

[kurodo3]

Fixed. _stored_pipeline_path: tuple[str, ...] = () is now initialized in __init__ on both FunctionNode and OperatorNode, making it a proper always-present instance field. node_identity_path accesses it directly as self._stored_pipeline_path — no getattr needed. from_descriptor() continues to overwrite it with the deserialized value when one is present in the descriptor.

[kurodo3]

Extended the fix to cover all descriptor-derived fields, not just _stored_pipeline_path. Both FunctionNode and OperatorNode now initialize every field that was previously set only by from_descriptor() — _load_status, _stored_content_hash, _stored_pipeline_hash, _stored_schema, _stored_node_uri, _stored_pipeline_path, _stored_result_record_path (FunctionNode only), and _descriptor — directly in __init__ with appropriate defaults. All getattr(self, "_stored_*", ...) fallback reads are replaced with direct attribute access.

[kurodo3]

Round 3 update (commit 28ab132)

Single fix this round:

_stored_pipeline_path is now a proper initialized field on both FunctionNode and OperatorNode.

Previously node_identity_path used getattr(self, "_stored_pipeline_path", None) to defensively access a field that was only set dynamically by from_descriptor(). This is now replaced with a proper __init__ declaration:

self._stored_pipeline_path: tuple[str, ...] = ()

node_identity_path accesses it directly as self._stored_pipeline_path — no getattr needed. from_descriptor() continues to overwrite it with the deserialized value when one is present in the descriptor. All 3046 tests pass.

[kurodo3]

Round 4 update (commit fe26874)

Scope expanded from previous round: The getattr-for-possibly-absent-field concern was originally raised against _stored_pipeline_path, which was fixed in round 3. The same principle applies to every other field set exclusively by from_descriptor(), so this round extends the fix to all of them.

Both FunctionNode and OperatorNode now initialize the full set of descriptor-derived fields directly in __init__:

Field	Default
_load_status	LoadStatus.FULL
_stored_content_hash	None
_stored_pipeline_hash	None
_stored_schema	{}
_stored_node_uri	()
_stored_pipeline_path	()
_stored_result_record_path	() (FunctionNode only)
_descriptor	{}

All getattr(self, "_stored_*", ...) fallback reads throughout both classes are replaced with direct self._stored_* attribute access. from_descriptor() continues to overwrite these with deserialized values when present. All 3046 tests pass.