fix: 隔离图片创作商品会话

core/backend/apps/ai/tests.py

@@ -1012,6 +1012,42 @@ class ImageConversationTests(TestCase):
         self.assertEqual(self.client.get("/api/ai/image-conversations/?mode=image").json()["results"], [])
         self.assertTrue(ImageConversation.objects.get(id=conv_id).is_deleted)
 
+    def test_listing_filters_product_and_unbound_scopes(self):
+        product = Product.objects.create(team=self.team, created_by=self.user, title="范围商品")
+        unbound = ImageConversation.objects.create(team=self.team, created_by=self.user, title="通用会话")
+        bound = ImageConversation.objects.create(team=self.team, created_by=self.user, title="商品会话", product=product)
+
+        unbound_rows = self.client.get(
+            "/api/ai/image-conversations/?mode=image&scope=unbound"
+        ).json()["results"]
+        product_rows = self.client.get(
+            f"/api/ai/image-conversations/?mode=image&product_id={product.id}"
+        ).json()["results"]
+
+        self.assertEqual([row["id"] for row in unbound_rows], [str(unbound.id)])
+        self.assertEqual([row["id"] for row in product_rows], [str(bound.id)])
+        conflict = self.client.get(
+            f"/api/ai/image-conversations/?mode=image&scope=unbound&product_id={product.id}"
+        )
+        self.assertEqual(conflict.status_code, 400)
+
+    def test_conversation_scope_rejects_cross_team_product(self):
+        other = User.objects.create_user(username="conv-other", password="pass")
+        other_team = Team.objects.create(name="Conv Other", owner=other)
+        other_product = Product.objects.create(team=other_team, created_by=other, title="其他团队商品")
+
+        listed = self.client.get(
+            f"/api/ai/image-conversations/?mode=image&product_id={other_product.id}"
+        )
+        created = self.client.post(
+            "/api/ai/image-conversations/",
+            {"mode": "image", "title": "越权会话", "product": str(other_product.id)},
+            format="json",
+        )
+
+        self.assertEqual(listed.status_code, 400)
+        self.assertEqual(created.status_code, 400)
+
     def test_trash_restore_and_purge_conversation_keeps_record(self):
         conv = ImageConversation.objects.create(team=self.team, created_by=self.user, title="待删", mode=ImageConversation.Mode.IMAGE)
         provider = ModelProvider.objects.create(name="conv-trash-provider", display_name="Conv Trash Provider")
@@ -1162,13 +1198,79 @@ class ImageConversationTests(TestCase):
         self.assertTrue(all(t.conversation_id == conv.id for t in tasks))
         self.assertEqual(conv.tasks.count(), 2)
 
+    def test_image_generate_without_product_keeps_conversation_and_task_unbound(self):
+        """图片生成首页进入的图片创作不应静默补入当前/首个商品。"""
+        CreditAccount.objects.create(team=self.team, balance="100.0000")
+        Product.objects.create(team=self.team, created_by=self.user, title="不应自动关联的商品")
+        with patch("apps.ai.tasks.generate_standalone_image_task.delay"):
+            response = self.client.post(
+                "/api/ai/generate-image/",
+                {"prompt": "一只宇航猫", "mode": "image", "count": 1},
+                format="json",
+            )
+
+        self.assertEqual(response.status_code, 202, response.content)
+        conversation = ImageConversation.objects.get(id=response.json()["conversation_id"])
+        task = AITask.objects.get(conversation=conversation)
+        self.assertIsNone(conversation.product_id)
+        self.assertIsNone(task.request_payload.get("product_id"))
+
+    def test_image_generate_with_explicit_product_keeps_product_binding(self):
+        """商品详情显式进入图片创作时，仍保留商品归属。"""
+        CreditAccount.objects.create(team=self.team, balance="100.0000")
+        product = Product.objects.create(team=self.team, created_by=self.user, title="显式关联商品")
+        with patch("apps.ai.tasks.generate_standalone_image_task.delay"):
+            response = self.client.post(
+                "/api/ai/generate-image/",
+                {"prompt": "商品海报", "mode": "image", "count": 1, "product_id": str(product.id)},
+                format="json",
+            )
+
+        self.assertEqual(response.status_code, 202, response.content)
+        conversation = ImageConversation.objects.get(id=response.json()["conversation_id"])
+        task = AITask.objects.get(conversation=conversation)
+        self.assertEqual(conversation.product_id, product.id)
+        self.assertEqual(task.request_payload.get("product_id"), str(product.id))
+
+    def test_new_generation_with_mismatched_conversation_creates_correct_scope(self):
+        CreditAccount.objects.create(team=self.team, balance="100.0000")
+        original_product = Product.objects.create(team=self.team, created_by=self.user, title="原商品")
+        target_product = Product.objects.create(team=self.team, created_by=self.user, title="目标商品")
+        original_conversation = ImageConversation.objects.create(
+            team=self.team,
+            created_by=self.user,
+            title="原商品会话",
+            product=original_product,
+        )
+
+        with patch("apps.ai.tasks.generate_standalone_image_task.delay"):
+            response = self.client.post(
+                "/api/ai/generate-image/",
+                {
+                    "prompt": "目标商品海报",
+                    "mode": "image",
+                    "count": 1,
+                    "product_id": str(target_product.id),
+                    "conversation_id": str(original_conversation.id),
+                },
+                format="json",
+            )
+
+        self.assertEqual(response.status_code, 202, response.content)
+        self.assertNotEqual(response.json()["conversation_id"], str(original_conversation.id))
+        created_conversation = ImageConversation.objects.get(id=response.json()["conversation_id"])
+        self.assertEqual(created_conversation.product_id, target_product.id)
+        self.assertEqual(original_conversation.tasks.count(), 0)
+        self.assertEqual(created_conversation.tasks.get().request_payload.get("product_id"), str(target_product.id))
+
     def test_tasks_endpoint_returns_grouped_history(self):
