diff --git a/src/wf_core/models/input_bindings.py b/src/wf_core/models/input_bindings.py index 744b5fbe..85749e77 100644 --- a/src/wf_core/models/input_bindings.py +++ b/src/wf_core/models/input_bindings.py @@ -1,6 +1,6 @@ from __future__ import annotations -from collections.abc import Mapping +from collections.abc import Iterator, Mapping from typing import Annotated, Literal, TypeAliasType from pydantic import BaseModel, ConfigDict, Field, field_validator @@ -103,6 +103,22 @@ type InputExpression = Annotated[ ] +def walk_expression_paths( + expression: InputExpression, +) -> Iterator[tuple[GraphSourcePath, str]]: + """Yield graph path leaves with their stable expression locations.""" + if isinstance(expression, PathExpression): + yield expression.path, "path" + elif isinstance(expression, ArrayExpression): + for index, item in enumerate(expression.items): + for path, suffix in walk_expression_paths(item): + yield path, f"items[{index}].{suffix}" + elif isinstance(expression, ObjectExpression): + for field, item in expression.fields.items(): + for path, suffix in walk_expression_paths(item): + yield path, f"fields.{field}.{suffix}" + + class OccurrenceExpression(BaseModel): """Reference one typed schedule-occurrence field. diff --git a/src/wf_core/runtime/input_sources.py b/src/wf_core/runtime/input_sources.py index cdde45cf..7f9b80c3 100644 --- a/src/wf_core/runtime/input_sources.py +++ b/src/wf_core/runtime/input_sources.py @@ -16,7 +16,7 @@ runtime values); budget parity there comes from the same model-level from __future__ import annotations -from collections.abc import Iterator, Mapping, Sequence +from collections.abc import Mapping, Sequence from typing import Any, Protocol from wf_core.errors import WorkflowExecutionError @@ -227,23 +227,3 @@ def resolve_schedule_input_bindings( raise WorkflowExecutionError(f"{label} {location}: {exc}") from exc return payload - -def walk_expression_paths( - expression: InputExpression, -) -> Iterator[tuple[GraphSourcePath, str]]: - """Yield ``(graph path, location)`` leaves in one shared traversal order. - - Locations use the same ``items[i]`` / ``fields.name`` / ``path`` suffix - shape as :func:`wf_core.validation.steps._validate_expression_sources`, - so validation and runtime traverse leaves in the same order. - """ - if isinstance(expression, PathExpression): - yield expression.path, "path" - elif isinstance(expression, ArrayExpression): - for index, item in enumerate(expression.items): - for path, suffix in walk_expression_paths(item): - yield path, f"items[{index}].{suffix}" - elif isinstance(expression, ObjectExpression): - for field, item in expression.fields.items(): - for path, suffix in walk_expression_paths(item): - yield path, f"fields.{field}.{suffix}" diff --git a/src/wf_core/validation/steps.py b/src/wf_core/validation/steps.py index a76003ea..f86b1737 100644 --- a/src/wf_core/validation/steps.py +++ b/src/wf_core/validation/steps.py @@ -10,7 +10,11 @@ from wf_core.models.conditions import ( PathOperand, VariadicCondition, ) -from wf_core.models.input_bindings import InputExpression, InputExpressionBinding +from wf_core.models.input_bindings import ( + InputExpression, + InputExpressionBinding, + walk_expression_paths, +) from wf_core.models.schemas import NodeDef from wf_core.models.steps import ( ConditionNode, @@ -33,7 +37,6 @@ from wf_core.paths import ( is_valid_destination_path, is_valid_source_path, ) -from wf_core.runtime.input_sources import walk_expression_paths from wf_core.validation.issues import ValidationIssueCode, ValidationReport @@ -247,7 +250,7 @@ def _validate_expression_sources( """Validate every graph path leaf while keeping one top-level target atomic. Leaf traversal order is the canonical order defined by - :func:`wf_core.runtime.input_sources.walk_expression_paths`. + :func:`wf_core.models.input_bindings.walk_expression_paths`. """ for path, suffix in walk_expression_paths(expression): _validate_source_path( diff --git a/src/wf_scheduling/dispatch.py b/src/wf_scheduling/dispatch.py index 69f488f2..fb7425e3 100644 --- a/src/wf_scheduling/dispatch.py +++ b/src/wf_scheduling/dispatch.py @@ -14,7 +14,7 @@ from __future__ import annotations from dataclasses import dataclass from datetime import datetime -from typing import Protocol, Union +from typing import Protocol from wf_artifacts.runs.models import RunAdmission from wf_core import RunState @@ -32,7 +32,7 @@ class StillRunning: """Dispatched with an unknown outcome; the run stays admitted.""" -DispatchResult = Union[Stopped, StillRunning] +DispatchResult = Stopped | StillRunning class RunDispatcher(Protocol): diff --git a/src/wf_scheduling/poll.py b/src/wf_scheduling/poll.py index d7100069..109dff27 100644 --- a/src/wf_scheduling/poll.py +++ b/src/wf_scheduling/poll.py @@ -17,7 +17,7 @@ contains no fixture input, fixture environments, or canned results. from __future__ import annotations from datetime import datetime, timezone -from typing import Any +from typing import Any, cast from wf_scheduling.calendar import ( CronSource, @@ -31,7 +31,7 @@ from wf_scheduling.history import ( HistoryEntry, HistoryRecorder, ) -from wf_scheduling.models import PendingCandidate +from wf_scheduling.models import OccurrenceKind, PendingCandidate from wf_scheduling.ownership import ( SchedulerOwnership, SecondOwnerError, @@ -258,7 +258,7 @@ class Scheduler: def _record( self, *, - kind: str, + kind: OccurrenceKind, sched_id: str, intended: datetime | None = None, run_id: str | None = None, @@ -275,7 +275,7 @@ class Scheduler: self.history.record( HistoryEntry( schedule_id=sched_id, - kind=kind, # type: ignore[arg-type] + kind=kind, resolved_at=intended, run_id=run_id, revision=revision, @@ -504,13 +504,16 @@ class Scheduler: run_id=run_id, ) self.run_store.clear_executing(run_id) - kind = { - StoredRunStatus.COMPLETED: "completed", - StoredRunStatus.INTERRUPTED: "interrupted", - StoredRunStatus.FAILED: "failed", - }[stopped.status] + kind = cast( + OccurrenceKind, + { + StoredRunStatus.COMPLETED: "completed", + StoredRunStatus.INTERRUPTED: "interrupted", + StoredRunStatus.FAILED: "failed", + }[stopped.status], + ) self._record( - kind=kind, # type: ignore[arg-type] + kind=kind, sched_id=admission.schedule_id, intended=_admission_intended(self.run_store, run_id), run_id=run_id, @@ -563,13 +566,16 @@ class Scheduler: run_id=run_id, ) self.run_store.clear_executing(run_id) - kind = { - StoredRunStatus.COMPLETED: "completed", - StoredRunStatus.INTERRUPTED: "interrupted", - StoredRunStatus.FAILED: "failed", - }[stopped.status] + kind = cast( + OccurrenceKind, + { + StoredRunStatus.COMPLETED: "completed", + StoredRunStatus.INTERRUPTED: "interrupted", + StoredRunStatus.FAILED: "failed", + }[stopped.status], + ) self._record( - kind=kind, # type: ignore[arg-type] + kind=kind, sched_id=admission.schedule_id, intended=_admission_intended(self.run_store, run_id), run_id=run_id, @@ -597,11 +603,14 @@ class Scheduler: history. Returns whether an entry was appended. """ self._require_ownership() - kind = { - "completed": "completed", - "interrupted": "interrupted", - "failed": "failed", - }.get(status_value) + kind = cast( + OccurrenceKind | None, + { + "completed": "completed", + "interrupted": "interrupted", + "failed": "failed", + }.get(status_value), + ) if kind is None: return False try: @@ -614,7 +623,7 @@ class Scheduler: if self.history.has_terminal(sched_id, run_id, kind, checkpoint_id): return False self._record( - kind=kind, # type: ignore[arg-type] + kind=kind, sched_id=sched_id, intended=admission.scheduled_at, run_id=run_id, diff --git a/src/wf_scheduling/recovery.py b/src/wf_scheduling/recovery.py index 7eb1a489..791b5a7d 100644 --- a/src/wf_scheduling/recovery.py +++ b/src/wf_scheduling/recovery.py @@ -62,6 +62,7 @@ from wf_scheduling.history import ( HistoryEntry, HistoryRecorder, ) +from wf_scheduling.models import OccurrenceKind from wf_scheduling.ownership import ( SchedulerOwnership, SecondOwnerError, @@ -676,7 +677,7 @@ def _reconcile_terminal( history: HistoryRecorder, sched_id: str | None, run_id: str, - kind: str, + kind: OccurrenceKind, checkpoint_id: str | None, intended: datetime | None, revision: int | None, @@ -697,7 +698,7 @@ def _reconcile_terminal( history.record( HistoryEntry( schedule_id=sched_id, - kind=kind, # type: ignore[arg-type] + kind=kind, resolved_at=intended, run_id=run_id, revision=revision,