Altium - the Altium Bridge

"""progress.py — the MEASURED-progress contract (wiki issue #32).

John's rule, verbatim from the issue: every slow verb reports measured progress,
never "can take a couple of minutes". A shrug is not an estimate: it cannot be
drawn as a bar, it never narrows, and it hides the difference between slow and
wedged.

So this module owns three things:

1. A rolling, PER-MACHINE history of how long each phase actually took, split by
   cold vs warm start, persisted next to the bridge's other state
   (%USERPROFILE%/.adom/altium-bridge-timings.json). Estimates come from this
   machine's own p50 - never from a hardcoded literal, because the literal is
   what made the old message useless.

2. Honest CONFIDENCE. A fresh box has no samples, and labelling a shipped
   default as "measured" would be a lie:
       0 samples  -> confidence "unknown"  (seeded from the shipped default)
       1-2        -> confidence "typical"
       3 or more  -> confidence "measured" (p50 of this machine's own history)
   This rule is the caveat I raised on #32 and it lives here, in one place.

3. The block itself, in the shape ab consumes verbatim - plus the two amendments
   I proposed on the issue: phase ids are NAMESPACED ("kicad.cold_start") so a UI
   can group histories across bridges without collisions, and `blockedBy` names a
   human-controlled wait (a modal, an unsaved-changes prompt, a UAC dialog) so a
   UI can say "waiting for you" instead of animating a bar through something that
   cannot progress.
"""
import json
import os
import threading
import time
from pathlib import Path

PREFIX = "altium"
_KEEP = 10         # rolling window per key (fusion parity, John 2026-08-21)
_lock = threading.Lock()

# Shipped defaults, used ONLY to seed a machine with no history of its own.
# Every one of these is a measured figure from this fleet, not a guess.
_DEFAULT_SECONDS = {
    "cold_start": 60.0,       # X2.exe from nothing (AD26, measured once — refine)
    "warm_start": 1.0,        # Altium already running
    "agent_start": 8.0,       # bootstrapping the resident DelphiScript agent
    "show_symbol": 12.0,
    "show_footprint": 20.0,   # the Footprint Editor loads its libraries as it opens
    "show_3d_chip": 25.0,
    "show_library": 12.0,
    "show_schematic": 15.0,
    "show_2d_board": 15.0,
    "show_3d_board": 30.0,
    "show_project": 10.0,
    "install_plugin": 6.0,
    "run_outjob": 60.0,
    "export": 8.0,
    "demo": 210.0,
}


def _path() -> Path:
    base = os.environ.get("USERPROFILE") or str(Path.home())
    d = Path(base) / ".adom"
    try:
        d.mkdir(parents=True, exist_ok=True)
    except Exception:
        pass
    return d / "altium-bridge-timings.json"


def _load() -> dict:
    try:
        return json.loads(_path().read_text(encoding="utf-8"))
    except Exception:
        return {}


def _save(data: dict) -> None:
    try:
        _path().write_text(json.dumps(data, indent=2), encoding="utf-8")
    except Exception:
        pass


def _key(phase: str, start_kind: str) -> str:
    return f"{phase}:{start_kind or 'warm'}"


def _pct(samples: list, q: float) -> float:
    if not samples:
        return 0.0
    s = sorted(samples)
    i = min(len(s) - 1, max(0, int(round(q * (len(s) - 1)))))
    return round(s[i], 2)


def record(phase: str, seconds: float, start_kind: str = "warm") -> None:
    """Add one real measurement. This is what makes confidence 'measured'."""
    if seconds is None or seconds < 0:
        return
    with _lock:
        data = _load()
        k = _key(phase, start_kind)
        samples = list(data.get(k) or [])
        samples.append(round(float(seconds), 2))
        data[k] = samples[-_KEEP:]
        _save(data)


