#!/usr/bin/env python3 # -*- coding: utf-8 -*- """stack_gate.py - 栈水位趋势判据（task-32 定稿的可执行形态，纯标准库）。 模块名称：台架栈水位判据器（趋势 + 采集健康门） 模块功能：把 N 段启动日志里的 [STACK] 水位汇成"本轮最大值"，与基线轮最大值按每任务 容差比较，输出四档结论；**先过采集健康门，再谈比较**。 创建日期：2026-10-27（裁定见 docs/内存管理/RAM使用分析与优化方案.md §4.0.6 第③条） 为什么要有它：旧判据「栈水位不劣于上一轮」拿单次运行时峰值跨轮比较，而同一份镜像自己 就会摆（实测 uiTask 4116~4188 = 72 B、adcTask 920~1000 = 80 B），于是既假红又漏报。 本脚本把那条判据拆成三件事，等级不同、互不越权： 1) 硬线（唯一拦提交项）= scripts/stack_report.py 的每任务 verdict： margin = cfg - max(静态最坏链, 运行时峰值) >= max(isr_margin, cfg*30%) 本脚本不重算也不替代它，只在末尾提示其退出码（接不接 pre-commit 由 Lead 定）。 2) 趋势（告警级，永不单独拦提交）= 同一身份戳、>=3 圈取最大值，与基线轮最大值比， 超每任务容差 -> WARN。容差按字节，来源见 scripts/stack_tolerance.tsv。 3) 作废（VOID，优先于前两者）= 采集健康门不过：本轮既不判过也不判不过，也不得写进台账 结论。判据不许依赖"采集看起来很健康"这种自我感觉——2026-10-27 实证：常驻 daemon 死后 status 会自动拉起一个新 daemon，仍报"正常"而 rx_lines=0。 退出码（与 verify_log.py 同族口径：测量问题绝不伪装成设备故障）： 0 = PASS（含通过但有趋势 WARN） 1 = FAIL（运行时绝对底线破线：free_min < 25% of cfg —— 设备侧真消息） 2 = VOID_MEASUREMENT（采集健康门未过：本轮不可用，也不算设备的账） 3 = 用法错误 """ import argparse import os import re import sys RUNTIME_FLOOR_PCT = 25 # 运行时绝对底线（与静态 30% 硬线错开一档：峰值只是下界证据） MIN_CIRCLES = 3 # 少于 3 圈 = 噪声样本，不做跨轮比较 DEFAULT_TOL = 160 # 未登记任务的兜底容差（字节），推导见 scripts/stack_tolerance.tsv 头注 STACK_RE = re.compile(r"\[STACK\]\s+(\w+)\s+peak_used=(\d+)/(\d+)\s+free_min=(\d+)\s*\((\d+)%\)") # __DATE__ 的月份是**英文名**（"Oct 26 2026 22:52:09"），不是数字——早先写成 \d{3} 会让 # 每一份真实日志都判成"没有身份戳"而整轮作废（判据全红比判据漏更响，但也别拿假红当纪律）。 STAMP_RE = re.compile(r"fw build:\s*([A-Za-z]{3}\s+\d{1,2}\s+\d{4}\s+\d{2}:\d{2}:\d{2})\s+suite=(\S+)\s+dstr=(\S+)") ANCHOR = "reset cause:" def parse_block(text): """取一段日志的 (每任务峰值, 身份戳, 锚点数, 有效行数)。 同一任务在一块里出现多次时取最大值——栈峰值的语义就是"最坏那一次"，不做平均。 """ payload = [l for l in text.splitlines() if l.strip() and not l.lstrip().startswith("#")] peaks = {} for ln in payload: m = STACK_RE.search(ln) if m: name = m.group(1) used, cfg, free = int(m.group(2)), int(m.group(3)), int(m.group(4)) prev = peaks.get(name) if prev is None or used > prev["used"]: peaks[name] = {"used": used, "cfg": cfg, "free": free} stamps = STAMP_RE.findall("\n".join(payload)) uniq = sorted(set(stamps)) stamp = None if uniq: stamp = uniq[0] if len(uniq) == 1 else ("MIXED:" + str(len(uniq))) return {"peaks": peaks, "anchors": sum(1 for l in payload if ANCHOR in l), "stamp": stamp, "lines": len(payload)} def measurement_gate(blocks, label): """(ok, reasons)。任一条不满足就整轮作废：宁可不出结论，也不出错结论。""" reasons = [] if len(blocks) < MIN_CIRCLES: reasons.append("%s: only %d circle(s), need >= %d for a peak-vs-peak comparison" % (label, len(blocks), MIN_CIRCLES)) stamps = set() for i, b in enumerate(blocks): n = i + 1 if b["anchors"] != 1: reasons.append("%s block#%d: boot anchor count=%d (want exactly 1) -> capture did " "not start at a reset, or the windows overlap" % (label, n, b["anchors"])) if b["lines"] < 20: reasons.append("%s block#%d: only %d payload line(s) -> the collector was not " "carrying this boot (a respawned daemon still answers 'ok')" % (label, n, b["lines"])) if not b["peaks"]: reasons.append("%s block#%d: zero [STACK] lines (a missing line is not zero used)" % (label, n)) if not b["stamp"]: reasons.append("%s block#%d: no 'fw build:' identity stamp -> cannot prove which " "image these numbers belong to" % (label, n)) elif str(b["stamp"]).startswith("MIXED"): reasons.append("%s block#%d: %s identity stamps inside one block -> the capture " "crossed two images" % (label, n, b["stamp"])) else: stamps.add(b["stamp"]) # 同一轮内部出现多个身份戳 = 采集窗跨了一次重烧，这轮的数字不属于任何一份镜像 -> 作废。 # 注意：**跨轮**戳不同是正常的（那正是要比的两份镜像），不能按作废处理，见 evaluate()。 if len(stamps) > 1: reasons.append("%s: this round mixes %d identity stamps %s -> one circle was reflashed " "mid-round, these peaks belong to no single image" % (label, len(stamps), sorted(stamps))) return (not reasons), reasons def load_tolerances(path): """容差表：每行 taskbytes[来源]，# 开头为注释。缺文件则全用默认值。""" tol = {} if not path or not os.path.isfile(path): return tol, "no tolerance file -> default %d B for every task" % DEFAULT_TOL for raw in open(path, "r", encoding="utf-8").read().splitlines(): line = raw.strip() if not line or line.startswith("#"): continue parts = line.split("\t") if len(parts) >= 2 and parts[1].isdigit(): tol[parts[0]] = int(parts[1]) return tol, "loaded %d task tolerance(s) from %s" % (len(tol), os.path.basename(path)) def worst_of(blocks): """把本轮各圈压成"每任务最大值"（峰值就该取最坏一次）。""" agg = {} for b in blocks: for name, info in b["peaks"].items(): prev = agg.get(name) if prev is None or info["used"] > prev["used"]: agg[name] = dict(info) return agg def decide(cur, base, tol_map): """跨轮趋势：只出 WARN。FAIL 的权限只给绝对底线，硬线归 stack_report.py。""" rows, warns = [], [] for name in sorted(cur): info = cur[name] bl = base.get(name) tol = tol_map.get(name, DEFAULT_TOL) pct = (100 * info["free"] // info["cfg"]) if info["cfg"] else 0 row = {"task": name, "cur_max": info["used"], "cfg": info["cfg"], "free_min": info["free"], "floor_pct": pct, "tol": tol, "base_max": (bl or {}).get("used"), "delta": (info["used"] - bl["used"]) if bl else None, "level": "PASS"} if pct < RUNTIME_FLOOR_PCT: row["level"] = "FAIL" warns.append("%s free_min=%d/%d (%d%%) below the %d%% runtime floor" % (name, info["free"], info["cfg"], pct, RUNTIME_FLOOR_PCT)) elif bl is not None and (info["used"] - bl["used"]) > tol: row["level"] = "WARN" warns.append("%s peak rose %d B over baseline max (%d -> %d, tol=%d B): trend only, " "the blocking decision is stack_report.py's margin verdict" % (name, info["used"] - bl["used"], bl["used"], info["used"], tol)) rows.append(row) return rows, warns def evaluate(cur_blocks, base_blocks, tol_map): """(exit_code, verdict, 报告行)。顺序：先门，再绝对底线，最后才是趋势告警。""" out = [] ok_cur, why_cur = measurement_gate(cur_blocks, "current") if not ok_cur: out.append("[VOID_MEASUREMENT] this round is not usable:") out.extend(" - " + r for r in why_cur) out.append(" -> fix the measurement first (open the port and watch rx_lines grow,") out.append(" capture from a reset, prove the identity stamp). A dead collector must") out.append(" not become a stack FAIL, and this round must NOT enter the ledger.") return 2, "VOID_MEASUREMENT", out if base_blocks: ok_base, why_base = measurement_gate(base_blocks, "baseline") if not ok_base: out.append("[WARN] baseline round failed its own gate -> trend skipped this run:") out.extend(" - " + r for r in why_base) out.append(" (we refuse to compare against numbers we cannot trust; the current") out.append(" round still gets its absolute-floor check)") base_blocks = [] cur = worst_of(cur_blocks) base = worst_of(base_blocks) if base_blocks else {} rows, warns = decide(cur, base, tol_map) cur_stamp = cur_blocks[0]["stamp"] base_stamp = base_blocks[0]["stamp"] if base_blocks else None cross_image = bool(base_blocks) and cur_stamp != base_stamp out.append("%-9s %8s %9s %6s %7s %6s %5s %s" % ("task", "cur_max", "base_max", "delta", "cfg", "free%", "tol", "level")) for r in rows: out.append("%-9s %8d %9s %6s %7d %5d%% %5d %s" % ( r["task"], r["cur_max"], "-" if r["base_max"] is None else r["base_max"], "-" if r["delta"] is None else ("%+d" % r["delta"]), r["cfg"], r["floor_pct"], r["tol"], r["level"])) fails = [w for w in warns if "floor" in w] trend = [w for w in warns if "floor" not in w] for w in fails: out.append("[FAIL] " + w) for w in trend: out.append("[WARN] " + w) if not base_blocks: out.append("note: no usable baseline -> trend comparison off this run (levels come from " "the absolute runtime floor only)") elif cross_image: out.append("note: CROSS-IMAGE comparison (current=%s vs baseline=%s). Different stamps " "are normal here -- but a delta is only attributable to a given change if the " "source diff between those two builds is attached; write BOTH stamps into the " "ledger row, never the commit SHA alone (SHAs moved three times on 2026-10-26)." % (cur_stamp[0], base_stamp[0])) else: out.append("note: same-image comparison (stamps match: %s) -> a delta over tolerance is " "measurement noise or an unexplained regression; re-measure before writing it up." % (cur_stamp[0],)) out.append("hard line unchanged: run scripts/stack_report.py; its per-task verdict " "(margin >= max(isr_margin, 30% of cfg), computed from the larger of the static " "chain and the runtime peak) is the only stack criterion that blocks a commit.") if fails: return 1, "FAIL", out if trend: return 0, "PASS_WITH_WARN", out return 0, "PASS", out def synth(stamp, anchors, entries, filler=24): """造一段日志文本（selftest 用）。entries = [(task, used, cfg, free)]。""" L = ["[RX] 2026-10-27T10:00:%02d.000+08:00 [PROF] filler=%d" % (i % 60, i) for i in range(filler)] if stamp: L.append("[RX] 2026-10-27T10:00:00.000+08:00 [2026-10-27 02:00:00][APP] fw build: " + stamp + " suite=NONE dstr=off") for _ in range(anchors): L.append("[RX] 2026-10-27T10:00:00.100+08:00 [2026-10-27 02:00:00][APP] reset cause: " "software (NVIC_SystemReset) (CSR=0x14000000)") for task, used, cfg, free in entries: L.append("[RX] 2026-10-27T10:00:01.000+08:00 [2026-10-27 02:00:01][STACK] %s " "peak_used=%d/%d free_min=%d (%d%%)" % (task, used, cfg, free, 100 * free // cfg)) return "\n".join(L) + "\n" def selftest(out=print): """证明四档都能真的亮——不能失败的判据等于没有判据。""" cases = [] def chk(name, got, want): cases.append((name, got == want, "got=%r want=%r" % (got, want))) tasks = ("default", "net", "ftp", "ui", "timesync", "adc", "iec104", "protect", "test") clean = [(t, 1000, 4096, 3096) for t in tasks] S = "Oct 26 2026 22:52:09" b1 = parse_block(synth(S, 1, clean)) b2 = parse_block(synth(S, 1, [(t, 1040, 4096, 3056) for t in tasks])) b3 = parse_block(synth(S, 1, [(t, 990, 4096, 3106) for t in tasks])) chk("three clean circles -> PASS", evaluate([b1, b2, b3], [], {})[1], "PASS") base72 = [parse_block(synth("Oct 26 2026 19:43:23", 1, [("ui", 4116, 16384, 12268)]))] cur72 = [parse_block(synth("Oct 26 2026 19:43:23", 1, [("ui", 4188, 16384, 12196)]))] * 3 chk("the measured 72 B swing is not a red flag", evaluate(cur72, base72, {"ui": 160})[1], "PASS") cur_big = [parse_block(synth("Oct 26 2026 19:43:23", 1, [("ui", 4388, 16384, 11996)]))] * 3 base72r = base72 * 3 # 基线轮也要 >=3 圈：单圈基线是噪声样本，不许拿它比 chk("delta over tolerance = WARN not FAIL", evaluate(cur_big, base72r, {"ui": 160})[1], "PASS_WITH_WARN") chk("single-circle baseline is refused (trend skipped, no false red)", evaluate(cur_big, base72, {"ui": 160})[1], "PASS") chk("cross-round different stamps still compares (with attribution note)", 0, evaluate([parse_block(synth("Oct 26 2026 22:52:09", 1, [("ui", 4400, 16384, 11984)]))] * 3, base72r, {"ui": 160})[0]) low = [parse_block(synth("Oct 26 2026 19:43:23", 1, [("ui", 13000, 16384, 3384)]))] * 3 chk("runtime floor breach = FAIL", evaluate(low, [], {})[1], "FAIL") dead = parse_block(synth(S, 0, [])) chk("dead collector -> VOID (not FAIL)", evaluate([dead] * 3, [], {})[1], "VOID_MEASUREMENT") two = parse_block(synth(S, 2, clean)) chk("two boot anchors -> VOID", evaluate([two, two, two], [], {})[1], "VOID_MEASUREMENT") nostamp = parse_block(synth("", 1, clean)) chk("missing identity stamp -> VOID", evaluate([nostamp] * 3, [], {})[1], "VOID_MEASUREMENT") mixed = parse_block(synth(S, 1, clean) + synth("Oct 25 2026 09:00:00", 1, clean)) chk("mixed stamps in one block -> VOID", evaluate([mixed] * 3, [], {})[1], "VOID_MEASUREMENT") cross = parse_block(synth("Oct 25 2026 09:00:00", 1, clean)) chk("round that mixes two stamps -> VOID", evaluate([b1, b2, cross], [], {})[1], "VOID_MEASUREMENT") chk("only two circles -> VOID", evaluate([b1, b2], [], {})[1], "VOID_MEASUREMENT") chk("unusable baseline -> trend skipped, floor still enforced", evaluate([b1, b2, b3], [dead], {})[1], "PASS") chk("peak takes the worst circle, not the last", worst_of([b1, b3, b2])["ui"]["used"], 1040) bad = 0 for name, ok, detail in cases: out("%-5s %s%s" % ("[OK]" if ok else "[FAIL]", name, "" if ok else " " + detail)) if not ok: bad += 1 out("SELFTEST: %d case(s), %d failure(s) -> %s" % (len(cases), bad, "PASS" if bad == 0 else "FAIL")) return 0 if bad == 0 else 1 def main(argv=None): ap = argparse.ArgumentParser(description="bench stack water-level trend gate (task-32 ruling)") ap.add_argument("--block", action="append", default=[], help="current-round boot log block, repeatable (need >= %d)" % MIN_CIRCLES) ap.add_argument("--baseline-block", action="append", default=[], help="previous-round boot log block, repeatable; omit = trend comparison off") ap.add_argument("--tolerance-tsv", default="scripts/stack_tolerance.tsv", help="per-task tolerance table (task TAB bytes); missing task -> %d B" % DEFAULT_TOL) ap.add_argument("--selftest", action="store_true", help="run built-in classification cases and exit") a = ap.parse_args(argv) if a.selftest: return selftest() if not a.block: print("usage error: give at least one --block (or --selftest)") return 3 tol_map, note = load_tolerances(a.tolerance_tsv) blocks = [parse_block(open(p, "r", encoding="utf-8", errors="replace").read()) for p in a.block] base = [parse_block(open(p, "r", encoding="utf-8", errors="replace").read()) for p in a.baseline_block] print("stack trend gate -- %s (floor=%d%% of cfg, min circles=%d)" % (note, RUNTIME_FLOOR_PCT, MIN_CIRCLES)) code, verdict, lines = evaluate(blocks, base, tol_map) for ln in lines: print(ln) print("VERDICT: %s" % verdict) return code if __name__ == "__main__": sys.exit(main())