fix(HEL-562): 人气榜批内已合并重复键不再硬失败

扩展软数据集（popularity 等）在 staging 已按业务键去重后，
quality gate 仍把原始抓取的 duplicate keys 记为 hard_fail，
导致 20260915 人气榜 integrity_gate 拒发。现降为 warning（soft_fail/
degraded 仍可发布），核心七类与 moneyflow/auction 判重口径不变。

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

xiaobai-datahub/admin/app.js

@@ -162,6 +162,30 @@ function hm(iso) {
   if (idx < 0 || s.length < idx + 6) return s.slice(0, 5) || '—';
   return s.slice(idx + 1, idx + 6);
 }
+/* 大白话错误摘要：主表只放一行短句，完整原文放 title 悬停提示。
+   不猜业务结论，只做确定性文案映射 + 定长截断。 */
+function plainError(raw, max = 18) {
+  const s = String(raw || '');
+  if (!s) return '';
+  const rules = [
+    [/UNIQUE constraint failed/i, '重复数据写入冲突'],
+    [/circuit open|breaker/i, '熔断保护中'],
+    [/timeout|timed out|ETIMEDOUT/i, '请求超时'],
+    [/row ratio/i, '行数校验未达标'],
+    [/field gate/i, '字段完整率未达标'],
+    [/empty official batch|no_rows_returned|empty_or_incomplete/i, '上游返回空数据'],
+    [/release group not switched/i, '发布组未切换'],
+    [/missing_fields/i, '返回缺字段'],
+    [/data_age_exceeds/i, '数据过期超阈值'],
+    [/blocked|captcha|安全验证|访问异常/i, '被上游拦截'],
+  ];
+  for (const [re, text] of rules) if (re.test(s)) return text;
+  return s.length > max ? s.slice(0, max) + '…' : s;
+}
+/* 单行截断单元格：class ellip（styles.css 生产补充段），title 放全文 */
+function ellipHtml(text, color, fontSize, title) {
+  return `<span class="ellip" style="color:${color};font-size:${fontSize || ''}" title="${esc(title ?? text)}">${esc(text)}</span>`;
+}
 const p2 = (x) => String(x).padStart(2, '0');
 
 /* ================= sparkline（真实历史：来自 provider_call_log） ================= */