-        conv = ImageConversation.objects.create(team=self.team, created_by=self.user, title="历史")
+        product = Product.objects.create(team=self.team, created_by=self.user, title="历史商品")
+        conv = ImageConversation.objects.create(team=self.team, created_by=self.user, title="历史", product=product)
         mc = ModelConfig.objects.filter(capability=ModelConfig.Capability.IMAGE).first()
         AITask.objects.create(
             team=self.team, created_by=self.user, conversation=conv, task_type=AITask.Type.PRODUCT_IMAGE,
             status=AITask.Status.SUCCEEDED, model_config=mc, idempotency_key="conv-test-1",
-            request_payload={"prompt": "猫", "batch_id": "bx", "ratio": "1:1"},
+            request_payload={"prompt": "猫", "batch_id": "bx", "ratio": "1:1", "product_id": str(product.id)},
         )
         r = self.client.get(f"/api/ai/image-conversations/{conv.id}/tasks/")
         self.assertEqual(r.status_code, 200, r.content)
@@ -1176,6 +1278,40 @@ class ImageConversationTests(TestCase):
         self.assertEqual(len(data), 1)
         self.assertEqual(data[0]["prompt"], "猫")
         self.assertEqual(data[0]["batch_id"], "bx")
+        self.assertEqual(data[0]["product_id"], str(product.id))
+
+    def test_rerun_uses_original_batch_product_instead_of_request_product(self):
+        CreditAccount.objects.create(team=self.team, balance="100.0000")
+        original_product = Product.objects.create(team=self.team, created_by=self.user, title="原批次商品")
+        other_product = Product.objects.create(team=self.team, created_by=self.user, title="当前路由商品")
+        with patch("apps.ai.tasks.generate_standalone_image_task.delay"):
+            first = self.client.post(
+                "/api/ai/generate-image/",
+                {"prompt": "原商品海报", "mode": "image", "count": 1, "product_id": str(original_product.id)},
+                format="json",
+            )
+            self.assertEqual(first.status_code, 202, first.content)
+            original_task = AITask.objects.get(conversation_id=first.json()["conversation_id"])
+            original_task.status = AITask.Status.FAILED
+            original_task.save(update_fields=["status"])
+            rerun = self.client.post(
+                "/api/ai/generate-image/",
+                {
+                    "prompt": "原商品海报",
+                    "mode": "image",
+                    "count": 1,
+                    "product_id": str(other_product.id),
+                    "conversation_id": first.json()["conversation_id"],
+                    "batch_id": first.json()["batch_id"],
+                    "retry_of_task_id": str(original_task.id),
+                },
+                format="json",
+            )
+
+        self.assertEqual(rerun.status_code, 202, rerun.content)
+        self.assertEqual(rerun.json()["conversation_id"], first.json()["conversation_id"])
+        newest = AITask.objects.filter(conversation_id=first.json()["conversation_id"]).order_by("created_at").last()
+        self.assertEqual(newest.request_payload.get("product_id"), str(original_product.id))
 
     def test_generate_with_refs_persists_and_tasks_endpoint_returns_them(self):
         """HTTP 全链路:带 reference_image_ids 提交 → 任务 payload 落 ids → tasks 接口把参考图解析回 {name,url}。