reap_stuck_jobs.py
5.39 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
#!/data/data/com.termux/files/usr/bin/python3
"""Stuck job reaper — scan hermes_jobs for orphaned 'running' jobs and fix them.
Also GC orphaned absorb step checkpoint files (learn_absorb_*.stepN.done/failed)."""
import sqlite3, os, sys, re, glob, time
from datetime import datetime, timedelta
DB = "/data/data/com.termux/files/home/wechat-viewer/circle.db"
LOG_DIR = "/data/data/com.termux/files/home/tmp/hermes_jobs"
TMP_DIR = "/data/data/com.termux/files/home/tmp"
# ── Step checkpoint orphan GC ──
# Inspired by 三天实现 Kafka 原型：Orphan GC — 孤儿对象先判定再用，不直接删
# .stepN.done/.stepN.failed 文件在 _on_absorb_done 的 finally 块中清理，
# 但如果 _on_absorb_done 本身崩溃，这些文件会成为孤儿。
# 策略：检查文件 mtime > 30min → 关联的 absorb task 已结束或异常 → 安全删除
def _gc_step_checkpoints():
gc_count = 0
cutoff = time.time() - 1800 # 30 minutes
patterns = [
os.path.join(TMP_DIR, "learn_absorb_*.step[1-9].done"),
os.path.join(TMP_DIR, "learn_absorb_*.step[1-9].failed"),
]
for pat in patterns:
for f in glob.glob(pat):
try:
if os.path.getmtime(f) < cutoff:
os.remove(f)
gc_count += 1
except OSError:
pass
if gc_count:
print(f"[GC] cleaned {gc_count} orphan step checkpoint files")
# ── Applied-without-audit orphan GC ──
# Inspired by 三天实现 Kafka 原型：ACK 在所有副作用落盘后才算成功
# 检测 learn_queue 中 status='applied' 但无 apply_audit 结果的项
# 这是 ACK_PREMATURE 故障模式的恢复路径
def _gc_applied_without_audit():
db = sqlite3.connect(DB)
rows = db.execute('''
SELECT lq.id, lq.title, lq.error_msg
FROM learn_queue lq
WHERE lq.status='applied'
AND NOT EXISTS (
SELECT 1 FROM learning_results lr
WHERE lr.learn_queue_id = lq.id AND lr.result_type='apply_audit'
)
ORDER BY lq.id DESC
''').fetchall()
db.close()
if not rows:
return
with_step6 = []
without_step6 = []
for r in rows:
msg = r[2] or ''
if 'step6_apply' in msg:
with_step6.append(r)
else:
without_step6.append(r)
trace_parts = []
if with_step6:
trace_parts.append(
f'step6_callback_fired_but_no_audit_result={len(with_step6)}')
if without_step6:
trace_parts.append(
f'step6_callback_never_fired={len(without_step6)}')
gate_trace = (
f"[node=reaper,attempt=1,gate=ACK_PREMATURE,"
f"action={'REVISE' if without_step6 else 'CONTINUE'}] "
f"applied-without-apply_audit: total={len(rows)}, "
f"{', '.join(trace_parts)}"
)
print(gate_trace)
if without_step6:
print(" Items where Step6 callback never fired (建议手动 /relearn):")
for r in without_step6:
print(f" #{r[0]} {r[1][:60] if r[1] else '(no title)'}")
def main():
_gc_step_checkpoints() # 先 GC，再处理 stuck jobs
_gc_applied_without_audit() # ACK_PREMATURE 孤儿检测
db = sqlite3.connect(DB)
# Find stuck jobs: status='running' and started > 5 min ago
cutoff = (datetime.now() - timedelta(minutes=5)).strftime("%Y-%m-%d %H:%M:%S")
stuck = db.execute(
"SELECT task_id, job_type, job_name, status, started_at, log_file "
"FROM hermes_jobs WHERE status='running' AND started_at < ?",
(cutoff,)
).fetchall()
if not stuck:
print("No stuck jobs found.")
db.close()
return
print(f"Found {len(stuck)} stuck jobs:")
completed_pattern = re.compile(r"Session:\s+\S+")
fixed_completed = 0
fixed_failed = 0
for task_id, job_type, job_name, status, started_at, log_file in stuck:
log_path = log_file if os.path.isabs(log_file) else os.path.join(LOG_DIR, os.path.basename(log_file))
# Check if log file exists and has completion evidence
new_status = "failed" # default
if os.path.exists(log_path):
try:
with open(log_path) as f:
content = f.read()
# Check for successful completion markers
if "Exit code: 0" in content or "Duration:" in content:
new_status = "completed"
elif "TIMEOUT" in content:
new_status = "failed"
elif completed_pattern.search(content):
new_status = "completed"
except Exception:
new_status = "failed"
else:
new_status = "failed"
db.execute("UPDATE hermes_jobs SET status=? WHERE task_id=?", (new_status, task_id))
print(f" {task_id[:8]}... {job_type}:{job_name[:30]} → {new_status}")
if new_status == "completed":
fixed_completed += 1
else:
fixed_failed += 1
db.commit()
db.close()
total_fixed = fixed_completed + fixed_failed
gate_trace = (
f"[node=reaper,attempt=1,gate=STUCK_JOBS,"
f"action={'CONTINUE' if total_fixed > 0 else 'PASS'}] "
f"fixed {fixed_completed} completed + {fixed_failed} failed out of {len(stuck)} stuck"
)
print(f"\nDone: {fixed_completed} completed, {fixed_failed} failed.")
print(gate_trace)
if __name__ == "__main__":
main()