Fixes #791 Fixed a bug with non-idempotent item creation retries causing 'already in use' error

src/fabric_cicd/_common/_fabric_endpoint.py

@@ -293,7 +293,7 @@ def _handle_response(
         msg = f"The executing identity is not authorized to call {method} on '{url}'."
         raise Exception(msg)
 
-    # Handle item name conflicts
+    # Handle item name conflicts (temporarily reserved)
     elif (
         response.status_code == 400
         and response.headers.get("x-ms-public-api-error-code") == "ItemDisplayNameNotAvailableYet"
@@ -307,6 +307,16 @@ def _handle_response(
             prepend_message="Item name is reserved.",
         )
 
+    # Handle item name already exists (permanent conflict - item exists in workspace)
+    elif (
+        response.status_code == 400
+        and response.headers.get("x-ms-public-api-error-code") == "ItemDisplayNameAlreadyInUse"
+    ):
+        response_json = response.json() if response.text else {}
+        item_name = response_json.get("message", "").replace("Requested '", "").split("'")[0] if response_json else ""
+        msg = f"Item '{item_name}' already exists in the workspace but was not found during initial scan. "
+        raise Exception(msg)
+
     # Handle scenario where library removed from environment before being removed from repo
     elif response.status_code == 400 and "is not present in the environment." in response.json().get(
         "message", "No message provided"

src/fabric_cicd/fabric_workspace.py

@@ -656,14 +656,57 @@ def _publish_item(
 
             # Create a new item if it does not exist
             # https://learn.microsoft.com/en-us/rest/api/fabric/core/items/create-item
-            item_create_response = self.endpoint.invoke(
-                method="POST", url=f"{self.base_api_url}/items", body=combined_body
-            )
-            api_response = item_create_response
-            item_guid = item_create_response["body"]["id"]
-            self.repository_items[item_type][item_name].guid = item_guid
+            try:
+                item_create_response = self.endpoint.invoke(
+                    method="POST", url=f"{self.base_api_url}/items", body=combined_body
+                )
+                api_response = item_create_response
+                item_guid = item_create_response["body"]["id"]
+                self.repository_items[item_type][item_name].guid = item_guid
+            except Exception as e:
+                # Handle race condition: item may have been created during a throttled retry
+                # or exists due to stale cache from API throttling delays during deployment.
+                # Check for both the error message and the specific API error code.
+                error_str = str(e).lower()
+                if "already in use" in error_str or "itemdisplaynamealreadyinuse" in error_str:
+                    logger.warning(
+                        f"Item '{item_name}' already exists (possible throttling race condition). "
+                        "Attempting to recover by fetching current state."
+                    )
+                    # Re-fetch the item's GUID from the workspace using existing lookup function
+                    try:
+                        item_guid = self._lookup_item_attribute(self.workspace_id, item_type, item_name, "id")
+                    except InputError:
+                        item_guid = None
+                    if item_guid:
+                        self.repository_items[item_type][item_name].guid = item_guid
+                        is_deployed = True
+                        # Update deployed_items cache to ensure folder move logic works correctly
+                        if item_type not in self.deployed_items:
+                            self.deployed_items[item_type] = {}
+                        self.deployed_items[item_type][item_name] = Item(
+                            type=item_type,
+                            name=item_name,
+                            description=item.description,
+                            guid=item_guid,
+                            folder_id="",  # Unknown at this point, folder move logic will handle if needed
+                        )
+                        # Set api_response for response tracking to indicate recovery occurred
+                        api_response = {
+                            "recovered": True,
+                            "body": {"id": item_guid, "displayName": item_name, "type": item_type},
+                            "status_code": 200,
+                            "header": {},
+                        }
+                        logger.info(
+                            f"{constants.INDENT}Recovered item GUID: {item_guid}. Will update instead of create."
+                        )
+                    else:
+                        raise  # Re-raise if we couldn't recover
+                else:
+                    raise  # Re-raise for other errors
 
-        elif is_deployed and not shell_only_publish:
+        if is_deployed and not shell_only_publish:
             # Update the item's definition if full publish is required
             # https://learn.microsoft.com/en-us/rest/api/fabric/core/items/update-item-definition
             update_response = self.endpoint.invoke(

tests/test__fabric_endpoint.py

@@ -476,6 +476,31 @@ def test_handle_response_item_display_name_already_in_use(setup_mocks, monkeypat
     assert dl.messages == [expected]
 
 
+def test_handle_response_item_display_name_already_in_use_permanent(setup_mocks):
+    """
+    Test _handle_response raises an exception when item display name is already in use (permanent conflict).
+
+    This error code indicates the item already exists in the workspace, not a temporary reservation.
+    """
+    _, _mock_requests = setup_mocks
+    response = Mock(
+        status_code=400,
+        headers={"x-ms-public-api-error-code": "ItemDisplayNameAlreadyInUse", "Content-Type": "application/json"},
+        text='{"message": "Requested \'Test Item\' is already in use."}',
+    )
+    response.json.return_value = {"message": "Requested 'Test Item' is already in use."}
+
+    with pytest.raises(Exception, match="already exists in the workspace"):
+        _handle_response(
+            response=response,
+            method="POST",
+            url="http://example.com/items",
+            body="{}",
+            long_running=False,
+            iteration_count=1,
+        )
+
+
 def test_handle_response_environment_libraries_not_found(setup_mocks):
     """Test _handle_response exits loop when environment libraries are not found (404)."""
     _, _mock_requests = setup_mocks