fix(runtime): reconcile workflow timeouts fail closed

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.

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)

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)

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,

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

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)