${fact('步数', `${step.toLocaleString()}${total ? ' / ' + total.toLocaleString() : ''}`)}
${fact('已用时间', hms(elapsed))}
- ${fact('剩余时间', finished ? '—' : hms(eta))}
+ ${fact('剩余时间', finished ? '—' : hmsShort(eta))} // 预估只到分钟（用时才精确到秒）
${fact('预计完成', fin)}
${kind === 'apg' ? fact('本批展开', r.plans ? `${r.plans} 次推理` : '—') : fact('学习率', isNum(r.lr) ? r.lr.toExponential(2) : '—')}
${fact('日志更新', lastMtime ? ago(lastMtime) : '—', running ? '' : 'warn')}
@@ -508,7 +508,7 @@ function stages(s) {
const r = P.stages[k], nm = SM[k].n;
if (!r) return [nm, 'pending', '—'];
if (P.current === k) return [nm, 'active', P.progress && P.progress.frac != null ? pct(P.progress.frac, 0) : '进行中'];
- if (r.rc === 0) return [nm, 'done', '用时 ' + hmsShort(el[k] != null ? el[k] : r.end - r.start)];
+ if (r.rc === 0) return [nm, 'done', '用时 ' + hms(el[k] != null ? el[k] : r.end - r.start)];
if (r.rc === -15 || r.rc === 143) return [nm, 'fail', '已停止'];
if (r.rc != null) return [nm, 'fail', `失败（返回码 ${r.rc}）`];
return [nm, 'fail', '中断'];
@@ -631,7 +631,7 @@ function renderPipe(s) {
const doneN = plan.filter(k => P.stages[k] && P.stages[k].rc === 0 && P.current !== k).length;
const failed = plan.find(k => P.stages[k] && P.stages[k].rc != null && P.stages[k].rc !== 0);
const allDone = !cur && doneN === plan.length;
- let remain = 0; for (const k of plan) { const r = P.stages[k]; if (!r || r.rc == null) remain += (SM[k]?.w || 0.1) * (k === cur ? 1 - (pr.frac || 0) : 1); }
+ let remain = 0, calib = false; for (const k of plan) { const r = P.stages[k]; if (!r || r.rc == null) { if (k === cur && pr.eta_s != null) { remain += pr.eta_s / 3600; calib = true; } else remain += (SM[k]?.w || 0.1) * (k === cur ? 1 - (pr.frac || 0) : 1); } }
const head = cur ? `
正在运行${SM[cur]?.n || cur}${pr.label || ''}
`
: allDone ? `
状态全部完成共 ${plan.length} 步
`
: failed ? `
状态停在「${SM[failed]?.n || failed}」${(P.stages[failed].info || {}).timed_out || P.stages[failed].rc === 124 ? '超过时间上限，判定卡死' : P.stages[failed].rc === -15 || P.stages[failed].rc === 143 ? '已停止' : '失败，看终端输出'}
`
@@ -644,8 +644,8 @@ function renderPipe(s) {
总用时
${hms(plan.reduce((a, k) => a + ((P.elapsed || {})[k] || 0), 0))}
本步用时
${cur ? hms((P.elapsed || {})[cur] || 0) : '—'}
-
本步粗估 / 上限
${cur ? hmsShort((SM[cur]?.w || 0) * 3600) + ' / ' + hmsShort((cr && cr.timeout_h || SM[cur]?.to || 0) * 3600) : '—'}
-
全流程剩余（粗估）
${allDone ? '—' : '约 ' + hmsShort(remain * 3600)}
+
${cur && pr.eta_s != null ? '本步预计剩余（实测速率）' : '本步粗估 / 上限'}
${cur ? (pr.eta_s != null ? hmsShort(pr.eta_s) + ' / 上限 ' + hmsShort((cr && cr.timeout_h || SM[cur]?.to || 0) * 3600) : hmsShort((SM[cur]?.w || 0) * 3600) + ' / ' + hmsShort((cr && cr.timeout_h || SM[cur]?.to || 0) * 3600)) : '—'}
+
全流程剩余（${calib ? '实测外推' : '粗估'}）
${allDone ? '—' : '约 ' + hmsShort(remain * 3600)}
${btns}`;
}
function renderJob() {
diff --git a/ml/monitor.py b/ml/monitor.py
index 0388f5b..d1b577c 100644
--- a/ml/monitor.py
+++ b/ml/monitor.py
@@ -117,7 +117,8 @@ def mahm():
psutil 在 Windows 没有 sensors_temperatures；ACPI 热区只有个 27.9 °C 的板温；Power Meter 计数器没有实例。
条目布局实测：头 32 字节（签名 MAHM / 版本 / 头大小 / 条目数 / 条目大小 / 时间戳），每条 1324 字节，
名字在前 260 字节、单位在其后 260、数值是偏移 1300 的 float（用 GPU1 power 对 nvidia-smi 校准过）。
- 读数比 10 秒旧就当没读到：Afterburner 关了或停了就别显示一个冻住的数。"""
+ 读数比 30 秒旧才当没读到：Afterburner 的硬件轮询周期实测 10~13 秒（不是 1 秒），阈值卡 10 秒会
+ 正好落在两次刷新之间、把还在用的读数丢掉（表现为温度和功耗一闪一闪地消失）。"""
now = time.time()
if now - _MAHM['t'] < 1.0: return _MAHM['v']
_MAHM['t'] = now
@@ -135,7 +136,7 @@ def mahm():
hd = ctypes.string_at(p, 32)
num, esize, hsize = (int.from_bytes(hd[i * 4:i * 4 + 4], 'little') for i in (3, 4, 2))
upd = int.from_bytes(hd[20:24], 'little')
- if 0 < num < 5000 and 32 < esize < 65536 and now - upd <= 10:
+ if 0 < num < 5000 and 32 < esize < 65536 and now - upd <= 30:
buf = ctypes.string_at(p, hsize + num * esize)
core = {}
for i in range(num):
@@ -221,7 +222,7 @@ def newer(pattern, t0):
def stage_progress(st, s):
- """当前阶段的进度：返回 {'frac': 0..1 或 None（无法计量）, 'label': 文字}。"""
+ """当前阶段的进度：返回 {'frac': 0..1 或 None（无法计量）, 'label': 文字, 'eta_s': 按实测速率外推的剩余秒数或 None}。"""
t0, info = s['start'], s.get('info') or {}
if st == 'gen':
# 分子算全部（含之前轮次已完成的分片），分母是累计目标：补量时显示 50,008/79,688 而不是 0/79,688。
@@ -230,37 +231,72 @@ def stage_progress(st, s):
n = sum(LC.count(f) for f in allf)
tgt = info.get('target_tasks') or 0
new = sum(LC.count(f) for f in newer(f'{DATA}/shard_*.jsonl', t0) + newer(f'{DATA}/shard_*.jsonl.part', t0))
- lab = f'已生成 {n:,} / {tgt:,} 条任务' + (f'（本轮补 {new:,} 条）' if 0 < new < n else '')
- return {'frac': min(1, n / tgt) if tgt else None, 'label': lab}
+ # 剩余时间按"本轮已写条数 ÷ 本轮已用时"外推（条数是实时数出来的，速率就是实测的），不再用静态估时
+ el = max(1.0, time.time() - t0)
+ rate = new / el if new > 0 else 0
+ eta = ((tgt - n) / rate if rate > 0 else None) if tgt > n else (0.0 if rate > 0 else None)
+ lab = f'已生成 {n:,} / {tgt:,} 条任务' + (f'（本轮补 {new:,} 条 · {rate * 60:.0f} 条/分）' if 0 < new < n else '')
+ return {'frac': min(1, n / tgt) if tgt else None, 'label': lab, 'eta_s': eta}
+ if st == 'gate1':
+ # 一致性门禁每 10 个任务往 stderr 打一行 "k/300"（consistency.cjs:41）。进度只读控制台日志最后 64 KB，
+ # 不跑任何计算、不占 CPU —— 正在生成的节点不受影响。日志里可能还留着上一次门禁的进度行，
+ # 所以只数最后一个 "consistency.cjs" 命令行之后的部分。
+ txt = console(0).get('text') or ''
+ cut = txt.rfind('consistency.cjs')
+ m = re.findall(r'^(\d+)/(\d+)\s*$', txt[cut:] if cut >= 0 else '', re.M)
+ n, N = (int(m[-1][0]), int(m[-1][1])) if m else (0, 300)
+ el = time.time() - t0
+ rate = n / el if n > 0 and el > 2 else 0
+ return {'frac': n / N if N else None,
+ 'label': f'门禁 {n} / {N} 个任务' + (f' · {rate * 60:.0f} 个/分' if rate else '') if n else '准备中（日志里还没有进度行）',
+ 'eta_s': (N - n) / rate if rate > 0 else None}
if st == 'prep':
nsh = len(glob.glob(os.path.join(ROOT, DATA, 'shard_*.jsonl')))
done = len(newer(f'{DATA}/samples/*.X.npy', t0))
- if os.path.exists(os.path.join(ROOT, DATA, 'r2.json')) and newer(f'{DATA}/r2.json', t0): return {'frac': 1, 'label': '打包完成，R2 检查完成'}
- return {'frac': min(0.95, done / nsh) if nsh else None, 'label': f'已打包 {done} / {nsh} 个分片' + ('，R2 检查中' if nsh and done >= nsh else '')}
+ if os.path.exists(os.path.join(ROOT, DATA, 'r2.json')) and newer(f'{DATA}/r2.json', t0): return {'frac': 1, 'label': '打包完成，R2 检查完成', 'eta_s': 0}
+ el = time.time() - t0
+ eta = (nsh - done) / (done / el) if done > 0 and el > 1 and nsh > done else None
+ return {'frac': min(0.95, done / nsh) if nsh else None, 'label': f'已打包 {done} / {nsh} 个分片' + ('，R2 检查中' if nsh and done >= nsh else ''), 'eta_s': eta}
if st == 'train':
tag, bs = info.get('tag', 'base'), info.get('bench_steps', 2000)
b, m = last_rec(f'bench_{tag}') if newer(f'{RUNS}/bench_{tag}/log.jsonl', t0) else None, last_rec(tag) if newer(f'{RUNS}/{tag}/log.jsonl', t0) else None
if m and m.get('total'):
- return {'frac': (bs + m['step']) / (bs + m['total']), 'label': f'正式训练 {m["step"]:,} / {m["total"]:,} 步', 'run': tag}
+ return {'frac': (bs + m['step']) / (bs + m['total']), 'label': f'正式训练 {m["step"]:,} / {m["total"]:,} 步', 'run': tag,
+ 'eta_s': m['eta_h'] * 3600 if m.get('eta_h') else None} # 训练脚本自己按实测样本/s 算的
if b: return {'frac': b['step'] / (bs + 60000), 'label': f'测速 {b["step"]:,} / {bs:,} 步', 'run': f'bench_{tag}'}
return {'frac': 0, 'label': '加载数据中'}
if st == 'apg':
r = last_rec('apg') if newer(f'{RUNS}/apg/log.jsonl', t0) else None
- return {'frac': r['it'] / r['total'] if r and r.get('total') else 0, 'label': f'APG {r["it"]:,} / {r["total"]:,} 步' if r else '准备中', 'run': 'apg'}
+ eta = None
+ if r and r.get('total'):
+ it, mn = r.get('it') or 0, r.get('min') or 0
+ eta = (r['total'] - it) / (it / mn) * 60 if it > 0 and mn > 0.1 else None # 步/分实测外推
+ return {'frac': r['it'] / r['total'] if r and r.get('total') else 0, 'label': f'APG {r["it"]:,} / {r["total"]:,} 步' if r else '准备中', 'run': 'apg', 'eta_s': eta}
if st == 'mine':
R = info.get('rounds', 6)
k = len(newer(f'{RUNS}/mine/r*/gate.json', t0))
cur = sorted(newer(f'{RUNS}/apg_r*/log.jsonl', t0), key=os.path.getmtime)
- frac_in, lab = 0, f'第 {k + 1} 轮：挖错出题'
- if cur and len(cur) > k:
- name = os.path.basename(os.path.dirname(cur[-1])); r = last_rec(name)
- if r and r.get('total'): frac_in = 0.3 + 0.6 * r['it'] / r['total']; lab = f'第 {k + 1} 轮：APG {r["it"]:,} / {r["total"]:,} 步'
- return {'frac': min(1, (k + frac_in) / R), 'label': f'{lab}（共最多 {R} 轮，提升不足会提前停）', 'run': name}
+ lab = f'第 {k + 1} 轮：挖错出题'
+ if cur:
+ name = os.path.basename(os.path.dirname(cur[-1])) # ml/runs/apg_rN
+ rn = re.sub(r'\D', '', name) or '1'
+ gate = os.path.join(ROOT, RUNS, 'mine', f'r{rn}', 'gate.json')
+ r = last_rec(name)
+ if r and r.get('total') and not os.path.exists(gate): # 这一轮还在跑（还没出 gate）
+ frac_in = 0.3 + 0.6 * r['it'] / r['total']
+ lab = f'第 {k + 1} 轮：APG {r["it"]:,} / {r["total"]:,} 步'
+ it, mn = r.get('it') or 0, r.get('min') or 0
+ eta = (r['total'] - it) / (it / mn) * 60 if it > 0 and mn > 0.1 else None
+ if eta is not None and k >= 1: # 剩余轮次按已完成轮次的平均时长外推
+ eta += max(0, R - k - 1) * (time.time() - t0) / k
+ return {'frac': min(1, (k + frac_in) / R), 'label': f'{lab}（共最多 {R} 轮，提升不足会提前停）', 'run': name, 'eta_s': eta}
return {'frac': min(1, k / R), 'label': f'{lab}（共最多 {R} 轮，提升不足会提前停）'}
if st == 'm5':
tag = info.get('tag', 'base')
n = len(newer(f'{RUNS}/m5_{tag}_policy.json', t0)) + len(newer(f'{RUNS}/m5_{tag}_safety.json', t0))
- return {'frac': n / 2, 'label': ['纯策略评估中', '加安全层评估中', '汇总'][min(n, 2)]}
+ el = time.time() - t0
+ eta = (2 - n) * (el / n) if n == 1 else (0.0 if n >= 2 else None) # 两次评估（纯策略 / 加安全层）按已完成的平均耗时外推
+ return {'frac': n / 2, 'label': ['纯策略评估中', '加安全层评估中', '汇总'][min(n, 2)], 'eta_s': eta}
return {'frac': None, 'label': '进行中'}