HEL-543 返工: 补充 observability 运行时可关闭开关 (DATAHUB_OBSERVABILITY)

总工复核 🔴：安全边界要求新增观测功能必须可关闭、关闭后现有功能完全照旧，但此前实现没有任何运行时开关。修复： - settings.py: 新增 Settings.observability_enabled 字段，沿用既有 DATAHUB_SCHEDULER 的环境变量模式，读取 DATAHUB_OBSERVABILITY (0/false/off 关闭，默认开启)。 - hub.py: Hub.__init__ 把 settings.observability_enabled 挂到 self.db 上，让 pipeline/realtime_serve/steward/admin_api 已经在传的 db 参数直接带上开关，零额外改造。 - observability.py: 新增 is_enabled(db)，缺失该属性时默认按启用处理 (向后兼容裸 HubDB 用例/测试)。关闭时 observe() 变成纯透传（不计时、不分类、不碰数据库)，record_call() 直接 no-op。 - admin_api.py: 4 个新只读端点关闭时返回明确的 {"enabled": false, ...空结构} 而不是静默返回旧数据。 - 新增 10 个测试：开关默认值/环境变量解析、关闭后 observe() 的透传语义 (含异常原样重新抛出)、关闭后 record_call() 零写入、关闭后重新开启恢复记录、4 个 admin 端点在关闭态的响应结构。全量测试 183/183 通过（新增 10 个，含此前 173 个零回归）。 Co-authored-by: Cursor <cursoragent@cursor.com> Co-authored-by: multica-agent <github@multica.ai>
HEL-543: add data hub observability side-channel (provider status, source catalog, lineage)
2026-09-14 01:10:58 +08:00 · 2026-09-14 00:59:03 +08:00
19 changed files with 1988 additions and 1653 deletions
@@ -3,15 +3,14 @@
 <head>
  <meta charset="UTF-8" />
  <meta name="viewport" content="width=device-width, initial-scale=1" />
-  <title>xiaobai-datahub 管理后台 · 数据中枢</title>
+  <title>xiaobai-datahub 管理后台</title>
  <link rel="stylesheet" href="/admin/styles.css" />
 </head>
 <body>
  <div id="app">
    <section id="login-view" class="panel auth-panel">
      <span class="badge-sim">内网 · 8766</span>
      <h1>数据中枢</h1>
-      <p class="muted">四路来源持续汇流、调度、发布与审计的运转空间。</p>
+      <p class="muted">内网管理后台，用于查看源状态、调度和盘后发布批次。</p>
      <form id="login-form">
        <label>账号 <input name="username" value="hub_admin" autocomplete="username" /></label>
        <label>密码 <input name="password" type="password" autocomplete="current-password" /></label>
@@ -31,46 +30,22 @@
    </section>
    <section id="shell" hidden>
-      <header id="topbar">
+      <header class="top">
-        <div class="logo">小白复盘 <em>·</em> 数据中枢</div>
+        <strong>xiaobai-datahub</strong>
        <span class="crumb" id="crumb">8766 · 四源汇流 · 持续运转</span>
        <span id="phase" class="pill"></span>
-        <span class="spacer"></span>
+        <span id="who" class="muted"></span>
        <span id="who" class="muted who"></span>
        <button type="button" id="theme-btn" class="ghost">夜间</button>
        <button type="button" id="logout-btn" class="ghost">退出</button>
      </header>
-
+      <nav>
-      <!-- 动效版：一台连续空间舞台，六幕滚动切换，同一核心与流道贯穿始终 -->
+        <button data-page="overview" class="active">总览</button>
-      <div id="stageWrap">
+        <button data-page="sources">数据源</button>
-        <canvas id="scene"></canvas>
+        <button data-page="jobs">调度任务</button>
-        <div id="cabin"></div>
+        <button data-page="release">盘后发布</button>
-        <div id="legend">
+        <button data-page="datasets">数据集</button>
-          <span class="lampico li-link"></span>LINK 链路常亮（低亮）<br>
+        <button data-page="audit">审计</button>
-          <span class="lampico li-act"></span>ACT 活动灯 · 事件成簇短闪
+      </nav>
-        </div>
+      <main id="page"></main>
        <div id="rail"></div>
        <div id="hint">滚 动 推 进 镜 头</div>
        <div id="detail" class="closed">
          <div class="dtag"></div>
          <button id="detailClose" type="button" aria-label="关闭详情">×</button>
          <div id="detailBody"></div>
        </div>
      </div>
      <div id="track" aria-hidden="true">
        <section data-scene="0"></section>
        <section data-scene="1"></section>
        <section data-scene="2"></section>
        <section data-scene="3"></section>
        <section data-scene="4"></section>
        <section data-scene="5"></section>
      </div>
      <!-- 减少动态效果版：六幕静态空间图 + 固定文字状态，信息与功能完全等价 -->
      <main id="staticShell" hidden>
        <nav id="staticNav"></nav>
        <div id="staticScenes"></div>
      </main>
    </section>
  </div>
  <script src="/admin/app.js"></script>
@@ -1,278 +1,51 @@
 /* xiaobai-datahub 数据中枢后台 —— 第八版「星港中枢」
   整页星空 + 中央星体 + 四颗数据源卫星的连续舞台，六幕滚动换场。
   统一动效 Token：进入/退出用强 ease-out，屏内移动用强 ease-in-out，持续流动用 linear。
   禁止 ease-in、禁止 transition:all —— 全文件遵守。 */
 :root {
  color-scheme: light;
-  --bar-h: 52px;
+  --bg: #f4f5f7;
  --radius: 14px;
  --pad: 16px;
  --font-cn: "Segoe UI", "PingFang SC", "Noto Sans SC", "Microsoft YaHei", sans-serif;
  --font-mono: "JetBrains Mono", "DejaVu Sans Mono", Consolas, monospace;
  --ease-out: cubic-bezier(.22, 1, .36, 1);
  --ease-in-out: cubic-bezier(.65, 0, .35, 1);
  --ease-linear: linear;
  /* day (default) */
  --bg0: #EDF0F7;
  --bg1: #F8FAFF;
  --surface: #ffffff;
-  --text: #1B2547;
+  --text: #1f2329;
-  --muted: #5B6785;
+  --muted: #646a73;
-  --faint: #93A0BD;
+  --line: #dee0e3;
-  --line: rgba(47, 99, 214, .18);
+  --action: #3370ff;
-  --action: #2F63D6;
+  --danger: #e04536;
-  --cyan: #0A9C93;
+  --ok: #16a34a;
-  --amber: #B97A0E;
+  --warn: #b45309;
-  --danger: #D64F4F;
+  --radius: 8px;
-  --ok: #1E9E6A;
+  --pad: 16px;
-  --warn: #B97A0E;
+  font-family: "Segoe UI", "PingFang SC", "Noto Sans SC", sans-serif;
  --glass: rgba(255, 255, 255, .82);
  --chip: rgba(47, 99, 214, .08);
  /* 星港中枢 · 画布调色（与 app.js PAL.day 保持一致，仅作文档留存） */
  --star-hi: #FFFFFF; --star-a: #EAF6FF; --star-b: #A9CCF0; --star-c: #5488C4; --star-d: #2E5A94;
  --orbit-line: rgba(90, 120, 180, .40);
  --sat1: #3997B4; --sat2: #BF873F; --sat3: #537EBF; --sat4: #BF525F;
 }
 :root[data-theme="night"] {
  color-scheme: dark;
-  --bg0: #05080F;
+  --bg: #111318;
-  --bg1: #0A1020;
+  --surface: #1b1e24;
-  --surface: #12172A;
+  --text: #e8eaed;
-  --text: #E8EEFC;
+  --muted: #9aa0a6;
-  --muted: #8B97B8;
+  --line: #2a2f38;
-  --faint: #59637F;
+  --action: #5b8cff;
  --line: rgba(120, 160, 255, .16);
  --action: #4A86FF;
  --cyan: #2FD8CE;
  --amber: #FFB84D;
  --danger: #FF6B6B;
  --ok: #3ED598;
  --warn: #FFB84D;
  --glass: rgba(13, 20, 38, .72);
  --chip: rgba(74, 134, 255, .10);
  /* 星港中枢 · 画布调色（与 app.js PAL.night 保持一致，仅作文档留存） */
  --star-hi: #F2FAFF; --star-a: #C9E7FF; --star-b: #7FB2EA; --star-c: #3E6FAE; --star-d: #16305E;
  --orbit-line: rgba(140, 180, 255, .22);
  --sat1: #4CC9F0; --sat2: #FFB454; --sat3: #6FA8FF; --sat4: #FF6E7F;
 }
 * { box-sizing: border-box; }
