创意审核检查删除 fix

examples/auto_put_ad_mini/.env.example

@@ -384,6 +384,16 @@ EXECUTION_ENABLED=false
 # 实时调控 DB 锁名，默认 tencent_realtime_control
 # RTC_DB_LOCK_NAME=tencent_realtime_control
 
+# -- 拒审创意自动删除限频 --
+# 删除并发 worker，默认 4，代码最大限制 16
+# TENCENT_AD_DELETE_WORKERS=4
+# 所有删除 worker 共享的请求启动最小间隔（秒）
+# TENCENT_AD_DELETE_MIN_INTERVAL_SECONDS=0.25
+# 仅 HTTP 429 或腾讯明确业务限频时重试，默认 2 次
+# TENCENT_AD_DELETE_RATE_LIMIT_RETRIES=2
+# 未返回 Retry-After 时的指数退避基数（秒）
+# TENCENT_AD_DELETE_RATE_LIMIT_BACKOFF_SECONDS=1
+
 # -- 人群包授权 --
 # 人群包来源账户 ID，默认 55615440
 # TENCENT_AUDIENCE_SOURCE_ACCOUNT_ID=55615440

examples/auto_put_ad_mini/docs/unified_services_deployment.md

@@ -89,6 +89,12 @@ DAILY_REJECTED_CREATIVE_CLEANUP_ENABLED=0
 DAILY_REJECTED_CREATIVE_APPLY_ENABLED=0
 DAILY_PARTIAL_CREATIVE_COST_THRESHOLD_YUAN=50
 DAILY_WECHAT_MINI_PROGRAM_CREATIVE_COST_THRESHOLD_YUAN=100
+DAILY_REJECTED_CREATIVE_DELETE_CLAIM_STALE_MINUTES=30
+DAILY_REJECTED_CREATIVE_NOTIFICATION_LOCK_NAME=ad_rejected_creative_notification
+TENCENT_AD_DELETE_WORKERS=4
+TENCENT_AD_DELETE_MIN_INTERVAL_SECONDS=0.25
+TENCENT_AD_DELETE_RATE_LIMIT_RETRIES=2
+TENCENT_AD_DELETE_RATE_LIMIT_BACKOFF_SECONDS=1
 DAILY_REJECTED_CREATIVE_CLEANUP_HOUR=11
 DAILY_REJECTED_CREATIVE_CLEANUP_MINUTE=0
 DAILY_REJECTED_CREATIVE_RUN_ON_STARTUP=0
@@ -155,10 +161,20 @@ T-3 至 T-1 三个完整日内 `SUM(成本)>0` 的 `账号id`，枚举这些账
 `TENCENT_AD_GET_RETRY_ATTEMPTS`（默认总尝试 3 次）和指数退避自动重试。腾讯查询、数据库
 审计和清理操作仍保持原有顺序。账户级创意、广告和审核结果读取按
 `TENCENT_AD_ACCOUNT_SCAN_WORKERS`（默认 8）有限并发，每个账户任务使用独立 HTTP Session；
-候选落库、删除、回读和通知仍在主线程顺序执行。写操作前在锁内重新读取创意和审核结果，
-只有整创意删除动作保持一致才执行。历史遗留的组件删除候选不会被加载执行。清理前后使用
-`RTC_DB_LOCK_NAME` 与实时 CPM/ROI 写操作互斥，并将每条写前状态、回读结果、失败状态和
-通知状态持久化；超时导致写结果未知时，下次运行先回读确认，不会盲目重复写入。
+候选并发落库时单条失败不会中断其他创意。删除前先原子认领数据库记录，再在
+`RTC_DB_LOCK_NAME` 写锁内并发回读、复审和删除；只有认领成功且整创意删除动作保持一致才执行，
+历史遗留的组件删除候选不会被加载执行。认领后仅允许持有 `DELETING` 状态的进程回写结果；
+超过 `DAILY_REJECTED_CREATIVE_DELETE_CLAIM_STALE_MINUTES`（默认 30 分钟）的认领可由下次任务恢复，
+并先回读腾讯状态，避免进程异常退出造成永久卡住或重复删除。删除并发数由
+`TENCENT_AD_DELETE_WORKERS` 控制（默认 4，最大 16）；所有 worker 共享
+`TENCENT_AD_DELETE_MIN_INTERVAL_SECONDS`（默认 0.25 秒）的进程内请求节流器，避免多账户并发
+绕过限频；HTTP 429 或明确的业务限频码最多按 `TENCENT_AD_DELETE_RATE_LIMIT_RETRIES`（默认 2 次）
+并使用 `Retry-After` 或指数退避重试。网络超时、连接中断、408、5xx 和无效响应不会自动重发删除，
+统一记为结果未知并由下一轮回读确认。飞书通知使用独立的
+`DAILY_REJECTED_CREATIVE_NOTIFICATION_LOCK_NAME` MySQL 锁（默认
+`ad_rejected_creative_notification`），拿锁后重新读取待通知记录，避免多实例重复发送；该锁不会在
+腾讯写操作期间持有。每条写前状态、回读结果、失败状态和通知状态均持久化；超时导致写结果未知时，
+下次运行先回读确认，不会盲目重复写入。
 
 自动清理成功项和待人工判断报警项按日级 ROI 数据中的代理归属生成一代理一份
 `YYYYMMDD_代理名称_创意审核异常处理_批次摘要.xlsx`。代理报表仅展示代理、账户、广告、创意、

examples/auto_put_ad_mini/test_creative_review_scan.py

@@ -5,7 +5,7 @@ import unittest
 from contextlib import contextmanager
 from datetime import date, datetime
 from pathlib import Path
-from unittest.mock import Mock, patch
+from unittest.mock import Mock, call, patch
 from zoneinfo import ZoneInfo
 
 import pandas as pd
@@ -286,160 +286,6 @@ class CreativeRejectionCleanupTests(unittest.TestCase):
         )
         self.assertEqual(request.kwargs["params"]["time_line"], "REQUEST_TIME")
 
