#!/usr/bin/env python3
"""Run EtherGold stage/promote/rollback/tip-apply/ops as topbox (ethergold-http is often root)."""
from __future__ import annotations

import json
import os
import re
import subprocess
import time
from pathlib import Path
from typing import Any, Dict, List, Optional

EG_ROOT = Path(os.environ.get("ETHERGOLD_ROOT", "/home/topbox/EtherGold"))
EG_SYNC = EG_ROOT / "sync"
TOPBOX_USER = os.environ.get("ETHERGOLD_USER", os.environ.get("DYZINEFRAME_USER", "topbox"))
STAGE_TIMEOUT_SEC = int(os.environ.get("EG_DEPLOY_STAGE_TIMEOUT_SEC", "180"))
PROMOTE_TIMEOUT_SEC = int(os.environ.get("EG_DEPLOY_PROMOTE_TIMEOUT_SEC", "180"))
TIP_APPLY_TIMEOUT_SEC = int(os.environ.get("EG_DEPLOY_TIP_APPLY_TIMEOUT_SEC", "600"))
BRIDGE_TIMEOUT_SEC = int(os.environ.get("EG_DEPLOY_BRIDGE_TIMEOUT_SEC", "30"))
SYNC_TIMEOUT_SEC = int(os.environ.get("EG_DEPLOY_SYNC_TIMEOUT_SEC", "600"))

APPLY_JOB_NAME = "apply-job.json"


def _read_text(path: Path) -> str:
    try:
        return path.read_text(encoding="utf-8", errors="replace").strip()
    except OSError:
        return ""


def _stamp_tuple(stamp: str):
    parts = []
    for tok in (stamp or "").strip().split("."):
        if not tok.isdigit():
            return None
        parts.append(int(tok))
    if not parts:
        return None
    while len(parts) < 4:
        parts.append(0)
    return tuple(parts[:4])


def _stamp_is_older(left: str, right: str) -> bool:
    a, b = _stamp_tuple(left), _stamp_tuple(right)
    if a is None or b is None:
        return False
    return a < b


def _cmd_prefix() -> List[str]:
    try:
        if hasattr(os, "geteuid") and os.geteuid() == 0:
            return ["runuser", "-u", TOPBOX_USER, "--"]
    except OSError:
        pass
    return []


def _first_script(*names: str) -> str:
    for name in names:
        if (EG_SYNC / name).is_file():
            return name
    return names[0]


def _staging_sync() -> Path:
    home = Path("/home") / TOPBOX_USER
    return Path(os.environ.get("ETHERGOLD_STAGING", str(home / "EtherGold-Staging"))) / "sync"


def _run_eg_script(
    script: str,
    extra_env: Optional[Dict[str, str]] = None,
    timeout: int = 120,
    prefer_staging: bool = False,
) -> Dict[str, Any]:
    path = EG_SYNC / script
    if prefer_staging:
        staged = _staging_sync() / Path(script).name
        if staged.is_file():
            path = staged
    if not path.is_file():
        return {"ok": False, "error": f"missing {path}", "exit": 127}
    env = os.environ.copy()
    if extra_env:
        env.update(extra_env)
    env.setdefault("HOME", str(Path("/home") / TOPBOX_USER))
    cmd = _cmd_prefix() + ["/bin/bash", str(path)]
    started = time.monotonic()
    try:
        proc = subprocess.run(
            cmd,
            cwd=str(path.parent),
            capture_output=True,
            text=True,
            timeout=timeout,
            check=False,
            env=env,
        )
    except subprocess.TimeoutExpired as exc:
        return {
            "ok": False,
            "error": f"timeout after {timeout}s",
            "cmd": " ".join(cmd),
            "stdout": (exc.stdout or "")[-2000:] if isinstance(exc.stdout, str) else "",
            "stderr": (exc.stderr or "")[-2000:] if isinstance(exc.stderr, str) else "",
        }
    except OSError as exc:
        return {"ok": False, "error": str(exc), "cmd": " ".join(cmd)}
    return {
        "ok": proc.returncode == 0,
        "exit": proc.returncode,
        "seconds": round(time.monotonic() - started, 1),
        "stdout": (proc.stdout or "")[-4000:],
        "stderr": (proc.stderr or "")[-2000:],
        "cmd": " ".join(cmd),
    }


def _version(root: Path) -> str:
    return _read_text(root / "VERSION.txt") or _read_text(root / "prod" / "VERSION.txt")


