fix(HEL-461): 切换事务失败写入 release-group 审计日志

整组切换中断时除回滚与废弃批次外，同步记录
action=release-group 的失败审计，便于后台追踪。

Co-authored-by: Cursor <cursoragent@cursor.com>
Co-authored-by: multica-agent <github@multica.ai>

xiaobai-datahub/README.md

@@ -91,7 +91,7 @@ python -m datahub history-backfill
 - 流程：组内全部成员先在暂存表完成拉取、字段质量门、覆盖检查和跨数据集交叉校验（`cross_gates` 配置 ts_code 覆盖重叠率下限），全部达标后才在**一个 SQLite 事务**里复制正式表并翻转全部 `publications` 指针。
 - 任一成员失败（拉取失败、质量门拒绝、交叉校验不过、切换事务中断）→ 整批不切换，对外继续提供上一份完整正式版本，失败原因写入 `batches.error` 与 `audit_log`（`action=release-group`），等待晚间自动重试。
 - 读取侧任何时刻只会看到"旧完整版本"或"新完整版本"：发布指针在单事务内统一翻转，容器重启/事务中断自动回滚，不暴露字段残缺或跨数据集混合版本。
-- 幂等：已发布数据集自动跳过；重复执行、并发重试不会生成重复批次或覆盖正常版本（调度器另有 EOD 互斥锁）。
+- 幂等：仅当一致性边界内全部成员都已发布时才整组跳过；边界内任有缺失则整组重暂存后统一切换，避免旧批次与新批次混在同一次重发中。重复执行、并发重试不会在完整边界已就绪时生成重复批次（调度器另有 EOD 互斥锁）。
 
 ## 股票主档每日刷新与发布
 
@@ -115,11 +115,13 @@ python -m datahub moneyflow-backfill        # --trading-days 60 --end-date --for
 
 ```bash
 cd xiaobai-datahub
-python -m datahub eod-refresh --trade-date 20260904                  # 只补缺失数据集
+python -m datahub eod-refresh --trade-date 20260904                  # 补不完整的 A/B 边界
 python -m datahub eod-refresh --trade-date 20260904 --force --dataset valuation
-# 强制重取重发：仍走全部质量门，生成新批次，上一批次保留可回滚
+# --force 按一致性边界整组重发：valuation/daily/moneyflow/auction/stocks → A 组；
+# index_daily → B 组。不可再单独切换某一个正式数据集。
 ```
 
+管理后台「补数」对盘后正式数据集同样走 `force_republish_boundary`，不会绕过 A/B 整批边界。
 
 ## 备份
 

xiaobai-datahub/admin/app.js

