修复前端问题

supply_infra/pipeline/run_service.py

@@ -273,30 +273,41 @@ def resume_pipeline_run(run_id: str, *, step_key: str | None = None) -> bool:
             return False
         if run.status not in {
             RunStatus.FAILED.value,
+            RunStatus.CANCELLED.value,
             RunStatus.DEADLINE_EXCEEDED.value,
         }:
             return False
         latest = step_repo.latest_attempts(run_id)
-        failed = next(
+        resumable_statuses = (
+            {StepStatus.CANCELLED.value}
+            if run.status == RunStatus.CANCELLED.value
+            else {
+                StepStatus.FAILED.value,
+                StepStatus.TIMED_OUT.value,
+                StepStatus.BLOCKED.value,
+            }
+        )
+        resume_from = next(
             (
                 item
                 for item in sorted(latest.values(), key=lambda row: row.step_order)
-                if item.status
-                in {
-                    StepStatus.FAILED.value,
-                    StepStatus.TIMED_OUT.value,
-                    StepStatus.BLOCKED.value,
-                }
+                if item.status in resumable_statuses
             ),
             None,
         )
-        if failed is None:
+        if resume_from is None:
             return False
-        if step_key is not None and failed.step_key != step_key:
+        if step_key is not None and resume_from.step_key != step_key:
             return False
-        retry = step_repo.create_retry_attempt(failed, status=StepStatus.READY.value)
+        retry = step_repo.create_retry_attempt(
+            resume_from,
+            status=StepStatus.READY.value,
+        )
         for item in latest.values():
-            if item.step_order > failed.step_order and item.status == StepStatus.BLOCKED.value:
+            if item.step_order > resume_from.step_order and item.status in {
+                StepStatus.BLOCKED.value,
+                StepStatus.CANCELLED.value,
+            }:
                 item.status = StepStatus.PENDING.value
                 item.error_code = None
                 item.error_message = None
@@ -305,6 +316,7 @@ def resume_pipeline_run(run_id: str, *, step_key: str | None = None) -> bool:
         if run.deadline_at is not None and run.deadline_at <= utc_now():
             run.deadline_at = next_schedule_utc()
         run.finished_at = None
+        run.summary_json = None
         run.error_code = None
         run.error_message = None
         return True

tests/supply_infra/pipeline/test_resume_cancelled_run.py

@@ -0,0 +1,153 @@
+from __future__ import annotations
+
+from contextlib import contextmanager
+
+from sqlalchemy import create_engine
+from sqlalchemy.orm import Session, sessionmaker
+
+from supply_infra.db.models.pipeline_run import PipelineRun
+from supply_infra.db.models.pipeline_step_run import PipelineStepRun
+from supply_infra.pipeline import run_service
+
+
+def _session_factory() -> sessionmaker[Session]:
+    engine = create_engine("sqlite+pysqlite:///:memory:")
+    PipelineRun.__table__.create(engine)
+    PipelineStepRun.__table__.create(engine)
+    return sessionmaker(bind=engine, expire_on_commit=False)
+
+
+def _run(status: str = "cancelled") -> PipelineRun:
+    return PipelineRun(
+        run_id="run-cancelled",
+        dedupe_key="supply_pipeline:20260727:full",
+        pipeline_key="supply_pipeline",
+        biz_dt="20260727",
+        trigger_type="api",
+        run_mode="full",
+        dry_run=True,
+        status=status,
+        current_step=None,
+        deadline_at=None,
+        config_snapshot_json={},
+        date_snapshot_json={},
+        summary_json={"old": "summary"},
+        error_code="cancellation_requested",
+        error_message="cancelled",
+    )
+
+
+def _step(
+    *,
+    step_run_id: str,
+    key: str,
+    order: int,
+    status: str,
+) -> PipelineStepRun:
+    return PipelineStepRun(
+        step_run_id=step_run_id,
+        run_id="run-cancelled",
+        step_key=key,
+        step_order=order,
+        attempt=1,
+        status=status,
+        critical=True,
+        dependency_snapshot_json=[],
+        input_snapshot_json={},
+        timeout_seconds=60,
+        max_attempts=1,
+        retryable=False,
+        error_code="cancelled" if status == "cancelled" else None,
+        error_message="Cancellation requested" if status == "cancelled" else None,
+    )
+
+
+def _patch_session(monkeypatch, factory: sessionmaker[Session]) -> None:
+    @contextmanager
+    def get_test_session():
+        session = factory()
+        try:
+            yield session
+            session.commit()
+        except Exception:
+            session.rollback()
+            raise
+        finally:
+            session.close()
+
+    monkeypatch.setattr(run_service, "get_session", get_test_session)
+
+
+def test_cancelled_run_resumes_from_first_cancelled_step(monkeypatch) -> None:
+    factory = _session_factory()
+    _patch_session(monkeypatch, factory)
+    with factory() as session:
+        session.add_all(
+            [
+                _run(),
+                _step(
+                    step_run_id="step-1",
+                    key="global_tree_sync",
+                    order=1,
+                    status="succeeded",
+                ),
+                _step(
+                    step_run_id="step-2",
+                    key="source_video_sync",
+                    order=2,
+                    status="cancelled",
+                ),
+                _step(
+                    step_run_id="step-3",
+                    key="demand_grade",
+                    order=3,
+                    status="cancelled",
+                ),
+            ]
+        )
+        session.commit()
+
+    assert run_service.resume_pipeline_run("run-cancelled") is True
+
+    with factory() as session:
+        run = session.get(PipelineRun, "run-cancelled")
+        assert run is not None
+        assert run.status == "queued"
+        assert run.current_step == "source_video_sync"
+        assert run.finished_at is None
+        assert run.summary_json is None
+        assert run.error_code is None
+        assert run.error_message is None
+
+        steps = (
+            session.query(PipelineStepRun)
+            .filter(PipelineStepRun.run_id == run.run_id)
+            .order_by(PipelineStepRun.step_order, PipelineStepRun.attempt)
+            .all()
+        )
+        assert [(step.step_key, step.attempt, step.status) for step in steps] == [
+            ("global_tree_sync", 1, "succeeded"),
+            ("source_video_sync", 1, "cancelled"),
+            ("source_video_sync", 2, "ready"),
+            ("demand_grade", 1, "pending"),
+        ]
+
+
+def test_cancelling_run_cannot_resume_before_cancellation_finishes(monkeypatch) -> None:
+    factory = _session_factory()
+    _patch_session(monkeypatch, factory)
+    with factory() as session:
+        session.add_all(
+            [
+                _run(status="cancelling"),
+                _step(
+                    step_run_id="step-1",
+                    key="source_video_sync",
+                    order=1,
+                    status="running",
+                ),
+            ]
+        )
+        session.commit()
+
+    assert run_service.resume_pipeline_run("run-cancelled") is False

web/src/components/pipeline/PipelineRunActions.vue

@@ -8,10 +8,10 @@ defineEmits<{ resume: []; cancel: [] }>()
 <template>
   <div class="pipeline-actions">
     <button
-      :disabled="busy || !['failed', 'deadline_exceeded'].includes(run.status)"
+      :disabled="busy || !['failed', 'deadline_exceeded', 'cancelled'].includes(run.status)"
       @click="$emit('resume')"
     >
-      从失败点恢复
+      {{ run.status === 'cancelled' ? '从取消点重新执行' : '从失败点恢复' }}
     </button>
     <button
       :disabled="busy || ['succeeded', 'failed', 'deadline_exceeded', 'cancelling', 'cancelled'].includes(run.status)"