def _http_working_directory(
    show_cmd: Optional[List[str]] = None,
) -> str:
    """Return ethergold-http WorkingDirectory, or empty if unavailable."""
    cmd = show_cmd or [
        "systemctl",
        "show",
        "ethergold-http.service",
        "-p",
        "WorkingDirectory",
        "--value",
    ]
    try:
        proc = subprocess.run(
            cmd,
            capture_output=True,
            text=True,
            timeout=5,
            check=False,
        )
    except (OSError, subprocess.TimeoutExpired):
        return ""
    if proc.returncode != 0:
        return ""
    return (proc.stdout or "").strip()


def _classify_run_tree(wd: str, eg_root: Path | None = None) -> str:
    """Map HTTP WD to ``dev`` | ``prod`` | ``other`` | ``unknown``."""
    raw = (wd or "").strip()
    if not raw:
        return "unknown"
    try:
        path = Path(raw).resolve()
    except OSError:
        path = Path(raw)
    name = path.name.lower()
    parent = path.parent.name.lower()
    if name == "dev" and parent == "ethergold":
        return "dev"
    if name == "prod" and parent == "ethergold":
        return "prod"
    root = eg_root or EG_ROOT
    try:
        resolved_root = root.resolve()
    except OSError:
        resolved_root = root
    if path == resolved_root / "dev":
        return "dev"
    if path == resolved_root / "prod":
        return "prod"
    return "other"


def _running_live_fields(
    eg_root: Path | None = None,
    http_wd: Optional[str] = None,
) -> Dict[str, str]:
    """Live stamp from the HTTP run tree; OTA package stays separate."""
    root = eg_root or EG_ROOT
    ota_live = _version(root)
    wd = http_wd if http_wd is not None else _http_working_directory()
    run_tree = _classify_run_tree(wd, root)
    run_root = (wd or "").strip()
    live_version = ""
    if run_root:
        live_version = _version(Path(run_root))
    if not live_version and run_tree == "dev":
        live_version = _version(root / "dev")
    if not live_version and run_tree == "prod":
        live_version = _version(root / "prod") or ota_live
    if not live_version:
        tip = _version(root / "dev")
        if tip and tip != ota_live:
            live_version = tip
            if run_tree == "unknown":
                run_tree = "dev"
                run_root = str(root / "dev")
        else:
            live_version = ota_live or tip
            if run_tree == "unknown" and ota_live:
                run_tree = "prod"
                run_root = str(root / "prod")
    live_source = (
        "promote"
        if run_tree == "prod" and live_version and live_version == ota_live
        else "outside_promote"
    )
    return {
        "live_version": live_version,
        "ota_live_version": ota_live,
        "run_tree": run_tree,
        "run_root": run_root,
        "live_source": live_source,
    }


def _parse_core_versions(tip: Path) -> Dict[str, str]:
    core = tip / "pog422_core.py"
    text = _read_text(core)
    eg = ""
    abi = ""
    slug = ""
    m = re.search(r'^EG_VERSION\s*=\s*"([^"]+)"', text, re.M)
    if m:
        eg = m.group(1)
    m = re.search(r'^UARTD_ABI_VERSION\s*=\s*"([^"]+)"', text, re.M)
    if m:
        abi = m.group(1)
    m = re.search(r'^EG_VERSION_SLUG\s*=\s*"([^"]+)"', text, re.M)
    if m:
        slug = m.group(1)
    return {"eg_version": eg, "uartd_abi_expected": abi, "eg_version_slug": slug}


def _binary_best_stamp(path: Path) -> str:
    if not path.is_file():
        return ""
    try:
        proc = subprocess.run(
            ["strings", str(path)],
            capture_output=True,
            text=True,
            timeout=10,
            check=False,
        )
    except (OSError, subprocess.TimeoutExpired):
        return ""
    best = ""
    for line in (proc.stdout or "").splitlines():
        tok = line.strip()
        if re.fullmatch(r"\d+\.\d+\.\d+\.\d+", tok):
            if not best or _stamp_is_older(best, tok) or best == tok:
                # keep highest
                if not best or (_stamp_tuple(tok) or ()) >= (_stamp_tuple(best) or ()):
                    best = tok
    return best


def _tty_present() -> bool:
    try:
        return any(Path("/dev").glob("ttyXRUSB*"))
    except OSError:
        return False


def _apply_job_path(eg_root: Path | None = None) -> Path:
    root = eg_root or EG_ROOT
    return root / "run" / APPLY_JOB_NAME


