@@ -0,0 +1,310 @@
#!/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 % o f 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__ 的月份是**英文名**（
# 每一份真实日志都判成"没有身份戳"而整轮作废（判据全红比判据漏更响，但也别拿假红当纪律）。
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 ) :
""" 容差表：每行 task<TAB>bytes[<TAB>来源]，# 开头为注释。缺文件则全用默认值。 """
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 % o f 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 ( ) )
