sched: add durable resume-attempt marker with store-backed identities (T09)

src/wf_api/runs.py

@@ -216,6 +216,34 @@ class WorkflowRunApi:
                 steps_executed=stopped_run.steps_executed,
                 steps_remaining=stopped_run.steps_remaining,
             )
+        # Durable resume-attempt marker: persist ACTIVE with a store-backed
+        # attempt identity before re-executing. A crash during resume leaves
+        # 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)
+        store.save_resume_attempt(
+            ResumeAttempt(
+                run_id=run_id,
+                attempt_id=attempt_id,
+                state="ACTIVE",
+                created_at=now_marker,
+                updated_at=now_marker,
+            )
+        )
         plan = raw_plan_from_artifact(environment.root_artifact)
         tree = saved_subgraph_tree_from_snapshots(environment.child_artifacts)
         run = await self.context.runtime.resume_workflow_from_plan(
@@ -227,11 +255,25 @@ class WorkflowRunApi:
             artifact=environment.root_artifact,
             saved_subgraph_tree=tree,
         )
+        # Granular completion: stopped persist, attempt-clear, and history
+        # are separate persists with fault boundaries between each pair. A
+        # resumed run may interrupt again (durable re-interruption).
         next_record = persist_stopped_run(
-            store=self._run_store(),
+            store=store,
             environment=environment,
             run=run,
             run_id=run_id,
+            attempt_id=attempt_id,
+        )
+        cleared_at = _datetime.now(_UTC)
+        store.save_resume_attempt(
+            ResumeAttempt(
+                run_id=run_id,
+                attempt_id=attempt_id,
+                state="DONE",
+                created_at=now_marker,
+                updated_at=cleared_at,
+            )
         )
         return _run_payload(
             deployment=environment.deployment,