@@ -377,7 +401,7 @@ function heroHtml(vm) {
     `</div><span class="flex1"></span>` +
     `<div class="kpis">` +
     kpiHtml('已发布数据集', `<span class="glow-mint">${vm.publishedCount}</span><span style="color:#54637e;font-size:18px">/${vm.totalOfficial}</span>`, vm.unpublished.length ? `${vm.unpublished.map((d) => DATASET_NAME[d] || d).join(' · ')} 未发布` : '全部已发布') +
-    kpiHtml('今日调用成功率', vm.successRate == null ? '<span style="color:#54637e">—</span>' : `<span>${vm.successRate.toFixed(1)}%</span>`, null, 'mint', vm.successRate != null) +
+    kpiHtml('今日调用成功率', vm.successRate == null ? '<span style="color:#54637e">—</span>' : `<span data-rate>${vm.successRate.toFixed(1)}%</span>`, null, 'mint', vm.successRate != null) +
     kpiHtml('主站回退页面', `${vm.fallbackPages.length} 页`, vm.fallbackPages.length ? esc(vm.fallbackPages[0]) : '实时层主源正常', vm.fallbackPages.length ? 'amb' : 'txt') +
     kpiHtml('EOD 尝试', `${vm.eod.attempts || 0} 次`, '上限 5 · 23:30 止', vm.eod.attempts ? 'rd' : 'txt') +
     `</div></div>`;
@@ -389,7 +413,7 @@ function datasetStripHtml(vm) {
     let note = '';
     if (d.time) note += `<span class="num" style="color:#8b9bb4">${esc(d.time)} · </span>`;
     if (d.rows) note += `<span class="num">${Number(d.rows).toLocaleString('en-US')} 行</span>`;
-    if (d.err) { const brief = String(d.err).length > 22 ? String(d.err).slice(0, 22) + '…' : d.err; note += `<span style="color:#f87171" title="${esc(d.err)}">${esc(brief)}</span>`; }
+    if (d.err) note += `<span class="ellip" style="color:#f87171" title="${esc(d.err)}">${esc(plainError(d.err, 16))}</span>`;
     if (!d.time && !d.rows && !d.err) note = '<span style="color:#54637e">今日尚未发布</span>';
     return `<div class="panel dsc ${ring}" data-dataset="${d.id}">` +
       `<div class="dsc-top"><span class="dsc-name">${esc(d.name)}</span>${pillHtml(d.bucket)}</div>` +
@@ -403,11 +427,11 @@ function observatoryHtml(vm) {
   const right = `<span class="legend">${Object.keys(legendByProv).map((p) => `<span>${ledHtml('ok')}${legendByProv[p]}</span>`).join('')}</span>`;
   const rows = vm.observers.length ? vm.observers.map((o) => `<tr>` +
     `<td style="color:#e8f1ff">${esc(o.name)}</td>` +
-    `<td style="color:#8b9bb4">${o.provider === 'eastmoney' ? '东方财富' : '腾讯行情'}</td>` +
+    `<td style="color:#8b9bb4;white-space:nowrap">${o.provider === 'eastmoney' ? '东方财富' : '腾讯行情'}</td>` +
     `<td>${ageHtml(o.lastOk)}</td>` +
     `<td>${latHtml(o.lat, o.bucket !== 'ok')}</td>` +
     `<td>${sparkHtml(o.jid, o.bucket === 'slow' ? 'amb' : 'mint', 64, 16)}</td>` +
-    `<td><span style="display:flex;align-items:center;gap:8px">${pillHtml(o.bucket)}${o.note ? `<span class="obs-note">${esc(o.note)}</span>` : ''}</span></td>` +
+    `<td style="max-width:260px"><span style="display:flex;align-items:center;gap:8px;min-width:0">${pillHtml(o.bucket)}${o.note ? ellipHtml(plainError(o.note, 20), '', '10px', o.note) : ''}</span></td>` +
     `</tr>`).join('') : `<tr><td colspan="6" class="empty-hint">尚无真实观测记录（等待首次探测/调用）</td></tr>`;
   const body = `<table class="dtable"><thead><tr><th>观察项</th><th>当前来源</th><th>缓存年龄</th><th>延迟</th><th>趋势</th><th>状态 · 备注</th></tr></thead><tbody>${rows}</tbody></table>`;
   return panelHtml('实时观察层', right, body, false);
@@ -418,7 +442,7 @@ function siteImpactHtml(vm) {
   const rows = bad.length ? bad.slice(0, 5).map((s) => `<tr>` +
     `<td style="color:#e8f1ff;width:40%">${esc(s.page)}</td>` +
     `<td style="width:20%">${pillHtml(s.bucket)}</td>` +
-    `<td style="font-size:11px;color:#54637e">${s.note ? esc(s.note) : '—'}</td>` +
+    `<td style="font-size:11px;color:#54637e;max-width:220px">${s.note ? ellipHtml(plainError(s.note, 22), '#54637e', '11px', s.note) : '—'}</td>` +
     `<td style="text-align:right"><button class="tbtn mini" data-goto-lineage="${esc(s.dataset)}">查看血缘</button></td>` +
     `</tr>`).join('') : `<tr><td style="color:#e8f1ff;width:40%">全部页面正常</td><td>${pillHtml('ok')}</td><td style="font-size:11px;color:#54637e">—</td><td></td></tr>`;
   const footPages = vm.siteImpact.map((s) => ({ name: s.short, bucket: s.bucket }));
@@ -479,7 +503,7 @@ function incidentsHtml(vm) {
     return `<div class="inc ${it.sev}">${tag}` +
       `<div style="flex:1;min-width:0">` +
       `<div class="inc-title"${it.sev === 'off' ? ' style="color:#8b9bb4"' : ''}>${esc(it.title)}</div>` +
-      `<div class="inc-meta">${meta}</div>` +
+      `<div class="inc-meta ellip" title="${esc(String(it.meta).replace(/<[^>]+>/g, ''))}">${meta}</div>` +
       `</div>` +
       `<button class="tbtn" style="flex:none" data-incident-action data-dataset="${esc(it.dataset || '')}" data-provider="${esc(it.provider || '')}">去处理 →</button>` +
       `</div>`;
@@ -589,7 +613,7 @@ function sourceCardHtml(card) {
     `<span>接口 <span class="num" style="color:${card.bucket === 'off' ? '#54637e' : '#22d3ee'}">${card.usedCount}/${card.totalCount}</span> 在用</span>` +
     `<span class="msep">|</span><span>最后探测 ${probeCell}</span>` +
     `<span class="msep">|</span><span>延迟 ${latHtml(card.latency)}</span>` +
-    (card.error ? `<span class="msep">|</span><span style="color:#f87171">最近错误 ${esc(card.error)}</span>` : (card.bucket !== 'off' ? `<span class="msep">|</span><span style="color:#54637e">无最近错误</span>` : '')) +
+    (card.error ? `<span class="msep">|</span><span style="color:#f87171;display:inline-flex;align-items:center;gap:4px;min-width:0">最近错误 ${ellipHtml(plainError(card.error, 26), '#f87171', '11px', card.error)}</span>` : (card.bucket !== 'off' ? `<span class="msep">|</span><span style="color:#54637e">无最近错误</span>` : '')) +
     `</div>`;
   if (ui.open && card.bucket !== 'off') {
     h += `<div class="mx-grid">` + card.groups.map((g) => `<div style="min-width:0">` +
@@ -600,7 +624,7 @@ function sourceCardHtml(card) {
         `<td style="color:#8b9bb4;font-size:11px;white-space:nowrap">${esc(it.use)}</td>` +
         `<td class="num" style="font-size:11px;color:${it.ok.includes('×') || it.ok === '昨日' ? '#f87171' : '#8b9bb4'}">${esc(it.ok)}</td>` +
         `<td style="font-size:11px">${latHtml(it.lat)}</td>` +
-        `<td style="white-space:nowrap"><span style="display:flex;align-items:center;gap:6px">${pillHtml(it.st, it.st === 'plan' && it.obs === false ? '已配置 · 待观测' : '')}${it.err ? `<span style="font-size:10px;color:#f87171">${esc(it.err)}</span>` : ''}</span></td>` +
+        `<td style="white-space:nowrap"><span style="display:flex;align-items:center;gap:6px">${pillHtml(it.st, it.st === 'plan' && it.obs === false ? '已配置 · 待观测' : '')}${it.err ? ellipHtml(plainError(it.err, 16), '#f87171', '10px', it.err) : ''}</span></td>` +
         `</tr>`).join('') +
       `</tbody></table></div>`).join('') + `</div>`;
   }
@@ -765,7 +789,7 @@ function lineageGraphHtml(vm) {
     const p = gpos['s:' + s.id]; if (!p) return;
     h += `<g data-nid="${esc(s.id)}" data-gkind="src" style="cursor:pointer;transition:opacity .18s">` +
       `<circle cx="${p.x}" cy="${p.y}" r="16" fill="rgba(34,211,238,.05)" stroke="#243152"/>` +
-      `<circle cx="${p.x}" cy="${p.y}" r="5" fill="${stColorOf(s.st)}"${s.st !== 'off' ? ' class="gnode-pulse"' : ''}/>` +
+      `<circle cx="${p.x}" cy="${p.y}" r="5" fill="${stColorOf(s.st)}"${s.st !== 'off' ? ` class="gnode-pulse" style="animation-delay:${(vm.sources.indexOf(s) * 0.37).toFixed(2)}s"` : ''}/>` +
       `<text x="${p.x + 24}" y="${p.y + 4}" font-size="12" fill="#d7e1f0" font-weight="600">${esc(s.name)}</text></g>`;
   });
   vm.datasets.forEach((d) => {
@@ -838,7 +862,7 @@ function renderLinRows(vm) {
     `<td class="num" style="color:#8b9bb4;font-size:11px;white-space:nowrap">${esc(r.freq)}</td>` +
     `<td class="num" style="font-size:11px;white-space:nowrap;color:#8b9bb4">${r.ok ? esc(hm(r.ok)) : '—'}</td>` +
     `<td class="num" style="font-size:11px;color:${r.lat != null && r.lat > 1000 ? '#fbbf24' : '#8b9bb4'}">${r.lat != null ? Number(r.lat).toLocaleString('en-US') + 'ms' : '—'}</td>` +
-    `<td><span style="display:flex;align-items:center;gap:8px;white-space:nowrap">${pillHtml(r.st, stLabel(r))}${r.note && !stLabel(r) ? `<span style="font-size:10px;color:#54637e">${esc(r.note)}</span>` : ''}</span></td>` +
+    `<td><span style="display:flex;align-items:center;gap:8px;white-space:nowrap">${pillHtml(r.st, stLabel(r))}${r.note && !stLabel(r) ? ellipHtml(plainError(r.note, 20), '#54637e', '10px', r.note) : ''}</span></td>` +
     `</tr>`).join('') || `<tr><td colspan="9" class="empty-hint">无匹配记录 · 调整筛选条件</td></tr>`;
   if (bodyN) bodyN.innerHTML = rows.map((r) => `<tr class="${cls(r.st)}">` +
     `<td style="color:#e8f1ff;white-space:nowrap">${esc(r.page)}<span class="sub" style="color:#8b9bb4">${esc(r.item)}</span></td>` +
@@ -846,7 +870,7 @@ function renderLinRows(vm) {
     `<td style="color:#8b9bb4;font-size:11px;word-break:break-all">${esc(r.via)}</td>` +
     `<td class="num" style="color:#8b9bb4;font-size:11px;white-space:nowrap">${esc(r.freq)}<span class="sub">${esc(r.role)}</span></td>` +
     `<td class="num" style="font-size:11px;white-space:nowrap;color:#8b9bb4">${r.ok ? esc(hm(r.ok)) : '—'}<span class="sub">${r.lat != null ? r.lat + 'ms' : '—'}</span></td>` +
-    `<td>${pillHtml(r.st, stLabel(r))}${r.note && !stLabel(r) ? `<span class="sub">${esc(r.note)}</span>` : ''}</td>` +
+    `<td>${pillHtml(r.st, stLabel(r))}${r.note && !stLabel(r) ? `<span class="sub ellip" title="${esc(r.note)}">${esc(plainError(r.note, 20))}</span>` : ''}</td>` +
     `</tr>`).join('') || `<tr><td colspan="6" class="empty-hint">无匹配记录 · 调整筛选条件</td></tr>`;
 }
 function lineageHtml() {
@@ -1212,6 +1236,16 @@ function paintSparks() {
 }
 const prevLat = new Map();
 function flashChangedLatencies() {
+  document.querySelectorAll('[data-rate]').forEach((el) => {
+    const v = el.textContent;
+    const prev = prevLat.get(el);
+    if (prev !== undefined && prev !== v) {
+      el.classList.remove('flash');
+      void el.offsetWidth;
+      el.classList.add('flash');
+    }
+    prevLat.set(el, v);
+  });
   document.querySelectorAll('[data-lat]').forEach((el) => {
     const jid = el.closest('[data-jid]') ? el.closest('[data-jid]').dataset.jid : (el.closest('tr') ? 'row:' + [...el.closest('tr').children].indexOf(el) : '');
     const v = el.dataset.lat;
@@ -1369,7 +1403,7 @@ async function startApp() {
   renderPage(pageFromHash());
   updateTape(); // 预装最近真实事件并开始滚动
   updateClock();
-  setInterval(() => { updateClock(); updateDynamics(); }, 200); // 打样 tick 节奏
+  setInterval(() => { if (document.hidden) return; updateClock(); updateDynamics(); }, 200); // 打样 tick 节奏；页面隐藏时不做任何工作
   await Poller.start();
 }
 

xiaobai-datahub/admin/styles.css

@@ -375,6 +375,18 @@ body {
   0%   { transform: scale(.4); opacity: .8; }
   80%, 100% { transform: scale(3.2); opacity: 0; }
 }
+
+/* ---------- HEL-529 第二轮返工生产补充（布局鲁棒 + 错峰编排） ----------
+   1) .ellip：表格/卡片里的长错误与备注统一单行截断，完整原文走 title；
+   2) spark-end / LED 相位错峰：同一动画不同 delay，避免所有元素同频同步。
+   均复用既有 token 与动效曲线，不新增缓动、不改打样规则。 ---------- */
+.ellip { white-space: nowrap; overflow: hidden; text-overflow: ellipsis; min-width: 0; max-width: 100%; display: inline-block; vertical-align: bottom; }
+.ellip:empty { display: none; }
+table.dtable tbody tr:nth-child(2n) .spark-end { animation-delay: -.7s; }
+table.dtable tbody tr:nth-child(3n) .spark-end { animation-delay: -.35s; }
+.ds-grid .dsc:nth-child(2n) .spark-end { animation-delay: -.5s; }
+.ds-grid .dsc:nth-child(3n) .spark-end { animation-delay: -.9s; }
+.ds-grid .dsc:nth-child(5n) .spark-end { animation-delay: -.2s; }
 .lin-filter { display: flex; flex-wrap: wrap; align-items: center; gap: 8px; padding: 10px 12px; border-bottom: 1px solid #1a2540; }
 .lin-sel, .lin-q { background: #0c1220; border: 1px solid #243152; border-radius: 4px; font-size: 11px; color: #8b9bb4; padding: 6px 8px; outline: none; font-family: inherit; }
 .lin-q { color: #d7e1f0; padding: 6px 10px; width: 220px; }

xiaobai-datahub/datahub/lineage.py

@@ -184,9 +184,31 @@ DATASETS: list[dict[str, Any]] = [
         "tier": "licensed",
         "update_freq": "按需调用",
         "v1_endpoint": "/v1/query (api_name=ifind_wencai)",
-        "primary_source": "ifind:smart_stock_picking",
+        "primary_source": "ifind:wencai",
         "backup_source": None,
-        "known_consumers": ["问师(自然语言选股，需 iFinD 凭证)"],
+        # Real call site: backend/features/pools/service.py:82 (wencai(query,"stock"))
+        # feeds 股票池's ifind_event_enrichment_v1 (涨停/炸板/跌停原因、首末
+        # 涨停时间、开板次数). 问师 does NOT call wencai anywhere — its main
+        # body is LLM + 主站市场快照 (mentor/service.py builds context from
+        # dashboard/popularity snapshots only).
+        "known_consumers": ["股票池(涨停/炸板/跌停事件补充 enrichment，需 iFinD 凭证)"],
+    },
+    {
+        "dataset": "ifind_history",
+        "tier": "licensed",
+        "update_freq": "问师问询时按需 · 45 日回看",
+        "v1_endpoint": "/v1/query (api_name=ifind_history)",
+        "primary_source": "ifind:history",
+        "backup_source": None,
+        # Real call site: backend/features/mentor/service.py:433-457
+        # (_mentor_market_matrix → ifind.history(close/volume/amount, 45 日回看)).
+        # Only the trend/macro thinking-model profiles use it, and only when
+        # iFinD is configured — it fails open to [] otherwise. 问师其余子能力
+        # (本体问答/低吸/人气上下文等) 不依赖 iFinD.
+        "known_consumers": [
+            "问师·趋势思维模型(指数动量矩阵，可选)",
+            "问师·宏观思维模型(宽基指数与核心ETF矩阵，可选)",
+        ],
     },
 ]
 

xiaobai-datahub/datahub/pipeline.py

@@ -1491,8 +1491,20 @@ class Pipeline:
         else:
             keys = [(row.get("ts_code"), row.get("trade_date")) for row in rows]
         dup = row_n - len(set(keys))
+        # Extended soft datasets (popularity/dragon_tiger/…) already collapse
+        # within-batch dups in _dedupe_staging_rows before INSERT. Upstream
+        # ths/dc (and similar) routinely emit duplicate business keys; those
+        # collapsed dups must not hard-fail publish (HEL-562). Datasets without
+        # a staging business key — and all hard/core soft gates — still treat
+        # raw duplicate keys as errors.
+        staging_collapses_dups = (
+            dataset in EXTENDED_SOFT_DATASETS and dataset in STAGING_KEY_FIELDS
+        )
         if dup:
-            errors.append(f"duplicate keys: {dup}")
+            if staging_collapses_dups:
+                warnings.append(f"duplicate keys: {dup}")
+            else:
+                errors.append(f"duplicate keys: {dup}")
         bad_date = sum(1 for row in rows if str(row.get("trade_date")) != trade_date)
         if bad_date:
             errors.append(f"date mismatch rows: {bad_date}")
@@ -1511,7 +1523,8 @@ class Pipeline:
         field_report = self._field_gate(dataset, trade_date, rows, errors)
         if dataset in SOFT_DATASETS:
             allow_empty = dataset in {"popularity", "dragon_tiger", "moneyflow", "auction"}
-            hard_fail = bool(dup or bad_date or (empty and not allow_empty))
+            hard_dup = 0 if staging_collapses_dups else dup
+            hard_fail = bool(hard_dup or bad_date or (empty and not allow_empty))
         else:
             hard_fail = bool(errors) and (dataset in HARD_DATASETS or dataset == STOCKS_DATASET)
         report = {

xiaobai-datahub/tests/test_hel529_rework.py

@@ -108,6 +108,91 @@ class StagingDedupePublishTests(_Base):
         kept = dt[0]
         self.assertEqual(kept["buy_amount"], 300)
 
+    def test_hel562_popularity_collapsed_dups_publish_not_hard_fail(self) -> None:
+        """HEL-562: staging dedupe alone is not enough — quality gate used to
+        hard-fail on the same within-batch dups after they were already
+        collapsed (live 20260915: duplicate keys: 3 → integrity_gate)."""
+        trade_date = "20240902"
+        # 3 within-batch dups on (ts_code, trade_date, source=dc) — mirrors
+        # ths+dc merge where dc_hot repeats the same keys.
+        rows = [
+            {"ts_code": "600000.SH", "trade_date": trade_date, "source": "ths",
+             "ts_name": "浦发银行", "rank": 1, "pct_change": 1.2, "current_price": 10.2,
+             "hot": 90.0, "concept": "银行", "data_type": "热股"},
+            {"ts_code": "000001.SZ", "trade_date": trade_date, "source": "dc",
+             "ts_name": "平安银行", "rank": 1, "pct_change": 2.0, "current_price": 11.0,
+             "hot": 88.0, "concept": "银行", "data_type": "A股市场"},
+            {"ts_code": "600000.SH", "trade_date": trade_date, "source": "dc",
+             "ts_name": "浦发银行", "rank": 2, "pct_change": 1.2, "current_price": 10.2,
+             "hot": 80.0, "concept": "银行", "data_type": "A股市场"},
+            {"ts_code": "600000.SH", "trade_date": trade_date, "source": "dc",
+             "ts_name": "浦发银行", "rank": 3, "pct_change": 1.3, "current_price": 10.3,
+             "hot": 81.0, "concept": "银行", "data_type": "A股市场"},
+            {"ts_code": "600000.SH", "trade_date": trade_date, "source": "dc",
+             "ts_name": "浦发银行", "rank": 4, "pct_change": 1.4, "current_price": 10.4,
+             "hot": 82.0, "concept": "银行", "data_type": "A股市场"},
+        ]
+        report = self.hub.pipeline.validate("popularity", "b-gate", trade_date, rows)
+        self.assertFalse(report["hard_fail"])
+        self.assertEqual(report["errors"], [])
+        self.assertEqual(report["warnings"], ["duplicate keys: 2"])
+
+        result = self.hub.pipeline.run_dataset("popularity", trade_date, prepared_rows=rows)
+        # warnings present → soft_fail → publication state is degraded (still served)
+        self.assertEqual(result["state"], "degraded")
+        self.assertFalse(result["quality"]["hard_fail"])
+        self.assertTrue(result["quality"]["soft_fail"])
+        self.assertIn("duplicate keys: 2", result["quality"]["warnings"])
+        eod = self.hub.db.fetchall(
+            "SELECT * FROM eod_popularity WHERE trade_date = ? AND batch_id = ?",
+            (trade_date, result["batch_id"]),
+        )
+        # 5 raw → 3 unique keys after staging collapse (ths + two dc codes)
+        self.assertEqual(len(eod), 3)
+        pub = self.hub.db.fetchone(
+            "SELECT * FROM publications WHERE dataset='popularity' AND trade_date=?",
+            (trade_date,),
+        )
+        self.assertEqual(pub["active_batch"], result["batch_id"])
+        self.assertEqual(pub["state"], "degraded")
+        # Serving path accepts degraded the same as published (no DATASET_NOT_PUBLISHED)
+        from datahub.serving import V1API
+
+        api = V1API(self.hub.db, self.hub.pipeline, self.hub.settings)
+        payload = api.handle("/v1/popularity", {"date": [trade_date]})
+        self.assertEqual(len(payload["data"]), 3)
+        self.assertEqual(payload["meta"]["state"], "degraded")
+        self.assertEqual(payload["meta"]["batch_id"], result["batch_id"])
+
+    def test_hel562_core_soft_still_hard_fails_on_duplicate_keys(self) -> None:
+        """moneyflow/auction stay on the old soft gate: raw dups → hard_fail."""
+        trade_date = "20240902"
+        rows = [
+            {"ts_code": "600000.SH", "trade_date": trade_date,
+             "buy_sm_amount": 1, "sell_sm_amount": 1, "buy_md_amount": 1, "sell_md_amount": 1,
+             "buy_lg_amount": 1, "sell_lg_amount": 1, "buy_elg_amount": 1, "sell_elg_amount": 1,
+             "net_mf_amount": 0},
+            {"ts_code": "600000.SH", "trade_date": trade_date,
+             "buy_sm_amount": 2, "sell_sm_amount": 2, "buy_md_amount": 2, "sell_md_amount": 2,
+             "buy_lg_amount": 2, "sell_lg_amount": 2, "buy_elg_amount": 2, "sell_elg_amount": 2,
+             "net_mf_amount": 0},
+        ]
+        report = self.hub.pipeline.validate("moneyflow", "b-mf", trade_date, rows)
+        self.assertTrue(report["hard_fail"])
+        self.assertIn("duplicate keys: 1", report["errors"])
+        self.assertEqual(report["warnings"], [])
+
+    def test_hel562_popularity_date_mismatch_still_hard_fails(self) -> None:
+        """Collapsed-dup carve-out must not weaken other soft integrity checks."""
+        rows = [
+            {"ts_code": "600000.SH", "trade_date": "20240901", "source": "ths",
+             "ts_name": "浦发银行", "rank": 1, "pct_change": 1.2, "current_price": 10.2,
+             "hot": 90.0, "concept": "银行", "data_type": "热股"},
+        ]
+        report = self.hub.pipeline.validate("popularity", "b-bad-date", "20240902", rows)
+        self.assertTrue(report["hard_fail"])
+        self.assertIn("date mismatch rows: 1", report["errors"])
+
 
 class OverviewAnomalyConvergenceTests(_Base):
     def _seed_batches(self, today: str) -> None:
@@ -209,5 +294,52 @@ class SourceCatalogJoinTests(_Base):
             self.assertTrue(item.get("update_freq"), f"missing update_freq for {item['dataset']}")
 
 
+class LineageMentorIfindTests(_Base):
+    """问师 → iFinD 血缘修正（第二轮返工）：不按名称猜关系，按真实调用代码。"""
+
+    REPO = Path(__file__).resolve().parents[2]
+
+    def test_no_mentor_dependency_on_ifind_wencai(self) -> None:
+        items = self.hub.admin.lineage("20240902")["items"]
+        wencai = [i for i in items if i["dataset"] == "ifind_wencai"]
+        self.assertEqual(len(wencai), 1)
+        consumers = wencai[0]["known_consumers"]
+        for consumer in consumers:
+            self.assertNotIn("问师", consumer, f"wencai consumer must not be 问师: {consumer}")
+        all_consumers = " ".join(c for i in items for c in i["known_consumers"])
+        self.assertNotIn("问师(自然语言选股", all_consumers)
+
+    def test_wencai_real_consumer_is_pools_enrichment_with_code_evidence(self) -> None:
+        items = self.hub.admin.lineage("20240902")["items"]
+        wencai = [i for i in items if i["dataset"] == "ifind_wencai"][0]
+        self.assertTrue(any("股票池" in c for c in wencai["known_consumers"]))
+        # Real call evidence in the main-site source tree:
+        pools_src = (self.REPO / "backend" / "features" / "pools" / "service.py").read_text(encoding="utf-8")
+        self.assertIn("ifind.wencai(", pools_src)
+        self.assertIn("ifind_event_enrichment_v1", pools_src)
+        # And 问师 itself never calls wencai:
+        mentor_src = (self.REPO / "backend" / "features" / "mentor" / "service.py").read_text(encoding="utf-8")
+        self.assertNotIn(".wencai(", mentor_src)
+
+    def test_mentor_optional_ifind_history_subcapabilities(self) -> None:
+        items = self.hub.admin.lineage("20240902")["items"]
+        history = [i for i in items if i["dataset"] == "ifind_history"]
+        self.assertEqual(len(history), 1)
+        consumers = history[0]["known_consumers"]
+        self.assertTrue(any("趋势思维模型" in c for c in consumers))
+        self.assertTrue(any("宏观思维模型" in c for c in consumers))
+        # every consumer must be a 问师 sub-capability, not the whole board
+        for consumer in consumers:
+            self.assertIn("·", consumer, f"not a sub-capability mapping: {consumer}")
+        # Real call evidence: mentor builds market matrices via ifind.history
+        mentor_src = (self.REPO / "backend" / "features" / "mentor" / "service.py").read_text(encoding="utf-8")
+        self.assertIn("ifind.history(", mentor_src)
+        self.assertIn("_mentor_market_matrix", mentor_src)
+        self.assertIn("MENTOR_INDEX_UNIVERSE", mentor_src)
+        self.assertIn("MENTOR_ETF_UNIVERSE", mentor_src)
+        # Optional dependency: fails open when ifind is not configured
+        self.assertIn("if not ifind or not ifind.configured", mentor_src)
+
+
 if __name__ == "__main__":
     unittest.main()