#!/usr/bin/env python3
"""The fleet night loop (phase 2): every 60 s reads each box's swap log and prover state; launches the prover on a box
whose replay reached the tip (once); re-runs the exec restart on a box whose executed tip fell back to 0 for more than
two minutes (the deep-reorg class of 15:43Z, 6 October 2026); every 15 minutes writes a night row (provers up, claims,
submitted, paid segments, shards paid, the chain's segment window from the hub) to ~/Desktop/fleet/night.json and the
page, and appends the hourly row to the plan's fleet night section when the hour turns. Log: ~/Desktop/fleet/night.log."""
import sys, os, time, json, datetime, subprocess
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))); import fleet
from concurrent.futures import ThreadPoolExecutor
SHA = "d6350586fe837b1f71696546628ec071b6accce2531aef776cf2b1e5487a8cdc"; DIG = "b18ed271f75dd46406d230f4156c37472127415a4c32c558bac662f6f840e61c"
LOG = os.path.join(fleet.ROOT, "night.log"); NIGHT = os.path.join(fleet.ROOT, "night.json")
def now(): return datetime.datetime.now(datetime.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
def log(s): open(LOG, "a").write(f"{now()} {s}\n"); print(now(), s, flush=True)
PROBE = r"""grep -E '^RESULT (swap_done|swap_failed)' /root/fleet/out/node-swap.log 2>/dev/null | grep -v step=binary | tail -1 | cut -c1-60
curl -s -m 6 -X POST -H 'Content-Type: application/json' --data '{"jsonrpc":"2.0","id":1,"method":"igneum_getExecStatus","params":[]}' http://127.0.0.1:26790/ | python3 -c 'import sys,json; r=sys.stdin.read(); d=json.loads(r).get("result",{}) if r.strip() else {}; print("exec", int(d.get("executedTip","0x0"),16), "blocked", "yes" if d.get("blocked") else "no", "from", str(d.get("startedFrom",""))[:30])' 2>/dev/null
curl -s -m 6 -X POST -H 'Content-Type: application/json' --data '{"jsonrpc":"2.0","id":1,"method":"igneum_getProvingStatus","params":[]}' http://127.0.0.1:26790/ | python3 -c 'import sys,json; r=sys.stdin.read(); d=json.loads(r).get("result",{}) if r.strip() else {}; v=d.get("v1",{}); w=v.get("segmentsInWindow",{}); print("tipDaa", int(d.get("tipDaa","0x0"),16), "fresh", v.get("freshRuleActive"), "paidShards", d.get("paidShards"), "pending", w.get("pending"), "proven", w.get("proven"), "unproven", w.get("unproven"), "paidSeg", v.get("paidSegments"))' 2>/dev/null
/opt/igneum/pkg/bin/igneum-miner watch 1 grpc://127.0.0.1:26610 2>/dev/null | grep -o 'daa=[0-9]*' | tail -1
pgrep -c -f '[p]ython3 -u /root/fleet/in/box-prover.py'
for f in /root/fleet/out/prover-state.json /root/fleet/card*/out/prover-state.json; do [ -f $f ] && python3 -c 'import json,sys; s=json.load(open(sys.argv[1])); print("ps", s.get("claimed",0), s.get("submitted",0), s.get("paid",0), s.get("shards_accepted",0), s.get("shards_refused",0), s.get("segment_refused",0), s.get("held",0), s.get("miner_mhs",0))' $f; done; true"""
def probe(b):
    r, out, err = fleet.ssh(b, PROBE, timeout=45)
    if r != 0 or not out.strip(): return None
    l = out.strip().split("\n"); d = {"swap": "", "exec": None, "blocked": "", "from": "", "tipDaa": None, "fresh": "", "paidShards": None, "pending": None, "proven": None, "unproven": None, "paidSeg": None, "daa": None, "provers": 0, "ps": []}
    for x in l:
        if x.startswith("RESULT swap"): d["swap"] = x
        elif x.startswith("exec "): p = x.split(); d["exec"] = int(p[1]); d["blocked"] = p[3]; d["from"] = " ".join(p[5:])
        elif x.startswith("tipDaa"): p = x.split(); d.update(tipDaa=int(p[1]), fresh=p[3], paidShards=p[5], pending=p[7], proven=p[9], unproven=p[11], paidSeg=p[13])
        elif x.startswith("daa="): d["daa"] = int(x[4:])
        elif x.isdigit(): d["provers"] = int(x)
        elif x.startswith("ps "): d["ps"].append([float(v) for v in x.split()[1:]])
    return d
zero_since = {}; launched = set(); last_row = 0; last_hour = None
hub = next(b for b in fleet.load().values() if b.get("hub") and b.get("state") != "destroyed" and b.get("hub_peer"))
while True:
    reg = fleet.load()
    live = [(iid, b) for iid, b in reg.items() if b.get("state") != "destroyed" and b.get("ssh_host") and (b.get("phase") in ("1", "2") or b.get("hub"))]
    with ThreadPoolExecutor(14) as ex: res = dict(zip([i for i, _ in live], ex.map(lambda x: probe(x[1]), live)))
    tot = {"provers": 0, "claimed": 0, "submitted": 0, "paid": 0, "shards": 0, "refused": 0, "segref": 0, "held": 0, "exec_ok": 0, "boxes": 0}
    hubd = None
    for iid, b in live:
        d = res.get(iid)
        if not d: continue
        tot["boxes"] += 1
        if b.get("hub"): hubd = d
        at_tip = d["exec"] is not None and d["tipDaa"] and d["daa"] and d["tipDaa"] >= d["daa"] - 30 and d["exec"] > 0
        if at_tip: tot["exec_ok"] += 1; zero_since.pop(iid, None)
        elif d["exec"] == 0 and "swap_done" in d["swap"]:
            zero_since.setdefault(iid, time.time())
            if time.time() - zero_since[iid] > 150:
                fleet.ssh(b, f"cd /root/fleet && bash in/box-kill.sh >/dev/null 2>&1; mv out/node-swap.log out/node-swap-reset-$(date +%s).log 2>/dev/null; STEP=file NODE_SHA256={SHA} EXPECT_DIGEST={DIG} HUB_PEER={hub['hub_peer']} HUB_SSH={hub['ssh_host']} HUB_PORT={hub['ssh_port']} setsid nohup in/box-node-swap.sh </dev/null >/dev/null 2>&1 &", timeout=60)
                log(f"{b['label']}: exec at 0 for 150 s after a completed replay; exec restart re-run"); zero_since.pop(iid, None); launched.discard(iid)
        if at_tip and iid not in launched and d["provers"] == 0 and b.get("stage") != "mine-only":
            if (b.get("num_gpus") or 1) > 1: fleet.ssh(b, f"cd /root/fleet && LABEL={b['label']} WALLET={b['wallet']} setsid nohup in/box-rig-prover.sh </dev/null > out/rig-provers.log 2>&1 &", timeout=60)
            else: fleet.ssh(b, f"cd /root/fleet && mv out/prover.log out/prover-prev-$(date +%s).log 2>/dev/null; LABEL={b['label']} WALLET={b['wallet']} HUB_PEER={hub['hub_peer']} setsid nohup in/box-prover.sh </dev/null >/dev/null 2>&1 &", timeout=60)
            launched.add(iid); fleet.patch(iid, stage="prover", state="running", doing=f"phase 2 prover on the executed tip ({d['exec']})"); log(f"{b['label']}: replay at the tip (exec {d['exec']}), prover started")
        tot["provers"] += d["provers"]
        for p in d["ps"]: tot["claimed"] += p[0]; tot["submitted"] += p[1]; tot["paid"] += p[2]; tot["shards"] += p[3]; tot["refused"] += p[4]; tot["segref"] += p[5]; tot["held"] += p[6]
        fleet.patch(iid, last_line=f"exec {d['exec']} tip {d['tipDaa']} provers {d['provers']} " + (" ".join(f"{int(p[0])}/{int(p[1])}/{int(p[2])}" for p in d["ps"])))
    if time.time() - last_row > 900:
        row = {"hour": now()[:13], "at": now(), "provers": tot["provers"], "boxes_executing": tot["exec_ok"], "boxes": tot["boxes"], "claimed": tot["claimed"], "segments_submitted": tot["submitted"], "segments_done": tot["paid"], "shards_paid": tot["shards"], "shards_refused": tot["refused"], "segment_records_refused": tot["segref"], "held": tot["held"],
               "chain": ({"tipDaa": hubd["tipDaa"], "fresh": hubd["fresh"], "pending": hubd["pending"], "proven": hubd["proven"], "unproven": hubd["unproven"], "paidSeg": hubd["paidSeg"], "paidShards": hubd["paidShards"]} if hubd else {}),
               "coverage_pct": (round(100 * int(hubd["proven"]) / max(1, int(hubd["proven"]) + int(hubd["pending"]) + int(hubd["unproven"])), 1) if hubd and hubd["proven"] not in (None, "None") else None),
               "note": f"{tot['exec_ok']} of {tot['boxes']} boxes executing"}
        rows = json.load(open(NIGHT)) if os.path.exists(NIGHT) else []; rows.append(row); json.dump(rows, open(NIGHT, "w"), indent=1)
        log("row " + json.dumps(row)); last_row = time.time()
        subprocess.run([sys.executable, os.path.join(fleet.HERE, "page.py"), "publish"], capture_output=True)
    time.sleep(60)