fix: docker 默认 runner 支持 env + 并发开通竞态复用

翻转实弹暴露的两个 #88 缺口：
- _docker_default_runner 不接受 env，_call_runner 遇 env 即 loud fail，
  导致 sub2api 路径下所有 docker 任务失败（xcmapi 路径不传 env，
  此前未暴露）。默认 runner 现接受 env 并与进程环境合并后传给
  docker CLI 子进程（-e <名> 由此解析，值绝不进 argv）。
- 同一用户两任务并发开通时建号撞 email 已存在：现轮询复用
  胜出者入库的行，等不到再原错抛出。
测试：sub2api 16、docker 沙箱 29；全量 cloud 789（4 跳过）、
host 65、deploy 19 全绿。

cloud/tests/test_sub2api.py

@@ -9,6 +9,7 @@ import sqlite3
 import sys
 import unittest
 from pathlib import Path
+from unittest import mock
 
 try:
     from cloud import sub2api
@@ -38,6 +39,8 @@ class FakeApi:
         self.fail_refresh_once = False
         self.balance_calls = []
         self.fail_balance = False
+        self.fail_create_user = False
+        self.create_user_hook = None
 
     def __call__(self, method, url, body=None, token=None):
         path = url.split("28084", 1)[-1] if "28084" in url else url
@@ -62,6 +65,10 @@ class FakeApi:
             return 200, {"data": {"balance": body.get("balance")}}
         if path == "/api/v1/admin/users" and method == "POST":
             self.next_user_id += 1
+            if self.create_user_hook is not None:
+                self.create_user_hook(self.next_user_id)
+            if self.fail_create_user:
+                return 400, {"message": "email already exists"}
             return 200, {"data": {"id": self.next_user_id}}
         if path == "/api/v1/admin/groups" and method == "GET":
             return 200, {"data": self.groups}
@@ -180,6 +187,35 @@ class TestEnsureGatewayKey(unittest.TestCase):
                 os.environ["CLOUD_SUB2API_OPENING_BALANCE_USD"] = old
         self.assertEqual(self.api.balance_calls, [])
 
+    def test_provision_race_reuses_winner_row(self):
+        conn = _conn()
+
+        def hook(_new_id):
+            conn.execute(
+                "INSERT INTO user_gateway_keys(user_id, sub2api_user_id,"
+                " api_key, refresh_token, created_at, updated_at)"
+                " VALUES(?, ?, ?, ?, ?, ?)",
+                ("user-race", 99, "sk-race", "ref-race", 0, 0))
+            conn.commit()
+
+        self.api.create_user_hook = hook
+        self.api.fail_create_user = True
+        key = ensure_user_gateway_key(conn, "user-race", admin=self.admin,
+                                      request=self.api)
+        self.assertEqual(key, "sk-race")
+
+    def test_provision_race_no_row_reraises(self):
+        self.api.fail_create_user = True
+        conn = _conn()
+        with mock.patch("time.sleep", return_value=None):
+            with self.assertRaises(sub2api.Sub2apiError):
+                ensure_user_gateway_key(conn, "user-race2",
+                                        admin=self.admin, request=self.api)
+        row = conn.execute(
+            "SELECT COUNT(*) FROM user_gateway_keys"
+            " WHERE user_id = ?", ("user-race2",)).fetchone()
+        self.assertEqual(row[0], 0)
+
 
 class TestProviderCommand(unittest.TestCase):
     def test_server_command_sub2api(self):