def _load_apply_job(eg_root: Path | None = None) -> Dict[str, Any]:
    path = _apply_job_path(eg_root)
    raw = _read_text(path)
    if not raw:
        return {}
    try:
        data = json.loads(raw)
    except json.JSONDecodeError:
        return {"phase": "failed", "error": "corrupt apply-job.json"}
    return data if isinstance(data, dict) else {}


def _save_apply_job(job: Dict[str, Any], eg_root: Path | None = None) -> None:
    root = eg_root or EG_ROOT
    run = root / "run"
    run.mkdir(parents=True, exist_ok=True)
    path = run / APPLY_JOB_NAME
    job = dict(job)
    job["updated_at"] = time.strftime("%Y-%m-%dT%H:%M:%S")
    path.write_text(json.dumps(job, indent=2) + "\n", encoding="utf-8")


def _clear_apply_job(eg_root: Path | None = None) -> None:
    path = _apply_job_path(eg_root)
    try:
        path.unlink(missing_ok=True)  # type: ignore[call-arg]
    except TypeError:
        if path.is_file():
            path.unlink()
    except OSError:
        pass


COLLECTOR_UNIT = os.environ.get("ETHERGOLD_COLLECTOR_UNIT", "ethergold-collector.service")


def _unit_active(unit: str) -> bool:
    try:
        proc = subprocess.run(
            ["systemctl", "is-active", unit],
            capture_output=True,
            text=True,
            timeout=5,
            check=False,
        )
    except (OSError, subprocess.TimeoutExpired):
        return False
    return (proc.stdout or "").strip() == "active"


def _role(eg_root: Path | None = None) -> str:
    """ETHERGOLD_ROLE as run_ethergold.sh resolves it (env, then role.env)."""
    role = (os.environ.get("ETHERGOLD_ROLE") or "").strip().lower()
    if role:
        return role
    root = eg_root or EG_ROOT
    for path in (Path("/etc/ethergold/role.env"), root / "sync" / "role.env"):
        for line in _read_text(path).splitlines():
            if line.startswith("ETHERGOLD_ROLE="):
                return line.split("=", 1)[1].strip().strip('"').strip("'").lower()
    return "topbox"


def _worker_ownership(eg_root: Path | None = None) -> Dict[str, Any]:
    """Who owns the EG worker: systemd collector, or manual run_ethergold.sh.

    Start/Stop EG must not fight the collector — pkill of pog422_listen.py comes
    back in RestartSec (see stop_fixture_session.sh).
    """
    collector_active = _unit_active(COLLECTOR_UNIT)
    return {
        "role": _role(eg_root),
        "collector_unit": COLLECTOR_UNIT,
        "collector_active": collector_active,
        "worker_owner": "collector" if collector_active else "manual",
        "run_eg_allowed": not collector_active,
    }


def _tip_fields(eg_root: Path | None = None) -> Dict[str, Any]:
    root = eg_root or EG_ROOT
    tip = root / "dev"
    core = _parse_core_versions(tip) if tip.is_dir() else {}
    uartd = tip / "inject-shim" / "pog422_uartd"
    so = tip / "inject-shim" / "libpog422_inject.so"
    uartd_ver = _binary_best_stamp(uartd)
    so_ver = _binary_best_stamp(so)
    abi = core.get("uartd_abi_expected") or ""
    tip_ver = _version(tip) or core.get("eg_version") or ""
    tip_abi_ok = bool(abi) and uartd_ver == abi and (not so_ver or so_ver == abi)
    job = _load_apply_job(root)
    return {
        "tip_version": tip_ver,
        "eg_version_slug": core.get("eg_version_slug") or "",
        "uartd_abi_expected": abi,
        "uartd_binary_version": uartd_ver,
        "inject_so_version": so_ver,
        "tip_abi_ok": tip_abi_ok,
        "tty_present": _tty_present(),
        "apply_phase": job.get("phase") or "",
        "apply_job": job,
        "usb_want_driver_ver": job.get("want_driver") or "",
        "last_apply": _read_text(root / "run" / "last-apply.txt"),
        "last_dev_overlay": _read_text(root / "run" / "last-dev-overlay.txt"),
    }


