Hung plugin hook silently discards prompts after prompt_async returns 204

"""
Concurrent prompt_async dispatches silently never start.

  POST /session                        -> create session
  POST /session/:id/prompt_async       -> fire-and-forget prompt (returns 204)
  GET  /session/:id/message            -> count persisted messages

A session that never starts ends with ZERO messages. A healthy one has >= 1.
"""
import argparse, json, sys, time, urllib.request, urllib.error, threading, collections

def call(base, method, path, payload=None, timeout=30):
    url = base.rstrip("/") + path
    data = json.dumps(payload).encode() if payload is not None else None
    req = urllib.request.Request(url, data=data, method=method,
                                 headers={"Content-Type": "application/json"})
    try:
        with urllib.request.urlopen(req, timeout=timeout) as r:
            body = r.read().decode("utf-8", "replace")
            return r.status, (json.loads(body) if body.strip() else None)
    except urllib.error.HTTPError as e:
        return e.code, e.read().decode("utf-8", "replace")[:200]
    except Exception as e:
        return -1, str(e)[:200]

def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--port", type=int, required=True)
    ap.add_argument("--host", default="127.0.0.1")
    ap.add_argument("--levels", default="1,2,3,4,6")
    ap.add_argument("--trials", type=int, default=3)
    ap.add_argument("--settle", type=int, default=25, help="seconds to wait before counting")
    ap.add_argument("--model", default=None)
    ap.add_argument("--agent", default=None)
    a = ap.parse_args()

    base = "http://%s:%d" % (a.host, a.port)
    st, _ = call(base, "GET", "/session?limit=1")
    if st not in (200, 204):
        sys.exit("Cannot reach OpenCode at %s (status=%s)" % (base, st))
    print("connected to %s\n" % base)

    levels = [int(x) for x in a.levels.split(",") if x.strip()]
    results = collections.OrderedDict()

    for n in levels:
        empties = total = 0
        for trial in range(a.trials):
            ids = []
            for i in range(n):
                st, body = call(base, "POST", "/session",
                                {"title": "repro-n%d-t%d-%d" % (n, trial, i)})
                if st in (200, 201) and isinstance(body, dict) and body.get("id"):
                    ids.append(body["id"])
                else:
                    print("  session create failed: status=%s body=%.120s" % (st, body))
            if len(ids) != n:
                print("  skipping trial (only %d/%d sessions)" % (len(ids), n)); continue

            barrier = threading.Barrier(n)
            statuses = {}
            def fire(sid):
                payload = {"parts": [{"type": "text", "text":
                            "Reply with exactly the word: ok"}]}
                if a.model:
                    prov, _, mid = a.model.partition("/")
                    payload["model"] = {"providerID": prov, "modelID": mid}
                if a.agent:
                    payload["agent"] = a.agent
                barrier.wait()
                statuses[sid] = call(base, "POST", "/session/%s/prompt_async" % sid, payload)[0]

            threads = [threading.Thread(target=fire, args=(s,)) for s in ids]
            t0 = time.time()
            for t in threads: t.start()
            for t in threads: t.join()
            spread = (time.time() - t0) * 1000

            time.sleep(a.settle)
            empty_ids = []
            for sid in ids:
                st, body = call(base, "GET", "/session/%s/message" % sid)
                cnt = len(body) if isinstance(body, list) else 0
                if cnt == 0: empty_ids.append(sid)
            empties += len(empty_ids); total += len(ids)
            print("  n=%-2d trial%-2d spread=%5.0fms  http=%s  EMPTY %d/%d %s"
                  % (n, trial + 1, spread,
                     ",".join(str(v) for v in statuses.values()),
                     len(empty_ids), len(ids),
                     "<-- REPRODUCED" if empty_ids else ""))
        results[n] = (empties, total)
        print()

    print("%-12s %8s %8s %8s" % ("CONCURRENCY", "EMPTY", "TOTAL", "RATE"))
    for n, (e, t) in results.items():
        print("%-12d %8d %8d %7.0f%%" % (n, e, t, (100.0 * e / t) if t else 0))

if __name__ == "__main__":
    main()