-    def test_read_only_account_scan_runs_concurrently_with_isolated_clients(self):
-        import threading
-
-        from report_rejected_creatives_once import collect_rejected_creatives
-
-        barrier = threading.Barrier(2, timeout=2)
-        created_clients = []
-        cost_requests = []
-
-        class Session:
-            def __init__(self):
-                self.closed = False
-
-            def close(self):
-                self.closed = True
-
-        class Client:
-            def __init__(self):
-                self.session = Session()
-                created_clients.append(self)
-
-            def get_dynamic_creatives(self, account_id):
-                barrier.wait()
-                return [{
-                    "dynamic_creative_id": account_id * 10,
-                    "dynamic_creative_name": f"创意{account_id}",
-                    "adgroup_id": account_id * 100,
-                    "configured_status": "AD_STATUS_NORMAL",
-                    "creative_set_approval_status": (
-                        "CREATIVE_SET_APPROVAL_STATUS_DENIED"
-                    ),
-                    "system_status": "DYNAMIC_CREATIVE_STATUS_DENIED",
-                }]
-
-            def get_ads(self, account_id):
-                return [{
-                    "adgroup_id": account_id * 100,
-                    "adgroup_name": f"广告{account_id}",
-                }]
-
-            def get_dynamic_creative_costs(
-                self, account_id, creative_ids, start_date, end_date
-            ):
-                cost_requests.append(
-                    (account_id, list(creative_ids), start_date, end_date)
-                )
-                return {creative_id: account_id * 100 for creative_id in creative_ids}
-
-        rows, errors, scanned = collect_rejected_creatives(
-            accounts=[{"account_id": 1}, {"account_id": 2}],
-            agency_context={
-                "creative_agencies": {},
-                "account_agencies": {},
-                "account_names": {},
-            },
-            tencent=Mock(),
-            tencent_factory=Client,
-            review_fetcher=lambda _account, _ids: [],
-            checked_at=datetime(2026, 8, 12, 11, 0),
-            max_workers=2,
-        )
-
-        self.assertEqual(errors, [])
-        self.assertEqual(scanned, 2)
-        self.assertEqual([row["账户ID"] for row in rows], [1, 2])
-        self.assertTrue(all(row["配置状态"] == "投放中" for row in rows))
-        self.assertTrue(
-            all(
-                row["创意审核状态"] == "审核拒绝"
-                for row in rows
-            )
-        )
-        self.assertEqual(
-            [row["近3天历史消耗(元)"] for row in rows],
-            ["1.00", "2.00"],
-        )
-        self.assertTrue(all(row["执行操作"] == "删除创意" for row in rows))
-        self.assertEqual(len(cost_requests), 2)
-        self.assertTrue(
-            all(request[2].isoformat() == "2026-08-09" for request in cost_requests)
-        )
-        self.assertTrue(
-            all(request[3].isoformat() == "2026-08-11" for request in cost_requests)
-        )
-        self.assertEqual(len(created_clients), 2)
-        self.assertTrue(all(client.session.closed for client in created_clients))
-
-    def test_read_only_report_marks_partial_at_threshold_for_manual_review(self):
-        from report_rejected_creatives_once import collect_rejected_creatives
-
-        tencent = Mock()
-        tencent.get_dynamic_creatives.return_value = [{
-            "dynamic_creative_id": 3,
-            "dynamic_creative_name": "创意三",
-            "adgroup_id": 13,
-            "configured_status": "AD_STATUS_NORMAL",
-            "creative_set_approval_status": (
-                "CREATIVE_SET_APPROVAL_STATUS_PARTIAL_NORMAL"
-            ),
-        }]
-        tencent.get_ads.return_value = [{"adgroup_id": 13, "adgroup_name": "广告三"}]
-        tencent.get_dynamic_creative_costs.return_value = {3: 3000}
-
-        rows, errors, scanned = collect_rejected_creatives(
-            accounts=[{"account_id": 1}],
-            agency_context={
-                "creative_agencies": {},
-                "account_agencies": {},
-                "account_names": {},
-            },
-            tencent=tencent,
-            review_fetcher=lambda _account, _ids: [],
-            checked_at=datetime(2026, 8, 12, 11, 0),
-            max_workers=1,
-        )
-
-        self.assertEqual(errors, [])
-        self.assertEqual(scanned, 1)
-        self.assertEqual(len(rows), 1)
-        self.assertEqual(rows[0]["近3天历史消耗(元)"], "30.00")
-        self.assertEqual(rows[0]["消耗日期范围"], "2026-08-09 ~ 2026-08-11")
-        self.assertEqual(rows[0]["执行操作"], "删除创意")
-
-    def test_read_only_workbook_writes_cost_and_action_columns(self):
-        from report_rejected_creatives_once import REPORT_COLUMNS, write_workbook
-
-        row = {column: "" for column in REPORT_COLUMNS}
-        row.update(
-            {
-                "代理名称": "棱镜",
-                "账户ID": 1000000000001,
-                "广告ID": 2000000000002,
-                "创意ID": 3000000000003,
-                "近3天历史消耗(元)": "29.99",
-                "消耗日期范围": "2026-08-10 ~ 2026-08-12",
-                "执行操作": "删除创意",
-                "操作判断原因": "近3天实时消耗低于30元",
-            }
-        )
-        with tempfile.TemporaryDirectory() as directory:
-            path = Path(directory) / "read-only.xlsx"
-            write_workbook([row], path)
-            sheet = load_workbook(path)["审核不通过创意"]
-
-        headers = [cell.value for cell in sheet[1]]
-        self.assertEqual(headers, list(REPORT_COLUMNS))
-        columns = {cell.value: cell.column for cell in sheet[1]}
-        self.assertEqual(sheet.cell(2, columns["近3天历史消耗(元)"]).value, "29.99")
-        self.assertEqual(sheet.cell(2, columns["执行操作"]).value, "删除创意")
-        for column_name in ("账户ID", "广告ID", "创意ID"):
-            cell = sheet.cell(2, columns[column_name])
-            self.assertEqual(cell.data_type, "s")
-            self.assertEqual(cell.number_format, "@")
-
     def test_account_scope_uses_recent_three_day_opengid_spend(self):
         from roi_control.data_source import (
             build_recent_spend_accounts_sql,
@@ -718,10 +564,16 @@ class CreativeRejectionCleanupTests(unittest.TestCase):
         self.assertEqual(values_sql.count("%s"), 19)
         self.assertEqual(len(insert_params), 19)
         self.assertEqual(insert_params[-1], "ALERT_PENDING")
+        update_sql, update_params = cursor.calls[1]
         self.assertIn(
-            "NOT (recent_cost_fen <=> VALUES(recent_cost_fen))",
-            insert_sql,
+            "NOT (item.recent_cost_fen <=> incoming.recent_cost_fen)",
+            update_sql,
         )
+        self.assertIn(
+            "item.cleanup_status NOT IN ('DELETING','CREATIVE_DELETED')",
+            update_sql,
+        )
+        self.assertEqual(update_params, insert_params)
         self.assertEqual(result, {"id": 7})
         connection.close.assert_called_once()
 
@@ -740,8 +592,53 @@ class CreativeRejectionCleanupTests(unittest.TestCase):
 
         sql = cursor.execute.call_args.args[0]
         self.assertIn("cleanup_action='DELETE_CREATIVE'", sql)
+        self.assertIn("item.cleanup_status='DELETING'", sql)
+        self.assertIn("DATE_SUB(NOW(), INTERVAL %s MINUTE)", sql)
         self.assertNotIn("COMPONENTS_PARTIAL", sql)
 
+    def test_cleanup_item_claim_is_atomic_and_reports_contention(self):
+        from tools import creative_rejection_cleanup as cleanup
+
+        cursor = Mock()
+        cursor.__enter__ = Mock(return_value=cursor)
+        cursor.__exit__ = Mock(return_value=None)
+        cursor.rowcount = 0
+        connection = Mock()
+        connection.cursor.return_value = cursor
+
+        with patch.object(cleanup, "get_connection", return_value=connection):
+            self.assertFalse(cleanup.claim_cleanup_item(7))
+
+        sql, params = cursor.execute.call_args.args
+        self.assertIn("SET cleanup_status='DELETING'", sql)
+        self.assertIn("cleanup_action='DELETE_CREATIVE'", sql)
+        self.assertIn("cleanup_status IN", sql)
+        self.assertIn("cleanup_status='DELETING'", sql)
+        self.assertEqual(params, (7, cleanup.DEFAULT_DELETE_CLAIM_STALE_MINUTES))
+
+    def test_owned_cleanup_update_requires_deleting_status(self):
+        from tools import creative_rejection_cleanup as cleanup
+
+        cursor = Mock()
+        cursor.__enter__ = Mock(return_value=cursor)
+        cursor.__exit__ = Mock(return_value=None)
+        cursor.rowcount = 1
+        connection = Mock()
+        connection.cursor.return_value = cursor
+
+        with patch.object(cleanup, "get_connection", return_value=connection):
+            self.assertTrue(
+                cleanup._update_owned_cleanup_item(
+                    7,
+                    cleanup_status="CREATIVE_DELETED",
+                    error_message=None,
+                )
+            )
+
+        sql, params = cursor.execute.call_args.args
+        self.assertIn("WHERE id=%s AND cleanup_status=%s", sql)
+        self.assertEqual(params, ["CREATIVE_DELETED", None, 7, "DELETING"])
+
     def test_dry_run_notification_query_includes_discovered_candidates(self):
         from tools import creative_rejection_cleanup as cleanup
 
@@ -830,41 +727,6 @@ class CreativeRejectionCleanupTests(unittest.TestCase):
         )
         self.assertIsNone(action)
 
-    def test_read_only_report_skips_non_rule_element_denial(self):
-        from report_rejected_creatives_once import collect_rejected_creatives
-
-        tencent = Mock()
-        tencent.get_dynamic_creatives.return_value = [{
-            "dynamic_creative_id": 3,
-            "adgroup_id": 2,
-            "configured_status": "AD_STATUS_NORMAL",
-            "creative_set_approval_status": "CREATIVE_SET_APPROVAL_STATUS_NORMAL",
-        }]
-        tencent.get_ads.return_value = [{"adgroup_id": 2}]
-        review = {
-            "dynamic_creative_id": 3,
-            "element_result_list": [{
-                "element_name": "图片",
-                "review_status": "AD_STATUS_DENIED",
-            }],
-        }
-
-        rows, errors, _ = collect_rejected_creatives(
-            accounts=[{"account_id": 1}],
-            agency_context={
-                "creative_agencies": {},
-                "account_agencies": {},
-                "account_names": {},
-            },
-            tencent=tencent,
-            review_fetcher=lambda _account, _ids: [review],
-            checked_at=datetime(2026, 8, 12, 11, 0),
-            max_workers=1,
-        )
-
-        self.assertEqual(errors, [])
-        self.assertEqual(rows, [])
-
     def test_tencent_client_uses_official_creative_delete_endpoint(self):
         from tencent_client import TencentClient
 
@@ -889,6 +751,234 @@ class CreativeRejectionCleanupTests(unittest.TestCase):
         )
         self.assertEqual(result["configured_status"], "AD_STATUS_DELETED")
 
+    def test_tencent_creative_delete_retries_explicit_rate_limit_only(self):
+        import tencent_client
+
+        TencentClient = tencent_client.TencentClient
+
+        limited = Mock(status_code=429, text="too many requests", headers={})
+        limited.raise_for_status.side_effect = AssertionError(
+            "429 should be handled before raise_for_status"
+        )
+        success = Mock(status_code=200, text="ok", headers={})
+        success.raise_for_status.return_value = None
+        success.json.return_value = {"code": 0, "data": {}}
+        client = TencentClient()
+        client.session.post = Mock(side_effect=[limited, success])
+        client._common_params = Mock(
+            side_effect=[{"nonce": "1"}, {"nonce": "2"}]
+        )
+        client._user_token = Mock(return_value="user-token")
+        client.get_dynamic_creative = Mock(
+            side_effect=RuntimeError("Dynamic creative not found: account=1 creative=3")
+        )
+
+        limiter = Mock()
+        with patch.dict(
+            os.environ,
+            {
+                "TENCENT_AD_DELETE_MIN_INTERVAL_SECONDS": "0",
+                "TENCENT_AD_DELETE_RATE_LIMIT_RETRIES": "1",
+                "TENCENT_AD_DELETE_RATE_LIMIT_BACKOFF_SECONDS": "0",
+            },
+            clear=False,
+        ), patch.object(tencent_client, "_CREATIVE_DELETE_RATE_LIMITER", limiter):
+            result = client.delete_dynamic_creative(1, 3)
+
+        self.assertEqual(result["configured_status"], "AD_STATUS_DELETED")
+        self.assertEqual(client.session.post.call_count, 2)
+        self.assertEqual(client._common_params.call_count, 2)
+        self.assertEqual(limiter.wait.call_args_list, [call(0.0)] * 2)
+        limiter.defer.assert_called_once_with(0.0)
+
+    def test_tencent_creative_delete_honors_shared_retry_after_cooldown(self):
+        import tencent_client
+
+        limited = Mock(
+            status_code=429,
+            text="too many requests",
+            headers={"Retry-After": "2"},
+        )
+        success = Mock(status_code=200, text="ok", headers={})
+        success.raise_for_status.return_value = None
+        success.json.return_value = {"code": 0, "data": {}}
+        client = tencent_client.TencentClient()
+        client.session.post = Mock(side_effect=[limited, success])
+        client._common_params = Mock(return_value={"nonce": "1"})
+        client._user_token = Mock(return_value="user-token")
+        client.get_dynamic_creative = Mock(
+            side_effect=RuntimeError("Dynamic creative not found: account=1 creative=3")
+        )
+        limiter = Mock()
+
+        with patch.dict(
+            os.environ,
+            {
+                "TENCENT_AD_DELETE_MIN_INTERVAL_SECONDS": "0.25",
+                "TENCENT_AD_DELETE_RATE_LIMIT_RETRIES": "1",
+                "TENCENT_AD_DELETE_RATE_LIMIT_BACKOFF_SECONDS": "1",
+            },
+            clear=False,
+        ), patch.object(tencent_client, "_CREATIVE_DELETE_RATE_LIMITER", limiter):
+            client.delete_dynamic_creative(1, 3)
+
+        self.assertEqual(limiter.wait.call_args_list, [call(0.25)] * 2)
+        limiter.defer.assert_called_once_with(2.0)
+
+    def test_tencent_creative_delete_exhausted_rate_limit_still_cools_workers(self):
+        import tencent_client
+
+        client = tencent_client.TencentClient()
+        limited = Mock(
+            status_code=429,
+            text="too many requests",
+            headers={"Retry-After": "3"},
+        )
+        client.session.post = Mock(return_value=limited)
+        client._common_params = Mock(return_value={"nonce": "1"})
+        client._user_token = Mock(return_value="user-token")
+        limiter = Mock()
+
+        with patch.dict(
+            os.environ,
+            {
+                "TENCENT_AD_DELETE_MIN_INTERVAL_SECONDS": "0.25",
+                "TENCENT_AD_DELETE_RATE_LIMIT_RETRIES": "0",
+                "TENCENT_AD_DELETE_RATE_LIMIT_BACKOFF_SECONDS": "1",
+            },
+            clear=False,
+        ), patch.object(tencent_client, "_CREATIVE_DELETE_RATE_LIMITER", limiter):
+            with self.assertRaises(tencent_client.TencentWriteRateLimitedError):
+                client.delete_dynamic_creative(1, 3)
+
+        limiter.wait.assert_called_once_with(0.25)
+        limiter.defer.assert_called_once_with(3.0)
+        self.assertEqual(client.session.post.call_count, 1)
+
+    def test_tencent_creative_delete_invalid_retry_after_uses_backoff(self):
+        import tencent_client
+
+        client = tencent_client.TencentClient()
+        limited = Mock(
+            status_code=429,
+            text="too many requests",
+            headers={"Retry-After": "NaN"},
+        )
+        client.session.post = Mock(return_value=limited)
+        client._common_params = Mock(return_value={"nonce": "1"})
+        client._user_token = Mock(return_value="user-token")
+        limiter = Mock()
+
+        with patch.dict(
+            os.environ,
+            {
+                "TENCENT_AD_DELETE_MIN_INTERVAL_SECONDS": "0",
+                "TENCENT_AD_DELETE_RATE_LIMIT_RETRIES": "0",
+                "TENCENT_AD_DELETE_RATE_LIMIT_BACKOFF_SECONDS": "1.5",
+            },
+            clear=False,
+        ), patch.object(tencent_client, "_CREATIVE_DELETE_RATE_LIMITER", limiter):
+            with self.assertRaises(tencent_client.TencentWriteRateLimitedError):
+                client.delete_dynamic_creative(1, 3)
+
+        limiter.defer.assert_called_once_with(1.5)
+
+    def test_tencent_creative_delete_retries_business_rate_limit_payload(self):
+        import tencent_client
+
+        client = tencent_client.TencentClient()
+        limited = Mock(status_code=200, text="limited", headers={})
+        limited.raise_for_status.return_value = None
+        limited.json.return_value = {
+            "code": 12002,
+            "message_cn": "请求频繁，请稍后重试",
+        }
+        success = Mock(status_code=200, text="ok", headers={})
+        success.raise_for_status.return_value = None
+        success.json.return_value = {"code": 0, "data": {}}
+        client.session.post = Mock(side_effect=[limited, success])
+        client._common_params = Mock(
+            side_effect=[{"nonce": "1"}, {"nonce": "2"}]
+        )
+        client._user_token = Mock(return_value="user-token")
+        client.get_dynamic_creative = Mock(
+            side_effect=RuntimeError("Dynamic creative not found: account=1 creative=3")
+        )
+        limiter = Mock()
+
+        with patch.dict(
+            os.environ,
+            {
+                "TENCENT_AD_DELETE_MIN_INTERVAL_SECONDS": "0",
+                "TENCENT_AD_DELETE_RATE_LIMIT_RETRIES": "1",
+                "TENCENT_AD_DELETE_RATE_LIMIT_BACKOFF_SECONDS": "2",
+            },
+            clear=False,
+        ), patch.object(tencent_client, "_CREATIVE_DELETE_RATE_LIMITER", limiter):
+            result = client.delete_dynamic_creative(1, 3)
+
+        self.assertEqual(result["configured_status"], "AD_STATUS_DELETED")
+        self.assertEqual(client.session.post.call_count, 2)
+        self.assertEqual(limiter.defer.call_args_list, [call(2.0)])
+
+    def test_tencent_creative_delete_does_not_retry_success_message_about_qps(self):
+        import tencent_client
+
+        client = tencent_client.TencentClient()
+        success = Mock(status_code=200, text="ok", headers={})
+        success.raise_for_status.return_value = None
+        success.json.return_value = {
+            "code": 0,
+            "message": "success; current QPS quota is available",
+            "data": {},
+        }
+        client.session.post = Mock(return_value=success)
+        client._common_params = Mock(return_value={"nonce": "1"})
+        client._user_token = Mock(return_value="user-token")
+        client.get_dynamic_creative = Mock(
+            side_effect=RuntimeError("Dynamic creative not found: account=1 creative=3")
+        )
+        limiter = Mock()
+
+        with patch.dict(
+            os.environ,
+            {
+                "TENCENT_AD_DELETE_MIN_INTERVAL_SECONDS": "0",
+                "TENCENT_AD_DELETE_RATE_LIMIT_RETRIES": "2",
+                "TENCENT_AD_DELETE_RATE_LIMIT_BACKOFF_SECONDS": "1",
+            },
+            clear=False,
+        ), patch.object(tencent_client, "_CREATIVE_DELETE_RATE_LIMITER", limiter):
+            result = client.delete_dynamic_creative(1, 3)
+
+        self.assertEqual(result["configured_status"], "AD_STATUS_DELETED")
+        self.assertEqual(client.session.post.call_count, 1)
+        limiter.defer.assert_not_called()
+
+    def test_tencent_creative_delete_does_not_retry_unknown_network_failure(self):
+        from tencent_client import TencentClient, TencentWriteOutcomeUnknownError
+
+        client = TencentClient()
+        client.session.post = Mock(
+            side_effect=__import__("requests").RequestException("connection reset")
+        )
+        client._common_params = Mock(return_value={"nonce": "1"})
+        client._user_token = Mock(return_value="user-token")
+
+        with patch.dict(
+            os.environ,
+            {
+                "TENCENT_AD_DELETE_MIN_INTERVAL_SECONDS": "0",
+                "TENCENT_AD_DELETE_RATE_LIMIT_RETRIES": "3",
+                "TENCENT_AD_DELETE_RATE_LIMIT_BACKOFF_SECONDS": "0",
+            },
+            clear=False,
+        ):
+            with self.assertRaises(TencentWriteOutcomeUnknownError):
+                client.delete_dynamic_creative(1, 3)
+
+        self.assertEqual(client.session.post.call_count, 1)
+
     def test_tencent_client_uses_component_id_for_component_delete(self):
         from tencent_client import TencentClient
 