def deploy_status(
    eg_root: Path | None = None,
    http_wd: Optional[str] = None,
    staging_root: Path | None = None,
    rollback_root: Path | None = None,
) -> Dict[str, Any]:
    root = eg_root or EG_ROOT
    pending = root / "run" / "staging-pending"
    home = Path("/home") / TOPBOX_USER
    staging = staging_root or Path(
        os.environ.get("ETHERGOLD_STAGING", str(home / "EtherGold-Staging"))
    )
    rollback = rollback_root or Path(
        os.environ.get("ETHERGOLD_ROLLBACK", str(home / "EtherGold-Rollback"))
    )
    rollback_empty = True
    if rollback.is_dir():
        try:
            rollback_empty = not any(rollback.iterdir())
        except OSError:
            rollback_empty = True
    hold = _read_text(root / "run" / "promote-hold")
    running = _running_live_fields(eg_root=root, http_wd=http_wd)
    ota_live_version = running["ota_live_version"]
    live_version = running["live_version"]
    staging_version = _version(staging)
    staging_older = _stamp_is_older(staging_version, ota_live_version)
    pending_file = pending.is_file()
    tip = _tip_fields(root)
    out: Dict[str, Any] = {
        "ok": True,
        "product": "ethergold",
        "eg_root": str(root),
        "ota_channel": _ota_channel(root),
        "live_version": live_version,
        "ota_live_version": ota_live_version,
        "run_tree": running["run_tree"],
        "run_root": running["run_root"],
        "live_source": running["live_source"],
        "staging_version": staging_version,
        "rollback_version": _version(rollback),
        "staging_pending": pending_file and not staging_older,
        "staging_older": staging_older,
        "staging_pending_text": _read_text(pending) if pending_file else "",
        "promote_hold": bool(hold),
        "promote_hold_text": hold,
        "last_promote": _read_text(root / "run" / "last-promote.txt"),
        "last_rollback": _read_text(root / "run" / "last-rollback.txt"),
        "last_rollback_directive": _read_text(root / "run" / "last-rollback-directive"),
        "rollback_available": not rollback_empty,
        "rollback_dir": str(rollback),
        "staging_dir": str(staging),
    }
    out.update(tip)
    out.update(_worker_ownership(root))
    return out


def check_for_updates() -> Dict[str, Any]:
    """rclone SGVI → Staging only. Does not promote or restart listen/HTTP."""
    stage = _run_eg_script(
        _first_script("stage_from_dropbox.sh", "stage-from-dropbox.sh"),
        timeout=STAGE_TIMEOUT_SEC,
    )
    status = deploy_status()
    pending = bool(status.get("staging_pending"))
    available = bool(stage.get("ok")) and pending
    return {
        "ok": bool(stage.get("ok")),
        "step": "check",
        "update_available": available,
        "stage": stage,
        "status": status,
    }


def _parse_kv_stdout(text: str) -> Dict[str, str]:
    out: Dict[str, str] = {}
    for line in (text or "").splitlines():
        if "=" not in line:
            continue
        k, _, v = line.partition("=")
        k = k.strip()
        if k:
            out[k] = v.strip()
    return out


def _run_tip_phase(phase: str, extra_env: Optional[Dict[str, str]] = None) -> Dict[str, Any]:
    env = {"PHASE": phase, "WEB_APPLY": "1"}
    if extra_env:
        env.update(extra_env)
    return _run_eg_script(
        "web_tip_apply.sh",
        extra_env=env,
        timeout=TIP_APPLY_TIMEOUT_SEC,
    )


def _ota_channel(eg_root: Path | None = None) -> str:
    """Resolve OTA_CHANNEL (env, then sync/rclone.env). Default field."""
    ch = (os.environ.get("OTA_CHANNEL") or "").strip().lower()
    if ch:
        return ch
    root = eg_root or EG_ROOT
    for path in (EG_SYNC / "rclone.env", root / "sync" / "rclone.env"):
        raw = _read_text(path)
        for line in raw.splitlines():
            if line.startswith("OTA_CHANNEL="):
                return line.split("=", 1)[1].strip().strip('"').strip("'").lower()
    return "field"


def resolve_ota_channel(eg_root: Path | None = None) -> str:
    """Public alias for /health, /status, and /config OTA channel badges."""
    return _ota_channel(eg_root)


def _should_tip_apply(eg_root: Path | None = None) -> bool:
    """Dev-channel bench overlays /dev + rebuilds tip; field stays prod-only."""
    if (os.environ.get("APPLY_DEV_TIP") or "").strip() == "1":
        return True
    if _ota_channel(eg_root) == "dev":
        return True
    root = eg_root or EG_ROOT
    tip = root / "dev"
    if not tip.is_dir():
        return False
    running = _running_live_fields(eg_root=root)
    return running.get("run_tree") == "dev"


