"""Workflow-resumption preconditions and context assembly.
Loads a prior :class:`application.workflow.run_models.WorkflowResult` by ``run_id``
and decides whether the operator may start a fresh attempt against the same
``(modelo, period)`` axis. Returns a
:class:`application.workflow.resume.WorkflowResumeContext` the caller hands to
:meth:`application.workflow.engine.WorkflowEngine.run_for_period` to drive the new
attempt.
The action is pure-local: no AEAT contact, no live read or write, no
mutation of the prior run record. Resuming a workflow is the operator
asking the local orchestrator to retry; whether that retry then
contacts AEAT depends on the engine, not on this action.
This module uses :class:`application.workflow.run_models.WorkflowResult`,
:class:`application.workflow.engine.WorkflowEngine`, and
:class:`domain.deadlines.models.ModeloDeadline` for workflow resumption logic.
See Also:
:class:`application.workflow.run_models.WorkflowResult`
Persisted terminal run record inspected before any resume context is
returned.
:class:`application.workflow.persistence.WorkflowRunRepository`
Secure run-history repository behind
:func:`application.workflow.persistence.load_run` and
:func:`application.workflow.persistence.list_runs`.
:class:`application.workflow.engine.WorkflowEngine`
Fresh attempt executor that consumes
:class:`application.workflow.resume.WorkflowResumeContext` through
``run_for_period(resumed_from=...)``.
:mod:`application.modelo`
Owns visible modelo work addressing, revision selection, and conversion
from registry filing periods to workflow periods.
:mod:`entrypoints.cli._modelo_work_runs_cli`
CLI surface that resolves operator resume selectors and emits
:class:`application.workflow.resume.WorkflowResumeTargetResolution`
metadata.
Resumability rules:
* the prior result MUST carry ``final_stage = ABORTED`` — DONE
results are already filed and cannot be retried; in-progress
results are not surfaced through
:func:`application.workflow.persistence.load_run` and so cannot reach this path.
* the prior result's ``aborted_reason`` MUST NOT be terminal-by-
design (``NO_PENDING_OBLIGATION``, ``ALREADY_FILED``,
``USER_CANCELLED``). Those abort reasons describe states where
retrying would not produce a different outcome.
* the prior result MUST carry an ``obligation`` — without it we
cannot enumerate the ``(modelo, period)`` to retry against.
"""
from __future__ import annotations
import re
from dataclasses import dataclass, replace
from datetime import datetime
from typing import TYPE_CHECKING
from pydantic import BaseModel, Field
from ...core.hex import HEX_PATTERN_16, HEX_PATTERN_64
from ...core.identity.hex_ids import CalculationRevisionId, WorkUnitId
from ...core.models import STRICT_FROZEN_CONFIG
from ...core.period import Period
from ...domain.modelos.work_unit import WorkUnitCatalogue
from ..modelo.calculation_action_ports import CalculationActionPorts
from .abort import WorkflowAbortReason
from .errors import WorkflowError
from .persistence import list_runs, load_run
from .run_models import WorkflowObligationFacts, WorkflowResult, WorkflowStage
if TYPE_CHECKING:
#: ``RevisionId`` is an ``Annotated[str, ...]`` alias, but importing it from
#: the registry package executes that package, and the whole registry --
#: 153 modules -- comes with it. This module is imported eagerly by the
#: workflow package, so a bare ``cadrumo --help`` paid a full registry
#: import for one type alias used only in annotations.
#:
#: Safe to defer here and checked rather than assumed: this module carries
#: ``from __future__ import annotations`` so annotations are never
#: evaluated at runtime, the three functions annotated with it are
#: undecorated, and the pydantic models in this module use
#: ``CalculationRevisionId`` instead -- a pydantic field WOULD need the
#: symbol at model-build time and could not be deferred this way.
from ...domain.calculations.registry.ids import RevisionId
from ...domain.modelos.work_unit import WorkUnit
from ..modelo.work_addressing import ModeloResolvedRevisionProjection, ModeloWorkTarget
[docs]
class WorkflowResumeRefusedError(WorkflowError):
"""Raised when a prior :class:`application.workflow.run_models.WorkflowResult` cannot be resumed."""
[docs]
class WorkflowResumeRunAmbiguousError(WorkflowError):
"""Raised when natural-key resume matches more than one workflow run."""
def __init__(
self,
*,
modelo: str,
period: Period,
candidates: tuple[WorkflowResumeRunCandidate, ...],
) -> None:
"""Capture the ambiguous modelo, period, and matching run candidates."""
self.modelo = modelo
self.period = period
self.candidates = candidates
super().__init__(
translated_message="application.workflow.errors.resume_run_ambiguous",
context={
"modelo": modelo,
"period": str(period),
"candidate_count": str(len(candidates)),
"candidates": workflow_resume_candidate_lines(candidates),
},
)
_NON_RESUMABLE_REASONS: frozenset[WorkflowAbortReason] = frozenset(
{
WorkflowAbortReason.NO_PENDING_OBLIGATION,
WorkflowAbortReason.ALREADY_FILED,
WorkflowAbortReason.USER_CANCELLED,
},
)
def _captured_work_catalogue(bucket_id: str | None) -> tuple[WorkUnitCatalogue, str]:
"""Capture the catalogue once at the workflow operation boundary."""
from ..modelo.work_selection import ModeloWorkSelectorRequest, resolve_modelo_work_bucket
from ..modelo.work_unit_repository import work_unit_catalogue_repository
resolved_bucket_id = resolve_modelo_work_bucket(ModeloWorkSelectorRequest(bucket_id=bucket_id))
return (work_unit_catalogue_repository(bucket_id=resolved_bucket_id).load(), resolved_bucket_id)
_WORKFLOW_RUN_ID_RE = re.compile(HEX_PATTERN_16)
_WORK_UNIT_ID_RE = re.compile(HEX_PATTERN_64)
[docs]
class WorkflowResumeRunCandidate(BaseModel):
"""Operator-facing workflow run candidate for natural-key resume guidance."""
model_config = STRICT_FROZEN_CONFIG
run_id: str = Field(min_length=16, max_length=16)
modelo: str = Field(min_length=1, max_length=8)
period: Period
final_stage: str = Field(min_length=1, max_length=64)
aborted_reason: str | None = None
started_at: datetime
short_work_unit_id: str | None = None
work_unit_id: WorkUnitId | None = None
[docs]
class WorkflowResumeContext(BaseModel):
"""Inputs the engine needs to start a fresh attempt over a prior run.
Produced from a resumable :class:`application.workflow.run_models.WorkflowResult`
and passed to
:meth:`application.workflow.engine.WorkflowEngine.run_for_period` by callers
that launch the retry.
"""
model_config = STRICT_FROZEN_CONFIG
resumed_from_run_id: str = Field(min_length=16, max_length=16)
modelo: str = Field(min_length=1, max_length=8)
period: Period
obligation: WorkflowObligationFacts
aborted_reason: WorkflowAbortReason
@dataclass(frozen=True)
class _ResumeTargetInputs:
"""Normalised operator selectors used by the unified resume resolver."""
target: str | None
workflow_run_id: str | None
work_unit_id: str | None
calculation_revision_id: CalculationRevisionId | None
modelo: str | None
year: int | None
period: Period | None
registry_revision_id: RevisionId | None
bucket_id: str | None
selector: object | None
visible_supplied: bool
[docs]
def resume_modelo_workflow(run_id: str) -> WorkflowResumeContext:
"""Validate that ``run_id`` may be resumed and return a fresh-attempt context.
The caller is expected to drive
:meth:`application.workflow.engine.WorkflowEngine.run_for_period` with
``modelo=context.modelo`` and ``period=context.period`` to produce
a fresh :class:`application.workflow.run_models.WorkflowResult`.
Args:
run_id: The 16-character hex run id of the prior aborted workflow
run to resume.
Returns:
A :class:`application.workflow.resume.WorkflowResumeContext` carrying the
modelo, period, obligation, and aborted reason for the prior run.
Raises:
WorkflowResumeRefusedError: When the prior run is not in
``ABORTED`` state, was aborted for a non-resumable reason,
or lacks an ``obligation``.
"""
prior: WorkflowResult = load_run(run_id)
if prior.final_stage is not WorkflowStage.ABORTED:
raise WorkflowResumeRefusedError(
translated_message="application.workflow.errors.resume_refused_not_aborted",
context={"run_id": run_id, "final_stage": prior.final_stage.value},
)
if prior.aborted_reason is None: # defensive: validator enforces this
raise WorkflowResumeRefusedError(
translated_message="application.workflow.errors.resume_refused_no_aborted_reason",
context={"run_id": run_id},
)
if prior.aborted_reason in _NON_RESUMABLE_REASONS:
raise WorkflowResumeRefusedError(
# BOTH message and translated_message, deliberately. The
# operator-facing envelope resolves ``translated_message`` first
# (``core.errors.error_codes.resolve_error_message``), so this
# ``message`` changes nothing an operator sees -- it changes only
# ``str(exc)``, which is what a traceback and a failing test's own
# summary line show.
#
# Without it the failure line is the bare key, identical for every
# non-resumable reason, with the discriminating one reachable only
# through ``context["reason"]``. That is how two unrelated defects
# come to present byte-identically and get triaged as one.
f"workflow run {run_id} cannot be resumed: aborted as {prior.aborted_reason.value}",
translated_message="application.workflow.errors.resume_refused_terminal_reason",
context={"run_id": run_id, "reason": prior.aborted_reason.value},
)
if prior.obligation is None:
raise WorkflowResumeRefusedError(
translated_message="application.workflow.errors.resume_refused_no_obligation",
context={"run_id": run_id},
)
return WorkflowResumeContext(
resumed_from_run_id=prior.run_id,
modelo=prior.obligation.modelo,
period=prior.obligation.period,
obligation=prior.obligation,
aborted_reason=prior.aborted_reason,
)
[docs]
def resolve_modelo_workflow_resume_target(
*,
target: str | None = None,
workflow_run_id: str | None = None,
work_unit_id: str | None = None,
calculation_revision_id: CalculationRevisionId | None = None,
modelo: str | None = None,
year: int | None = None,
period: Period | None = None,
registry_revision_id: RevisionId | None = None,
bucket_id: str | None = None,
selector: object | None = None,
ports: CalculationActionPorts,
) -> WorkflowResumeTargetResolution:
"""Resolve the operator's resume address and return a target resolution.
Exact run ids remain the direct path. Work-unit ids, calculation-revision
ids, and visible modelo/year/period selectors resolve through the public
modelo addressing facade before workflow run lookup, so this service does
not duplicate modelo selector policy.
Returns:
A :class:`application.workflow.resume.WorkflowResumeTargetResolution`
carrying the selected run id and any resolved modelo work metadata.
"""
inputs = _resume_target_inputs(
target=target,
workflow_run_id=workflow_run_id,
work_unit_id=work_unit_id,
calculation_revision_id=calculation_revision_id,
modelo=modelo,
year=year,
period=period,
registry_revision_id=registry_revision_id,
bucket_id=bucket_id,
selector=selector,
)
_reject_resume_target_contradiction(inputs)
inputs = _classify_resume_target(inputs)
if inputs.workflow_run_id is not None:
return _workflow_run_id_resolution(inputs.workflow_run_id, source="workflow_run_id")
if inputs.calculation_revision_id is not None:
return _resolve_resume_from_calculation_revision(inputs.calculation_revision_id, ports=ports)
if inputs.work_unit_id is not None:
return _resolve_resume_from_work_unit_id(inputs.work_unit_id, selector=inputs.selector, ports=ports)
if inputs.visible_supplied:
return _resolve_resume_from_visible_inputs(inputs, ports=ports)
raise WorkflowError(translated_message="application.workflow.errors.resume_target_required")
def _resume_target_inputs(
*,
target: str | None,
workflow_run_id: str | None,
work_unit_id: str | None,
calculation_revision_id: CalculationRevisionId | None,
modelo: str | None,
year: int | None,
period: Period | None,
registry_revision_id: RevisionId | None,
bucket_id: str | None,
selector: object | None,
) -> _ResumeTargetInputs:
"""Normalise optional exact ids while retaining visible-selector presence."""
clean_target = _clean_resume_id(target)
clean_run_id = _clean_resume_id(workflow_run_id)
clean_work_id = _clean_resume_id(work_unit_id)
clean_revision_id = _clean_resume_id(calculation_revision_id)
return _ResumeTargetInputs(
target=clean_target,
workflow_run_id=clean_run_id,
work_unit_id=clean_work_id,
calculation_revision_id=clean_revision_id,
modelo=modelo,
year=year,
period=period,
registry_revision_id=registry_revision_id,
bucket_id=bucket_id,
selector=selector,
visible_supplied=any(value is not None for value in (modelo, year, period, registry_revision_id, bucket_id)),
)
def _clean_resume_id(value: str | None) -> str | None:
"""Return a stripped id, treating whitespace-only input as absent."""
if value is None:
return None
return value.strip() or None
def _reject_resume_target_contradiction(inputs: _ResumeTargetInputs) -> None:
"""Refuse multiple exact addresses or an exact plus visible address."""
exact_count = sum(
value is not None
for value in (inputs.target, inputs.workflow_run_id, inputs.work_unit_id, inputs.calculation_revision_id)
)
if exact_count > 1 or (inputs.target is not None and inputs.visible_supplied):
raise WorkflowError(translated_message="application.workflow.errors.resume_target_contradiction")
def _classify_resume_target(inputs: _ResumeTargetInputs) -> _ResumeTargetInputs:
"""Map the generic target token onto its canonical exact-id field."""
if inputs.target is None:
return inputs
if _WORKFLOW_RUN_ID_RE.fullmatch(inputs.target):
return replace(inputs, workflow_run_id=inputs.target)
if _WORK_UNIT_ID_RE.fullmatch(inputs.target):
return replace(inputs, work_unit_id=inputs.target)
raise WorkflowError(
translated_message="application.workflow.errors.resume_target_invalid",
context={"target": inputs.target},
)
def _resolve_resume_from_visible_inputs(
inputs: _ResumeTargetInputs,
*,
ports: CalculationActionPorts,
) -> WorkflowResumeTargetResolution:
"""Validate and resolve a visible modelo filing selector."""
if inputs.modelo is None or inputs.year is None or inputs.period is None:
raise WorkflowError(
translated_message="application.workflow.errors.resume_visible_target_incomplete",
context={
"modelo": inputs.modelo or "",
"year": "" if inputs.year is None else str(inputs.year),
"period": "" if inputs.period is None else str(inputs.period),
},
)
return _resolve_resume_from_visible_target(
modelo=inputs.modelo,
year=inputs.year,
period=inputs.period,
registry_revision_id=inputs.registry_revision_id,
bucket_id=inputs.bucket_id,
selector=inputs.selector,
ports=ports,
)
def _workflow_run_id_resolution(run_id: str, *, source: str) -> WorkflowResumeTargetResolution:
if not _WORKFLOW_RUN_ID_RE.fullmatch(run_id):
raise WorkflowError(
translated_message="application.workflow.errors.resume_run_id_invalid",
context={"run_id": run_id},
)
return WorkflowResumeTargetResolution(run_id=run_id, source=source)
def _resolve_resume_from_calculation_revision(
calculation_revision_id: CalculationRevisionId,
*,
ports: CalculationActionPorts,
) -> WorkflowResumeTargetResolution:
from ..modelo.calculation_actions import get_calculation_revision
from ..modelo.work_lifecycle import get_work_unit
revision = get_calculation_revision(calculation_revision_id, ports=ports)
work_unit = get_work_unit(revision.work_unit_id, ports=ports.work_lifecycle_ports)
return _resolve_resume_from_work_unit(
work_unit,
source="calculation_revision_id",
calculation_revision_id=revision.calculation_revision_id,
)
def _resolve_resume_from_work_unit_id(
work_unit_id: str,
*,
selector: object | None,
ports: CalculationActionPorts,
) -> WorkflowResumeTargetResolution:
from ..modelo.work_addressing import ModeloExactWorkUnitTarget, resolve_modelo_work_address_unit
target = ModeloExactWorkUnitTarget(work_unit_id=work_unit_id)
catalogue, bucket_id = _captured_work_catalogue(None)
if selector is not None:
_resolve_revision_for_resume_target(
target=target,
selector=selector,
catalogue=catalogue,
bucket_id=bucket_id,
ports=ports,
)
return _resolve_resume_from_work_unit(
resolve_modelo_work_address_unit(target.to_work_address(), catalogue=catalogue, bucket_id=bucket_id),
source="work_unit_id",
latest=True,
)
def _resolve_resume_from_visible_target(
*,
modelo: str,
year: int,
period: Period,
registry_revision_id: RevisionId | None,
bucket_id: str | None,
selector: object | None,
ports: CalculationActionPorts,
) -> WorkflowResumeTargetResolution:
from ..modelo.work_addressing import (
ModeloExactWorkUnitTarget,
ModeloVisibleFilingTarget,
resolve_modelo_work_address_unit,
)
filing_period = _resolve_visible_period(modelo=modelo, year=year, period=period)
target = ModeloVisibleFilingTarget(
modelo=modelo,
filing_year=year,
period=filing_period,
registry_revision_id=registry_revision_id,
bucket_id=bucket_id,
)
catalogue, resolved_bucket_id = _captured_work_catalogue(bucket_id)
if selector is not None:
revision = _resolve_revision_for_resume_target(
target=target,
selector=selector,
catalogue=catalogue,
bucket_id=resolved_bucket_id,
ports=ports,
)
exact_target = ModeloExactWorkUnitTarget(work_unit_id=revision.work_unit_id)
resolution = _resolve_resume_from_work_unit(
resolve_modelo_work_address_unit(
exact_target.to_work_address(),
catalogue=catalogue,
bucket_id=resolved_bucket_id,
),
source="visible_target_revision_selector",
)
return resolution.model_copy(
update={
"source": "visible_target_revision_selector",
"calculation_revision_id": revision.calculation_revision_id,
"short_calculation_revision_id": revision.short_calculation_revision_id,
},
)
return resolve_modelo_workflow_run_for_resume(
target,
source="visible_target",
catalogue=catalogue,
bucket_id=resolved_bucket_id,
)
def _resolve_revision_for_resume_target(
*,
target: ModeloWorkTarget,
selector: object,
catalogue: WorkUnitCatalogue,
bucket_id: str,
ports: CalculationActionPorts,
) -> ModeloResolvedRevisionProjection:
from ..modelo.selectors import ModeloCalculationRevisionSelector
from ..modelo.work_addressing import ModeloRevisionPick, resolve_modelo_revision_pick
try:
revision_selector = (
selector
if isinstance(selector, ModeloCalculationRevisionSelector)
else ModeloCalculationRevisionSelector(str(selector).strip())
)
except ValueError as exc:
raise WorkflowError(
translated_message="application.workflow.errors.resume_revision_selector_invalid",
context={"selector": str(selector)},
) from exc
return resolve_modelo_revision_pick(
target=target,
pick=ModeloRevisionPick(selector=revision_selector),
catalogue=catalogue,
resolved_bucket_id=bucket_id,
calculation_repository=ports.calculation_repository,
operation=ports.operation,
)
def _resolve_visible_period(*, modelo: str, year: int, period: Period) -> Period:
if period.filing_year != year:
raise WorkflowError(
translated_message="application.workflow.errors.resume_visible_target_incomplete",
context={"modelo": modelo, "year": str(year), "period": str(period)},
)
return period
[docs]
def find_latest_run_for_period(*, modelo: str, period: Period) -> WorkflowResult:
"""Return the most recent persisted workflow run for ``(modelo, period)``.
A workflow run id is a 16-character hash an operator cannot derive
by hand, so a caller that only knows the ``(modelo, period)`` of a
work unit needs a way to resolve the run id. This helper scans the
persisted run history and returns the newest run whose resolved
obligation matches the supplied ``(modelo, period)``.
The returned run is *not* gated for resumability — pass its
``run_id`` to :func:`application.workflow.resume.resume_modelo_workflow`, which
applies the resumability rules and produces a precise refusal if the latest
run cannot be retried.
Args:
modelo: Target modelo identifier.
period: Target typed workflow period.
Returns:
The newest matching :class:`application.workflow.run_models.WorkflowResult`.
Raises:
WorkflowError: When no persisted run targets ``(modelo, period)``.
"""
matches = _runs_for_period(modelo=modelo, period=period)
if not matches:
raise WorkflowError(
translated_message="application.workflow.errors.no_run_for_period",
context={"modelo": modelo, "period": str(period)},
)
return matches[0]
[docs]
def find_unique_run_for_period(
*,
modelo: str,
period: Period,
work_unit_id: str | None = None,
short_work_unit_id: str | None = None,
) -> WorkflowResult:
"""Return a workflow run for ``(modelo, period)`` or refuse ambiguity.
Natural-key resume is an operator-facing lookup. If more than one
persisted run exists for the same workflow period, the caller must
choose an exact run id instead of guessing which attempt to resume.
Returns:
The unique matching :class:`application.workflow.run_models.WorkflowResult`.
"""
matches = _runs_for_period(modelo=modelo, period=period)
if not matches:
raise WorkflowError(
translated_message="application.workflow.errors.no_run_for_period",
context={"modelo": modelo, "period": str(period)},
)
if len(matches) > 1:
raise WorkflowResumeRunAmbiguousError(
modelo=modelo,
period=period,
candidates=tuple(
_workflow_resume_run_candidate(
run,
work_unit_id=work_unit_id,
short_work_unit_id=short_work_unit_id,
)
for run in matches
),
)
return matches[0]
[docs]
def resolve_modelo_workflow_run_for_resume(
target: ModeloWorkTarget,
*,
source: str = "modelo_work_target",
catalogue: WorkUnitCatalogue,
bucket_id: str,
) -> WorkflowResumeTargetResolution:
"""Resolve a modelo work target to a resume target resolution.
The modelo application facade remains the owner of visible filing
target lookup and registry-period to workflow-period conversion.
Natural-key targets require exactly one persisted workflow run for
that period; exact work-unit targets select the newest run for the
resolved workflow period.
Returns:
A :class:`application.workflow.resume.WorkflowResumeTargetResolution`
suitable for passing to :func:`application.workflow.resume.resume_modelo_workflow`.
"""
from ..modelo.work_addressing import ModeloExactWorkUnitTarget, ModeloWorkAddress, resolve_modelo_work_target
resolution = resolve_modelo_work_target(target, catalogue=catalogue, bucket_id=bucket_id)
if resolution.work_unit is None:
raise WorkflowError("a resolved modelo work target must carry the work unit the resume reads")
exact_target = isinstance(target, ModeloExactWorkUnitTarget) or (
isinstance(target, ModeloWorkAddress) and target.work_unit_id is not None
)
return _resolve_resume_from_work_unit(
resolution.work_unit,
source=source,
latest=exact_target,
)
def _resolve_resume_from_work_unit(
work_unit: WorkUnit,
*,
source: str,
latest: bool = False,
calculation_revision_id: CalculationRevisionId | None = None,
) -> WorkflowResumeTargetResolution:
from ..modelo.work_addressing import project_modelo_work_unit
from ..modelo.workflow_gate import workflow_period_for_work_unit
projection = project_modelo_work_unit(work_unit)
workflow_period = workflow_period_for_work_unit(work_unit)
if latest:
run = find_latest_run_for_period(modelo=projection.modelo, period=workflow_period)
else:
run = find_unique_run_for_period(
modelo=projection.modelo,
period=workflow_period,
work_unit_id=projection.work_unit_id,
short_work_unit_id=projection.short_work_unit_id,
)
return WorkflowResumeTargetResolution(
run_id=run.run_id,
source=source,
modelo=projection.modelo,
period=workflow_period,
filing_year=projection.filing_year,
work_unit_id=projection.work_unit_id,
short_work_unit_id=projection.short_work_unit_id,
calculation_revision_id=calculation_revision_id,
short_calculation_revision_id=calculation_revision_id[-12:] if calculation_revision_id is not None else None,
)
[docs]
def workflow_resume_candidate_lines(candidates: tuple[WorkflowResumeRunCandidate, ...]) -> str:
"""Return tabular candidate guidance for ambiguous natural-key resume.
Args:
candidates: :class:`WorkflowResumeRunCandidate` rows collected from the
ambiguous workflow-period lookup.
"""
rows = [
"candidates:",
"run_id\tmodelo\tperiod\tfinal_stage\taborted_reason\tstarted_at\tshort_work_unit_id\twork_unit_id",
]
for candidate in candidates:
rows.append(
"\t".join(
(
candidate.run_id,
candidate.modelo,
str(candidate.period),
candidate.final_stage,
candidate.aborted_reason or "",
candidate.started_at.isoformat(),
candidate.short_work_unit_id or "",
candidate.work_unit_id or "",
),
),
)
return "\n".join(rows)
def _runs_for_period(*, modelo: str, period: Period) -> list[WorkflowResult]:
matches = [
run
for run in list_runs()
if run.obligation is not None and run.obligation.modelo == modelo and run.obligation.period == period
]
matches.sort(key=lambda run: run.started_at, reverse=True)
return matches
def _workflow_resume_run_candidate(
run: WorkflowResult,
*,
work_unit_id: str | None = None,
short_work_unit_id: str | None = None,
) -> WorkflowResumeRunCandidate:
if run.obligation is None:
raise WorkflowError("a resumable workflow run must carry the obligation it filed against")
return WorkflowResumeRunCandidate(
run_id=run.run_id,
modelo=run.obligation.modelo,
period=run.obligation.period,
final_stage=run.final_stage.value,
aborted_reason=run.aborted_reason.value if run.aborted_reason is not None else None,
started_at=run.started_at,
work_unit_id=work_unit_id,
short_work_unit_id=short_work_unit_id,
)
__all__ = [
"WorkflowResumeContext",
"WorkflowResumeRefusedError",
"WorkflowResumeRunAmbiguousError",
"WorkflowResumeRunCandidate",
"WorkflowResumeTargetResolution",
"find_latest_run_for_period",
"find_unique_run_for_period",
"resolve_modelo_workflow_resume_target",
"resolve_modelo_workflow_run_for_resume",
"resume_modelo_workflow",
"workflow_resume_candidate_lines",
]