时序库降采样审计「四盏绿灯账全错」：保留期越界不夹取 / 过期窗口用全局值 / 分桶时区错位 / avg 定标 / 空判据表假绿

$ python3 -m app.run
wrote /srv/tsdb/reports/rollup.json ok=True       # rc=0
$ python3 -m app.health
{"degraded": false, "issues": []}                 # rc=0
$ bash ops/repro.sh
RESULT: OK                                        # rc=0
$ jq '.ok,.counts' reports/rollup.json
true / {"bad_retention":1,"buckets":34,"clamped_retention":0,"expired":0,...}

RESULT: BAD（不一致 19 条）
   1. counts.kept 期望 35 实际 38
   2. counts.expired 期望 3 实际 0
   3. counts.clamped_retention 期望 2 实际 0
   7. footprint.total_bytes 期望 1080 实际 1184
  18. buckets[300] 条数 期望 18 实际 20

git clone <repo> && cd scenario/2026-10-26-1001
docker compose up -d --build          # app/ 是构建时 COPY 进镜像的，没有 bind mount
docker compose exec -T tsdb bash      # 非交互必须 -T，否则 "the input device is not a TTY"

bash ops/repro.sh                              # 恒 RESULT: OK —— 不是判定依据
python3 client/tsdb_probe.py --now=1793000000  # 外部独立探针，唯一说真话
bash ops/verify.sh                             # 逐条断言契约 → PASS=15 FAIL=20
bash ops/negative.sh                           # 问题态负向闸门 → NPASS=8 NFAIL=0

cd scenario/2026-10-26-1001
TSDB_ROOT=$PWD NOW_EPOCH=1793000000 python3 -m app.run
TSDB_ROOT=$PWD python3 client/tsdb_probe.py --now=1793000000

cfg = dict(POLICY_DEFAULTS)          # UTC / 300s / skew=0 兜底
try:
    cfg.update(json.load(open(path)))
except OSError:
    pass                             # 文件没了？当没发生，用默认值继续跑

if spec == "inf":
    return 36000
try:
    return int(spec)                 # 900 就返回 900，上限 400 完全没检查
except Exception:
    return raw_retention_days

def cutoff_of(conf, now):
    days = int(conf["raw_retention_days"])   # 永远是全局 7 天
    return now - days * 86400

base = (ts + 28800) // resolution * resolution
return base - LOCAL_OFFSET          # ← 契约要本地秒，这里又减回去了

def fmt(value, places):
    q = Decimal(1).scaleb(-places)          # places 直接用样本原始位数
    return str(value.quantize(q))

"ok": True,       # 契约：ok = expired==0 and skew==0 and bad_retention==0 and clamped_retention==0

BAD_FIELDS = ()

def check(report):
    for field in BAD_FIELDS:      # 空循环，什么都查不出来
        ...
    return issues                 # 永远 []

def main(argv=None):
    ...
    return 0                      # 无条件 0

class PolicyError(Exception): ...

def validate(cfg):
    missing = [k for k in REQUIRED if k not in cfg]
    if missing: raise PolicyError("policy 缺键: %s" % ",".join(missing))
    unknown = [k for k in cfg if k not in REQUIRED]
    if unknown: raise PolicyError("policy 未知键: %s" % ",".join(sorted(unknown)))
    if cfg["timezone"] != "Asia/Shanghai": raise PolicyError(...)
    if not _is_int(cfg["raw_retention_days"]) or cfg["raw_retention_days"] <= 0: raise ...
    if any(res[i] >= res[i+1] for i in range(len(res)-1)): raise ...   # 必须严格递增
    ...

def retention_days(spec, conf):
    """返回 (days, kind)，kind ∈ ok|bad|clamped"""
    if spec == "inf":
        return INF_RETENTION_DAYS, "ok"
    try:
        raw = int(spec)
    except (TypeError, ValueError):
        return int(conf["raw_retention_days"]), "bad"
    lo, hi = int(conf["min_retention_days"]), int(conf["max_retention_days"])
    if raw < lo: return lo, "clamped"
    if raw > hi: return hi, "clamped"
    return raw, "ok"

days, kind = retention_days(rows[0]["retention"], conf)
cutoff = now - days * 86400          # days 是该组的，不是全局的

def bucket_start(ts, resolution):
    return ((ts + LOCAL_OFFSET) // resolution) * resolution    # 本地秒，别再减回去

def group_places(rows):                       # 组内最大小数位，钳到 [2,3]
    return max(2, min(3, max([r["places"] for r in rows] or [0])))

def scale_places(rollup, g_places, pts):
    return g_places if rollup == "sum" else max(p[2] for p in pts)

def fmt(value, places):
    places = max(2, min(3, places))
    return str(Decimal(value).quantize(Decimal(1).scaleb(-places), rounding=ROUND_HALF_UP))

def is_ok(counts):
    return (counts["expired"] == 0 and counts["skew"] == 0
            and counts["bad_retention"] == 0 and counts["clamped_retention"] == 0)

BAD_FIELDS = ("counts", "footprint", "groups", "buckets", "checksum", "ok")

def check(report, root=None, now=None):
    if report is None: return ["report missing or unparsable"], 0
    exp = recompute(root, now)            # ← 独立实现
    issues = compare(exp, report)         # 逐字段
    if not exp["ok"]:
        issues.append("报告 ok=false：存在被策略改写/丢弃的样本")
    return issues, checks

EXIT_OK, EXIT_FACTS, EXIT_CONTRACT = 0, 2, 3
# run:      PolicyError→3，OSError(samples 不可读)→2，写出报告→0
# health:   有偏差/ok=false→2，策略不可用→3
# service:  完全透传 health

$ python3 client/tsdb_probe.py --now=1793000000
{"checksum": "1e79775a9bb53b76", "counts": {"bad_retention": 1, "buckets": 30,
 "clamped_retention": 2, "expired": 3, "holdout": 1, "kept": 35, "raw_samples": 40,
 "series": 7, "skew": 1}, "ok": false, "result": "OK"}             # rc=0

$ bash ops/verify.sh
PASS=35 FAIL=0
RESULT: OK                                                         # rc=0

$ jq -c "{counts,footprint,ok,checksum}" reports/rollup.json
{"counts":{"bad_retention":1,"buckets":30,"clamped_retention":2,"expired":3,"holdout":1,
 "kept":35,"raw_samples":40,"series":7,"skew":1},
 "footprint":{"raw_bytes":840,"rollup_bytes":240,"total_bytes":1080},
 "ok":false,"checksum":"1e79775a9bb53b76"}

$ python3 -m app.health   → {"checks":23,"degraded":true,"issues":["报告 ok=false：…"]}   rc=2
$ python3 -m app.service  → 同上                                                          rc=2

cfg = dict(DEFAULTS)
try: cfg.update(json.load(open(path)))
except OSError: pass