def _continue_tip_apply_after_promote(eg_root: Path | None = None) -> Dict[str, Any]:
    """Plan → inject → maybe pause for USB → finish. Never silent SKIP_DRIVER."""
    root = eg_root or EG_ROOT
    if not _should_tip_apply(root):
        # Still restart HTTP so Promote's deferred WEB_APPLY path completes.
        fin = _run_tip_phase("finish")
        return {
            "ok": bool(fin.get("ok")),
            "pending": False,
            "step": "tip_apply_skipped_field",
            "phase": "complete" if fin.get("ok") else "failed",
            "status": deploy_status(eg_root=root),
        }
    plan = _run_tip_phase("plan")
    kv = _parse_kv_stdout(plan.get("stdout") or "")
    need_inject = kv.get("need_inject") == "1" or plan.get("exit") == 11
    need_usb = kv.get("need_usb") == "1" or plan.get("exit") == 10
    # exit 12 = nothing; exit 10 = usb; exit 11 = inject only
    if plan.get("exit") == 10:
        need_usb = True
    if plan.get("exit") == 11:
        need_inject = True
    if plan.get("exit") == 12:
        need_inject = False
        need_usb = False

    job: Dict[str, Any] = {
        "phase": "running",
        "need_inject": need_inject,
        "need_usb": need_usb,
        "plan": kv,
        "want_driver": kv.get("want_driver") or "",
    }

    if need_inject:
        inj = _run_tip_phase("inject")
        job["inject"] = {
            "ok": inj.get("ok"),
            "exit": inj.get("exit"),
            "stdout": inj.get("stdout"),
            "stderr": inj.get("stderr"),
        }
        if not inj.get("ok"):
            job["phase"] = "failed"
            job["error"] = "inject rebuild failed"
            _save_apply_job(job, root)
            return {
                "ok": False,
                "pending": False,
                "step": "tip_inject_failed",
                "job": job,
                "status": deploy_status(eg_root=root),
            }

    if need_usb:
        job["phase"] = "await_unplug"
        job["message"] = (
            "Disconnect the Exar hat USB cable and leave it out, "
            "then click Disconnected."
        )
        _save_apply_job(job, root)
        return {
            "ok": False,
            "pending": True,
            "step": "await_unplug",
            "phase": "await_unplug",
            "job": job,
            "status": deploy_status(eg_root=root),
        }

    fin = _run_tip_phase("finish")
    job["phase"] = "complete" if fin.get("ok") else "failed"
    job["finish"] = {"ok": fin.get("ok"), "exit": fin.get("exit")}
    if fin.get("ok"):
        _clear_apply_job(root)
        # keep a breadcrumb
        _save_apply_job({**job, "phase": "complete"}, root)
    else:
        job["error"] = "finish failed"
        _save_apply_job(job, root)
    return {
        "ok": bool(fin.get("ok")),
        "pending": False,
        "step": "tip_apply_complete" if fin.get("ok") else "tip_finish_failed",
        "phase": job["phase"],
        "job": job,
        "status": deploy_status(eg_root=root),
    }


def apply_usb_unplugged() -> Dict[str, Any]:
    """Operator confirmed disconnect (update.sh Enter 1/2)."""
    root = EG_ROOT
    job = _load_apply_job(root)
    if job.get("phase") != "await_unplug":
        return {
            "ok": False,
            "error": f"expected await_unplug, got {job.get('phase') or 'none'}",
            "status": deploy_status(),
        }
    res = _run_tip_phase("usb_unplug")
    if not res.get("ok"):
        job["phase"] = "failed"
        job["error"] = "usb_unplug failed"
        job["usb_unplug"] = res
        _save_apply_job(job, root)
        return {
            "ok": False,
            "step": "usb_unplug_failed",
            "result": res,
            "status": deploy_status(),
        }
    job["phase"] = "await_replug"
    job["message"] = (
        "Reconnect the Exar hat USB cable, then click Reconnected."
    )
    job["usb_unplug"] = {"ok": True}
    _save_apply_job(job, root)
    return {
        "ok": False,
        "pending": True,
        "step": "await_replug",
        "phase": "await_replug",
        "job": job,
        "status": deploy_status(),
    }


