diff --git a/src/core/runtime_config.py b/src/core/runtime_config.py
index ad8c254..42494a5 100644
--- a/src/core/runtime_config.py
+++ b/src/core/runtime_config.py
@@ -127,6 +127,21 @@ async def get_runtime_value(key: str) -> str:
     return getattr(settings, key, "") or ""
 
 
+async def set_runtime_value(key: str, value: str) -> None:
+    """写入/更新 DB 覆盖值(调用方需先校验 key 在白名单内)。
+
+    敏感项(SETTING_DEFS.sensitive)以 Fernet 加密存储,库里不落明文。
+    """
+    defn = SETTING_DEFS.get(key)
+    stored = crypto.encrypt_value(value) if defn and defn.sensitive else value
+    async with AsyncSessionLocal() as db:
+        stmt = pg_insert(AppSetting).values(key=key, value=stored)
+        stmt = stmt.on_conflict_do_update(index_elements=["key"], set_={"value": stored})
+        await db.execute(stmt)
+        await db.commit()
+    logger.info("运行时配置 %s 已更新", key)
+
+
 async def get_setting_origin(key: str) -> tuple[str, str]:
     """返回 (origin, 当前生效值)。origin ∈ db / env / none。
 
@@ -159,6 +174,16 @@ async def get_setting_origin(key: str) -> tuple[str, str]:
         return ("env", env_value) if env_value else ("none", "")
 
 
+async def clear_runtime_value(key: str) -> None:
+    """清除 DB 覆盖值,回落 .env(调用方需先校验 key 在白名单内)。"""
+    async with AsyncSessionLocal() as db:
+        row = await db.get(AppSetting, key)
+        if row is not None:
+            await db.delete(row)
+            await db.commit()
+    logger.info("运行时配置 %s 已清除覆盖", key)
+
+
 async def migrate_plaintext_sensitive_settings() -> int:
     """一次性迁移:把库中仍是明文的敏感项加密(幂等,启动时执行)。
 
diff --git a/tests/test_events_bronze.py b/tests/test_events_bronze.py
index 00d0392..1a84fb6 100644
--- a/tests/test_events_bronze.py
+++ b/tests/test_events_bronze.py
@@ -72,6 +72,7 @@ class _FakeDB:
             RawEvent: list(raw_events),
         }
         self._teams_by_id: dict[int, Team] = {t.id: t for t in teams if getattr(t, "id", None)}
+        self._teams_by_name: dict[str, Team] = {t.name: t for t in teams if getattr(t, "id", None)}
         self._aliases: dict[str, TeamAlias] = {}
         self._next_id = max((t.id for t in teams if getattr(t, "id", None)), default=0)
 
@@ -86,6 +87,52 @@ class _FakeDB:
         return None
 
     async def execute(self, stmt):
+        # P1-K:LeagueRepository/TeamRepository.get_or_create 使用 pg_insert(Insert 语句,无 column_descriptions)
+        if not hasattr(stmt, "column_descriptions"):
+            # 解析 INSERT 值并注册到 _by_entity,使二次查询可命中
+            try:
+                values = stmt.compile().params
+            except Exception:
+                values = {}
+            table = getattr(stmt, "table", None)
+            table_name = getattr(table, "name", None) if table is not None else None
+            entity_map = {"leagues": League, "teams": Team}
+            entity = entity_map.get(table_name)
+            if entity is not None:
+                obj = entity()
+                for k, v in values.items():
+                    if k in ("code", "name", "country", "name_zh"):
+                        setattr(obj, k, v)
+                if getattr(obj, "id", None) is None:
+                    self._next_id += 1
+                    obj.id = self._next_id
+                if entity not in self._by_entity:
+                    self._by_entity[entity] = []
+                self._by_entity[entity].append(obj)
+                if entity is Team:
+                    self._teams_by_id[obj.id] = obj
+                    self._teams_by_name[obj.name] = obj
+            class _Empty:
+                def scalar_one_or_none(self_inner):
+                    return None
+
+                def scalars(self_inner):
+                    return self_inner
+
+                def all(self_inner):
+                    return []
+
+            return _Empty()
+        # P1-K:Team 查询回退到内存映射 name(避免队列/实体分发不完全匹配)
+        try:
+            params = stmt.compile().params
+        except Exception:
+            params = {}
+        name_val = next((v for k, v in params.items() if k.startswith("name")), None)
+        if isinstance(name_val, str) and name_val:
+            team = self._teams_by_name.get(name_val)
+            if team:
+                return _FakeResult([team])
         entities = set()
         for d in (stmt.column_descriptions or []):
             entities.add(d.get("entity") or d.get("type"))
