From 56c01365dfa362a821bc33ab930e0661db55bc18 Mon Sep 17 00:00:00 2001 From: Emile Riberdy <38288472+magix022@users.noreply.github.com> Date: Wed, 22 Jul 2026 14:20:18 +0000 Subject: [PATCH 1/2] Add first-class skipped node outcomes --- CHANGELOG.md | 8 + docs/dag-api.md | 42 ++++ src/avalanche/__init__.py | 4 +- src/avalanche/dag.py | 213 ++++++++++++++++---- src/avalanche/types.py | 33 ++- src/runtime/executor.py | 28 ++- src/runtime/operator/convert.py | 5 + src/runtime/operator/hooks.py | 1 + src/runtime/operator/models.py | 4 +- src/runtime/operator/operator.py | 13 +- src/runtime/operator/proto/operator.proto | 2 + src/runtime/operator/proto/operator_pb2.py | 26 +-- src/runtime/operator/proto/operator_pb2.pyi | 8 +- src/runtime/operator/run_worker.py | 9 + src/tui/widgets/log_panel.py | 23 ++- test/fixtures/sample_workflows.py | 15 ++ test/operator_tests/test_operator.py | 23 +++ test/skipped_outcome_test.py | 170 ++++++++++++++++ test/tui_test.py | 25 +++ 19 files changed, 583 insertions(+), 69 deletions(-) create mode 100644 test/skipped_outcome_test.py diff --git a/CHANGELOG.md b/CHANGELOG.md index d1e7cdf..fe8954f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,6 +1,14 @@ # Changelog ## Unreleased +### First-class skipped node outcomes + +- Added `ava.skip(reason, metadata=None)` as a successful, non-value node + outcome across local, Ray, operator, protobuf, and TUI paths. +- Authored skips record `SKIPPED`, reason, metadata, and timestamps while + satisfying downstream dependencies; skipped fan-in slots are omitted and + contribute no persisted row or lineage entry. + ### Worker execution services diff --git a/docs/dag-api.md b/docs/dag-api.md index 7cc376b..44f1178 100644 --- a/docs/dag-api.md +++ b/docs/dag-api.md @@ -106,6 +106,48 @@ load_documents() >> (validate_chunks() & build_index()) >> publish_report() load_documents() >> validate_chunks() & build_index() ``` +## Intentional skips + +Return `ava.skip(reason, metadata=None)` when a node successfully determines +that it has no value to produce. A skip does not fail the run. The operator +records the node as `SKIPPED` with its reason, optional metadata, and lifecycle +timestamps; the TUI shows the authored reason and metadata. + +Skipped results satisfy dependency ordering but do not occupy a data slot. +This makes `&` the fan-in mechanism for optional branches: downstream implicit +binding receives only values produced by non-skipped branches. A downstream +node whose inputs may all skip should declare an appropriate default and may +return its own skip before performing persistence. + +```python +@ava.source +def optional_rows(): + rows = fetch_optional_rows() + if not rows: + return ava.skip("No rows for partition", {"partition": "2026-07-22"}) + return rows + +@ava.source +def required_rows(): + return fetch_required_rows() + +@ava.dest +def persist(rows=None): + if rows is None: + return ava.skip("Nothing to persist", {"rows": 0}) + return table.append(rows) + +@ava.workflow +def optional_flow(): + return (optional_rows() & required_rows()) >> persist() +``` + +Skipped nodes produce no persisted row and contribute no value or producer +entry to downstream lineage. Reruns execute them normally when selected; a +fresh skip remains a successful no-value outcome. If a skipped node is the +workflow return, `RunHandle.result()` returns `None`. + + ## Multi-return nodes Declare `num_returns` when a node returns multiple values that downstream nodes diff --git a/src/avalanche/__init__.py b/src/avalanche/__init__.py index 170ff44..85db973 100644 --- a/src/avalanche/__init__.py +++ b/src/avalanche/__init__.py @@ -61,7 +61,7 @@ from .storage import Namespace, NamespaceConfig, ScanResult, Table, TableGroup # Types -from .types import AppendResult, SnapshotMetadata, SnapshotState +from .types import AppendResult, SkipOutcome, SnapshotMetadata, SnapshotState, skip __version__ = "0.1.0rc1" @@ -95,6 +95,7 @@ def __getattr__(name: str): "workflow", "pipeline", "input", + "skip", # Workflow "Workflow", "Pipeline", @@ -126,6 +127,7 @@ def __getattr__(name: str): "ProgressStore", # Types "AppendResult", + "SkipOutcome", "SnapshotState", "SnapshotMetadata", "Json", diff --git a/src/avalanche/dag.py b/src/avalanche/dag.py index e75d3ad..d24ccce 100644 --- a/src/avalanche/dag.py +++ b/src/avalanche/dag.py @@ -1476,6 +1476,10 @@ def _unwrap_lineaged_tree(value: Any) -> Any: Only unwraps envelopes; other containers/values pass through unchanged so user args keep their structure. """ + from .types import SkipOutcome + + if isinstance(value, SkipOutcome): + return None from .types import LineagedResult if isinstance(value, LineagedResult): @@ -1552,17 +1556,20 @@ def _wrap_lineaged_result(value: Any, context: Any, *, num_returns: int) -> Any: Multi-return nodes wrap each item individually so executor multi-return (e.g. Ray) still sees the expected number of results. """ - from .types import LineagedResult + from .types import LineagedResult, SkipOutcome lineage = dict(context.lineage_vector) if context.node_slug is not None: lineage[context.node_slug] = context.run_id + def wrap(item: Any) -> Any: + return item if isinstance(item, SkipOutcome) else LineagedResult(item, lineage) + if num_returns > 1 and isinstance(value, tuple): - return tuple(LineagedResult(item, lineage) for item in value) + return tuple(wrap(item) for item in value) if num_returns > 1 and isinstance(value, list): - return [LineagedResult(item, lineage) for item in value] - return LineagedResult(value, lineage) + return [wrap(item) for item in value] + return wrap(value) _STREAM_PARENT_KWARG_PREFIX = "__ava_stream_parent_" @@ -1622,8 +1629,82 @@ def get_data(ref: Any) -> Any: return materialize_append_handles(value, get_data) +_IMPLICIT_VALUE_KWARG_PREFIX = "__ava_implicit_value_" + + +def _bind_deferred_implicit_values( + args: tuple[Any, ...], + kwargs: dict[str, Any], + *, + binding_fn: Callable[..., Any] | None, + skip_param_names: set[str], + position_consuming_param_names: set[str], + adapt_positionals: bool, +) -> tuple[tuple[Any, ...], dict[str, Any]]: + """Bind implicit inputs after distributed refs have materialized worker-side.""" + if binding_fn is None: + return args, kwargs + + from .types import skip_outcome_from_result + + implicit_keys = sorted( + ( + key + for key in kwargs + if isinstance(key, str) and key.startswith(_IMPLICIT_VALUE_KWARG_PREFIX) + ), + key=lambda key: int(key.removeprefix(_IMPLICIT_VALUE_KWARG_PREFIX)), + ) + upstream_values = [kwargs.pop(key) for key in implicit_keys] + value_inputs = [ + value for value in upstream_values if skip_outcome_from_result(value) is None + ] + implicit_args, implicit_kwargs = _bind_implicit_parent_results( + binding_fn, + value_inputs, + skip_param_names, + position_consuming_param_names, + kwargs, + adapt_positionals=adapt_positionals, + ) + kwargs.update(implicit_kwargs) + return (*args, *implicit_args), kwargs + + +def _prepare_user_call( + args: tuple[Any, ...], kwargs: dict[str, Any] +) -> tuple[tuple[Any, ...], dict[str, Any]]: + """Remove skipped top-level data slots while preserving dependency-only kwargs.""" + from .types import skip_outcome_from_result + + filtered_args = tuple(value for value in args if skip_outcome_from_result(value) is None) + filtered_kwargs = { + key: value + for key, value in kwargs.items() + if (isinstance(key, str) and key.startswith(_STREAM_PARENT_KWARG_PREFIX)) + or skip_outcome_from_result(value) is None + } + return filtered_args, filtered_kwargs + + +def _shape_skipped_result(value: Any, *, num_returns: int) -> Any: + """Give whole-node skips the physical shape required by multi-return executors.""" + from .types import SkipOutcome + + if isinstance(value, SkipOutcome) and num_returns > 1: + return tuple(value for _ in range(num_returns)) + return value + + def _with_current_run_context( - fn: Callable[..., Any], context: Any, *, num_returns: int = 1 + fn: Callable[..., Any], + context: Any, + *, + num_returns: int = 1, + implicit_binding_fn: Callable[..., Any] | None = None, + implicit_skip_param_names: set[str] | None = None, + implicit_position_consuming_param_names: set[str] | None = None, + adapt_implicit_positionals: bool = False, ) -> Callable[..., Any]: """Wrap a node function so framework helpers can read its RunContext. @@ -1641,6 +1722,18 @@ def _with_current_run_context( from .runtime import RunContext from .runtime.context import _run_with_context + def expose_framework_signature(wrapped_fn: Callable[..., Any]) -> Callable[..., Any]: + if implicit_binding_fn is not None: + import inspect + + wrapped_fn.__signature__ = inspect.Signature( + parameters=( + inspect.Parameter("args", inspect.Parameter.VAR_POSITIONAL), + inspect.Parameter("kwargs", inspect.Parameter.VAR_KEYWORD), + ) + ) + return wrapped_fn + if not isinstance(context, RunContext): # Still unwrap parent envelopes so non-context nodes never receive an # internal transport type. @@ -1648,9 +1741,21 @@ def _with_current_run_context( def plain(*args: Any, **kwargs: Any) -> Any: args = _materialize_append_handles_for_worker(args) kwargs = _materialize_worker_kwargs(kwargs) - return fn(*_unwrap_lineaged_tree(args), **_unwrap_lineaged_tree(kwargs)) + args, kwargs = _bind_deferred_implicit_values( + args, + kwargs, + binding_fn=implicit_binding_fn, + skip_param_names=implicit_skip_param_names or set(), + position_consuming_param_names=( + implicit_position_consuming_param_names or set() + ), + adapt_positionals=adapt_implicit_positionals, + ) + args, kwargs = _prepare_user_call(args, kwargs) + result = fn(*_unwrap_lineaged_tree(args), **_unwrap_lineaged_tree(kwargs)) + return _shape_skipped_result(result, num_returns=num_returns) - return plain + return expose_framework_signature(plain) @wraps(fn) def wrapped(*args: Any, **kwargs: Any) -> Any: @@ -1667,13 +1772,23 @@ def wrapped(*args: Any, **kwargs: Any) -> Any: merged = dict(context.lineage_vector) merged.update(parent_lineage) context.lineage_vector = merged + args, kwargs = _bind_deferred_implicit_values( + args, + kwargs, + binding_fn=implicit_binding_fn, + skip_param_names=implicit_skip_param_names or set(), + position_consuming_param_names=(implicit_position_consuming_param_names or set()), + adapt_positionals=adapt_implicit_positionals, + ) + args, kwargs = _prepare_user_call(args, kwargs) unwrapped_args = _unwrap_lineaged_tree(args) unwrapped_kwargs = _unwrap_lineaged_tree(kwargs) result = _run_with_context(context, fn, *unwrapped_args, **unwrapped_kwargs) + result = _shape_skipped_result(result, num_returns=num_returns) return _wrap_lineaged_result(result, context, num_returns=num_returns) - return wrapped + return expose_framework_signature(wrapped) def _inspect_runtime_params( @@ -2313,7 +2428,12 @@ def _run_driver( or self.returns is None or ( hooks - and (hooks.on_node_success or hooks.on_node_failure or hooks.unwrap_result) + and ( + hooks.on_node_success + or hooks.on_node_failure + or hooks.on_node_skip + or hooks.unwrap_result + ) ) ) ) @@ -2484,7 +2604,10 @@ def submit_node(node_id: str) -> tuple[NodeFuture, Any]: # This enables: `a() >> b() >> c()` without manual wiring. # Injectable params (Logger, Stream, etc.) are excluded — they're # handled separately by the provider system. - if not node_ref.args and not binding_plan.has_explicit_provider_selectors: + deferred_implicit_binding = bool( + not node_ref.args and not binding_plan.has_explicit_provider_selectors + ) + if deferred_implicit_binding: upstream_values = _collect_implicit_parent_results( node_ref, result_refs, @@ -2492,16 +2615,8 @@ def submit_node(node_id: str) -> tuple[NodeFuture, Any]: node_id, executor=executor, ) - implicit_args, implicit_kwargs = _bind_implicit_parent_results( - node_ref.node.fn, - upstream_values, - skip_param_names, - position_consuming_params, - resolved_kwargs, - adapt_positionals=binding_plan.positional_call_adapter is not None, - ) - resolved_args = implicit_args - resolved_kwargs.update(implicit_kwargs) + for index, value in enumerate(upstream_values): + resolved_kwargs[f"{_IMPLICIT_VALUE_KWARG_PREFIX}{index}"] = value # Handle injectable parameters that may need wrappers (using provider pattern) # @@ -2569,6 +2684,10 @@ def submit_node(node_id: str) -> tuple[NodeFuture, Any]: actual_fn, node_run_context, num_returns=node_ref.node.num_returns, + implicit_binding_fn=(node_ref.node.fn if deferred_implicit_binding else None), + implicit_skip_param_names=skip_param_names, + implicit_position_consuming_param_names=position_consuming_params, + adapt_implicit_positionals=(binding_plan.positional_call_adapter is not None), ) try: @@ -2675,7 +2794,12 @@ def resolve_submitted_result(node_ref: NodeFuture, result: Any) -> Any: cancel_requested is not None or ( hooks - and (hooks.on_node_success or hooks.on_node_failure or hooks.unwrap_result) + and ( + hooks.on_node_success + or hooks.on_node_failure + or hooks.on_node_skip + or hooks.unwrap_result + ) ) ) ) @@ -2694,24 +2818,30 @@ def ray_refs_for_result(result: Any) -> list[Any]: return [] def complete_ray_node(node_id: str) -> None: + from .types import skip_outcome_from_result + node_ref = self.nodes[node_id] result = result_refs[node_id] try: - if hooks and hooks.unwrap_result: + outcome = None + if node_id in status_refs: + status_value = executor.get([status_refs[node_id]])[0] + outcome = skip_outcome_from_result(status_value) + if hooks and hooks.unwrap_result and outcome is None: # unwrap_result needs the user-facing value, so a payload # fetch here is intentional. Keep the lineage-preserving # envelope in result_refs for downstream dataflow, # reattaching any hook replacement. resolved_val = resolve_submitted_result(node_ref, result) - user_val = _unwrap_lineaged_tree(resolved_val) - replacement = hooks.unwrap_result(node_id, user_val) - result_refs[node_id] = _reattach_lineage(replacement, resolved_val) - elif node_id in status_refs: - # Progress-only: fetch just the tiny status ref to - # surface a task failure. Never materialize the payload; - # result_refs keeps the payload ref for downstream tasks. - executor.get([status_refs[node_id]]) - if hooks and hooks.on_node_success: + outcome = skip_outcome_from_result(resolved_val) + if outcome is None: + user_val = _unwrap_lineaged_tree(resolved_val) + replacement = hooks.unwrap_result(node_id, user_val) + result_refs[node_id] = _reattach_lineage(replacement, resolved_val) + if outcome is not None: + if hooks and hooks.on_node_skip: + hooks.on_node_skip(node_id, outcome) + elif hooks and hooks.on_node_success: hooks.on_node_success(node_id) completed_nodes.add(node_id) except Exception as exc: @@ -2824,21 +2954,29 @@ def dependencies_ready(node_id: str) -> bool: # on_node_success fires after actual completion (not just # submission). This serializes execution but gives accurate # per-node progress — the right tradeoff for the operator. - if hooks and (hooks.on_node_success or hooks.unwrap_result): + if hooks and ( + hooks.on_node_success or hooks.on_node_skip or hooks.unwrap_result + ): + from .types import skip_outcome_from_result + # Wait for completion and resolve to actual value. # For multi-return nodes (num_returns > 1), result is # a tuple of refs; for single-return it's one ref/value. resolved_val = resolve_submitted_result(node_ref, result) + outcome = skip_outcome_from_result(resolved_val) # Unwrap side-channel data (e.g. logs from Ray workers). # Hooks see user-facing values; result_refs keeps the # lineage-preserving envelope for downstream dataflow. - if hooks.unwrap_result: + if hooks.unwrap_result and outcome is None: user_val = _unwrap_lineaged_tree(resolved_val) replacement = hooks.unwrap_result(node_id, user_val) result = _reattach_lineage(replacement, resolved_val) else: result = resolved_val - if hooks.on_node_success: + if outcome is not None: + if hooks.on_node_skip: + hooks.on_node_skip(node_id, outcome) + elif hooks.on_node_success: hooks.on_node_success(node_id) result_refs[node_id] = result @@ -2847,7 +2985,12 @@ def dependencies_ready(node_id: str) -> bool: if self.returns is None: already_observed = bool( hooks - and (hooks.on_node_success or hooks.on_node_failure or hooks.unwrap_result) + and ( + hooks.on_node_success + or hooks.on_node_failure + or hooks.on_node_skip + or hooks.unwrap_result + ) ) if already_observed: # Per-node completion/failure was already observed above (hooks). diff --git a/src/avalanche/types.py b/src/avalanche/types.py index 136e9e5..5e5ce8d 100644 --- a/src/avalanche/types.py +++ b/src/avalanche/types.py @@ -10,7 +10,7 @@ from abc import ABC, abstractmethod from dataclasses import dataclass from enum import Enum -from typing import Any, Callable, ClassVar, Generic, TypeVar, Union, cast +from typing import Any, Callable, ClassVar, Generic, Mapping, TypeVar, Union, cast import polars as pl import pyarrow as pa @@ -19,6 +19,37 @@ ModelT = TypeVar("ModelT", bound=BaseModel) +@dataclass(frozen=True) +class SkipOutcome: + """Successful node outcome that intentionally carries no data value.""" + + reason: str + metadata: dict[str, Any] | None = None + + +def skip(reason: str, metadata: Mapping[str, Any] | None = None) -> SkipOutcome: + """Mark the current node as successfully skipped.""" + if not isinstance(reason, str): + raise TypeError("skip reason must be a string") + if metadata is not None and not isinstance(metadata, Mapping): + raise TypeError("skip metadata must be a mapping or None") + return SkipOutcome(reason=reason, metadata=dict(metadata) if metadata is not None else None) + + +def skip_outcome_from_result(value: Any) -> SkipOutcome | None: + """Extract a whole-node skip outcome from an internal result envelope.""" + if isinstance(value, SkipOutcome): + return value + if isinstance(value, LineagedResult): + return skip_outcome_from_result(value.value) + if isinstance(value, (tuple, list)) and value: + outcomes = [skip_outcome_from_result(item) for item in value] + first = outcomes[0] + if first is not None and all(outcome == first for outcome in outcomes): + return first + return None + + @dataclass class AppendResult(Generic[ModelT]): """ diff --git a/src/runtime/executor.py b/src/runtime/executor.py index 6be2811..bd9b823 100644 --- a/src/runtime/executor.py +++ b/src/runtime/executor.py @@ -62,19 +62,22 @@ def wrapper(*args: Any, **kwargs: Any) -> Any: def _wrap_with_status(fn: Callable, user_num_returns: int) -> Callable: - """Wrap a task so it also emits a small status marker as its last return. + """Wrap a task so it also emits its small control outcome as the last return. The status value is produced by the *same* task as the payload, so fetching - only the status ref surfaces a task exception without materializing the - payload on the driver (or in a separate worker). ``None`` on success. + only the status ref surfaces both task exceptions and intentional skips + without materializing the payload on the driver. """ @wraps(fn) def wrapper(*args: Any, **kwargs: Any) -> Any: + from avalanche.types import skip_outcome_from_result + result = _normalize_distributed_result(call_sync_or_async(fn, *args, **kwargs)) + status = skip_outcome_from_result(result) if user_num_returns > 1: - return (*result, None) - return result, None + return (*result, status) + return result, status return wrapper @@ -123,9 +126,12 @@ def _distributed_execution_services_task( num_returns=user_num_returns, normalize_result=_normalize_distributed_result, ) + from avalanche.types import skip_outcome_from_result + + status = skip_outcome_from_result(result) if user_num_returns > 1: - return (*result, receipt, None) - return result, receipt, None + return (*result, receipt, status) + return result, receipt, status class Executor(Protocol): @@ -422,10 +428,10 @@ def submit_with_status( """Submit a Ray task that also emits a tiny status marker. The task is created with ``num_returns + 1`` return values: the user - payload(s) followed by a small status value (``None`` on success). The - status ref lets the driver observe completion/failure without fetching - the payload, and it is produced by the *same* task so no payload is - deserialized in a separate worker. + payload(s) followed by a small control status. The status ref carries + a ``SkipOutcome`` for an intentional skip and ``None`` for an ordinary + success, letting the driver observe completion without fetching the + payload. """ if hasattr(fn, "remote"): raise TypeError( diff --git a/src/runtime/operator/convert.py b/src/runtime/operator/convert.py index e436c11..c4a10ea 100644 --- a/src/runtime/operator/convert.py +++ b/src/runtime/operator/convert.py @@ -2,6 +2,7 @@ from __future__ import annotations +import json from pathlib import Path, PureWindowsPath from .models import ( @@ -91,6 +92,8 @@ def node_state_to_proto(ns: NodeState) -> pb.NodeStateMsg: status=ns.status.value, started_at=ns.started_at or 0.0, ended_at=ns.ended_at or 0.0, + reason=ns.reason or "", + metadata_json=json.dumps(ns.metadata) if ns.metadata is not None else "", ) @@ -102,6 +105,8 @@ def node_state_from_proto(msg: pb.NodeStateMsg) -> NodeState: status=NodeStatus(msg.status), started_at=msg.started_at if msg.started_at else None, ended_at=msg.ended_at if msg.ended_at else None, + reason=msg.reason or None, + metadata=json.loads(msg.metadata_json) if msg.metadata_json else None, ) diff --git a/src/runtime/operator/hooks.py b/src/runtime/operator/hooks.py index 1cdf9ee..7aae806 100644 --- a/src/runtime/operator/hooks.py +++ b/src/runtime/operator/hooks.py @@ -18,6 +18,7 @@ class RunHooks: on_node_start: Callable[[str], None] | None = None on_node_success: Callable[[str], None] | None = None on_node_failure: Callable[[str, Exception], None] | None = None + on_node_skip: Callable[[str, Any], None] | None = None cancel_requested: Callable[[], bool] | None = None wrap_fn: Callable[[str, Callable], Callable] | None = None """Optional function wrapper applied before executor.submit(). diff --git a/src/runtime/operator/models.py b/src/runtime/operator/models.py index f3ebaa8..e5779f8 100644 --- a/src/runtime/operator/models.py +++ b/src/runtime/operator/models.py @@ -7,7 +7,7 @@ from datetime import datetime from enum import Enum from types import MappingProxyType -from typing import Literal, Mapping +from typing import Any, Literal, Mapping class NodeStatus(Enum): @@ -49,6 +49,8 @@ class NodeState: status: NodeStatus = NodeStatus.PENDING started_at: float | None = None ended_at: float | None = None + reason: str | None = None + metadata: dict[str, Any] | None = None @property def elapsed(self) -> float | None: diff --git a/src/runtime/operator/operator.py b/src/runtime/operator/operator.py index de39b9b..23642dc 100644 --- a/src/runtime/operator/operator.py +++ b/src/runtime/operator/operator.py @@ -521,12 +521,17 @@ def _apply_event(self, run_id: str, event: dict[str, Any]) -> bool: "node_started": NodeStatus.RUNNING, "node_succeeded": NodeStatus.SUCCESS, "node_failed": NodeStatus.FAILED, + "node_skipped": NodeStatus.SKIPPED, }[event_type] node.status = status if status == NodeStatus.RUNNING: node.started_at = event["timestamp"] else: node.ended_at = event["timestamp"] + if status == NodeStatus.SKIPPED: + node.reason = event["reason"] + metadata = event["metadata"] + node.metadata = dict(metadata) if metadata is not None else None elif event_type == "log": if run.status in { RunStatus.SUCCESS, @@ -661,10 +666,11 @@ def _notify_log(self, entry: LogEntry) -> None: "node_started", "node_succeeded", "node_failed", + "node_skipped", "log", "terminal", } -_NODE_EVENT_TYPES = {"node_started", "node_succeeded", "node_failed"} +_NODE_EVENT_TYPES = {"node_started", "node_succeeded", "node_failed", "node_skipped"} _TERMINAL_STATUSES = {"success", "failed", "cancelled"} @@ -720,6 +726,11 @@ def _validate_run_event(event: object) -> str: _timestamp_field(event, "timestamp") if event_type == "node_failed": _string_field(event, "error") + elif event_type == "node_skipped": + _string_field(event, "reason") + metadata = _required_field(event, "metadata") + if metadata is not None and type(metadata) is not dict: + raise _CoordinatorProtocolError("field 'metadata' must be a dict or None") elif event_type == "log": timestamp = _timestamp_field(event, "timestamp") try: diff --git a/src/runtime/operator/proto/operator.proto b/src/runtime/operator/proto/operator.proto index d1eae10..6ced8af 100644 --- a/src/runtime/operator/proto/operator.proto +++ b/src/runtime/operator/proto/operator.proto @@ -107,6 +107,8 @@ message NodeStateMsg { string status = 4; double started_at = 5; double ended_at = 6; + string reason = 7; + string metadata_json = 8; } message LogEntryMsg { diff --git a/src/runtime/operator/proto/operator_pb2.py b/src/runtime/operator/proto/operator_pb2.py index 66886d8..629acaf 100644 --- a/src/runtime/operator/proto/operator_pb2.py +++ b/src/runtime/operator/proto/operator_pb2.py @@ -24,7 +24,7 @@ -DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(b'\n\x0eoperator.proto\x12\x12\x61valanche.operator\"\x07\n\x05\x45mpty\"\xef\x01\n\x0fStartRunRequest\x12\x11\n\tflow_name\x18\x01 \x01(\t\x12\x12\n\ninput_json\x18\x02 \x01(\t\x12\x14\n\x0c\x63ontext_json\x18\x03 \x01(\t\x12\x37\n\x0binput_files\x18\x04 \x03(\x0b\x32\".avalanche.operator.FileAttachment\x12;\n\x0einput_s3_files\x18\x05 \x03(\x0b\x32#.avalanche.operator.S3FileReference\x12\x0e\n\x06run_id\x18\x06 \x01(\t\x12\x19\n\x11workflow_selector\x18\x07 \x01(\t\"i\n\x0e\x46ileAttachment\x12\x12\n\nfield_name\x18\x01 \x01(\t\x12\x0c\n\x04name\x18\x02 \x01(\t\x12\x0f\n\x07\x63ontent\x18\x03 \x01(\x0c\x12\x14\n\x0c\x63ontent_type\x18\x04 \x01(\t\x12\x0e\n\x06sha256\x18\x05 \x01(\t\"\x8e\x01\n\x0fS3FileReference\x12\x12\n\nfield_name\x18\x01 \x01(\t\x12\x0b\n\x03uri\x18\x02 \x01(\t\x12\x12\n\nversion_id\x18\x03 \x01(\t\x12\x0c\n\x04\x65tag\x18\x04 \x01(\t\x12\x12\n\nsize_bytes\x18\x05 \x01(\x04\x12\x14\n\x0c\x63ontent_type\x18\x06 \x01(\t\x12\x0e\n\x06sha256\x18\x07 \x01(\t\"\"\n\x10StartRunResponse\x12\x0e\n\x06run_id\x18\x01 \x01(\t\"\"\n\x10\x43\x61ncelRunRequest\x12\x0e\n\x06run_id\x18\x01 \x01(\t\"?\n\x0fListRunsRequest\x12\x11\n\tflow_name\x18\x01 \x01(\t\x12\x19\n\x11workflow_selector\x18\x02 \x01(\t\"\x1f\n\rGetRunRequest\x12\x0e\n\x06run_id\x18\x01 \x01(\t\"\'\n\rStreamRequest\x12\x16\n\x0esince_sequence\x18\x01 \x01(\x04\"\x1d\n\tNodeEdges\x12\x10\n\x08\x63hildren\x18\x01 \x03(\t\"\xe3\x04\n\x0b\x46lowInfoMsg\x12\x0c\n\x04name\x18\x01 \x01(\t\x12\x11\n\tfile_path\x18\x02 \x01(\t\x12\x10\n\x08node_ids\x18\x03 \x03(\t\x12\x39\n\x05graph\x18\x04 \x03(\x0b\x32*.avalanche.operator.FlowInfoMsg.GraphEntry\x12\x42\n\nnode_types\x18\x05 \x03(\x0b\x32..avalanche.operator.FlowInfoMsg.NodeTypesEntry\x12H\n\rdisplay_names\x18\x06 \x03(\x0b\x32\x31.avalanche.operator.FlowInfoMsg.DisplayNamesEntry\x12\x0c\n\x04\x63ron\x18\x07 \x01(\t\x12\x13\n\x0bnext_run_at\x18\x08 \x01(\x01\x12\x13\n\x0blast_run_at\x18\t \x01(\x01\x12\x13\n\x0bworkflow_id\x18\n \x01(\t\x12\x14\n\x0c\x64isplay_name\x18\x0b \x01(\t\x12\x12\n\nroot_alias\x18\x0c \x01(\t\x12\x15\n\rrelative_file\x18\r \x01(\t\x12\x16\n\x0e\x62uilder_symbol\x18\x0e \x01(\t\x1aK\n\nGraphEntry\x12\x0b\n\x03key\x18\x01 \x01(\t\x12,\n\x05value\x18\x02 \x01(\x0b\x32\x1d.avalanche.operator.NodeEdges:\x02\x38\x01\x1a\x30\n\x0eNodeTypesEntry\x12\x0b\n\x03key\x18\x01 \x01(\t\x12\r\n\x05value\x18\x02 \x01(\t:\x02\x38\x01\x1a\x33\n\x11\x44isplayNamesEntry\x12\x0b\n\x03key\x18\x01 \x01(\t\x12\r\n\x05value\x18\x02 \x01(\t:\x02\x38\x01\"{\n\x08\x46lowList\x12.\n\x05\x66lows\x18\x01 \x03(\x0b\x32\x1f.avalanche.operator.FlowInfoMsg\x12?\n\x0b\x64iagnostics\x18\x02 \x03(\x0b\x32*.avalanche.operator.DiscoveryDiagnosticMsg\"E\n\x16\x44iscoveryDiagnosticMsg\x12\x0c\n\x04path\x18\x01 \x01(\t\x12\x0c\n\x04kind\x18\x02 \x01(\t\x12\x0f\n\x07message\x18\x03 \x01(\t\"v\n\x0cNodeStateMsg\x12\x0f\n\x07node_id\x18\x01 \x01(\t\x12\x0c\n\x04name\x18\x02 \x01(\t\x12\x11\n\tnode_type\x18\x03 \x01(\t\x12\x0e\n\x06status\x18\x04 \x01(\t\x12\x12\n\nstarted_at\x18\x05 \x01(\x01\x12\x10\n\x08\x65nded_at\x18\x06 \x01(\x01\"Q\n\x0bLogEntryMsg\x12\x11\n\ttimestamp\x18\x01 \x01(\x01\x12\r\n\x05level\x18\x02 \x01(\t\x12\x0f\n\x07node_id\x18\x03 \x01(\t\x12\x0f\n\x07message\x18\x04 \x01(\t\"\x90\x02\n\x0bRunStateMsg\x12\x0e\n\x06run_id\x18\x01 \x01(\t\x12\x11\n\tflow_name\x18\x02 \x01(\t\x12\x0e\n\x06status\x18\x03 \x01(\t\x12\x12\n\nstarted_at\x18\x04 \x01(\x01\x12\x10\n\x08\x65nded_at\x18\x05 \x01(\x01\x12/\n\x05nodes\x18\x06 \x03(\x0b\x32 .avalanche.operator.NodeStateMsg\x12-\n\x04logs\x18\x07 \x03(\x0b\x32\x1f.avalanche.operator.LogEntryMsg\x12\x14\n\x0ctriggered_by\x18\x08 \x01(\t\x12\x13\n\x0bworkflow_id\x18\t \x01(\t\x12\x1d\n\x15workflow_display_name\x18\n \x01(\t\"8\n\x07RunList\x12-\n\x04runs\x18\x01 \x03(\x0b\x32\x1f.avalanche.operator.RunStateMsg\"K\n\tRunUpdate\x12\x10\n\x08sequence\x18\x01 \x01(\x04\x12,\n\x03run\x18\x02 \x01(\x0b\x32\x1f.avalanche.operator.RunStateMsg2\xed\x03\n\x0fOperatorService\x12\x44\n\tListFlows\x12\x19.avalanche.operator.Empty\x1a\x1c.avalanche.operator.FlowList\x12U\n\x08StartRun\x12#.avalanche.operator.StartRunRequest\x1a$.avalanche.operator.StartRunResponse\x12L\n\tCancelRun\x12$.avalanche.operator.CancelRunRequest\x1a\x19.avalanche.operator.Empty\x12L\n\x08ListRuns\x12#.avalanche.operator.ListRunsRequest\x1a\x1b.avalanche.operator.RunList\x12L\n\x06GetRun\x12!.avalanche.operator.GetRunRequest\x1a\x1f.avalanche.operator.RunStateMsg\x12S\n\rStreamUpdates\x12!.avalanche.operator.StreamRequest\x1a\x1d.avalanche.operator.RunUpdate0\x01\x62\x06proto3') +DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(b'\n\x0eoperator.proto\x12\x12\x61valanche.operator\"\x07\n\x05\x45mpty\"\xef\x01\n\x0fStartRunRequest\x12\x11\n\tflow_name\x18\x01 \x01(\t\x12\x12\n\ninput_json\x18\x02 \x01(\t\x12\x14\n\x0c\x63ontext_json\x18\x03 \x01(\t\x12\x37\n\x0binput_files\x18\x04 \x03(\x0b\x32\".avalanche.operator.FileAttachment\x12;\n\x0einput_s3_files\x18\x05 \x03(\x0b\x32#.avalanche.operator.S3FileReference\x12\x0e\n\x06run_id\x18\x06 \x01(\t\x12\x19\n\x11workflow_selector\x18\x07 \x01(\t\"i\n\x0e\x46ileAttachment\x12\x12\n\nfield_name\x18\x01 \x01(\t\x12\x0c\n\x04name\x18\x02 \x01(\t\x12\x0f\n\x07\x63ontent\x18\x03 \x01(\x0c\x12\x14\n\x0c\x63ontent_type\x18\x04 \x01(\t\x12\x0e\n\x06sha256\x18\x05 \x01(\t\"\x8e\x01\n\x0fS3FileReference\x12\x12\n\nfield_name\x18\x01 \x01(\t\x12\x0b\n\x03uri\x18\x02 \x01(\t\x12\x12\n\nversion_id\x18\x03 \x01(\t\x12\x0c\n\x04\x65tag\x18\x04 \x01(\t\x12\x12\n\nsize_bytes\x18\x05 \x01(\x04\x12\x14\n\x0c\x63ontent_type\x18\x06 \x01(\t\x12\x0e\n\x06sha256\x18\x07 \x01(\t\"\"\n\x10StartRunResponse\x12\x0e\n\x06run_id\x18\x01 \x01(\t\"\"\n\x10\x43\x61ncelRunRequest\x12\x0e\n\x06run_id\x18\x01 \x01(\t\"?\n\x0fListRunsRequest\x12\x11\n\tflow_name\x18\x01 \x01(\t\x12\x19\n\x11workflow_selector\x18\x02 \x01(\t\"\x1f\n\rGetRunRequest\x12\x0e\n\x06run_id\x18\x01 \x01(\t\"\'\n\rStreamRequest\x12\x16\n\x0esince_sequence\x18\x01 \x01(\x04\"\x1d\n\tNodeEdges\x12\x10\n\x08\x63hildren\x18\x01 \x03(\t\"\xe3\x04\n\x0b\x46lowInfoMsg\x12\x0c\n\x04name\x18\x01 \x01(\t\x12\x11\n\tfile_path\x18\x02 \x01(\t\x12\x10\n\x08node_ids\x18\x03 \x03(\t\x12\x39\n\x05graph\x18\x04 \x03(\x0b\x32*.avalanche.operator.FlowInfoMsg.GraphEntry\x12\x42\n\nnode_types\x18\x05 \x03(\x0b\x32..avalanche.operator.FlowInfoMsg.NodeTypesEntry\x12H\n\rdisplay_names\x18\x06 \x03(\x0b\x32\x31.avalanche.operator.FlowInfoMsg.DisplayNamesEntry\x12\x0c\n\x04\x63ron\x18\x07 \x01(\t\x12\x13\n\x0bnext_run_at\x18\x08 \x01(\x01\x12\x13\n\x0blast_run_at\x18\t \x01(\x01\x12\x13\n\x0bworkflow_id\x18\n \x01(\t\x12\x14\n\x0c\x64isplay_name\x18\x0b \x01(\t\x12\x12\n\nroot_alias\x18\x0c \x01(\t\x12\x15\n\rrelative_file\x18\r \x01(\t\x12\x16\n\x0e\x62uilder_symbol\x18\x0e \x01(\t\x1aK\n\nGraphEntry\x12\x0b\n\x03key\x18\x01 \x01(\t\x12,\n\x05value\x18\x02 \x01(\x0b\x32\x1d.avalanche.operator.NodeEdges:\x02\x38\x01\x1a\x30\n\x0eNodeTypesEntry\x12\x0b\n\x03key\x18\x01 \x01(\t\x12\r\n\x05value\x18\x02 \x01(\t:\x02\x38\x01\x1a\x33\n\x11\x44isplayNamesEntry\x12\x0b\n\x03key\x18\x01 \x01(\t\x12\r\n\x05value\x18\x02 \x01(\t:\x02\x38\x01\"{\n\x08\x46lowList\x12.\n\x05\x66lows\x18\x01 \x03(\x0b\x32\x1f.avalanche.operator.FlowInfoMsg\x12?\n\x0b\x64iagnostics\x18\x02 \x03(\x0b\x32*.avalanche.operator.DiscoveryDiagnosticMsg\"E\n\x16\x44iscoveryDiagnosticMsg\x12\x0c\n\x04path\x18\x01 \x01(\t\x12\x0c\n\x04kind\x18\x02 \x01(\t\x12\x0f\n\x07message\x18\x03 \x01(\t\"\x9d\x01\n\x0cNodeStateMsg\x12\x0f\n\x07node_id\x18\x01 \x01(\t\x12\x0c\n\x04name\x18\x02 \x01(\t\x12\x11\n\tnode_type\x18\x03 \x01(\t\x12\x0e\n\x06status\x18\x04 \x01(\t\x12\x12\n\nstarted_at\x18\x05 \x01(\x01\x12\x10\n\x08\x65nded_at\x18\x06 \x01(\x01\x12\x0e\n\x06reason\x18\x07 \x01(\t\x12\x15\n\rmetadata_json\x18\x08 \x01(\t\"Q\n\x0bLogEntryMsg\x12\x11\n\ttimestamp\x18\x01 \x01(\x01\x12\r\n\x05level\x18\x02 \x01(\t\x12\x0f\n\x07node_id\x18\x03 \x01(\t\x12\x0f\n\x07message\x18\x04 \x01(\t\"\x90\x02\n\x0bRunStateMsg\x12\x0e\n\x06run_id\x18\x01 \x01(\t\x12\x11\n\tflow_name\x18\x02 \x01(\t\x12\x0e\n\x06status\x18\x03 \x01(\t\x12\x12\n\nstarted_at\x18\x04 \x01(\x01\x12\x10\n\x08\x65nded_at\x18\x05 \x01(\x01\x12/\n\x05nodes\x18\x06 \x03(\x0b\x32 .avalanche.operator.NodeStateMsg\x12-\n\x04logs\x18\x07 \x03(\x0b\x32\x1f.avalanche.operator.LogEntryMsg\x12\x14\n\x0ctriggered_by\x18\x08 \x01(\t\x12\x13\n\x0bworkflow_id\x18\t \x01(\t\x12\x1d\n\x15workflow_display_name\x18\n \x01(\t\"8\n\x07RunList\x12-\n\x04runs\x18\x01 \x03(\x0b\x32\x1f.avalanche.operator.RunStateMsg\"K\n\tRunUpdate\x12\x10\n\x08sequence\x18\x01 \x01(\x04\x12,\n\x03run\x18\x02 \x01(\x0b\x32\x1f.avalanche.operator.RunStateMsg2\xed\x03\n\x0fOperatorService\x12\x44\n\tListFlows\x12\x19.avalanche.operator.Empty\x1a\x1c.avalanche.operator.FlowList\x12U\n\x08StartRun\x12#.avalanche.operator.StartRunRequest\x1a$.avalanche.operator.StartRunResponse\x12L\n\tCancelRun\x12$.avalanche.operator.CancelRunRequest\x1a\x19.avalanche.operator.Empty\x12L\n\x08ListRuns\x12#.avalanche.operator.ListRunsRequest\x1a\x1b.avalanche.operator.RunList\x12L\n\x06GetRun\x12!.avalanche.operator.GetRunRequest\x1a\x1f.avalanche.operator.RunStateMsg\x12S\n\rStreamUpdates\x12!.avalanche.operator.StreamRequest\x1a\x1d.avalanche.operator.RunUpdate0\x01\x62\x06proto3') _globals = globals() _builder.BuildMessageAndEnumDescriptors(DESCRIPTOR, _globals) @@ -69,16 +69,16 @@ _globals['_FLOWLIST']._serialized_end=1520 _globals['_DISCOVERYDIAGNOSTICMSG']._serialized_start=1522 _globals['_DISCOVERYDIAGNOSTICMSG']._serialized_end=1591 - _globals['_NODESTATEMSG']._serialized_start=1593 - _globals['_NODESTATEMSG']._serialized_end=1711 - _globals['_LOGENTRYMSG']._serialized_start=1713 - _globals['_LOGENTRYMSG']._serialized_end=1794 - _globals['_RUNSTATEMSG']._serialized_start=1797 - _globals['_RUNSTATEMSG']._serialized_end=2069 - _globals['_RUNLIST']._serialized_start=2071 - _globals['_RUNLIST']._serialized_end=2127 - _globals['_RUNUPDATE']._serialized_start=2129 - _globals['_RUNUPDATE']._serialized_end=2204 - _globals['_OPERATORSERVICE']._serialized_start=2207 - _globals['_OPERATORSERVICE']._serialized_end=2700 + _globals['_NODESTATEMSG']._serialized_start=1594 + _globals['_NODESTATEMSG']._serialized_end=1751 + _globals['_LOGENTRYMSG']._serialized_start=1753 + _globals['_LOGENTRYMSG']._serialized_end=1834 + _globals['_RUNSTATEMSG']._serialized_start=1837 + _globals['_RUNSTATEMSG']._serialized_end=2109 + _globals['_RUNLIST']._serialized_start=2111 + _globals['_RUNLIST']._serialized_end=2167 + _globals['_RUNUPDATE']._serialized_start=2169 + _globals['_RUNUPDATE']._serialized_end=2244 + _globals['_OPERATORSERVICE']._serialized_start=2247 + _globals['_OPERATORSERVICE']._serialized_end=2740 # @@protoc_insertion_point(module_scope) diff --git a/src/runtime/operator/proto/operator_pb2.pyi b/src/runtime/operator/proto/operator_pb2.pyi index 90d9c8a..5496f1c 100644 --- a/src/runtime/operator/proto/operator_pb2.pyi +++ b/src/runtime/operator/proto/operator_pb2.pyi @@ -170,20 +170,24 @@ class DiscoveryDiagnosticMsg(_message.Message): def __init__(self, path: _Optional[str] = ..., kind: _Optional[str] = ..., message: _Optional[str] = ...) -> None: ... class NodeStateMsg(_message.Message): - __slots__ = ("node_id", "name", "node_type", "status", "started_at", "ended_at") + __slots__ = ("node_id", "name", "node_type", "status", "started_at", "ended_at", "reason", "metadata_json") NODE_ID_FIELD_NUMBER: _ClassVar[int] NAME_FIELD_NUMBER: _ClassVar[int] NODE_TYPE_FIELD_NUMBER: _ClassVar[int] STATUS_FIELD_NUMBER: _ClassVar[int] STARTED_AT_FIELD_NUMBER: _ClassVar[int] ENDED_AT_FIELD_NUMBER: _ClassVar[int] + REASON_FIELD_NUMBER: _ClassVar[int] + METADATA_JSON_FIELD_NUMBER: _ClassVar[int] node_id: str name: str node_type: str status: str started_at: float ended_at: float - def __init__(self, node_id: _Optional[str] = ..., name: _Optional[str] = ..., node_type: _Optional[str] = ..., status: _Optional[str] = ..., started_at: _Optional[float] = ..., ended_at: _Optional[float] = ...) -> None: ... + reason: str + metadata_json: str + def __init__(self, node_id: _Optional[str] = ..., name: _Optional[str] = ..., node_type: _Optional[str] = ..., status: _Optional[str] = ..., started_at: _Optional[float] = ..., ended_at: _Optional[float] = ..., reason: _Optional[str] = ..., metadata_json: _Optional[str] = ...) -> None: ... class LogEntryMsg(_message.Message): __slots__ = ("timestamp", "level", "node_id", "message") diff --git a/src/runtime/operator/run_worker.py b/src/runtime/operator/run_worker.py index 2e19339..9cbf80c 100644 --- a/src/runtime/operator/run_worker.py +++ b/src/runtime/operator/run_worker.py @@ -131,6 +131,15 @@ def wrap_fn(node_id: str, fn: Callable[..., Any]) -> Callable[..., Any]: on_node_success=lambda node_id: event_queue.put( {"type": "node_succeeded", "node_id": node_id, "timestamp": time.monotonic()} ), + on_node_skip=lambda node_id, outcome: event_queue.put( + { + "type": "node_skipped", + "node_id": node_id, + "timestamp": time.monotonic(), + "reason": outcome.reason, + "metadata": outcome.metadata, + } + ), on_node_failure=lambda node_id, exc: event_queue.put( { "type": "node_failed", diff --git a/src/tui/widgets/log_panel.py b/src/tui/widgets/log_panel.py index f8b34c5..100012b 100644 --- a/src/tui/widgets/log_panel.py +++ b/src/tui/widgets/log_panel.py @@ -2,6 +2,8 @@ from __future__ import annotations +import json + from rich.style import Style from rich.text import Text from textual.widgets import Static @@ -76,15 +78,28 @@ def render(self) -> Text: text.append(" Waiting for dependencies…\n", DIM_STYLE) return text if status == NodeStatus.SKIPPED: - text.append(" Skipped — upstream dependency failed.\n", Style(color=ICE_WARN)) + node_state = ( + store.current_run.nodes.get(selected_node.name) + if store.current_run is not None + else None + ) + if node_state is not None and node_state.reason is not None: + text.append(f" Skipped — {node_state.reason}\n", Style(color=ICE_WARN)) + if node_state.metadata is not None: + metadata = json.dumps(node_state.metadata, sort_keys=True) + text.append(f" Metadata: {metadata}\n", DIM_STYLE) + else: + text.append( + " Skipped — upstream dependency failed.\n", + Style(color=ICE_WARN), + ) return text # Filter entries for selected node visible = logs if selected_node: visible = [ - e for e in logs - if e.node_id in (selected_node.name, selected_node.display_name) + e for e in logs if e.node_id in (selected_node.name, selected_node.display_name) ] if not visible and selected_node: @@ -192,5 +207,5 @@ def _append_highlighted( break if idx > pos: text.append(msg[pos:idx], base_style) - text.append(msg[idx:idx + len(query)], hl) + text.append(msg[idx : idx + len(query)], hl) pos = idx + len(query) diff --git a/test/fixtures/sample_workflows.py b/test/fixtures/sample_workflows.py index 5ea9515..0aca018 100644 --- a/test/fixtures/sample_workflows.py +++ b/test/fixtures/sample_workflows.py @@ -306,3 +306,18 @@ def order_workflow(): enriched = enrich(validated) agg = aggregate(enriched) (save_warehouse(agg) & notify(agg)) + + +@source +def optional_source(): + return ava.skip("No matching files", {"partition": "2026-07-22"}) + + +@step +def after_optional(): + return "completed" + + +@workflow +def skipped_workflow(): + optional_source() >> after_optional() diff --git a/test/operator_tests/test_operator.py b/test/operator_tests/test_operator.py index 4e612c3..a12236f 100644 --- a/test/operator_tests/test_operator.py +++ b/test/operator_tests/test_operator.py @@ -129,6 +129,29 @@ def test_all_nodes_succeed(self): assert ns.started_at is not None assert ns.ended_at is not None + def test_authored_skip_is_recorded_and_run_succeeds(self): + op = self._make_operator() + run_id = op.start_run("skipped_workflow") + + deadline = time.monotonic() + 5 + while time.monotonic() < deadline: + run = op.get_run(run_id) + if run and run.status in (RunStatus.SUCCESS, RunStatus.FAILED): + break + time.sleep(0.05) + + run = op.get_run(run_id) + assert run is not None + assert run.status == RunStatus.SUCCESS + skipped = run.nodes["optional_source_1"] + assert skipped.status == NodeStatus.SKIPPED + assert skipped.reason == "No matching files" + assert skipped.metadata == {"partition": "2026-07-22"} + assert skipped.started_at is not None + assert skipped.ended_at is not None + assert skipped.ended_at >= skipped.started_at + assert run.nodes["after_optional_1"].status == NodeStatus.SUCCESS + def test_node_transitions_observed(self): """Verify nodes go through PENDING -> RUNNING -> SUCCESS in order.""" op = self._make_operator() diff --git a/test/skipped_outcome_test.py b/test/skipped_outcome_test.py new file mode 100644 index 0000000..0210e56 --- /dev/null +++ b/test/skipped_outcome_test.py @@ -0,0 +1,170 @@ +from __future__ import annotations + +from typing import Any + +import pytest + +import avalanche as ava +from avalanche.operator.hooks import RunHooks + +EXECUTOR_FACTORIES = [ + ava.LocalExecutor, + pytest.param(ava.RayExecutor, marks=pytest.mark.ray), +] + + +@pytest.mark.parametrize("executor_factory", EXECUTOR_FACTORIES) +def test_skipped_branch_is_non_value_and_fan_in_completes(executor_factory): + if executor_factory is ava.RayExecutor: + pytest.importorskip("ray") + + events: list[tuple[str, str, Any]] = [] + + @ava.source(slug="optional") + def optional(): + return ava.skip("No optional rows", {"partition": "2026-07-22"}) + + @ava.source(slug="required") + def required(): + return "required-value" + + @ava.dest(slug="persist") + def persist(value: str): + return f"persisted:{value}" + + @ava.workflow + def optional_workflow(): + return (optional() & required()) >> persist() + + hooks = RunHooks( + on_node_success=lambda node_id: events.append(("success", node_id, None)), + on_node_skip=lambda node_id, outcome: events.append(("skipped", node_id, outcome)), + ) + executor = executor_factory() + try: + result = optional_workflow().run(executor=executor, hooks=hooks).result() + finally: + ray = getattr(executor, "ray", None) + if ray is not None and ray.is_initialized(): + ray.shutdown() + + skipped = [(node_id, outcome) for kind, node_id, outcome in events if kind == "skipped"] + assert result == "persisted:required-value" + assert len(skipped) == 1 + assert skipped[0][0] == "optional_1" + assert skipped[0][1] == ava.SkipOutcome( + reason="No optional rows", metadata={"partition": "2026-07-22"} + ) + assert {node_id for kind, node_id, _ in events if kind == "success"} == { + "required_1", + "persist_1", + } + + +def test_skipped_persistence_step_appends_no_row(): + appended: list[Any] = [] + skipped: list[tuple[str, ava.SkipOutcome]] = [] + + @ava.source + def optional(): + return ava.skip("Source is empty") + + @ava.dest + def persist(value: Any = None): + if value is None: + return ava.skip("Nothing to persist", {"rows": 0}) + appended.append(value) + return value + + @ava.workflow + def persistence_workflow(): + return optional() >> persist() + + result = ( + persistence_workflow() + .run( + executor=ava.LocalExecutor(), + hooks=RunHooks( + on_node_skip=lambda node_id, outcome: skipped.append((node_id, outcome)) + ), + ) + .result() + ) + + assert result is None + assert appended == [] + assert skipped == [ + ("optional_1", ava.SkipOutcome("Source is empty")), + ("persist_1", ava.SkipOutcome("Nothing to persist", {"rows": 0})), + ] + + +def test_rerun_skip_contributes_no_value_or_lineage(): + observed: dict[str, Any] = {} + + @ava.source(slug="optional") + def optional(): + return ava.skip("Not present in rerun") + + @ava.source(slug="required") + def required(): + return "replayed-value" + + @ava.step(slug="combine") + def combine(value: str, ctx: ava.RunContext): + observed["value"] = value + observed["lineage"] = dict(ctx.lineage_vector) + return value + + @ava.workflow + def rerunnable_workflow(): + return (optional() & required()) >> combine() + + result = ( + rerunnable_workflow() + .run( + executor=ava.LocalExecutor(), + run_id="rerun_execution", + rerun=ava.Rerun( + run_id="original_run", + start=["optional", "required"], + mode="autorun", + ), + ) + .result() + ) + + assert result == "replayed-value" + assert observed == { + "value": "replayed-value", + "lineage": {"required": "rerun_execution"}, + } + + +def test_skip_copies_caller_metadata(): + metadata = {"attempt": 1} + + outcome = ava.skip("No work", metadata) + metadata["attempt"] = 2 + + assert outcome == ava.SkipOutcome("No work", {"attempt": 1}) + + +def test_skipped_node_state_round_trips_over_operator_wire_format(): + from avalanche.operator.convert import node_state_from_proto, node_state_to_proto + from avalanche.operator.models import NodeState, NodeStatus + + state = NodeState( + node_id="optional_1", + name="optional", + node_type="source", + status=NodeStatus.SKIPPED, + started_at=10.0, + ended_at=11.5, + reason="No rows", + metadata={"partition": "west", "attempt": 2}, + ) + + restored = node_state_from_proto(node_state_to_proto(state)) + + assert restored == state diff --git a/test/tui_test.py b/test/tui_test.py index ed7e0d2..90aba55 100644 --- a/test/tui_test.py +++ b/test/tui_test.py @@ -1702,6 +1702,31 @@ def test_log_panel_renders_datetime(self): rendered = w.render().plain assert "2026-03-27 14:35:01" in rendered + def test_log_panel_renders_authored_skip_reason_and_metadata(self): + from avalanche.tui.widgets.log_panel import LogWidget + + store = UIStore(MockStateProvider()) + node = store.all_nodes[0] + store.select_node(node) + store.current_run = RunState(run_id="skip_run", flow_name="skip_workflow") + store.current_run.nodes[node.name] = NodeState( + node_id=node.name, + name=node.display_name, + node_type=node.node_type, + status=NodeStatus.SKIPPED, + started_at=10.0, + ended_at=11.0, + reason="No matching files", + metadata={"partition": "2026-07-22"}, + ) + widget = LogWidget() + widget._test_store = store + + rendered = widget.render().plain + + assert "Skipped — No matching files" in rendered + assert 'Metadata: {"partition": "2026-07-22"}' in rendered + # ── Headless interaction tests (Textual pilot) ──────────────────────────── From a2e1cf2c4c61ba5cc59626a75f78617bd413fff2 Mon Sep 17 00:00:00 2001 From: Emile Riberdy <38288472+magix022@users.noreply.github.com> Date: Wed, 22 Jul 2026 14:57:57 +0000 Subject: [PATCH 2/2] Fix deferred Ray input materialization --- src/avalanche/dag.py | 23 +++++++++++++---------- 1 file changed, 13 insertions(+), 10 deletions(-) diff --git a/src/avalanche/dag.py b/src/avalanche/dag.py index d24ccce..e94551e 100644 --- a/src/avalanche/dag.py +++ b/src/avalanche/dag.py @@ -1593,19 +1593,18 @@ def _materialize_append_handles_for_worker(value: Any) -> Any: def _materialize_worker_kwargs(kwargs: dict[str, Any]) -> dict[str, Any]: - """Materialize worker kwargs, EXCEPT hidden Stream parent kwargs. - - A deferred Stream upstream is lifted into a top-level hidden kwarg named - ``__ava_stream_parent_*`` so Ray tracks the producer as a dependency. That - value must NOT be generic-materialized here: ``stream_wrapper`` owns its - control/data split and pops the hidden kwarg itself, converting the small - ``AppendResultHandle`` into an ``AppendResult`` only when it decides to. - Fetching the handle's ``data_ref`` here would defeat the split (early frame - fetch) and bypass Stream's own worker-side resolver. + """Materialize worker kwargs except dependency-only framework carriers. + + Deferred Stream and implicit upstream values are lifted into top-level hidden + kwargs so Ray tracks their producers as dependencies. The implicit carriers + are inspected only for skip outcomes before binding; Stream owns its + control/data split. Generic materialization here would eagerly fetch an + ``AppendResultHandle`` that neither path passes to ordinary user code. """ return { key: value - if isinstance(key, str) and key.startswith(_STREAM_PARENT_KWARG_PREFIX) + if isinstance(key, str) + and key.startswith((_STREAM_PARENT_KWARG_PREFIX, _IMPLICIT_VALUE_KWARG_PREFIX)) else _materialize_append_handles_for_worker(value) for key, value in kwargs.items() } @@ -1751,6 +1750,8 @@ def plain(*args: Any, **kwargs: Any) -> Any: ), adapt_positionals=adapt_implicit_positionals, ) + args = _materialize_append_handles_for_worker(args) + kwargs = _materialize_worker_kwargs(kwargs) args, kwargs = _prepare_user_call(args, kwargs) result = fn(*_unwrap_lineaged_tree(args), **_unwrap_lineaged_tree(kwargs)) return _shape_skipped_result(result, num_returns=num_returns) @@ -1780,6 +1781,8 @@ def wrapped(*args: Any, **kwargs: Any) -> Any: position_consuming_param_names=(implicit_position_consuming_param_names or set()), adapt_positionals=adapt_implicit_positionals, ) + args = _materialize_append_handles_for_worker(args) + kwargs = _materialize_worker_kwargs(kwargs) args, kwargs = _prepare_user_call(args, kwargs) unwrapped_args = _unwrap_lineaged_tree(args)