def apply_usb_replugged() -> Dict[str, Any]:
    """Operator confirmed reconnect (update.sh Enter 2/2)."""
    root = EG_ROOT
    job = _load_apply_job(root)
    if job.get("phase") != "await_replug":
        return {
            "ok": False,
            "error": f"expected await_replug, got {job.get('phase') or 'none'}",
            "status": deploy_status(),
        }
    env = {}
    if job.get("want_driver"):
        env["WANT_DRIVER"] = str(job["want_driver"])
    res = _run_tip_phase("usb_replug", extra_env=env)
    if not res.get("ok"):
        job["phase"] = "failed"
        job["error"] = "usb_replug failed"
        job["usb_replug"] = res
        _save_apply_job(job, root)
        return {
            "ok": False,
            "step": "usb_replug_failed",
            "result": res,
            "status": deploy_status(),
        }
    # Persist past await_replug BEFORE finish. finish used to restart HTTP under
    # WEB_APPLY and kill this handler — job stayed await_replug and the modal looped.
    job["phase"] = "finishing"
    job["usb_replug"] = {"ok": True}
    job["message"] = "USB reconnect confirmed — finishing tip apply…"
    _save_apply_job(job, root)

    fin = _run_tip_phase("finish")
    job["phase"] = "complete" if fin.get("ok") else "failed"
    job["finish"] = {"ok": fin.get("ok"), "exit": fin.get("exit")}
    if fin.get("ok"):
        # Clear so Status polls do not keep a sticky apply_phase.
        _clear_apply_job(root)
    else:
        job["error"] = "finish failed"
        _save_apply_job(job, root)
    return {
        "ok": bool(fin.get("ok")),
        "pending": False,
        "step": "tip_apply_complete" if fin.get("ok") else "tip_finish_failed",
        "phase": job["phase"],
        "job": job,
        "status": deploy_status(eg_root=root),
    }


def apply_usb_cancel() -> Dict[str, Any]:
    """Dismiss / cancel a paused USB apply job so Status stops re-showing the modal."""
    root = EG_ROOT
    job = _load_apply_job(root)
    phase = job.get("phase") or ""
    if phase in ("await_unplug", "await_replug", "finishing", "failed", "complete"):
        _clear_apply_job(root)
        return {
            "ok": True,
            "pending": False,
            "step": "usb_apply_cancelled",
            "phase": "",
            "cleared_phase": phase,
            "status": deploy_status(eg_root=root),
        }
    if not job:
        return {
            "ok": True,
            "pending": False,
            "step": "usb_apply_idle",
            "phase": "",
            "status": deploy_status(eg_root=root),
        }
    return {
        "ok": False,
        "error": f"cannot cancel phase {phase or 'none'}",
        "status": deploy_status(eg_root=root),
    }


def update_now() -> Dict[str, Any]:
    stage = _run_eg_script(
        _first_script("stage_from_dropbox.sh", "stage-from-dropbox.sh"),
        timeout=STAGE_TIMEOUT_SEC,
    )
    if not stage.get("ok"):
        return {"ok": False, "step": "stage", "result": stage, "status": deploy_status()}
    # APPLY_DEV_TIP only when OTA_CHANNEL=dev (promote_staging.sh); do not force on field.
    promote = _run_eg_script(
        _first_script("promote_staging.sh", "promote-staging.sh"),
        extra_env={
            "FORCE": "1",
            "PROMOTE_RESTART": "1",
            "WEB_APPLY": "1",
        },
        timeout=PROMOTE_TIMEOUT_SEC,
        prefer_staging=True,
    )
    if not promote.get("ok"):
        return {
            "ok": False,
            "step": "promote_failed",
            "stage": stage,
            "promote": promote,
            "status": deploy_status(),
        }
    tip = _continue_tip_apply_after_promote()
    return {
        "ok": bool(tip.get("ok")) if not tip.get("pending") else False,
        "pending": bool(tip.get("pending")),
        "step": tip.get("step") or "promote",
        "phase": tip.get("phase"),
        "stage": stage,
        "promote": promote,
        "tip_apply": tip,
        "status": tip.get("status") or deploy_status(),
    }


def promote_now(confirm_hold: bool = False) -> Dict[str, Any]:
    # Overlay /dev only when OTA_CHANNEL=dev (see promote_staging.sh). Field = prod-only.
    env = {
        "FORCE": "1",
        "PROMOTE_RESTART": "1",
        "WEB_APPLY": "1",
    }
    if confirm_hold:
        env["CONFIRM_PROMOTE_HOLD"] = "1"
    promote = _run_eg_script(
        _first_script("promote_staging.sh", "promote-staging.sh"),
        extra_env=env,
        timeout=PROMOTE_TIMEOUT_SEC,
        prefer_staging=True,
    )
    if not promote.get("ok"):
        return {
            "ok": False,
            "step": "promote_failed",
            "promote": promote,
            "status": deploy_status(),
        }
    tip = _continue_tip_apply_after_promote()
    return {
        "ok": bool(tip.get("ok")) if not tip.get("pending") else False,
        "pending": bool(tip.get("pending")),
        "step": tip.get("step") or "promote",
        "phase": tip.get("phase"),
        "promote": promote,
        "tip_apply": tip,
        "status": tip.get("status") or deploy_status(),
    }