-html, body { margin: 0; background: var(--bg0); color: var(--text); font-family: var(--font-cn); }
+body { margin: 0; background: var(--bg); color: var(--text); }
-body { overflow-x: hidden; }
+.panel, header.top, nav, main { background: var(--surface); }
-.mono { font-family: var(--font-mono); font-variant-numeric: tabular-nums; }
+.auth-panel { max-width: 420px; margin: 12vh auto; padding: 28px; border-radius: var(--radius); border: 1px solid var(--line); }
 label { display: block; margin: 12px 0; }
 input, select { width: 100%; padding: 8px 10px; border: 1px solid var(--line); border-radius: 4px; background: var(--bg); color: var(--text); }
 button { background: var(--action); color: #fff; border: 0; border-radius: 4px; padding: 8px 14px; cursor: pointer; }
 button.ghost { background: transparent; color: var(--text); border: 1px solid var(--line); }
 button.danger { background: var(--danger); }
 .muted { color: var(--muted); }
 .error { color: var(--danger); }
 .top { display: flex; gap: 12px; align-items: center; padding: 10px var(--pad); border-bottom: 1px solid var(--line); }
 nav { display: flex; gap: 4px; padding: 8px var(--pad); border-bottom: 1px solid var(--line); }
 nav button { background: transparent; color: var(--muted); }
 nav button.active { color: var(--action); background: transparent; font-weight: 600; }
 main { padding: var(--pad); min-height: calc(100vh - 96px); }
 .cards { display: grid; grid-template-columns: repeat(auto-fit, minmax(180px, 1fr)); gap: 12px; margin-bottom: 16px; }
 .card { border: 1px solid var(--line); border-radius: var(--radius); padding: 12px; }
 table { width: 100%; border-collapse: collapse; font-size: 13px; }
 th, td { text-align: left; padding: 8px; border-bottom: 1px solid var(--line); vertical-align: top; }
 .pill { font-size: 12px; padding: 2px 8px; border-radius: 999px; border: 1px solid var(--line); }
 .ok { color: var(--ok); }
 .warn { color: var(--warn); }
 .fail { color: var(--danger); }
-
+.toolbar { display: flex; gap: 8px; flex-wrap: wrap; margin: 12px 0; align-items: end; }
-/* ---------- 登录 / 改密（未进入舞台的前置流程） ---------- */
+.toolbar label { margin: 0; }
 .auth-panel {
  max-width: 420px; margin: 12vh auto; padding: 30px 32px; border-radius: var(--radius);
  background: var(--glass); border: 1px solid var(--line); backdrop-filter: blur(14px);
  box-shadow: 0 30px 70px -32px rgba(20, 30, 60, .35);
 }
 .auth-panel h1 { margin: 0 0 6px; font-size: 21px; }
 .auth-panel label { display: block; margin: 12px 0; font-size: 13px; color: var(--muted); }
 .auth-panel input {
  width: 100%; margin-top: 6px; padding: 9px 11px; border: 1px solid var(--line); border-radius: 8px;
  background: var(--chip); color: var(--text); font-family: var(--font-cn);
 }
 .auth-panel input:focus { outline: 2px solid var(--action); outline-offset: 1px; }
 .auth-panel .badge-sim { display: inline-block; margin-bottom: 10px; }
 button { font-family: var(--font-cn); cursor: pointer; }
 .btn, button.primary, .auth-panel button[type="submit"] {
  background: linear-gradient(135deg, var(--action), var(--cyan)); color: #fff; border: 0; border-radius: 8px;
  padding: 8px 16px; font-size: 13px; transition: transform .12s var(--ease-out), opacity .12s var(--ease-out);
 }
 .btn:active, button:active { transform: translateY(1px); }
 .btn.ghost, button.ghost {
  background: var(--chip); color: var(--text); border: 1px solid var(--line);
 }
 .btn.warn { background: transparent; color: var(--warn); border: 1px solid var(--warn); }
 .btn.danger, button.danger { background: var(--danger); color: #fff; border: 0; }
 .btn.pri { background: linear-gradient(135deg, var(--action), var(--cyan)); color: #fff; border: 0; }
 .btn { padding: 6px 13px; border-radius: 8px; font-size: 12px; }
 .badge-sim {
  font-size: 11px; padding: 3px 10px; border-radius: 99px; letter-spacing: .1em;
  color: var(--amber); border: 1px solid var(--amber); opacity: .85;
 }
 /* ---------- 通用表格 / 卡片（cabin、detail、reduced 三处共用同一来源） ---------- */
 table { width: 100%; border-collapse: collapse; font-size: 12.5px; }
 th, td { text-align: left; padding: 7px 6px; border-bottom: 1px dashed var(--line); vertical-align: top; }
 th { color: var(--faint); font-weight: 500; font-size: 11px; letter-spacing: .04em; }
 .pill { font-size: 12px; padding: 2px 10px; border-radius: 999px; border: 1px solid var(--line); color: var(--muted); }
 .cards { display: grid; grid-template-columns: repeat(auto-fit, minmax(170px, 1fr)); gap: 10px; margin-bottom: 14px; }
 .card { border: 1px solid var(--line); border-radius: 10px; padding: 10px 12px; background: var(--chip); }
 .card strong { font-size: 18px; }
 .toolbar { display: flex; gap: 8px; flex-wrap: wrap; margin: 10px 0; align-items: center; }
 .toolbar label { font-size: 12px; color: var(--muted); display: flex; align-items: center; gap: 6px; }
 .toolbar input { padding: 6px 9px; border: 1px solid var(--line); border-radius: 6px; background: var(--chip); color: var(--text); }
 h2 { font-size: 16px; margin: 4px 0 10px; }
 h3 { font-size: 13px; margin: 14px 0 6px; color: var(--muted); }
 /* ---------- 顶部总控条 ---------- */
 #topbar {
  height: var(--bar-h); display: flex; align-items: center; gap: 14px; padding: 0 22px;
  border-bottom: 1px solid var(--line); background: linear-gradient(180deg, var(--bg1), transparent);
  backdrop-filter: blur(10px);
  position: sticky; top: 0; z-index: 4;
 }
 #topbar .logo { font-weight: 700; font-size: 15px; letter-spacing: .04em; }
 #topbar .logo em { font-style: normal; color: var(--cyan); }
 #topbar .crumb { font-size: 12px; color: var(--faint); }
 #topbar .spacer { flex: 1; }
 #topbar .who { font-size: 12px; }
 /* ---------- 全幅星空舞台（100vw × (100vh-52px)，六幕连续场景） ----------
   第八版取消卡片式舞台：星空即整页背景，无圆角、无边框、无外边距。
   #track 的六段高幕（每段 130vh）在它身后正常参与文档流把页面撑高，
   #stageWrap 用 sticky 钉在视口里，滚动时只有 #track 的高度被“划过”，
   舞台本体在这段距离内始终可见、位置不变，直到 #track 撑出的空间耗尽才随之离场。
   星系几何（星体/轨道/卫星/去向端/双灯）的安全盒与缩放公式见 app.js Stage.resize()。 */
 #stageWrap {
  flex: none; width: 100vw; height: calc(100vh - var(--bar-h)); min-height: 480px;
  margin: 0; border-radius: 0; overflow: hidden; border: 0;
  box-shadow: inset 0 0 160px rgba(0, 0, 0, .10);
  position: sticky; top: var(--bar-h); z-index: 2;
 }
 #scene { position: absolute; inset: 0; width: 100%; height: 100%; display: block; cursor: crosshair; }
 /* ---------- 控制舱：当前场景的关键数据与操作 ---------- */
 #cabin {
  position: absolute; left: 24px; top: 24px; width: 300px; z-index: 5; max-height: calc(100% - 48px);
  overflow: auto; background: var(--glass); border: 1px solid var(--line); border-radius: 14px;
  backdrop-filter: blur(14px); padding: 16px 18px;
  transition: opacity .45s var(--ease-out), transform .45s var(--ease-out);
 }
 #cabin.hide { opacity: 0; transform: translateX(-14px); pointer-events: none; }
 #cabin .eyebrow { font-size: 10px; letter-spacing: .22em; color: var(--cyan); margin-bottom: 6px; }
 #cabin h2 { font-size: 18px; margin: 0 0 4px; }
 #cabin .sub { font-size: 11.5px; color: var(--muted); line-height: 1.6; margin-bottom: 10px; }
 #cabin .rows { display: flex; flex-direction: column; gap: 7px; margin-bottom: 4px; }
 #cabin .row { display: flex; align-items: center; gap: 8px; font-size: 12px; color: var(--muted); }
 #cabin .row b { color: var(--text); font-weight: 600; }
 #cabin .row .fill { flex: 1; }
 #cabin .dotlamp { width: 7px; height: 7px; border-radius: 99px; background: var(--ok); box-shadow: 0 0 8px var(--ok); flex: none; }
 #cabin .dotlamp.warn { background: var(--warn); box-shadow: 0 0 8px var(--warn); }
 #cabin .dotlamp.fail { background: var(--danger); box-shadow: 0 0 8px var(--danger); }
 #cabin .actions { display: flex; gap: 8px; margin-top: 12px; flex-wrap: wrap; }
 /* ---------- 底部章节轨（导航等同滚动跳转） ---------- */
 #rail {
  position: absolute; left: 50%; bottom: 16px; transform: translateX(-50%); z-index: 5;
  display: flex; gap: 6px; align-items: center; padding: 8px 12px; border-radius: 99px;
  background: var(--glass); border: 1px solid var(--line); backdrop-filter: blur(14px);
 }
 #rail .stop {
  display: flex; align-items: center; gap: 7px; padding: 5px 11px; border-radius: 99px; cursor: pointer;
  font-size: 11.5px; color: var(--faint); background: transparent; border: 0;
  transition: color .25s var(--ease-in-out), background .25s var(--ease-in-out);
  font-family: var(--font-cn); white-space: nowrap;
 }
 #rail .stop i { width: 6px; height: 6px; border-radius: 99px; background: var(--faint); transition: background .25s var(--ease-in-out), box-shadow .25s var(--ease-in-out); display: inline-block; }
 #rail .stop.on { color: var(--text); background: var(--chip); }
 #rail .stop.on i { background: var(--cyan); box-shadow: 0 0 10px var(--cyan); }
 #rail .sep { width: 12px; height: 1px; background: var(--line); }
 /* ---------- 图例 ---------- */
 #legend {
  position: absolute; right: 22px; top: 22px; z-index: 5; font-size: 10.5px; color: var(--muted);
  background: var(--glass); border: 1px solid var(--line); border-radius: 12px; padding: 10px 13px;
  backdrop-filter: blur(14px); line-height: 1.9;
 }
 #legend .lampico { display: inline-block; width: 14px; height: 5px; border-radius: 99px; vertical-align: middle; margin-right: 6px; }
 #legend .li-link { background: rgba(47, 216, 206, .5); box-shadow: 0 0 6px rgba(47, 216, 206, .6); }
 #legend .li-act { background: var(--amber); box-shadow: 0 0 8px var(--amber); }
 /* ---------- 详情面板：从端口/来源空间位置展开的功能抽屉 ---------- */
 #detail {
  position: absolute; z-index: 8; width: 360px; max-width: calc(100% - 40px); max-height: calc(100% - 40px);
  overflow: auto; pointer-events: auto; background: var(--glass); border: 1px solid var(--line);
  border-radius: 14px; backdrop-filter: blur(16px); padding: 16px 18px;
  transition: transform .4s var(--ease-out), opacity .3s var(--ease-out);
  box-shadow: 0 24px 60px -20px rgba(0, 0, 0, .45);
 }
 #detail.closed { transform: scale(.14); opacity: 0; pointer-events: none; }
 #detail h3 { font-size: 14px; margin: 0 0 2px; color: var(--text); }
 #detail .dsub { font-size: 10.5px; color: var(--faint); margin-bottom: 10px; letter-spacing: .04em; }
 #detailClose {
  position: absolute; right: 10px; top: 10px; width: 24px; height: 24px; border-radius: 8px;
  border: 1px solid var(--line); background: transparent; color: var(--muted); cursor: pointer; font-size: 13px; line-height: 1;
 }
 #detail .dtag {
  position: absolute; left: -7px; top: 26px; width: 14px; height: 14px; transform: rotate(45deg);
  background: var(--glass); border-left: 1px solid var(--line); border-bottom: 1px solid var(--line);
 }
 /* ---------- 滚动提示 ---------- */
 #hint {
  position: absolute; left: 50%; bottom: 100px; transform: translateX(-50%); z-index: 5;
  font-size: 11px; letter-spacing: .3em; color: var(--faint); transition: opacity .5s var(--ease-out);
 }
 #hint.off { opacity: 0; }
 /* 滚动章节占位：仅提供滚动行程，无可见内容 */
 #track { position: relative; z-index: 1; pointer-events: none; }
 #track section { height: 130vh; }
 /* ---------- 危险操作 / 确认弹层，复用 dialog ---------- */
 dialog { border: 1px solid var(--line); border-radius: var(--radius); background: var(--surface); color: var(--text); padding: 20px; }
 /* ================================================================
   减少动态效果（prefers-reduced-motion）：舞台切换为六幕静态空间图 + 固定文字状态
   信息与功能与动效版完全一致，只是去掉滚动镜头、旋转与闪烁。
   ================================================================ */
 #staticShell { display: none; }
 html.reduced #stageWrap, html.reduced #track, html.reduced #hint { display: none !important; }
 html.reduced #staticShell { display: block; }
 #staticNav {
  display: flex; gap: 4px; padding: 8px var(--pad); border-bottom: 1px solid var(--line);
  background: var(--surface); flex-wrap: wrap; position: sticky; top: 0; z-index: 3;
 }
 #staticNav button {
  background: transparent; color: var(--muted); border: 0; padding: 7px 12px; border-radius: 8px; font-size: 13px;
 }
 #staticNav button.active { color: var(--action); font-weight: 600; background: var(--chip); }
 .scene-static { padding: var(--pad); border-bottom: 1px solid var(--line); }
 .scene-static .scene-head { display: flex; align-items: baseline; gap: 10px; margin-bottom: 10px; }
 .scene-static .scene-head .eyebrow { font-size: 11px; letter-spacing: .18em; color: var(--cyan); }
 .scene-static .diagram {
  width: 100%; max-width: 620px; height: 190px; border-radius: 12px; border: 1px solid var(--line);
  background: var(--chip); display: block; margin-bottom: 12px;
 }
 .scene-static .lamprow { display: flex; flex-wrap: wrap; gap: 10px; margin-bottom: 10px; }
 .lampchip {
  display: inline-flex; align-items: center; gap: 6px; font-size: 11.5px; color: var(--muted);
  border: 1px solid var(--line); border-radius: 999px; padding: 4px 10px; background: var(--chip);
 }
 .lampchip .lk, .lampchip .ac { width: 7px; height: 7px; border-radius: 99px; background: var(--faint); display: inline-block; }
 .lampchip .lk.on { background: var(--cyan); box-shadow: 0 0 5px var(--cyan); }
 .lampchip .ac.on { background: var(--amber); box-shadow: 0 0 6px var(--amber); }
 .lampchip .ac.recent { outline: 1px solid var(--amber); transition: outline-color .8s var(--ease-out); }
 /* ---------- 主内容区（main/reduced 内容容器统一样式，供动效版 detail 与静态版共用） ---------- */
 main#page { padding: var(--pad); min-height: calc(100vh - 96px); background: var(--surface); }
 /* ---------- 响应式：1440 / 1280 / 1024 三档桌面宽（须与 app.js cabinW 三档同步） ---------- */
@media (max-width: 1280px) {
  #cabin { width: 260px; }
  #detail { width: 320px; }
 }
@media (max-width: 1024px) {
  #cabin { width: 230px; padding: 13px 14px; }
  #legend { display: none; }
 }
 /* ---------- 系统级兜底：即便 JS 未及时接管，也不留半成品动效 ---------- */
@media (prefers-reduced-motion: reduce) {
  #stageWrap, #track, #hint { display: none !important; }
  #staticShell { display: block !important; }
  * { animation-duration: .001ms !important; animation-iteration-count: 1 !important; transition-duration: .001ms !important; }
 }
@@ -6,9 +6,12 @@ from typing import Any
 from datahub.adapters import RESERVED
 from datahub.auth import AuthService
 from datahub.db import HubDB
 from datahub import lineage as lineage_module
 from datahub import observability
 from datahub.pipeline import OFFICIAL_DATASETS, STOCKS_DATASET, Pipeline
 from datahub.scheduler import Scheduler
 from datahub.serving import ApiError
 from datahub import source_catalog
 from datahub.timeutil import isoformat, now_shanghai, session_phase, yyyymmdd
