From 7e92f5a7d750629b7be78c275b2fe913f9c09502 Mon Sep 17 00:00:00 2001
From: Raphael Malikian <rtmalikian@gmail.com>
Date: Wed, 7 Oct 2026 11:15:17 -0700
Subject: [PATCH] Fix StorageTask listing failing on tagged storage inputs
 (#1348)

---
 src/python/txtai/workflow/task/storage.py |  4 +--
 test/python/testworkflow.py               | 30 +++++++++++++++++++++++
 2 files changed, 32 insertions(+), 2 deletions(-)

diff --git a/src/python/txtai/workflow/task/storage.py b/src/python/txtai/workflow/task/storage.py
index 24ca8fd5..4b9ebced 100644
--- a/src/python/txtai/workflow/task/storage.py
+++ b/src/python/txtai/workflow/task/storage.py
@@ -54,8 +54,8 @@ class StorageTask(Task):
         outputs = []
         for element in elements:
             if self.matches(element):
-                # Get directory listing and run actions
-                outputs.extend(super().__call__(self.list(element), executor))
+                # Unpack tagged input, list container and re-tag each listing to match input format
+                outputs.extend(self.pack(element, result) for result in super().__call__(self.list(self.upack(element, True)), executor))
             else:
                 outputs.append(element)
 
diff --git a/test/python/testworkflow.py b/test/python/testworkflow.py
index 0d754742..14546f87 100644
--- a/test/python/testworkflow.py
+++ b/test/python/testworkflow.py
@@ -524,6 +524,36 @@ class TestWorkflow(unittest.TestCase):
 
         self.assertEqual(len(results), 22)
 
+    def testStorageTaggedWorkflow(self):
+        """
+        Test a storage task with tagged (id, data, tag) inputs (#1328)
+        """
+
+        task = StorageTask()
+
+        # matches() accepts untagged and tagged storage urls
+        self.assertIsNotNone(task.matches("local://" + Utils.PATH))
+        self.assertIsNotNone(task.matches(("id", "local://" + Utils.PATH, "tag")))
+        self.assertIsNone(task.matches("test string"))
+
+        workflow = Workflow([StorageTask()])
+
+        # Tagged inputs list the container and re-tag each file with the input (id, tag)
+        url = "local://" + Utils.PATH
+        results = list(workflow([("id", url, "tag")]))
+        self.assertEqual(len(results), 21)
+        for result in results:
+            self.assertEqual(len(result), 3)
+            self.assertEqual(result[0], "id")
+            # normpath: libcloud joins listing paths with os.path.join, which uses
+            # backslashes on Windows while Utils.PATH uses forward slashes
+            self.assertTrue(os.path.normpath(result[1]).startswith(os.path.normpath(Utils.PATH)))
+            self.assertEqual(result[2], "tag")
+
+        # Non-storage inputs pass through unmodified
+        results = list(workflow([("id2", "test string", "tag2")]))
+        self.assertEqual(results, [("id2", "test string", "tag2")])
+
     def testTemplateInput(self):
         """
         Test template task input