def rollback_now() -> Dict[str, Any]:
    result = _run_eg_script(
        "rollback_live.sh",
        extra_env={"RESTART": "1"},
        timeout=PROMOTE_TIMEOUT_SEC,
    )
    return {
        "ok": bool(result.get("ok")),
        "step": "rollback",
        "result": result,
        "status": deploy_status(),
    }


def _job_log_path(name: str) -> Path:
    return EG_ROOT / "run" / f"web-{name}.log"


def _write_job_meta(name: str, meta: Dict[str, Any]) -> None:
    run = EG_ROOT / "run"
    run.mkdir(parents=True, exist_ok=True)
    (run / f"web-{name}-job.json").write_text(
        json.dumps(meta, indent=2) + "\n", encoding="utf-8"
    )


def bridge_start() -> Dict[str, Any]:
    """Start run_ethergold.sh in background from tip /dev (role picks listen vs bridge)."""
    own = _worker_ownership()
    if not own["run_eg_allowed"]:
        return {
            "ok": False,
            "error": (
                f"{own['collector_unit']} owns the EG worker — a second listen would "
                "contend for uartd / Pi HS. Use Recover stack instead."
            ),
            "worker_owner": own["worker_owner"],
            "status": deploy_status(),
        }
    tip = EG_ROOT / "dev" / "run_ethergold.sh"
    if not tip.is_file():
        tip = EG_ROOT / "prod" / "run_ethergold.sh"
    if not tip.is_file():
        return {"ok": False, "error": "missing run_ethergold.sh", "status": deploy_status()}
    log = _job_log_path("bridge")
    pid_file = EG_ROOT / "run" / "web-bridge.pid"
    EG_ROOT.joinpath("run").mkdir(parents=True, exist_ok=True)
    # Already running?
    if pid_file.is_file():
        try:
            old = int(_read_text(pid_file) or "0")
        except ValueError:
            old = 0
        if old > 0:
            try:
                os.kill(old, 0)
                return {
                    "ok": True,
                    "step": "bridge_already_running",
                    "pid": old,
                    "status": deploy_status(),
                }
            except OSError:
                pass
    cmd = _cmd_prefix() + [
        "/bin/bash",
        "-c",
        f"nohup bash {tip} >>{log} 2>&1 & echo $! > {pid_file}",
    ]
    try:
        proc = subprocess.run(
            cmd,
            capture_output=True,
            text=True,
            timeout=BRIDGE_TIMEOUT_SEC,
            check=False,
            env={**os.environ, "HOME": str(Path("/home") / TOPBOX_USER)},
        )
    except (OSError, subprocess.TimeoutExpired) as exc:
        return {"ok": False, "error": str(exc), "status": deploy_status()}
    pid = 0
    try:
        pid = int(_read_text(pid_file) or "0")
    except ValueError:
        pid = 0
    meta = {
        "ok": proc.returncode == 0 and pid > 0,
        "pid": pid,
        "log": str(log),
        "started_at": time.strftime("%Y-%m-%dT%H:%M:%S"),
    }
    _write_job_meta("bridge", meta)
    return {
        "ok": bool(meta["ok"]),
        "step": "bridge_start",
        "pid": pid,
        "log": str(log),
        "status": deploy_status(),
    }