@@ -105,6 +108,53 @@ class AdminAPI:
            raise ApiError("INVALID_ARGUMENT", f"unknown provider: {provider}")
        return adapter.probe()
    # ------------------------------------------------------------------
    # HEL-543: read-only side-channel status/catalog/lineage. These never
    # change routing, credentials, or adapters; they only read the
    # provider_call_log/provider_health tables (observability.py) plus the
    # static registries in source_catalog.py / lineage.py.
    # ------------------------------------------------------------------
    def providers_status(self, provider: str = "", limit: int = 50) -> dict[str, Any]:
        if not observability.is_enabled(self.db):
            return {"enabled": False, "health": [], "recent_calls": []}
        limit = max(1, min(int(limit or 50), 200))
        health_sql = "SELECT * FROM provider_health"
        params: tuple[Any, ...] = ()
        if provider:
            health_sql += " WHERE provider = ?"
            params = (provider,)
        health_sql += " ORDER BY provider, interface"
        health = self.db.fetchall(health_sql, params)
        calls_sql = "SELECT * FROM provider_call_log"
        if provider:
            calls_sql += " WHERE provider = ?"
        calls_sql += " ORDER BY id DESC LIMIT ?"
        recent = self.db.fetchall(calls_sql, (*params, limit))
        return {"enabled": True, "health": health, "recent_calls": recent}
    def source_catalog(self) -> dict[str, Any]:
        if not observability.is_enabled(self.db):
            return {"enabled": False, "items": []}
        return {"enabled": True, "items": source_catalog.snapshot(self.db, self.auth)}
    def lineage(self, trade_date: str = "") -> dict[str, Any]:
        day = yyyymmdd(trade_date) if trade_date else yyyymmdd(now_shanghai())
        if not observability.is_enabled(self.db):
            return {"enabled": False, "trade_date": day, "items": []}
        return {"enabled": True, "trade_date": day, "items": lineage_module.snapshot(self.db, day)}
    def lineage_affected(self, provider: str = "", interface: str = "") -> dict[str, Any]:
        if not provider:
            raise ApiError("INVALID_ARGUMENT", "provider is required")
        if not observability.is_enabled(self.db):
            return {"enabled": False, "provider": provider, "interface": interface, "items": []}
        return {
            "enabled": True,
            "provider": provider,
            "interface": interface,
            "items": lineage_module.affected(self.db, provider, interface),
        }
    def jobs(self) -> dict[str, Any]:
        runs = self.db.fetchall("SELECT * FROM job_runs ORDER BY id DESC LIMIT 100")
        stocks_times = "/".join(self.pipeline.settings.stocks_refresh_times) or "20:00"
@@ -8,6 +8,7 @@ from pathlib import Path
 from typing import Any
 from datahub.datasets_ext import EXTENDED_DATASET_TABLES, EXTENDED_SCHEMA
 from datahub.observability import OBS_SCHEMA
 from datahub.timeutil import isoformat
 _BASE_SCHEMA = """
@@ -296,7 +297,7 @@ CREATE INDEX IF NOT EXISTS idx_eod_bars_date ON eod_bars(trade_date, batch_id);
 CREATE INDEX IF NOT EXISTS idx_calendar_open ON trade_calendar(is_open, cal_date);
 """