@@ -269,7 +269,7 @@ function renderRelease(data) {
 
 async function dangerous(kind, dataset) {
   const date = ($("rel-date") && $("rel-date").value) || "";
-  const ds = dataset || prompt("数据集（daily / valuation / moneyflow / auction / index_daily / reference）", "daily");
+  const ds = dataset || prompt("数据集（daily/valuation/moneyflow/auction/stocks→A组整批；index_daily→B组；或 reference）", "daily");
   if (!ds) return;
   const password = prompt("二次确认：输入管理密码");
   if (!password) return;

xiaobai-datahub/datahub/admin_api.py

@@ -6,7 +6,7 @@ from typing import Any
 from datahub.adapters import RESERVED
 from datahub.auth import AuthService
 from datahub.db import HubDB
-from datahub.pipeline import Pipeline
+from datahub.pipeline import OFFICIAL_DATASETS, STOCKS_DATASET, Pipeline
 from datahub.scheduler import Scheduler
 from datahub.serving import ApiError
 from datahub.timeutil import isoformat, now_shanghai, session_phase, yyyymmdd
@@ -143,8 +143,22 @@ class AdminAPI:
             self._dangerous(password, confirm, f"{dataset}:{day}")
             if dataset == "reference":
                 result = self.pipeline.ingest_reference(day)
+            elif dataset in OFFICIAL_DATASETS or dataset == STOCKS_DATASET:
+                # Manual same-day republish must rebuild the full A/B boundary.
+                # Gate failures and mid-switch exceptions both surface as
+                # FAILED_PRECONDITION so the admin API never leaks raw
+                # transaction errors to the client.
+                try:
+                    result = self.pipeline.force_republish_boundary(dataset, day)
+                    failures = self.pipeline.eod_failures(result)
+                    if failures:
+                        raise ApiError("FAILED_PRECONDITION", "; ".join(failures))
+                except ApiError:
+                    raise
+                except Exception as exc:
+                    raise ApiError("FAILED_PRECONDITION", str(exc)) from exc
             else:
-                result = self.pipeline.run_dataset(dataset, day)
+                raise ApiError("INVALID_ARGUMENT", f"unsupported backfill dataset: {dataset}")
         self.pipeline.audit(actor, "backfill", f"{dataset}:{day}", json.dumps({"ok": True}))
         return result
 

xiaobai-datahub/datahub/cli.py

@@ -7,7 +7,7 @@ import json
 import sys
 
 from datahub.hub import build_hub
-from datahub.pipeline import OFFICIAL_DATASETS
+from datahub.pipeline import EOD_A_DATASETS, OFFICIAL_DATASETS, STOCKS_DATASET
 from datahub.settings import load_settings
 from datahub.timeutil import yyyymmdd
 
@@ -19,15 +19,15 @@ def main(argv: list[str] | None = None) -> int:
     history.add_argument("--calendar-start", default=None, help="日历起点，默认配置 calendar_start")
     history.add_argument("--index-days", type=int, default=None, help="指数回补交易日数量，默认 260")
     history.add_argument("--force", action="store_true", help="覆盖已发布的指数日期")
-    refresh = sub.add_parser("eod-refresh", help="对指定交易日补跑盘后正式数据（跳过已发布数据集，仍走质量门禁）")
+    refresh = sub.add_parser("eod-refresh", help="对指定交易日补跑盘后正式数据（跳过已完整发布的一致性边界，仍走质量门禁）")
     refresh.add_argument("--trade-date", default=None, help="交易日 YYYYMMDD，默认今天")
     refresh.add_argument(
         "--force", action="store_true",
-        help="对 --dataset 指定的数据集强制重取重发（生成新批次，保留上一批次可回滚）",
+        help="强制重发 --dataset 所属的完整一致性边界（A 组或 B 组），生成新批次并保留上一批次可回滚",
     )
     refresh.add_argument(
         "--dataset", default=None,
-        help="配合 --force 使用：只强制重发该数据集（如 valuation）",
+        help="配合 --force：指定边界内任一成员（如 valuation→整组 A；index_daily→整组 B）",
     )
     stocks_refresh = sub.add_parser("stocks-refresh", help="刷新股票主档并发布正式快照（幂等：无变化则跳过）")
     stocks_refresh.add_argument("--trade-date", default=None, help="交易日 YYYYMMDD，默认今天")
@@ -54,26 +54,27 @@ def main(argv: list[str] | None = None) -> int:
     if args.command == "eod-refresh":
         day = yyyymmdd(args.trade_date) if args.trade_date else yyyymmdd()
         if args.force:
-            datasets = tuple(sorted({args.dataset} & OFFICIAL_DATASETS)) if args.dataset else ()
+            allowed = set(OFFICIAL_DATASETS) | {STOCKS_DATASET}
-            if args.dataset and not datasets:
+            if not args.dataset:
-                parser.error(f"unknown dataset: {args.dataset}")
-            if not datasets:
                 parser.error("--force requires --dataset (e.g. --dataset valuation)")
-            result = {}
+            if args.dataset not in allowed:
-            for dataset in datasets:
+                parser.error(f"unknown dataset: {args.dataset}")
-                result[dataset] = hub.pipeline.run_dataset(dataset, day)
+            result = hub.pipeline.force_republish_boundary(args.dataset, day)
+            boundary = "A" if args.dataset in EOD_A_DATASETS or args.dataset == STOCKS_DATASET else "B"
         else:
             result = hub.pipeline.run_eod_missing(day)
+            boundary = None
         hub.pipeline.audit("cli", "eod-refresh", f"eod:{day}", json.dumps(
-            {"force": bool(args.force), "dataset": args.dataset,
+            {"force": bool(args.force), "dataset": args.dataset, "boundary": boundary,
              **{name: item.get("state") for name, item in result.items() if isinstance(item, dict)}},
             ensure_ascii=False,
         ))
         if args.force:
-            payload = {"trade_date": day, "datasets": result}
+            failures = hub.pipeline.eod_failures(result)
+            payload = {"trade_date": day, "boundary": boundary, "datasets": result}
             json.dump(payload, sys.stdout, ensure_ascii=False, indent=2, default=str)
             sys.stdout.write("\n")
-            return 0
+            return 0 if not failures else 1
         missing = hub.pipeline.missing_official_datasets(day)
         payload = {"trade_date": day, "datasets": result, "missing_after": missing}
         json.dump(payload, sys.stdout, ensure_ascii=False, indent=2, default=str)

xiaobai-datahub/datahub/pipeline.py

@@ -348,11 +348,20 @@ class Pipeline:
         is published; ``force`` re-publishes unconditionally. New listings,
         renames (incl. N/C prefix removal) and status changes all flow into
         the snapshot, which carries batch_id/published_at metadata.
+
+        The ``stock_master`` UPSERT happens inside the same publish
+        transaction as the snapshot switch — fetch / quality-gate / switch
+        failures leave the master on the previous complete values.
         """
         day = yyyymmdd(trade_date or self.clock())
-        rows = self._fetch_dataset(STOCKS_DATASET, day)
+        try:
-        with self.db.write() as connection:
+            rows = self._fetch_dataset(STOCKS_DATASET, day)
-            self._upsert_stock_master(connection, rows, isoformat(self.clock()))
+        except Exception as exc:
+            self.audit(
+                "pipeline", "stocks-refresh", f"{STOCKS_DATASET}:{day}",
+                json.dumps({"state": "failed", "error": str(exc)}, ensure_ascii=False),
+            )
+            raise
         if not force:
             active, snapshot = self.published_stock_snapshot(day)
             if active is not None:
@@ -369,7 +378,14 @@ class Pipeline:
                         "batch_id": active,
                         "rows": len(snapshot),
                     }
-        result = self.run_dataset(STOCKS_DATASET, day, prepared_rows=rows)
+        try:
+            result = self.run_dataset(STOCKS_DATASET, day, prepared_rows=rows)
+        except Exception as exc:
+            self.audit(
+                "pipeline", "stocks-refresh", f"{STOCKS_DATASET}:{day}",
+                json.dumps({"state": "failed", "error": str(exc)}, ensure_ascii=False),
+            )
+            raise
         self.audit(
             "pipeline", "stocks-refresh", f"{STOCKS_DATASET}:{day}",
             json.dumps({"batch_id": result["batch_id"], "rows": result["rows"]}, ensure_ascii=False),
@@ -637,53 +653,90 @@ class Pipeline:
         return [dataset for dataset in sorted(OFFICIAL_DATASETS) if dataset not in published]
 
     def run_eod_missing(self, trade_date: str) -> dict[str, Any]:
-        """Fetch/publish every official dataset still missing for the date.
+        """Republish every incomplete EOD consistency group for the date.
 
-        Idempotent: datasets with an existing publication are skipped, so
+        A-group (daily/valuation/moneyflow/auction + stocks) and B-group
-        repeats never overwrite the current official batch. All missing
+        (index_daily) are separate boundaries. Within a group, either the
-        members stage first and switch in one atomic release group; a single
+        whole boundary is already published (idempotent skip) or every
-        member failure keeps the previous complete official version serving.
+        member is re-staged and switched together — never fill only the
+        missing members on top of older batches from an earlier partial run.
         """
-        return self.run_release_group(tuple(sorted(OFFICIAL_DATASETS)), trade_date, include_stocks=True)
+        results: dict[str, Any] = {}
+        results.update(self.run_eod_batch_a(trade_date))
+        results.update(self.run_eod_batch_b(trade_date))
+        return results
 
-    def run_eod_batch_a(self, trade_date: str) -> dict[str, Any]:
+    def run_eod_batch_a(self, trade_date: str, force: bool = False) -> dict[str, Any]:
-        return self.run_release_group(EOD_A_DATASETS, trade_date, include_stocks=True)
+        return self.run_release_group(EOD_A_DATASETS, trade_date, include_stocks=True, force=force)
 
-    def run_eod_batch_b(self, trade_date: str) -> dict[str, Any]:
+    def run_eod_batch_b(self, trade_date: str, force: bool = False) -> dict[str, Any]:
-        return self.run_release_group(EOD_B_DATASETS, trade_date)
+        return self.run_release_group(EOD_B_DATASETS, trade_date, force=force)
+
+    def force_republish_boundary(self, dataset: str, trade_date: str) -> dict[str, Any]:
+        """Force-republish the full A/B consistency boundary that owns ``dataset``.
+
+        CLI ``eod-refresh --force`` and admin manual backfill must not publish a
+        single official member alone — that would mix old and new batches inside
+        the same trade date. Naming any A-group member (or stocks) rebuilds the
+        whole A group; naming ``index_daily`` rebuilds B.
+        """
+        name = str(dataset or "").strip()
+        if name in EOD_A_DATASETS or name == STOCKS_DATASET:
+            return self.run_eod_batch_a(trade_date, force=True)
+        if name in EOD_B_DATASETS:
+            return self.run_eod_batch_b(trade_date, force=True)
+        raise ValueError(f"dataset is not part of an EOD release boundary: {dataset}")
 
     def run_release_group(
         self,
         datasets: tuple[str, ...],
         trade_date: str,
         include_stocks: bool = False,
+        force: bool = False,
     ) -> dict[str, Any]:
         """One post-market publish/republish becomes one atomic visibility flip.
 
         Consistency boundary: every member (official datasets, plus the daily
-        stocks snapshot when ``include_stocks`` and not yet published) is
+        stocks snapshot when ``include_stocks``) is fetched, staged,
-        fetched, staged, field-gated and cross-validated BEFORE any reader can
+        field-gated and cross-validated BEFORE any reader can see it. Only
-        see it. Only when the whole group passes does a single SQLite
+        when the whole group passes does a single SQLite transaction copy
-        transaction copy all staging batches to the official tables and flip
+        all staging batches to the official tables and flip every
-        every ``publications`` row at once. Any member failure aborts the
+        ``publications`` row at once. Any member failure aborts the group:
-        group: the previous complete official version keeps serving and the
+        the previous complete official version keeps serving and the reason
-        reason is recorded on the batches and in the audit log.
+        is recorded on the batches and in the audit log.
+
+        Skip is all-or-nothing for the boundary unless ``force``: if every
+        official member (and stocks when required) is already published, the
+        group is skipped. If any official member is still missing — or
+        ``force`` is set — every official member is re-staged, so a retry or
+        manual republish never mixes old and new batches in one release.
         """
         day = yyyymmdd(trade_date)
         results: dict[str, Any] = {}
         staged: dict[str, dict[str, Any]] = {}
         failure: str | None = None
-        pending: list[str] = []
+        missing_official = [dataset for dataset in datasets if self.active_batch(dataset, day) is None]
-        for dataset in datasets:
+        stocks_missing = include_stocks and self.active_batch(STOCKS_DATASET, day) is None
-            if self.active_batch(dataset, day) is not None:
+
+        if not force and not missing_official and not stocks_missing:
+            for dataset in datasets:
                 results[dataset] = {
                     "dataset": dataset,
                     "trade_date": day,
                     "state": "skipped",
                     "reason": "already_published",
                 }
-            else:
+            if include_stocks:
-                pending.append(dataset)
+                results[STOCKS_DATASET] = {
+                    "dataset": STOCKS_DATASET,
+                    "trade_date": day,
+                    "state": "skipped",
+                    "reason": "already_published",
+                }
+            return results
+
+        # Incomplete or forced boundary → restage every official member together.
+        pending = list(datasets)
 
         for dataset in pending:
             if failure is not None:
@@ -705,9 +758,13 @@ class Pipeline:
                     "error": str(exc),
                 }
 
-        if include_stocks and failure is None and self.active_batch(STOCKS_DATASET, day) is None:
+        # Stocks join the same switch when the official boundary is being
+        # rebuilt (missing or forced), or when only the stocks snapshot is
+        # still missing.
+        rebuild_official = bool(force or missing_official)
+        if include_stocks and failure is None and (rebuild_official or stocks_missing):
             try:
-                stocks_plan = self._stage_stocks_snapshot(day)
+                stocks_plan = self._stage_stocks_snapshot(day, force=rebuild_official)
             except Exception as exc:
                 failure = f"{STOCKS_DATASET}: {exc}"
                 results[STOCKS_DATASET] = {
@@ -756,8 +813,32 @@ class Pipeline:
             try:
                 self._switch_release_group(day, staged)
             except Exception as exc:
+                reason = f"release group switch failed: {exc}"
                 for item in staged.values():
-                    self._abandon_batch(item["batch_id"], f"release group switch failed: {exc}")
+                    self._abandon_batch(item["batch_id"], reason)
+                LOGGER.warning(
+                    "release group switch failed, previous official version keeps serving",
+                    extra={
+                        "hub": {
+                            "trade_date": day,
+                            "datasets": sorted(staged),
+                            "reason": reason,
+                            "event": "release_group_switch_failed",
+                        }
+                    },
+                )
+                self.audit(
+                    "pipeline", "release-group", f"eod:{day}",
+                    json.dumps(
+                        {
+                            "state": "failed",
+                            "reason": reason,
+                            "switched": [],
+                            "force": bool(force),
+                        },
+                        ensure_ascii=False,
+                    ),
+                )
                 raise
             for dataset, item in staged.items():
                 results[dataset] = {
@@ -770,7 +851,8 @@ class Pipeline:
             self.audit(
                 "pipeline", "release-group", f"eod:{day}",
                 json.dumps(
-                    {"state": "ok", "switched": sorted(staged)}, ensure_ascii=False
+                    {"state": "ok", "switched": sorted(staged), "force": bool(force)},
+                    ensure_ascii=False,
                 ),
             )
         return results
@@ -820,18 +902,18 @@ class Pipeline:
             "state": "degraded" if report["soft_fail"] else "published",
         }
 
-    def _stage_stocks_snapshot(self, trade_date: str) -> dict[str, Any] | None:
+    def _stage_stocks_snapshot(self, trade_date: str, force: bool = False) -> dict[str, Any] | None:
         """Stage the daily stocks snapshot for a release group switch.
 
         Returns None when the published snapshot is already identical to
-        upstream (idempotent skip). The stock_master upsert is deferred into
+        upstream (idempotent skip) unless ``force`` is set. The stock_master
-        the group switch transaction so the master never runs ahead of the
+        upsert is deferred into the group switch / publish transaction so the
-        published snapshot.
+        master never runs ahead of the published snapshot.
         """
         day = yyyymmdd(trade_date)
         active, snapshot = self.published_stock_snapshot(day)
         rows = self._fetch_dataset(STOCKS_DATASET, day)
-        if active is not None:
+        if active is not None and not force:
             upstream = sorted(
                 tuple(str(row.get(field)) for field in STOCK_SNAPSHOT_FIELDS) for row in rows
             )
@@ -1077,6 +1159,16 @@ class Pipeline:
         with self.db.write() as connection:
             _staging_count_or_raise(connection, dataset, trade_date, batch_id)
             connection.execute(EOD_COPY[dataset], (batch_id,))
+            if dataset == STOCKS_DATASET:
+                staging = DATASET_TABLES[STOCKS_DATASET][1]
+                stock_rows = [
+                    dict(row)
+                    for row in connection.execute(
+                        f"SELECT * FROM {staging} WHERE batch_id = ?",
+                        (batch_id,),
+                    ).fetchall()
+                ]
+                self._upsert_stock_master(connection, stock_rows, published_at)
             if self.before_commit:
                 self.before_commit()
             _upsert_publication(connection, dataset, trade_date, batch_id, state, published_at)

xiaobai-datahub/tests/test_atomic_release.py

@@ -21,14 +21,14 @@ class GroupTransport:
     def __init__(self) -> None:
         self.empty: set[str] = set()
         self.keep_rows: dict[str, int] = {}
-        self.stocks: list[dict] = []
+        self.stocks: list[dict] | None = None
         self.calls: list[str] = []
 
     def __call__(self, api_name: str, params: dict, fields: str):
         self.calls.append(api_name)
         if api_name in self.empty:
             return []
-        if api_name == "stock_basic" and self.stocks:
+        if api_name == "stock_basic" and self.stocks is not None:
             return [dict(row) for row in self.stocks]
         rows = fake_transport(api_name, params, fields)
         keep = self.keep_rows.get(api_name)
@@ -113,19 +113,48 @@ class ReleaseGroupSwitchTests(unittest.TestCase):
         first = self.pipe.run_dataset("daily", TRADE_DATE)
         self.transport.empty = {"daily_basic"}
         results = self.pipe.run_eod_missing(TRADE_DATE)
-        self.assertEqual(results["daily"]["state"], "skipped")
+        # incomplete A-group restages daily with the others; valuation fails → no A switch
+        self.assertEqual(results["daily"]["state"], "failed")
         self.assertEqual(results["valuation"]["state"], "failed")
-        # the already-published complete batch is untouched and keeps serving
+        # the already-published daily batch is untouched and keeps serving
         self.assertEqual(self.pipe.active_batch("daily", TRADE_DATE), first["batch_id"])
-        self.assertEqual(
+        pubs = publications_map(self.db, TRADE_DATE)
-            publications_map(self.db, TRADE_DATE),
+        self.assertEqual(pubs["daily"], first["batch_id"])
-            {"daily": first["batch_id"]},
+        self.assertNotIn("valuation", pubs)
-        )
+        self.assertNotIn("moneyflow", pubs)
+        self.assertNotIn("auction", pubs)
+        # B-group is an independent boundary and may still publish
+        self.assertEqual(results["index_daily"]["state"], "published")
         payload = V1API(self.db, self.pipe, self.pipe.settings).handle(
             "/v1/bars/daily", {"date": [TRADE_DATE], "code": ["600000.SH"]}
         )
         self.assertEqual(payload["meta"]["batch_id"], first["batch_id"])
 
+    def test_partial_group_retry_does_not_mix_batches(self) -> None:
+        """Already-published A members must be restaged with missing ones."""
+        first_daily = self.pipe.run_dataset("daily", TRADE_DATE)
+        first_moneyflow = self.pipe.run_dataset("moneyflow", TRADE_DATE)
+        results = self.pipe.run_eod_missing(TRADE_DATE)
+        # A-group switched as one boundary; B-group (index) also published
+        for name in (*GROUP_A, "stocks"):
+            self.assertEqual(results[name]["state"], "published", name)
+        self.assertEqual(results["index_daily"]["state"], "published")
+        pubs = self.db.fetchall(
+            "SELECT dataset, active_batch, published_at FROM publications WHERE trade_date = ?",
+            (TRADE_DATE,),
+        )
+        by_ds = {str(row["dataset"]): row for row in pubs}
+        # old partial batches replaced — no cross-batch mix of the first wave
+        self.assertNotEqual(by_ds["daily"]["active_batch"], first_daily["batch_id"])
+        self.assertNotEqual(by_ds["moneyflow"]["active_batch"], first_moneyflow["batch_id"])
+        a_times = {by_ds[name]["published_at"] for name in (*GROUP_A, "stocks")}
+        self.assertEqual(len(a_times), 1)
+        # serving resolves the new complete A-group batches
+        api = V1API(self.db, self.pipe, self.pipe.settings)
+        daily = api.handle("/v1/bars/daily", {"date": [TRADE_DATE], "code": ["600000.SH"]})
+        self.assertEqual(daily["meta"]["batch_id"], results["daily"]["batch_id"])
+        self.assertEqual(daily["meta"]["batch_id"], by_ds["daily"]["active_batch"])
+
     def test_reads_during_switch_see_old_state_until_commit(self) -> None:
         snapshots: list[dict] = []
 
@@ -155,6 +184,13 @@ class ReleaseGroupSwitchTests(unittest.TestCase):
         for table in ("eod_bars", "eod_valuation", "eod_moneyflow", "eod_auction", "eod_stocks"):
             rows = self.db.fetchall(f"SELECT * FROM {table} WHERE trade_date = ?", (TRADE_DATE,))
             self.assertEqual(rows, [], table)
+        audit = self.db.fetchone(
+            "SELECT * FROM audit_log WHERE action = 'release-group' ORDER BY id DESC"
+        )
+        self.assertIsNotNone(audit)
+        detail = str(audit["detail"])
+        self.assertIn("killed mid-switch", detail)
+        self.assertIn("failed", detail)
 
     def test_duplicate_runs_are_idempotent(self) -> None:
         self.pipe.run_eod_batch_a(TRADE_DATE)
@@ -231,5 +267,142 @@ class ReleaseGroupSwitchTests(unittest.TestCase):
         self.assertIsNone(stocks_pub)
 
 
+class StocksRefreshAtomicTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.transport = GroupTransport()
+        self.pipe, self.db = make_pipe(self.transport)
+        self.pipe.ingest_reference(TRADE_DATE)
+        self.transport.stocks = [
+            {"ts_code": "600000.SH", "symbol": "600000", "name": "浦发银行", "area": "上海",
+             "industry": "银行", "market": "主板", "list_status": "L", "list_date": "19991110"},
+            {"ts_code": "920071.BJ", "symbol": "920071", "name": "N金钛", "area": "辽宁",
+             "industry": "小金属", "market": "北交所", "list_status": "L", "list_date": "20240901"},
+        ]
+        first = self.pipe.refresh_stocks(TRADE_DATE)
+        self.assertEqual(first["state"], "published")
+        self.first_batch = first["batch_id"]
+
+    def test_refresh_keeps_master_when_snapshot_publish_fails(self) -> None:
+        self.transport.stocks = [
+            {"ts_code": "600000.SH", "symbol": "600000", "name": "浦发银行", "area": "上海",
+             "industry": "银行", "market": "主板", "list_status": "L", "list_date": "19991110"},
+            {"ts_code": "920071.BJ", "symbol": "920071", "name": "金钛股份", "area": "辽宁",
+             "industry": "小金属", "market": "北交所", "list_status": "L", "list_date": "20240901"},
+        ]
+
+        def explode() -> None:
+            raise RuntimeError("snapshot switch killed")
+
+        self.pipe.before_commit = explode
+        with self.assertRaises(RuntimeError):
+            self.pipe.refresh_stocks(TRADE_DATE)
+        master = self.db.fetchone("SELECT name FROM stock_master WHERE ts_code = '920071.BJ'")
+        self.assertEqual(master["name"], "N金钛")  # rename not applied
+        self.assertEqual(self.pipe.active_batch("stocks", TRADE_DATE), self.first_batch)
+        audit = self.db.fetchone(
+            "SELECT * FROM audit_log WHERE action = 'stocks-refresh' ORDER BY id DESC"
+        )
+        self.assertIn("failed", str(audit["detail"]))
+        self.assertIn("snapshot switch killed", str(audit["detail"]))
+
+    def test_refresh_keeps_master_when_quality_gate_rejects(self) -> None:
+        self.transport.stocks = []  # empty → hard fail before publish
+        with self.assertRaises(Exception):
+            self.pipe.refresh_stocks(TRADE_DATE)
+        master = self.db.fetchone("SELECT name FROM stock_master WHERE ts_code = '920071.BJ'")
+        self.assertEqual(master["name"], "N金钛")
+        self.assertEqual(self.pipe.active_batch("stocks", TRADE_DATE), self.first_batch)
+        audit = self.db.fetchone(
+            "SELECT * FROM audit_log WHERE action = 'stocks-refresh' ORDER BY id DESC"
+        )
+        self.assertIn("failed", str(audit["detail"]))
+
+
+class ForceBoundaryEntryTests(unittest.TestCase):
+    """CLI force / admin backfill must rebuild the full A/B boundary."""
+
+    def setUp(self) -> None:
+        self.transport = GroupTransport()
+        self.pipe, self.db = make_pipe(self.transport)
+        self.pipe.ingest_reference(TRADE_DATE)
+        self.first = self.pipe.run_eod_batch_a(TRADE_DATE)
+        self.pipe.run_eod_batch_b(TRADE_DATE)
+
+    def test_force_republish_valuation_rebuilds_whole_a_group(self) -> None:
+        before = publications_map(self.db, TRADE_DATE)
+        results = self.pipe.force_republish_boundary("valuation", TRADE_DATE)
+        self.assertEqual({item["state"] for item in results.values()}, {"published"})
+        after = publications_map(self.db, TRADE_DATE)
+        for name in (*GROUP_A, "stocks"):
+            self.assertNotEqual(after[name], before[name], name)
+            self.assertEqual(after[name], results[name]["batch_id"], name)
+        # B-group left alone
+        self.assertEqual(after["index_daily"], before["index_daily"])
+        pubs = self.db.fetchall(
+            "SELECT dataset, published_at FROM publications WHERE trade_date = ?",
+            (TRADE_DATE,),
+        )
+        a_times = {row["published_at"] for row in pubs if row["dataset"] in {*GROUP_A, "stocks"}}
+        self.assertEqual(len(a_times), 1)
+
+    def test_force_republish_index_rebuilds_only_b_group(self) -> None:
+        before = publications_map(self.db, TRADE_DATE)
+        results = self.pipe.force_republish_boundary("index_daily", TRADE_DATE)
+        self.assertEqual(results["index_daily"]["state"], "published")
+        after = publications_map(self.db, TRADE_DATE)
+        self.assertNotEqual(after["index_daily"], before["index_daily"])
+        for name in GROUP_A:
+            self.assertEqual(after[name], before[name], name)
+
+    def test_admin_backfill_official_dataset_uses_boundary(self) -> None:
+        from datahub.admin_api import AdminAPI
+        from datahub.auth import AuthService
+        from datahub.crypto import SecretVault
+        from datahub.scheduler import Scheduler
+        from datahub.serving import ApiError
+
+        vault = SecretVault(self.pipe.settings.encryption_key)
+        auth = AuthService(self.db, vault, self.pipe.settings.api_token, "StartPass1")
+        admin = AdminAPI(self.db, self.pipe, Scheduler(self.db, self.pipe), auth)
+        before = publications_map(self.db, TRADE_DATE)
+        result = admin.backfill("moneyflow", TRADE_DATE, "StartPass1", f"moneyflow:{TRADE_DATE}", "tester")
+        self.assertEqual(result["moneyflow"]["state"], "published")
+        after = publications_map(self.db, TRADE_DATE)
+        for name in (*GROUP_A, "stocks"):
+            self.assertNotEqual(after[name], before[name], name)
+        # bad password / wrong confirm still rejected
+        with self.assertRaises(ApiError):
+            admin.backfill("daily", TRADE_DATE, "wrong", f"daily:{TRADE_DATE}", "tester")
+
+    def test_admin_backfill_switch_crash_is_failed_precondition(self) -> None:
+        from datahub.admin_api import AdminAPI
+        from datahub.auth import AuthService
+        from datahub.crypto import SecretVault
+        from datahub.scheduler import Scheduler
+        from datahub.serving import ApiError
+
+        vault = SecretVault(self.pipe.settings.encryption_key)
+        auth = AuthService(self.db, vault, self.pipe.settings.api_token, "StartPass1")
+        admin = AdminAPI(self.db, self.pipe, Scheduler(self.db, self.pipe), auth)
+        before = publications_map(self.db, TRADE_DATE)
+
+        def explode() -> None:
+            raise RuntimeError("killed mid-switch")
+
+        self.pipe.before_commit = explode
+        with self.assertRaises(ApiError) as ctx:
+            admin.backfill("valuation", TRADE_DATE, "StartPass1", f"valuation:{TRADE_DATE}", "tester")
+        self.assertEqual(ctx.exception.code, "FAILED_PRECONDITION")
+        self.assertIn("killed mid-switch", ctx.exception.message)
+        # previous complete A/B versions keep serving
+        self.assertEqual(publications_map(self.db, TRADE_DATE), before)
+        audit = self.db.fetchone(
+            "SELECT * FROM audit_log WHERE action = 'release-group' ORDER BY id DESC"
+        )
+        self.assertIsNotNone(audit)
+        self.assertIn("failed", str(audit["detail"]))
+        self.assertIn("killed mid-switch", str(audit["detail"]))
+
+
 if __name__ == "__main__":
     unittest.main()

xiaobai-datahub/tests/test_quality_gates.py

@@ -228,25 +228,34 @@ class GateRetryInterplayTests(unittest.TestCase):
 
 
 class ForceRepublishTests(unittest.TestCase):
-    def test_run_dataset_over_published_keeps_prev_for_rollback(self) -> None:
+    def test_force_boundary_republish_keeps_prev_for_rollback(self) -> None:
         transport = ValuationTransport()
         pipe, db = make_pipe(transport)
         pipe.ingest_reference(TRADE_DATE)
-        first = pipe.run_dataset("valuation", TRADE_DATE)
+        first = pipe.run_eod_batch_a(TRADE_DATE)
+        first_val = first["valuation"]["batch_id"]
+        first_daily = first["daily"]["batch_id"]
         transport.mode = "vr_all_null"
-        with self.assertRaises(QualityError):
+        blocked = pipe.force_republish_boundary("valuation", TRADE_DATE)
-            pipe.run_dataset("valuation", TRADE_DATE)  # gate holds: bad re-publish refused
+        self.assertEqual(blocked["valuation"]["state"], "failed")
+        self.assertEqual(pipe.active_batch("valuation", TRADE_DATE), first_val)
+        self.assertEqual(pipe.active_batch("daily", TRADE_DATE), first_daily)
         transport.mode = "ok"
-        second = pipe.run_dataset("valuation", TRADE_DATE)  # CLI --force path
+        second = pipe.force_republish_boundary("valuation", TRADE_DATE)
-        self.assertNotEqual(first["batch_id"], second["batch_id"])
+        self.assertEqual(second["valuation"]["state"], "published")
-        pub = db.fetchone(
+        self.assertNotEqual(second["valuation"]["batch_id"], first_val)
-            "SELECT * FROM publications WHERE dataset='valuation' AND trade_date=?",
+        self.assertNotEqual(second["daily"]["batch_id"], first_daily)
+        pubs = db.fetchall(
+            "SELECT dataset, active_batch, prev_batch, published_at FROM publications WHERE trade_date=?",
             (TRADE_DATE,),
         )
-        self.assertEqual(pub["active_batch"], second["batch_id"])
+        by_ds = {str(row["dataset"]): row for row in pubs}
-        self.assertEqual(pub["prev_batch"], first["batch_id"])
+        a_times = {by_ds[name]["published_at"] for name in ("daily", "valuation", "moneyflow", "auction", "stocks")}
+        self.assertEqual(len(a_times), 1)
+        self.assertEqual(by_ds["valuation"]["active_batch"], second["valuation"]["batch_id"])
+        self.assertEqual(by_ds["valuation"]["prev_batch"], first_val)
         rolled = pipe.rollback("valuation", TRADE_DATE, actor="cli")
-        self.assertEqual(rolled["active_batch"], first["batch_id"])
+        self.assertEqual(rolled["active_batch"], first_val)
 
 
 if __name__ == "__main__":