From 85476e66f9a16945bd9e45030b28e9e39983757c Mon Sep 17 00:00:00 2001 From: Oliver Mohr Date: Sun, 14 Jun 2026 20:08:34 -0400 Subject: [PATCH 01/10] feat: promote ErrorData to public API and add timing/attempts to TaskResult - Rename internal _ErrorData dataclass to public ErrorData, exported from the package root, while keeping _ErrorContextFormatter private. - TaskResult now records elapsed_seconds and attempts for every run, laying the groundwork for a process execution report. --- .gitignore | 4 ++- src/processes/__init__.py | 1 + src/processes/_error_data.py | 8 +++--- src/processes/_webhook_internals.py | 8 +++--- src/processes/task.py | 39 ++++++++++++++++++++++++++--- 5 files changed, 48 insertions(+), 12 deletions(-) diff --git a/.gitignore b/.gitignore index 47e92eb..43673bb 100644 --- a/.gitignore +++ b/.gitignore @@ -182,4 +182,6 @@ mail_config.toml logs/* logs -PR_DESCRIPTION.md \ No newline at end of file +PR_DESCRIPTION.md + +FEATURE_PLAN.md \ No newline at end of file diff --git a/src/processes/__init__.py b/src/processes/__init__.py index 535262a..411fc9f 100644 --- a/src/processes/__init__.py +++ b/src/processes/__init__.py @@ -1,6 +1,7 @@ from importlib.metadata import PackageNotFoundError as _pnfe from importlib.metadata import version as _v +from ._error_data import ErrorData as ErrorData from .email_config import HTMLEmailStyle as HTMLEmailStyle from .email_config import SMTPConfig as SMTPConfig from .exceptions import ( diff --git a/src/processes/_error_data.py b/src/processes/_error_data.py index 061d3e1..2aa7a4b 100644 --- a/src/processes/_error_data.py +++ b/src/processes/_error_data.py @@ -6,7 +6,7 @@ @dataclass(frozen=True) -class _ErrorData: +class ErrorData: """Typed view of a task failure, extracted from ``record.task_context``. Attributes @@ -46,7 +46,7 @@ class _ErrorData: class _ErrorContextFormatter(logging.Formatter): """Base formatter providing typed access to a record's failure context.""" - def _error_data(self, record: logging.LogRecord) -> _ErrorData: + def _error_data(self, record: logging.LogRecord) -> ErrorData: """Extract the failure context from a log record. Parameters @@ -57,12 +57,12 @@ def _error_data(self, record: logging.LogRecord) -> _ErrorData: Returns ------- - _ErrorData + ErrorData Typed view of ``record.task_context``, with defaults filled in for any missing fields. """ ctx = getattr(record, "task_context", None) or {} - return _ErrorData( + return ErrorData( task_name=str(ctx.get("task_name", "?")), function=str(ctx.get("function", "?")), args=ctx.get("args", ()), diff --git a/src/processes/_webhook_internals.py b/src/processes/_webhook_internals.py index f8f7d1f..749b927 100644 --- a/src/processes/_webhook_internals.py +++ b/src/processes/_webhook_internals.py @@ -7,7 +7,7 @@ import urllib.request from typing import Any -from ._error_data import _ErrorContextFormatter, _ErrorData +from ._error_data import ErrorData, _ErrorContextFormatter from .webhook_config import WebhookConfig _SIGNATURE_HEADER = "X-Signature-SHA256" @@ -46,8 +46,8 @@ def format(self, record: logging.LogRecord) -> str: payload = {**generic_payload, **self._extra_payload} return json.dumps(payload) - def _build_payload(self, error: _ErrorData) -> dict[str, Any]: - """Build the JSON-serializable payload dict from ``_ErrorData``. + def _build_payload(self, error: ErrorData) -> dict[str, Any]: + """Build the JSON-serializable payload dict from ``ErrorData``. Subclasses targeting a specific webhook service can override this to reshape the payload, while reusing ``format`` and the rest of @@ -55,7 +55,7 @@ def _build_payload(self, error: _ErrorData) -> dict[str, Any]: Parameters ---------- - error : _ErrorData + error : ErrorData Typed failure context for the record being formatted. Returns diff --git a/src/processes/task.py b/src/processes/task.py index af17cf6..6f1f555 100644 --- a/src/processes/task.py +++ b/src/processes/task.py @@ -1,6 +1,7 @@ from __future__ import annotations import concurrent.futures +import time from collections.abc import Callable from typing import TYPE_CHECKING, Any @@ -9,6 +10,7 @@ import logging +from ._error_data import ErrorData from ._tb_utils import _build_traced_vars, _build_traced_vars_location, _format_traceback from .exceptions import CircularDependencyError from .notification_channels import NotificationChannel, _FileChannel @@ -29,12 +31,33 @@ class TaskResult: The return value of the task's function if execution succeeded, None if failed. exception : Exception | None The exception object if execution failed, None if successful. + error_data : ErrorData | None + Structured failure context (function, args, kwargs, traceback, traced + variables, downstream impact) when execution failed; None if the task + succeeded. + elapsed_seconds : float + Wall-clock time spent running the task across all attempts, in + seconds. Defaults to ``0.0``. + attempts : int + Number of attempts actually executed (1 or more if the task ran, + 0 if it never ran). Defaults to ``0``. """ - def __init__(self, worked: bool, result: Any, exception: Exception | None): + def __init__( + self, + worked: bool, + result: Any, + exception: Exception | None, + error_data: ErrorData | None = None, + elapsed_seconds: float = 0.0, + attempts: int = 0, + ): self.worked = worked self.result = result self.exception = exception + self.error_data = error_data + self.elapsed_seconds = elapsed_seconds + self.attempts = attempts class TaskDependency: @@ -405,12 +428,14 @@ def run(self, executing_process: Process | None = None) -> TaskResult: self.logger.info(f"Starting {self.name}.") last_exc: Exception | None = None + start = time.monotonic() for attempt in range(1, max_attempts + 1): try: result = self._call_with_timeout(final_args, final_kwargs) self.logger.info(f"Finished {self.name}.") - return TaskResult(True, result, None) + elapsed = time.monotonic() - start + return TaskResult(True, result, None, elapsed_seconds=elapsed, attempts=attempt) except Exception as e: last_exc = e retryable = self.retries >= 1 and attempt < max_attempts @@ -422,6 +447,14 @@ def run(self, executing_process: Process | None = None) -> TaskResult: break assert last_exc is not None + elapsed = time.monotonic() - start task_context = self._build_failure_context(last_exc, executing_process) self.logger.error(str(last_exc), exc_info=last_exc, extra={"task_context": task_context}) - return TaskResult(False, None, last_exc) + return TaskResult( + False, + None, + last_exc, + error_data=ErrorData(**task_context), + elapsed_seconds=elapsed, + attempts=attempt, + ) From 6042090699b33aac31210b2c0c204833b86937e7 Mon Sep 17 00:00:00 2001 From: Oliver Mohr Date: Sun, 14 Jun 2026 20:09:32 -0400 Subject: [PATCH 02/10] feat: track failed task results in ProcessRunner and ProcessResult Add failed_results/failed_tasks_results dicts alongside the existing failed_tasks set so callers can access the TaskResult (including ErrorData) for each errored task, not just its name. --- src/processes/process.py | 24 ++++++++++++++++++++++-- 1 file changed, 22 insertions(+), 2 deletions(-) diff --git a/src/processes/process.py b/src/processes/process.py index 8deaf41..d050a27 100644 --- a/src/processes/process.py +++ b/src/processes/process.py @@ -3,6 +3,7 @@ from types import TracebackType from typing import Literal, Self +from ._error_data import ErrorData from .exceptions import CircularDependencyError, DependencyNotFoundError, TaskNotFoundError from .task import Task, TaskResult @@ -29,6 +30,9 @@ class ProcessResult: errored_tasks : set[str] Subset of ``failed_tasks``: tasks whose function actually raised an exception (``failed_tasks - skipped_tasks``). + failed_tasks_results : dict[str, TaskResult] + Mapping of errored task names (``errored_tasks``) to the TaskResult + produced by their failed run. Cascade-skipped tasks have no entry here. """ def __init__( @@ -36,11 +40,13 @@ def __init__( passed_tasks_results: dict[str, TaskResult], failed_tasks: set[str], skipped_tasks: set[str], + failed_tasks_results: dict[str, TaskResult] | None = None, ): self.passed_tasks_results = passed_tasks_results self.failed_tasks = failed_tasks self.skipped_tasks = skipped_tasks self.errored_tasks: set[str] = failed_tasks - skipped_tasks + self.failed_tasks_results: dict[str, TaskResult] = failed_tasks_results or {} class Process: @@ -298,6 +304,9 @@ class ProcessRunner: Results from successfully executed tasks. failed_tasks : set[str] Names of tasks that failed during execution. + failed_results : dict[str, TaskResult] + Results from tasks whose function raised an exception, keyed by task + name. Cascade-skipped tasks have no entry here. skipped_tasks : set[str] Names of tasks that were never run because an upstream dependency failed. submitted_tasks : set[str] @@ -308,6 +317,7 @@ def __init__(self, process_ref: Process): self.process = process_ref self.passed_results: dict[str, TaskResult] = {} self.failed_tasks: set[str] = set() + self.failed_results: dict[str, TaskResult] = {} self.skipped_tasks: set[str] = set() self.submitted_tasks: set[str] = set() @@ -341,7 +351,9 @@ def run(self, parallel: bool, max_workers: int) -> ProcessResult: self._run_parallel(max_workers) else: self._run_sequential() - return ProcessResult(self.passed_results, self.failed_tasks, self.skipped_tasks) + return ProcessResult( + self.passed_results, self.failed_tasks, self.skipped_tasks, self.failed_results + ) def _is_unrunnable(self, task: Task) -> bool: """Check if a task cannot be run due to failed dependencies. @@ -390,6 +402,7 @@ def _run_sequential(self) -> None: self.passed_results[task.name] = res else: self.failed_tasks.add(task.name) + self.failed_results[task.name] = res def _run_parallel(self, max_workers: int) -> None: """Execute tasks in parallel using a thread pool while respecting dependencies. @@ -437,8 +450,15 @@ def _run_parallel(self, max_workers: int) -> None: self.passed_results[name] = res else: self.failed_tasks.add(name) - except Exception: + self.failed_results[name] = res + except Exception as e: self.failed_tasks.add(name) + self.failed_results[name] = TaskResult( + False, + None, + e, + error_data=ErrorData(task_name=name, exception=str(e)), + ) else: # No running tasks and no new candidates. The # ``_is_unrunnable`` side effect above may have From 1b6dadb758803e545d85b7e9b15c46925201a022 Mon Sep 17 00:00:00 2001 From: Oliver Mohr Date: Sun, 14 Jun 2026 20:11:20 -0400 Subject: [PATCH 03/10] feat: add ProcessExecutionReport for per-task execution summaries Introduce TaskStatus, TaskReportEntry, and ProcessExecutionReport with a from_result() builder that classifies every task as success, errored, or skipped and exposes its function, args, kwargs, result/error, elapsed time, and attempt count. --- src/processes/__init__.py | 3 + src/processes/execution_report.py | 154 ++++++++++++++++++++++++++++++ tests/test_execution_report.py | 125 ++++++++++++++++++++++++ tests/test_timeout_retry.py | 5 + 4 files changed, 287 insertions(+) create mode 100644 src/processes/execution_report.py create mode 100644 tests/test_execution_report.py diff --git a/src/processes/__init__.py b/src/processes/__init__.py index 411fc9f..4648115 100644 --- a/src/processes/__init__.py +++ b/src/processes/__init__.py @@ -13,6 +13,9 @@ from .exceptions import ( TaskNotFoundError as TaskNotFoundError, ) +from .execution_report import ProcessExecutionReport as ProcessExecutionReport +from .execution_report import TaskReportEntry as TaskReportEntry +from .execution_report import TaskStatus as TaskStatus from .notification_channels import EmailChannel as EmailChannel from .notification_channels import NotificationChannel as NotificationChannel from .notification_channels import WebhookChannel as WebhookChannel diff --git a/src/processes/execution_report.py b/src/processes/execution_report.py new file mode 100644 index 0000000..61cd133 --- /dev/null +++ b/src/processes/execution_report.py @@ -0,0 +1,154 @@ +from __future__ import annotations + +from dataclasses import dataclass, field +from enum import Enum +from typing import TYPE_CHECKING, Any + +from ._error_data import ErrorData + +if TYPE_CHECKING: + from .process import Process, ProcessResult + + +class TaskStatus(Enum): + """Outcome of a task within a process execution. + + Attributes + ---------- + SUCCESS + The task ran and its function returned without raising. + ERRORED + The task ran and its function raised an exception (after exhausting + retries, if any). + SKIPPED + The task never ran because an upstream dependency failed. + """ + + SUCCESS = "success" + ERRORED = "errored" + SKIPPED = "skipped" + + +@dataclass(frozen=True) +class TaskReportEntry: + """Per-task entry in a :class:`ProcessExecutionReport`. + + Attributes + ---------- + name : str + The task's name. + function : str + Name of the function the task runs. + args : tuple[Any, ...] + Positional arguments the task was constructed with. + kwargs : dict[str, Any] + Keyword arguments the task was constructed with. + status : TaskStatus + Outcome of the task: ``SUCCESS``, ``ERRORED``, or ``SKIPPED``. + elapsed_seconds : float + Wall-clock time spent running the task across all attempts. ``0.0`` + for skipped tasks. + attempts : int + Number of attempts actually executed. ``0`` for skipped tasks. + result : Any | None + The task's return value if ``status`` is ``SUCCESS``, else ``None``. + error : ErrorData | None + Structured failure context if ``status`` is ``ERRORED``, else ``None``. + """ + + name: str + function: str + args: tuple[Any, ...] + kwargs: dict[str, Any] + status: TaskStatus + elapsed_seconds: float + attempts: int + result: Any | None = None + error: ErrorData | None = None + + +@dataclass(frozen=True) +class ProcessExecutionReport: + """Per-task breakdown of a finished :meth:`Process.run` call. + + Attributes + ---------- + entries : dict[str, TaskReportEntry] + Mapping of task name to its report entry, ordered the same way as + ``process.tasks`` (topological order). + """ + + entries: dict[str, TaskReportEntry] = field(default_factory=dict) + + def _filter(self, status: TaskStatus) -> dict[str, TaskReportEntry]: + return {name: entry for name, entry in self.entries.items() if entry.status == status} + + @property + def successes(self) -> dict[str, TaskReportEntry]: + """Entries for tasks whose status is ``SUCCESS``.""" + return self._filter(TaskStatus.SUCCESS) + + @property + def errored(self) -> dict[str, TaskReportEntry]: + """Entries for tasks whose status is ``ERRORED``.""" + return self._filter(TaskStatus.ERRORED) + + @property + def skipped(self) -> dict[str, TaskReportEntry]: + """Entries for tasks whose status is ``SKIPPED``.""" + return self._filter(TaskStatus.SKIPPED) + + @classmethod + def from_result(cls, process: Process, process_result: ProcessResult) -> ProcessExecutionReport: + """Build a report from a finished process run. + + Parameters + ---------- + process : Process + The process that was run. Used for task definitions (name, + function, args, kwargs) and topological ordering. + process_result : ProcessResult + The result returned by :meth:`Process.run`. + + Returns + ------- + ProcessExecutionReport + One entry per task in ``process.tasks``, in topological order. + """ + entries: dict[str, TaskReportEntry] = {} + for task in process.tasks: + if task.name in process_result.skipped_tasks: + entries[task.name] = TaskReportEntry( + name=task.name, + function=task.func.__name__, + args=task.args, + kwargs=task.kwargs, + status=TaskStatus.SKIPPED, + elapsed_seconds=0.0, + attempts=0, + ) + elif task.name in process_result.passed_tasks_results: + passed = process_result.passed_tasks_results[task.name] + entries[task.name] = TaskReportEntry( + name=task.name, + function=task.func.__name__, + args=task.args, + kwargs=task.kwargs, + status=TaskStatus.SUCCESS, + elapsed_seconds=passed.elapsed_seconds, + attempts=passed.attempts, + result=passed.result, + ) + else: + failed = process_result.failed_tasks_results[task.name] + entries[task.name] = TaskReportEntry( + name=task.name, + function=task.func.__name__, + args=task.args, + kwargs=task.kwargs, + status=TaskStatus.ERRORED, + elapsed_seconds=failed.elapsed_seconds, + attempts=failed.attempts, + error=failed.error_data, + ) + return cls(entries) diff --git a/tests/test_execution_report.py b/tests/test_execution_report.py new file mode 100644 index 0000000..465a215 --- /dev/null +++ b/tests/test_execution_report.py @@ -0,0 +1,125 @@ +"""Tests for ``ProcessExecutionReport.from_result``. + +Covers status classification (success / errored / skipped), the data carried +by each ``TaskReportEntry`` (result, error, elapsed_seconds, attempts), entry +ordering, and the ``successes``/``errored``/``skipped`` filter accessors — +for both sequential and parallel execution. +""" + +from __future__ import annotations + +from processes import ( + ErrorData, + Process, + ProcessExecutionReport, + Task, + TaskDependency, + TaskStatus, +) + +from .base_test import BaseTest + + +class TestProcessExecutionReport(BaseTest): + def _build_process(self) -> tuple[Process, Task, Task, Task]: + def load() -> str: + return "data" + + def apply(_data: str) -> None: + raise ValueError("apply failed") + + def notify() -> None: + pass + + load_task = Task("load", load, self._log("report_load.log")) + apply_task = Task( + "apply", + apply, + self._log("report_apply.log"), + dependencies=[TaskDependency("load", use_result_as_additional_args=True)], + ) + notify_task = Task( + "notify", + notify, + self._log("report_notify.log"), + dependencies=[TaskDependency("apply")], + ) + + process = Process([load_task, apply_task, notify_task]) + return process, load_task, apply_task, notify_task + + def test_sequential_report_classifies_each_task(self) -> None: + process, load_task, apply_task, notify_task = self._build_process() + with process: + result = process.run(parallel=False) + report = ProcessExecutionReport.from_result(process, result) + + assert list(report.entries) == ["load", "apply", "notify"] + + load_entry = report.entries["load"] + assert load_entry.status == TaskStatus.SUCCESS + assert load_entry.result == "data" + assert load_entry.error is None + assert load_entry.attempts == 1 + assert load_entry.elapsed_seconds >= 0.0 + assert load_entry.function == "load" + assert load_entry.args == () + assert load_entry.kwargs == {} + + apply_entry = report.entries["apply"] + assert apply_entry.status == TaskStatus.ERRORED + assert apply_entry.result is None + assert isinstance(apply_entry.error, ErrorData) + assert apply_entry.error.task_name == "apply" + assert apply_entry.error.exception == "apply failed" + assert apply_entry.attempts == 1 + + notify_entry = report.entries["notify"] + assert notify_entry.status == TaskStatus.SKIPPED + assert notify_entry.result is None + assert notify_entry.error is None + assert notify_entry.attempts == 0 + assert notify_entry.elapsed_seconds == 0.0 + + self._close_handlers(load_task, apply_task, notify_task) + + def test_filter_accessors_partition_entries(self) -> None: + process, load_task, apply_task, notify_task = self._build_process() + with process: + result = process.run(parallel=False) + report = ProcessExecutionReport.from_result(process, result) + + assert set(report.successes) == {"load"} + assert set(report.errored) == {"apply"} + assert set(report.skipped) == {"notify"} + + self._close_handlers(load_task, apply_task, notify_task) + + def test_parallel_report_classifies_each_task(self) -> None: + process, load_task, apply_task, notify_task = self._build_process() + with process: + result = process.run(parallel=True, max_workers=4) + report = ProcessExecutionReport.from_result(process, result) + + assert report.entries["load"].status == TaskStatus.SUCCESS + assert report.entries["apply"].status == TaskStatus.ERRORED + assert report.entries["notify"].status == TaskStatus.SKIPPED + + self._close_handlers(load_task, apply_task, notify_task) + + def test_all_success_report_has_no_errored_or_skipped(self) -> None: + def step() -> int: + return 42 + + task = Task("step", step, self._log("report_all_success.log")) + process = Process([task]) + with process: + result = process.run(parallel=False) + report = ProcessExecutionReport.from_result(process, result) + + assert report.entries["step"].status == TaskStatus.SUCCESS + assert report.entries["step"].result == 42 + assert report.errored == {} + assert report.skipped == {} + + self._close_handlers(task) diff --git a/tests/test_timeout_retry.py b/tests/test_timeout_retry.py index e9f266b..a08bb35 100644 --- a/tests/test_timeout_retry.py +++ b/tests/test_timeout_retry.py @@ -44,6 +44,8 @@ def fast() -> int: assert result.worked assert result.result == 42 assert result.exception is None + assert result.attempts == 1 + assert result.elapsed_seconds >= 0.0 def test_timeout_in_sequential_process(self) -> None: """Timeout propagates through Process.run(parallel=False).""" @@ -135,6 +137,7 @@ def flaky() -> str: assert result.worked assert result.result == "ok" assert len(calls) == 2 + assert result.attempts == 2 def test_retry_exhausted_returns_failed(self) -> None: """A task that fails all attempts returns TaskResult(False, ...).""" @@ -153,6 +156,8 @@ def always_fails() -> None: assert not result.worked assert isinstance(result.exception, ConnectionError) assert len(calls) == 3 # 1 original + 2 retries + assert result.attempts == 3 + assert result.elapsed_seconds >= 0.0 def test_retry_zero_means_no_retry(self) -> None: """retries=0 (default) never retries, even for a retryable exception.""" From 65100f9c0075c8d090625754e49b081713505ad8 Mon Sep 17 00:00:00 2001 From: Oliver Mohr Date: Sun, 14 Jun 2026 20:12:18 -0400 Subject: [PATCH 04/10] docs: document ProcessExecutionReport, ErrorData, and TaskResult timing fields --- README.md | 55 ++++++++++++++++++++++++++++++++++++++++++++++- docs/reference.md | 5 +++++ 2 files changed, 59 insertions(+), 1 deletion(-) diff --git a/README.md b/README.md index 700f959..3c1a76b 100644 --- a/README.md +++ b/README.md @@ -277,12 +277,65 @@ process.run(parallel: bool | None = None, max_workers: int = 4) -> ProcessResult ```python result.passed_tasks_results # dict[str, TaskResult] — name → TaskResult for every task that succeeded result.failed_tasks # set[str] — all tasks that did not produce a result (errored + skipped) +result.failed_tasks_results # dict[str, TaskResult] — name → TaskResult for every task that errored result.errored_tasks # set[str] — tasks whose function actually raised result.skipped_tasks # set[str] — tasks skipped because an upstream dependency failed -TaskResult(worked: bool, result: Any, exception: Exception | None) +TaskResult( + worked: bool, + result: Any, + exception: Exception | None, + error_data: ErrorData | None = None, + elapsed_seconds: float = 0.0, # wall-clock time across all attempts + attempts: int = 0, # attempts actually executed (0 if never run) +) +``` + +### `ProcessExecutionReport` + +```python +ProcessExecutionReport.from_result(process: Process, result: ProcessResult) -> ProcessExecutionReport + +report.entries # dict[str, TaskReportEntry] — one entry per task, in topological order +report.successes # dict[str, TaskReportEntry] — entries with status == TaskStatus.SUCCESS +report.errored # dict[str, TaskReportEntry] — entries with status == TaskStatus.ERRORED +report.skipped # dict[str, TaskReportEntry] — entries with status == TaskStatus.SKIPPED +``` + +A `TaskReportEntry` carries `name`, `function`, `args`, `kwargs`, `status` +(`TaskStatus.SUCCESS | ERRORED | SKIPPED`), `elapsed_seconds`, `attempts`, +plus `result` (set when `SUCCESS`) and `error: ErrorData | None` (set when +`ERRORED`). + +```python +with Process([load_task, apply_task, notify_task]) as process: + result = process.run() + +report = ProcessExecutionReport.from_result(process, result) +for name, entry in report.entries.items(): + if entry.status is TaskStatus.ERRORED: + print(f"{name} failed after {entry.attempts} attempt(s): {entry.error.exception}") ``` +### `ErrorData` + +```python +ErrorData( + task_name: str, + function: str, + args: tuple[Any, ...], + kwargs: dict[str, Any], + downstream_impact: list[str], + exception: str, + traceback_str: str, + traced_vars: dict[str, str], + traced_vars_location: str, +) +``` + +Structured failure context for a single task, available via +`TaskResult.error_data` and `TaskReportEntry.error`. + ### `SMTPConfig` ```python diff --git a/docs/reference.md b/docs/reference.md index 20cb014..cf31beb 100644 --- a/docs/reference.md +++ b/docs/reference.md @@ -6,5 +6,10 @@ This page is automatically generated from the source code docstrings. ::: processes.process.ProcessResult ::: processes.task.Task ::: processes.task.TaskDependency +::: processes.task.TaskResult +::: processes.execution_report.ProcessExecutionReport +::: processes.execution_report.TaskReportEntry +::: processes.execution_report.TaskStatus +::: processes._error_data.ErrorData ::: processes.email_config.SMTPConfig ::: processes.email_config.HTMLEmailStyle \ No newline at end of file From 58f900b861c195b2655986ebd58b2e6692917296 Mon Sep 17 00:00:00 2001 From: Oliver Mohr Date: Sun, 14 Jun 2026 20:44:15 -0400 Subject: [PATCH 05/10] docs: add plan --- FEATURE_PLAN1.MD | 86 ++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 86 insertions(+) create mode 100644 FEATURE_PLAN1.MD diff --git a/FEATURE_PLAN1.MD b/FEATURE_PLAN1.MD new file mode 100644 index 0000000..f04a0e6 --- /dev/null +++ b/FEATURE_PLAN1.MD @@ -0,0 +1,86 @@ +USER: Could ProcessResult be ditched and replaced by ProcessExecutionReport? Show impacts. Don't code yet. + +Claude: Could ProcessResult be replaced by ProcessExecutionReport? + +Structurally, yes — ProcessExecutionReport.entries is a superset of what ProcessResult carries (it adds function/args/kwargs/status/elapsed_seconds/attempts for every task, success/errored/skipped alike, in topological order). The raw data ProcessRunner already collects (passed_results, failed_results, skipped_tasks, process.tasks) is exactly what from_result consumes — so ProcessRunner.run() could build a ProcessExecutionReport directly instead of a ProcessResult, without from_result needing a ProcessResult input at all (avoids the current process.py ↔ execution_report.py TYPE_CHECKING coupling — execution_report.py becomes self-sufficient). + +What breaks / changes + +1. Lost: raw Exception object access +TaskResult.exception is a live Exception instance (tests do isinstance(result.exception, TimeoutError)). TaskReportEntry.error is ErrorData | None, whose .exception field is a str(exception). Any caller that needs to isinstance()-check or re-raise the original exception loses that — ErrorData would need an added field, or TaskReportEntry would need to carry the raw TaskResult too. + +2. API shape changes for every consumer +- result.passed_tasks_results["x"].result → report.successes["x"].result (or report.entries["x"].result) +- result.failed_tasks (a set[str]) → no direct equivalent; becomes set(report.errored) | set(report.skipped) +- result.errored_tasks / result.skipped_tasks → set(report.errored) / set(report.skipped) +- result.failed_tasks_results["x"] (a TaskResult) → report.entries["x"].error (an ErrorData, narrower — see #1) +- .worked / boolean checks → .status == TaskStatus.SUCCESS etc. + +3. Test impact — large +Every test file that touches these attributes needs rewriting: +test_args_kwargs.py, test_complex_dag_failures.py, test_normal_run_no_errors.py, test_parallel_race_conditions.py, test_timeout_retry.py, plus 3 manual-test scripts. This is the bulk of the effort — dozens of assertion sites. + +4. Docs impact +README's ProcessResult section (added in this session) and any usage examples would need replacing with ProcessExecutionReport equivalents; docs/reference.md would drop the ProcessResult mkdocstrings entry. + +5. Breaking change vs. additive discipline +This session deliberately kept everything additive (FEATURE_PLAN's stance). Removing ProcessResult and changing Process.run()'s return type is a hard breaking change — would warrant a major version bump and a migration note, not a patch/minor release. + +6. Minor ergonomics loss +Simple "did everything pass?" checks (if not result.failed_tasks) become if not report.errored and not report.skipped unless a convenience property (e.g. report.ok) is added. + +Net assessment + +Feasible and arguably a cleaner unified model long-term, but it's a breaking API change with wide blast radius across tests/docs, and needs at least one new field on ErrorData/TaskReportEntry to avoid losing the raw exception object. If you want this, I'd suggest either (a) a v6 major bump doing a clean replacement, or (b) keep ProcessResult as a thin deprecated wrapper for one release while Process.run() returns ProcessExecutionReport and ProcessResult.from_report(...) is offered for back-compat. + +✻ Cooked for 49s + +❯ /advisor off + ⎿ Advisor disabled + +❯ is ProcessExecutionReport being created anywhere yet? + + Searched for 1 pattern + +Every test file that touches these attributes needs rewriting: +test_args_kwargs.py, test_complex_dag_failures.py, test_normal_run_no_errors.py, test_parallel_race_conditions.py, test_timeout_retry.py, plus 3 manual-test scripts. This is the bulk of the effort — dozens of assertion sites. + +4. Docs impact +README's ProcessResult section (added in this session) and any usage examples would need replacing with ProcessExecutionReport equivalents; docs/reference.md would drop the ProcessResult mkdocstrings entry. + +5. Breaking change vs. additive discipline +This session deliberately kept everything additive (FEATURE_PLAN's stance). Removing ProcessResult and changing Process.run()'s return type is a hard breaking change — would warrant a major version bump and a migration note, not a patch/minor release. + +6. Minor ergonomics loss +Simple "did everything pass?" checks (if not result.failed_tasks) become if not report.errored and not report.skipped unless a convenience property (e.g. report.ok) is added. + +Net assessment + +Feasible and arguably a cleaner unified model long-term, but it's a breaking API change with wide blast radius across tests/docs, and needs at least one new field on ErrorData/TaskReportEntry to avoid losing the raw exception object. If you want this, I'd suggest either (a) a v6 major bump doing a clean replacement, or (b) keep ProcessResult as a thin deprecated wrapper for one release while Process.run() returns ProcessExecutionReport and ProcessResult.from_report(...) is offered for back-compat. + +USER: failed_results and passed_results and failed_tasks are separated. Shouldn't they be in a single dict with k:v task.name:task_result_obj ? + +Claude: Yes — and it'd fix a latent inconsistency: currently failed_tasks (a set) = errored_tasks ∪ skipped_tasks, but failed_results (a dict) only has entries for errored tasks, not skipped ones. So failed_tasks and failed_results.keys() silently diverge — anyone iterating failed_tasks and indexing into failed_results will KeyError on skipped tasks. + +Proposal + +Maintain one internal dict in ProcessRunner: self.results: dict[str, TaskResult], one entry per task regardless of outcome (success, errored, or skipped — synthesizing a TaskResult for skipped tasks the same way the parallel defensive branch already does for unexpected exceptions). + +Discriminator problem: TaskResult.worked: bool alone can't tell errored apart from skipped (both worked=False). Two options: +- Implicit: attempts == 0 ⇒ skipped. Works, but relies on an unstated convention — fragile if someone constructs a TaskResult manually. +- Explicit (cleaner): give TaskResult a status: TaskStatus field (reuse the enum from execution_report.py). Then results[name].status is authoritative everywhere. + +Impact + +- Internals: _is_done, _all_deps_met, _is_unrunnable currently check membership across passed_results/failed_tasks/skipped_tasks/failed_results — all rewritten to query self.results by status. Net simplification. +- ProcessResult public API: can stay byte-for-byte unchanged. passed_tasks_results, failed_tasks, skipped_tasks, errored_tasks, failed_tasks_results all become derived views filtered from self.results at the end of run() — zero breaking change, fixes the divergence bug above as a side effect. +- Ties into the ProcessExecutionReport question: if TaskResult gains status: TaskStatus, then self.results is already shaped almost exactly like ProcessExecutionReport.entries (just missing name/function/args/kwargs, which come from Task). from_result/ProcessRunner.run() could build the report directly from this dict with minimal mapping — makes that future migration much cheaper. +- Effort: moderate, contained to process.py (ProcessRunner) + adding a field to TaskResult (additive, default-valued — safe). No test changes needed if ProcessResult's shape is preserved. + +This seems like a good incremental step worth doing on its own regardless of the ProcessExecutionReport replacement decision. + +USER: Tasks could have an extra attr TaskStatus already defined. Status could be expanded to PENDING, SUCCESS, ERRORED, SKIPPED. These changes should simplify or solve some of the problems stated by you. + + + +INSTRUCTION: IMPLEMENT BOTH THINGS. Removal of ProcessResult and use of single dict self.results. \ No newline at end of file From 103863961116f7ff1b05897311ee3d363c8ccb1a Mon Sep 17 00:00:00 2001 From: Oliver Mohr Date: Sun, 14 Jun 2026 22:58:24 -0400 Subject: [PATCH 06/10] feat: expand TaskStatus with PENDING and add status field to TaskResult TaskStatus moves to task.py (PENDING/SUCCESS/ERRORED/SKIPPED) so TaskResult can carry it directly. worked becomes a derived property (status == SUCCESS), keeping the existing call sites working. --- src/processes/__init__.py | 2 +- src/processes/execution_report.py | 21 +------------- src/processes/process.py | 4 +-- src/processes/task.py | 46 ++++++++++++++++++++++++++----- 4 files changed, 43 insertions(+), 30 deletions(-) diff --git a/src/processes/__init__.py b/src/processes/__init__.py index 4648115..0f8f50e 100644 --- a/src/processes/__init__.py +++ b/src/processes/__init__.py @@ -15,7 +15,6 @@ ) from .execution_report import ProcessExecutionReport as ProcessExecutionReport from .execution_report import TaskReportEntry as TaskReportEntry -from .execution_report import TaskStatus as TaskStatus from .notification_channels import EmailChannel as EmailChannel from .notification_channels import NotificationChannel as NotificationChannel from .notification_channels import WebhookChannel as WebhookChannel @@ -23,6 +22,7 @@ from .task import Task as Task from .task import TaskDependency as TaskDependency from .task import TaskResult as TaskResult +from .task import TaskStatus as TaskStatus from .webhook_config import WebhookConfig as WebhookConfig try: diff --git a/src/processes/execution_report.py b/src/processes/execution_report.py index 61cd133..c56b352 100644 --- a/src/processes/execution_report.py +++ b/src/processes/execution_report.py @@ -1,34 +1,15 @@ from __future__ import annotations from dataclasses import dataclass, field -from enum import Enum from typing import TYPE_CHECKING, Any from ._error_data import ErrorData +from .task import TaskStatus if TYPE_CHECKING: from .process import Process, ProcessResult -class TaskStatus(Enum): - """Outcome of a task within a process execution. - - Attributes - ---------- - SUCCESS - The task ran and its function returned without raising. - ERRORED - The task ran and its function raised an exception (after exhausting - retries, if any). - SKIPPED - The task never ran because an upstream dependency failed. - """ - - SUCCESS = "success" - ERRORED = "errored" - SKIPPED = "skipped" - - @dataclass(frozen=True) class TaskReportEntry: """Per-task entry in a :class:`ProcessExecutionReport`. diff --git a/src/processes/process.py b/src/processes/process.py index d050a27..66ebb3b 100644 --- a/src/processes/process.py +++ b/src/processes/process.py @@ -5,7 +5,7 @@ from ._error_data import ErrorData from .exceptions import CircularDependencyError, DependencyNotFoundError, TaskNotFoundError -from .task import Task, TaskResult +from .task import Task, TaskResult, TaskStatus __all__ = ["CircularDependencyError", "DependencyNotFoundError", "TaskNotFoundError"] @@ -454,7 +454,7 @@ def _run_parallel(self, max_workers: int) -> None: except Exception as e: self.failed_tasks.add(name) self.failed_results[name] = TaskResult( - False, + TaskStatus.ERRORED, None, e, error_data=ErrorData(task_name=name, exception=str(e)), diff --git a/src/processes/task.py b/src/processes/task.py index 6f1f555..0403cce 100644 --- a/src/processes/task.py +++ b/src/processes/task.py @@ -3,6 +3,7 @@ import concurrent.futures import time from collections.abc import Callable +from enum import Enum from typing import TYPE_CHECKING, Any if TYPE_CHECKING: @@ -16,17 +17,41 @@ from .notification_channels import NotificationChannel, _FileChannel +class TaskStatus(Enum): + """Outcome of a task within a process execution. + + Attributes + ---------- + PENDING + The task has not been executed yet. + SUCCESS + The task ran and its function returned without raising. + ERRORED + The task ran and its function raised an exception (after exhausting + retries, if any). + SKIPPED + The task never ran because an upstream dependency failed. + """ + + PENDING = "pending" + SUCCESS = "success" + ERRORED = "errored" + SKIPPED = "skipped" + + class TaskResult: """ Container for the result of a task execution. - Holds the outcome of running a task, including whether it succeeded, - its return value, and any exception that occurred. + Holds the outcome of running a task, including its status, return value, + and any exception that occurred. Attributes ---------- + status : TaskStatus + Outcome of the task: ``PENDING``, ``SUCCESS``, ``ERRORED``, or ``SKIPPED``. worked : bool - True if the task executed successfully, False if an exception occurred. + True if ``status`` is ``SUCCESS``, False otherwise. result : Any The return value of the task's function if execution succeeded, None if failed. exception : Exception | None @@ -45,20 +70,25 @@ class TaskResult: def __init__( self, - worked: bool, + status: TaskStatus, result: Any, exception: Exception | None, error_data: ErrorData | None = None, elapsed_seconds: float = 0.0, attempts: int = 0, ): - self.worked = worked + self.status = status self.result = result self.exception = exception self.error_data = error_data self.elapsed_seconds = elapsed_seconds self.attempts = attempts + @property + def worked(self) -> bool: + """True if the task executed successfully, False otherwise.""" + return self.status == TaskStatus.SUCCESS + class TaskDependency: """ @@ -435,7 +465,9 @@ def run(self, executing_process: Process | None = None) -> TaskResult: result = self._call_with_timeout(final_args, final_kwargs) self.logger.info(f"Finished {self.name}.") elapsed = time.monotonic() - start - return TaskResult(True, result, None, elapsed_seconds=elapsed, attempts=attempt) + return TaskResult( + TaskStatus.SUCCESS, result, None, elapsed_seconds=elapsed, attempts=attempt + ) except Exception as e: last_exc = e retryable = self.retries >= 1 and attempt < max_attempts @@ -451,7 +483,7 @@ def run(self, executing_process: Process | None = None) -> TaskResult: task_context = self._build_failure_context(last_exc, executing_process) self.logger.error(str(last_exc), exc_info=last_exc, extra={"task_context": task_context}) return TaskResult( - False, + TaskStatus.ERRORED, None, last_exc, error_data=ErrorData(**task_context), From 8c049fd1bc809a16f9880695bd076739ee4356db Mon Sep 17 00:00:00 2001 From: Oliver Mohr Date: Sun, 14 Jun 2026 23:03:17 -0400 Subject: [PATCH 07/10] feat!: remove ProcessResult, have Process.run() return ProcessExecutionReport ProcessRunner now tracks a single dict[str, TaskResult] (self.results), seeded with PENDING entries for every task and transitioned to SUCCESS/ERRORED/SKIPPED as tasks resolve. This replaces the separate passed_results/failed_tasks/failed_results/skipped_tasks bookkeeping and fixes the latent divergence between failed_tasks and failed_results for cascade-skipped tasks. Process.run() now returns a ProcessExecutionReport directly, built via ProcessExecutionReport.from_results(process, results). ProcessResult and from_result are removed as part of this breaking change. BREAKING CHANGE: Process.run() returns ProcessExecutionReport instead of ProcessResult. Use report.successes / report.errored / report.skipped instead of passed_tasks_results / errored_tasks / skipped_tasks, and report.entries[name].result / .error instead of passed_tasks_results[name] / failed_tasks_results[name]. --- src/processes/execution_report.py | 59 +++----- src/processes/process.py | 135 +++++------------- src/processes/task.py | 2 +- tests/manual_tests/manual_pipeline_inspect.py | 4 +- .../manual_tests/manual_themed_tracebacks.py | 17 +-- tests/manual_tests/manual_webhook_inspect.py | 4 +- tests/test_args_kwargs.py | 26 ++-- tests/test_complex_dag_failures.py | 33 ++--- tests/test_execution_report.py | 15 +- tests/test_normal_run_no_errors.py | 20 ++- tests/test_parallel_race_conditions.py | 31 ++-- tests/test_timeout_retry.py | 29 ++-- 12 files changed, 144 insertions(+), 231 deletions(-) diff --git a/src/processes/execution_report.py b/src/processes/execution_report.py index c56b352..2997580 100644 --- a/src/processes/execution_report.py +++ b/src/processes/execution_report.py @@ -4,10 +4,10 @@ from typing import TYPE_CHECKING, Any from ._error_data import ErrorData -from .task import TaskStatus +from .task import TaskResult, TaskStatus if TYPE_CHECKING: - from .process import Process, ProcessResult + from .process import Process @dataclass(frozen=True) @@ -80,7 +80,9 @@ def skipped(self) -> dict[str, TaskReportEntry]: return self._filter(TaskStatus.SKIPPED) @classmethod - def from_result(cls, process: Process, process_result: ProcessResult) -> ProcessExecutionReport: + def from_results( + cls, process: Process, results: dict[str, TaskResult] + ) -> ProcessExecutionReport: """Build a report from a finished process run. Parameters @@ -88,8 +90,9 @@ def from_result(cls, process: Process, process_result: ProcessResult) -> Process process : Process The process that was run. Used for task definitions (name, function, args, kwargs) and topological ordering. - process_result : ProcessResult - The result returned by :meth:`Process.run`. + results : dict[str, TaskResult] + One ``TaskResult`` per task, keyed by task name, as produced by + :class:`~processes.process.ProcessRunner`. Returns ------- @@ -98,38 +101,16 @@ def from_result(cls, process: Process, process_result: ProcessResult) -> Process """ entries: dict[str, TaskReportEntry] = {} for task in process.tasks: - if task.name in process_result.skipped_tasks: - entries[task.name] = TaskReportEntry( - name=task.name, - function=task.func.__name__, - args=task.args, - kwargs=task.kwargs, - status=TaskStatus.SKIPPED, - elapsed_seconds=0.0, - attempts=0, - ) - elif task.name in process_result.passed_tasks_results: - passed = process_result.passed_tasks_results[task.name] - entries[task.name] = TaskReportEntry( - name=task.name, - function=task.func.__name__, - args=task.args, - kwargs=task.kwargs, - status=TaskStatus.SUCCESS, - elapsed_seconds=passed.elapsed_seconds, - attempts=passed.attempts, - result=passed.result, - ) - else: - failed = process_result.failed_tasks_results[task.name] - entries[task.name] = TaskReportEntry( - name=task.name, - function=task.func.__name__, - args=task.args, - kwargs=task.kwargs, - status=TaskStatus.ERRORED, - elapsed_seconds=failed.elapsed_seconds, - attempts=failed.attempts, - error=failed.error_data, - ) + res = results[task.name] + entries[task.name] = TaskReportEntry( + name=task.name, + function=task.func.__name__, + args=task.args, + kwargs=task.kwargs, + status=res.status, + elapsed_seconds=res.elapsed_seconds, + attempts=res.attempts, + result=res.result if res.status == TaskStatus.SUCCESS else None, + error=res.error_data if res.status == TaskStatus.ERRORED else None, + ) return cls(entries) diff --git a/src/processes/process.py b/src/processes/process.py index 66ebb3b..2afd3d6 100644 --- a/src/processes/process.py +++ b/src/processes/process.py @@ -5,50 +5,12 @@ from ._error_data import ErrorData from .exceptions import CircularDependencyError, DependencyNotFoundError, TaskNotFoundError +from .execution_report import ProcessExecutionReport from .task import Task, TaskResult, TaskStatus __all__ = ["CircularDependencyError", "DependencyNotFoundError", "TaskNotFoundError"] -class ProcessResult: - """ - Container for the results of a process execution. - - Holds the outcomes of all tasks executed in a process, separating successful - and failed tasks with their respective results. - - Attributes - ---------- - passed_tasks_results : dict[str, TaskResult] - Mapping of task names to TaskResult objects for all tasks that executed successfully. - failed_tasks : set[str] - All tasks that did not produce a result — union of ``errored_tasks`` and - ``skipped_tasks``. - skipped_tasks : set[str] - Subset of ``failed_tasks``: tasks that were never run because an upstream - dependency failed (cascade-skipped). - errored_tasks : set[str] - Subset of ``failed_tasks``: tasks whose function actually raised an exception - (``failed_tasks - skipped_tasks``). - failed_tasks_results : dict[str, TaskResult] - Mapping of errored task names (``errored_tasks``) to the TaskResult - produced by their failed run. Cascade-skipped tasks have no entry here. - """ - - def __init__( - self, - passed_tasks_results: dict[str, TaskResult], - failed_tasks: set[str], - skipped_tasks: set[str], - failed_tasks_results: dict[str, TaskResult] | None = None, - ): - self.passed_tasks_results = passed_tasks_results - self.failed_tasks = failed_tasks - self.skipped_tasks = skipped_tasks - self.errored_tasks: set[str] = failed_tasks - skipped_tasks - self.failed_tasks_results: dict[str, TaskResult] = failed_tasks_results or {} - - class Process: """ Manages and executes a collection of interdependent tasks. @@ -217,7 +179,7 @@ def get_task(self, task_name: str) -> Task: except KeyError as err: raise TaskNotFoundError(task_name) from err - def run(self, parallel: bool | None = None, max_workers: int = 4) -> ProcessResult: + def run(self, parallel: bool | None = None, max_workers: int = 4) -> ProcessExecutionReport: """Execute all tasks in the process. Runs tasks sequentially or in parallel while respecting dependencies. @@ -236,9 +198,8 @@ def run(self, parallel: bool | None = None, max_workers: int = 4) -> ProcessResu Returns ------- - ProcessResult - Contains passed_tasks_results (dict mapping task names to TaskResult) - and failed_tasks (set of task names that failed). + ProcessExecutionReport + Per-task breakdown of the run, in topological order. """ if parallel is None: parallel = len(self.tasks) >= 10 @@ -247,8 +208,7 @@ def run(self, parallel: bool | None = None, max_workers: int = 4) -> ProcessResu if parallel: if max_workers == 1: parallel = False # Fallback to sequential if only one worker - process_result = self.runner.run(parallel, max_workers) - return process_result + return self.runner.run(parallel, max_workers) def get_dependant_tasks(self, task_name: str) -> list[Task]: """Retrieve all tasks that directly or indirectly depend on a given task. @@ -293,46 +253,38 @@ class ProcessRunner: """ Executes tasks in a Process, handling both sequential and parallel execution. - Manages task execution state, tracks passed and failed tasks, and coordinates - dependencies during execution. + Manages task execution state and coordinates dependencies during execution. Attributes ---------- process : Process Reference to the parent Process being executed. - passed_results : dict[str, TaskResult] - Results from successfully executed tasks. - failed_tasks : set[str] - Names of tasks that failed during execution. - failed_results : dict[str, TaskResult] - Results from tasks whose function raised an exception, keyed by task - name. Cascade-skipped tasks have no entry here. - skipped_tasks : set[str] - Names of tasks that were never run because an upstream dependency failed. + results : dict[str, TaskResult] + One entry per task, keyed by task name. Every task starts out + ``PENDING`` and transitions to ``SUCCESS``, ``ERRORED``, or + ``SKIPPED`` once it is resolved. submitted_tasks : set[str] Names of tasks that have been submitted for execution. """ def __init__(self, process_ref: Process): self.process = process_ref - self.passed_results: dict[str, TaskResult] = {} - self.failed_tasks: set[str] = set() - self.failed_results: dict[str, TaskResult] = {} - self.skipped_tasks: set[str] = set() + self.results: dict[str, TaskResult] = { + task.name: TaskResult(TaskStatus.PENDING, None, None) for task in process_ref.tasks + } self.submitted_tasks: set[str] = set() def _is_done(self) -> bool: - """Check whether every task has either passed or failed. + """Check whether every task has been resolved. Returns ------- bool - True if every task has a recorded result (passed or failed), - False otherwise. + True if no task is still ``PENDING``, False otherwise. """ - return len(self.passed_results) + len(self.failed_tasks) >= len(self.process.tasks) + return all(res.status != TaskStatus.PENDING for res in self.results.values()) - def run(self, parallel: bool, max_workers: int) -> ProcessResult: + def run(self, parallel: bool, max_workers: int) -> ProcessExecutionReport: """Execute all tasks in the process using the specified execution mode. Parameters @@ -344,19 +296,17 @@ def run(self, parallel: bool, max_workers: int) -> ProcessResult: Returns ------- - ProcessResult - The combined results of all task executions. + ProcessExecutionReport + Per-task breakdown of the run, in topological order. """ if parallel: self._run_parallel(max_workers) else: self._run_sequential() - return ProcessResult( - self.passed_results, self.failed_tasks, self.skipped_tasks, self.failed_results - ) + return ProcessExecutionReport.from_results(self.process, self.results) def _is_unrunnable(self, task: Task) -> bool: - """Check if a task cannot be run due to failed dependencies. + """Check if a task cannot be run due to failed or skipped dependencies. Parameters ---------- @@ -366,13 +316,15 @@ def _is_unrunnable(self, task: Task) -> bool: Returns ------- bool - True if any of the task's dependencies have failed, False otherwise. - If True, the task is recorded in both ``failed_tasks`` and - ``skipped_tasks`` (cascade-skip). + True if any of the task's dependencies errored or were skipped, + False otherwise. If True, the task is recorded as ``SKIPPED`` + (cascade-skip). """ - if any(d.task_name in self.failed_tasks for d in task.dependencies): - self.failed_tasks.add(task.name) - self.skipped_tasks.add(task.name) + if any( + self.results[d.task_name].status in (TaskStatus.ERRORED, TaskStatus.SKIPPED) + for d in task.dependencies + ): + self.results[task.name] = TaskResult(TaskStatus.SKIPPED, None, None) return True return False @@ -387,9 +339,11 @@ def _all_deps_met(self, task: Task) -> bool: Returns ------- bool - True if all dependencies have passed, False otherwise. + True if all dependencies succeeded, False otherwise. """ - return all(d.task_name in self.passed_results for d in task.dependencies) + return all( + self.results[d.task_name].status == TaskStatus.SUCCESS for d in task.dependencies + ) def _run_sequential(self) -> None: """Execute all tasks sequentially in dependency order.""" @@ -397,12 +351,7 @@ def _run_sequential(self) -> None: if self._is_unrunnable(task): continue if self._all_deps_met(task): - res = task.run(self.process) - if res.worked: - self.passed_results[task.name] = res - else: - self.failed_tasks.add(task.name) - self.failed_results[task.name] = res + self.results[task.name] = task.run(self.process) def _run_parallel(self, max_workers: int) -> None: """Execute tasks in parallel using a thread pool while respecting dependencies. @@ -425,7 +374,7 @@ def _run_parallel(self, max_workers: int) -> None: t for t in self.process.tasks if t.name not in self.submitted_tasks - and t.name not in self.failed_tasks + and self.results[t.name].status == TaskStatus.PENDING and not self._is_unrunnable(t) and self._all_deps_met(t) ] @@ -437,7 +386,7 @@ def _run_parallel(self, max_workers: int) -> None: self.submitted_tasks.add(task.name) # If there are tasks pending, wait. As soon one is completed, - # save as passed or failed and remove from futures. + # save its result and remove from futures. if fut_to_name: done, _ = concurrent.futures.wait( fut_to_name.keys(), return_when="FIRST_COMPLETED" @@ -445,15 +394,9 @@ def _run_parallel(self, max_workers: int) -> None: for fut in done: name = fut_to_name.pop(fut) try: - res = fut.result() - if res.worked: - self.passed_results[name] = res - else: - self.failed_tasks.add(name) - self.failed_results[name] = res + self.results[name] = fut.result() except Exception as e: - self.failed_tasks.add(name) - self.failed_results[name] = TaskResult( + self.results[name] = TaskResult( TaskStatus.ERRORED, None, e, @@ -462,7 +405,7 @@ def _run_parallel(self, max_workers: int) -> None: else: # No running tasks and no new candidates. The # ``_is_unrunnable`` side effect above may have - # marked the remaining tasks as failed, completing + # marked the remaining tasks as skipped, completing # the DAG. Re-check the loop condition before # declaring a stall. if not self._is_done(): diff --git a/src/processes/task.py b/src/processes/task.py index 0403cce..df6af9f 100644 --- a/src/processes/task.py +++ b/src/processes/task.py @@ -405,7 +405,7 @@ def _resolve_args(self, executing_process: Process | None) -> tuple[list[Any], d final_kwargs = self.kwargs.copy() if executing_process is not None: for dep in self.dependencies: - dep_result = executing_process.runner.passed_results[dep.task_name].result + dep_result = executing_process.runner.results[dep.task_name].result if dep.use_result_as_additional_args: final_args.append(dep_result) if dep.use_result_as_additional_kwargs: diff --git a/tests/manual_tests/manual_pipeline_inspect.py b/tests/manual_tests/manual_pipeline_inspect.py index ebc4a21..9982771 100644 --- a/tests/manual_tests/manual_pipeline_inspect.py +++ b/tests/manual_tests/manual_pipeline_inspect.py @@ -406,10 +406,10 @@ def main() -> int: else: print("-" * 72) print("passed:") - for name in sorted(result.passed_tasks_results): + for name in sorted(result.successes): print(f" + {name}") print("failed (includes cascading-skipped):") - for name in sorted(result.failed_tasks): + for name in sorted(set(result.errored) | set(result.skipped)): print(f" - {name}") if exit_code == 0: diff --git a/tests/manual_tests/manual_themed_tracebacks.py b/tests/manual_tests/manual_themed_tracebacks.py index 7a2c9de..1455421 100644 --- a/tests/manual_tests/manual_themed_tracebacks.py +++ b/tests/manual_tests/manual_themed_tracebacks.py @@ -224,20 +224,21 @@ def _run_one_combo( return False, 2 print("passed:") - for name in sorted(result.passed_tasks_results): + for name in sorted(result.successes): print(f" + {name}") print("failed (includes cascading-skipped):") - for name in sorted(result.failed_tasks): + failed = set(result.errored) | set(result.skipped) + for name in sorted(failed): print(f" - {name}") # Strict post-conditions: downstream tasks were never invoked. ok = ( - "risky_step" in result.failed_tasks - and "child_a" in result.failed_tasks - and "child_b" in result.failed_tasks - and "risky_step" not in result.passed_tasks_results - and "child_a" not in result.passed_tasks_results - and "child_b" not in result.passed_tasks_results + "risky_step" in failed + and "child_a" in failed + and "child_b" in failed + and "risky_step" not in result.successes + and "child_a" not in result.successes + and "child_b" not in result.successes ) if not ok: print( diff --git a/tests/manual_tests/manual_webhook_inspect.py b/tests/manual_tests/manual_webhook_inspect.py index ac4cf3b..fbfe0c8 100644 --- a/tests/manual_tests/manual_webhook_inspect.py +++ b/tests/manual_tests/manual_webhook_inspect.py @@ -171,10 +171,10 @@ def main() -> int: print("-" * 72) print("passed:") - for name in sorted(result.passed_tasks_results): + for name in sorted(result.successes): print(f" + {name}") print("failed (includes cascading-skipped):") - for name in sorted(result.failed_tasks): + for name in sorted(set(result.errored) | set(result.skipped)): print(f" - {name}") return 0 diff --git a/tests/test_args_kwargs.py b/tests/test_args_kwargs.py index b53f7f1..8b101b6 100644 --- a/tests/test_args_kwargs.py +++ b/tests/test_args_kwargs.py @@ -72,12 +72,13 @@ def div(a: int, b: int) -> int: dependencies=[TaskDependency("task_1", use_result_as_additional_args=True)], ) with Process([t1, t2]) as process: - process_result = process.run() + report = process.run() - assert len(process_result.failed_tasks) == 0 - assert len(process_result.passed_tasks_results) == 2 - assert process_result.passed_tasks_results["task_1"].result == 2 - assert process_result.passed_tasks_results["task_2"].result == 5 + assert len(report.errored) == 0 + assert len(report.skipped) == 0 + assert len(report.successes) == 2 + assert report.successes["task_1"].result == 2 + assert report.successes["task_2"].result == 5 def test_add_extra_args_kwargs(self) -> None: """Test passing extra arguments and keyword arguments to a Task @@ -116,10 +117,11 @@ def div(a: int, b: int, c: int = 5) -> int: ], ) with Process([t0, t1, t2, t3]) as process: - process_result = process.run() - - assert len(process_result.failed_tasks) == 0 - assert len(process_result.passed_tasks_results) == 4 - assert process_result.passed_tasks_results["task_1"].result == 10 - assert process_result.passed_tasks_results["task_2"].result == 5 - assert process_result.passed_tasks_results["task_3"].result == 4 + report = process.run() + + assert len(report.errored) == 0 + assert len(report.skipped) == 0 + assert len(report.successes) == 4 + assert report.successes["task_1"].result == 10 + assert report.successes["task_2"].result == 5 + assert report.successes["task_3"].result == 4 diff --git a/tests/test_complex_dag_failures.py b/tests/test_complex_dag_failures.py index 0a75f32..630a6d0 100644 --- a/tests/test_complex_dag_failures.py +++ b/tests/test_complex_dag_failures.py @@ -138,24 +138,21 @@ def make_task(name: str, deps, fail: bool = False) -> Task: assert call_counts[name] == 1, ( f"Independent task {name} did not execute exactly once (got {call_counts[name]})" ) - assert name in result.passed_tasks_results, ( - f"Independent task {name} missing from passed_results" - ) - assert result.passed_tasks_results[name].worked is True - assert result.passed_tasks_results[name].result == f"payload_{name}" - assert result.passed_tasks_results[name].exception is None - - assert len(result.passed_tasks_results) == len(independent_task_names) - assert len(result.failed_tasks) == len(expected_failed) - assert result.errored_tasks == failing_task_names, ( - f"errored_tasks should be exactly the two tasks whose func raised, " - f"got {result.errored_tasks}" + assert name in result.successes, f"Independent task {name} missing from successes" + assert result.successes[name].result == f"payload_{name}" + assert result.successes[name].error is None + + assert len(result.successes) == len(independent_task_names) + assert len(result.errored) + len(result.skipped) == len(expected_failed) + assert set(result.errored) == failing_task_names, ( + f"errored entries should be exactly the two tasks whose func raised, " + f"got {set(result.errored)}" ) - assert result.skipped_tasks == skipped_task_names, ( - f"skipped_tasks should be the three cascade-skipped tasks, got {result.skipped_tasks}" + assert set(result.skipped) == skipped_task_names, ( + f"skipped entries should be the three cascade-skipped tasks, got {set(result.skipped)}" ) - assert result.errored_tasks.isdisjoint(result.skipped_tasks), ( - "errored_tasks and skipped_tasks must be disjoint" + assert set(result.errored).isdisjoint(result.skipped), ( + "errored and skipped entries must be disjoint" ) # OUTCOME #2 — Cascading Skip Control @@ -167,9 +164,9 @@ def make_task(name: str, deps, fail: bool = False) -> Task: assert call_counts[name] == 0, ( f"Skipped task {name} func was called {call_counts[name]} times — must be 0" ) - assert result.failed_tasks == expected_failed + assert set(result.errored) | set(result.skipped) == expected_failed for name in skipped_task_names: - assert name not in result.passed_tasks_results + assert name not in result.successes # OUTCOME #3 — Data-Driven Logger Payload for name in failing_task_names: diff --git a/tests/test_execution_report.py b/tests/test_execution_report.py index 465a215..a36c99c 100644 --- a/tests/test_execution_report.py +++ b/tests/test_execution_report.py @@ -1,4 +1,4 @@ -"""Tests for ``ProcessExecutionReport.from_result``. +"""Tests for ``ProcessExecutionReport`` as returned by ``Process.run()``. Covers status classification (success / errored / skipped), the data carried by each ``TaskReportEntry`` (result, error, elapsed_seconds, attempts), entry @@ -11,7 +11,6 @@ from processes import ( ErrorData, Process, - ProcessExecutionReport, Task, TaskDependency, TaskStatus, @@ -51,8 +50,7 @@ def notify() -> None: def test_sequential_report_classifies_each_task(self) -> None: process, load_task, apply_task, notify_task = self._build_process() with process: - result = process.run(parallel=False) - report = ProcessExecutionReport.from_result(process, result) + report = process.run(parallel=False) assert list(report.entries) == ["load", "apply", "notify"] @@ -86,8 +84,7 @@ def test_sequential_report_classifies_each_task(self) -> None: def test_filter_accessors_partition_entries(self) -> None: process, load_task, apply_task, notify_task = self._build_process() with process: - result = process.run(parallel=False) - report = ProcessExecutionReport.from_result(process, result) + report = process.run(parallel=False) assert set(report.successes) == {"load"} assert set(report.errored) == {"apply"} @@ -98,8 +95,7 @@ def test_filter_accessors_partition_entries(self) -> None: def test_parallel_report_classifies_each_task(self) -> None: process, load_task, apply_task, notify_task = self._build_process() with process: - result = process.run(parallel=True, max_workers=4) - report = ProcessExecutionReport.from_result(process, result) + report = process.run(parallel=True, max_workers=4) assert report.entries["load"].status == TaskStatus.SUCCESS assert report.entries["apply"].status == TaskStatus.ERRORED @@ -114,8 +110,7 @@ def step() -> int: task = Task("step", step, self._log("report_all_success.log")) process = Process([task]) with process: - result = process.run(parallel=False) - report = ProcessExecutionReport.from_result(process, result) + report = process.run(parallel=False) assert report.entries["step"].status == TaskStatus.SUCCESS assert report.entries["step"].result == 42 diff --git a/tests/test_normal_run_no_errors.py b/tests/test_normal_run_no_errors.py index 7fd3e58..95fe983 100644 --- a/tests/test_normal_run_no_errors.py +++ b/tests/test_normal_run_no_errors.py @@ -103,14 +103,12 @@ def _build_dependent_task_graph(self) -> list[Task]: def test_run_dependent_tasks_sequential(self) -> None: with Process(self._build_dependent_task_graph()) as process: t_start = time.time() - process_result = process.run(parallel=False) + report = process.run(parallel=False) t_end = time.time() - assert len(process_result.passed_tasks_results) == 6, ( - f"Expected 6 passed tasks. Got {len(process_result.passed_tasks_results)}" - ) - assert len(process_result.failed_tasks) == 0, ( - f"Expected 0 failed tasks. Got {len(process_result.failed_tasks)}" + assert len(report.successes) == 6, f"Expected 6 passed tasks. Got {len(report.successes)}" + assert len(report.errored) + len(report.skipped) == 0, ( + f"Expected 0 failed tasks. Got {len(report.errored) + len(report.skipped)}" ) elapsed = t_end - t_start assert 4.0 <= elapsed < 4.8, f"Sequential run took {elapsed} seconds. Expected ~4 seconds." @@ -121,14 +119,12 @@ def test_run_dependent_tasks_parallel(self) -> None: n_workers = os.cpu_count() with Process(self._build_dependent_task_graph()) as process: t_start = time.time() - process_result = process.run(parallel=True, max_workers=n_workers) + report = process.run(parallel=True, max_workers=n_workers) t_end = time.time() - assert len(process_result.passed_tasks_results) == 6, ( - f"Expected 6 passed tasks. Got {len(process_result.passed_tasks_results)}" - ) - assert len(process_result.failed_tasks) == 0, ( - f"Expected 0 failed tasks. Got {len(process_result.failed_tasks)}" + assert len(report.successes) == 6, f"Expected 6 passed tasks. Got {len(report.successes)}" + assert len(report.errored) + len(report.skipped) == 0, ( + f"Expected 0 failed tasks. Got {len(report.errored) + len(report.skipped)}" ) if n_workers and n_workers > 2: assert int(round(t_end - t_start, 0)) == 2, ( diff --git a/tests/test_parallel_race_conditions.py b/tests/test_parallel_race_conditions.py index 5e1e7be..dbfde53 100644 --- a/tests/test_parallel_race_conditions.py +++ b/tests/test_parallel_race_conditions.py @@ -100,10 +100,11 @@ def collector(*sibling_results): f"{tag} collector received {call_counts['__collector_arg_count']} " f"args, expected {sibling_count}" ) - assert not result.failed_tasks, f"{tag} unexpected failures: {result.failed_tasks}" - assert len(result.passed_tasks_results) == sibling_count + 2, f"{tag} passed count mismatch" + failed = set(result.errored) | set(result.skipped) + assert not failed, f"{tag} unexpected failures: {failed}" + assert len(result.successes) == sibling_count + 2, f"{tag} passed count mismatch" - stored = result.passed_tasks_results["collector"].result + stored = result.successes["collector"].result assert stored == sorted(f"sibling_{i}_payload" for i in range(sibling_count)), ( f"{tag} collector result missing sibling payloads: {stored}" ) @@ -201,23 +202,23 @@ def _child(*_args, **_kwargs): expected_failed = {n for chain in failing_chains for n in chain} expected_passed = {n for chain in passing_chains for n in chain} - assert result.failed_tasks == expected_failed, ( + failed = set(result.errored) | set(result.skipped) + passed = set(result.successes) + + assert failed == expected_failed, ( f"{tag} failed set mismatch: " - f"missing={expected_failed - result.failed_tasks}, " - f"extra={result.failed_tasks - expected_failed}" + f"missing={expected_failed - failed}, " + f"extra={failed - expected_failed}" ) - assert set(result.passed_tasks_results) == expected_passed, ( + assert passed == expected_passed, ( f"{tag} passed set mismatch: " - f"missing={expected_passed - set(result.passed_tasks_results)}, " - f"extra={set(result.passed_tasks_results) - expected_passed}" - ) - assert not (result.failed_tasks & set(result.passed_tasks_results)), ( - f"{tag} a task appears in both passed and failed sets" + f"missing={expected_passed - passed}, " + f"extra={passed - expected_passed}" ) - assert len(result.failed_tasks) + len(result.passed_tasks_results) == len(tasks), ( + assert not (failed & passed), f"{tag} a task appears in both passed and failed sets" + assert len(failed) + len(passed) == len(tasks), ( f"{tag} task accounting drifted: " - f"{len(result.failed_tasks)} failed + " - f"{len(result.passed_tasks_results)} passed != {len(tasks)} total" + f"{len(failed)} failed + {len(passed)} passed != {len(tasks)} total" ) diff --git a/tests/test_timeout_retry.py b/tests/test_timeout_retry.py index a08bb35..7713714 100644 --- a/tests/test_timeout_retry.py +++ b/tests/test_timeout_retry.py @@ -56,13 +56,10 @@ def slow() -> str: task = Task("t_seq", slow, self._log("seq.log"), timeout=0.05) with Process([task]) as process: - pr = process.run(parallel=False) + report = process.run(parallel=False) - assert "t_seq" in pr.failed_tasks - assert "t_seq" in pr.errored_tasks - assert isinstance( - pr.passed_tasks_results.get("t_seq", None) or pr.failed_tasks, set - ) # task did not pass + assert "t_seq" in report.errored + assert "t_seq" not in report.successes def test_timeout_in_parallel_process(self) -> None: """Timeout propagates through Process.run(parallel=True).""" @@ -73,10 +70,10 @@ def slow() -> str: task = Task("t_par", slow, self._log("par.log"), timeout=0.05) with Process([task]) as process: - pr = process.run(parallel=True) + report = process.run(parallel=True) - assert "t_par" in pr.failed_tasks - assert "t_par" in pr.errored_tasks + assert "t_par" in report.errored + assert "t_par" not in report.successes def test_timeout_cascades_to_dependants(self) -> None: """A timed-out task causes its dependants to be cascade-skipped.""" @@ -96,11 +93,11 @@ def child(x: str) -> str: dependencies=[TaskDependency("root", use_result_as_additional_args=True)], ) with Process([t_root, t_child]) as process: - pr = process.run(parallel=False) + report = process.run(parallel=False) - assert "root" in pr.errored_tasks - assert "child" in pr.skipped_tasks - assert "child" not in pr.passed_tasks_results + assert "root" in report.errored + assert "child" in report.skipped + assert "child" not in report.successes # --- validation --- @@ -267,10 +264,10 @@ def flaky() -> str: task = Task("proc_retry", flaky, self._log("pr.log"), retries=1) with Process([task]) as process: - pr = process.run(parallel=False) + report = process.run(parallel=False) - assert "proc_retry" in pr.passed_tasks_results - assert pr.passed_tasks_results["proc_retry"].result == "ok" + assert "proc_retry" in report.successes + assert report.successes["proc_retry"].result == "ok" assert len(calls) == 2 # --- validation --- From 3dc988dfd373cee8bf734cdce90c5cd4fe46188d Mon Sep 17 00:00:00 2001 From: Oliver Mohr Date: Sun, 14 Jun 2026 23:06:47 -0400 Subject: [PATCH 08/10] docs: update README and reference docs for ProcessExecutionReport API Remove stale references to ProcessResult, passed_tasks_results, failed_tasks, errored_tasks, and skipped_tasks now that Process.run() returns a ProcessExecutionReport backed by a single results dict. --- README.md | 38 +++++++++++--------------- docs/reference.md | 1 - tests/test_parallel_race_conditions.py | 2 +- 3 files changed, 17 insertions(+), 24 deletions(-) diff --git a/README.md b/README.md index 3c1a76b..4916819 100644 --- a/README.md +++ b/README.md @@ -37,7 +37,7 @@ A `Process` holds a list of `Task`s. At construction it validates names, types, When you call `process.run()`, tasks are topologically sorted and scheduled: dependencies first, independent tasks in parallel. -A `TaskDependency` can forward an upstream result directly into a downstream function, as a positional or keyword argument. The result is a `ProcessResult` with `passed_tasks_results` and `failed_tasks` for inspection. +A `TaskDependency` can forward an upstream result directly into a downstream function, as a positional or keyword argument. The result is a `ProcessExecutionReport` with `successes`, `errored`, and `skipped` for inspection. --- @@ -70,7 +70,7 @@ tasks = [ with Process(tasks) as p: result = p.run(parallel=True) -print(result.passed_tasks_results["enrich"].result) +print(result.successes["enrich"].result) # [{'id': 1, 'name': 'user-1'}, {'id': 2, 'name': 'user-2'}, {'id': 3, 'name': 'user-3'}] ``` @@ -201,11 +201,11 @@ tasks = [ with Process(tasks) as process: result = process.run(parallel=True) -print("passed:", sorted(result.passed_tasks_results)) +print("passed:", sorted(result.successes)) # archive_report, build_report, compute_revenue, compute_stock, fetch_inventory, fetch_orders -print("failed:", sorted(result.failed_tasks)) +print("failed:", sorted(set(result.errored) | set(result.skipped))) # notify_slack -print("report:", result.passed_tasks_results["build_report"].result) +print("report:", result.successes["build_report"].result) # daily-report | revenue=59.50 stock=262.50 ``` @@ -265,24 +265,18 @@ TaskDependency( ```python Process(tasks: list[Task]) # validates types, names, deps, cycles -process.run(parallel: bool | None = None, max_workers: int = 4) -> ProcessResult +process.run(parallel: bool | None = None, max_workers: int = 4) -> ProcessExecutionReport ``` - Raises `DependencyNotFoundError`, `CircularDependencyError`, `TypeError`, `ValueError` on construction if the workflow is malformed. - `parallel=None` auto-parallelises when `len(tasks) >= 10`; `max_workers=1` is always sequential. - Use as a context manager — it cleans up `FileHandler`s on exit. -### `ProcessResult` +### `TaskResult` ```python -result.passed_tasks_results # dict[str, TaskResult] — name → TaskResult for every task that succeeded -result.failed_tasks # set[str] — all tasks that did not produce a result (errored + skipped) -result.failed_tasks_results # dict[str, TaskResult] — name → TaskResult for every task that errored -result.errored_tasks # set[str] — tasks whose function actually raised -result.skipped_tasks # set[str] — tasks skipped because an upstream dependency failed - TaskResult( - worked: bool, + status: TaskStatus, result: Any, exception: Exception | None, error_data: ErrorData | None = None, @@ -291,11 +285,12 @@ TaskResult( ) ``` +- `status` — `TaskStatus.PENDING | SUCCESS | ERRORED | SKIPPED`. +- `worked` — `True` if `status == TaskStatus.SUCCESS`. + ### `ProcessExecutionReport` ```python -ProcessExecutionReport.from_result(process: Process, result: ProcessResult) -> ProcessExecutionReport - report.entries # dict[str, TaskReportEntry] — one entry per task, in topological order report.successes # dict[str, TaskReportEntry] — entries with status == TaskStatus.SUCCESS report.errored # dict[str, TaskReportEntry] — entries with status == TaskStatus.ERRORED @@ -309,9 +304,8 @@ plus `result` (set when `SUCCESS`) and `error: ErrorData | None` (set when ```python with Process([load_task, apply_task, notify_task]) as process: - result = process.run() + report = process.run() -report = ProcessExecutionReport.from_result(process, result) for name, entry in report.entries.items(): if entry.status is TaskStatus.ERRORED: print(f"{name} failed after {entry.attempts} attempt(s): {entry.error.exception}") @@ -442,10 +436,10 @@ body, so receivers can verify the payload wasn't tampered with. When a task raises: -1. The exception is caught and stored in `TaskResult.exception`; the task name goes into `failed_tasks` and `errored_tasks`. -2. **Every task that depends on it (directly or indirectly) is skipped** — added to `failed_tasks` and `skipped_tasks` without running. +1. The exception is caught and stored in `TaskResult.exception`; the task's entry in the report gets `status == TaskStatus.ERRORED`. +2. **Every task that depends on it (directly or indirectly) is skipped** — its entry gets `status == TaskStatus.SKIPPED` without running. 3. **Every other independent part of the workflow keeps running.** With `parallel=True` they keep running concurrently on the worker pool. -4. After `run()` returns, `ProcessResult.errored_tasks` and `ProcessResult.skipped_tasks` let you distinguish root failures from cascade skips for triage or alerting. +4. After `run()` returns, `ProcessExecutionReport.errored` and `ProcessExecutionReport.skipped` let you distinguish root failures from cascade skips for triage or alerting. When a task has `retries >= 1`, a failure matching `retry_on` triggers another attempt before the task is declared failed and its dependants are skipped. This gives transient errors (network blips, connection resets) a chance to resolve without aborting downstream work. @@ -475,7 +469,7 @@ This makes the library a good fit for fan-out / fan-in pipelines, "best-effort" - **Shared log file** — pass the same `log_path` to every `Task` for a single combined run.log; pass distinct paths for per-task isolation. - **Auto-parallel** — `Process.run()` with no argument runs sequentially for small workflows and switches to parallel for `len(tasks) >= 10`. Pass `parallel=True` or `parallel=False` to force the mode. -- **Result inspection** — iterate `result.passed_tasks_results.items()` to log or post-process every successful task; iterate `result.failed_tasks` for triage. +- **Result inspection** — iterate `report.successes.items()` to log or post-process every successful task; iterate `set(report.errored) | set(report.skipped)` for triage. - **Re-raising** — wrap `process.run()` in `try/except` if you need a non-zero exit code on any failure; the library itself does not raise on partial failure. diff --git a/docs/reference.md b/docs/reference.md index cf31beb..1ed66a3 100644 --- a/docs/reference.md +++ b/docs/reference.md @@ -3,7 +3,6 @@ This page is automatically generated from the source code docstrings. ::: processes.process.Process -::: processes.process.ProcessResult ::: processes.task.Task ::: processes.task.TaskDependency ::: processes.task.TaskResult diff --git a/tests/test_parallel_race_conditions.py b/tests/test_parallel_race_conditions.py index dbfde53..a4f25a1 100644 --- a/tests/test_parallel_race_conditions.py +++ b/tests/test_parallel_race_conditions.py @@ -2,7 +2,7 @@ Parallel-execution race-condition stress tests. Two scenarios designed to maximise contention on ``ProcessRunner``'s -shared state (``passed_results``, ``failed_tasks``, ``submitted_tasks``): +shared state (``results``, ``submitted_tasks``): 1. ``test_diamond_fan_in_race`` — wide fan-in diamond. Many sibling tasks gate on a ``threading.Barrier`` and release simultaneously, From 51f41928e0317054a49b2ab954ac3e2b80a849dd Mon Sep 17 00:00:00 2001 From: Oliver Mohr Date: Mon, 15 Jun 2026 19:32:37 -0400 Subject: [PATCH 09/10] fix: make Task.run never propagate dependency-resolution failures _resolve_args ran outside the retry try/except, so a failure injecting an upstream result propagated out of Task.run. Sequential execution crashed the whole run while parallel execution swallowed it -- an asymmetry. Move arg resolution inside the error handling so it always returns an ERRORED result (attempts=0), and factor the failure-logging path into _errored_result. Adds TaskResult named constructors (pending/skipped/success/errored) used by the new error path, plus tests pinning the no-propagation guarantee. --- src/processes/task.py | 87 ++++++++++++++++++++++++------ tests/test_resolve_args_failure.py | 52 ++++++++++++++++++ 2 files changed, 123 insertions(+), 16 deletions(-) create mode 100644 tests/test_resolve_args_failure.py diff --git a/src/processes/task.py b/src/processes/task.py index df6af9f..6abe8bd 100644 --- a/src/processes/task.py +++ b/src/processes/task.py @@ -89,6 +89,46 @@ def worked(self) -> bool: """True if the task executed successfully, False otherwise.""" return self.status == TaskStatus.SUCCESS + @classmethod + def pending(cls) -> TaskResult: + """Build the placeholder result for a task that has not run yet.""" + return cls(TaskStatus.PENDING, None, None) + + @classmethod + def skipped(cls) -> TaskResult: + """Build the result for a task skipped after an upstream dependency failed.""" + return cls(TaskStatus.SKIPPED, None, None) + + @classmethod + def success(cls, result: Any, *, elapsed_seconds: float = 0.0, attempts: int = 0) -> TaskResult: + """Build the result for a task whose function returned without raising.""" + return cls( + TaskStatus.SUCCESS, + result, + None, + elapsed_seconds=elapsed_seconds, + attempts=attempts, + ) + + @classmethod + def errored( + cls, + exception: Exception, + *, + error_data: ErrorData | None = None, + elapsed_seconds: float = 0.0, + attempts: int = 0, + ) -> TaskResult: + """Build the result for a task whose function raised after exhausting retries.""" + return cls( + TaskStatus.ERRORED, + None, + exception, + error_data=error_data, + elapsed_seconds=elapsed_seconds, + attempts=attempts, + ) + class TaskDependency: """ @@ -434,9 +474,30 @@ def _build_failure_context( "traced_vars_location": _build_traced_vars_location(exc_tb, self._frame_filter), } + def _errored_result( + self, + exc: Exception, + executing_process: Process | None, + elapsed_seconds: float, + attempts: int, + ) -> TaskResult: + """Log ``exc`` through this task's handlers and wrap it in a TaskResult.""" + task_context = self._build_failure_context(exc, executing_process) + self.logger.error(str(exc), exc_info=exc, extra={"task_context": task_context}) + return TaskResult.errored( + exc, + error_data=ErrorData(**task_context), + elapsed_seconds=elapsed_seconds, + attempts=attempts, + ) + def run(self, executing_process: Process | None = None) -> TaskResult: """Execute the task, retrying on transient failures up to ``retries`` times. + Never propagates: a failure while resolving dependency arguments (before + any attempt runs) is captured and returned as an ``ERRORED`` result with + ``attempts=0``, just like a failure inside the task's function. + Parameters ---------- executing_process : Process, optional @@ -449,24 +510,26 @@ def run(self, executing_process: Process | None = None) -> TaskResult: ``worked=True`` with the return value on success; ``worked=False`` with the last exception on failure. """ - final_args, final_kwargs = self._resolve_args(executing_process) - max_attempts = self.retries + 1 effective_retry_on = ( self.retry_on if self.retry_on is not None else (ConnectionError, TimeoutError) ) self.logger.info(f"Starting {self.name}.") - last_exc: Exception | None = None start = time.monotonic() + try: + final_args, final_kwargs = self._resolve_args(executing_process) + except Exception as e: + return self._errored_result(e, executing_process, time.monotonic() - start, attempts=0) + + last_exc: Exception | None = None for attempt in range(1, max_attempts + 1): try: result = self._call_with_timeout(final_args, final_kwargs) self.logger.info(f"Finished {self.name}.") - elapsed = time.monotonic() - start - return TaskResult( - TaskStatus.SUCCESS, result, None, elapsed_seconds=elapsed, attempts=attempt + return TaskResult.success( + result, elapsed_seconds=time.monotonic() - start, attempts=attempt ) except Exception as e: last_exc = e @@ -479,14 +542,6 @@ def run(self, executing_process: Process | None = None) -> TaskResult: break assert last_exc is not None - elapsed = time.monotonic() - start - task_context = self._build_failure_context(last_exc, executing_process) - self.logger.error(str(last_exc), exc_info=last_exc, extra={"task_context": task_context}) - return TaskResult( - TaskStatus.ERRORED, - None, - last_exc, - error_data=ErrorData(**task_context), - elapsed_seconds=elapsed, - attempts=attempt, + return self._errored_result( + last_exc, executing_process, time.monotonic() - start, attempts=attempt ) diff --git a/tests/test_resolve_args_failure.py b/tests/test_resolve_args_failure.py new file mode 100644 index 0000000..9b9b0e9 --- /dev/null +++ b/tests/test_resolve_args_failure.py @@ -0,0 +1,52 @@ +"""``Task.run`` must never propagate. + +A failure while resolving dependency arguments (before any attempt runs) has to +become an ``ERRORED`` result, just like a failure inside the task's function. +This previously crashed sequential execution while parallel execution swallowed +it, an asymmetry these tests pin down. +""" + +from __future__ import annotations + +from typing import Any + +from processes import Process, Task, TaskDependency, TaskStatus + +from .base_test import BaseTest + + +def _boom(_executing_process: Any) -> Any: + raise RuntimeError("resolve blew up") + + +class TestResolveArgsFailure(BaseTest): + def test_run_wraps_resolution_failure(self) -> None: + """A failure in ``_resolve_args`` yields ``ERRORED`` (attempts=0), not a raise.""" + task = Task("t", lambda: 1, self._log("resolve.log")) + task._resolve_args = _boom # type: ignore[method-assign] + try: + result = task.run(executing_process=None) + finally: + self._close_handlers(task) + + assert not result.worked + assert result.status == TaskStatus.ERRORED + assert isinstance(result.exception, RuntimeError) + assert result.attempts == 0 + + def test_sequential_process_survives_resolution_failure(self) -> None: + """Sequential ``Process.run`` does not crash when a task's args fail to resolve.""" + producer = Task("producer", lambda: 1, self._log("producer.log")) + consumer = Task( + "consumer", + lambda x: x, + self._log("consumer.log"), + dependencies=[TaskDependency("producer", use_result_as_additional_args=True)], + ) + consumer._resolve_args = _boom # type: ignore[method-assign] + + with Process([producer, consumer]) as process: + report = process.run(parallel=False) + + assert report.entries["producer"].status == TaskStatus.SUCCESS + assert report.entries["consumer"].status == TaskStatus.ERRORED From 99704d73cbfede1b9ef8ba016e020cf80daec31d Mon Sep 17 00:00:00 2001 From: Oliver Mohr Date: Mon, 15 Jun 2026 19:33:09 -0400 Subject: [PATCH 10/10] refactor: drop _is_unrunnable side effect and reuse TaskResult helpers Split the cascade-skip query from its mutation: _has_failed_dep is now a pure predicate and callers record SKIPPED explicitly. Adopt the TaskResult named constructors and the existing `worked` property instead of re-deriving `status == SUCCESS` by hand. --- src/processes/execution_report.py | 2 +- src/processes/process.py | 45 ++++++++++++++++--------------- 2 files changed, 25 insertions(+), 22 deletions(-) diff --git a/src/processes/execution_report.py b/src/processes/execution_report.py index 2997580..06b1a95 100644 --- a/src/processes/execution_report.py +++ b/src/processes/execution_report.py @@ -110,7 +110,7 @@ def from_results( status=res.status, elapsed_seconds=res.elapsed_seconds, attempts=res.attempts, - result=res.result if res.status == TaskStatus.SUCCESS else None, + result=res.result if res.worked else None, error=res.error_data if res.status == TaskStatus.ERRORED else None, ) return cls(entries) diff --git a/src/processes/process.py b/src/processes/process.py index 2afd3d6..2237b74 100644 --- a/src/processes/process.py +++ b/src/processes/process.py @@ -270,7 +270,7 @@ class ProcessRunner: def __init__(self, process_ref: Process): self.process = process_ref self.results: dict[str, TaskResult] = { - task.name: TaskResult(TaskStatus.PENDING, None, None) for task in process_ref.tasks + task.name: TaskResult.pending() for task in process_ref.tasks } self.submitted_tasks: set[str] = set() @@ -305,8 +305,11 @@ def run(self, parallel: bool, max_workers: int) -> ProcessExecutionReport: self._run_sequential() return ProcessExecutionReport.from_results(self.process, self.results) - def _is_unrunnable(self, task: Task) -> bool: - """Check if a task cannot be run due to failed or skipped dependencies. + def _has_failed_dep(self, task: Task) -> bool: + """Return whether any dependency errored or was skipped. + + Pure query with no side effects: it does not record the task as + ``SKIPPED``. Callers decide when to mark the cascade-skip. Parameters ---------- @@ -317,16 +320,12 @@ def _is_unrunnable(self, task: Task) -> bool: ------- bool True if any of the task's dependencies errored or were skipped, - False otherwise. If True, the task is recorded as ``SKIPPED`` - (cascade-skip). + False otherwise. """ - if any( + return any( self.results[d.task_name].status in (TaskStatus.ERRORED, TaskStatus.SKIPPED) for d in task.dependencies - ): - self.results[task.name] = TaskResult(TaskStatus.SKIPPED, None, None) - return True - return False + ) def _all_deps_met(self, task: Task) -> bool: """Check if all dependencies of a task have been successfully executed. @@ -341,14 +340,13 @@ def _all_deps_met(self, task: Task) -> bool: bool True if all dependencies succeeded, False otherwise. """ - return all( - self.results[d.task_name].status == TaskStatus.SUCCESS for d in task.dependencies - ) + return all(self.results[d.task_name].worked for d in task.dependencies) def _run_sequential(self) -> None: """Execute all tasks sequentially in dependency order.""" for task in self.process.tasks: - if self._is_unrunnable(task): + if self._has_failed_dep(task): + self.results[task.name] = TaskResult.skipped() continue if self._all_deps_met(task): self.results[task.name] = task.run(self.process) @@ -369,13 +367,21 @@ def _run_parallel(self, max_workers: int) -> None: with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor: fut_to_name = {} while not self._is_done(): - # Look for candidates to execute now + # Record cascade-skips first (a dependency errored or was + # skipped), then look for candidates whose deps all succeeded. + for t in self.process.tasks: + if ( + t.name not in self.submitted_tasks + and self.results[t.name].status == TaskStatus.PENDING + and self._has_failed_dep(t) + ): + self.results[t.name] = TaskResult.skipped() + candidates = [ t for t in self.process.tasks if t.name not in self.submitted_tasks and self.results[t.name].status == TaskStatus.PENDING - and not self._is_unrunnable(t) and self._all_deps_met(t) ] @@ -396,11 +402,8 @@ def _run_parallel(self, max_workers: int) -> None: try: self.results[name] = fut.result() except Exception as e: - self.results[name] = TaskResult( - TaskStatus.ERRORED, - None, - e, - error_data=ErrorData(task_name=name, exception=str(e)), + self.results[name] = TaskResult.errored( + e, error_data=ErrorData(task_name=name, exception=str(e)) ) else: # No running tasks and no new candidates. The