#!/usr/bin/env python3
"""Campaign 10 — Phase B soak runner. Runs on DooPlex (outside the VM it abuses).

Shape (brief B1): a set of atomic operations composed into cycles in varying order,
with EVERY invariant checked EVERY cycle — drift at iteration 38 is the point.

Every atom and every invariant here was proven BY HAND before automation
(see the audit §5); nothing in this file asserts a behaviour that was not first
observed live.

Writes: journal.tsv (one row per invariant check), runner.log, status.txt (heartbeat).
Halts cleanly on: disk floor breach, VM gone, I11 breach (secret leak).
"""
import json, os, random, subprocess, sys, time, traceback
import c10lib as L

EV = "/mnt/5_hdd/felhom.eu/git/felhom.eu/documentation/tests/campaign10-evidence-2026-07-31"
STATE = os.path.join(EV, "state")
os.makedirs(STATE, exist_ok=True)
JOURNAL = os.path.join(STATE, "journal.tsv")
RUNLOG = os.path.join(STATE, "runner.log")
STATUS = os.path.join(STATE, "status.txt")

APPS = ["rallly", "homebox", "grafana", "papra"]      # docker-volume apps
DRIVE_APP = "calibre-web"                              # binds /mnt/felhom-drives/adatok
TARGET_DISK, TARGET_VOL = "scsi2", "c10-scratch:311/vm-311-disk-3.qcow2"   # mentes = backup target
NONTGT_DISK, NONTGT_VOL = "scsi1", "c10-scratch:311/vm-311-disk-2.qcow2"   # adatok
FLOOR_GB = 120          # halt if /mnt/nvme-1tb free drops below this
MAX_CYCLES = int(os.environ.get("C10_CYCLES", "40"))
DEADLINE = time.time() + float(os.environ.get("C10_HOURS", "6")) * 3600

CYCLE = 0
VIOL = []


def log(msg):
    line = "%s  %s" % (time.strftime("%H:%M:%S"), msg)
    print(line, flush=True)
    with open(RUNLOG, "a") as fh:
        fh.write(line + "\n")


def rec(inv, ok, detail=""):
    """One journal row per invariant check. Violations do NOT stop the run (brief B2)."""
    with open(JOURNAL, "a") as fh:
        fh.write("%d\t%s\t%s\t%s\t%s\n" % (CYCLE, time.strftime("%FT%TZ", time.gmtime()),
                                           inv, "PASS" if ok else "VIOLATION", detail[:400]))
    if not ok:
        VIOL.append((CYCLE, inv, detail[:300]))
        log("  !! %s VIOLATION: %s" % (inv, detail[:220]))


def heartbeat(phase):
    with open(STATUS, "w") as fh:
        fh.write("cycle=%d\nphase=%s\nutc=%s\nviolations=%d\n"
                 % (CYCLE, phase, time.strftime("%FT%TZ", time.gmtime()), len(VIOL)))


# ------------------------------------------------------------------ helpers

def hub_events(since="3m"):
    """Hub-side event stream: type + severity, the observable I1/I2 assert on."""
    p = subprocess.run(["sudo", "kubectl", "-n", "felhom-system", "logs", "deploy/hub",
                        "--since=" + since], capture_output=True, text=True, timeout=120)
    out = []
    for line in p.stdout.splitlines():
        if "Event from c10-soak:" in line:
            seg = line.split("Event from c10-soak:", 1)[1].strip()
            typ = seg.split()[0]
            sev = seg.split("(", 1)[1].split(")", 1)[0] if "(" in seg else "?"
            out.append((typ, sev, seg))
    return out


def target_state():
    return (L.jparse(L.api("GET", "/api/storage/backup-target")) or {}).get("data", {}) or {}


def disks():
    return ((L.jparse(L.api("GET", "/api/disks")) or {}).get("data", {}) or {}).get("disks", [])


def wait_backup(timeout=900):
    t0 = time.time()
    while time.time() - t0 < timeout:
        d = ((L.jparse(L.api("GET", "/api/backup/status")) or {}).get("data") or {})
        if not d.get("running"):
            return d
        time.sleep(10)
    return {"timeout": True}


def wait_restore(timeout=900):
    t0 = time.time()
    while time.time() - t0 < timeout:
        d = ((L.jparse(L.api("GET", "/api/backup/restore-status")) or {}).get("data") or {})
        if d and not d.get("running"):
            return d
        time.sleep(10)
    return {"timeout": True}


def settle(seconds=130):
    """Agent tick + host-report + hub event propagation. Measured at ~25-60s; 130 gives margin."""
    time.sleep(seconds)


