Discriminate input from output in staging manifest

pulsar/client/action_mapper.py

@@ -536,20 +536,24 @@ class JsonTransferAction(BaseAction):
     def __init__(self, source, file_lister=None, url=None):
         super().__init__(source, file_lister)
         self.url = url
-        self._path = None
+        self._from_path = None
+        self._to_path = None
 
     @classmethod
     def from_dict(cls, action_dict):
         return JsonTransferAction(source=action_dict["source"], url=action_dict["url"])
 
     def write_to_path(self, path):
-        self._path = path
+        self._to_path = path
 
     def write_from_path(self, pulsar_path: str):
-        self._path = pulsar_path
+        self._from_path = pulsar_path
 
     def finalize(self):
-        return {"url": self.url, "path": self.path}
+        if self._to_path:
+            return {"url": self.url, "to_path": self._to_path}
+        else:
+            return {"url": self.url, "from_path": self._from_path}
 
 
 class RemoteObjectStoreCopyAction(BaseAction):