UiPath · andreitava-uip · Aug 13, 2026
@@ -1,6 +1,6 @@
 [project]
 name = "uipath-runtime"
-version = "0.13.1"
+version = "0.13.2"
 description = "Runtime abstractions and interfaces for building agents and automation scripts in the UiPath ecosystem"
 readme = { file = "README.md", content-type = "text/markdown" }
 requires-python = ">=3.11"

@@ -2,6 +2,7 @@
 
 import hashlib
 import os
+from collections.abc import Iterable
 from datetime import datetime, timezone
 from pathlib import Path
 from typing import Any
@@ -126,8 +127,6 @@ async def dehydrate(
 
             if existing and existing.sha256 == digest and existing.size == size:
                 current[virtual_path] = existing
-                if self.current_job_key:
-                    await self.link_attachment(existing.attachment_key)
                 continue
 
             attachment_name = self._attachment_name_for_virtual_path(virtual_path)
@@ -146,22 +145,44 @@ async def dehydrate(
             )
             current[virtual_path] = entry
 
-            if self.current_job_key:
-                await self.link_attachment(entry.attachment_key)
+        await self._link_attachments(entry.attachment_key for entry in current.values())
 
         return self._dump_registry(current)
 
     async def link_attachment(self, attachment_key: str) -> None:
-        """Link an already uploaded attachment to the current job."""
+        """Link an attachment unless it is already associated with the current job."""
+        await self._link_attachments((attachment_key,))
+
+    async def _link_attachments(self, attachment_keys: Iterable[str]) -> None:
         if self.jobs is None or not self.current_job_key:
             return
 
-        await self.jobs.link_attachment_async(
-            job_key=UUID(self.current_job_key),
-            attachment_key=UUID(attachment_key),
-            folder_key=self.folder_key,
-            folder_path=self.folder_path,
+        requested_attachment_keys = list(
+            dict.fromkeys(UUID(key) for key in attachment_keys)
         )
+        if not requested_attachment_keys:
+            return
+
+        job_key = UUID(self.current_job_key)
+        linked_attachment_keys = {
+            UUID(key)
+            for key in await self.jobs.list_attachments_async(
+                job_key=job_key,
+                folder_key=self.folder_key,
+                folder_path=self.folder_path,
+            )
+        }
+        for attachment_key in requested_attachment_keys:
+            if attachment_key in linked_attachment_keys:
+                continue
+
+            await self.jobs.link_attachment_async(
+                job_key=job_key,
+                attachment_key=attachment_key,
+                folder_key=self.folder_key,
+                folder_path=self.folder_path,
+            )
+            linked_attachment_keys.add(attachment_key)
 
     def _iter_files(self) -> list[Path]:
         if not self.workspace_path.exists():

@@ -117,6 +117,10 @@ async def link_attachment_async(
         folder_key: str | None = None,
         folder_path: str | None = None,
     ) -> None:
+        linked_attachments = self.attachments.setdefault(str(job_key), [])
+        if str(attachment_key) in linked_attachments:
+            raise RuntimeError("The association already exists")
+        linked_attachments.append(str(attachment_key))
         self.links.append((job_key, attachment_key))
 
 
@@ -932,6 +936,39 @@ async def test_dehydrate_relinks_unchanged_file_without_reupload(
     assert jobs.links == [(current_job, key)]
 
 
+@pytest.mark.asyncio
+async def test_dehydrate_does_not_relink_attachment_already_linked_to_job(
+    tmp_path: Path,
+) -> None:
+    workspace = Workspace.create(tmp_path / "workspace")
+    attachments = FakeAttachments()
+    jobs = FakeJobs()
+    current_job = uuid.uuid4()
+    (workspace.path / "notes.txt").write_text("same", encoding="utf-8")
+
+    first_hydrator = WorkspaceHydrator(
+        workspace_path=workspace.path,
+        attachments=attachments,
+        jobs=jobs,
+        current_job_key=str(current_job),
+    )
+    registry = await first_hydrator.dehydrate({})
+
+    resumed_hydrator = WorkspaceHydrator(
+        workspace_path=workspace.path,
+        attachments=attachments,
+        jobs=jobs,
+        current_job_key=str(current_job),
+    )
+    result = await resumed_hydrator.dehydrate(registry)
+
+    assert result == registry
+    assert attachments.uploads == 1
+    assert jobs.links == [
+        (current_job, uuid.UUID(registry["notes.txt"]["attachment_key"]))
+    ]
+
+
 @pytest.mark.asyncio
 async def test_attachment_names_are_single_segment_for_nested_files(
     tmp_path: Path,