#!/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()