package_courier: fix sampler (tps_ewma field, run-lifetime sampling, w-mode) + report (x-domain clipping, t0 anchor, aggregate-tps/max-ctx tiles) — bug-2223

bench/package_courier/agent_app.py

@@ -160,6 +160,7 @@ class Runtime:
         self.target = cfg.workers
         self.live = 0  # threads actually running; idx is monotonic, never a count
         self.no_more_packages = threading.Event()
+        self.run_done = threading.Event()  # set after the last worker joins
         self.events: list[dict] = []
         self.workers: list[Worker] = []
 
@@ -383,18 +384,26 @@ def engine_props(url: str) -> dict:
 
 
 def sampler(rt: Runtime, out_path: Path):
+    # "w" not "a": a relaunch into the same --out dir must not keep stale
+    # samples from a crashed attempt. Runs until the last worker exits
+    # (run_done) — the queue empties while episodes are still decoding.
     http = httpx.Client(base_url=rt.cfg.engine, timeout=10)
-    with out_path.open("a") as f:
-        while not rt.no_more_packages.is_set():
+    with out_path.open("w") as f:
+        while not rt.run_done.is_set():
             row = {"t": round(time.time(), 3)}
             try:
                 r = http.get("/polykv/tps", params={"once": "1"})
                 if r.status_code == 200:
                     d = r.json()
                     row["mean_active_tps"] = d.get("mean_active_tps")
+                    # engine field is tps_ewma (recorded here as "tps"); keep
+                    # "active" so idle sessions' stale EWMA isn't aggregated
                     row["sessions"] = [
-                        {k: s.get(k) for k in
-                         ("session_id", "tps", "ctx_used", "ctx_headroom_tokens")}
+                        {"session_id": s.get("session_id"),
+                         "tps": s.get("tps_ewma"),
+                         "active": s.get("active"),
+                         "ctx_used": s.get("ctx_used"),
+                         "ctx_headroom_tokens": s.get("ctx_headroom_tokens")}
                         for s in d.get("sessions", [])]
             except Exception:
                 row["err"] = True
@@ -490,6 +499,7 @@ def main():
         rt.spawn_worker(schemas)
     for w in rt.workers:
         w.join()
+    rt.run_done.set()
 
     work = WorkClient(cfg.work)
     work.signal_done()

bench/package_courier/report_html.py

@@ -24,6 +24,11 @@ def svg_chart(series: dict[str, list[tuple[float, float]]], title: str,
               step_series: set[str] | None = None) -> str:
     pad_l, pad_r, pad_t, pad_b = 56, 16, 30, 40
     step_series = step_series or set()
+    if x_domain:
+        # clip to the shared clock — SVG polylines happily render outside the
+        # plot area, so stray points (e.g. stale samples) would smear the chart
+        series = {n: [p for p in s if x_domain[0] <= p[0] <= x_domain[1]]
+                  for n, s in series.items()}
     pts = [p for s in series.values() for p in s]
     if not pts:
         return f"<p>{title}: no data</p>"
@@ -132,7 +137,8 @@ def build(run_dir: Path) -> Path:
         x_domain=xdom, step_series={"delivered"}))
 
     charts.append(svg_chart(
-        {"delivery time": [(t, d) for t, d in episodes if d is not None]},
+        {"delivery time": [(0.0, 0.0)]
+         + [(t, d) for t, d in episodes if d is not None]},
         "Delivery time per package (does the run slow down?)",
         "completion run time (s)", "episode duration (s)", x_domain=xdom))
 
@@ -157,6 +163,31 @@ def build(run_dir: Path) -> Path:
             x_domain=xdom, step_series={"worker target", "live workers"}))
 
     dt = rep.get("delivery_time_s") or {}
+    # aggregate engine throughput per sample = sum of ACTIVE sessions' gen tps;
+    # window to the run clock so stale samples from a reused --out dir don't count
+    win_rows = [r for r in tps_rows if t0 <= r["t"] <= t_end]
+    agg = [sum(s.get("tps") or 0.0 for s in (r.get("sessions") or [])
+               if s.get("active", True))
+           for r in win_rows]
+    agg = [a for a in agg if a > 0]
+    if not agg:
+        # older runs sampled the wrong field (tps always null) — reconstruct
+        # from turn events: each turn generated out_toks at gen_tps, so it
+        # occupied [t - out/gen, t]; aggregate at time g = Σ gen_tps overlapping
+        turns = [(e["t"] - t0 - e["out_toks"] / e["gen_tps"], e["t"] - t0, e["gen_tps"])
+                 for e in events if e.get("ev") == "turn"
+                 and (e.get("gen_tps") or 0) > 0 and (e.get("out_toks") or 0) > 0]
+        grid = [g * 5.0 for g in range(int(T / 5.0) + 1)]
+        agg = [a for g in grid
+               if (a := sum(tps for b, e, tps in turns if b <= g <= e)) > 0]
+    agg.sort()
+    agg_avg = sum(agg) / len(agg) if agg else None
+    agg_p90 = agg[int(0.9 * (len(agg) - 1))] if agg else None
+    max_ctx = max((s.get("ctx_used") or 0 for r in win_rows
+                   for s in (r.get("sessions") or [])), default=0)
+    if not max_ctx:
+        max_ctx = max((e.get("in_toks") or 0 for e in events
+                       if e.get("ev") == "turn"), default=0)
     tiles = [
         ("Score", f"{rep.get('score_pct')}%"),
         ("Paid", f"${rep.get('paid_usd')}"),
@@ -168,6 +199,9 @@ def build(run_dir: Path) -> Path:
         ("Delivery mean / p90",
          " / ".join(f"{dt.get(k):.1f}s" if dt.get(k) is not None else "—"
                     for k in ("mean", "p90"))),
+        ("Aggregate gen tps avg / p90",
+         f"{agg_avg:.1f} / {agg_p90:.1f}" if agg else "—"),
+        ("Max ctx / session", f"{max_ctx:,} tok" if max_ctx else "—"),
         ("Tool errors", stats.get("tool_errors")),
         ("Workers", f"{stats.get('workers_initial', '?')} → {stats.get('workers_final')}"),
         ("tps floor", stats.get("tps_floor")),