Cloudbird-Software · randypanding · Aug 21, 2026 · Aug 21, 2026 · Aug 21, 2026 · Aug 21, 2026
@@ -196,6 +196,165 @@ def sli_stuck(repos):
     return stuck
 
 
+# @w5c4-pure-begin —— 纯函数区（governance/tests/test-metrics-wiring.sh 按标记对
+# 提取本块离线单测——不复制实现，防"测试测影子"；标记对缺失=测试红）
+DRILL_MARKS = ("演练", "演习", "[drill]")  # 演习数据约定标记（sli-report"演练"+ADR-0069"演习"双词兼容）
+
+
+def _ts(s):
+    """ISO→datetime（失败 None）；本块自包含（不依赖模块级 _iso——提取测试可独立运行）。"""
+    try:
+        return _dt.datetime.fromisoformat(str(s or "").replace("Z", "+00:00"))
+    except ValueError:
+        return None
-def _ts(s):
-    """ISO→datetime（失败 None）；本块自包含（不依赖模块级 _iso——提取测试可独立运行）。"""
-    try:
-        return _dt.datetime.fromisoformat(str(s or "").replace("Z", "+00:00"))
-    except ValueError:
-        return None
+def _ts(s):
+    """ISO→datetime（失败 None）；本块自包含（不依赖模块级 _iso——提取测试可独立运行）。"""
+    try:
+        value = _dt.datetime.fromisoformat(str(s or "").replace("Z", "+00:00"))
+        if value.tzinfo is None:
+            return None
+        return value.astimezone(_dt.timezone.utc)
+    except ValueError:
+        return None
-def _ts(s):
-    """ISO→datetime（失败 None）；本块自包含（不依赖模块级 _iso——提取测试可独立运行）。"""
-    try:
-        return _dt.datetime.fromisoformat(str(s or "").replace("Z", "+00:00"))
-    except ValueError:
-        return None
+def _ts(s):
+    """ISO→datetime（失败 None）；本块自包含（不依赖模块级 _iso——提取测试可独立运行）。"""
+    try:
+        value = _dt.datetime.fromisoformat(str(s or "").replace("Z", "+00:00"))
+        if value.tzinfo is None:
+            return None
+        return value.astimezone(_dt.timezone.utc)
+    except ValueError:
+        return None
+
+
+def _is_drill_text(*texts):
+    joined = " ".join(str(t or "") for t in texts)
+    return any(m in joined for m in DRILL_MARKS)
+
+
+def partition_escapes(prs, p0s, now):
+    """双窗逃逸事件：current=[now-7d,now)，previous=[now-14d,now-7d)。
+
+    prs=merged PR 节点（[auto-revert] 标题约定）；p0s=post-merge 冒烟 P0 issue。
+    演习数据（title/body 含约定标记）从分子排除且 drills_excluded 计数可见——
+    过滤不可见=作弊通道（sli-report 先例）。reverts_current 供回滚率护栏（分子
+    只算 revert，不含 P0）。
+    """
+    w = _dt.timedelta(days=7)
+    cur = prev = drills = reverts_cur = 0
+    for p in prs or []:
+        if "[auto-revert]" not in str(p.get("title") or ""):
+            continue
+        if _is_drill_text(p.get("title"), p.get("body")):
+            drills += 1
+            continue
+        ts = _ts(p.get("mergedAt"))
+        if ts and now - 2 * w < ts <= now:
+            if ts > now - w:
+                cur += 1
+                reverts_cur += 1
+            else:
+                prev += 1
+    for i in p0s or []:
+        if _is_drill_text(i.get("title"), i.get("body")):
+            drills += 1
+            continue
+        ts = _ts(i.get("created_at"))
+        if ts and now - 2 * w < ts <= now:
+            if ts > now - w:
+                cur += 1
+            else:
+                prev += 1
+    return {"current": cur, "previous": prev, "reverts_current": reverts_cur,
+            "drills_excluded": drills}
+
+
+def drill_redrate_lines(lines):
+    """drill history.jsonl 行→红率输入（kind=seed-drill，red/denom=red+green——
+    与 drill.py redrate 同口径，no-surface 与 failclose 演习不入分母；畸形行计入 bad 可见）。"""
+    red = denom = bad = 0
+    for ln in lines or []:
+        ln = ln.strip()
+        if not ln or ln.startswith("#"):
+            continue
+        try:
+            rec = json.loads(ln)
+        except ValueError:
+            bad += 1
+            continue
+        if rec.get("kind") != "seed-drill":
+            continue
+        v = rec.get("verdict")
+        if v in ("red", "green"):
+            denom += 1
+            if v == "red":
+                red += 1
+    return {"red": red, "denom": denom, "bad_lines": bad}
+
+
+def false_decision_parse(text, now, window_days):
+    """arbiter 误放行台账文本→（窗内 false-allow 数, 窗内 false-deny 数, 全部行列表）。
+    `#` 注释行跳过（台账文件头约定）；date 出窗不计。"""
+    allow = deny = 0
+    lines = []
+    for ln in str(text or "").splitlines():
+        ln = ln.strip()
+        if not ln or ln.startswith("#"):
+            continue
+        try:
+            rec = json.loads(ln)
+        except ValueError:
+            continue
+        lines.append(rec)
+        ts = _ts(rec.get("date"))
+        if ts and (now - ts).days <= window_days:
+            if rec.get("kind") == "false-allow":
+                allow += 1
+            elif rec.get("kind") == "false-deny":
+                deny += 1
+    return allow, deny, lines
+
+
+def sign_durations(timelines):
+    """type:intent issue 的 timeline 事件列表→（签署耗时秒列表, 在途 draft 数）。
+
+    耗时=首个 labeled state:ir-signed 时刻 − 首个 labeled state:ir-draft 时刻
+    （宪法 §7：签署耗时如实计入判断预算）。无 draft 事件的已签 IR 不可算→跳过
+    （不造 0）；有 draft 无 signed→在途。"""
+    durations, in_flight = [], 0
+    for events in timelines or []:
+        draft = signed = None
+        for ev in events or []:
+            if ev.get("event") != "labeled":
+                continue
+            name = (ev.get("label") or {}).get("name")
+            ts = _ts(ev.get("created_at"))
+            if not ts:
+                continue
+            if name == "state:ir-draft" and draft is None:
+                draft = ts
+            elif name == "state:ir-signed" and signed is None and draft is not None:
+                signed = ts
+        if draft and signed and signed > draft:
+            durations.append(round((signed - draft).total_seconds()))
+        elif draft and not signed:
+            in_flight += 1
+    return durations, in_flight
+
+
+def dwell_hours(events, now, label="state:needs-human"):
+    """timeline 事件→进入 label 态至今停留小时数（取最近一次 labeled 时刻——
+    反复进出取当前段）。无该事件→None。"""
+    latest = None
+    for ev in events or []:
+        if ev.get("event") != "labeled":
+            continue
+        if (ev.get("label") or {}).get("name") != label:
+            continue
+        ts = _ts(ev.get("created_at"))
+        if ts and (latest is None or ts > latest):
+            latest = ts
+    if latest is None or latest > now:
+        return None
+    return round((now - latest).total_seconds() / 3600, 2)
+
+
+def user_metric_from(content_text):
+    """产品仓 user-result.yaml 文本→指标 dict（缺 metric_key/value=不完整→None）。
+    本地 import yaml——提取测试独立运行不依赖模块级导入。"""
+    import yaml
+    try:
+        d = yaml.safe_load(content_text)
+    except Exception:
+        return None
+    if isinstance(d, dict) and d.get("metric_key") and "value" in d:
+        return d
+    return None
+# @w5c4-pure-end
+
+
 def build_payload(repos, cards, purl=""):
     rate, denom = sli_automerge(repos)
     sli = {"automerge_rate": rate, "human_touch_per_pr": None, "escape_rate": None,

@@ -0,0 +1,148 @@
+#!/usr/bin/env bash
+# test-metrics-wiring.sh —— dashboard 采集层纯函数自测（W5-C4 .github#227，ADR-0073）
+#
+# 从 dashboard-update.py 按 @w5c4-pure 标记对提取纯函数区（不复制实现——防"测试
+# 测影子"，test-ir0002.sh 同模式；标记对缺失=fail-closed 红），fixture 断言：
+#   逃逸双窗分割（sustained 无状态化）· 演习过滤可见 · 演习红率分母口径
+#   误放行台账窗过滤 · 签署耗时 timeline 差（无 draft 不造 0）· needs-human 停留
+#   产品仓用户结果指标读取位（缺失/畸形=pending 不造数）
+# 用法：bash governance/tests/test-metrics-wiring.sh（零网络零真实 gh）
+set -uo pipefail
+HERE="$(cd "$(dirname "$0")" && pwd)"
+GOV="$(cd "$HERE/.." && pwd)"
+
+PASS=0; FAIL=0
+pass() { PASS=$((PASS+1)); echo "PASS  $1"; }
+fail() { FAIL=$((FAIL+1)); echo "FAIL  $1"; }
+
+PY=""
+for c in "${PYTHON:-}" python3 python py -3; do
+  [[ -n "$c" ]] || continue
+  "$c" -c 'import sys, yaml; print("ok")' >/dev/null 2>&1 || continue
+  PY="$c"; break
+done
+[[ -n "$PY" ]] || { echo "::error::无可用 python（含 pyyaml）"; exit 2; }
+
+TMP=$(mktemp -d); trap 'rm -rf "$TMP"' EXIT
+SRC="$GOV/dashboard-update.py"
+[[ -f "$SRC" ]] || { echo "FATAL: dashboard-update.py 不存在"; exit 2; }
+
+# --- 提取被测纯函数区（标记对缺失=fail-closed） ---
+awk '/@w5c4-pure-begin/{f=1} f{print} /@w5c4-pure-end/{exit}' "$SRC" >"$TMP/pure_body.py"
+if ! grep -q '^def partition_escapes(' "$TMP/pure_body.py"; then
+  echo "FATAL: 标记对内未找到纯函数定义（提取失效——实现与测试脱钩）"; exit 2
+fi
+# 垫片头：提取块只依赖 stdlib（datetime/json）+局部 yaml——补导入即可独立运行
+{ echo 'import datetime as _dt'; echo 'import json'; cat "$TMP/pure_body.py"; } >"$TMP/pure.py"
-# --- 提取被测纯函数区（标记对缺失=fail-closed） ---
-awk '/@w5c4-pure-begin/{f=1} f{print} /@w5c4-pure-end/{exit}' "$SRC" >"$TMP/pure_body.py"
-if ! grep -q '^def partition_escapes(' "$TMP/pure_body.py"; then
-  echo "FATAL: 标记对内未找到纯函数定义（提取失效——实现与测试脱钩）"; exit 2
-fi
-# 垫片头：提取块只依赖 stdlib（datetime/json）+局部 yaml——补导入即可独立运行
-{ echo 'import datetime as _dt'; echo 'import json'; cat "$TMP/pure_body.py"; } >"$TMP/pure.py"
+# --- 提取被测纯函数区（标记对缺失=fail-closed） ---
+awk '/@w5c4-pure-begin/{f=1} f{print} /@w5c4-pure-end/{exit}' "$SRC" >"$TMP/pure_body.py"
+if ! grep -q '^# @w5c4-pure-end$' "$TMP/pure_body.py" ||
+   ! grep -q '^def partition_escapes(' "$TMP/pure_body.py"; then
+  echo "FATAL: 标记对内未找到纯函数定义（提取失效——实现与测试脱钩）"; exit 2
+fi
+# 垫片头：提取块只依赖 stdlib（datetime/json）+局部 yaml——补导入即可独立运行
+{ echo 'import datetime as _dt'; echo 'import json'; cat "$TMP/pure_body.py"; } >"$TMP/pure.py"
-# --- 提取被测纯函数区（标记对缺失=fail-closed） ---
-awk '/@w5c4-pure-begin/{f=1} f{print} /@w5c4-pure-end/{exit}' "$SRC" >"$TMP/pure_body.py"
-if ! grep -q '^def partition_escapes(' "$TMP/pure_body.py"; then
-  echo "FATAL: 标记对内未找到纯函数定义（提取失效——实现与测试脱钩）"; exit 2
-fi
-# 垫片头：提取块只依赖 stdlib（datetime/json）+局部 yaml——补导入即可独立运行
-{ echo 'import datetime as _dt'; echo 'import json'; cat "$TMP/pure_body.py"; } >"$TMP/pure.py"
+# --- 提取被测纯函数区（标记对缺失=fail-closed） ---
+awk '/@w5c4-pure-begin/{f=1} f{print} /@w5c4-pure-end/{exit}' "$SRC" >"$TMP/pure_body.py"
+if ! grep -q '^# @w5c4-pure-end$' "$TMP/pure_body.py" ||
+   ! grep -q '^def partition_escapes(' "$TMP/pure_body.py"; then
+  echo "FATAL: 标记对内未找到纯函数定义（提取失效——实现与测试脱钩）"; exit 2
+fi
+# 垫片头：提取块只依赖 stdlib（datetime/json）+局部 yaml——补导入即可独立运行
+{ echo 'import datetime as _dt'; echo 'import json'; cat "$TMP/pure_body.py"; } >"$TMP/pure.py"
+
+cat >>"$TMP/pure.py" <<'PYEOF'
+
+# ==== 驱动断言（now 固定——离线可复现，owner 复算同款） ====
+NOW = _dt.datetime(2026, 8, 22, 0, 0, tzinfo=_dt.timezone.utc)
+W7 = _dt.timedelta(days=7)
+results = []
+
+def check(name, cond):
+    results.append((name, bool(cond)))
+
+# 1) 逃逸双窗分割：current=[now-7d,now) previous=[now-14d,now-7d)；演习排除且可见
+prs = [
+    {"title": "[auto-revert] #11 bad fix", "body": "x",
+     "mergedAt": (NOW - _dt.timedelta(days=1)).isoformat()},          # 本窗 revert
+    {"title": "[auto-revert] #10 older", "body": "x",
+     "mergedAt": (NOW - _dt.timedelta(days=10)).isoformat()},         # 上一窗 revert
+    {"title": "[auto-revert] drill tail", "body": "演习收尾",
+     "mergedAt": (NOW - _dt.timedelta(days=2)).isoformat()},          # 演习排除
+    {"title": "feat: normal", "body": "x",
+     "mergedAt": (NOW - _dt.timedelta(days=2)).isoformat()},          # 非 revert
+    {"title": "[auto-revert] ancient", "body": "x",
+     "mergedAt": (NOW - _dt.timedelta(days=30)).isoformat()},         # 出 14d 窗
+]
+p0s = [
+    {"title": "post-merge 冒烟失败 #91", "created_at": (NOW - _dt.timedelta(days=3)).isoformat()},
+    {"title": "post-merge 冒烟失败 #80", "created_at": (NOW - _dt.timedelta(days=9)).isoformat()},
+    {"title": "post-merge 冒烟失败 drill", "body": "[drill]", "created_at": (NOW - _dt.timedelta(days=1)).isoformat()},
+]
+esc = partition_escapes(prs, p0s, NOW)
+check("逃逸双窗 current=2（1 revert+1 P0）", esc["current"] == 2)
+check("逃逸双窗 previous=2（1 revert+1 P0）", esc["previous"] == 2)
+check("回滚率分子 reverts_current=1（不含 P0）", esc["reverts_current"] == 1)
+check("演习排除计数可见=2（过滤不可见=作弊通道）", esc["drills_excluded"] == 2)
+check("空输入零窗", partition_escapes([], [], NOW) == {"current": 0, "previous": 0, "reverts_current": 0, "drills_excluded": 0})
+
+# 2) 演习红率：kind=seed-drill + no-surface 不入分母（drill.py redrate 同口径）；畸形行计数可见
+lines = [
+    '# 台账头注释',
+    '{"kind":"seed-drill","verdict":"red"}',
+    '{"kind":"seed-drill","verdict":"green"}',
+    '{"kind":"seed-drill","verdict":"no-surface"}',
+    '{"kind":"failclose-drill","outcome":"pass"}',   # 非种子演习——不入红率分母
+    'not-json-line',
+]
+agg = drill_redrate_lines(lines)
+check("演习红率 red=1 denom=2（no-surface 与 failclose 出分母）", agg["red"] == 1 and agg["denom"] == 2)
+check("畸形行 bad_lines=1 可见", agg["bad_lines"] == 1)
+
+# 3) 误放行台账：注释跳过 · 窗过滤 · infra 不算误拒（ADR-0054 §7）
+ledger = '\n'.join([
+    '# arbiter 误放行/误拒台账（false decision ledger，ADR-0054 §7）',
+    '{"date": "2026-08-15T00:00:00Z", "kind": "false-allow"}',
+    '{"date": "2026-08-20T00:00:00Z", "kind": "false-deny"}',
+    '{"date": "2026-05-01T00:00:00Z", "kind": "false-allow"}',
+    '{"date": "2026-08-21T00:00:00Z", "kind": "infra"}',
+])
+allow, deny, fd_lines = false_decision_parse(ledger, NOW, 30)
+check("误放行窗内=1（窗外不计）", allow == 1)
+check("误拒窗内=1（infra 不计）", deny == 1)
+check("行列表=4（注释行剔除）", len(fd_lines) == 4)
+
+# 4) 签署耗时：draft→signed timeline 差；在途计数；无 draft 不造 0
+t_ok = [
+    {"event": "labeled", "label": {"name": "state:ir-draft"}, "created_at": "2026-08-20T10:00:00Z"},
+    {"event": "labeled", "label": {"name": "state:ir-signed"}, "created_at": "2026-08-20T11:30:00Z"},
+]
+t_fast = [
+    {"event": "labeled", "label": {"name": "state:ir-draft"}, "created_at": "2026-08-21T10:00:00Z"},
+    {"event": "labeled", "label": {"name": "state:ir-signed"}, "created_at": "2026-08-21T10:00:30Z"},
+]
+t_inflight = [{"event": "labeled", "label": {"name": "state:ir-draft"}, "created_at": "2026-08-18T00:00:00Z"}]
+t_nodraft = [{"event": "labeled", "label": {"name": "state:ir-signed"}, "created_at": "2026-08-19T00:00:00Z"}]
+durs, inflight = sign_durations([t_ok, t_fast, t_inflight, t_nodraft])
+check("签署耗时=[5400, 30]（90min 与 30s 快签）", durs == [5400, 30])
+check("在途 draft=1", inflight == 1)
+check("signed 无 draft 不入统计（不造 0）", len(durs) == 2)
+
+# 5) needs-human 停留：取最近一次 labeled 时刻（反复进出取当前段）
+ev = [
+    {"event": "labeled", "label": {"name": "state:needs-human"}, "created_at": "2026-08-10T00:00:00Z"},
+    {"event": "labeled", "label": {"name": "state:in-progress"}, "created_at": "2026-08-15T00:00:00Z"},
+    {"event": "labeled", "label": {"name": "state:needs-human"}, "created_at": "2026-08-21T00:00:00Z"},
+]
+h = dwell_hours(ev, NOW)
+check("停留=24h（最近一次 needs-human）", h == 24.0)
+check("无事件→None", dwell_hours([], NOW) is None)
+check("未来时戳→None（不造负数）", dwell_hours(
+    [{"event": "labeled", "label": {"name": "state:needs-human"}, "created_at": "2026-08-23T00:00:00Z"}], NOW) is None)
+
+# 6) 用户结果读取位：完整→ok；缺 value→None；非 yaml→None（pending 不造数）
+check("完整声明→dict", user_metric_from("metric_key: dau\nvalue: 42\nunit: 人\n") == {"metric_key": "dau", "value": 42, "unit": "人"})
+check("缺 metric_key→None", user_metric_from("value: 42\n") is None)
+check("畸形 yaml→None", user_metric_from(":::not yaml[") is None)
+check("空文本→None", user_metric_from("") is None)
+
+bad = [n for n, ok in results if not ok]
+for n, ok in results:
+    print(("PASS  " if ok else "FAIL  ") + n)
+raise SystemExit(1 if bad else 0)
+PYEOF
+
+if "$PY" "$TMP/pure.py" >"$TMP/run.txt" 2>"$TMP/err.txt"; then
+  sed 's/^/  /' "$TMP/run.txt"
+  pass "dashboard 采集层纯函数 fixture 全过（$(grep -c '^PASS' "$TMP/run.txt") 项）"
+else
+  sed 's/^/  /' "$TMP/run.txt" "$TMP/err.txt" 2>/dev/null
+  fail "采集层纯函数断言失败（详见上行）"
+fi
+
+echo "== test-metrics-wiring: pass=$PASS fail=$FAIL =="
+[[ $FAIL -eq 0 ]]