# ------------------------------------------------------------------ atoms

def atom_backup():
    L.api("POST", "/api/backup/run", {})
    d = wait_backup()
    ok = bool(d.get("db_dump", {}).get("success", True)) and not d.get("timeout")
    rec("BACKUP", ok, json.dumps(d)[:300])
    return ok


def atom_restore_verify():
    """I7 — the whole point: a restore must return the discriminator asked for."""
    app = "rallly"                                  # the DB app; its canary is a real DB row
    # PRECONDITION, not politeness: a canary lives inside an app container. Reading or writing
    # one before the app is back measures the harness, not the product. Run 2a produced a fake
    # "stale restore" this exact way, after a hard reset left rallly down.
    if not L.apps_ready([app]):
        rec("I7-SKIP", True, "apps not ready before seed — atom skipped, NOT counted as a violation")
        return
    genA = "C10-C%03d-A-%s" % (CYCLE, time.strftime("%H%M%S"))
    sa = L.seed(genA)
    if not sa.get(app):
        rec("I7-SKIP", True, "seed of %s did not land (app unhealthy) — atom skipped, not a violation" % genA)
        return
    if not atom_backup():
        return
    genB = "C10-C%03d-B-%s" % (CYCLE, time.strftime("%H%M%S"))
    sb = L.seed(genB)                               # must NOT survive the restore
    if not sb.get(app):
        rec("I7-SKIP", True, "seed of %s did not land — cannot distinguish stale from unwritten" % genB)
        return
    snaps = L.jparse(L.api("GET", "/api/backup/snapshots?stack=%s" % app)) or {}
    pts = snaps.get("data") or []
    if not pts:
        rec("I7", False, "no restore points for %s" % app)
        return
    sid = pts[0].get("short_id")
    t0 = time.time()
    L.guest('/root/c10api.sh POST /backup/restore "stack_name=%s&snapshot_id=%s" 2>/dev/null' % (app, sid))
    d = wait_restore()
    rto = time.time() - t0
    got = L.read_canaries().get(app)
    rec("I7", got == genA,
        "app=%s want=%s got=%s snap=%s restore_ok=%s rto=%.1fs"
        % (app, genA, got, sid, (d.get("last") or {}).get("ok"), rto))
    # RTO/RPO byproduct (brief B3): to app-serving-correct-data, not to "restore returned"
    with open(os.path.join(STATE, "rto.tsv"), "a") as fh:
        fh.write("%d\t%s\t%s\ttier1\t%.1f\t%s\n" % (CYCLE, time.strftime("%FT%TZ", time.gmtime()),
                                                    app, rto, got == genA))


def _detach(disk, vol, label, expect_type, expect_recover):
    """I1/I2 — the absent event must be the RIGHT one, and the pair must match."""
    before_boot = L.guest("stat -c %Y /proc/1")[1].strip()
    L.hp("qm set 311 --delete %s" % disk)
    settle()
    evs = hub_events("4m")
    absent = [e for e in evs if e[0] in ("backup_target_absent", "storage_disconnected")]
    got = absent[-1][0] if absent else None
    sev = absent[-1][1] if absent else None
    rec("I1" if expect_type == "backup_target_absent" else "I2",
        got == expect_type and sev == "error",
        "%s absent -> got=%s sev=%s (want %s/error)" % (label, got, sev, expect_type))

    # I3: an unusable namespace must not read as bound
    for d in disks():
        if d.get("guest_path", "") and label in str(d.get("guest_path", "")):
            rec("I3", d.get("bound_under_parent") is False,
                "%s bound_under_parent=%s while absent" % (label, d.get("bound_under_parent")))

    L.hp("qm set 311 --%s %s" % (disk, vol))
    settle()
    evs2 = hub_events("4m")
    back = [e for e in evs2 if e[0] in ("backup_target_restored", "storage_reconnected")]
    got2 = back[-1][0] if back else None
    sev2 = back[-1][1] if back else None
    rec("I1-pair" if expect_type == "backup_target_absent" else "I2-pair",
        got2 == expect_recover and sev2 == "info",
        "%s return -> got=%s sev=%s (want %s/info)" % (label, got2, sev2, expect_recover))

    # I4: the bind repairs WITHOUT the guest restarting
    after_boot = L.guest("stat -c %Y /proc/1")[1].strip()
    rec("I4", before_boot == after_boot and before_boot != "",
        "guest init boot before=%s after=%s (must be equal)" % (before_boot, after_boot))


def atom_detach_target():
    _detach(TARGET_DISK, TARGET_VOL, "mentes", "backup_target_absent", "backup_target_restored")


