feat: 接入模型动态 Fallback 与调用审计

core/backend/apps/adminpanel/tests.py

@@ -368,6 +368,35 @@ class AdminTaskMonitorTests(TestCase):
         self.assertEqual(r.status_code, 200)
         self.assertGreaterEqual(r.data["count"], 4)
 
+    def test_list_defers_large_payload_columns_but_detail_keeps_them(self):
+        from django.db import connection
+        from django.test.utils import CaptureQueriesContext
+
+        self.t_ok.request_payload = {"prompt": "x" * 100_000}
+        self.t_ok.response_payload = {"image": "y" * 100_000}
+        self.t_ok.error_message = "z" * 100_000
+        self.t_ok.save(update_fields=["request_payload", "response_payload", "error_message", "updated_at"])
+
+        with CaptureQueriesContext(connection) as captured:
+            response = self.ac.get("/api/admin/tasks/?page_size=10")
+        self.assertEqual(response.status_code, 200)
+        task_selects = [
+            item["sql"]
+            for item in captured.captured_queries
+            if "FROM \"ai_aitask\"" in item["sql"] and "COUNT(" not in item["sql"]
+        ]
+        self.assertTrue(task_selects)
+        for sql in task_selects:
+            self.assertNotIn("request_payload", sql)
+            self.assertNotIn("response_payload", sql)
+            self.assertNotIn("error_message", sql)
+
+        detail = self.ac.get(f"/api/admin/tasks/{self.t_ok.id}/")
+        self.assertEqual(detail.status_code, 200)
+        self.assertEqual(detail.data["request_payload"], self.t_ok.request_payload)
+        self.assertEqual(detail.data["response_payload"], self.t_ok.response_payload)
+        self.assertEqual(detail.data["error_message"], self.t_ok.error_message)
+
     def test_filter_status_type_anomaly(self):
         self.assertTrue(all(t["status"] == "failed" for t in self.ac.get("/api/admin/tasks/?status=failed").data["results"]))
         self.assertTrue(all(t["task_type"] == "person_image" for t in self.ac.get("/api/admin/tasks/?task_type=person_image").data["results"]))
@@ -380,10 +409,34 @@ class AdminTaskMonitorTests(TestCase):
         self.assertFalse(self.ac.get(f"/api/admin/tasks/{self.t_ok.id}/").data["cost_anomaly"])
 
     def test_detail_has_payloads(self):
+        from django.utils import timezone
+
+        from apps.ai.models import AIModelAttempt
+
+        AIModelAttempt.objects.create(
+            task=self.t_failed,
+            sequence=1,
+            provider=self.mc.provider,
+            model_config=self.mc,
+            provider_name=self.mc.provider.name,
+            provider_display_name=self.mc.provider.display_name,
+            model_name=self.mc.name,
+            model_display_name=self.mc.display_name,
+            public_model_name="AirShelf Image",
+            capability="image",
+            operation="image_generate",
+            status=AIModelAttempt.Status.FAILED,
+            started_at=timezone.now(),
+            error_type="provider_unavailable",
+        )
         d = self.ac.get(f"/api/admin/tasks/{self.t_failed.id}/")
         self.assertEqual(d.status_code, 200)
         self.assertIn("request_payload", d.data)
         self.assertIn("response_payload", d.data)
+        self.assertEqual(len(d.data["attempts"]), 1)
+        self.assertEqual(d.data["attempts"][0]["model_name"], self.mc.name)
+        row = next(item for item in self.ac.get("/api/admin/tasks/").data["results"] if item["id"] == str(self.t_failed.id))
+        self.assertNotIn("attempts", row)
 
     def test_retry_dispatch_and_audit(self):
         with patch("apps.ai.tasks.generate_standalone_image_task.delay") as mock_delay: