fix: transplant PG init backfill rollback scope + audit-loop state upsert registration (code-to10 v2, from 854f810)

- db: translate INSERT OR IGNORE ... SELECT to ON CONFLICT DO NOTHING;
  register task_audit_loop_state (task_id) in _OR_REPLACE_PK (direct port).
- server: _backfill_default_quotas uses SAVEPOINT containment, no longer
  commits the caller's transaction (direct port; callers are init_db only).
- worker: NOT ported — integration _coord_child_rows has no column-fallback
  retry loop, so the InFailedSqlTransaction cascade site does not exist here.
- tests: rewritten for the integration base (no cloud.audit_loop /
  _coord_audit_loop_merge here); real-PG cases skip without DSN/driver.

(cherry picked from commit 79e6843610fcd275af874e7b4afc57af7ffa579e)

cloud/tests/test_pg_init_fix.py

@@ -1,5 +1,10 @@
 """PG init/backfill 回滚与审计状态 upsert 登记修复测试。
 
+演练注：add/add 冲突取并集——以 main 版（真实接口版）为基底，
+另并入移植版独有的 2 项纯 _translate_sql 回归（无列清单 SELECT
+形、VALUES 形不变）；移植版其余用例均已被 main 版同名/更强
+用例覆盖（save_loop_state、worker 合流、PG upsert 行更新）。
+
 只覆盖两处修复，不跑真实模型回合、不联网：
 - 问题一：新鲜 PG 库 init 后 tasks 缺 paused_by_user /
   interrupted_reason，合流随后连环报 InFailedSqlTransaction。
@@ -83,6 +88,28 @@ class TranslateTestCase(unittest.TestCase):
         self.assertIn("%s", sql)
         self.assertNotIn("?", sql)
 
+    def test_ignore_select_form_without_column_list_translated(self):
+        # 正则里列清单是可选组；无列清单的 SELECT 形同样要翻译，
+        # 不能原样透传到 PG 报 syntax error。
+        sql, kind = _translate_sql(
+            "INSERT OR IGNORE INTO user_quotas"
+            " SELECT u.id, ?, ? FROM users u"
+            " LEFT JOIN user_quotas q ON q.user_id = u.id"
+            " WHERE q.user_id IS NULL")
+        self.assertEqual(kind, "normal")
+        self.assertNotIn("OR IGNORE", sql)
+        self.assertIn("ON CONFLICT DO NOTHING", sql)
+        self.assertIn("SELECT u.id", sql)
+
+    def test_ignore_values_form_unchanged(self):
+        # 回归：既有 VALUES 形翻译不受新增 SELECT 形分支影响。
+        sql, kind = _translate_sql(
+            "INSERT OR IGNORE INTO user_quotas(user_id,"
+            " quota_usd_micros, updated_at) VALUES(?, ?, ?)")
+        self.assertEqual(kind, "normal")
+        self.assertNotIn("OR IGNORE", sql)
+        self.assertIn("ON CONFLICT DO NOTHING", sql)
+
     def test_audit_loop_state_or_replace_updates(self):
         sql, kind = _translate_sql(
             "INSERT OR REPLACE INTO task_audit_loop_state"