@@ -94,6 +141,10 @@ class _FakeDB:
                 return _FakeResult(items)
         return _FakeResult([])
 
+    @staticmethod
+    def _filter(entity, items, stmt):
+        return items
+
     async def flush(self):
         for obj in self.added:
             if getattr(obj, "id", None) is None:
@@ -261,8 +312,8 @@ class TestEventsBronzeIsBestEffort:
             raise RuntimeError("infra down")
 
         # Bronze 写入助手直接 import 到 bzzoiro_events 命名空间,需 patch 该处
-        monkeypatch.setattr(bz_events, "_write_raw_event", _boom)
-        monkeypatch.setattr(bz_events, "_write_lineage", _boom)
+        monkeypatch.setattr("src.data.pipeline_write._write_raw_event", _boom)
+        monkeypatch.setattr("src.data.pipeline_write._write_lineage", _boom)
         _patch_fetch(monkeypatch, [_event()])
         db = _FakeDB()
 
diff --git a/tests/test_review_required_fixes.py b/tests/test_review_required_fixes.py
index df12e80..e72961f 100644
--- a/tests/test_review_required_fixes.py
+++ b/tests/test_review_required_fixes.py
@@ -17,7 +17,7 @@ import re
 
 import pytest
 
-from src.db.models import Team, TeamAlias
+from src.db.models import League, Team, TeamAlias
 from src.data.key_ring import _mask
 from src.llm import backtest as bt_mod
 from src.llm.agents import orchestrator as orch_mod
@@ -107,6 +107,7 @@ class _FakeDb:
         self.flush_count = 0
         self._next_id = 1000
         self._teams_by_id: dict[int, Team] = {}
+        self._teams_by_name: dict[str, Team] = {}
         self._aliases: dict[str, TeamAlias] = {}
 
     async def get(self, cls, key):
@@ -117,6 +118,38 @@ class _FakeDb:
         return None
 
     async def execute(self, _stmt):
+        # P1-K:get_or_create 使用 pg_insert(Insert 语句)→ 解析值并注册实体
+        if not hasattr(_stmt, "column_descriptions"):
+            try:
+                values = _stmt.compile().params
+            except Exception:
+                values = {}
+            table = getattr(_stmt, "table", None)
+            table_name = getattr(table, "name", None) if table is not None else None
+            entity_map = {"leagues": League, "teams": Team}
+            entity = entity_map.get(table_name)
+            if entity is not None:
+                obj = entity()
+                for k, v in values.items():
+                    if k in ("code", "name", "country", "name_zh"):
+                        setattr(obj, k, v)
+                if getattr(obj, "id", None) is None:
+                    self._next_id += 1
+                    obj.id = self._next_id
+                if entity is Team:
+                    self._teams_by_id[obj.id] = obj
+                    self._teams_by_name[obj.name] = obj
+            return _FakeResult([])
+        # P1-K:Team 查询回退到内存映射(避免队列耗尽)
+        try:
+            params = _stmt.compile().params
+        except Exception:
+            params = {}
+        name_val = next((v for k, v in params.items() if "name" in k), None)
+        if isinstance(name_val, str) and name_val:
+            team = self._teams_by_name.get(name_val)
+            if team:
+                return _FakeResult([team])
         if self._results:
             return self._results.pop(0)
         return _FakeResult([])