def estimate(phase: str, start_kind: str = "warm") -> tuple:
    """(estimatedSec, confidence) for this phase ON THIS MACHINE."""
    data = _load()
    samples = data.get(_key(phase, start_kind)) or []
    if len(samples) >= 3:
        return _pct(samples, 0.5), "measured"
    default = _DEFAULT_SECONDS.get(phase)
    if samples:
        # blend the shipped default with what little we have measured
        base = default if default is not None else samples[-1]
        return round((sum(samples) / len(samples) + base) / 2, 2), "typical"
    if default is not None:
        # A phase-SPECIFIC seeded literal is "typical" per the ladder (John's
        # directive: measured > typical [1-2 samples, or a seeded literal] >
        # unknown). It was mislabelled "unknown" here.
        return float(default), "typical"
    # No samples AND no seed: genuinely unknown. Return 0 so block() renders an
    # indeterminate marquee. The old fallback borrowed the generic warm default
    # (1.0s) for phases it had never heard of, which made a 5s run look deeply
    # overdue at 92% - a guessed number wearing a measurement's clothes, the
    # exact failure mode this system exists to kill.
    return 0.0, "unknown"


def history() -> dict:
    """Everything we know, for describe/status: p50 + p90 + sample counts."""
    out = {}
    for k, samples in (_load() or {}).items():
        if not samples:
            continue
        phase, _, kind = k.partition(":")
        out.setdefault(phase, {})[kind] = {
            "samples": len(samples), "p50": _pct(samples, 0.5), "p90": _pct(samples, 0.9),
        }
    return out


def start_kind_for_altium() -> str:
    """cold = X2.exe not running yet; warm = Altium Designer is already up."""
    try:
        from handlers.altium_detect import running_pids
        return "warm" if running_pids() else "cold"
    except Exception:
        return "cold"

def block(phase: str, step_label: str, started_at: float, percent=None,
          start_kind: str = "warm", blocked_by: str = None, done: bool = False) -> dict:
    """Build the progress block. `percent` may be None while genuinely unknown -
    the contract allows it, as long as stepLabel keeps moving."""
    elapsed = max(0.0, time.time() - started_at)
    est, confidence = estimate(phase, start_kind)
    # Adopted verbatim-in-spirit from fusion's timings.py (John's directive,
    # 2026-08-21: copy the method, do not redesign). Their comments record two
    # live failures these formulas exist to prevent, and both apply here:
    #   * est = elapsed * 1.25 alone pins percent at exactly 80 forever - a
    #     spinner wearing a progress bar's clothes;
    #   * pre-overdue and overdue formulas that do not MEET at elapsed == est0
    #     walk percent BACKWARDS at the crossover (91 then 84, live).
    est0 = est
    overdue = bool(est0 and elapsed > est0)
    if overdue:
        est = round(elapsed * 1.25, 1)     # revised, so eta never lies at 0
        if confidence == "measured":
            confidence = "typical"          # past what this box ever measured
        if elapsed > 2 * est0:
            confidence = "unknown"          # stop claiming to know at all
    if done:
        percent = 100
        est = round(elapsed, 2)
    elif percent is None:
        if not est0:
            percent = None                  # unknown -> indeterminate, never 0%
        elif not overdue:
            percent = int(80 * elapsed / est0)          # 0 -> 80 across the estimate
        else:
            over = elapsed - est0
            percent = int(80 + 15 * (over / (over + est0)))   # 80 -> 95, asymptotic
    eta = None if percent is None else max(0.0, round(est - elapsed, 1))
    out = {
        "phase": f"{PREFIX}.{phase}",
        "stepLabel": step_label,
        "percent": percent,
        "elapsedSec": round(elapsed, 1),
        "estimatedSec": round(est, 1),
        "etaSec": eta,
        "confidence": confidence,
        "startKind": start_kind,
    }
    if not done and est0:
        out["overdue"] = overdue
        if overdue and confidence == "unknown":
            out["stepLabel"] = f"{step_label} (taking far longer than this machine ever has)"
    if blocked_by:
        out["blockedBy"] = blocked_by
        out["stepLabel"] = f"waiting for you: {blocked_by}"
    if done:
        out["measured"] = {"phase": f"{PREFIX}.{phase}", "seconds": round(elapsed, 2),
                           "startKind": start_kind}
    return out


