From 6f8e63eb160ada7484c79f658349a89e6ea8f24d Mon Sep 17 00:00:00 2001 From: lda Date: Thu, 10 Sep 2026 05:26:48 +0700 Subject: [PATCH] refactor: reuse resume run store --- src/wf_api/runs.py | 20 +++++--------------- 1 file changed, 5 insertions(+), 15 deletions(-) diff --git a/src/wf_api/runs.py b/src/wf_api/runs.py index f6eb67fc..2208ab71 100644 --- a/src/wf_api/runs.py +++ b/src/wf_api/runs.py @@ -2,10 +2,12 @@ from __future__ import annotations import asyncio from dataclasses import asdict +from datetime import UTC, datetime from typing import Any, Protocol from wf_artifacts import ( DependencyDiagnostic, + ResumeAttempt, RunStore, StoredRunStatus, WorkflowArtifact, @@ -266,7 +268,7 @@ class WorkflowRunApi: ) if has_blocking_diagnostics(diagnostics): blocked = mark_resume_blocked( - store=self._run_store(), + store=store, record=record, diagnostics=diagnostics, ) @@ -291,20 +293,8 @@ class WorkflowRunApi: # the ACTIVE marker so recovery fails closed instead of presenting # the old checkpoint as safe to retry. Every stopped result echoes # the attempt identity back for matching. - from datetime import UTC as _UTC - from datetime import datetime as _datetime - - from wf_artifacts.runs.models import ResumeAttempt - - store = self._run_store() - existing_attempt = store.get_resume_attempt(run_id) - if existing_attempt is not None and existing_attempt.state == "ACTIVE": - raise ValueError( - f"workflow run {run_id!r} has an ambiguous active resume attempt; " - "recovery must fail it closed before retry" - ) attempt_id = store.allocate_resume_attempt_id() - now_marker = _datetime.now(_UTC) + now_marker = datetime.now(UTC) store.save_resume_attempt( ResumeAttempt( run_id=run_id, @@ -335,7 +325,7 @@ class WorkflowRunApi: run_id=run_id, attempt_id=attempt_id, ) - cleared_at = _datetime.now(_UTC) + cleared_at = datetime.now(UTC) store.save_resume_attempt( ResumeAttempt( run_id=run_id,