def bridge_stop() -> Dict[str, Any]:
    """Stop bridge/listen workers; SKIP_SYNC=1 (sync is explicit button)."""
    own = _worker_ownership()
    if not own["run_eg_allowed"]:
        return {
            "ok": False,
            "error": (
                f"{own['collector_unit']} owns the EG worker — pkill comes back in "
                "RestartSec. Stop the unit on the Pi, or use Recover stack."
            ),
            "worker_owner": own["worker_owner"],
            "status": deploy_status(),
        }
    candidates = [
        EG_ROOT / "dev" / "stop_fixture_session.sh",
        EG_ROOT / "sync" / "stop_fixture_session.sh",
        EG_SYNC / "stop_fixture_session.sh",
    ]
    script = next((p for p in candidates if p.is_file()), None)
    log = _job_log_path("bridge-stop")
    if script is not None:
        env = {
            "SKIP_SYNC": "1",
            "HOME": str(Path("/home") / TOPBOX_USER),
        }
        cmd = _cmd_prefix() + ["/bin/bash", str(script)]
        try:
            with log.open("a", encoding="utf-8") as fh:
                fh.write(f"\n--- stop {time.strftime('%Y-%m-%dT%H:%M:%S')} ---\n")
                proc = subprocess.run(
                    cmd,
                    stdout=fh,
                    stderr=subprocess.STDOUT,
                    timeout=BRIDGE_TIMEOUT_SEC,
                    check=False,
                    env={**os.environ, **env},
                    cwd=str(script.parent),
                )
            ok = proc.returncode == 0
        except (OSError, subprocess.TimeoutExpired) as exc:
            return {"ok": False, "error": str(exc), "status": deploy_status()}
    else:
        # Fallback: kill known workers
        for pat in (
            "run_ethergold.sh",
            "pog422_bridge.py",
            "pog422_fixture_bridge",
            "pog422_uartd",
        ):
            subprocess.run(
                _cmd_prefix() + ["pkill", "-f", pat],
                capture_output=True,
                check=False,
            )
        ok = True
    pid_file = EG_ROOT / "run" / "web-bridge.pid"
    try:
        pid_file.unlink(missing_ok=True)  # type: ignore[call-arg]
    except TypeError:
        if pid_file.is_file():
            pid_file.unlink()
    except OSError:
        pass
    return {
        "ok": ok,
        "step": "bridge_stop",
        "log": str(log),
        "status": deploy_status(),
    }


def sync_capture() -> Dict[str, Any]:
    """Upload capture via sync_to_dropbox.sh."""
    script = EG_SYNC / "sync_to_dropbox.sh"
    if not script.is_file():
        return {
            "ok": False,
            "error": f"missing {script}",
            "status": deploy_status(),
        }
    log = _job_log_path("sync")
    cmd = _cmd_prefix() + ["/bin/bash", str(script)]
    started = time.monotonic()
    try:
        with log.open("a", encoding="utf-8") as fh:
            fh.write(f"\n--- sync {time.strftime('%Y-%m-%dT%H:%M:%S')} ---\n")
            proc = subprocess.run(
                cmd,
                stdout=fh,
                stderr=subprocess.STDOUT,
                timeout=SYNC_TIMEOUT_SEC,
                check=False,
                env={**os.environ, "HOME": str(Path("/home") / TOPBOX_USER)},
                cwd=str(script.parent),
            )
    except subprocess.TimeoutExpired:
        return {
            "ok": False,
            "error": f"timeout after {SYNC_TIMEOUT_SEC}s",
            "log": str(log),
            "status": deploy_status(),
        }
    except OSError as exc:
        return {"ok": False, "error": str(exc), "status": deploy_status()}
    meta = {
        "ok": proc.returncode == 0,
        "exit": proc.returncode,
        "seconds": round(time.monotonic() - started, 1),
        "log": str(log),
        "finished_at": time.strftime("%Y-%m-%dT%H:%M:%S"),
    }
    _write_job_meta("sync", meta)
    return {
        "ok": bool(meta["ok"]),
        "step": "sync_capture",
        "exit": proc.returncode,
        "seconds": meta["seconds"],
        "log": str(log),
        "status": deploy_status(),
    }


def job_status(name: str) -> Dict[str, Any]:
    """Return last job meta + log tail."""
    meta_path = EG_ROOT / "run" / f"web-{name}-job.json"
    log_path = _job_log_path(name if name != "bridge-stop" else "bridge-stop")
    if name == "bridge":
        log_path = _job_log_path("bridge")
    elif name == "sync":
        log_path = _job_log_path("sync")
    meta: Dict[str, Any] = {}
    raw = _read_text(meta_path)
    if raw:
        try:
            meta = json.loads(raw)
        except json.JSONDecodeError:
            meta = {"error": "corrupt job json"}
    tail = ""
    try:
        if log_path.is_file():
            data = log_path.read_text(encoding="utf-8", errors="replace")
            tail = data[-4000:]
    except OSError:
        pass
    return {
        "ok": True,
        "name": name,
        "meta": meta,
        "log_tail": tail,
        "apply_job": _load_apply_job(),
        "status": deploy_status(),
    }