fix(workflows): reject mismatched run state IDs

src/specify_cli/workflows/engine.py

@@ -732,6 +732,11 @@ def load(cls, run_id: str, project_root: Path) -> RunState:
                 "Invalid run state: missing required field(s): "
                 + ", ".join(missing_fields)
             )
+        if state_data["run_id"] != run_id:
+            raise ValueError(
+                f"Invalid run state: stored run_id {state_data['run_id']!r} "
+                f"does not match requested run_id {run_id!r}"
+            )
 
         workflow_id = state_data["workflow_id"]
         if not isinstance(workflow_id, str) or not _ID_PATTERN.fullmatch(

tests/test_workflows.py

@@ -6768,6 +6768,35 @@ def test_load_not_found(self, project_dir):
         with pytest.raises(FileNotFoundError):
             RunState.load("nonexistent", project_dir)
 
+    def test_load_rejects_stored_run_id_mismatch(self, project_dir):
+        """The state payload cannot redirect later writes to another run."""
+        from specify_cli.workflows.engine import RunState
+
+        run_dir = (
+            project_dir
+            / ".specify"
+            / "workflows"
+            / "runs"
+            / "requested-run"
+        )
+        run_dir.mkdir(parents=True)
+        (run_dir / "state.json").write_text(
+            json.dumps(
+                {
+                    "run_id": "other-run",
+                    "workflow_id": "test-workflow",
+                    "status": "created",
+                }
+            ),
+            encoding="utf-8",
+        )
+
+        with pytest.raises(
+            ValueError,
+            match="stored run_id 'other-run' does not match requested run_id 'requested-run'",
+        ):
+            RunState.load("requested-run", project_dir)
+
     @pytest.mark.parametrize(
         ("installed_workflow_id", "installed_registry_root"),
         [