@@ -429,7 +462,7 @@ async def test_r4_dispatch_passes_override_to_specialists(monkeypatch):
 
     monkeypatch.setattr(orch_mod, "load_match_header", _fake_header, raising=True)
     monkeypatch.setattr(orch_mod, "run_specialists", _fake_specialists, raising=True)
-    monkeypatch.setattr(orch_mod, "_upsert_prediction", _fake_upsert, raising=True)
+    monkeypatch.setattr(orch_mod, "_insert_or_find_by_fingerprint", _fake_upsert, raising=True)
     monkeypatch.setattr(orch_mod, "get_uow", _FakeUow, raising=True)
 
     assert get_uow is not None  # 确保 import 生效,session 未被真实打开
diff --git a/tests/test_standings_bronze.py b/tests/test_standings_bronze.py
index 4d2b926..7d9cabf 100644
--- a/tests/test_standings_bronze.py
+++ b/tests/test_standings_bronze.py
@@ -80,6 +80,7 @@ class _FakeDB:
         }
         # session.get 查找表(Team/TeamAlias)
         self._teams_by_id: dict[int, Team] = {t.id: t for t in teams if getattr(t, "id", None)}
+        self._teams_by_name: dict[str, Team] = {t.name: t for t in teams if getattr(t, "id", None)}
         self._aliases: dict[str, TeamAlias] = {}
         self._next_id = max((t.id for t in teams if getattr(t, "id", None)), default=0)
 
@@ -94,6 +95,51 @@ class _FakeDB:
         return None
 
     async def execute(self, stmt):
+        # P1-K:LeagueRepository.get_or_create 使用 pg_insert(Insert 语句,无 column_descriptions)
+        if not hasattr(stmt, "column_descriptions"):
+            # 解析 INSERT 值并注册到 _by_entity,使二次查询可命中
+            try:
+                values = stmt.compile().params
+            except Exception:
+                values = {}
+            # 推断实体类型(League/Team)从 inserted 的表名
+            table = getattr(stmt, "table", None)
+            table_name = getattr(table, "name", None) if table is not None else None
+            entity_map = {"leagues": League, "teams": Team}
+            entity = entity_map.get(table_name)
+            if entity is not None:
+                obj = entity()
+                for k, v in values.items():
+                    if k in ("code", "name", "country", "name_zh"):
+                        setattr(obj, k, v)
+                # 分配 id 并注册
+                if getattr(obj, "id", None) is None:
+                    self._next_id += 1
+                    obj.id = self._next_id
+                if entity not in self._by_entity:
+                    self._by_entity[entity] = []
+                self._by_entity[entity].append(obj)
+                if entity is Team:
+                    self._teams_by_id[obj.id] = obj
+                    self._teams_by_name[obj.name] = obj
+            class _Empty:
+                def scalar_one_or_none(self_inner):
+                    return None
+                def scalars(self_inner):
+                    return self_inner
+                def all(self_inner):
+                    return []
+            return _Empty()
+        # P1-K:Team 查询回退到内存映射 name
+        try:
+            params = stmt.compile().params
+        except Exception:
+            params = {}
+        name_val = next((v for k, v in params.items() if k.startswith("name")), None)
+        if isinstance(name_val, str) and name_val:
+            team = self._teams_by_name.get(name_val)
+            if team:
+                return _FakeResult([team])
         entities = set()
         for d in (stmt.column_descriptions or []):
             entities.add(d.get("entity") or d.get("type"))
@@ -267,8 +313,8 @@ class TestStandingsBronzeIsBestEffort:
             raise RuntimeError("infra down")
 
         # Bronze 写入助手直接 import 到 bzzoiro_standings 命名空间,需 patch 该处
-        monkeypatch.setattr(bz_standings, "_write_raw_event", _boom)
-        monkeypatch.setattr(bz_standings, "_write_lineage", _boom)
+        monkeypatch.setattr("src.data.pipeline_write._write_raw_event", _boom)
+        monkeypatch.setattr("src.data.pipeline_write._write_lineage", _boom)
         _patch_fetch(monkeypatch, _payload())
         db = _FakeDB(leagues=[_preset_league()])