def finish(phase: str, started_at: float, start_kind: str = "warm", ok: bool = True,
           step_label: str = "done") -> dict:
    """Terminal frame: carries the ACTUALS and feeds the history a new sample."""
    elapsed = max(0.0, time.time() - started_at)
    if ok:
        record(phase, elapsed, start_kind)
    b = block(phase, step_label, started_at, percent=100, start_kind=start_kind, done=True)
    if not ok:
        b["percent"] = None
        b["stepLabel"] = step_label
        b.pop("measured", None)
    return b


# ── LIVE, IN-FLIGHT progress (0.9.195) ──────────────────────────────────────
# John's contract to the team (2026-08-19): "ongoing streaming feedback so
# progress bars could be shown in the front-end" and "show live streaming of the
# progress in your text status line". Until now `block()` was only ever built
# ONCE, on the terminal frame - which is a receipt, not a progress bar. A caller
# watching a 24s cold footprint open saw nothing at all until it finished.
#
# So a running verb publishes its current frame here, and anything that can
# poll a verb (ab's `watch` wrapper, the dashboard's SSE hub, a status line)
# can read it while the work is still happening. In-memory on purpose: this is
# liveness, and a stale frame surviving a bridge restart would be a lie.
_live: dict = {}
_live_seq = 0


def _caller() -> str:
    """Who is asking. Used to keep concurrent callers out of each other's steps."""
    try:
        import caller_identity
        return str(caller_identity.thread_name() or "unknown")
    except Exception:
        return "unknown"


def begin(phase: str, start_kind: str = "warm", step_label: str = "starting",
          caller: str = None) -> None:
    """Mark a phase as in flight, SCOPED TO THE CALLER.

    Wiki issue #46 (John): a second caller's steps appeared inside a first
    caller's response - "locating 'OPA2134'" in a response about nRF54L15,
    because someone used a wiki Quick Action on one KiCad while a matrix run
    drove the same box. Two independent web callers, one global step stream.

    Two distinct bugs, both from keying this registry on the phase alone:
      * a reader saw steps that were never theirs, which reads as a wrong-part
        failure to anyone who does not go and check the pixels;
      * worse and quieter, the second caller OVERWROTE the first's entry, so a
        caller could lose its own progress to someone else's call.

    Keying on (caller, phase) fixes both. It degrades honestly rather than
    perfectly: two browser tabs from one origin still share an identity until ab
    ships a per-page discriminator, but every other pair of callers separates
    correctly today, and every frame now says who produced it."""
    global _live_seq
    who = caller or _caller()
    with _lock:
        _live_seq += 1
        _live[(who, phase)] = {"startedAt": time.time(), "startKind": start_kind,
                               "stepLabel": step_label, "percent": None,
                               "blockedBy": None, "seq": _live_seq, "caller": who}


def step(phase: str, step_label: str, percent=None, blocked_by: str = None) -> None:
    """Advance the visible sub-step of an in-flight phase.

    This is what turns one opaque 24-second bar into "launching KiCad" ->
    "indexing symbol libraries" -> "navigating to IN-S42ATR", which is the
    sub-phase breakdown Colby asked for on wiki issue #35."""
    who = _caller()
    with _lock:
        cur = _live.get((who, phase))
        if cur is None:
            # A handler thread may not carry the caller context; fall back to the
            # only in-flight entry for this phase, but never guess when several
            # callers are running the same phase - a wrong attribution is the
            # bug this issue is about.
            matches = [k for k in _live if k[1] == phase]
            if len(matches) != 1:
                return
            cur = _live[matches[0]]
        cur["stepLabel"] = step_label
        cur["blockedBy"] = blocked_by
        if percent is not None:
            # never let a bar go backwards: a bar that retreats reads as broken
            cur["percent"] = max(percent, cur.get("percent") or 0)


def end(phase: str, caller: str = None) -> None:
    who = caller or _caller()
    with _lock:
        if _live.pop((who, phase), None) is None:
            for k in [k for k in _live if k[1] == phase and k[0] == who]:
                _live.pop(k, None)


