@@ -44,7 +44,7 @@ def upload_image(
         Optional[bool],
         typer.Option(
             "--annotation-overwrite/--no-annotation-overwrite",
-            help="Zip flow: overwrite existing annotations on duplicate images "
+            help="Overwrite existing annotations on duplicate images "
             "(default: off, except classification projects where the API requires it on)",
         ),
     ] = None,
 
@@ -540,7 +540,7 @@ def upload_dataset(
             num_retries (int, optional): number of times to retry uploading an image if the upload fails. Defaults to 0.
             is_prediction (bool, optional): whether the annotations provided in the dataset are predictions and not ground truth. Defaults to False.
             use_zip_upload (bool, optional): opt-in to the zip flow for a directory input (the SDK zips it client-side). Ignored when dataset_path is already a `.zip`.
-            annotation_overwrite (bool, optional): zip flow only — overwrite existing annotations on duplicate images. Defaults to False, except classification projects where it defaults to True (the API requires it).
+            annotation_overwrite (bool, optional): overwrite existing annotations on duplicate images. Defaults to False, except classification projects where it defaults to True (the API requires it).
             tags (list[str], optional): zip flow only — tags to apply to the uploaded batch.
             split (str, optional): dataset split for the uploaded batch. In per-image directory
                 uploads, this overrides inferred splits for every image.
@@ -569,6 +569,9 @@ def upload_dataset(
                 "Call upload_dataset without use_zip_upload for prediction uploads."
             )
 
+        if annotation_overwrite is None:
+            annotation_overwrite = project.type == "classification"
+
         if use_zip_flow:
             project_slug = project.id.rsplit("/")[1]
             temp_zip = None
@@ -579,9 +582,6 @@ def upload_dataset(
                     zip_path = temp_zip = _zip_directory(dataset_path)
                     print(f"Zipped {dataset_path} -> {zip_path}")
 
-                if annotation_overwrite is None:
-                    annotation_overwrite = project.type == "classification"
-
                 init = rfapi.init_zip_upload(
                     self.__api_key,
                     self.url,
@@ -635,6 +635,8 @@ def _log_img_upload(
             if annotation:
                 if annotation.get("success"):
                     msg += f" / annotations = OK {annotation_time_str}"
+                elif annotation.get("warn") == "already annotated":
+                    msg += " / annotations = kept existing (pass annotation_overwrite=True to replace)"
                 elif annotation.get("warn"):
                     msg += f" / annotations = WARN: {annotation['warn']} {annotation_time_str}"
                 else:
@@ -687,6 +689,7 @@ def _save_annotation(image_id, imagedesc):
                 job_name=batch_name,
                 num_retry_uploads=num_retries,
                 is_prediction=is_prediction,
+                annotation_overwrite=annotation_overwrite,
             )
 
             return annotation, upload_time
 
@@ -1218,3 +1218,61 @@ def test_poll_loop_timeout(self):
 
             if _os.path.exists(zip_path):
                 _os.unlink(zip_path)
+
+
+class TestPerImageAnnotationOverwrite(RoboflowTest):
+    """upload_dataset on a directory (per-image flow) forwards annotation_overwrite to the annotate POST."""
+
+    def _upload(self, image, annotate_status=200, annotate_json=None, **kwargs):
+        import re
+
+        dataset = {
+            "location": "/test/location/",
+            "images": [
+                {
+                    "file": "img.jpg",
+                    "split": "train",
+                    "annotationfile": {"name": "annotation.coco.json", "rawText": "{}"},
+                }
+            ],
+        }
+        responses.add(
+            responses.POST,
+            re.compile(rf"{API_URL}/dataset/{PROJECT_NAME}/annotate/img-id\?.*"),
+            json=annotate_json or {"success": True},
+            status=annotate_status,
+        )
+        with (
+            patch("roboflow.util.folderparser.parsefolder", return_value=dataset),
+            patch("roboflow.core.project.Project.upload_image", return_value=(image, 0.1, 0)),
+            patch("roboflow.core.workspace.Workspace._get_or_create_project", return_value=(self.project, False)),
+            patch("builtins.print") as mock_print,
+        ):
+            self.workspace.upload_dataset(dataset_path="/test/dataset", project_name=PROJECT_NAME, **kwargs)
+        annotate_calls = [c for c in responses.calls if "/annotate/" in c.request.url]
+        logs = " ".join(str(c.args[0]) for c in mock_print.call_args_list if c.args)
+        return annotate_calls, logs
+
+    @responses.activate
+    def test_fresh_image_annotates_without_overwrite(self):
+        calls, logs = self._upload({"id": "img-id", "success": True})
+        self.assertEqual(len(calls), 1)
+        self.assertNotIn("overwrite=true", calls[0].request.url)
+        self.assertIn("annotations = OK", logs)
+
+    @responses.activate
+    def test_duplicate_without_overwrite_keeps_existing_annotations(self):
+        calls, logs = self._upload(
+            {"id": "img-id", "duplicate": True},
+            annotate_status=409,
+            annotate_json={"error": {"message": "Image was already annotated. To overwrite, pass overwrite=true."}},
+        )
+        self.assertEqual(len(calls), 1)
+        self.assertNotIn("overwrite=true", calls[0].request.url)
+        self.assertIn("annotations = kept existing (pass annotation_overwrite=True to replace)", logs)
+
+    @responses.activate
+    def test_duplicate_with_overwrite_true_sends_overwrite(self):
+        calls, _ = self._upload({"id": "img-id", "duplicate": True}, annotation_overwrite=True)
+        self.assertEqual(len(calls), 1)
+        self.assertIn("overwrite=true", calls[0].request.url)