def atom_detach_nontarget():
    _detach(NONTGT_DISK, NONTGT_VOL, "adatok", "storage_disconnected", "storage_reconnected")


def atom_reboot_guest():
    L.box("pct reboot 9201", timeout=300)
    for _ in range(40):
        time.sleep(15)
        if "felhom-controller" in L.containers():
            break
    rec("REBOOT", "felhom-controller" in L.containers(), "controller back after guest reboot")


def atom_kill_controller():
    L.guest("docker restart felhom-controller >/dev/null 2>&1 &")
    time.sleep(45)
    rec("KILL-CTRL", "felhom-controller" in L.containers(), "controller recovered after kill")


def atom_redeploy_app():
    app = random.choice(["homebox", "grafana"])
    L.api("POST", "/api/stacks/%s/remove" % app, {})
    time.sleep(20)
    sec = L.SECRETS.get(app, {}).get("values", {})
    if sec:
        L.api("POST", "/api/stacks/%s/deploy" % app, {"values": sec})
    for _ in range(20):
        time.sleep(15)
        if app in L.containers():
            break
    rec("REDEPLOY", app in L.containers(), "%s redeployed" % app)


RAW_ADATOK = "/mnt/adatok"
ADATOK_UUID = "13a43656-b393-4879-8d05-37de560def5f"


def _adatok_row():
    for d in disks():
        if str(d.get("guest_path", "")).endswith("adatok"):
            return d
    return {}


def atom_abort_fs_inplace():
    """R-117 Q7 / I4 second clause — the filesystem dies WITHOUT the device disappearing.
    Before agent v0.117.0 this state reported healthy and restarted apps onto it with
    nothing emitted on any channel. Assert it SURFACES and does not silently retry."""
    L.box("mount -o remount,abort %s" % RAW_ADATOK)
    opts = L.box("grep ' %s ' /proc/mounts" % RAW_ADATOK)[1].strip()
    if not ("abort" in opts or "emergency_ro" in opts):
        rec("I4-abort", False, "could not abort the fs in place; /proc/mounts=%s" % opts[:120])
        return
    settle(150)
    row = _adatok_row()
    dev_present = "sd" in L.box("lsblk -dno NAME | tr '\\n' ' '")[1]
    rec("I4-abort", row.get("bound_under_parent") is False and dev_present,
        "aborted-in-place: bound=%s (want False), device still present=%s, opts=%s"
        % (row.get("bound_under_parent"), dev_present, opts.split("ext4")[-1][:60]))
    # the gate must STOP the app whose data is on the dead namespace, not restart it onto it
    rec("I3-abort", DRIVE_APP not in L.containers(),
        "%s must be stopped while its namespace is aborted; running=%s"
        % (DRIVE_APP, DRIVE_APP in L.containers()))
    # recovery needs a full device close — a remount lands on the same aborted superblock
    L.hp("qm set 311 --delete %s" % NONTGT_DISK)
    time.sleep(8)
    L.hp("qm set 311 --%s %s" % (NONTGT_DISK, NONTGT_VOL))
    settle(150)
    row = _adatok_row()
    rec("I4-abort-recover", row.get("bound_under_parent") is True and DRIVE_APP in L.containers(),
        "after re-seat: bound=%s app_up=%s" % (row.get("bound_under_parent"), DRIVE_APP in L.containers()))


def atom_kill_agent_mid_backup():
    """Kill the AGENT while a backup runs, then prove the next backup still succeeds."""
    L.api("POST", "/api/backup/run", {})
    time.sleep(6)
    L.box("systemctl kill -s KILL felhom-agent || true")
    time.sleep(20)
    L.box("systemctl start felhom-agent || true")
    wait_backup()
    time.sleep(20)
    up = "active" in L.box("systemctl is-active felhom-agent")[1]
    ok2 = atom_backup()
    rec("KILL-AGENT", up and ok2, "agent active after kill=%s; subsequent backup ok=%s" % (up, ok2))


def atom_hard_reset_mid_write():
    """Hard-reset the VM mid-backup. Assert it returns AND the canaries survive intact."""
    before = L.read_canaries()
    L.api("POST", "/api/backup/run", {})
    time.sleep(8)
    L.hp("qm reset 311")
    ok = False
    for _ in range(40):
        time.sleep(20)
        if "felhom-controller" in L.containers():
            ok = True
            break
    # the canaries live in the APP containers — wait for those, or this measures startup latency
    ready = L.apps_ready() if ok else False
    after = L.read_canaries() if ready else {}
    intact = all(after.get(a) == before.get(a) for a in before) if ready else None
    detail = "VM returned=%s apps_ready=%s canaries_intact=%s" % (ok, ready, intact)
    if not ready:
        rec("HARD-RESET", False, detail + " — apps did not return within 10 min (that IS the failure)")
    else:
        rec("HARD-RESET", bool(intact), detail + " (a hard reset mid-write must not corrupt them)")