@@ -1453,6 +1543,8 @@ class CreativeRejectionCleanupTests(unittest.TestCase):
         ), patch.object(
             cleanup, "load_retryable_cleanup_items", return_value=[retry_item]
         ), patch.object(
+            cleanup, "claim_cleanup_item", return_value=True
+        ) as claim_item, patch.object(
             cleanup,
             "update_cleanup_item",
             side_effect=lambda item_id, **values: updates.append((item_id, values)),
@@ -1495,6 +1587,7 @@ class CreativeRejectionCleanupTests(unittest.TestCase):
         self.assertEqual(summary["account_ids"], [1])
         self.assertEqual(summary["tokens_prefetched"], 1)
         self.assertEqual(summary["deleted"], 1)
+        claim_item.assert_called_once_with(7)
         self.assertTrue(
             any(
                 values.get("cleanup_status") == "CREATIVE_DELETED"
@@ -1504,6 +1597,92 @@ class CreativeRejectionCleanupTests(unittest.TestCase):
         mark_notified.assert_called_once_with([7], current)
         mark_operator_notified.assert_called_once_with([7], current)
 
+    def test_cleanup_defers_item_when_delete_rate_limit_is_exhausted(self):
+        from tencent_client import TencentWriteRateLimitedError
+        from tools import creative_rejection_cleanup as cleanup
+
+        @contextmanager
+        def acquired_lock(_name):
+            yield True
+
+        tencent = Mock()
+        tencent.get_dynamic_creative.return_value = {
+            "dynamic_creative_id": 3,
+            "adgroup_id": 2,
+            "creative_set_approval_status": cleanup.CREATIVE_DENIED_STATUS,
+        }
+        tencent.delete_dynamic_creative.side_effect = TencentWriteRateLimitedError(
+            "Tencent HTTP 429"
+        )
+        retry_item = {
+            "id": 7,
+            "account_id": 1,
+            "agency_name": "代理A",
+            "adgroup_id": 2,
+            "dynamic_creative_id": 3,
+            "cleanup_action": cleanup.DELETE_CREATIVE,
+            "cleanup_status": "WRITE_OUTCOME_UNKNOWN",
+        }
+        updates = []
+
+        with tempfile.TemporaryDirectory() as directory, patch.dict(
+            os.environ,
+            {
+                "DAILY_REJECTED_CREATIVE_APPLY_ENABLED": "1",
+                "ROI_AGENCY_WEBHOOK_ENABLED": "1",
+                "ROI_AGENCY_WEBHOOKS_JSON": json.dumps({
+                    "代理A": (
+                        "https://open.feishu.cn/open-apis/bot/v2/hook/"
+                        "test-cleanup-route"
+                    )
+                }),
+                "FEISHU_AD_PROJECT_CHAT_ID": "chat-operator",
+            },
+            clear=False,
+        ), patch.object(cleanup, "initialize_schema"), patch.object(
+            cleanup, "resolve_end_date", return_value="20260812"
+        ), patch.object(
+            cleanup, "fetch_daily_data", return_value=pd.DataFrame()
+        ), patch.object(
+            cleanup, "fetch_recent_spend_accounts", return_value=[]
+        ), patch.object(
+            cleanup, "fetch_account_agency_fallbacks", return_value={}
+        ), patch.object(
+            cleanup, "prefetch_account_access_tokens", return_value={}
+        ), patch.object(
+            cleanup, "load_retryable_cleanup_items", return_value=[retry_item]
+        ), patch.object(
+            cleanup, "claim_cleanup_item", return_value=True
+        ), patch.object(
+            cleanup, "cleanup_precondition_failure", return_value=None
+        ), patch.object(
+            cleanup,
+            "update_cleanup_item",
+            side_effect=lambda item_id, **values: updates.append((item_id, values)),
+        ), patch.object(
+            cleanup, "load_unnotified_deleted_items", return_value=[]
+        ), patch.object(
+            cleanup, "advisory_lock", side_effect=acquired_lock
+        ):
+            summary = cleanup.run_rejected_creative_cleanup(
+                output_dir=Path(directory),
+                now=datetime(2026, 8, 13, 11, 0, tzinfo=ZoneInfo("Asia/Shanghai")),
+                tencent=tencent,
+                odps=Mock(),
+            )
+
+        self.assertEqual(summary["deleted"], 0)
+        self.assertEqual(summary["deferred"], 1)
+        self.assertTrue(any(
+            values.get("cleanup_status") == "DEFERRED"
+            and "429" in values.get("error_message", "")
+            for _, values in updates
+        ))
+        self.assertFalse(any(
+            values.get("cleanup_status") == "WRITE_OUTCOME_UNKNOWN"
+            for _, values in updates
+        ))
+
     def test_apply_deletes_partial_below_threshold_and_alerts_at_threshold(self):
         from tools import creative_rejection_cleanup as cleanup
 
@@ -1625,6 +1804,8 @@ class CreativeRejectionCleanupTests(unittest.TestCase):
             side_effect=lambda record: candidates.append(record) or {"id": len(candidates)},
         ), patch.object(
             cleanup, "load_retryable_cleanup_items", return_value=[retry_item]
+        ), patch.object(
+            cleanup, "claim_cleanup_item", return_value=True
         ), patch.object(
             cleanup,
             "update_cleanup_item",
@@ -1743,6 +1924,8 @@ class CreativeRejectionCleanupTests(unittest.TestCase):
             cleanup, "upsert_cleanup_candidate"
         ), patch.object(
             cleanup, "load_retryable_cleanup_items", return_value=[retry_item]
+        ), patch.object(
+            cleanup, "claim_cleanup_item", return_value=True
         ), patch.object(
             cleanup,
             "update_cleanup_item",
@@ -2139,6 +2322,8 @@ class CreativeRejectionCleanupTests(unittest.TestCase):
             cleanup, "prefetch_account_access_tokens", return_value={}
         ), patch.object(cleanup, "upsert_cleanup_candidate"), patch.object(
             cleanup, "load_retryable_cleanup_items", return_value=[retry_item]
+        ), patch.object(
+            cleanup, "claim_cleanup_item", return_value=True
         ), patch.object(
             cleanup,
             "update_cleanup_item",
@@ -2208,6 +2393,8 @@ class CreativeRejectionCleanupTests(unittest.TestCase):
             cleanup, "prefetch_account_access_tokens", return_value={}
         ), patch.object(cleanup, "upsert_cleanup_candidate"), patch.object(
             cleanup, "load_retryable_cleanup_items", return_value=[retry_item]
+        ), patch.object(
+            cleanup, "claim_cleanup_item", return_value=True
         ), patch.object(
             cleanup,
             "update_cleanup_item",
@@ -2244,7 +2431,11 @@ class CreativeRejectionCleanupTests(unittest.TestCase):
             "reject_reason": "需人工判断",
         }
 
-        def run_with(row):
+        def run_with(row, *, lock_acquired=True):
+            @contextmanager
+            def notification_lock(_name):
+                yield lock_acquired
+
             with tempfile.TemporaryDirectory() as directory, patch.dict(
                 os.environ,
                 {
@@ -2266,6 +2457,8 @@ class CreativeRejectionCleanupTests(unittest.TestCase):
             ), patch.object(cleanup, "prefetch_account_access_tokens", return_value={}), patch.object(
                 cleanup, "load_retryable_cleanup_items", return_value=[]
             ), patch.object(cleanup, "load_unnotified_deleted_items", return_value=[row]), patch.object(
+                cleanup, "advisory_lock", side_effect=notification_lock
+            ), patch.object(
                 cleanup, "publish_agency_reports", return_value=[{
                     "agency_name": "代理A", "status": "SENT"
                 }]
@@ -2307,6 +2500,19 @@ class CreativeRejectionCleanupTests(unittest.TestCase):
         mark_agency.assert_called_once()
         mark_operator.assert_not_called()
 
+        agency, operator, mark_agency, mark_operator = run_with(
+            {
+                **base_row,
+                "agency_notified_at": None,
+                "operator_notified_at": None,
+            },
+            lock_acquired=False,
+        )
+        agency.assert_not_called()
+        operator.assert_not_called()
+        mark_agency.assert_not_called()
+        mark_operator.assert_not_called()
+
     def test_cleanup_entry_returns_nonzero_for_notification_failure(self):
         from examples.auto_put_ad_mini import cleanup_rejected_creatives as entry
 

examples/auto_put_ad_mini/tools/creative_rejection_cleanup.py

@@ -6,6 +6,7 @@ import hashlib
 import json
 import logging
 import os
+import threading
 from collections import defaultdict
 from concurrent.futures import ThreadPoolExecutor, as_completed
 from datetime import date, datetime, timedelta
@@ -43,6 +44,7 @@ DELETE_CREATIVE = "DELETE_CREATIVE"
 ALERT_ONLY = "ALERT_ONLY"
 DEFAULT_PARTIAL_CREATIVE_COST_THRESHOLD_YUAN = 50.0
 DEFAULT_WECHAT_MINI_PROGRAM_COST_THRESHOLD_YUAN = 100.0
+DEFAULT_DELETE_CLAIM_STALE_MINUTES = 30
 AGENCY_REPORT_COLUMNS = (
     "代理名称",
     "账户ID",
@@ -370,6 +372,36 @@ def _resolve_agency(
 
 
 def upsert_cleanup_candidate(record: dict[str, Any]) -> dict[str, Any]:
+    component_ids_json = _json(record.get("component_ids") or [])
+    element_ids_json = _json(record.get("element_ids") or [])
+    review_result_json = _json(record.get("review_result") or {})
+    pre_state_json = _json(record.get("pre_state") or {})
+    cleanup_status = (
+        "ALERT_PENDING"
+        if record["cleanup_action"] == ALERT_ONLY
+        else "DISCOVERED"
+    )
+    insert_values = (
+        record["account_id"],
+        record.get("account_name"),
+        record.get("agency_name"),
+        record["adgroup_id"],
+        record.get("adgroup_name"),
+        record["dynamic_creative_id"],
+        record.get("dynamic_creative_name"),
+        record["check_date"],
+        record["cleanup_action"],
+        component_ids_json,
+        element_ids_json,
+        record.get("recent_cost_fen"),
+        record.get("cost_start_date"),
+        record.get("cost_end_date"),
+        record.get("action_reason") or record["reject_reason"],
+        record["reject_reason"],
+        review_result_json,
+        pre_state_json,
+        cleanup_status,
+    )
     connection = get_connection()
     try:
         with connection.cursor() as cursor:
@@ -384,101 +416,108 @@ def upsert_cleanup_candidate(record: dict[str, Any]) -> dict[str, Any]:
                      cost_start_date, cost_end_date, action_reason, reject_reason,
                      review_result_json, pre_state_json, cleanup_status)
                 VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)
-                ON DUPLICATE KEY UPDATE
-                    account_name=COALESCE(NULLIF(VALUES(account_name),''), account_name),
-                    agency_name=COALESCE(NULLIF(VALUES(agency_name),''), agency_name),
-                    adgroup_id=VALUES(adgroup_id),
-                    adgroup_name=COALESCE(NULLIF(VALUES(adgroup_name),''), adgroup_name),
-                    dynamic_creative_name=COALESCE(
-                        NULLIF(VALUES(dynamic_creative_name),''), dynamic_creative_name
+                ON DUPLICATE KEY UPDATE id=LAST_INSERT_ID(id)
+                """,
+                insert_values,
+            )
+            cursor.execute(
+                """
+                UPDATE creative_rejection_cleanup_item AS item
+                JOIN (
+                    SELECT
+                        %s AS account_id, %s AS account_name, %s AS agency_name,
+                        %s AS adgroup_id, %s AS adgroup_name,
+                        %s AS dynamic_creative_id, %s AS dynamic_creative_name,
+                        %s AS check_date, %s AS cleanup_action,
+                        %s AS target_component_ids_json,
+                        %s AS target_element_ids_json, %s AS recent_cost_fen,
+                        %s AS cost_start_date, %s AS cost_end_date,
+                        %s AS action_reason, %s AS reject_reason,
+                        %s AS review_result_json, %s AS pre_state_json,
+                        %s AS cleanup_status
+                ) AS incoming
+                  ON incoming.account_id=item.account_id
+                 AND incoming.dynamic_creative_id=item.dynamic_creative_id
+                 AND incoming.check_date=item.check_date
+                SET
+                    item.account_name=COALESCE(
+                        NULLIF(incoming.account_name,''), item.account_name
+                    ),
+                    item.agency_name=COALESCE(
+                        NULLIF(incoming.agency_name,''), item.agency_name
+                    ),
+                    item.adgroup_id=incoming.adgroup_id,
+                    item.adgroup_name=COALESCE(
+                        NULLIF(incoming.adgroup_name,''), item.adgroup_name
                     ),
-                    action_reason=VALUES(action_reason),
-                    reject_reason=VALUES(reject_reason),
-                    review_result_json=VALUES(review_result_json),
-                    pre_state_json=VALUES(pre_state_json),
+                    item.dynamic_creative_name=COALESCE(
+                        NULLIF(incoming.dynamic_creative_name,''),
+                        item.dynamic_creative_name
+                    ),
+                    item.action_reason=incoming.action_reason,
+                    item.reject_reason=incoming.reject_reason,
+                    item.review_result_json=incoming.review_result_json,
+                    item.pre_state_json=incoming.pre_state_json,
                     notified_at=CASE
-                        WHEN NOT (cleanup_action <=> VALUES(cleanup_action))
-                          OR NOT (target_component_ids_json <=> VALUES(target_component_ids_json))
-                          OR NOT (target_element_ids_json <=> VALUES(target_element_ids_json))
-                          OR NOT (recent_cost_fen <=> VALUES(recent_cost_fen))
-                          OR NOT (cost_end_date <=> VALUES(cost_end_date))
-                            THEN NULL ELSE notified_at
+                        WHEN NOT (item.cleanup_action <=> incoming.cleanup_action)
+                          OR NOT (item.target_component_ids_json <=> incoming.target_component_ids_json)
+                          OR NOT (item.target_element_ids_json <=> incoming.target_element_ids_json)
+                          OR NOT (item.recent_cost_fen <=> incoming.recent_cost_fen)
+                          OR NOT (item.cost_end_date <=> incoming.cost_end_date)
+                            THEN NULL ELSE item.notified_at
                     END,
-                    agency_notified_at=CASE
-                        WHEN NOT (cleanup_action <=> VALUES(cleanup_action))
-                          OR NOT (target_component_ids_json <=> VALUES(target_component_ids_json))
-                          OR NOT (target_element_ids_json <=> VALUES(target_element_ids_json))
-                          OR NOT (recent_cost_fen <=> VALUES(recent_cost_fen))
-                          OR NOT (cost_end_date <=> VALUES(cost_end_date))
-                            THEN NULL ELSE agency_notified_at
+                    item.agency_notified_at=CASE
+                        WHEN NOT (item.cleanup_action <=> incoming.cleanup_action)
+                          OR NOT (item.target_component_ids_json <=> incoming.target_component_ids_json)
+                          OR NOT (item.target_element_ids_json <=> incoming.target_element_ids_json)
+                          OR NOT (item.recent_cost_fen <=> incoming.recent_cost_fen)
+                          OR NOT (item.cost_end_date <=> incoming.cost_end_date)
+                            THEN NULL ELSE item.agency_notified_at
                     END,
-                    operator_notified_at=CASE
-                        WHEN NOT (cleanup_action <=> VALUES(cleanup_action))
-                          OR NOT (target_component_ids_json <=> VALUES(target_component_ids_json))
-                          OR NOT (target_element_ids_json <=> VALUES(target_element_ids_json))
-                          OR NOT (recent_cost_fen <=> VALUES(recent_cost_fen))
-                          OR NOT (cost_end_date <=> VALUES(cost_end_date))
-                            THEN NULL ELSE operator_notified_at
+                    item.operator_notified_at=CASE
+                        WHEN NOT (item.cleanup_action <=> incoming.cleanup_action)
+                          OR NOT (item.target_component_ids_json <=> incoming.target_component_ids_json)
+                          OR NOT (item.target_element_ids_json <=> incoming.target_element_ids_json)
+                          OR NOT (item.recent_cost_fen <=> incoming.recent_cost_fen)
+                          OR NOT (item.cost_end_date <=> incoming.cost_end_date)
+                            THEN NULL ELSE item.operator_notified_at
                     END,
-                    deleted_at=CASE
-                        WHEN NOT (cleanup_action <=> VALUES(cleanup_action))
-                          OR NOT (target_component_ids_json <=> VALUES(target_component_ids_json))
-                          OR NOT (target_element_ids_json <=> VALUES(target_element_ids_json))
-                          OR NOT (recent_cost_fen <=> VALUES(recent_cost_fen))
-                          OR NOT (cost_end_date <=> VALUES(cost_end_date))
-                            THEN NULL ELSE deleted_at
+                    item.deleted_at=CASE
+                        WHEN NOT (item.cleanup_action <=> incoming.cleanup_action)
+                          OR NOT (item.target_component_ids_json <=> incoming.target_component_ids_json)
+                          OR NOT (item.target_element_ids_json <=> incoming.target_element_ids_json)
+                          OR NOT (item.recent_cost_fen <=> incoming.recent_cost_fen)
+                          OR NOT (item.cost_end_date <=> incoming.cost_end_date)
+                            THEN NULL ELSE item.deleted_at
                     END,
-                    readback_json=CASE
-                        WHEN NOT (cleanup_action <=> VALUES(cleanup_action))
-                          OR NOT (target_component_ids_json <=> VALUES(target_component_ids_json))
-                          OR NOT (target_element_ids_json <=> VALUES(target_element_ids_json))
-                          OR NOT (recent_cost_fen <=> VALUES(recent_cost_fen))
-                          OR NOT (cost_end_date <=> VALUES(cost_end_date))
-                            THEN NULL ELSE readback_json
+                    item.readback_json=CASE
+                        WHEN NOT (item.cleanup_action <=> incoming.cleanup_action)
+                          OR NOT (item.target_component_ids_json <=> incoming.target_component_ids_json)
+                          OR NOT (item.target_element_ids_json <=> incoming.target_element_ids_json)
+                          OR NOT (item.recent_cost_fen <=> incoming.recent_cost_fen)
+                          OR NOT (item.cost_end_date <=> incoming.cost_end_date)
+                            THEN NULL ELSE item.readback_json
                     END,
-                    cleanup_status=CASE
-                        WHEN NOT (cleanup_action <=> VALUES(cleanup_action))
-                          OR NOT (target_component_ids_json <=> VALUES(target_component_ids_json))
-                          OR NOT (target_element_ids_json <=> VALUES(target_element_ids_json))
-                          OR NOT (recent_cost_fen <=> VALUES(recent_cost_fen))
-                          OR NOT (cost_end_date <=> VALUES(cost_end_date))
-                            THEN VALUES(cleanup_status)
-                        WHEN cleanup_status='SKIPPED_REVIEW_NOT_RECONFIRMED'
+                    item.cleanup_status=CASE
+                        WHEN NOT (item.cleanup_action <=> incoming.cleanup_action)
+                          OR NOT (item.target_component_ids_json <=> incoming.target_component_ids_json)
+                          OR NOT (item.target_element_ids_json <=> incoming.target_element_ids_json)
+                          OR NOT (item.recent_cost_fen <=> incoming.recent_cost_fen)
+                          OR NOT (item.cost_end_date <=> incoming.cost_end_date)
+                            THEN incoming.cleanup_status
+                        WHEN item.cleanup_status='SKIPPED_REVIEW_NOT_RECONFIRMED'
                             THEN 'DISCOVERED'
-                        ELSE cleanup_status
+                        ELSE item.cleanup_status
                     END,
-                    target_component_ids_json=VALUES(target_component_ids_json),
-                    target_element_ids_json=VALUES(target_element_ids_json),
-                    recent_cost_fen=VALUES(recent_cost_fen),
-                    cost_start_date=VALUES(cost_start_date),
-                    cost_end_date=VALUES(cost_end_date),
-                    cleanup_action=VALUES(cleanup_action)
+                    item.target_component_ids_json=incoming.target_component_ids_json,
+                    item.target_element_ids_json=incoming.target_element_ids_json,
+                    item.recent_cost_fen=incoming.recent_cost_fen,
+                    item.cost_start_date=incoming.cost_start_date,
+                    item.cost_end_date=incoming.cost_end_date,
+                    item.cleanup_action=incoming.cleanup_action
+                WHERE item.cleanup_status NOT IN ('DELETING','CREATIVE_DELETED')
                 """,
-                (
-                    record["account_id"],
-                    record.get("account_name"),
-                    record.get("agency_name"),
-                    record["adgroup_id"],
-                    record.get("adgroup_name"),
-                    record["dynamic_creative_id"],
-                    record.get("dynamic_creative_name"),
-                    record["check_date"],
-                    record["cleanup_action"],
-                    _json(record.get("component_ids") or []),
-                    _json(record.get("element_ids") or []),
-                    record.get("recent_cost_fen"),
-                    record.get("cost_start_date"),
-                    record.get("cost_end_date"),
-                    record.get("action_reason") or record["reject_reason"],
-                    record["reject_reason"],
-                    _json(record.get("review_result") or {}),
-                    _json(record.get("pre_state") or {}),
-                    (
-                        "ALERT_PENDING"
-                        if record["cleanup_action"] == ALERT_ONLY
-                        else "DISCOVERED"
-                    ),
-                ),
+                insert_values,
             )
             cursor.execute(
                 """
@@ -497,6 +536,16 @@ def upsert_cleanup_candidate(record: dict[str, Any]) -> dict[str, Any]:
 
 
 def load_retryable_cleanup_items() -> list[dict[str, Any]]:
+    stale_minutes = int(
+        os.getenv(
+            "DAILY_REJECTED_CREATIVE_DELETE_CLAIM_STALE_MINUTES",
+            str(DEFAULT_DELETE_CLAIM_STALE_MINUTES),
+        )
+    )
+    if stale_minutes <= 0:
+        raise ValueError(
+            "DAILY_REJECTED_CREATIVE_DELETE_CLAIM_STALE_MINUTES must be positive"
+        )
     connection = get_connection()
     try:
         with connection.cursor() as cursor:
@@ -512,18 +561,64 @@ def load_retryable_cleanup_items() -> list[dict[str, Any]]:
                   ON latest.account_id=item.account_id
                  AND latest.dynamic_creative_id=item.dynamic_creative_id
                  AND latest.check_date=item.check_date
-                WHERE item.cleanup_status IN
-                    ('DISCOVERED','DEFERRED','FAILED','WRITE_OUTCOME_UNKNOWN')
+                WHERE (
+                    item.cleanup_status IN
+                        ('DISCOVERED','DEFERRED','FAILED','WRITE_OUTCOME_UNKNOWN')
+                    OR (
+                        item.cleanup_status='DELETING'
+                        AND item.updated_at < DATE_SUB(NOW(), INTERVAL %s MINUTE)
+                    )
+                )
                   AND item.cleanup_action='DELETE_CREATIVE'
                 ORDER BY item.id
-                """
+                """,
+                (stale_minutes,),
             )
             return list(cursor.fetchall())
     finally:
         connection.close()
 
 
-def update_cleanup_item(item_id: int, **values: Any) -> None:
+def claim_cleanup_item(item_id: int) -> bool:
+    """Atomically claim a retryable deletion row for this run."""
+
+    stale_minutes = int(
+        os.getenv(
+            "DAILY_REJECTED_CREATIVE_DELETE_CLAIM_STALE_MINUTES",
+            str(DEFAULT_DELETE_CLAIM_STALE_MINUTES),
+        )
+    )
+    if stale_minutes <= 0:
+        raise ValueError(
+            "DAILY_REJECTED_CREATIVE_DELETE_CLAIM_STALE_MINUTES must be positive"
+        )
+    connection = get_connection()
+    try:
+        with connection.cursor() as cursor:
+            cursor.execute(
+                """
+                UPDATE creative_rejection_cleanup_item
+                SET cleanup_status='DELETING', error_message=NULL, updated_at=NOW()
+                WHERE id=%s
+                  AND cleanup_action='DELETE_CREATIVE'
+                  AND (
+                      cleanup_status IN
+                          ('DISCOVERED','DEFERRED','FAILED','WRITE_OUTCOME_UNKNOWN')
+                      OR (
+                          cleanup_status='DELETING'
+                          AND updated_at < DATE_SUB(NOW(), INTERVAL %s MINUTE)
+                      )
+                  )
+                """,
+                (item_id, stale_minutes),
+            )
+            return cursor.rowcount == 1
+    finally:
+        connection.close()
+
+
+def update_cleanup_item(item_id: int, **values: Any) -> bool:
+    expected_cleanup_status = values.pop("_expected_cleanup_status", None)
     allowed = {
         "agency_name",
         "cleanup_action",
@@ -548,15 +643,29 @@ def update_cleanup_item(item_id: int, **values: Any) -> None:
     if unknown:
         raise ValueError(f"Unsupported cleanup fields: {sorted(unknown)}")
     if not values:
-        return
+        return False
     assignments = ", ".join(f"{name}=%s" for name in values)
     connection = get_connection()
     try:
         with connection.cursor() as cursor:
+            where = "WHERE id=%s"
+            params = [*values.values(), item_id]
+            if expected_cleanup_status is not None:
+                if isinstance(expected_cleanup_status, (tuple, list, set, frozenset)):
+                    statuses = list(expected_cleanup_status)
+                    if not statuses:
+                        return False
+                    placeholders = ",".join(["%s"] * len(statuses))
+                    where += f" AND cleanup_status IN ({placeholders})"
+                    params.extend(statuses)
+                else:
+                    where += " AND cleanup_status=%s"
+                    params.append(expected_cleanup_status)
             cursor.execute(
-                f"UPDATE creative_rejection_cleanup_item SET {assignments} WHERE id=%s",
-                [*values.values(), item_id],
+                f"UPDATE creative_rejection_cleanup_item SET {assignments} {where}",
+                params,
             )
+            return cursor.rowcount > 0
     finally:
         connection.close()
 
@@ -651,6 +760,16 @@ def mark_cleanup_items_notified(
     mark_cleanup_items_agency_notified(item_ids, notified_at)
 
 
+def _update_owned_cleanup_item(item_id: int, **values: Any) -> bool:
+    """Update a row only while this run owns its DELETING claim."""
+
+    return update_cleanup_item(
+        item_id,
+        _expected_cleanup_status="DELETING",
+        **values,
+    )
+
+
 def upsert_cleanup_delivery(record: dict[str, Any]) -> dict[str, Any]:
     connection = get_connection()
     try:
@@ -1456,8 +1575,23 @@ def run_rejected_creative_cleanup(
             max_workers=process_workers,
             thread_name_prefix="creative-process",
         ) as executor:
-            results = executor.map(process_creative, tasks)
-            for completed, result in enumerate(results, start=1):
+            futures = {
+                executor.submit(process_creative, task): task for task in tasks
+            }
+            for completed, future in enumerate(as_completed(futures), start=1):
+                task = futures[future]
+                account_id = task[1]
+                creative_id = _as_int(task[2].get("dynamic_creative_id"))
+                try:
+                    result = future.result()
+                except Exception as exc:
+                    error = (
+                        "creative processing failed "
+                        f"account={account_id} creative={creative_id}: {exc}"
+                    )
+                    scan_errors.append(error)
+                    logger.exception(error)
+                    continue
                 if result is None:
                     continue
                 account_id, creative_id, action = result
@@ -1484,9 +1618,16 @@ def run_rejected_creative_cleanup(
             item_id = int(item["id"])
             account_id = int(item["account_id"])
             creative_id = int(item["dynamic_creative_id"])
+            snapshot_status = str(item.get("cleanup_status") or "")
+            if snapshot_status == "DELETING":
+                # A stale claim is recovered only after obtaining the global
+                # Tencent write lock; do not mutate a potentially live owner.
+                deletable_items.append(item)
+                continue
             if item.get("cleanup_action") != DELETE_CREATIVE:
                 update_cleanup_item(
                     item_id,
+                    _expected_cleanup_status=snapshot_status,
                     cleanup_status="SKIPPED_REVIEW_NOT_RECONFIRMED",
                     error_message="当前规则仅允许删除整条创意",
                 )
@@ -1495,7 +1636,12 @@ def run_rejected_creative_cleanup(
                 context, account_id, creative_id
             )
             if agency and agency != item.get("agency_name"):
-                update_cleanup_item(item_id, agency_name=agency)
+                update_cleanup_item(
+                    item_id,
+                    _expected_cleanup_status=snapshot_status,
+                    agency_name=agency,
+                )
+                item["agency_name"] = agency
             webhook_url = resolve_agency_webhook(
                 agency, webhook_config.webhooks or {}
             )
@@ -1505,14 +1651,19 @@ def run_rejected_creative_cleanup(
                     if not agency
                     else f"代理商 {agency} 未配置通知群，禁止自动删除"
                 )
-                update_cleanup_item(
+                updated = update_cleanup_item(
                     item_id,
+                    _expected_cleanup_status=snapshot_status,
                     cleanup_status="DEFERRED",
                     error_message=reason,
                 )
-                deferred += 1
+                if updated is not False:
+                    deferred += 1
                 continue
-            if item.get("cleanup_status") != "WRITE_OUTCOME_UNKNOWN":
+            if item.get("cleanup_status") not in {
+                "WRITE_OUTCOME_UNKNOWN",
+                "DELETING",
+            }:
                 precondition_failure = cleanup_precondition_failure(
                     item,
                     scanned_accounts,
@@ -1520,12 +1671,13 @@ def run_rejected_creative_cleanup(
                 )
                 if precondition_failure:
                     status, reason = precondition_failure
-                    update_cleanup_item(
+                    updated = update_cleanup_item(
                         item_id,
+                        _expected_cleanup_status=snapshot_status,
                         cleanup_status=status,
                         error_message=reason,
                     )
-                    if status == "DEFERRED":
+                    if status == "DEFERRED" and updated is not False:
                         deferred += 1
                     continue
             deletable_items.append(item)
@@ -1534,11 +1686,23 @@ def run_rejected_creative_cleanup(
         # 每 worker 用独立 TencentClient（requests.Session 非线程安全），
         # 避免共享 session 并发导致不可预测行为（与扫描阶段一致）。
         if deletable_items:
+            configured_delete_workers = int(
+                os.getenv("TENCENT_AD_DELETE_WORKERS", "4")
+            )
+            if configured_delete_workers < 1:
+                raise ValueError("TENCENT_AD_DELETE_WORKERS must be at least 1")
             delete_workers = min(
-                int(os.getenv("TENCENT_AD_DELETE_WORKERS", "4")),
+                configured_delete_workers,
                 len(deletable_items),
                 16,
             )
+            if not owned_tencent:
+                # A caller-supplied client may wrap a requests.Session and is
+                # not assumed to be thread-safe.
+                delete_workers = 1
+            delete_worker_local = threading.local()
+            delete_worker_clients = []
+            delete_worker_clients_lock = threading.Lock()
 
             def delete_one(item, delete_client):
                 """锁内单条创意：回读 → 复审 → 删除；返回 (deleted, deferred, error)。"""
@@ -1553,22 +1717,28 @@ def run_rejected_creative_cleanup(
                     except Exception as read_exc:
                         if (
                             item.get("cleanup_action") == DELETE_CREATIVE
-                            and item.get("cleanup_status")
-                            == "WRITE_OUTCOME_UNKNOWN"
                             and str(read_exc).startswith(
                                 "Dynamic creative not found:"
                             )
                         ):
-                            update_cleanup_item(
+                            updated = _update_owned_cleanup_item(
                                 item_id,
                                 cleanup_status="CREATIVE_DELETED",
                                 error_message=None,
                                 readback_json=_json({"deleted_from_listing": True}),
                                 deleted_at=effective_now,
                             )
-                            return 1, 0, None
+                            if updated is not False:
+                                return 1, 0, None
+                            return 0, 0, (
+                                f"account={account_id} creative={creative_id}: "
+                                "delete result ignored because claim ownership was lost"
+                            )
                         raise
-                    if item.get("cleanup_status") == "WRITE_OUTCOME_UNKNOWN":
+                    if item.get("cleanup_status") in {
+                        "WRITE_OUTCOME_UNKNOWN",
+                        "DELETING",
+                    }:
                         precondition_failure = cleanup_precondition_failure(
                             item,
                             scanned_accounts,
@@ -1576,7 +1746,7 @@ def run_rejected_creative_cleanup(
                         )
                         if precondition_failure:
                             status, reason = precondition_failure
-                            update_cleanup_item(
+                            _update_owned_cleanup_item(
                                 item_id,
                                 cleanup_status=status,
                                 error_message=reason,
@@ -1629,7 +1799,7 @@ def run_rejected_creative_cleanup(
                             fresh_action
                             and fresh_action.get("cleanup_action") == ALERT_ONLY
                         ):
-                            update_cleanup_item(
+                            _update_owned_cleanup_item(
                                 item_id,
                                 cleanup_action=ALERT_ONLY,
                                 target_component_ids_json="[]",
@@ -1651,7 +1821,7 @@ def run_rejected_creative_cleanup(
                                 notified_at=None,
                             )
                             return 0, 0, None
-                        update_cleanup_item(
+                        updated = _update_owned_cleanup_item(
                             item_id,
                             cleanup_status="SKIPPED_REVIEW_NOT_RECONFIRMED",
                             error_message="写锁内回读发现整创意删除条件已变化",
@@ -1662,7 +1832,7 @@ def run_rejected_creative_cleanup(
                         readback = delete_client.delete_dynamic_creative(
                             account_id, creative_id
                         )
-                        update_cleanup_item(
+                        updated = _update_owned_cleanup_item(
                             item_id,
                             cleanup_status="CREATIVE_DELETED",
                             error_message=None,
@@ -1670,12 +1840,18 @@ def run_rejected_creative_cleanup(
                             readback_json=_json(readback),
                             deleted_at=effective_now,
                         )
-                        return 1, 0, None
+                        if updated is not False:
+                            return 1, 0, None
+                        return 0, 0, (
+                            f"account={account_id} creative={creative_id}: "
+                            "delete result ignored because claim ownership was lost"
+                        )
                     return 0, 0, None
                 except Exception as exc:
                     from tencent_client import (
                         PostWriteVerificationError,
                         TencentWriteOutcomeUnknownError,
+                        TencentWriteRateLimitedError,
                     )
 
                     outcome_unknown = isinstance(
@@ -1685,63 +1861,166 @@ def run_rejected_creative_cleanup(
                             PostWriteVerificationError,
                         ),
                     )
-                    update_cleanup_item(
-                        item_id,
-                        cleanup_status=(
-                            "WRITE_OUTCOME_UNKNOWN" if outcome_unknown else "FAILED"
-                        ),
-                        error_message=str(exc)[:4000],
-                    )
-                    return 0, 0, f"account={account_id} creative={creative_id}: {exc}"
+                    rate_limited = isinstance(exc, TencentWriteRateLimitedError)
+                    error = f"account={account_id} creative={creative_id}: {exc}"
+                    try:
+                        _update_owned_cleanup_item(
+                            item_id,
+                            cleanup_status=(
+                                "DEFERRED"
+                                if rate_limited
+                                else (
+                                    "WRITE_OUTCOME_UNKNOWN"
+                                    if outcome_unknown
+                                    else "FAILED"
+                                )
+                            ),
+                            error_message=str(exc)[:4000],
+                        )
+                    except Exception as update_exc:
+                        logger.exception(
+                            "creative delete failure status update failed "
+                            "account=%d creative=%d",
+                            account_id,
+                            creative_id,
+                        )
+                        error += f"; status_update_failed={update_exc}"
+                    return 0, int(rate_limited), error
 
             def run_delete(item):
                 if owned_tencent:
-                    from tencent_client import TencentClient
-
-                    delete_client = TencentClient()
-                    delete_client.seed_access_tokens(prefetched_tokens)
+                    delete_client = getattr(delete_worker_local, "client", None)
+                    if delete_client is None:
+                        from tencent_client import TencentClient
+
+                        delete_client = TencentClient()
+                        delete_client.seed_access_tokens(prefetched_tokens)
+                        delete_worker_local.client = delete_client
+                        with delete_worker_clients_lock:
+                            delete_worker_clients.append(delete_client)
                 else:
                     delete_client = client
-                try:
-                    return delete_one(item, delete_client)
-                finally:
-                    if delete_client is not client:
-                        delete_client.session.close()
+                return delete_one(item, delete_client)
 
             with advisory_lock(write_lock_name) as acquired:
                 if not acquired:
                     for item in deletable_items:
-                        update_cleanup_item(
+                        snapshot_status = str(item.get("cleanup_status") or "")
+                        if snapshot_status == "DELETING":
+                            continue
+                        updated = update_cleanup_item(
                             int(item["id"]),
+                            _expected_cleanup_status=snapshot_status,
                             cleanup_status="DEFERRED",
                             error_message="腾讯写锁被实时调控占用",
                         )
-                        deferred += 1
+                        if updated is not False:
+                            deferred += 1
                 else:
+                    claimed_items = []
+                    for item in deletable_items:
+                        try:
+                            if claim_cleanup_item(int(item["id"])):
+                                claimed_items.append(item)
+                        except Exception as exc:
+                            error = (
+                                "creative delete claim failed "
+                                f"account={item['account_id']} "
+                                f"creative={item['dynamic_creative_id']}: {exc}"
+                            )
+                            delete_errors.append(error)
+                            logger.exception(error)
+                    executable_items = []
+                    for item in claimed_items:
+                        agency = str(item.get("agency_name") or "") or _resolve_agency(
+                            context,
+                            int(item["account_id"]),
+                            int(item["dynamic_creative_id"]),
+                        )
+                        webhook_url = resolve_agency_webhook(
+                            agency, webhook_config.webhooks or {}
+                        )
+                        if agency and webhook_url:
+                            if agency != item.get("agency_name"):
+                                _update_owned_cleanup_item(
+                                    int(item["id"]),
+                                    agency_name=agency,
+                                )
+                                item["agency_name"] = agency
+                            executable_items.append(item)
+                            continue
+                        reason = (
+                            "代理商归属为空，禁止自动删除"
+                            if not agency
+                            else f"代理商 {agency} 未配置通知群，禁止自动删除"
+                        )
+                        updated = _update_owned_cleanup_item(
+                            int(item["id"]),
+                            cleanup_status="DEFERRED",
+                            error_message=reason,
+                        )
+                        if updated is not False:
+                            deferred += 1
                     logger.info(
-                        "creative delete started items=%d workers=%d",
+                        "creative delete started candidates=%d claimed=%d workers=%d",
                         len(deletable_items),
+                        len(executable_items),
                         delete_workers,
                     )
-                    with ThreadPoolExecutor(
-                        max_workers=delete_workers,
-                        thread_name_prefix="creative-delete",
-                    ) as executor:
-                        for d_deleted, d_deferred, d_error in executor.map(
-                            run_delete, deletable_items
-                        ):
-                            deleted += d_deleted
-                            deferred += d_deferred
-                            if d_error:
-                                delete_errors.append(d_error)
+                    if executable_items:
+                        try:
+                            with ThreadPoolExecutor(
+                                max_workers=min(delete_workers, len(executable_items)),
+                                thread_name_prefix="creative-delete",
+                            ) as executor:
+                                futures = {
+                                    executor.submit(run_delete, item): item
+                                    for item in executable_items
+                                }
+                                for future in as_completed(futures):
+                                    item = futures[future]
+                                    try:
+                                        d_deleted, d_deferred, d_error = future.result()
+                                    except Exception as exc:
+                                        d_deleted = 0
+                                        d_deferred = 0
+                                        d_error = (
+                                            "creative delete worker failed "
+                                            f"account={item['account_id']} "
+                                            f"creative={item['dynamic_creative_id']}: {exc}"
+                                        )
+                                        logger.exception(d_error)
+                                        try:
+                                            _update_owned_cleanup_item(
+                                                int(item["id"]),
+                                                cleanup_status="FAILED",
+                                                error_message=str(exc)[:4000],
+                                            )
+                                        except Exception as update_exc:
+                                            logger.exception(
+                                                "creative delete worker failure status "
+                                                "update failed account=%s creative=%s",
+                                                item["account_id"],
+                                                item["dynamic_creative_id"],
+                                            )
+                                            d_error += (
+                                                f"; status_update_failed={update_exc}"
+                                            )
+                                    deleted += d_deleted
+                                    deferred += d_deferred
+                                    if d_error:
+                                        delete_errors.append(d_error)
+                        finally:
+                            for delete_worker_client in delete_worker_clients:
+                                delete_worker_client.session.close()
 
-        pending_notifications = load_unnotified_deleted_items(
-            include_discovered=not apply_enabled,
-        )
         deliveries: list[dict[str, object]] = []
         operator_deliveries: list[dict[str, object]] = []
         notification_errors: list[str] = []
-        if pending_notifications and webhook_config.enabled:
+
+        def publish_pending_notifications(
+            pending_notifications: list[dict[str, Any]],
+        ) -> None:
             owned_publisher = publisher is None
             sheet_publisher = publisher or RoiFeishuPublisher(require_chat_ids=False)
             try:
@@ -1822,6 +2101,27 @@ def run_rejected_creative_cleanup(
                 if owned_publisher:
                     sheet_publisher.close()
 
+        pending_notification_probe = load_unnotified_deleted_items(
+            include_discovered=not apply_enabled,
+        )
+        if pending_notification_probe and webhook_config.enabled:
+            notification_lock_name = os.getenv(
+                "DAILY_REJECTED_CREATIVE_NOTIFICATION_LOCK_NAME",
+                "ad_rejected_creative_notification",
+            )
+            with advisory_lock(notification_lock_name) as acquired:
+                if not acquired:
+                    logger.info(
+                        "creative cleanup notification skipped: lock busy name=%s",
+                        notification_lock_name,
+                    )
+                else:
+                    pending_notifications = load_unnotified_deleted_items(
+                        include_discovered=not apply_enabled,
+                    )
+                    if pending_notifications:
+                        publish_pending_notifications(pending_notifications)
+
         return {
             "apply_enabled": apply_enabled,
             "account_scope": "opengid_recent_3d_spend",

examples/tencent_realtime_control/tencent_client.py

@@ -4,7 +4,9 @@ from __future__ import annotations
 
 import json
 import logging
+import math
 import os
+import threading
 import time
 import uuid
 from datetime import date
@@ -18,6 +20,128 @@ from storage import connect
 logger = logging.getLogger(__name__)
 
 
+class _CreativeDeleteRateLimiter:
+    """Process-wide start-rate limiter shared by all TencentClient instances."""
+
+    def __init__(self) -> None:
+        self._condition = threading.Condition()
+        self._next_allowed_at = 0.0
+
+    def wait(self, min_interval_seconds: float) -> None:
+        min_interval_seconds = max(min_interval_seconds, 0.0)
+        with self._condition:
+            while True:
+                now = time.monotonic()
+                delay = self._next_allowed_at - now
+                if delay <= 0:
+                    self._next_allowed_at = now + min_interval_seconds
+                    return
+                # Condition.wait releases the lock so a concurrent 429 can
+                # immediately extend the shared cooldown before this request
+                # is allowed to start.
+                self._condition.wait(timeout=delay)
+
+    def defer(self, delay_seconds: float) -> None:
+        """Apply a shared cooldown after any worker receives a rate limit."""
+
+        if delay_seconds <= 0:
+            return
+        with self._condition:
+            self._next_allowed_at = max(
+                self._next_allowed_at,
+                time.monotonic() + delay_seconds,
+            )
+            self._condition.notify_all()
+
+
+_CREATIVE_DELETE_RATE_LIMITER = _CreativeDeleteRateLimiter()
+
+
+def _creative_delete_rate_limit_config() -> tuple[float, int, float]:
+    """Return (minimum interval, retries, backoff) for delete POST requests."""
+
+    try:
+        min_interval = float(os.getenv("TENCENT_AD_DELETE_MIN_INTERVAL_SECONDS", "0.25"))
+        retries = int(os.getenv("TENCENT_AD_DELETE_RATE_LIMIT_RETRIES", "2"))
+        backoff = float(os.getenv("TENCENT_AD_DELETE_RATE_LIMIT_BACKOFF_SECONDS", "1"))
+    except (TypeError, ValueError) as exc:
+        raise ValueError(
+            "TENCENT_AD_DELETE_MIN_INTERVAL_SECONDS, "
+            "TENCENT_AD_DELETE_RATE_LIMIT_RETRIES and "
+            "TENCENT_AD_DELETE_RATE_LIMIT_BACKOFF_SECONDS must be numeric"
+        ) from exc
+    if (
+        not math.isfinite(min_interval)
+        or not math.isfinite(backoff)
+        or min_interval < 0
+        or retries < 0
+        or backoff < 0
+    ):
+        raise ValueError(
+            "Tencent creative delete rate-limit settings must be finite and "
+            "not negative"
+        )
+    if retries > 10:
+        raise ValueError(
+            "TENCENT_AD_DELETE_RATE_LIMIT_RETRIES must not be greater than 10"
+        )
+    return min_interval, retries, backoff
+
+
+def _rate_limit_retry_after(response: requests.Response) -> float | None:
+    raw = response.headers.get("Retry-After") if response.headers else None
+    try:
+        value = float(raw)
+    except (TypeError, ValueError):
+        return None
+    if not math.isfinite(value) or value < 0:
+        return None
+    return min(value, 60.0)
+
+
+def _is_rate_limited_payload(payload: Any) -> bool:
+    if not isinstance(payload, dict):
+        return False
+    codes = [
+        str(payload.get(name)).strip()
+        for name in ("code", "error_code")
+        if payload.get(name) is not None
+        and str(payload.get(name)).strip()
+    ]
+    if "429" in codes:
+        return True
+    # Text matching is only safe on an explicit error response. A successful
+    # delete may legally include informational text mentioning QPS or quotas;
+    # retrying that POST could repeat a write that already succeeded.
+    if not codes or all(code == "0" for code in codes):
+        return False
+    values = [
+        payload.get("message"),
+        payload.get("message_cn"),
+        payload.get("error_message"),
+    ]
+    text = " ".join(str(value or "") for value in values).lower()
+    markers = (
+        "rate limit",
+        "rate_limit",
+        "rate-limit",
+        "throttle",
+        "qps",
+        "too many",
+        "too frequent",
+        "频率限制",
+        "访问频率",
+        "请求频繁",
+        "请求过于频繁",
+        "调用频次",
+        "调用次数超限",
+        "请求次数超限",
+        "限流",
+        "限频",
+    )
+    return any(marker in text for marker in markers)
+
+
 ACTIVE_STATUS = "AD_STATUS_NORMAL"
 SUSPEND_STATUS = "AD_STATUS_SUSPEND"
 DELETED_STATUS = "AD_STATUS_DELETED"
@@ -75,6 +199,10 @@ class TencentWriteRejectedError(RuntimeError):
     """Tencent explicitly rejected the write request."""
 
 
+class TencentWriteRateLimitedError(TencentWriteRejectedError):
+    """Tencent explicitly rejected the write because of rate limiting."""
+
+
 class TencentWriteOutcomeUnknownError(RuntimeError):
     """The request may have reached Tencent, but no definitive result exists."""
 
@@ -463,40 +591,114 @@ class TencentClient:
     ) -> dict[str, Any]:
         """Delete a creative through Tencent's dedicated delete endpoint."""
 
+        min_interval, rate_limit_retries, rate_limit_backoff = (
+            _creative_delete_rate_limit_config()
+        )
+
         try:
-            params = {
-                **self._common_params(account_id),
-                "user_token": self._user_token(account_id),
-            }
+            user_token = self._user_token(account_id)
         except Exception as exc:
             raise TencentWriteNotSentError(str(exc)) from exc
-        try:
-            response = self.session.post(
-                f"{self.base_url}/dynamic_creatives/delete",
-                params=params,
-                json={
-                    "account_id": account_id,
-                    "dynamic_creative_id": dynamic_creative_id,
-                },
-                timeout=self.timeout,
-            )
-        except requests.RequestException as exc:
-            raise TencentWriteOutcomeUnknownError(str(exc)) from exc
-        if response.status_code == 408 or response.status_code >= 500:
-            raise TencentWriteOutcomeUnknownError(
-                f"Tencent HTTP {response.status_code}: {response.text[:500]}"
+        payload_body = {
+            "account_id": account_id,
+            "dynamic_creative_id": dynamic_creative_id,
+        }
+        data: dict[str, Any] | None = None
+        for attempt in range(rate_limit_retries + 1):
+            _CREATIVE_DELETE_RATE_LIMITER.wait(min_interval)
+            try:
+                # Refresh timestamp/nonce on every explicit rate-limit retry;
+                # Tencent may reject a reused signed request even after the
+                # rate-limit window has elapsed.
+                params = {
+                    **self._common_params(account_id),
+                    "user_token": user_token,
+                }
+            except Exception as exc:
+                raise TencentWriteNotSentError(str(exc)) from exc
+            try:
+                response = self.session.post(
+                    f"{self.base_url}/dynamic_creatives/delete",
+                    params=params,
+                    json=payload_body,
+                    timeout=self.timeout,
+                )
+            except requests.RequestException as exc:
+                # The request may have reached Tencent.  Retrying a POST here
+                # could issue a duplicate delete, so let the next scheduled run
+                # resolve it by read-back.
+                raise TencentWriteOutcomeUnknownError(str(exc)) from exc
+
+            if response.status_code == 429:
+                delay = _rate_limit_retry_after(response)
+                if delay is None:
+                    delay = min(rate_limit_backoff * (2**attempt), 60.0)
+                # Publish every explicit limit, including the final exhausted
+                # attempt, so other delete workers do not immediately continue
+                # hitting the same Tencent quota window.
+                _CREATIVE_DELETE_RATE_LIMITER.defer(delay)
+                if attempt < rate_limit_retries:
+                    logger.warning(
+                        "Tencent creative delete rate-limited account=%s creative=%s "
+                        "attempt=%d/%d retry_in=%.2fs",
+                        account_id,
+                        dynamic_creative_id,
+                        attempt + 1,
+                        rate_limit_retries + 1,
+                        delay,
+                    )
+                    continue
+                raise TencentWriteRateLimitedError(
+                    f"Tencent HTTP 429: {response.text[:500]}"
+                )
+            if response.status_code == 408 or response.status_code >= 500:
+                raise TencentWriteOutcomeUnknownError(
+                    f"Tencent HTTP {response.status_code}: {response.text[:500]}"
+                )
+            try:
+                response.raise_for_status()
+                payload = response.json()
+            except requests.HTTPError as exc:
+                raise TencentWriteRejectedError(str(exc)) from exc
+            except (ValueError, TypeError) as exc:
+                raise TencentWriteOutcomeUnknownError(
+                    f"Tencent returned invalid JSON: {response.text[:500]}"
+                ) from exc
+            if not isinstance(payload, dict):
+                raise TencentWriteOutcomeUnknownError(
+                    "Tencent returned a non-object JSON response: "
+                    f"{response.text[:500]}"
+                )
+
+            if _is_rate_limited_payload(payload):
+                delay = _rate_limit_retry_after(response)
+                if delay is None:
+                    delay = min(rate_limit_backoff * (2**attempt), 60.0)
+                _CREATIVE_DELETE_RATE_LIMITER.defer(delay)
+                if attempt < rate_limit_retries:
+                    logger.warning(
+                        "Tencent creative delete business-rate-limited account=%s "
+                        "creative=%s attempt=%d/%d retry_in=%.2fs",
+                        account_id,
+                        dynamic_creative_id,
+                        attempt + 1,
+                        rate_limit_retries + 1,
+                        delay,
+                    )
+                    continue
+                raise TencentWriteRateLimitedError(
+                    "delete_dynamic_creative rate limited after retries: "
+                    f"{payload}"
+                )
+            try:
+                data = self._check(payload, "delete_dynamic_creative")
+            except RuntimeError as exc:
+                raise TencentWriteRejectedError(str(exc)) from exc
+            break
+        if data is None:
+            raise TencentWriteRejectedError(
+                "delete_dynamic_creative did not return a definitive response"
             )
-        try:
-            response.raise_for_status()
-            data = self._check(response.json(), "delete_dynamic_creative")
-        except requests.HTTPError as exc:
-            raise TencentWriteRejectedError(str(exc)) from exc
-        except (ValueError, TypeError) as exc:
-            raise TencentWriteOutcomeUnknownError(
-                f"Tencent returned invalid JSON: {response.text[:500]}"
-            ) from exc
-        except RuntimeError as exc:
-            raise TencentWriteRejectedError(str(exc)) from exc
 
         last_actual: dict[str, Any] = {}
         for attempt in range(1, self.verify_attempts + 1):