diff --git a/game-runtime/tools/newapi_pool_provision.py b/game-runtime/tools/newapi_pool_provision.py
index 9a1da68e..afc88704 100644
--- a/game-runtime/tools/newapi_pool_provision.py
+++ b/game-runtime/tools/newapi_pool_provision.py
@@ -235,6 +235,17 @@ def compute_grant(yuan: float, qpu: int, usd: float) -> int:
     return round(yuan / usd * qpu)
 
 
+def wait_for_db_visibility(query, attempts: int = 20, interval_seconds: float = 0.1):
+    """等待 new-api API 写入在 PostgreSQL 可见；超出固定窗口后返回 None。"""
+    for attempt in range(attempts):
+        value = query()
+        if value is not None:
+            return value
+        if attempt + 1 < attempts:
+            time.sleep(interval_seconds)
+    return None
+
+
 def validate_provision_request(start: int, count: int, prefix: str, hostname: str) -> list[str]:
     """校验离线补池边界，并返回确定性的目标用户名。
 
@@ -279,7 +290,7 @@ def provision_one(pg: Postgres, root_token: str, username: str, grant: int,
             "password": password,
             "display_name": username[:20],  # new-api DisplayName max=20，直接用 username（截断兜底）
         })
-        uid = pg.query_user_id(username)
+        uid = wait_for_db_visibility(lambda: pg.query_user_id(username))
         if uid is None:
             raise RuntimeError(f"建用户 {username} 后 postgres 未查到 uid")
         log.info("[1/4] 建用户 %s → uid=%d", username, uid)
@@ -304,7 +315,7 @@ def provision_one(pg: Postgres, root_token: str, username: str, grant: int,
             "unlimited_quota": False,
             "expired_time": -1,  # 不过期，避免自伤式过期（设计 §3.9）
         })
-        tok = pg.query_token(uid, username)
+        tok = wait_for_db_visibility(lambda: pg.query_token(uid, username))
         if tok is None:
             raise RuntimeError(f"建 token 后 postgres 未查到 user={uid} name={username}")
         log.info("[3/4] 建 token uid=%d → token_id=%d key=%s", uid, tok[0], _mask(tok[1]))
diff --git a/game-runtime/tools/test_newapi_pool_provision.py b/game-runtime/tools/test_newapi_pool_provision.py
index a7a66933..00bf1460 100755
--- a/game-runtime/tools/test_newapi_pool_provision.py
+++ b/game-runtime/tools/test_newapi_pool_provision.py
@@ -58,6 +58,22 @@ class ProvisionRequestTest(unittest.TestCase):
             self.assertEqual(stat.S_IMODE(sql_path.stat().st_mode), 0o600)
             self.assertIn("-- pool_entry username=neice_051 newapi_user_id=51", sql_path.read_text())
 
+    def test_waits_for_api_write_to_become_visible(self) -> None:
+        query = mock.Mock(side_effect=[None, None, 67])
+        with mock.patch.object(MODULE.time, "sleep") as sleep:
+            value = MODULE.wait_for_db_visibility(query, attempts=4, interval_seconds=0.1)
+        self.assertEqual(value, 67)
+        self.assertEqual(query.call_count, 3)
+        self.assertEqual(sleep.call_count, 2)
+
+    def test_db_visibility_wait_is_bounded(self) -> None:
+        query = mock.Mock(return_value=None)
+        with mock.patch.object(MODULE.time, "sleep") as sleep:
+            value = MODULE.wait_for_db_visibility(query, attempts=3, interval_seconds=0.1)
+        self.assertIsNone(value)
+        self.assertEqual(query.call_count, 3)
+        self.assertEqual(sleep.call_count, 2)
+
 
 if __name__ == "__main__":
     unittest.main()