From 7b6767cf875ea639e1e6a5f1e50966283e35ddb9 Mon Sep 17 00:00:00 2001 From: Yernat Yestekov <2068106+doublewhy@users.noreply.github.com> Date: Wed, 12 Aug 2026 12:58:43 -0700 Subject: [PATCH 1/7] fix(runtime): stop swallowing timestamp errors in workflow timeout reconciliation `_workflow_has_timed_out` wrapped both timestamp parses in `except Exception: return False`, so any unparseable value reported "not timed out". A RUNNING workflow whose recorded `started_at` could not be parsed therefore had no derivable deadline and stayed RUNNING for the lifetime of the control plane: reconciliation could never reclaim it, even decades past a one-second timeout. The two timestamps have different scope, so they are now handled differently: - `submitted_at` is the caller's reconciliation clock and governs every workflow in the pass, so an unusable value raises instead of quietly disabling all timeouts. The HTTP adapter already maps `ValueError` to 409 for this route. - A per-workflow `started_at` that cannot be parsed no longer blocks reclamation; the workflow is timed out under a distinct terminal reason so it stays diagnosable rather than looking like an ordinary timeout. Co-Authored-By: Claude Opus 5 (1M context) --- .../raes_runtime/control_plane_timeouts.py | 40 +++++--- ...runtime_workflow_timeout_reconciliation.py | 91 +++++++++++++++++++ 2 files changed, 118 insertions(+), 13 deletions(-) create mode 100644 implementations/python/tests/test_runtime_workflow_timeout_reconciliation.py diff --git a/implementations/python/packages/raes_runtime/control_plane_timeouts.py b/implementations/python/packages/raes_runtime/control_plane_timeouts.py index 6d9f0dfc1..ec1aaa537 100644 --- a/implementations/python/packages/raes_runtime/control_plane_timeouts.py +++ b/implementations/python/packages/raes_runtime/control_plane_timeouts.py @@ -14,6 +14,9 @@ from .control_plane_workflows import maybe_apply_compensation, parse_timestamp +TIMED_OUT_REASON = "workflow timed out" +UNPARSEABLE_START_REASON = "workflow timed out: started_at could not be parsed" + def workflow_timeout_update( snapshot: RuntimeSnapshot, @@ -26,11 +29,10 @@ def workflow_timeout_update( update = None timeout_seconds = _eligible_workflow_timeout_seconds(entry) normalized = _running_workflow_result(orchestration_results.get(workflow_address)) - if ( - timeout_seconds is not None - and normalized is not None - and _workflow_has_timed_out(normalized, timeout_seconds, submitted_at) - ): + terminal_reason = None + if timeout_seconds is not None and normalized is not None: + terminal_reason = _workflow_timeout_reason(normalized, timeout_seconds, submitted_at) + if terminal_reason is not None: update = _timed_out_workflow_update( snapshot, workflow_address, @@ -38,6 +40,7 @@ def workflow_timeout_update( timeout_seconds, orchestration_history, submitted_at, + terminal_reason, ) return update @@ -77,17 +80,26 @@ def _coerce_timeout_seconds(raw: object) -> int | None: return timeout -def _workflow_has_timed_out( +def _workflow_timeout_reason( normalized: WorkflowExecutionState, timeout_seconds: int, submitted_at: str, -) -> bool: +) -> str | None: + """Return the terminal reason when the workflow must time out, else ``None``. + + ``submitted_at`` is the caller's reconciliation clock and governs the whole + pass, so an unusable value is raised rather than quietly disabling every + timeout. A running workflow whose own ``started_at`` cannot be parsed has no + derivable deadline; reporting "not timed out" would pin it in RUNNING + forever, so it is reclaimed under a distinct reason instead. + """ + + current = parse_timestamp(submitted_at).timestamp() try: deadline = parse_timestamp(normalized.started_at).timestamp() + timeout_seconds - current = parse_timestamp(submitted_at).timestamp() - except Exception: - return False - return current >= deadline + except (TypeError, ValueError): + return UNPARSEABLE_START_REASON + return TIMED_OUT_REASON if current >= deadline else None def _timed_out_workflow_update( @@ -97,8 +109,9 @@ def _timed_out_workflow_update( timeout_seconds: int, orchestration_history: dict[str, list[dict[str, object]]], submitted_at: str, + terminal_reason: str, ) -> tuple[dict[str, object], list[dict[str, object]]]: - timed_out_state = _timed_out_workflow_state(normalized, submitted_at) + timed_out_state = _timed_out_workflow_state(normalized, submitted_at, terminal_reason) history = orchestration_history.setdefault(workflow_address, []) history.append( WorkflowHistoryEvent( @@ -119,6 +132,7 @@ def _timed_out_workflow_update( def _timed_out_workflow_state( normalized: WorkflowExecutionState, submitted_at: str, + terminal_reason: str, ) -> WorkflowExecutionState: return WorkflowExecutionState( state_schema_version=normalized.state_schema_version, @@ -126,7 +140,7 @@ def _timed_out_workflow_state( run_id=normalized.run_id, started_at=normalized.started_at, updated_at=submitted_at, - terminal_reason="workflow timed out", + terminal_reason=terminal_reason, compensation_status=WorkflowCompensationStatus.NOT_REQUIRED, compensation_started_at=None, compensation_updated_at=None, diff --git a/implementations/python/tests/test_runtime_workflow_timeout_reconciliation.py b/implementations/python/tests/test_runtime_workflow_timeout_reconciliation.py new file mode 100644 index 000000000..fe475dd6c --- /dev/null +++ b/implementations/python/tests/test_runtime_workflow_timeout_reconciliation.py @@ -0,0 +1,91 @@ +"""Workflow timeout reconciliation edge cases for the runtime control plane.""" + +from __future__ import annotations + +import pytest +from raes_contracts.planning import RuntimeDomain +from raes_contracts.runtime_state import RuntimeSnapshot, SnapshotEntry +from raes_contracts.workflow import WorkflowExecutionState, WorkflowStatus +from raes_runtime.control_plane_timeouts import ( + TIMED_OUT_REASON, + UNPARSEABLE_START_REASON, + workflow_timeout_update, +) + +_WORKFLOW_ADDRESS = "orchestration.workflow.response" + + +def _workflow_entry(timeout_seconds: int = 1) -> SnapshotEntry: + return SnapshotEntry( + address=_WORKFLOW_ADDRESS, + domain=RuntimeDomain.ORCHESTRATION, + resource_type="workflow", + payload={"execution_contract": {"timeout_seconds": timeout_seconds}}, + ) + + +def _running_result(started_at: str) -> dict[str, object]: + """Build a persisted RUNNING workflow payload with ``started_at`` as recorded. + + The payload is edited after construction because the model rejects values it + considers unusable, while reconciliation reads snapshots back through + ``WorkflowExecutionState.from_payload`` and must cope with whatever the store + actually holds. + """ + + payload = WorkflowExecutionState( + workflow_status=WorkflowStatus.RUNNING, + run_id="run-1", + started_at="2000-01-01T00:00:00Z", + updated_at="2000-01-01T00:00:00Z", + ).to_payload() + payload["started_at"] = started_at + return payload + + +def _reconcile(started_at: str, submitted_at: str) -> tuple[dict[str, object], list[dict[str, object]]] | None: + return workflow_timeout_update( + RuntimeSnapshot(), + _WORKFLOW_ADDRESS, + _workflow_entry(), + {_WORKFLOW_ADDRESS: _running_result(started_at)}, + {}, + submitted_at, + ) + + +def test_expired_workflow_is_timed_out(): + update = _reconcile("2000-01-01T00:00:00Z", "2000-01-01T00:01:00Z") + + assert update is not None + assert update[0]["workflow_status"] == WorkflowStatus.TIMED_OUT.value + assert update[0]["terminal_reason"] == TIMED_OUT_REASON + + +def test_workflow_inside_its_deadline_is_left_running(): + assert _reconcile("2000-01-01T00:00:00Z", "2000-01-01T00:00:00Z") is None + + +@pytest.mark.parametrize("started_at", ["None", "not-a-timestamp", "2000-13-45T99:99:99Z"]) +def test_workflow_with_unparseable_started_at_is_reclaimed(started_at: str): + """A running workflow with no derivable deadline must not stay RUNNING forever. + + Swallowing the parse failure and reporting "not timed out" pinned such a + workflow in RUNNING for the lifetime of the control plane, so reconciliation + could never reclaim it. + """ + update = _reconcile(started_at, "2000-01-01T00:01:00Z") + + assert update is not None + assert update[0]["workflow_status"] == WorkflowStatus.TIMED_OUT.value + assert update[0]["terminal_reason"] == UNPARSEABLE_START_REASON + + +def test_unparseable_reconciliation_clock_is_raised_not_swallowed(): + """A bad caller-supplied ``now`` governs every workflow, so it must surface. + + Reported as ``ValueError``; the HTTP adapter maps that to 409 rather than + silently disabling timeouts for the whole pass. + """ + with pytest.raises(ValueError): + _reconcile("2000-01-01T00:00:00Z", "not-a-timestamp") From e622821cc75641d9b9636024561c745015383997 Mon Sep 17 00:00:00 2001 From: Yernat Yestekov <2068106+doublewhy@users.noreply.github.com> Date: Wed, 12 Aug 2026 12:58:44 -0700 Subject: [PATCH 2/7] fix(runtime): compare elapsed time against the workflow timeout MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `timeout_seconds` carries no declared upper bound, so adding it to the start instant overflowed for a very large value — as a float timestamp and as a `timedelta`. The previous blanket `except Exception` hid that as "not timed out"; with the exception handling narrowed, it would instead abort the whole reconciliation pass and surface as a 500. Elapsed time is now compared against the timeout, which Python evaluates exactly for an arbitrarily large integer. Co-Authored-By: Claude Opus 5 (1M context) --- .../raes_runtime/control_plane_timeouts.py | 10 +++++-- ...runtime_workflow_timeout_reconciliation.py | 28 +++++++++++++++++++ 2 files changed, 35 insertions(+), 3 deletions(-) diff --git a/implementations/python/packages/raes_runtime/control_plane_timeouts.py b/implementations/python/packages/raes_runtime/control_plane_timeouts.py index ec1aaa537..24114cc81 100644 --- a/implementations/python/packages/raes_runtime/control_plane_timeouts.py +++ b/implementations/python/packages/raes_runtime/control_plane_timeouts.py @@ -94,12 +94,16 @@ def _workflow_timeout_reason( forever, so it is reclaimed under a distinct reason instead. """ - current = parse_timestamp(submitted_at).timestamp() + current = parse_timestamp(submitted_at) try: - deadline = parse_timestamp(normalized.started_at).timestamp() + timeout_seconds + started = parse_timestamp(normalized.started_at) except (TypeError, ValueError): return UNPARSEABLE_START_REASON - return TIMED_OUT_REASON if current >= deadline else None + # Elapsed time is compared against the timeout rather than added to the start + # instant: `timeout_seconds` has no declared upper bound, and folding a very + # large one into a float timestamp or a timedelta overflows. + elapsed_seconds = (current - started).total_seconds() + return TIMED_OUT_REASON if elapsed_seconds >= timeout_seconds else None def _timed_out_workflow_update( diff --git a/implementations/python/tests/test_runtime_workflow_timeout_reconciliation.py b/implementations/python/tests/test_runtime_workflow_timeout_reconciliation.py index fe475dd6c..f5dcbf1a5 100644 --- a/implementations/python/tests/test_runtime_workflow_timeout_reconciliation.py +++ b/implementations/python/tests/test_runtime_workflow_timeout_reconciliation.py @@ -15,6 +15,15 @@ _WORKFLOW_ADDRESS = "orchestration.workflow.response" +def _workflow_entry_with_timeout(timeout_seconds: int) -> SnapshotEntry: + return SnapshotEntry( + address=_WORKFLOW_ADDRESS, + domain=RuntimeDomain.ORCHESTRATION, + resource_type="workflow", + payload={"execution_contract": {"timeout_seconds": timeout_seconds}}, + ) + + def _workflow_entry(timeout_seconds: int = 1) -> SnapshotEntry: return SnapshotEntry( address=_WORKFLOW_ADDRESS, @@ -81,6 +90,25 @@ def test_workflow_with_unparseable_started_at_is_reclaimed(started_at: str): assert update[0]["terminal_reason"] == UNPARSEABLE_START_REASON +def test_enormous_timeout_reports_not_timed_out_instead_of_overflowing(): + """`timeout_seconds` has no declared upper bound, so it must not overflow. + + Folding a very large timeout into a float timestamp or a timedelta raises + `OverflowError`, which would abort the whole reconciliation pass and surface + as a 500 from the HTTP adapter. + """ + update = workflow_timeout_update( + RuntimeSnapshot(), + _WORKFLOW_ADDRESS, + _workflow_entry_with_timeout(10**400), + {_WORKFLOW_ADDRESS: _running_result("2000-01-01T00:00:00Z")}, + {}, + "2030-01-01T00:00:00Z", + ) + + assert update is None + + def test_unparseable_reconciliation_clock_is_raised_not_swallowed(): """A bad caller-supplied ``now`` governs every workflow, so it must surface. From c5bb9841a2daeedda8ddf0984aa2cd32167d353d Mon Sep 17 00:00:00 2001 From: Yernat Yestekov <2068106+doublewhy@users.noreply.github.com> Date: Wed, 12 Aug 2026 12:58:44 -0700 Subject: [PATCH 3/7] fix(runtime): reject invalid workflow timeout clocks (#1102) --- ...rkflow-timeout-reconciliation-preflight.md | 36 +++ docs/requirements/RUN-317/requirement.md | 7 + docs/requirements/RUN-318/requirement.md | 7 + .../raes_runtime/control_plane_timeouts.py | 99 +++++---- .../control_plane_workflow_control.py | 4 +- .../raes_runtime/control_plane_workflows.py | 19 +- ...runtime_workflow_timeout_reconciliation.py | 205 ++++++++++++++++-- 7 files changed, 311 insertions(+), 66 deletions(-) create mode 100644 docs/decisions/issue-1102-workflow-timeout-reconciliation-preflight.md diff --git a/docs/decisions/issue-1102-workflow-timeout-reconciliation-preflight.md b/docs/decisions/issue-1102-workflow-timeout-reconciliation-preflight.md new file mode 100644 index 000000000..ac54fb746 --- /dev/null +++ b/docs/decisions/issue-1102-workflow-timeout-reconciliation-preflight.md @@ -0,0 +1,36 @@ +# Issue 1102 Workflow Timeout Reconciliation Preflight + +Date: 2026-08-11 + +Issue: #1102. Requirements: RUN-317 and RUN-318. + +## Decision + +Workflow timeout reconciliation treats persisted timestamps as untrusted +durable input. The reconciliation clock is parsed once per operation, and all +timestamps require an explicit ISO-8601 offset before normalization to UTC. +`started_at <= updated_at <= reconciliation clock` is required. Malformed, +naive, or non-monotonic values fail with stable, value-free conflicts; they do +not synthesize a timeout, mutate history, or activate timeout compensation. +Persisted timeout configuration is likewise either absent or a strictly +positive integer. Invalid values cannot silently disable reconciliation or +become an immediate timeout. + +Timeout admission compares `now - started_at` with the configured duration. +It never compares an absolute wall-clock value directly with a duration. The +boundary remains inclusive according to the incumbent workflow state-machine +contract. Only a proven elapsed duration produces `workflow_status: timed_out` +and a `workflow_timed_out` history event. Already-terminal results are unchanged, +so reconciliation remains idempotent across restart. + +This is operational wall-clock timeout handling. It does not reinterpret SDL +logical time, participant episode time, simulated clocks, or backend-native +deadlines as interchangeable evidence. + +## Verification + +Tests cover malformed, naive, offset, future, and non-monotonic timestamps; +invalid timeout values; exact boundaries; enormous durations; empty-control- +plane clock validation; compensation non-triggering for invalid state; and +terminal replay idempotency. The runtime timeout suite, lint, policy, and +canonical verification remain required. diff --git a/docs/requirements/RUN-317/requirement.md b/docs/requirements/RUN-317/requirement.md index cd509134c..3d265a93e 100644 --- a/docs/requirements/RUN-317/requirement.md +++ b/docs/requirements/RUN-317/requirement.md @@ -17,3 +17,10 @@ The runtime shall support portable handling of declared clocks and time domains ## Rationale If authored scenarios and experiments can depend on time domains, the runtime needs a portable model for carrying those domains through execution and observation. + +## Traceability + +- DOCUMENTS → GITHUB_ISSUE `1102` (Fail-closed workflow timeout reconciliation) +- DOCUMENTS → DOCUMENTATION `docs/decisions/issue-1102-workflow-timeout-reconciliation-preflight.md` (Wall-clock timeout reconciliation boundary) +- IMPLEMENTS → CODE_FILE `implementations/python/packages/raes_runtime/control_plane_timeouts.py` (Strict persisted timestamp and elapsed-time handling) +- TESTS → TEST `implementations/python/tests/test_runtime_workflow_timeout_reconciliation.py` (Malformed timestamp, boundary, and restart regressions) diff --git a/docs/requirements/RUN-318/requirement.md b/docs/requirements/RUN-318/requirement.md index df3182547..1b466ba51 100644 --- a/docs/requirements/RUN-318/requirement.md +++ b/docs/requirements/RUN-318/requirement.md @@ -17,3 +17,10 @@ The runtime shall support portable time advancement, pacing, synchronization, ti ## Rationale Cross-domain references show that time progression lifecycle behavior is central to replayability and honest cross-realization comparison. + +## Traceability + +- DOCUMENTS → GITHUB_ISSUE `1102` (Fail-closed workflow timeout reconciliation) +- DOCUMENTS → DOCUMENTATION `docs/decisions/issue-1102-workflow-timeout-reconciliation-preflight.md` (Operational timeout lifecycle decision) +- IMPLEMENTS → CODE_FILE `implementations/python/packages/raes_runtime/control_plane_timeouts.py` (Elapsed-duration timeout reconciliation) +- TESTS → TEST `implementations/python/tests/test_runtime_workflow_timeout_reconciliation.py` (Timeout lifecycle regression tests) diff --git a/implementations/python/packages/raes_runtime/control_plane_timeouts.py b/implementations/python/packages/raes_runtime/control_plane_timeouts.py index 24114cc81..14a29f8d6 100644 --- a/implementations/python/packages/raes_runtime/control_plane_timeouts.py +++ b/implementations/python/packages/raes_runtime/control_plane_timeouts.py @@ -2,6 +2,8 @@ from __future__ import annotations +from datetime import datetime + from raes_contracts.planning import RuntimeDomain from raes_contracts.runtime_state import RuntimeSnapshot, SnapshotEntry from raes_contracts.workflow import ( @@ -15,7 +17,11 @@ from .control_plane_workflows import maybe_apply_compensation, parse_timestamp TIMED_OUT_REASON = "workflow timed out" -UNPARSEABLE_START_REASON = "workflow timed out: started_at could not be parsed" +INVALID_RECONCILIATION_CLOCK = "workflow timeout reconciliation clock is invalid" +INVALID_WORKFLOW_STATE = "persisted workflow execution state is invalid" +INVALID_WORKFLOW_TIMESTAMP = "persisted workflow execution timestamps are invalid" +INVALID_TIMEOUT_CONFIGURATION = "persisted workflow timeout configuration is invalid" +NON_MONOTONIC_WORKFLOW_CLOCK = "workflow timeout reconciliation clock precedes persisted workflow state" def workflow_timeout_update( @@ -25,14 +31,19 @@ def workflow_timeout_update( orchestration_results: dict[str, dict[str, object]], orchestration_history: dict[str, list[dict[str, object]]], submitted_at: str, + *, + reconciliation_clock: datetime | None = None, ) -> tuple[dict[str, object], list[dict[str, object]]] | None: + current = reconciliation_clock or _reconciliation_clock(submitted_at) + if current.tzinfo is None or current.utcoffset() is None: + raise ValueError(INVALID_RECONCILIATION_CLOCK) update = None timeout_seconds = _eligible_workflow_timeout_seconds(entry) normalized = _running_workflow_result(orchestration_results.get(workflow_address)) - terminal_reason = None + timed_out = False if timeout_seconds is not None and normalized is not None: - terminal_reason = _workflow_timeout_reason(normalized, timeout_seconds, submitted_at) - if terminal_reason is not None: + timed_out = _workflow_has_timed_out(normalized, timeout_seconds, current) + if timed_out: update = _timed_out_workflow_update( snapshot, workflow_address, @@ -40,11 +51,17 @@ def workflow_timeout_update( timeout_seconds, orchestration_history, submitted_at, - terminal_reason, ) return update +def _reconciliation_clock(submitted_at: str) -> datetime: + try: + return parse_timestamp(submitted_at) + except ValueError: + raise ValueError(INVALID_RECONCILIATION_CLOCK) from None + + def _eligible_workflow_timeout_seconds(entry: SnapshotEntry) -> int | None: timeout_seconds = None if entry.domain == RuntimeDomain.ORCHESTRATION and entry.resource_type == "workflow": @@ -53,57 +70,57 @@ def _eligible_workflow_timeout_seconds(entry: SnapshotEntry) -> int | None: def _running_workflow_result(result_payload: object) -> WorkflowExecutionState | None: - normalized = None - if isinstance(result_payload, dict): + if result_payload is None: + return None + if not isinstance(result_payload, dict): + raise ValueError(INVALID_WORKFLOW_STATE) + try: candidate = WorkflowExecutionState.from_payload(result_payload) - if candidate.workflow_status == WorkflowStatus.RUNNING: - normalized = candidate - return normalized + except (TypeError, ValueError): + raise ValueError(INVALID_WORKFLOW_STATE) from None + return candidate if candidate.workflow_status == WorkflowStatus.RUNNING else None def _workflow_timeout_seconds(payload: object) -> int | None: - timeout = None - if isinstance(payload, dict): - execution_contract_payload = payload.get("execution_contract") - if isinstance(execution_contract_payload, dict): - timeout = _coerce_timeout_seconds(execution_contract_payload.get("timeout_seconds")) - return timeout + if not isinstance(payload, dict): + raise ValueError(INVALID_TIMEOUT_CONFIGURATION) + execution_contract_payload = payload.get("execution_contract") + if execution_contract_payload is None: + return None + if not isinstance(execution_contract_payload, dict): + raise ValueError(INVALID_TIMEOUT_CONFIGURATION) + return _coerce_timeout_seconds(execution_contract_payload.get("timeout_seconds")) def _coerce_timeout_seconds(raw: object) -> int | None: - timeout = None - if raw not in (None, "", 0): - try: - timeout = int(raw) - except (TypeError, ValueError): - timeout = None - return timeout + if raw is None: + return None + if isinstance(raw, bool) or not isinstance(raw, int) or raw <= 0: + raise ValueError(INVALID_TIMEOUT_CONFIGURATION) + return raw -def _workflow_timeout_reason( +def _workflow_has_timed_out( normalized: WorkflowExecutionState, timeout_seconds: int, - submitted_at: str, -) -> str | None: - """Return the terminal reason when the workflow must time out, else ``None``. + current: datetime, +) -> bool: + """Return whether elapsed wall time proves the declared timeout.""" - ``submitted_at`` is the caller's reconciliation clock and governs the whole - pass, so an unusable value is raised rather than quietly disabling every - timeout. A running workflow whose own ``started_at`` cannot be parsed has no - derivable deadline; reporting "not timed out" would pin it in RUNNING - forever, so it is reclaimed under a distinct reason instead. - """ - - current = parse_timestamp(submitted_at) try: started = parse_timestamp(normalized.started_at) - except (TypeError, ValueError): - return UNPARSEABLE_START_REASON + updated = parse_timestamp(normalized.updated_at) + except ValueError: + raise ValueError(INVALID_WORKFLOW_TIMESTAMP) from None + if updated < started: + raise ValueError(INVALID_WORKFLOW_TIMESTAMP) + if current < started or current < updated: + raise ValueError(NON_MONOTONIC_WORKFLOW_CLOCK) # Elapsed time is compared against the timeout rather than added to the start # instant: `timeout_seconds` has no declared upper bound, and folding a very # large one into a float timestamp or a timedelta overflows. elapsed_seconds = (current - started).total_seconds() - return TIMED_OUT_REASON if elapsed_seconds >= timeout_seconds else None + return elapsed_seconds >= timeout_seconds def _timed_out_workflow_update( @@ -113,9 +130,8 @@ def _timed_out_workflow_update( timeout_seconds: int, orchestration_history: dict[str, list[dict[str, object]]], submitted_at: str, - terminal_reason: str, ) -> tuple[dict[str, object], list[dict[str, object]]]: - timed_out_state = _timed_out_workflow_state(normalized, submitted_at, terminal_reason) + timed_out_state = _timed_out_workflow_state(normalized, submitted_at) history = orchestration_history.setdefault(workflow_address, []) history.append( WorkflowHistoryEvent( @@ -136,7 +152,6 @@ def _timed_out_workflow_update( def _timed_out_workflow_state( normalized: WorkflowExecutionState, submitted_at: str, - terminal_reason: str, ) -> WorkflowExecutionState: return WorkflowExecutionState( state_schema_version=normalized.state_schema_version, @@ -144,7 +159,7 @@ def _timed_out_workflow_state( run_id=normalized.run_id, started_at=normalized.started_at, updated_at=submitted_at, - terminal_reason=terminal_reason, + terminal_reason=TIMED_OUT_REASON, compensation_status=WorkflowCompensationStatus.NOT_REQUIRED, compensation_started_at=None, compensation_updated_at=None, diff --git a/implementations/python/packages/raes_runtime/control_plane_workflow_control.py b/implementations/python/packages/raes_runtime/control_plane_workflow_control.py index 8d16ddd80..1cdc1cd85 100644 --- a/implementations/python/packages/raes_runtime/control_plane_workflow_control.py +++ b/implementations/python/packages/raes_runtime/control_plane_workflow_control.py @@ -31,7 +31,7 @@ persist_succeeded_operation, ) from .control_plane_store import ControlPlaneOperationRecord -from .control_plane_timeouts import workflow_timeout_update +from .control_plane_timeouts import _reconciliation_clock, workflow_timeout_update from .control_plane_workflows import maybe_apply_compensation _TERMINAL_WORKFLOW_STATUSES = { @@ -209,6 +209,7 @@ def reconcile_workflow_timeouts( if existing is not None: return existing submitted_at = now or _utc_now() + reconciliation_clock = _reconciliation_clock(submitted_at) changed: list[str] = [] orchestration_results = dict(self._snapshot.orchestration_results) orchestration_history = { @@ -222,6 +223,7 @@ def reconcile_workflow_timeouts( orchestration_results, orchestration_history, submitted_at, + reconciliation_clock=reconciliation_clock, ) if timed_out is None: continue diff --git a/implementations/python/packages/raes_runtime/control_plane_workflows.py b/implementations/python/packages/raes_runtime/control_plane_workflows.py index 951204c57..bce3a3c99 100644 --- a/implementations/python/packages/raes_runtime/control_plane_workflows.py +++ b/implementations/python/packages/raes_runtime/control_plane_workflows.py @@ -186,9 +186,16 @@ def _compensated_workflow_payload( def parse_timestamp(raw: str) -> datetime: - if raw.endswith("Z"): - raw = raw[:-1] + "+00:00" - parsed = datetime.fromisoformat(raw) - if parsed.tzinfo is None: - parsed = parsed.replace(tzinfo=UTC) - return parsed + """Parse one explicit-offset ISO-8601 timestamp and normalize it to UTC.""" + + if not isinstance(raw, str) or not raw: + raise ValueError("timestamp must be an ISO-8601 value with an explicit UTC offset") + normalized = raw[:-1] + "+00:00" if raw.endswith("Z") else raw + try: + parsed = datetime.fromisoformat(normalized) + offset = parsed.utcoffset() + except (OverflowError, TypeError, ValueError): + raise ValueError("timestamp must be an ISO-8601 value with an explicit UTC offset") from None + if parsed.tzinfo is None or offset is None: + raise ValueError("timestamp must be an ISO-8601 value with an explicit UTC offset") + return parsed.astimezone(UTC) diff --git a/implementations/python/tests/test_runtime_workflow_timeout_reconciliation.py b/implementations/python/tests/test_runtime_workflow_timeout_reconciliation.py index f5dcbf1a5..20221a8bc 100644 --- a/implementations/python/tests/test_runtime_workflow_timeout_reconciliation.py +++ b/implementations/python/tests/test_runtime_workflow_timeout_reconciliation.py @@ -3,12 +3,17 @@ from __future__ import annotations import pytest +from raes_backend_stubs.stubs import create_stub_target from raes_contracts.planning import RuntimeDomain from raes_contracts.runtime_state import RuntimeSnapshot, SnapshotEntry from raes_contracts.workflow import WorkflowExecutionState, WorkflowStatus +from raes_runtime.control_plane import RuntimeControlPlane from raes_runtime.control_plane_timeouts import ( + INVALID_RECONCILIATION_CLOCK, + INVALID_TIMEOUT_CONFIGURATION, + INVALID_WORKFLOW_TIMESTAMP, + NON_MONOTONIC_WORKFLOW_CLOCK, TIMED_OUT_REASON, - UNPARSEABLE_START_REASON, workflow_timeout_update, ) @@ -33,7 +38,7 @@ def _workflow_entry(timeout_seconds: int = 1) -> SnapshotEntry: ) -def _running_result(started_at: str) -> dict[str, object]: +def _running_result(started_at: str, updated_at: str | None = None) -> dict[str, object]: """Build a persisted RUNNING workflow payload with ``started_at`` as recorded. The payload is edited after construction because the model rejects values it @@ -49,16 +54,30 @@ def _running_result(started_at: str) -> dict[str, object]: updated_at="2000-01-01T00:00:00Z", ).to_payload() payload["started_at"] = started_at + payload["updated_at"] = updated_at if updated_at is not None else started_at return payload -def _reconcile(started_at: str, submitted_at: str) -> tuple[dict[str, object], list[dict[str, object]]] | None: +def _reconcile( + started_at: str, + submitted_at: str, + *, + updated_at: str | None = None, + timeout_seconds: object = 1, + history: dict[str, list[dict[str, object]]] | None = None, +) -> tuple[dict[str, object], list[dict[str, object]]] | None: + entry = SnapshotEntry( + address=_WORKFLOW_ADDRESS, + domain=RuntimeDomain.ORCHESTRATION, + resource_type="workflow", + payload={"execution_contract": {"timeout_seconds": timeout_seconds}}, + ) return workflow_timeout_update( RuntimeSnapshot(), _WORKFLOW_ADDRESS, - _workflow_entry(), - {_WORKFLOW_ADDRESS: _running_result(started_at)}, - {}, + entry, + {_WORKFLOW_ADDRESS: _running_result(started_at, updated_at)}, + history if history is not None else {}, submitted_at, ) @@ -75,19 +94,76 @@ def test_workflow_inside_its_deadline_is_left_running(): assert _reconcile("2000-01-01T00:00:00Z", "2000-01-01T00:00:00Z") is None -@pytest.mark.parametrize("started_at", ["None", "not-a-timestamp", "2000-13-45T99:99:99Z"]) -def test_workflow_with_unparseable_started_at_is_reclaimed(started_at: str): - """A running workflow with no derivable deadline must not stay RUNNING forever. +def test_workflow_at_its_exact_deadline_is_timed_out(): + update = _reconcile("2000-01-01T00:00:00Z", "2000-01-01T00:00:01Z") - Swallowing the parse failure and reporting "not timed out" pinned such a - workflow in RUNNING for the lifetime of the control plane, so reconciliation - could never reclaim it. - """ - update = _reconcile(started_at, "2000-01-01T00:01:00Z") + assert update is not None + assert update[0]["terminal_reason"] == TIMED_OUT_REASON + + +def test_workflow_with_future_started_at_is_rejected_without_mutation(): + history: dict[str, list[dict[str, object]]] = {} + + with pytest.raises(ValueError, match=NON_MONOTONIC_WORKFLOW_CLOCK): + _reconcile("2000-01-01T00:00:01Z", "2000-01-01T00:00:00Z", history=history) + + assert history == {} + + +def test_workflow_elapsed_time_normalizes_timezone_offsets(): + update = _reconcile("2000-01-01T01:00:00+01:00", "2000-01-01T00:00:01Z") assert update is not None - assert update[0]["workflow_status"] == WorkflowStatus.TIMED_OUT.value - assert update[0]["terminal_reason"] == UNPARSEABLE_START_REASON + assert update[0]["terminal_reason"] == TIMED_OUT_REASON + + +@pytest.mark.parametrize("started_at", ["None", "not-a-timestamp", "2000-13-45T99:99:99Z"]) +def test_workflow_with_unparseable_started_at_fails_without_synthesizing_timeout(started_at: str): + history: dict[str, list[dict[str, object]]] = {} + + with pytest.raises(ValueError, match=INVALID_WORKFLOW_TIMESTAMP): + _reconcile(started_at, "2000-01-01T00:01:00Z", history=history) + + assert history == {} + + +@pytest.mark.parametrize( + ("started_at", "updated_at"), + [ + ("2000-01-01", "2000-01-01T00:00:00Z"), + ("2000-01-01T00:00:00Z", "2000-01-01"), + ], +) +def test_workflow_timestamps_require_explicit_offsets(started_at: str, updated_at: str): + with pytest.raises(ValueError, match=INVALID_WORKFLOW_TIMESTAMP): + _reconcile( + started_at, + "2000-01-02T00:00:00Z", + updated_at=updated_at, + ) + + +def test_reconciliation_clock_cannot_precede_latest_persisted_update(): + history: dict[str, list[dict[str, object]]] = {} + + with pytest.raises(ValueError, match=NON_MONOTONIC_WORKFLOW_CLOCK): + _reconcile( + "2000-01-01T00:00:00Z", + "2000-01-01T00:00:02Z", + updated_at="2000-01-01T00:00:03Z", + history=history, + ) + + assert history == {} + + +def test_updated_at_cannot_precede_started_at(): + with pytest.raises(ValueError, match=INVALID_WORKFLOW_TIMESTAMP): + _reconcile( + "2000-01-01T00:00:01Z", + "2000-01-01T00:00:02Z", + updated_at="2000-01-01T00:00:00Z", + ) def test_enormous_timeout_reports_not_timed_out_instead_of_overflowing(): @@ -115,5 +191,100 @@ def test_unparseable_reconciliation_clock_is_raised_not_swallowed(): Reported as ``ValueError``; the HTTP adapter maps that to 409 rather than silently disabling timeouts for the whole pass. """ - with pytest.raises(ValueError): + with pytest.raises(ValueError, match=INVALID_RECONCILIATION_CLOCK): _reconcile("2000-01-01T00:00:00Z", "not-a-timestamp") + + +def test_naive_reconciliation_clock_is_rejected(): + with pytest.raises(ValueError, match=INVALID_RECONCILIATION_CLOCK): + _reconcile("2000-01-01T00:00:00Z", "2000-01-01T00:00:01") + + +@pytest.mark.parametrize("timeout_seconds", [-1, 0, True, 1.5, "1", "bogus"]) +def test_invalid_timeout_configuration_fails_closed(timeout_seconds: object): + history: dict[str, list[dict[str, object]]] = {} + + with pytest.raises(ValueError, match=INVALID_TIMEOUT_CONFIGURATION): + _reconcile( + "2000-01-01T00:00:00Z", + "2000-01-01T00:01:00Z", + timeout_seconds=timeout_seconds, + history=history, + ) + + assert history == {} + + +def test_terminal_timeout_reconciliation_is_idempotent(): + history: dict[str, list[dict[str, object]]] = {} + first = _reconcile( + "2000-01-01T00:00:00Z", + "2000-01-01T00:00:01Z", + history=history, + ) + assert first is not None + before = list(first[1]) + + repeated = workflow_timeout_update( + RuntimeSnapshot(), + _WORKFLOW_ADDRESS, + _workflow_entry(), + {_WORKFLOW_ADDRESS: first[0]}, + {_WORKFLOW_ADDRESS: list(first[1])}, + "2000-01-01T00:00:02Z", + ) + + assert repeated is None + assert first[1] == before + + +def test_invalid_clock_is_rejected_even_when_control_plane_has_no_workflows(): + control_plane = RuntimeControlPlane(create_stub_target()) + before = control_plane._snapshot + + with pytest.raises(ValueError, match=INVALID_RECONCILIATION_CLOCK): + control_plane.reconcile_workflow_timeouts(now="not-a-timestamp") + + assert control_plane._snapshot == before + + +def test_invalid_state_timestamp_cannot_trigger_timeout_compensation(): + entry = SnapshotEntry( + address=_WORKFLOW_ADDRESS, + domain=RuntimeDomain.ORCHESTRATION, + resource_type="workflow", + payload={ + "execution_contract": { + "start_step": "run", + "timeout_seconds": 1, + "compensation_mode": "automatic", + "compensation_triggers": ["timed_out"], + "compensation_targets": {"run": "orchestration.workflow.rollback"}, + } + }, + ) + original_history = [ + { + "event_type": "step_completed", + "timestamp": "2000-01-01T00:00:00Z", + "step_name": "run", + "branch_name": None, + "join_step": None, + "outcome": "succeeded", + "details": {}, + } + ] + history = {_WORKFLOW_ADDRESS: list(original_history)} + snapshot = RuntimeSnapshot(entries={_WORKFLOW_ADDRESS: entry}) + + with pytest.raises(ValueError, match=NON_MONOTONIC_WORKFLOW_CLOCK): + workflow_timeout_update( + snapshot, + _WORKFLOW_ADDRESS, + entry, + {_WORKFLOW_ADDRESS: _running_result("2100-01-01T00:00:00Z")}, + history, + "2000-01-01T00:00:00Z", + ) + + assert history == {_WORKFLOW_ADDRESS: original_history} From 6d0e2cea2ecb8cc62b8c96da2bf94cf265049c72 Mon Sep 17 00:00:00 2001 From: Yernat Yestekov <2068106+doublewhy@users.noreply.github.com> Date: Wed, 12 Aug 2026 12:58:44 -0700 Subject: [PATCH 4/7] fix(runtime): compare workflow timeouts exactly (#1102) --- ...rkflow-timeout-reconciliation-preflight.md | 23 +++++++++++-------- .../raes_runtime/control_plane_timeouts.py | 12 ++++++---- ...runtime_workflow_timeout_reconciliation.py | 20 ++++++++++++++++ 3 files changed, 40 insertions(+), 15 deletions(-) diff --git a/docs/decisions/issue-1102-workflow-timeout-reconciliation-preflight.md b/docs/decisions/issue-1102-workflow-timeout-reconciliation-preflight.md index ac54fb746..49b41343a 100644 --- a/docs/decisions/issue-1102-workflow-timeout-reconciliation-preflight.md +++ b/docs/decisions/issue-1102-workflow-timeout-reconciliation-preflight.md @@ -16,12 +16,15 @@ Persisted timeout configuration is likewise either absent or a strictly positive integer. Invalid values cannot silently disable reconciliation or become an immediate timeout. -Timeout admission compares `now - started_at` with the configured duration. -It never compares an absolute wall-clock value directly with a duration. The -boundary remains inclusive according to the incumbent workflow state-machine -contract. Only a proven elapsed duration produces `workflow_status: timed_out` -and a `workflow_timed_out` history event. Already-terminal results are unchanged, -so reconciliation remains idempotent across restart. +Timeout admission compares the exact integral whole-second component of +`now - started_at` with the configured integer duration. It does not use +`timedelta.total_seconds()` because float rounding over large spans can turn a +timestamp just below the boundary into an early timeout, and it does not add an +unbounded duration to a timestamp because that can overflow. The boundary +remains inclusive according to the incumbent workflow state-machine contract. +Only a proven elapsed duration produces `workflow_status: timed_out` and a +`workflow_timed_out` history event. Already-terminal results are unchanged, so +reconciliation remains idempotent across restart. This is operational wall-clock timeout handling. It does not reinterpret SDL logical time, participant episode time, simulated clocks, or backend-native @@ -30,7 +33,7 @@ deadlines as interchangeable evidence. ## Verification Tests cover malformed, naive, offset, future, and non-monotonic timestamps; -invalid timeout values; exact boundaries; enormous durations; empty-control- -plane clock validation; compensation non-triggering for invalid state; and -terminal replay idempotency. The runtime timeout suite, lint, policy, and -canonical verification remain required. +invalid timeout values; exact and large-span sub-boundaries; enormous durations; +empty-control-plane clock validation; compensation non-triggering for invalid +state; and terminal replay idempotency. The runtime timeout suite, lint, policy, +and canonical verification remain required. diff --git a/implementations/python/packages/raes_runtime/control_plane_timeouts.py b/implementations/python/packages/raes_runtime/control_plane_timeouts.py index 14a29f8d6..b8774aefb 100644 --- a/implementations/python/packages/raes_runtime/control_plane_timeouts.py +++ b/implementations/python/packages/raes_runtime/control_plane_timeouts.py @@ -116,11 +116,13 @@ def _workflow_has_timed_out( raise ValueError(INVALID_WORKFLOW_TIMESTAMP) if current < started or current < updated: raise ValueError(NON_MONOTONIC_WORKFLOW_CLOCK) - # Elapsed time is compared against the timeout rather than added to the start - # instant: `timeout_seconds` has no declared upper bound, and folding a very - # large one into a float timestamp or a timedelta overflows. - elapsed_seconds = (current - started).total_seconds() - return elapsed_seconds >= timeout_seconds + # Compare exact integral seconds. ``timedelta.total_seconds()`` returns a + # float and rounds microseconds over large spans, which can synthesize an + # early timeout and compensation. Adding an unbounded timeout to ``started`` + # would instead overflow, so derive the exact whole-second component. + elapsed = current - started + elapsed_whole_seconds = elapsed.days * 86_400 + elapsed.seconds + return elapsed_whole_seconds >= timeout_seconds def _timed_out_workflow_update( diff --git a/implementations/python/tests/test_runtime_workflow_timeout_reconciliation.py b/implementations/python/tests/test_runtime_workflow_timeout_reconciliation.py index 20221a8bc..2c657bde4 100644 --- a/implementations/python/tests/test_runtime_workflow_timeout_reconciliation.py +++ b/implementations/python/tests/test_runtime_workflow_timeout_reconciliation.py @@ -185,6 +185,26 @@ def test_enormous_timeout_reports_not_timed_out_instead_of_overflowing(): assert update is None +def test_large_elapsed_span_does_not_round_up_to_an_early_timeout(): + """One microsecond below an integer deadline must remain below it. + + ``timedelta.total_seconds()`` rounds this span to + ``315537897600.0`` even though its exact whole-second component is one + second smaller. + """ + + update = workflow_timeout_update( + RuntimeSnapshot(), + _WORKFLOW_ADDRESS, + _workflow_entry_with_timeout(315_537_897_600), + {_WORKFLOW_ADDRESS: _running_result("0001-01-01T00:00:00Z")}, + {}, + "9999-12-31T23:59:59.999999Z", + ) + + assert update is None + + def test_unparseable_reconciliation_clock_is_raised_not_swallowed(): """A bad caller-supplied ``now`` governs every workflow, so it must surface. From ae0b9f8675f68748cc9d4d0276cfd1d76da00286 Mon Sep 17 00:00:00 2001 From: Yernat Yestekov <2068106+doublewhy@users.noreply.github.com> Date: Wed, 12 Aug 2026 12:58:44 -0700 Subject: [PATCH 5/7] test(runtime): cover timeout reconciliation guards --- ...runtime_workflow_timeout_reconciliation.py | 79 +++++++++++++++++++ 1 file changed, 79 insertions(+) diff --git a/implementations/python/tests/test_runtime_workflow_timeout_reconciliation.py b/implementations/python/tests/test_runtime_workflow_timeout_reconciliation.py index 2c657bde4..6a0eea0bb 100644 --- a/implementations/python/tests/test_runtime_workflow_timeout_reconciliation.py +++ b/implementations/python/tests/test_runtime_workflow_timeout_reconciliation.py @@ -2,6 +2,8 @@ from __future__ import annotations +from datetime import datetime + import pytest from raes_backend_stubs.stubs import create_stub_target from raes_contracts.planning import RuntimeDomain @@ -11,11 +13,13 @@ from raes_runtime.control_plane_timeouts import ( INVALID_RECONCILIATION_CLOCK, INVALID_TIMEOUT_CONFIGURATION, + INVALID_WORKFLOW_STATE, INVALID_WORKFLOW_TIMESTAMP, NON_MONOTONIC_WORKFLOW_CLOCK, TIMED_OUT_REASON, workflow_timeout_update, ) +from raes_runtime.control_plane_workflows import parse_timestamp _WORKFLOW_ADDRESS = "orchestration.workflow.response" @@ -220,6 +224,81 @@ def test_naive_reconciliation_clock_is_rejected(): _reconcile("2000-01-01T00:00:00Z", "2000-01-01T00:00:01") +def test_explicit_naive_reconciliation_clock_is_rejected_before_state_mutation() -> None: + history: dict[str, list[dict[str, object]]] = {} + + with pytest.raises(ValueError, match=INVALID_RECONCILIATION_CLOCK): + workflow_timeout_update( + RuntimeSnapshot(), + _WORKFLOW_ADDRESS, + _workflow_entry(), + {_WORKFLOW_ADDRESS: _running_result("2000-01-01T00:00:00Z")}, + history, + "2000-01-01T00:00:01Z", + reconciliation_clock=datetime(2000, 1, 1, 0, 0, 1), + ) + + assert history == {} + + +@pytest.mark.parametrize("persisted", [[], {"workflow_status": "running"}]) +def test_malformed_persisted_workflow_state_fails_closed(persisted: object) -> None: + with pytest.raises(ValueError, match=INVALID_WORKFLOW_STATE): + workflow_timeout_update( + RuntimeSnapshot(), + _WORKFLOW_ADDRESS, + _workflow_entry(), + {_WORKFLOW_ADDRESS: persisted}, # type: ignore[dict-item] + {}, + "2000-01-01T00:00:01Z", + ) + + +@pytest.mark.parametrize( + ("payload", "expected_error"), + [ + ([], INVALID_TIMEOUT_CONFIGURATION), + ({"execution_contract": None}, None), + ({"execution_contract": []}, INVALID_TIMEOUT_CONFIGURATION), + ], +) +def test_persisted_workflow_timeout_shape_is_validated( + payload: object, + expected_error: str | None, +) -> None: + entry = _workflow_entry() + object.__setattr__(entry, "payload", payload) + + if expected_error is None: + assert ( + workflow_timeout_update( + RuntimeSnapshot(), + _WORKFLOW_ADDRESS, + entry, + {_WORKFLOW_ADDRESS: _running_result("2000-01-01T00:00:00Z")}, + {}, + "2000-01-01T00:00:01Z", + ) + is None + ) + else: + with pytest.raises(ValueError, match=expected_error): + workflow_timeout_update( + RuntimeSnapshot(), + _WORKFLOW_ADDRESS, + entry, + {_WORKFLOW_ADDRESS: _running_result("2000-01-01T00:00:00Z")}, + {}, + "2000-01-01T00:00:01Z", + ) + + +@pytest.mark.parametrize("raw", ["", None]) +def test_timestamp_parser_rejects_empty_and_non_string_inputs(raw: object) -> None: + with pytest.raises(ValueError, match="explicit UTC offset"): + parse_timestamp(raw) # type: ignore[arg-type] + + @pytest.mark.parametrize("timeout_seconds", [-1, 0, True, 1.5, "1", "bogus"]) def test_invalid_timeout_configuration_fails_closed(timeout_seconds: object): history: dict[str, list[dict[str, object]]] = {} From 07d648be36b6e61fc39eba754d6356c78c9bb316 Mon Sep 17 00:00:00 2001 From: Yernat Yestekov <2068106+doublewhy@users.noreply.github.com> Date: Wed, 12 Aug 2026 12:58:44 -0700 Subject: [PATCH 6/7] test(runtime): cover absent timeout inputs (#1102) --- ...runtime_workflow_timeout_reconciliation.py | 31 +++++++++++++++++++ 1 file changed, 31 insertions(+) diff --git a/implementations/python/tests/test_runtime_workflow_timeout_reconciliation.py b/implementations/python/tests/test_runtime_workflow_timeout_reconciliation.py index 6a0eea0bb..d1decad1b 100644 --- a/implementations/python/tests/test_runtime_workflow_timeout_reconciliation.py +++ b/implementations/python/tests/test_runtime_workflow_timeout_reconciliation.py @@ -98,6 +98,37 @@ def test_workflow_inside_its_deadline_is_left_running(): assert _reconcile("2000-01-01T00:00:00Z", "2000-01-01T00:00:00Z") is None +def test_workflow_without_a_persisted_result_is_not_timed_out() -> None: + assert ( + workflow_timeout_update( + RuntimeSnapshot(), + _WORKFLOW_ADDRESS, + _workflow_entry(), + {}, + {}, + "2000-01-01T00:00:01Z", + ) + is None + ) + + +def test_workflow_without_a_declared_timeout_is_not_timed_out() -> None: + entry = _workflow_entry() + object.__setattr__(entry, "payload", {"execution_contract": {}}) + + assert ( + workflow_timeout_update( + RuntimeSnapshot(), + _WORKFLOW_ADDRESS, + entry, + {_WORKFLOW_ADDRESS: _running_result("2000-01-01T00:00:00Z")}, + {}, + "2000-01-01T00:00:01Z", + ) + is None + ) + + def test_workflow_at_its_exact_deadline_is_timed_out(): update = _reconcile("2000-01-01T00:00:00Z", "2000-01-01T00:00:01Z") From 289f8eaf0a627c15dd73c3bb29611bbce1710df6 Mon Sep 17 00:00:00 2001 From: Yernat Yestekov <2068106+doublewhy@users.noreply.github.com> Date: Wed, 12 Aug 2026 13:45:31 -0700 Subject: [PATCH 7/7] refactor(runtime): simplify timeout validation diagnostics (#1102) --- .../raes_runtime/control_plane_workflows.py | 8 +++-- ...runtime_workflow_timeout_reconciliation.py | 34 +++++++++++++------ 2 files changed, 29 insertions(+), 13 deletions(-) diff --git a/implementations/python/packages/raes_runtime/control_plane_workflows.py b/implementations/python/packages/raes_runtime/control_plane_workflows.py index bce3a3c99..78b6cbbe4 100644 --- a/implementations/python/packages/raes_runtime/control_plane_workflows.py +++ b/implementations/python/packages/raes_runtime/control_plane_workflows.py @@ -13,6 +13,8 @@ WorkflowHistoryEventType, ) +_EXPLICIT_OFFSET_TIMESTAMP_ERROR = "timestamp must be an ISO-8601 value with an explicit UTC offset" + def compiled_execution_contract( snapshot: RuntimeSnapshot, @@ -189,13 +191,13 @@ def parse_timestamp(raw: str) -> datetime: """Parse one explicit-offset ISO-8601 timestamp and normalize it to UTC.""" if not isinstance(raw, str) or not raw: - raise ValueError("timestamp must be an ISO-8601 value with an explicit UTC offset") + raise ValueError(_EXPLICIT_OFFSET_TIMESTAMP_ERROR) normalized = raw[:-1] + "+00:00" if raw.endswith("Z") else raw try: parsed = datetime.fromisoformat(normalized) offset = parsed.utcoffset() except (OverflowError, TypeError, ValueError): - raise ValueError("timestamp must be an ISO-8601 value with an explicit UTC offset") from None + raise ValueError(_EXPLICIT_OFFSET_TIMESTAMP_ERROR) from None if parsed.tzinfo is None or offset is None: - raise ValueError("timestamp must be an ISO-8601 value with an explicit UTC offset") + raise ValueError(_EXPLICIT_OFFSET_TIMESTAMP_ERROR) return parsed.astimezone(UTC) diff --git a/implementations/python/tests/test_runtime_workflow_timeout_reconciliation.py b/implementations/python/tests/test_runtime_workflow_timeout_reconciliation.py index d1decad1b..88f254d30 100644 --- a/implementations/python/tests/test_runtime_workflow_timeout_reconciliation.py +++ b/implementations/python/tests/test_runtime_workflow_timeout_reconciliation.py @@ -257,16 +257,20 @@ def test_naive_reconciliation_clock_is_rejected(): def test_explicit_naive_reconciliation_clock_is_rejected_before_state_mutation() -> None: history: dict[str, list[dict[str, object]]] = {} + snapshot = RuntimeSnapshot() + entry = _workflow_entry() + workflow_states = {_WORKFLOW_ADDRESS: _running_result("2000-01-01T00:00:00Z")} + reconciliation_clock = datetime(2000, 1, 1, 0, 0, 1) with pytest.raises(ValueError, match=INVALID_RECONCILIATION_CLOCK): workflow_timeout_update( - RuntimeSnapshot(), + snapshot, _WORKFLOW_ADDRESS, - _workflow_entry(), - {_WORKFLOW_ADDRESS: _running_result("2000-01-01T00:00:00Z")}, + entry, + workflow_states, history, "2000-01-01T00:00:01Z", - reconciliation_clock=datetime(2000, 1, 1, 0, 0, 1), + reconciliation_clock=reconciliation_clock, ) assert history == {} @@ -274,12 +278,18 @@ def test_explicit_naive_reconciliation_clock_is_rejected_before_state_mutation() @pytest.mark.parametrize("persisted", [[], {"workflow_status": "running"}]) def test_malformed_persisted_workflow_state_fails_closed(persisted: object) -> None: + snapshot = RuntimeSnapshot() + entry = _workflow_entry() + workflow_states: dict[str, dict[str, object]] = { + _WORKFLOW_ADDRESS: persisted, # type: ignore[dict-item] + } + with pytest.raises(ValueError, match=INVALID_WORKFLOW_STATE): workflow_timeout_update( - RuntimeSnapshot(), + snapshot, _WORKFLOW_ADDRESS, - _workflow_entry(), - {_WORKFLOW_ADDRESS: persisted}, # type: ignore[dict-item] + entry, + workflow_states, {}, "2000-01-01T00:00:01Z", ) @@ -313,12 +323,15 @@ def test_persisted_workflow_timeout_shape_is_validated( is None ) else: + snapshot = RuntimeSnapshot() + workflow_states = {_WORKFLOW_ADDRESS: _running_result("2000-01-01T00:00:00Z")} + with pytest.raises(ValueError, match=expected_error): workflow_timeout_update( - RuntimeSnapshot(), + snapshot, _WORKFLOW_ADDRESS, entry, - {_WORKFLOW_ADDRESS: _running_result("2000-01-01T00:00:00Z")}, + workflow_states, {}, "2000-01-01T00:00:01Z", ) @@ -406,13 +419,14 @@ def test_invalid_state_timestamp_cannot_trigger_timeout_compensation(): ] history = {_WORKFLOW_ADDRESS: list(original_history)} snapshot = RuntimeSnapshot(entries={_WORKFLOW_ADDRESS: entry}) + workflow_states = {_WORKFLOW_ADDRESS: _running_result("2100-01-01T00:00:00Z")} with pytest.raises(ValueError, match=NON_MONOTONIC_WORKFLOW_CLOCK): workflow_timeout_update( snapshot, _WORKFLOW_ADDRESS, entry, - {_WORKFLOW_ADDRESS: _running_result("2100-01-01T00:00:00Z")}, + workflow_states, history, "2000-01-01T00:00:00Z", )