def atom_reboot_vm():
    L.hp("qm reboot 311")
    ok = False
    for _ in range(40):
        time.sleep(20)
        if "felhom-controller" in L.containers():
            ok = True
            break
    rec("REBOOT-VM", ok, "controller back after full VM reboot")


def atom_concurrent_backup_restore():
    """The controller must REFUSE the second op, not run both over the same data."""
    L.api("POST", "/api/backup/run", {})
    time.sleep(3)
    out = L.guest('/root/c10api.sh POST /backup/restore "stack_name=rallly&snapshot_id=helyi" 2>/dev/null')[1]
    st = ((L.jparse(L.api("GET", "/api/backup/restore-status")) or {}).get("data") or {})
    wait_backup(); wait_restore()
    # either the restore was refused, or it queued behind the backup — never both mutating at once
    rec("CONCURRENCY", True, "backup+restore raced; restore-status op=%s running=%s (recorded, not asserted)"
        % (st.get("op"), st.get("running")))


def atom_concurrent_backup_detach():
    """Pull the backup TARGET out from under a running backup."""
    L.api("POST", "/api/backup/run", {})
    time.sleep(5)
    L.hp("qm set 311 --delete %s" % TARGET_DISK)
    settle()
    evs = [e for e in hub_events("4m") if e[0] in ("backup_target_absent", "storage_disconnected")]
    got = evs[-1][0] if evs else None
    rec("I1-under-load", got == "backup_target_absent",
        "target pulled DURING a backup -> got=%s (want backup_target_absent)" % got)
    L.hp("qm set 311 --%s %s" % (TARGET_DISK, TARGET_VOL))
    settle()
    wait_backup()
    ts = target_state()
    rec("I1-under-load-recover", ts.get("degraded") is False,
        "after return degraded=%s target=%s" % (ts.get("degraded"), ts.get("target")))


def atom_fill_drive():
    """Fill the backup target to near-full and back up. A refusal is a PASS; a silent
    success that writes nothing is the failure this looks for."""
    avail = L.box("df --output=avail -BM /mnt/mentes | tail -1")[1].strip().rstrip("M")
    try:
        keep = max(int(avail) - 300, 50)
    except Exception:
        keep = 50
    L.box("fallocate -l %dM /mnt/mentes/c10-filler 2>&1 || dd if=/dev/zero of=/mnt/mentes/c10-filler bs=1M count=%d 2>/dev/null" % (keep, keep))
    free_after = L.box("df -h /mnt/mentes | tail -1")[1].strip()
    d = ((L.jparse(L.api("GET", "/api/backup/status")) or {}).get("data") or {})
    L.api("POST", "/api/backup/run", {})
    res = wait_backup()
    L.box("rm -f /mnt/mentes/c10-filler")
    ok = res.get("db_dump", {}).get("success") is not None
    rec("FILL-DRIVE", ok, "near-full (%s) backup outcome recorded: %s" % (free_after[-24:], json.dumps(res)[:200]))


# ------------------------------------------------------------------ per-cycle invariants

def check_healthy_baseline():
    """I5/I6 — the over-correction check: healthy must render nothing."""
    ts = target_state()
    degraded = ts.get("degraded")
    evs = [e for e in hub_events("3m") if e[0] in ("backup_target_absent", "storage_disconnected")]
    rec("I5", degraded is False and not evs,
        "degraded=%s absent_events=%d target=%s" % (degraded, len(evs), ts.get("target")))
    if degraded is False:
        rec("I6", "message" not in ts or not ts.get("message"),
            "healthy state must carry no degraded message; got=%s" % str(ts.get("message"))[:120])


UNIT = "/mnt/sys_drive/felhom-data/backups/primary/%s/compose/app.yaml"

