refactor: minor refactor

Signed-off-by: Noppanat Wadlom <noppanat.wad@gmail.com>

Author: kaiitunnz

docs/SERVICE_RESTARTS.md

@@ -54,9 +54,8 @@ restart safe:
   A transition's task records, workflow status-set membership, and schedule
   snapshot are written as a single atomic Redis transaction
   (`WorkflowRegistry.commit_transition`), so a crash mid-persist commits the whole
-  transition or none of it — never a half-applied state. Event-driven transitions
-  are additionally healed by replay; the API-driven workflow cancel, which has no
-  event to replay, relies on this atomicity alone.
+  transition or none of it. Event-driven transitions are additionally healed by replay;
+  the API-driven workflow cancel relies on this atomicity alone.
 - **Replayable task events.** Task lifecycle events flow through a durable Redis
   stream consumed from a persisted cursor. The ordering is what makes replay
   safe: a transition is written to durable scheduler state *before* its event is

src/server/task/runtime.py

@@ -371,19 +371,19 @@ def _persisted_task_locked(self, task_id: str) -> PersistedTask | None:
             epoch_index=self._task_epoch_index.get(task_id),
         )
 
-    def _sched_locked(self, workflow_id: str) -> WorkflowSched:
-        return WorkflowSched(
-            in_epoch_order=self._workflow_in_epoch_order.get(workflow_id, False),
-            epoch_frontier=self._workflow_epoch_frontier.get(workflow_id, 0),
-        )
-
     def _records_locked(self, *task_ids: str) -> list[PersistedTask]:
         return [
             persisted
             for task_id in dict.fromkeys(task_ids)
             if (persisted := self._persisted_task_locked(task_id))
         ]
 
+    def _sched_locked(self, workflow_id: str) -> WorkflowSched:
+        return WorkflowSched(
+            in_epoch_order=self._workflow_in_epoch_order.get(workflow_id, False),
+            epoch_frontier=self._workflow_epoch_frontier.get(workflow_id, 0),
+        )
+
     def _persist_locked(self, *task_ids: str) -> None:
         """Commit task records (no membership change) atomically, per workflow."""
         by_workflow: dict[str, list[str]] = defaultdict(list)