def live(mine_only: bool = True, asker: str = None) -> list:
    """Every phase currently in flight, as full progress blocks.

    By default a caller sees ONLY ITS OWN phases (#46). Pass mine_only=False for
    an operator view of everything in flight; each frame carries `caller`.

    Percent/ETA are recomputed at READ time from this machine's own history, so
    a poller gets a bar that keeps moving even if the verb has not reported a
    new sub-step since its last one."""
    asker = asker or _caller()
    out = []
    with _lock:
        snapshot = list(_live.items())
    for (who, phase), cur in snapshot:
        if mine_only and who != asker:
            continue
        try:
            # A step's explicit percent is a FLOOR, not a freeze. The first live
            # test pinned the bar at 10% for 47s while the ETA counted to zero,
            # because an explicit percent suppressed the elapsed-derived one -
            # a stuck bar next to a moving ETA reads as a hung job. Take
            # whichever is further along, so the bar always advances and never
            # goes backwards.
            # ONE derivation, not two. This used to compute its own percent
            # (min(95, elapsed/est*95)) alongside block()'s - exactly fusion's
            # crossover trap, where two formulas for one bar disagree and
            # percent walks backwards. Now block() owns the math and the
            # explicit step percent is applied only as a monotonic FLOOR.
            b = block(phase, cur["stepLabel"], cur["startedAt"],
                      percent=None, start_kind=cur.get("startKind", "warm"),
                      blocked_by=cur.get("blockedBy"))
            _floor = cur.get("percent")
            if _floor is not None and b.get("percent") is not None:
                b["percent"] = max(int(_floor), b["percent"])
            elif _floor is not None:
                b["percent"] = int(_floor)
            b["seq"] = cur.get("seq")
            b["caller"] = who
            out.append(b)
        except Exception:
            pass
    return out


# ── Per-VERB run times (fusion parity, John's directive 2026-08-21) ─────────
# Lifted from fusion's timings.py. Two stores because they answer different
# questions: the rolling window above drives the LIVE estimate (recent, tracks
# a machine that got faster or slower); the append-only log below is the
# HISTORICAL record, every run WITH its outcome - a verb that fails fast is a
# different bug from a slow one, and durations alone cannot tell them apart.
# A rolling window can only answer what it still remembers; a log can be
# re-analysed for questions nobody thought to ask up front.

_RUNLOG_MAX_BYTES = 4 * 1024 * 1024


def _runlog_path() -> Path:
    return _path().parent / "altium-verb-times.jsonl"


def log_run(verb: str, seconds: float, ok: bool = True, note: str = "") -> None:
    """Append one verb run to the historical log AND the rolling window.
    Called at the dispatch chokepoint for EVERY verb - one wrapper, not
    per-verb calls someone forgets. Never raises."""
    try:
        seconds = float(seconds)
        if seconds < 0 or seconds > 7200:
            return                      # a clock jump is not a sample
        record(f"verb:{verb}", seconds)
        p = _runlog_path()
        try:
            if p.exists() and p.stat().st_size > _RUNLOG_MAX_BYTES:
                p.replace(str(p) + ".1")    # one generation back, never unbounded
        except Exception:
            pass
        with open(p, "a", encoding="utf-8") as f:
            f.write(json.dumps({"at": int(time.time()), "verb": verb,
                                "seconds": round(seconds, 2),
                                "ok": bool(ok), "note": (note or "")[:120]}) + "\n")
    except Exception:
        pass


def verb_report(limit: int = 40) -> dict:
    """Aggregate the historical log: per verb count / p50 / p90 / max / failures,
    sorted by p90 descending - p90, not p50, is the column for hunting slow
    paths (p50 15s beside p90 400s is two code paths wearing one name)."""
    runs = {}
    try:
        with open(_runlog_path(), "r", encoding="utf-8") as f:
            for line in f:
                try:
                    r = json.loads(line)
                except Exception:
                    continue
                d = runs.setdefault(r.get("verb", "?"), {"secs": [], "fail": 0})
                d["secs"].append(float(r.get("seconds") or 0))
                if not r.get("ok"):
                    d["fail"] += 1
    except Exception:
        return {}
    out = {}
    for verb, d in runs.items():
        s = d["secs"]
        out[verb] = {"count": len(s), "p50": _pct(s, 0.5), "p90": _pct(s, 0.9),
                     "max": max(s) if s else None, "failures": d["fail"]}
    return dict(sorted(out.items(), key=lambda kv: -(kv[1]["p90"] or 0))[:limit])