-SCHEMA = _BASE_SCHEMA + EXTENDED_SCHEMA
+SCHEMA = _BASE_SCHEMA + EXTENDED_SCHEMA + OBS_SCHEMA
 DATASET_TABLES = {
    "daily": ("eod_bars", "staging_bars"),
@@ -122,6 +122,26 @@ class HubRequestHandler(BaseHTTPRequestHandler):
        if path == "/admin/api/sources" and method == "GET":
            self._json(self.hub.admin.sources(), HTTPStatus.OK)
            return
        if path == "/admin/api/providers/status" and method == "GET":
            query = parse_query(urlparse(self.path).query)
            provider = (query.get("provider") or [""])[0]
            limit = (query.get("limit") or ["50"])[0]
            self._json(self.hub.admin.providers_status(provider, int(limit or 50)), HTTPStatus.OK)
            return
        if path == "/admin/api/source-catalog" and method == "GET":
            self._json(self.hub.admin.source_catalog(), HTTPStatus.OK)
            return
        if path == "/admin/api/lineage" and method == "GET":
            query = parse_query(urlparse(self.path).query)
            date = (query.get("date") or [""])[0]
            self._json(self.hub.admin.lineage(date), HTTPStatus.OK)
            return
        if path == "/admin/api/lineage/affected" and method == "GET":
            query = parse_query(urlparse(self.path).query)
            provider = (query.get("provider") or [""])[0]
            interface = (query.get("interface") or [""])[0]
            self._json(self.hub.admin.lineage_affected(provider, interface), HTTPStatus.OK)
            return
        if path.startswith("/admin/api/sources/") and path.endswith("/probe") and method == "POST":
            provider = path.split("/")[4]
            self._json(self.hub.admin.probe(provider), HTTPStatus.OK)
@@ -24,6 +24,12 @@ class Hub:
            raise SystemExit("DATAHUB_ENCRYPTION_KEY 未配置")
        self.settings = settings
        self.db = HubDB(settings.db_path)
        # HEL-543: carry the observability kill switch on the db handle so
        # every call site that already threads `db` through (pipeline,
        # realtime_serve, steward, admin_api) picks it up for free with no
        # extra plumbing. Missing this attribute (e.g. a bare HubDB built
        # directly in tests) defaults to enabled — see observability.is_enabled.
        self.db.observability_enabled = settings.observability_enabled
        self.vault = SecretVault(settings.encryption_key)
        self.auth = AuthService(self.db, self.vault, settings.api_token, settings.admin_password)
        token = settings.tushare_token or self.auth.load_credential("tushare_token")
@@ -0,0 +1,258 @@
 """Read-only lineage/impact inventory (HEL-543).
 Answers, without changing any routing decision: for a given main-site data
 item, which datahub dataset backs it, which provider/interface currently
 serves it (primary and backup), and — when a provider/interface is
 unhealthy — which datasets and, best-effort, which main-site consumers are
 affected.
 Every row cites where it was verified so a reviewer does not have to trust
 a paraphrase:
 - ``v1_endpoint``/``primary_source``/``backup_source`` are taken verbatim
  from ``datahub/serving.py`` (the ``source=`` string literal passed to
  ``_published_rows``/``_official_meta``) or from the provider/interface
  pairs wired into ``datahub/realtime_serve.py`` for HEL-543.
 - ``known_consumers`` lists only call sites this round actually found via
  code search in the ``xiaobai-review`` website tree (cited as
  ``file:line`` in the comment above each dataset). Anything not backed by
  a citation is left out rather than guessed; a fuller page-by-page map is
  tracked separately (HEL-549) and can extend this table later without
  touching its shape.
 This module never talks to a provider and never mutates anything; it only
 reads ``provider_health``/``provider_call_log`` (HEL-543) and the existing
 ``publications``/``batches`` tables to attach live status to each row.
 """
 from __future__ import annotations
 from typing import Any
 # Verified against backend/features/screener/data_sync.py (calendar,
 # stock_basic, daily, daily_basic, index_daily-as-benchmark, stk_auction,
 # moneyflow, ths_hot, dc_hot all called via `self.client.query(...)`) and
 # backend/features/heaven/market_context.py (stock_basic, index_daily via
 # `self._tushare_client().query(...)` / `client.query(...)`).
 DATASETS: list[dict[str, Any]] = [
    {
        "dataset": "calendar",
        "tier": "official",
        "v1_endpoint": "/v1/calendar",
        "primary_source": "tushare:trade_cal",
        "backup_source": None,
        "known_consumers": ["智能选股(data_sync.py 交易日历解析)", "问天(交易日推算)"],
    },
    {
        "dataset": "stocks",
        "tier": "official",
        "v1_endpoint": "/v1/stocks",
        "primary_source": "tushare:stock_basic",
        "backup_source": None,
        "known_consumers": ["智能选股(股票主档)", "问天(market_context.py 股票代码/名称解析)"],
    },
    {
        "dataset": "daily",
        "tier": "official",
        "v1_endpoint": "/v1/bars/daily",
        "primary_source": "tushare:daily",
        "backup_source": None,
        "known_consumers": ["智能选股(data_sync.py 日K因子)", "交易复盘/个股详情日K图表"],
    },
    {
        "dataset": "valuation",
        "tier": "official",
        "v1_endpoint": "/v1/valuation",
        "primary_source": "tushare:daily_basic",
        "backup_source": None,
        "known_consumers": ["智能选股(data_sync.py 估值因子)"],
    },
    {
        "dataset": "moneyflow",
        "tier": "official",
        "v1_endpoint": "/v1/moneyflow",
        "primary_source": "tushare:moneyflow",
        "backup_source": None,
        "known_consumers": ["智能选股(data_sync.py 资金流因子)", "个股详情资金流"],
    },
    {
        "dataset": "auction",
        "tier": "official",
        "v1_endpoint": "/v1/auction",
        "primary_source": "tushare:stk_auction",
        "backup_source": None,
        "known_consumers": ["智能选股(data_sync.py 竞价快照)", "竞价板块"],
    },
    {
        "dataset": "index_daily",
        "tier": "official",
        "v1_endpoint": "/v1/indexes/bars",
        "primary_source": "tushare:index_daily",
        "backup_source": None,
        "known_consumers": ["问天(market_context.py 指数近20日走势)", "智能选股(基准回看)"],
    },
    {
        "dataset": "limit_events",
        "tier": "official",
        "v1_endpoint": "/v1/limit-events",
        "primary_source": "tushare:limit_list_d",
        "backup_source": None,
        "known_consumers": ["涨停梯队(历史/盘后视图)"],
    },
    {
        "dataset": "popularity",
        "tier": "official",
        "v1_endpoint": "/v1/popularity",
        "primary_source": "tushare:ths_hot+dc_hot",
        "backup_source": None,
        "known_consumers": ["人气榜", "智能选股(data_sync.py 人气因子)"],
    },
    {
        "dataset": "dragon_tiger",
        "tier": "official",
        "v1_endpoint": "/v1/dragon-tiger",
        "primary_source": "tushare:hm_detail",
        "backup_source": None,
        "known_consumers": ["龙虎榜"],
    },
    {
        "dataset": "sector_daily",
        "tier": "official",
        "v1_endpoint": "/v1/sectors",
        "primary_source": "tushare:ths_daily+dc_index+sw_daily",
        "backup_source": None,
        "known_consumers": ["主题轮动", "板块梯队"],
    },
    {
        "dataset": "quotes_latest",
        "tier": "provisional",
        "v1_endpoint": "/v1/quotes/latest",
        "primary_source": "eastmoney:ulist/clist",
        "backup_source": "tencent:qt",
        "known_consumers": ["竞价/股票池盘中价格", "情绪周期盘中快照"],
    },
    {
        "dataset": "index_quotes",
        "tier": "provisional",
        "v1_endpoint": "/v1/indexes/quotes",
        "primary_source": "eastmoney:ulist",
        "backup_source": "tencent:qt",
        "known_consumers": ["首页大盘指数条"],
    },
    {
        "dataset": "sectors_quote",
        "tier": "provisional",
        "v1_endpoint": "/v1/sectors/quote",
        "primary_source": "eastmoney:sw",
        "backup_source": None,
        "known_consumers": ["主题轮动盘中板块报价"],
    },
    {
        "dataset": "limit_pool",
        "tier": "provisional",
        "v1_endpoint": "/v1/limit-pool",
        "primary_source": "eastmoney:zt_pool",
        "backup_source": None,
        "known_consumers": ["涨停梯队盘中视图"],
    },
    {
        "dataset": "intraday_points",
        "tier": "provisional",
        "v1_endpoint": "/v1/intraday/points",
        "primary_source": "eastmoney:trends2",
        "backup_source": None,
        "known_consumers": ["个股详情分时图"],
    },
    {
        "dataset": "ifind_wencai",
        "tier": "licensed",
        "v1_endpoint": "/v1/query (api_name=ifind_wencai)",
        "primary_source": "ifind:smart_stock_picking",
        "backup_source": None,
        "known_consumers": ["问师(自然语言选股，需 iFinD 凭证)"],
    },
 ]
 _KNOWN_PROVIDERS_BY_SOURCE_PREFIX = ("tushare", "eastmoney", "tencent", "ifind")
 def _providers_for(primary_source: str, backup_source: str | None) -> list[str]:
    providers: list[str] = []
    for source in (primary_source, backup_source or ""):
        for provider in _KNOWN_PROVIDERS_BY_SOURCE_PREFIX:
            if source.startswith(provider) and provider not in providers:
                providers.append(provider)
    return providers
 def snapshot(db: Any, trade_date: str = "") -> list[dict[str, Any]]:
    """Attach live status to the static lineage table. Read-only; never
    raises (a per-row status lookup failure just leaves that row's status
    empty rather than failing the whole snapshot)."""
    result: list[dict[str, Any]] = []
    for entry in DATASETS:
        row = dict(entry)
        providers = _providers_for(entry["primary_source"], entry.get("backup_source"))
        row["providers"] = providers
        live: list[dict[str, Any]] = []
        try:
            if db is not None and providers:
                placeholders = ",".join("?" for _ in providers)
                live = db.fetchall(
                    f"SELECT provider, interface, state, last_error, last_fallback_reason, "
                    f"consec_failures, updated_at FROM provider_health WHERE provider IN ({placeholders})",
                    tuple(providers),
                )
        except Exception:
            live = []
        row["live_provider_health"] = live
        if entry["tier"] == "official":
            pub = None
            try:
                if db is not None and trade_date:
                    pub = db.fetchone(
                        "SELECT dataset, trade_date, state, published_at FROM publications "
                        "WHERE dataset = ? AND trade_date = ?",
                        (entry["dataset"], trade_date),
                    )
            except Exception:
                pub = None
            row["publication"] = pub
        result.append(row)
    return result
 def affected(db: Any, provider: str = "", interface: str = "") -> list[dict[str, Any]]:
    """Read-only: which datasets/pages are impacted by a given provider (and,
    optionally, a specific interface) right now. Does not change routing."""
    provider = str(provider or "").strip()
    interface = str(interface or "").strip()
    result: list[dict[str, Any]] = []
    for entry in DATASETS:
        providers = _providers_for(entry["primary_source"], entry.get("backup_source"))
        if provider and provider not in providers:
            continue
        row = dict(entry)
        row["providers"] = providers
        health: list[dict[str, Any]] = []
        try:
            if db is not None:
                if interface:
                    health = db.fetchall(
                        "SELECT provider, interface, state, last_error, last_fallback_reason, "
                        "consec_failures, updated_at FROM provider_health "
                        "WHERE provider = ? AND interface = ?",
                        (provider, interface),
                    )
                elif providers:
                    placeholders = ",".join("?" for _ in providers)
                    health = db.fetchall(
                        f"SELECT provider, interface, state, last_error, last_fallback_reason, "
                        f"consec_failures, updated_at FROM provider_health WHERE provider IN ({placeholders})",
                        tuple(providers),
                    )
        except Exception:
            health = []
        row["live_provider_health"] = health
        result.append(row)
    return result
@@ -0,0 +1,319 @@
 """Side-channel provider-call observability (HEL-543).
 This module is additive-only and must never change what any existing call
 returns or raises. It exists purely to answer, after the fact and without
 touching routing: which provider/interface was called, whether it
 succeeded, how stale/complete the payload looked, and why a fallback fired.
 Hard rules enforced here:
 - Every public entry point (`observe`, `record_call`) is wrapped so that a
  database failure, a classifier bug, or any other internal error is
  swallowed and logged at DEBUG level. It never raises into the caller and
  never delays/blocks the caller's real data path beyond a best-effort
  timing measurement.
 - `observe()` always returns exactly what `fn()` returned, and re-raises
  exactly what `fn()` raised (same exception object, unmodified). It does
  not retry, does not change ordering, and does not add new failure modes.
 - No mock data is ever produced or returned by this module.
 """
 from __future__ import annotations
 import re
 import time
 from typing import Any, Callable, TypeVar
 from datahub.timeutil import isoformat, now_shanghai
 _T = TypeVar("_T")
 # Additive-only schema: two new tables, no changes to any existing table.
 # Merged into datahub.db.SCHEMA the same way EXTENDED_SCHEMA is.
 OBS_SCHEMA = """
 CREATE TABLE IF NOT EXISTS provider_call_log (
  id INTEGER PRIMARY KEY AUTOINCREMENT,
  provider TEXT NOT NULL,
  interface TEXT NOT NULL,
  fetched_at TEXT NOT NULL,
  latency_ms INTEGER,
  status TEXT NOT NULL,
  error TEXT,
  fallback_reason TEXT,
  data_age_seconds INTEGER,
  created_at TEXT NOT NULL
 );
 CREATE TABLE IF NOT EXISTS provider_health (
  provider TEXT NOT NULL,
  interface TEXT NOT NULL,
  state TEXT NOT NULL,
  last_ok_at TEXT,
  last_error TEXT,
  last_fallback_reason TEXT,
  consec_failures INTEGER NOT NULL DEFAULT 0,
  last_latency_ms INTEGER,
  last_data_age_seconds INTEGER,
  updated_at TEXT NOT NULL,
  PRIMARY KEY (provider, interface)
 );
 CREATE INDEX IF NOT EXISTS idx_provider_call_log_created ON provider_call_log(created_at);
 CREATE INDEX IF NOT EXISTS idx_provider_call_log_provider ON provider_call_log(provider, interface, created_at);
 """
 DEFAULT_STALE_SECONDS = 300
 _BLOCKED_MARKERS = (
    "<!doctype", "<html", "expecting value", "verify you are human",
    "unusual traffic", "captcha", "安全验证", "访问异常", "请完成验证",
    "拦截", "禁止访问", "forbidden",
 )
 # Matches provider messages like "Eastmoney returned 0/3 indices" or
 # "Tencent returned 2/3 indices" (see adapters/eastmoney.py, adapters/tencent.py).
 _COUNT_MISMATCH_RE = re.compile(r"returned (\d+)\s*/\s*(\d+)")
 def _logger():
    from datahub.logutil import get_logger
    return get_logger()
 def is_enabled(db: Any) -> bool:
    """Runtime kill switch (``Settings.observability_enabled`` /
    ``DATAHUB_OBSERVABILITY``, wired onto the db handle in ``Hub.__init__``).
    Defaults to enabled when the attribute is absent — e.g. a bare ``HubDB``
    built directly in a test, or any call site that predates HEL-543 — so
    this can never accidentally disable an existing deployment. Never
    raises.
    """
    try:
        return bool(getattr(db, "observability_enabled", True))
    except Exception:  # pragma: no cover - defensive, must never raise
        return True
 def classify_error(message: str) -> tuple[str, str]:
    """Best-effort, side-reading classification of an exception message.
    Never raises. Unknown shapes fall back to a generic ``error`` status so a
    classifier miss can never be mistaken for a healthy call.
    """
    try:
        lower = (message or "").lower()
        if any(marker in lower for marker in _BLOCKED_MARKERS):
            return "blocked", "response_looks_like_intercept_page"
        if "timeout" in lower or "timed out" in lower:
            return "timeout", "request_timeout"
        mismatch = _COUNT_MISMATCH_RE.search(lower)
        if mismatch and int(mismatch.group(1)) == 0:
            return "empty", "empty_or_incomplete_response"
        if mismatch:
            return "degraded", "partial_or_mismatched_response"
        if "empty" in lower or "no intraday chart data" in lower or "missing" in lower:
            return "empty", "empty_or_incomplete_response"
        if "too small" in lower or "incomplete" in lower or "mismatch" in lower:
            return "degraded", "partial_or_mismatched_response"
        return "error", ""
    except Exception:  # pragma: no cover - defensive, must never raise
        return "error", ""
 def classify_rows(
    rows: Any,
    *,
    required_fields: tuple[str, ...] | None = None,
    freshness_field: str | None = "quote_time_epoch",
    max_age_seconds: int = DEFAULT_STALE_SECONDS,
 ) -> tuple[str, str, int | None]:
    """Read-only classification of an already-successful payload.
    Only ever called on a value a caller is about to use as-is; this never
    mutates ``rows`` and a classifier bug always degrades to ``("ok", "",
    None)`` rather than mislabeling a real success as a failure.
    """
    try:
        if isinstance(rows, dict):
            items = [rows] if rows else []
        elif isinstance(rows, (list, tuple)):
            items = [item for item in rows if isinstance(item, dict)]
        else:
            items = []
        if not items:
            return "empty", "no_rows_returned", None
        if required_fields:
            missing: set[str] = set()
            for item in items:
                for field in required_fields:
                    if item.get(field) in (None, ""):
                        missing.add(field)
            if missing:
                return "missing_fields", "missing:" + ",".join(sorted(missing)), None
        data_age: int | None = None
        if freshness_field:
            now_epoch = time.time()
            ages: list[int] = []
            for item in items:
                raw = item.get(freshness_field)
                try:
                    epoch = int(raw or 0)
                except (TypeError, ValueError):
                    epoch = 0
                if epoch > 0:
                    ages.append(max(0, int(now_epoch - epoch)))
            if ages:
                data_age = max(ages)
                if data_age > max_age_seconds:
                    return "stale", "data_age_exceeds_threshold", data_age
        return "ok", "", data_age
    except Exception:  # pragma: no cover - defensive, must never raise
        return "ok", "", None
 def record_call(
    db: Any,
    provider: str,
    interface: str,
    *,
    status: str,
    latency_ms: int | None = None,
    error: str = "",
    fallback_reason: str = "",
    data_age_seconds: int | None = None,
 ) -> None:
    """Fail-open recorder. Never raises; a write failure here must never be
    able to take down a real, otherwise-successful data path."""
    if db is None or not is_enabled(db):
        return
    try:
        now = isoformat(now_shanghai())
        ok = status == "ok"
        error_text = (error or "")[:500]
        reason_text = (fallback_reason or "")[:200]
        with db.write() as connection:
            connection.execute(
                "INSERT INTO provider_call_log("
                "provider, interface, fetched_at, latency_ms, status, error, "
                "fallback_reason, data_age_seconds, created_at) "
                "VALUES (?,?,?,?,?,?,?,?,?)",
                (provider, interface, now, latency_ms, status, error_text, reason_text, data_age_seconds, now),
            )
            connection.execute(
                """
                INSERT INTO provider_health(
                  provider, interface, state, last_ok_at, last_error, last_fallback_reason,
                  consec_failures, last_latency_ms, last_data_age_seconds, updated_at)
                VALUES (?,?,?,?,?,?,?,?,?,?)
                ON CONFLICT(provider, interface) DO UPDATE SET
                  state = excluded.state,
                  last_ok_at = CASE WHEN excluded.state = 'ok' THEN excluded.last_ok_at ELSE provider_health.last_ok_at END,
                  last_error = CASE WHEN excluded.state = 'ok' THEN '' ELSE excluded.last_error END,
                  last_fallback_reason = excluded.last_fallback_reason,
                  consec_failures = CASE WHEN excluded.state = 'ok' THEN 0 ELSE provider_health.consec_failures + 1 END,
                  last_latency_ms = excluded.last_latency_ms,
                  last_data_age_seconds = excluded.last_data_age_seconds,
                  updated_at = excluded.updated_at
                """,
                (
                    provider,
                    interface,
                    status,
                    now if ok else None,
                    "" if ok else (error_text or reason_text or "unknown_error"),
                    reason_text,
                    0 if ok else 1,
                    latency_ms,
                    data_age_seconds,
                    now,
                ),
            )
    except Exception:  # pragma: no cover - defensive, must never raise
        try:
            _logger().debug(
                "observability record_call failed (fail-open)",
                extra={"hub": {"provider": provider, "interface": interface}},
                exc_info=True,
            )
        except Exception:
            pass
 def _safe_record_failure(db: Any, provider: str, interface: str, latency_ms: int, exc: BaseException) -> None:
    if db is None:
        return
    try:
        message = str(exc)
        status, reason = classify_error(message)
        record_call(
            db, provider, interface,
            status=status, latency_ms=latency_ms, error=message, fallback_reason=reason,
        )
    except Exception:  # pragma: no cover - defensive, must never raise
        pass
 def _safe_record_success(
    db: Any,
    provider: str,
    interface: str,
    latency_ms: int,
    result: Any,
    classify: Callable[[Any], tuple[str, str, int | None] | None] | None,
 ) -> None:
    if db is None:
        return
    status, reason, data_age = "ok", "", None
    if classify is not None:
        try:
            classified = classify(result)
            if classified:
                status, reason, data_age = classified
        except Exception:  # pragma: no cover - defensive, must never raise
            # A classifier bug must never mislabel (or hide) a real success;
            # degrade to a plain "ok" call rather than skipping the log.
            status, reason, data_age = "ok", "", None
    try:
        record_call(
            db, provider, interface,
            status=status, latency_ms=latency_ms, fallback_reason=reason, data_age_seconds=data_age,
        )
    except Exception:  # pragma: no cover - defensive, must never raise
        pass
 def observe(
    db: Any,
    provider: str,
    interface: str,
    fn: Callable[[], _T],
    *,
    classify: Callable[[Any], tuple[str, str, int | None] | None] | None = None,
 ) -> _T:
    """Call ``fn()`` and record a side-channel status row.
    Returns exactly what ``fn()`` returns and re-raises exactly what
    ``fn()`` raises. ``db`` may be ``None`` (e.g. in call sites that are not
    wired to a database yet); in that case this is a transparent passthrough
    with no recording at all. Same when the ``DATAHUB_OBSERVABILITY`` kill
    switch is off (see ``is_enabled``): this becomes ``return fn()`` with no
    timing, no classification, and no db access whatsoever.
    """
    if db is not None and not is_enabled(db):
        return fn()
    started = time.perf_counter()
    try:
        result = fn()
    except Exception:
        latency_ms = round((time.perf_counter() - started) * 1000)
        import sys
        exc = sys.exc_info()[1]
        if exc is not None:
            _safe_record_failure(db, provider, interface, latency_ms, exc)
        raise
    latency_ms = round((time.perf_counter() - started) * 1000)
    _safe_record_success(db, provider, interface, latency_ms, result, classify)
    return result
@@ -20,6 +20,7 @@ from datahub.governance.ratelimit import TokenBucket
 from datahub.governance.retry import RetryError, retry_call
 from datahub.logutil import get_logger
 from datahub.normalize import finite_number, normalize_daily
 from datahub import observability
 from datahub.revision import (
    compare_fields,
    diff_published_vs_upstream,
@@ -1634,6 +1635,7 @@ class Pipeline:
                deleted += cur.rowcount
            connection.execute("DELETE FROM job_runs WHERE started_at < ?", (cutoff_jobs,))
            connection.execute("DELETE FROM src_calls WHERE created_at < ?", (cutoff_jobs,))
            connection.execute("DELETE FROM provider_call_log WHERE created_at < ?", (cutoff_jobs,))
        return {"staging_deleted": deleted}
    def audit(self, actor: str, action: str, target: str = "", detail: str = "") -> None:
@@ -1768,6 +1770,18 @@ class Pipeline:
            "INSERT INTO src_calls(provider, endpoint, ok, latency_ms, error, created_at) VALUES (?,?,?,?,?,?)",
            ("tushare", endpoint, 1 if ok else 0, latency_ms, error, isoformat(self.clock())),
        )
        # HEL-543 side channel: unified cross-provider call log/health. Kept
        # strictly additive and fail-open; the src_calls insert above (the
        # existing, already-compatible Tushare record) is unaffected either
        # way.
        if ok:
            status, reason = "ok", ""
        else:
            status, reason = observability.classify_error(error)
        observability.record_call(
            self.db, "tushare", endpoint,
            status=status, latency_ms=latency_ms, error=error, fallback_reason=reason,
        )
    def _persist_health(self, state: str, error: str = "") -> None:
        snap = self.breaker.snapshot()
@@ -18,6 +18,7 @@ from datahub.adapters.tencent import TencentAdapter
 from datahub.codes import resolve_code
 from datahub.db import HubDB
 from datahub.governance.lkg import LastKnownGood
 from datahub import observability
 from datahub.timeutil import isoformat, now_shanghai, yyyymmdd
 QUOTE_TTL = 60
@@ -25,6 +26,23 @@ INDEX_TTL = 60
 INTRADAY_TTL = 20
 QUOTE_BATCH = 60
 # HEL-543 side-channel classifiers. These only *read* an already-successful
 # payload to decide what to log; they never change the payload itself and a
 # classifier exception always degrades to "ok" (see observability.classify_rows).
 def _classify_quote_rows(rows: Any) -> tuple[str, str, int | None]:
    return observability.classify_rows(rows, freshness_field="quote_time_epoch")
 def _classify_rows_no_freshness(rows: Any) -> tuple[str, str, int | None]:
    return observability.classify_rows(rows, freshness_field=None)
 def _classify_intraday_payload(data: Any) -> tuple[str, str, int | None]:
    points = data.get("points") if isinstance(data, dict) else None
    return observability.classify_rows(points or [], freshness_field=None)
 class RealtimeApiError(RuntimeError):
    def __init__(self, code: str, message: str) -> None:
@@ -44,12 +62,17 @@ def fetch_index_quotes(db: HubDB) -> dict[str, Any]:
    cached = _read_cache(db, cache_key)
    if cached is not None:
        return cached
    eastmoney = EastmoneyAdapter()
    try:
-        rows = eastmoney.fetch_indices()
+        rows = observability.observe(
            db, "eastmoney", "indices", lambda: EastmoneyAdapter().fetch_indices(),
            classify=_classify_quote_rows,
        )
        source = "eastmoney:ulist"
    except Exception:
-        rows = TencentAdapter().fetch_indices()
+        rows = observability.observe(
            db, "tencent", "indices", lambda: TencentAdapter().fetch_indices(),
            classify=_classify_quote_rows,
        )
        source = "tencent:qt"
    if len(rows) < 3:
        raise RealtimeApiError("SOURCE_UNAVAILABLE", "index quotes incomplete")
@@ -77,7 +100,10 @@ def fetch_market_quotes(db: HubDB) -> dict[str, Any]:
    rows: list[dict[str, Any]] = []
    source = ""
    try:
-        rows = EastmoneyAdapter().fetch_market_quotes()
+        rows = observability.observe(
            db, "eastmoney", "market_quotes", lambda: EastmoneyAdapter().fetch_market_quotes(),
            classify=_classify_quote_rows,
        )
        source = "eastmoney:clist"
    except Exception as exc:
        errors.append(f"eastmoney:{exc}")
@@ -85,7 +111,10 @@ def fetch_market_quotes(db: HubDB) -> dict[str, Any]:
            listed = _listed_ts_codes(db)
            if not listed:
                raise AdapterError("no local stock master for tencent market snapshot")
-            rows = _tencent_named_quotes(listed)
+            rows = observability.observe(
                db, "tencent", "market_quotes_fallback", lambda: _tencent_named_quotes(listed),
                classify=_classify_quote_rows,
            )
            if len(rows) < 200:
                raise AdapterError(f"Tencent market snapshot too small: {len(rows)}")
            source = "tencent:qt"
@@ -130,7 +159,10 @@ def fetch_quotes(db: HubDB, codes: list[str]) -> dict[str, Any]:
    missing = [code for code in resolved if code not in by_code]
    try:
-        rows = _eastmoney_named_quotes(missing)
+        rows = observability.observe(
            db, "eastmoney", "named_quotes", lambda: _eastmoney_named_quotes(missing),
            classify=_classify_quote_rows,
        )
        by_code.update(_quote_map(rows, missing))
        if rows:
            sources.append("eastmoney:ulist")
@@ -140,7 +172,10 @@ def fetch_quotes(db: HubDB, codes: list[str]) -> dict[str, Any]:
    missing = [code for code in resolved if code not in by_code]
    if missing:
        try:
-            rows = _tencent_named_quotes(missing)
+            rows = observability.observe(
                db, "tencent", "named_quotes", lambda: _tencent_named_quotes(missing),
                classify=_classify_quote_rows,
            )
            by_code.update(_quote_map(rows, missing))
            if rows:
                sources.append("tencent:qt")
@@ -203,7 +238,10 @@ def fetch_sector_quote(db: HubDB, code: str, expected_date: str = "") -> dict[st
        return cached
    errors: list[str] = []
    try:
-        row = EastmoneyAdapter().fetch_shenwan_quote(ts_code)
+        row = observability.observe(
            db, "eastmoney", "sector_quote", lambda: EastmoneyAdapter().fetch_shenwan_quote(ts_code),
            classify=_classify_quote_rows,
        )
        if not _sector_row_matches(row, canonical_name):
            raise AdapterError(
                f"industry name mismatch: expected {canonical_name}, got {row.get('name') or '--'}"
@@ -270,7 +308,10 @@ def fetch_limit_pool(db: HubDB, trade_date: str = "") -> dict[str, Any]:
    if cached is not None:
        return cached
    try:
-        rows = EastmoneyAdapter().fetch_limit_pool(day)
+        rows = observability.observe(
            db, "eastmoney", "limit_pool", lambda: EastmoneyAdapter().fetch_limit_pool(day),
            classify=_classify_rows_no_freshness,
        )
        source = "eastmoney:zt_pool"
    except Exception as exc:
        recovered = _load_quotes_lkg(db, cache_key)
@@ -530,7 +571,10 @@ def warm_realtime(db: HubDB) -> dict[str, Any]:
            if master_code and master_name:
                canonical_names.setdefault(master_code, master_name)
        codes = list(canonical_names)
-        fetched_sector_rows = _eastmoney_sector_quotes(codes)
+        fetched_sector_rows = observability.observe(
            db, "eastmoney", "sector_quotes_batch", lambda: _eastmoney_sector_quotes(codes),
            classify=_classify_quote_rows,
        )
        for row in fetched_sector_rows:
            if _row_quote_date(row, today) != today:
                continue
@@ -588,7 +632,10 @@ def fetch_intraday(db: HubDB, code: str, date: str = "") -> dict[str, Any]:
        return cached
    adapter = EastmoneyAdapter()
    try:
-        payload_data = adapter.fetch_intraday(ts_code, date)
+        payload_data = observability.observe(
            db, "eastmoney", "intraday", lambda: adapter.fetch_intraday(ts_code, date),
            classify=_classify_intraday_payload,
        )
        source = "eastmoney:trends2"
    except Exception as exc:
        recovered = _load_intraday_lkg(db, ts_code, date)
@@ -33,6 +33,11 @@ class Settings:
    quality: dict[str, Any] = field(default_factory=dict)
    log_level: str = "INFO"
    scheduler_enabled: bool = True
    # HEL-543 kill switch: off disables the provider_call_log/provider_health
    # side channel entirely (observe()/record_call() become no-ops and the
    # new read-only admin endpoints report {"enabled": false}). Default on;
    # existing routing/fetch/publish behavior is identical either way.
    observability_enabled: bool = True
    @property
    def tushare_rate_per_minute(self) -> int:
@@ -126,4 +131,5 @@ def load_settings(
        quality=_load_quality(quality_path),
        log_level=environ.get("DATAHUB_LOG_LEVEL") or "INFO",
        scheduler_enabled=str(environ.get("DATAHUB_SCHEDULER") or "1") not in {"0", "false", "False"},
        observability_enabled=str(environ.get("DATAHUB_OBSERVABILITY") or "1") not in {"0", "false", "False", "off", "OFF"},
    )
@@ -0,0 +1,165 @@
 """Minimal, read-only source directory (HEL-543).
 Registers what already exists: providers, their concrete interfaces, what
 capability/dataset each interface serves, and whether the provider plays a
 primary or backup role. This module only *describes* the current adapters
 and datasets already wired in `datahub/hub.py`, `datahub/serving.py`, and
 `datahub/realtime_serve.py`; it does not add a way to configure or add a new
 source without code, and it never changes routing, retries, or fallback
 order.
 Every ``interfaces`` entry below is a docs-as-code mirror of a real call
 site, cross-referenced in comments so a reviewer can verify each row is
 accurate rather than aspirational:
 - tushare interfaces mirror ``datahub/serving.py``'s ``_official_meta``/``source=``
  strings and ``datahub/steward.py``'s live/published dataset table.
 - eastmoney/tencent interfaces mirror the ``observability.observe(...)``
  call sites added in ``datahub/realtime_serve.py`` for HEL-543.
 - ifind interfaces mirror ``datahub/steward.py``'s ``IFIND_APIS`` table.
 """
 from __future__ import annotations
 from typing import Any
 CATALOG: list[dict[str, Any]] = [
    {
        "provider": "tushare",
        "label": "Tushare",
        "role": "official_primary",
        "credential_key": "tushare_token",
        "status_source": "src_health (legacy, kept) + provider_health (unified, HEL-543)",
        "interfaces": [
            {"interface": "trade_cal", "capability": "交易日历", "datasets": ["calendar"]},
            {"interface": "stock_basic", "capability": "股票主档", "datasets": ["stocks"]},
            {"interface": "daily", "capability": "个股日K", "datasets": ["daily"]},
            {"interface": "adj_factor", "capability": "复权因子", "datasets": ["daily"]},
            {"interface": "daily_basic", "capability": "估值", "datasets": ["valuation"]},
            {"interface": "index_daily", "capability": "指数日K", "datasets": ["index_daily"]},
            {"interface": "moneyflow", "capability": "资金流", "datasets": ["moneyflow"]},
            {"interface": "stk_auction", "capability": "集合竞价", "datasets": ["auction"]},
            {"interface": "limit_list_d", "capability": "涨跌停池", "datasets": ["limit_events"]},
            {"interface": "ths_hot", "capability": "同花顺人气榜", "datasets": ["popularity"]},
            {"interface": "dc_hot", "capability": "东方财富人气榜", "datasets": ["popularity"]},
            {"interface": "hm_detail", "capability": "龙虎榜游资明细", "datasets": ["dragon_tiger"]},
            {"interface": "ths_daily", "capability": "同花顺概念行情", "datasets": ["sector_daily"]},
            {"interface": "dc_index", "capability": "东方财富板块行情", "datasets": ["sector_daily"]},
            {"interface": "sw_daily", "capability": "申万行业行情", "datasets": ["sector_daily"]},
        ],
    },
    {
        "provider": "eastmoney",
        "label": "东方财富",
        "role": "provisional_primary",
        "credential_key": None,
        "status_source": "provider_health (unified, HEL-543)",
        "interfaces": [
            {"interface": "indices", "capability": "指数实时报价", "datasets": ["index_quotes"]},
            {"interface": "market_quotes", "capability": "全市场实时快照", "datasets": ["quotes_latest"]},
            {"interface": "named_quotes", "capability": "指定个股实时报价", "datasets": ["quotes_latest"]},
            {"interface": "sector_quote", "capability": "申万板块实时报价(单个)", "datasets": ["sectors_quote"]},
            {"interface": "sector_quotes_batch", "capability": "申万板块批量报价(预热)", "datasets": ["sectors_quote"]},
            {"interface": "limit_pool", "capability": "涨停/炸板池(盘中)", "datasets": ["limit_pool"]},
            {"interface": "intraday", "capability": "分时走势", "datasets": ["intraday_points"]},
        ],
    },
    {
        "provider": "tencent",
        "label": "腾讯行情",
        "role": "provisional_backup",
        "credential_key": None,
        "status_source": "provider_health (unified, HEL-543)",
        "interfaces": [
            {"interface": "indices", "capability": "指数实时报价(东财失败时备用)", "datasets": ["index_quotes"]},
            {
                "interface": "market_quotes_fallback",
                "capability": "全市场快照(备用；按本地股票主档逐只请求拼接)",
                "datasets": ["quotes_latest"],
            },
            {"interface": "named_quotes", "capability": "指定个股实时报价(东财失败时备用)", "datasets": ["quotes_latest"]},
        ],
    },
    {
        "provider": "ifind",
        "label": "同花顺 iFinD",
        "role": "licensed_optional",
        "credential_key": "ifind_refresh_token",
        "status_source": "provider_health (unified, HEL-543) + adapter.status()",
        "interfaces": [
            {"interface": "wencai", "capability": "问财自然语言选股", "datasets": ["ifind_wencai"]},
            {"interface": "snapshots", "capability": "快照", "datasets": ["ifind_snapshots"]},
            {"interface": "history", "capability": "历史行情", "datasets": ["ifind_history"]},
            {"interface": "realtime", "capability": "实时行情", "datasets": ["ifind_realtime"]},
            {"interface": "intraday", "capability": "分时(高频)", "datasets": ["ifind_intraday"]},
        ],
    },
    {
        "provider": "ths",
        "label": "同花顺(预留)",
        "role": "reserved",
        "credential_key": None,
        "status_source": "adapter.probe()（占位，本阶段未接入真实数据）",
        "interfaces": [],
    },
    {
        "provider": "xgb",
        "label": "选股宝(预留)",
        "role": "reserved",
        "credential_key": None,
        "status_source": "adapter.probe()（占位，本阶段未接入真实数据）",
        "interfaces": [],
    },
    {
        "provider": "akshare",
        "label": "AKShare(预留)",
        "role": "reserved",
        "credential_key": None,
        "status_source": "adapter.probe()（占位，本阶段未接入真实数据）",
        "interfaces": [],
    },
 ]
 def snapshot(db: Any, auth: Any = None) -> list[dict[str, Any]]:
    """Merge the static catalog with live credential/health facts.
    Purely read-only: never touches routing, credentials, or adapters. Any
    failure while enriching one entry only degrades that entry's live data;
    it never drops the entry or raises, so a directory read can never break
    on a partially-unhealthy database.
    """
    result: list[dict[str, Any]] = []
    for entry in CATALOG:
        item: dict[str, Any] = {
            "provider": entry["provider"],
            "label": entry.get("label", entry["provider"]),
            "role": entry["role"],
            "status_source": entry["status_source"],
            "interfaces": [dict(i) for i in entry.get("interfaces", [])],
        }
        cred_key = entry.get("credential_key")
        if cred_key:
            cred = None
            try:
                if auth is not None:
                    cred = auth.credential_status(cred_key)
            except Exception:
                cred = None
            item["credential"] = cred or {"configured": False, "last4": "", "updated_at": ""}
        else:
            item["credential"] = {"configured": True, "last4": "", "updated_at": "", "note": "无需凭证"}
        health_rows: list[dict[str, Any]] = []
        try:
            if db is not None:
                health_rows = db.fetchall(
                    "SELECT interface, state, last_ok_at, last_error, last_fallback_reason, "
                    "consec_failures, last_latency_ms, last_data_age_seconds, updated_at "
                    "FROM provider_health WHERE provider = ? ORDER BY interface",
                    (entry["provider"],),
                )
        except Exception:
            health_rows = []
        item["live_interfaces"] = health_rows
        result.append(item)
    return result
@@ -12,6 +12,7 @@ from typing import Any
 from datahub.adapters.base import AdapterError
 from datahub.adapters.tushare import TUSHARE_FIELDS
 from datahub import observability
 from datahub.numbers import finite_number
 from datahub.realtime_serve import (
    RealtimeApiError,
@@ -152,8 +153,12 @@ def _ifind_query(api, api_name: str, params: dict[str, Any], fields: str) -> dic
        )
    if not adapter.configured:
        raise ApiError("SOURCE_UNAVAILABLE", "iFinD 尚未配置")
    db = getattr(api, "db", None)
    try:
-        rows = adapter.fetch(dataset, dict(params))
+        rows = observability.observe(
            db, "ifind", dataset, lambda: adapter.fetch(dataset, dict(params)),
            classify=lambda r: observability.classify_rows(r, freshness_field=None),
        )
    except AdapterError as exc:
        raise ApiError("SOURCE_UNAVAILABLE", str(exc)) from exc
    return envelope(
@@ -0,0 +1,180 @@
 from __future__ import annotations
 import json
 import tempfile
 import threading
 import unittest
 from http.server import ThreadingHTTPServer
 from pathlib import Path
 from urllib.request import Request, urlopen
 from datahub.adapters.tushare import TushareAdapter
 from datahub.crypto import SecretVault
 from datahub.httpapp import make_handler
 from datahub.hub import Hub
 from datahub.settings import Settings
 from tests.fixtures import TRADE_DATE, fake_transport
 class AdminObservabilityApiTests(unittest.TestCase):
    """HEL-543: new read-only admin endpoints for provider status, source
    catalog and lineage. These must never require write access and must
    never touch the existing routing/publish logic."""
    def setUp(self) -> None:
        self.tmp = tempfile.TemporaryDirectory()
        settings = Settings(
            encryption_key=SecretVault.generate_key(),
            api_token="z" * 32,
            admin_password="StartPass1",
            tushare_token="real-tushare-token-abcdef",
            db_path=Path(self.tmp.name) / "hub.db",
            scheduler_enabled=False,
        )
        self.hub = Hub(settings, adapter=TushareAdapter("real-tushare-token-abcdef", transport=fake_transport))
        handler = make_handler(self.hub)
        self.server = ThreadingHTTPServer(("127.0.0.1", 0), handler)
        threading.Thread(target=self.server.serve_forever, daemon=True).start()
        self.base = f"http://127.0.0.1:{self.server.server_address[1]}"
        _, body, cookie_header = self._json(
            "/admin/api/login", "POST", {"username": "hub_admin", "password": "StartPass1"}
        )
        cookie = cookie_header.split(";")[0]
        csrf = body["csrf"]
        self._json(
            "/admin/api/change-password",
            "POST",
            {"current": "StartPass1", "new_password": "NewPass123"},
            cookie=cookie,
            csrf=csrf,
        )
        self.cookie = cookie
        self.csrf = csrf
    def tearDown(self) -> None:
        self.server.shutdown()
        self.server.server_close()
        self.tmp.cleanup()
    def _json(self, path, method="GET", body=None, cookie="", csrf=""):
        data = None if body is None else json.dumps(body).encode()
        headers = {"Content-Type": "application/json"}
        if cookie:
            headers["Cookie"] = cookie
        if csrf:
            headers["X-CSRF-Token"] = csrf
        req = Request(self.base + path, data=data, headers=headers, method=method)
        with urlopen(req, timeout=5) as resp:
            set_cookie = resp.headers.get("Set-Cookie", "")
            return resp.status, json.loads(resp.read().decode()), set_cookie
    def _get(self, path):
        return self._json(path, cookie=self.cookie, csrf=self.csrf)
    def test_providers_status_reflects_real_pipeline_activity(self) -> None:
        pipeline = self.hub.pipeline
        pipeline.ingest_reference(TRADE_DATE)
        pipeline.run_dataset("daily", TRADE_DATE)
        status, body, _ = self._get("/admin/api/providers/status")
        self.assertEqual(status, 200)
        health = body["health"]
        self.assertTrue(any(item["provider"] == "tushare" and item["interface"] == "daily" for item in health))
        row = next(item for item in health if item["provider"] == "tushare" and item["interface"] == "daily")
        self.assertEqual(row["state"], "ok")
        recent = body["recent_calls"]
        self.assertTrue(any(item["provider"] == "tushare" and item["interface"] == "daily" for item in recent))
    def test_providers_status_filters_by_provider(self) -> None:
        pipeline = self.hub.pipeline
        pipeline.ingest_reference(TRADE_DATE)
        pipeline.run_dataset("daily", TRADE_DATE)
        status, body, _ = self._get("/admin/api/providers/status?provider=tushare")
        self.assertEqual(status, 200)
        self.assertTrue(all(item["provider"] == "tushare" for item in body["health"]))
        self.assertTrue(all(item["provider"] == "tushare" for item in body["recent_calls"]))
        status, body, _ = self._get("/admin/api/providers/status?provider=eastmoney")
        self.assertEqual(status, 200)
        self.assertEqual(body["health"], [])
        self.assertEqual(body["recent_calls"], [])
    def test_source_catalog_lists_known_providers_without_leaking_secrets(self) -> None:
        status, body, _ = self._get("/admin/api/source-catalog")
        self.assertEqual(status, 200)
        blob = json.dumps(body)
        self.assertNotIn("real-tushare-token-abcdef", blob)
        providers = {item["provider"] for item in body["items"]}
        self.assertIn("tushare", providers)
        self.assertIn("eastmoney", providers)
        self.assertIn("tencent", providers)
        self.assertIn("ifind", providers)
    def test_lineage_snapshot_and_affected_query(self) -> None:
        status, body, _ = self._get("/admin/api/lineage")
        self.assertEqual(status, 200)
        self.assertTrue(len(body["items"]) > 0)
        datasets = {item["dataset"] for item in body["items"]}
        self.assertIn("stocks", datasets)
        status, body, _ = self._get("/admin/api/lineage/affected?provider=tushare&interface=daily")
        self.assertEqual(status, 200)
        self.assertEqual(body["provider"], "tushare")
        self.assertEqual(body["interface"], "daily")
    def test_disabled_kill_switch_reports_enabled_false_with_empty_structure(self) -> None:
        # HEL-543 total-review 🔴: flip the runtime kill switch the same way
        # Hub.__init__ wires Settings.observability_enabled onto the db
        # handle, then confirm every new endpoint reports disabled with an
        # explicit empty structure rather than silently going quiet.
        self.hub.db.observability_enabled = False
        try:
            pipeline = self.hub.pipeline
            pipeline.ingest_reference(TRADE_DATE)
            pipeline.run_dataset("daily", TRADE_DATE)  # must still fully succeed
            status, body, _ = self._get("/admin/api/providers/status")
            self.assertEqual(status, 200)
            self.assertEqual(body, {"enabled": False, "health": [], "recent_calls": []})
            status, body, _ = self._get("/admin/api/source-catalog")
            self.assertEqual(status, 200)
            self.assertEqual(body, {"enabled": False, "items": []})
            status, body, _ = self._get("/admin/api/lineage")
            self.assertEqual(status, 200)
            self.assertFalse(body["enabled"])
            self.assertEqual(body["items"], [])
            status, body, _ = self._get("/admin/api/lineage/affected?provider=tushare")
            self.assertEqual(status, 200)
            self.assertEqual(
                body, {"enabled": False, "provider": "tushare", "interface": "", "items": []}
            )
            # Nothing was ever written while disabled.
            self.assertEqual(self.hub.db.fetchall("SELECT * FROM provider_call_log"), [])
        finally:
            self.hub.db.observability_enabled = True
    def test_must_change_password_blocks_new_endpoints_too(self) -> None:
        from urllib.error import HTTPError
        _, body, cookie_header = self._json(
            "/admin/api/login", "POST", {"username": "hub_admin", "password": "NewPass123"}
        )
        # Freshly logged-in user has already changed password in setUp, so
        # this login should not require a change; verify the endpoint is
        # reachable with a valid, non-must-change session (regression guard
        # against accidentally bypassing the must-change gate for these new
        # routes).
        cookie = cookie_header.split(";")[0]
        csrf = body["csrf"]
        status, _, _ = self._json("/admin/api/source-catalog", cookie=cookie, csrf=csrf)
        self.assertEqual(status, 200)
 if __name__ == "__main__":
    unittest.main()
@@ -0,0 +1,105 @@
 from __future__ import annotations
 import tempfile
 import unittest
 from pathlib import Path
 from datahub.adapters.ifind import IfindAdapter
 from datahub.db import HubDB
 from datahub.serving import ApiError
 from datahub.steward import steward_query
 class _Resp:
    def __init__(self, payload: dict, status: int = 200) -> None:
        import json
        self.status = status
        self._raw = json.dumps(payload).encode("utf-8")
    def read(self):
        return self._raw
    def __enter__(self):
        return self
    def __exit__(self, *args):
        return False
 class IfindObservabilityTests(unittest.TestCase):
    def setUp(self) -> None:
        self.tmp = tempfile.TemporaryDirectory()
        self.db = HubDB(Path(self.tmp.name) / "hub.db")
    def tearDown(self) -> None:
        self.tmp.cleanup()
    def _adapter_with_urlopen(self, urlopen) -> IfindAdapter:
        return IfindAdapter(refresh_token="rt", access_token="at", urlopen=urlopen)
    def test_successful_fetch_is_logged_without_changing_rows(self) -> None:
        def urlopen(request, timeout=None):
            return _Resp(
                {
                    "errorcode": 0,
                    "tables": [{"thscode": ["000001.SZ"], "table": {"涨停原因": ["重组"]}}],
                }
            )
        adapter = self._adapter_with_urlopen(urlopen)
        class _Api:
            ifind = adapter
            db = self.db
        payload = steward_query(
            _Api(), {"api_name": "ifind_wencai", "params": {"query": "涨停"}}
        )
        self.assertEqual(payload["data"][0]["thscode"], "000001.SZ")
        log = self.db.fetchone("SELECT * FROM provider_call_log WHERE provider = 'ifind'")
        self.assertIsNotNone(log)
        self.assertEqual(log["interface"], "wencai")
        self.assertEqual(log["status"], "ok")
    def test_failed_fetch_reraises_and_logs_error(self) -> None:
        def urlopen(request, timeout=None):
            return _Resp({"errorcode": -9999, "errmsg": "quota exceeded"})
        adapter = self._adapter_with_urlopen(urlopen)
        class _Api:
            ifind = adapter
            db = self.db
        with self.assertRaises(ApiError) as ctx:
            steward_query(_Api(), {"api_name": "ifind_wencai", "params": {"query": "涨停"}})
        self.assertEqual(ctx.exception.code, "SOURCE_UNAVAILABLE")
        log = self.db.fetchone("SELECT * FROM provider_call_log WHERE provider = 'ifind'")
        self.assertIsNotNone(log)
        self.assertEqual(log["status"], "error")
    def test_status_check_alone_does_not_dial_or_log_a_fetch_call(self) -> None:
        class _Api:
            ifind = IfindAdapter()
            db = self.db
        payload = steward_query(_Api(), {"api_name": "ifind_status", "params": {}})
        self.assertFalse(payload["data"][0]["configured"])
        log = self.db.fetchall("SELECT * FROM provider_call_log WHERE provider = 'ifind'")
        self.assertEqual(log, [])
    def test_api_double_without_db_attribute_still_works(self) -> None:
        # Mirrors tests/test_ifind_adapter.py's `_Api` double, which has no
        # `db` attribute at all. Observability must not require it.
        class _Api:
            ifind = IfindAdapter()
        payload = steward_query(_Api(), {"api_name": "ifind_status", "params": {}})
        self.assertFalse(payload["data"][0]["configured"])
        with self.assertRaises(ApiError):
            steward_query(_Api(), {"api_name": "ifind_wencai", "params": {"query": "涨停"}})
 if __name__ == "__main__":
    unittest.main()
@@ -0,0 +1,306 @@
 from __future__ import annotations
 import tempfile
 import unittest
 from contextlib import contextmanager
 from pathlib import Path
 from datahub import observability
 from datahub.db import HubDB
 class _BrokenDB:
    """A db double whose write() always raises, to prove fail-open."""
    @contextmanager
    def write(self):
        raise RuntimeError("disk is full")
        yield None  # pragma: no cover - unreachable, keeps this a generator
    def fetchall(self, sql, params=()):
        raise RuntimeError("disk is full")
    def fetchone(self, sql, params=()):
        raise RuntimeError("disk is full")
 class ClassifyRowsTests(unittest.TestCase):
    def test_empty_list_is_flagged_empty(self):
        status, reason, age = observability.classify_rows([])
        self.assertEqual(status, "empty")
        self.assertEqual(reason, "no_rows_returned")
        self.assertIsNone(age)
    def test_empty_dict_result_is_flagged_empty(self):
        status, reason, _ = observability.classify_rows({})
        self.assertEqual(status, "empty")
        self.assertEqual(reason, "no_rows_returned")
    def test_missing_required_field_is_flagged(self):
        rows = [{"ts_code": "600000.SH", "close": 10.2}, {"ts_code": "000001.SZ"}]
        status, reason, _ = observability.classify_rows(rows, required_fields=("close",), freshness_field=None)
        self.assertEqual(status, "missing_fields")
        self.assertIn("close", reason)
    def test_fresh_rows_are_ok(self):
        import time
        rows = [{"ts_code": "600000.SH", "close": 10.2, "quote_time_epoch": int(time.time())}]
        status, reason, age = observability.classify_rows(rows)
        self.assertEqual(status, "ok")
        self.assertEqual(reason, "")
        self.assertIsNotNone(age)
        self.assertLess(age, 5)
    def test_stale_rows_are_flagged(self):
        import time
        rows = [{"ts_code": "600000.SH", "close": 10.2, "quote_time_epoch": int(time.time()) - 3600}]
        status, reason, age = observability.classify_rows(rows, max_age_seconds=300)
        self.assertEqual(status, "stale")
        self.assertEqual(reason, "data_age_exceeds_threshold")
        self.assertGreaterEqual(age, 3600 - 5)
    def test_classifier_never_raises_on_garbage_input(self):
        status, reason, age = observability.classify_rows(object())
        self.assertEqual(status, "empty")
        self.assertIsNone(age)
        # Malformed rows inside a list must not raise either.
        status, _, _ = observability.classify_rows(["not-a-dict", 123, None])
        self.assertEqual(status, "empty")
 class ClassifyErrorTests(unittest.TestCase):
    def test_blocked_page_markers_are_detected(self):
        status, reason = observability.classify_error("eastmoney request failed: Expecting value: line 1 column 1")
        self.assertEqual(status, "blocked")
        self.assertEqual(reason, "response_looks_like_intercept_page")
    def test_timeout_is_detected(self):
        status, _ = observability.classify_error("tencent request failed: timed out")
        self.assertEqual(status, "timeout")
    def test_generic_error_falls_back(self):
        status, reason = observability.classify_error("connection reset by peer")
        self.assertEqual(status, "error")
        self.assertEqual(reason, "")
    def test_never_raises_on_none(self):
        status, reason = observability.classify_error(None)
        self.assertEqual(status, "error")
 class RecordCallTests(unittest.TestCase):
    def setUp(self) -> None:
        self.tmp = tempfile.TemporaryDirectory()
        self.db = HubDB(Path(self.tmp.name) / "hub.db")
    def tearDown(self) -> None:
        self.tmp.cleanup()
    def test_record_call_writes_log_and_health(self):
        observability.record_call(self.db, "eastmoney", "indices", status="ok", latency_ms=42)
        log_rows = self.db.fetchall("SELECT * FROM provider_call_log")
        self.assertEqual(len(log_rows), 1)
        self.assertEqual(log_rows[0]["provider"], "eastmoney")
        self.assertEqual(log_rows[0]["interface"], "indices")
        self.assertEqual(log_rows[0]["status"], "ok")
        health = self.db.fetchone(
            "SELECT * FROM provider_health WHERE provider = ? AND interface = ?",
            ("eastmoney", "indices"),
        )
        self.assertIsNotNone(health)
        self.assertEqual(health["state"], "ok")
        self.assertEqual(health["consec_failures"], 0)
    def test_consecutive_failures_increment_and_reset(self):
        observability.record_call(self.db, "tencent", "named_quotes", status="error", error="boom")
        observability.record_call(self.db, "tencent", "named_quotes", status="error", error="boom again")
        health = self.db.fetchone(
            "SELECT * FROM provider_health WHERE provider = ? AND interface = ?",
            ("tencent", "named_quotes"),
        )
        self.assertEqual(health["consec_failures"], 2)
        self.assertEqual(health["state"], "error")
        observability.record_call(self.db, "tencent", "named_quotes", status="ok")
        health = self.db.fetchone(
            "SELECT * FROM provider_health WHERE provider = ? AND interface = ?",
            ("tencent", "named_quotes"),
        )
        self.assertEqual(health["consec_failures"], 0)
        self.assertEqual(health["state"], "ok")
    def test_none_db_is_a_silent_noop(self):
        # Must not raise even though there is nowhere to write.
        observability.record_call(None, "ifind", "wencai", status="ok")
    def test_broken_db_write_does_not_raise(self):
        observability.record_call(_BrokenDB(), "eastmoney", "indices", status="ok")
 class ObserveTests(unittest.TestCase):
    def setUp(self) -> None:
        self.tmp = tempfile.TemporaryDirectory()
        self.db = HubDB(Path(self.tmp.name) / "hub.db")
    def tearDown(self) -> None:
        self.tmp.cleanup()
    def test_returns_exact_success_value_unmodified(self):
        sentinel = {"ts_code": "600000.SH", "close": 10.2}
        result = observability.observe(self.db, "eastmoney", "indices", lambda: sentinel)
        self.assertIs(result, sentinel)
        rows = self.db.fetchall("SELECT * FROM provider_call_log")
        self.assertEqual(len(rows), 1)
        self.assertEqual(rows[0]["status"], "ok")
    def test_reraises_exact_exception_on_failure(self):
        boom = ValueError("upstream exploded")
        def fn():
            raise boom
        with self.assertRaises(ValueError) as ctx:
            observability.observe(self.db, "eastmoney", "indices", fn)
        self.assertIs(ctx.exception, boom)
        rows = self.db.fetchall("SELECT * FROM provider_call_log")
        self.assertEqual(len(rows), 1)
        self.assertEqual(rows[0]["status"], "error")
        self.assertIn("upstream exploded", rows[0]["error"])
    def test_classify_downgrades_success_to_stale_without_changing_return_value(self):
        sentinel = [{"ts_code": "600000.SH", "quote_time_epoch": 1}]
        result = observability.observe(
            self.db, "eastmoney", "indices", lambda: sentinel,
            classify=lambda rows: observability.classify_rows(rows),
        )
        self.assertIs(result, sentinel)
        row = self.db.fetchone("SELECT * FROM provider_call_log")
        self.assertEqual(row["status"], "stale")
    def test_broken_db_never_breaks_a_successful_call(self):
        sentinel = {"ok": True}
        result = observability.observe(_BrokenDB(), "eastmoney", "indices", lambda: sentinel)
        self.assertIs(result, sentinel)
    def test_broken_db_never_masks_a_real_failure(self):
        def fn():
            raise RuntimeError("real upstream failure")
        with self.assertRaises(RuntimeError) as ctx:
            observability.observe(_BrokenDB(), "eastmoney", "indices", fn)
        self.assertEqual(str(ctx.exception), "real upstream failure")
    def test_classifier_exception_does_not_break_the_call(self):
        sentinel = {"ok": True}
        def bad_classify(_result):
            raise KeyError("classifier bug")
        result = observability.observe(self.db, "eastmoney", "indices", lambda: sentinel, classify=bad_classify)
        self.assertIs(result, sentinel)
        row = self.db.fetchone("SELECT * FROM provider_call_log")
        # A classifier bug must degrade to "ok", never silently drop the row
        # nor claim the call failed when it did not.
        self.assertEqual(row["status"], "ok")
    def test_none_db_is_transparent_passthrough(self):
        sentinel = object()
        result = observability.observe(None, "eastmoney", "indices", lambda: sentinel)
        self.assertIs(result, sentinel)
 class _ToggleDB(HubDB):
    """A real HubDB subclass so we can flip the HEL-543 kill switch the same
    way Hub.__init__ does, without needing a full Hub/Settings wiring."""
 class KillSwitchTests(unittest.TestCase):
    """HEL-543 total-review 🔴: the observability side channel must be
    disable-able at runtime, and disabling it must leave existing behavior
    completely unchanged (pure passthrough, zero db access)."""
    def setUp(self) -> None:
        self.tmp = tempfile.TemporaryDirectory()
        self.db = _ToggleDB(Path(self.tmp.name) / "hub.db")
    def tearDown(self) -> None:
        self.tmp.cleanup()
    def test_is_enabled_defaults_true_when_attribute_absent(self):
        # A bare HubDB (as used throughout the rest of this test suite, and
        # by any pre-HEL-543 call site) must default to enabled.
        self.assertTrue(observability.is_enabled(self.db))
        self.assertTrue(observability.is_enabled(None))
    def test_disabled_record_call_writes_nothing(self):
        self.db.observability_enabled = False
        observability.record_call(self.db, "eastmoney", "indices", status="ok", latency_ms=1)
        self.assertEqual(self.db.fetchall("SELECT * FROM provider_call_log"), [])
        self.assertEqual(self.db.fetchall("SELECT * FROM provider_health"), [])
    def test_disabled_observe_is_a_pure_passthrough_on_success(self):
        self.db.observability_enabled = False
        sentinel = {"ts_code": "600000.SH"}
        calls = {"n": 0}
        def fn():
            calls["n"] += 1
            return sentinel
        result = observability.observe(self.db, "eastmoney", "indices", fn)
        self.assertIs(result, sentinel)
        self.assertEqual(calls["n"], 1)
        self.assertEqual(self.db.fetchall("SELECT * FROM provider_call_log"), [])
    def test_disabled_observe_still_reraises_the_exact_exception(self):
        self.db.observability_enabled = False
        boom = RuntimeError("upstream exploded")
        def fn():
            raise boom
        with self.assertRaises(RuntimeError) as ctx:
            observability.observe(self.db, "eastmoney", "indices", fn)
        self.assertIs(ctx.exception, boom)
        self.assertEqual(self.db.fetchall("SELECT * FROM provider_call_log"), [])
    def test_re_enabling_resumes_recording(self):
        self.db.observability_enabled = False
        observability.observe(self.db, "eastmoney", "indices", lambda: {"ok": True})
        self.assertEqual(self.db.fetchall("SELECT * FROM provider_call_log"), [])
        self.db.observability_enabled = True
        observability.observe(self.db, "eastmoney", "indices", lambda: {"ok": True})
        self.assertEqual(len(self.db.fetchall("SELECT * FROM provider_call_log")), 1)
 class SettingsToggleTests(unittest.TestCase):
    """The kill switch follows the same env-var pattern as DATAHUB_SCHEDULER."""
    def test_defaults_to_enabled(self):
        from datahub.settings import load_settings
        settings = load_settings(env={})
        self.assertTrue(settings.observability_enabled)
    def test_datahub_observability_zero_disables(self):
        from datahub.settings import load_settings
        settings = load_settings(env={"DATAHUB_OBSERVABILITY": "0"})
        self.assertFalse(settings.observability_enabled)
    def test_datahub_observability_off_disables(self):
        from datahub.settings import load_settings
        settings = load_settings(env={"DATAHUB_OBSERVABILITY": "off"})
        self.assertFalse(settings.observability_enabled)
    def test_datahub_observability_one_keeps_enabled(self):
        from datahub.settings import load_settings
        settings = load_settings(env={"DATAHUB_OBSERVABILITY": "1"})
        self.assertTrue(settings.observability_enabled)
 if __name__ == "__main__":
    unittest.main()
@@ -0,0 +1,82 @@
 from __future__ import annotations
 import unittest
 from datahub.pipeline import RetryError
 from tests.fixtures import TRADE_DATE
 from tests.test_pipeline import make_pipeline
 class PipelineObservabilityTests(unittest.TestCase):
    """HEL-543: Tushare calls must keep writing the existing `src_calls`
    record unchanged, while also feeding the new cross-provider
    `provider_call_log` / `provider_health` side channel."""
    def test_successful_fetch_logs_to_both_src_calls_and_provider_call_log(self) -> None:
        pipe, db = make_pipeline()
        pipe.ingest_reference(TRADE_DATE)
        result = pipe.run_dataset("daily", TRADE_DATE)
        self.assertEqual(result["state"], "published")
        src_calls = db.fetchall("SELECT * FROM src_calls WHERE provider = 'tushare' AND endpoint = 'daily'")
        self.assertTrue(any(row["ok"] == 1 for row in src_calls))
        log = db.fetchall(
            "SELECT * FROM provider_call_log WHERE provider = 'tushare' AND interface = 'daily'"
        )
        self.assertTrue(len(log) >= 1)
        self.assertEqual(log[-1]["status"], "ok")
        health = db.fetchone(
            "SELECT * FROM provider_health WHERE provider = 'tushare' AND interface = 'daily'"
        )
        self.assertIsNotNone(health)
        self.assertEqual(health["state"], "ok")
        self.assertEqual(health["consec_failures"], 0)
    def test_failed_fetch_logs_error_to_both_channels_and_still_raises(self) -> None:
        pipe, db = make_pipeline()
        pipe.ingest_reference(TRADE_DATE)
        def boom(dataset, params):
            raise RuntimeError("tushare upstream 500")
        pipe.adapter.fetch = boom  # type: ignore[assignment]
        with self.assertRaises(RetryError):
            pipe.run_dataset("daily", TRADE_DATE, attempts=1)
        src_calls = db.fetchall("SELECT * FROM src_calls WHERE provider = 'tushare' AND endpoint = 'daily' AND ok = 0")
        self.assertTrue(len(src_calls) >= 1)
        self.assertIn("tushare upstream 500", src_calls[-1]["error"])
        log = db.fetchall(
            "SELECT * FROM provider_call_log WHERE provider = 'tushare' AND interface = 'daily' AND status != 'ok'"
        )
        self.assertTrue(len(log) >= 1)
        self.assertIn("tushare upstream 500", log[-1]["error"])
        health = db.fetchone(
            "SELECT * FROM provider_health WHERE provider = 'tushare' AND interface = 'daily'"
        )
        self.assertIsNotNone(health)
        self.assertNotEqual(health["state"], "ok")
        self.assertGreaterEqual(health["consec_failures"], 1)
    def test_provider_call_log_is_purged_by_existing_cleanup_job(self) -> None:
        pipe, db = make_pipeline()
        pipe.ingest_reference(TRADE_DATE)
        pipe.run_dataset("daily", TRADE_DATE)
        self.assertTrue(db.fetchall("SELECT * FROM provider_call_log"))
        # Force everything to look ancient so cleanup() sweeps it.
        db.execute("UPDATE provider_call_log SET created_at = '2000-01-01T00:00:00+08:00'")
        db.execute("UPDATE src_calls SET created_at = '2000-01-01T00:00:00+08:00'")
        db.execute("UPDATE job_runs SET started_at = '2000-01-01T00:00:00+08:00'")
        pipe.cleanup()
        self.assertEqual(db.fetchall("SELECT * FROM provider_call_log"), [])
 if __name__ == "__main__":
    unittest.main()
@@ -0,0 +1,163 @@
 from __future__ import annotations
 import tempfile
 import unittest
 from contextlib import contextmanager
 from pathlib import Path
 from unittest.mock import patch
 from datahub.adapters.base import AdapterError
 from datahub.db import HubDB
 from datahub.realtime_serve import fetch_index_quotes, fetch_intraday, fetch_quotes
 class _WriteBreaksDB:
    """Wraps a real HubDB but breaks only the write path, to prove the
    real serving path (reads/caches) is untouched by an observability
    failure while still exercising real fetch/cache code around it."""
    def __init__(self, real: HubDB) -> None:
        self._real = real
    def fetchall(self, sql, params=()):
        return self._real.fetchall(sql, params)
    def fetchone(self, sql, params=()):
        return self._real.fetchone(sql, params)
    def execute(self, sql, params=()):
        return self._real.execute(sql, params)
    def executemany(self, sql, rows):
        return self._real.executemany(sql, rows)
    @contextmanager
    def write(self):
        raise RuntimeError("db is not writable right now")
        yield None  # pragma: no cover
 class RealtimeObservabilityTests(unittest.TestCase):
    def setUp(self) -> None:
        self.tmp = tempfile.TemporaryDirectory()
        self.db = HubDB(Path(self.tmp.name) / "hub.db")
    def tearDown(self) -> None:
        self.tmp.cleanup()
    def test_eastmoney_success_is_logged_without_changing_payload(self) -> None:
        rows = [
            {"ts_code": "000001.SH", "code": "000001", "name": "上证指数", "price": 3000.0,
             "previous_close": 2990.0, "quote_time_epoch": 0, "source": "eastmoney_push2"},
            {"ts_code": "399001.SZ", "code": "399001", "name": "深证成指", "price": 9000.0,
             "previous_close": 8990.0, "quote_time_epoch": 0, "source": "eastmoney_push2"},
            {"ts_code": "399006.SZ", "code": "399006", "name": "创业板指", "price": 1800.0,
             "previous_close": 1790.0, "quote_time_epoch": 0, "source": "eastmoney_push2"},
        ]
        with patch("datahub.realtime_serve.EastmoneyAdapter") as mocked:
            mocked.return_value.fetch_indices.return_value = rows
            payload = fetch_index_quotes(self.db)
        self.assertEqual(payload["data"], rows)
        self.assertEqual(payload["meta"]["source"], "eastmoney:ulist")
        log = self.db.fetchall("SELECT * FROM provider_call_log WHERE provider = 'eastmoney'")
        self.assertEqual(len(log), 1)
        self.assertEqual(log[0]["interface"], "indices")
        self.assertEqual(log[0]["status"], "ok")
        health = self.db.fetchone(
            "SELECT * FROM provider_health WHERE provider = 'eastmoney' AND interface = 'indices'"
        )
        self.assertEqual(health["state"], "ok")
    def test_eastmoney_failure_falls_back_to_tencent_and_logs_both(self) -> None:
        tencent_rows = [
            {"ts_code": "000001.SH", "code": "000001", "name": "上证指数", "price": 3000.0,
             "previous_close": 2990.0, "quote_time_epoch": 0, "source": "tencent_qt"},
            {"ts_code": "399001.SZ", "code": "399001", "name": "深证成指", "price": 9000.0,
             "previous_close": 8990.0, "quote_time_epoch": 0, "source": "tencent_qt"},
            {"ts_code": "399006.SZ", "code": "399006", "name": "创业板指", "price": 1800.0,
             "previous_close": 1790.0, "quote_time_epoch": 0, "source": "tencent_qt"},
        ]
        with patch("datahub.realtime_serve.EastmoneyAdapter") as eastmoney, patch(
            "datahub.realtime_serve.TencentAdapter"
        ) as tencent:
            eastmoney.return_value.fetch_indices.side_effect = AdapterError("Eastmoney returned 0/3 indices")
            tencent.return_value.fetch_indices.return_value = tencent_rows
            payload = fetch_index_quotes(self.db)
        self.assertEqual(payload["meta"]["source"], "tencent:qt")
        self.assertEqual(payload["data"], tencent_rows)
        east_log = self.db.fetchone("SELECT * FROM provider_call_log WHERE provider = 'eastmoney'")
        self.assertEqual(east_log["status"], "empty")
        tencent_log = self.db.fetchone("SELECT * FROM provider_call_log WHERE provider = 'tencent'")
        self.assertEqual(tencent_log["status"], "ok")
    def test_observability_db_failure_never_breaks_a_real_successful_fetch(self) -> None:
        rows = [
            {"ts_code": "000001.SH", "price": 3000.0, "previous_close": 2990.0, "quote_time_epoch": 0},
            {"ts_code": "399001.SZ", "price": 9000.0, "previous_close": 8990.0, "quote_time_epoch": 0},
            {"ts_code": "399006.SZ", "price": 1800.0, "previous_close": 1790.0, "quote_time_epoch": 0},
        ]
        broken = _WriteBreaksDB(self.db)
        with patch("datahub.realtime_serve.EastmoneyAdapter") as mocked:
            mocked.return_value.fetch_indices.return_value = rows
            payload = fetch_index_quotes(broken)
        self.assertEqual(payload["data"], rows)
        self.assertEqual(payload["meta"]["source"], "eastmoney:ulist")
    def test_observability_db_failure_never_masks_a_real_source_outage(self) -> None:
        broken = _WriteBreaksDB(self.db)
        with patch("datahub.realtime_serve.EastmoneyAdapter") as eastmoney, patch(
            "datahub.realtime_serve.TencentAdapter"
        ) as tencent:
            eastmoney.return_value.fetch_indices.side_effect = AdapterError("down")
            tencent.return_value.fetch_indices.side_effect = AdapterError("also down")
            with self.assertRaises(Exception):
                fetch_index_quotes(broken)
    def test_named_quotes_records_both_providers_on_partial_merge(self) -> None:
        with patch("datahub.realtime_serve.EastmoneyAdapter") as eastmoney, patch(
            "datahub.realtime_serve.TencentAdapter"
        ) as tencent:
            eastmoney.return_value.fetch_quotes.return_value = [
                {"ts_code": "000001.SZ", "close": 10, "pre_close": 9, "quote_date": "20260907"},
            ]
            tencent.return_value.fetch_quotes.return_value = [
                {"ts_code": "000002.SZ", "close": 20, "pre_close": 19, "quote_date": "20260907"},
            ]
            payload = fetch_quotes(self.db, ["000001.SZ", "000002.SZ"])
        self.assertEqual(payload["meta"]["complete"], True)
        east_log = self.db.fetchone(
            "SELECT * FROM provider_call_log WHERE provider = 'eastmoney' AND interface = 'named_quotes'"
        )
        self.assertEqual(east_log["status"], "ok")
        tencent_log = self.db.fetchone(
            "SELECT * FROM provider_call_log WHERE provider = 'tencent' AND interface = 'named_quotes'"
        )
        self.assertEqual(tencent_log["status"], "ok")
    def test_intraday_success_is_logged_as_ok(self) -> None:
        payload_data = {
            "entity_type": "stock", "ts_code": "601318.SH", "trade_date": "2026-09-07",
            "previous_close": 55.8, "points": [{"date": "2026-09-07", "time": "09:30", "close": 55.9}],
        }
        with patch("datahub.realtime_serve.EastmoneyAdapter") as mocked:
            mocked.return_value.fetch_intraday.return_value = payload_data
            payload = fetch_intraday(self.db, "601318.SH")
        self.assertEqual(payload["data"], payload_data)
        log = self.db.fetchone(
            "SELECT * FROM provider_call_log WHERE provider = 'eastmoney' AND interface = 'intraday'"
        )
        self.assertEqual(log["status"], "ok")
    def test_intraday_failure_is_logged_as_empty(self) -> None:
        with patch("datahub.realtime_serve.EastmoneyAdapter") as mocked:
            mocked.return_value.fetch_intraday.side_effect = AdapterError("No intraday chart data returned")
            with self.assertRaises(Exception):
                fetch_intraday(self.db, "000001.SZ")
        log = self.db.fetchone(
            "SELECT * FROM provider_call_log WHERE provider = 'eastmoney' AND interface = 'intraday'"
        )
        self.assertEqual(log["status"], "empty")
 if __name__ == "__main__":
    unittest.main()