diff --git a/app/api/v1/ad.py b/app/api/v1/ad.py
index 6f8ab9b..9bc9cbe 100644
--- a/app/api/v1/ad.py
+++ b/app/api/v1/ad.py
@@ -120,6 +120,21 @@ def pangle_callback(request: Request, db: DbSession) -> PangleCallbackOut:
     ad_session_id = extra.get("ad_session_id")
     ecpm = params.get("ecpm")
 
+    # 环境隔离:激励视频 mediaExtra 里带「这次观看属于哪个后端环境」(srv_env=dev/prod,客户端按
+    # BuildConfig.DEBUG 决定),穿山甲 S2S 回调原样带回。回调 URL 在穿山甲后台只配一个(指向生产),
+    # 所以测试包(dev)看广告的 S2S 也会打到生产——若不拦,生产会把奖发给「本库里同 user_id 的另一个
+    # 真人」(user_id 是跨库不隔离的裸数字 → 跨库串号)。这里只处理「属于本服环境」的回调:环境不符
+    # 直接受理但不发币、不写任何记录,保证各环境后台广告收益页只含本环境用户。is_verify=true 让穿山甲
+    # 不再重试(测试用户的币由 localhost 的 test-grant 单独发,不依赖这条 S2S)。
+    # 兼容:旧客户端不带 srv_env(取不到)→ 视为本环境,照常处理,不误伤存量正式用户。
+    callback_env = extra.get("srv_env")
+    if callback_env and callback_env != settings.APP_ENV:
+        logger.info(
+            "pangle callback foreign env skip: callback_env=%s self_env=%s user_id=%d trans_id=%s",
+            callback_env, settings.APP_ENV, user_id, trans_id,
+        )
+        return PangleCallbackOut(is_verify=True, reason=REASON_OK)
+
     existing = crud_ad.find_by_trans(db, trans_id)
     if existing is not None:
         logger.info(
diff --git a/scripts/sim_pangle_callback.py b/scripts/sim_pangle_callback.py
index 52e767d..31fd490 100644
--- a/scripts/sim_pangle_callback.py
+++ b/scripts/sim_pangle_callback.py
@@ -12,6 +12,7 @@ FastAPI TestClient 在进程内打**真实** endpoint(`/api/v1/ad/pangle-callbac
 from __future__ import annotations
 
 import argparse
+import json
 import os
 import sys
 import uuid
@@ -56,6 +57,11 @@ def main() -> None:
     p = argparse.ArgumentParser(description="模拟穿山甲发奖回调")
     p.add_argument("user_id", nargs="?", type=int, default=None, help="目标 user_id(默认最近注册)")
     p.add_argument("--trans", default=None, help="交易号(默认随机);同号重跑验幂等")
+    p.add_argument(
+        "--srv-env", default=None,
+        help="模拟客户端 mediaExtra 里的后端环境标记(dev/prod)。与本服 APP_ENV 不符时"
+             "应被环境隔离拦下(Δ0、不串号);留空=不带该字段(旧客户端,照常发)",
+    )
     args = p.parse_args()
 
     if not settings.pangle_callback_configured:
@@ -75,6 +81,9 @@ def main() -> None:
         # 镜像生产:后台代码位"奖励数量"应配成 AD_REWARD_COIN,回调据此发金币
         "reward_amount": str(AD_REWARD_COIN),
     }
+    # 镜像穿山甲透传 mediaExtra:把后端环境标记放进 gromoreExtra(服务端据此做环境隔离)
+    if args.srv_env:
+        params["gromoreExtra"] = json.dumps({"srv_env": args.srv_env})
     params["sign"] = pangle.build_sign(trans_id, settings.PANGLE_REWARD_SECRET)
 
     resp = TestClient(app).get("/api/v1/ad/pangle-callback", params=params)
diff --git a/tests/test_ad_reward.py b/tests/test_ad_reward.py
index 1943317..00b85c0 100644
--- a/tests/test_ad_reward.py
+++ b/tests/test_ad_reward.py
@@ -5,6 +5,8 @@
 """
 from __future__ import annotations
 
+import json
+
 from sqlalchemy import select
 
 from app.core.config import settings
@@ -179,6 +181,38 @@ def test_callback_unknown_user(client) -> None:
     assert r.json() == {"is_verify": False, "reason": 2}
 
 
+def test_callback_foreign_env_skipped(client) -> None:
+    """回调 srv_env 与本服 APP_ENV 不符(测试包的 S2S 打到生产)→ 受理但不发币、不串号。
+
+    防的是跨库串号:同一 user_id 在测试库 / 正式库是两个人,回调地址只配生产,测试包(dev)看广告
+    的 S2S 也会打到生产。测试环境 APP_ENV=dev:srv_env=prod 视为「别的环境」→ 跳过,余额不变、
+    不写记录;srv_env=dev(本环境)照常发。srv_env 走 gromoreExtra 透传(穿山甲带回 mediaExtra 的字段)。
+    兼容:不带 srv_env 的旧客户端按本环境处理(其余用例已覆盖)。
+    """
+    phone = "13800003501"
+    token = _login(client, phone)
+    uid = _user_id(phone)
+
+    # 外环境(prod)回调:受理(is_verify=true 防穿山甲重试)但不发币、不留记录
+    foreign = _signed(uid, "trans_foreign_env", ecpm="200",
+                      gromoreExtra=json.dumps({"srv_env": "prod"}))
+    r = _callback(client, foreign)
+    assert r.status_code == 200, r.text
+    assert r.json() == {"is_verify": True, "reason": 0}
+    assert _coin_balance(client, token) == 0
+    st = client.get("/api/v1/ad/reward-status", headers=_auth(token)).json()
+    assert st["used_today"] == 0
+
+    # 本环境(dev)回调:照常发币
+    same = _signed(uid, "trans_same_env", ecpm="200",
+                   gromoreExtra=json.dumps({"srv_env": "dev"}))
+    r = _callback(client, same)
+    assert r.status_code == 200, r.text
+    assert _coin_balance(client, token) == calculate_ad_reward_coin("200", 1)
+    st = client.get("/api/v1/ad/reward-status", headers=_auth(token)).json()
+    assert st["used_today"] == 1
+
+
 def test_reward_status_requires_auth(client) -> None:
     """客户端进度接口需登录。"""
     r = client.get("/api/v1/ad/reward-status")