Sort scheduler reconciliation by UTC timestamp

Author: joeszilagyi

tests/test_scheduler_failure_state_reconciliation.py

@@ -566,6 +566,48 @@ def test_unknown_terminal_status_does_not_count_as_success(tmp_path: Path) -> No
     assert entry["recommendation"] == "replace"
 
 
+def test_runtime_ledger_and_outcomes_sort_by_normalized_timestamp(
+    tmp_path: Path,
+) -> None:
+    ledger_path = tmp_path / "runtime" / "ledgers" / "workspace-a.runtime-ledger.jsonl"
+    append_ledger_events(
+        ledger_path,
+        [
+            build_success_event(
+                workspace_id="workspace-a",
+                run_id="success-run",
+                occurred_at="2026-06-02T00:00:00Z",
+            ),
+            build_failure_event(
+                workspace_id="workspace-a",
+                run_id="failure-run",
+                occurred_at="2026-06-01T23:30:00-01:00",
+                message="later failure",
+            ),
+        ],
+    )
+
+    events = scheduler_reconciliation.read_runtime_ledger(ledger_path, workspace_id="workspace-a")
+    assert [event["run_id"] for event in events] == ["success-run", "failure-run"]
+
+    outcomes = scheduler_reconciliation.summarize_run_outcomes(events)
+    assert [outcome.run_id for outcome in outcomes] == ["success-run", "failure-run"]
+
+    derived, reasons, _ = scheduler_reconciliation.derive_failure_state(
+        current_failure_state=None,
+        run_budget=None,
+        retry_policy=None,
+        events=events,
+    )
+    assert derived == {
+        "status": "retryable",
+        "attempt_count": 1,
+        "last_failure_at": "2026-06-01T23:30:00-01:00",
+        "last_failure_reason": "later failure",
+    }
+    assert reasons == ["1 consecutive terminal runtime failure(s) since the last success"]
+
+
 def test_reconciliation_keeps_current_state_without_terminal_runs(tmp_path: Path) -> None:
     workspace_root = tmp_path / "workspace"
     workspace_root.mkdir()

tools/common/scheduler_failure_reconciliation.py

@@ -42,6 +42,10 @@ def parse_timestamp(raw_value: str, *, label: str) -> datetime:
     return parsed.astimezone(UTC)
 
 
+def occurrence_sort_key(raw_value: str, *, label: str, tie_breaker: str) -> tuple[datetime, str]:
+    return parse_timestamp(raw_value, label=label), tie_breaker
+
+
 def read_runtime_ledger(path: Path, *, workspace_id: str) -> list[dict[str, Any]]:
     if not path.exists():
         return []
@@ -64,7 +68,13 @@ def read_runtime_ledger(path: Path, *, workspace_id: str) -> list[dict[str, Any]
             )
         parse_timestamp(occurred_at, label=f"{path}:{line_number}:occurred_at")
         events.append(payload)
-    events.sort(key=lambda event: (event["occurred_at"], str(event.get("event_id", ""))))
+    events.sort(
+        key=lambda event: occurrence_sort_key(
+            str(event["occurred_at"]),
+            label=f"{path}:occurred_at",
+            tie_breaker=str(event.get("event_id", "")),
+        )
+    )
     return events
 
 
@@ -84,7 +94,11 @@ def summarize_run_outcomes(events: list[dict[str, Any]]) -> list[RunOutcome]:
         if not terminal_events:
             continue
         terminal_events.sort(
-            key=lambda event: (event["occurred_at"], str(event.get("event_id", "")))
+            key=lambda event: occurrence_sort_key(
+                str(event["occurred_at"]),
+                label="terminal runtime-ledger occurred_at",
+                tie_breaker=str(event.get("event_id", "")),
+            )
         )
         last_event = terminal_events[-1]
         event_type = last_event.get("event_type")
@@ -129,7 +143,13 @@ def summarize_run_outcomes(events: list[dict[str, Any]]) -> list[RunOutcome]:
             )
         )
 
-    outcomes.sort(key=lambda outcome: (outcome.occurred_at, outcome.run_id))
+    outcomes.sort(
+        key=lambda outcome: occurrence_sort_key(
+            outcome.occurred_at,
+            label="runtime-ledger outcome occurred_at",
+            tie_breaker=outcome.run_id,
+        )
+    )
     return outcomes