def check_secret_class_travel():
    """I10 — `type: secret` travels to the local unit; `type: password` NEVER does.
    Asserted in BOTH directions: the withheld class staying withheld is the half
    that silently rots. Key NAMES only; no value is ever read."""
    for app, meta in L.SECRETS.items():
        classes = meta.get("secret_classes", {})
        if not classes:
            continue
        rc, keys, _ = L.guest("grep -oE '^ *[A-Z_0-9]+:' " + (UNIT % app) +
                              " 2>/dev/null | tr -d ' :' | tr '\\n' ' '")
        present = set(keys.split())
        if not present:
            rec("I10", False, "%s: recovery unit unreadable/absent" % app)
            continue
        for ev, c in classes.items():
            if c["type"] == "secret":
                rec("I10", ev in present, "%s/%s type=secret must travel; in_unit=%s"
                    % (app, ev, ev in present))
            elif c["type"] == "password":
                rec("I10", ev not in present, "%s/%s type=password must NEVER travel; in_unit=%s"
                    % (app, ev, ev in present))


def check_secrets_absent():
    """I11 — no secret value in any log, event or hub payload. Breach STOPS the run."""
    vals = [v for a in L.SECRETS.values() for k, v in a.get("values", {}).items()
            if k not in ("DOMAIN", "SUBDOMAIN") and v and len(v) > 8]
    hay = L.guest("docker logs felhom-controller --since 20m 2>&1 | tail -c 200000")[1]
    p = subprocess.run(["sudo", "kubectl", "-n", "felhom-system", "logs", "deploy/hub", "--since=20m"],
                       capture_output=True, text=True, timeout=120)
    hay += p.stdout
    leaked = [v[:6] + "..." for v in vals if v in hay]
    rec("I11", not leaked, "leaked=%s" % leaked)
    return not leaked


def watchdog():
    rc, out, _ = L.hp("df --output=avail -BG /mnt/nvme-1tb | tail -1")
    try:
        free = int(out.strip().rstrip("G"))
    except Exception:
        free = 9999
    rc2, vm, _ = L.hp("qm status 311")
    alive = "running" in vm
    if free < FLOOR_GB:
        log("WATCHDOG: free %dG < floor %dG — halting cleanly" % (free, FLOOR_GB)); return False
    if not alive:
        log("WATCHDOG: VM 311 not running (%s) — halting cleanly" % vm.strip()); return False
    return True


# ------------------------------------------------------------------ main

# LIGHT atoms run every cycle. HEAVY ones are slow (VM reboots, fills, aborts), so a
# rotating slice joins each cycle and is SHUFFLED IN with the rest — the v1 runner appended
# reboot after the shuffle, so it never interleaved with a detach. That is fixed here.
ATOMS = [atom_backup, atom_restore_verify, atom_detach_target,
         atom_detach_nontarget, atom_kill_controller, atom_redeploy_app]

HEAVY = [atom_abort_fs_inplace, atom_kill_agent_mid_backup, atom_hard_reset_mid_write,
         atom_reboot_vm, atom_reboot_guest, atom_concurrent_backup_restore,
         atom_concurrent_backup_detach, atom_fill_drive]

def main():
    global CYCLE
    log("=== Campaign 10 Phase B start — max %d cycles, deadline %s ==="
        % (MAX_CYCLES, time.strftime("%H:%M", time.localtime(DEADLINE))))
    if not os.path.exists(JOURNAL):
        with open(JOURNAL, "w") as fh:
            fh.write("cycle\tutc\tinvariant\tverdict\tdetail\n")
    rng = random.Random(20260801)
    while CYCLE < MAX_CYCLES and time.time() < DEADLINE:
        CYCLE += 1
        heartbeat("start")
        if not watchdog():
            break
        order = ATOMS[:]
        # ONE heavy atom per cycle: every heavy atom is seen every 8 cycles, and the cycle
        # stays short enough to reach the tail the brief cares about (drift at ~38).
        order += [HEAVY[CYCLE % len(HEAVY)]]
        rng.shuffle(order)                       # permutation INCLUDING the heavy atoms
        log("cycle %d: %s" % (CYCLE, " -> ".join(a.__name__.replace("atom_", "") for a in order)))
        for a in order:
            heartbeat(a.__name__)
            try:
                a()
            except Exception as e:
                rec(a.__name__, False, "EXCEPTION %s" % e)
                log("  atom %s raised: %s" % (a.__name__, traceback.format_exc().splitlines()[-1]))
        heartbeat("invariants")
        try:
            check_healthy_baseline()
            check_secret_class_travel()
            if not check_secrets_absent():
                log("I11 BREACH — stopping the run immediately per brief B2"); break
        except Exception as e:
            log("invariant pass raised: %s" % e)
        log("cycle %d done — violations so far: %d" % (CYCLE, len(VIOL)))
    heartbeat("finished")
    log("=== finished after %d cycles, %d violations ===" % (CYCLE, len(VIOL)))
    for c, i, d in VIOL:
        log("  cycle %d %s: %s" % (c, i, d))


if __name__ == "__main__":
    main()