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 00000000..49b41343 --- /dev/null +++ b/docs/decisions/issue-1102-workflow-timeout-reconciliation-preflight.md @@ -0,0 +1,39 @@ +# 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 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 +deadlines as interchangeable evidence. + +## Verification + +Tests cover malformed, naive, offset, future, and non-monotonic timestamps; +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/docs/requirements/RUN-317/requirement.md b/docs/requirements/RUN-317/requirement.md index cd509134..3d265a93 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 df318254..1b466ba5 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 6d9f0dfc..b8774aef 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 ( @@ -14,6 +16,13 @@ from .control_plane_workflows import maybe_apply_compensation, parse_timestamp +TIMED_OUT_REASON = "workflow timed out" +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( snapshot: RuntimeSnapshot, @@ -22,15 +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)) - if ( - timeout_seconds is not None - and normalized is not None - and _workflow_has_timed_out(normalized, timeout_seconds, submitted_at) - ): + timed_out = False + if timeout_seconds is not None and normalized is not None: + timed_out = _workflow_has_timed_out(normalized, timeout_seconds, current) + if timed_out: update = _timed_out_workflow_update( snapshot, workflow_address, @@ -42,6 +55,13 @@ def workflow_timeout_update( 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": @@ -50,44 +70,59 @@ 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_has_timed_out( normalized: WorkflowExecutionState, timeout_seconds: int, - submitted_at: str, + current: datetime, ) -> bool: + """Return whether elapsed wall time proves the declared timeout.""" + try: - deadline = parse_timestamp(normalized.started_at).timestamp() + timeout_seconds - current = parse_timestamp(submitted_at).timestamp() - except Exception: - return False - return current >= deadline + started = parse_timestamp(normalized.started_at) + 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) + # 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( @@ -126,7 +161,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=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 8d16ddd8..1cdc1cd8 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 951204c5..78b6cbbe 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, @@ -186,9 +188,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(_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(_EXPLICIT_OFFSET_TIMESTAMP_ERROR) from None + if parsed.tzinfo is None or offset is None: + 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 new file mode 100644 index 00000000..88f254d3 --- /dev/null +++ b/implementations/python/tests/test_runtime_workflow_timeout_reconciliation.py @@ -0,0 +1,434 @@ +"""Workflow timeout reconciliation edge cases for the runtime control plane.""" + +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 +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_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" + + +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, + domain=RuntimeDomain.ORCHESTRATION, + resource_type="workflow", + payload={"execution_contract": {"timeout_seconds": timeout_seconds}}, + ) + + +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 + 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 + payload["updated_at"] = updated_at if updated_at is not None else started_at + return payload + + +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, + entry, + {_WORKFLOW_ADDRESS: _running_result(started_at, updated_at)}, + history if history is not None else {}, + 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 + + +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") + + 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]["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(): + """`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_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. + + Reported as ``ValueError``; the HTTP adapter maps that to 409 rather than + silently disabling timeouts for the whole pass. + """ + 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") + + +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( + snapshot, + _WORKFLOW_ADDRESS, + entry, + workflow_states, + history, + "2000-01-01T00:00:01Z", + reconciliation_clock=reconciliation_clock, + ) + + assert history == {} + + +@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( + snapshot, + _WORKFLOW_ADDRESS, + entry, + workflow_states, + {}, + "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: + snapshot = RuntimeSnapshot() + workflow_states = {_WORKFLOW_ADDRESS: _running_result("2000-01-01T00:00:00Z")} + + with pytest.raises(ValueError, match=expected_error): + workflow_timeout_update( + snapshot, + _WORKFLOW_ADDRESS, + entry, + workflow_states, + {}, + "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]]] = {} + + 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}) + 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_states, + history, + "2000-01-01T00:00:00Z", + ) + + assert history == {_WORKFLOW_ADDRESS: original_history}