KiCad - the KiCad Bridge

@@ -34,7 +34,7 @@ import time from pathlib import Path  PREFIX = "kicad"-_KEEP = 5          # samples retained per (phase, startKind)+_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.@@ -113,12 +113,21 @@ def estimate(phase: str, start_kind: str = "warm") -> tuple:     if len(samples) >= 3:         return _pct(samples, 0.5), "measured"     default = _DEFAULT_SECONDS.get(phase)-    if default is None:-        default = _DEFAULT_SECONDS.get("cold_start" if start_kind == "cold" else "warm_start", 10.0)     if samples:         # blend the shipped default with what little we have measured-        return round((sum(samples) / len(samples) + default) / 2, 2), "typical"-    return float(default), "unknown"+        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:@@ -150,13 +159,32 @@ def block(phase: str, step_label: str, started_at: float, percent=None,     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:-        # derive a coarse, MONOTONIC percent from elapsed vs estimate, capped-        # below 99 so it never claims completion it cannot see-        percent = int(min(95, (elapsed / est) * 95)) if est else 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}",@@ -168,6 +196,10 @@ def block(phase: str, step_label: str, started_at: float, percent=None,         "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}"@@ -299,17 +331,87 @@ def live(mine_only: bool = True, asker: str = None) -> list:             # 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.-            _elapsed = max(0.0, time.time() - cur["startedAt"])-            _est, _ = estimate(phase, cur.get("startKind", "warm"))-            _derived = int(min(95, (_elapsed / _est) * 95)) if _est else None-            _floor = cur.get("percent")-            _pct_now = max([v for v in (_derived, _floor) if v is not None], default=None)+            # 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=_pct_now, start_kind=cur.get("startKind", "warm"),+                      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 / "kicad-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])