// Hermetic tests for the multiplexed HTTP/2 client connection (h2_mux_conn.v):
// concurrent interleaved streams, flow-control blocking, GOAWAY mid-flight,
// per-stream cancellation that must not poison the connection, CONTINUATION
// assembly, and waking every waiter when the connection dies. The peer is a
// scripted in-process HTTP/2 server over an in-memory blocking pipe.
module http

import sync
import time

// MuxPipeBuf is a one-way in-memory FIFO with blocking, socket-like reads.
struct MuxPipeBuf {
mut:
    mu     &sync.Mutex = sync.new_mutex()
    data   []u8
    closed bool
}

fn (mut p MuxPipeBuf) write(buf []u8) !int {
    p.mu.lock()
    defer {
        p.mu.unlock()
    }
    if p.closed {
        return error('pipe: write to closed pipe')
    }
    p.data << buf
    return buf.len
}

fn (mut p MuxPipeBuf) read(mut buf []u8) !int {
    for {
        p.mu.lock()
        if p.data.len > 0 {
            mut n := p.data.len
            if n > buf.len {
                n = buf.len
            }
            for i in 0 .. n {
                buf[i] = p.data[i]
            }
            p.data = p.data[n..].clone()
            p.mu.unlock()
            return n
        }
        if p.closed {
            p.mu.unlock()
            return error('eof')
        }
        p.mu.unlock()
        time.sleep(time.millisecond)
    }
    return 0
}

fn (mut p MuxPipeBuf) close() {
    p.mu.lock()
    p.closed = true
    p.mu.unlock()
}

// MuxPipeEnd is one half of a bidirectional pipe.
@[heap]
struct MuxPipeEnd {
mut:
    incoming &MuxPipeBuf
    outgoing &MuxPipeBuf
}

fn (mut p MuxPipeEnd) read(mut buf []u8) !int {
    return p.incoming.read(mut buf)!
}

fn (mut p MuxPipeEnd) write(buf []u8) !int {
    return p.outgoing.write(buf)!
}

fn (mut p MuxPipeEnd) close_both() {
    p.incoming.close()
    p.outgoing.close()
}

fn new_mux_pipe() (&MuxPipeEnd, &MuxPipeEnd) {
    mut a := &MuxPipeBuf{}
    mut b := &MuxPipeBuf{}
    client := &MuxPipeEnd{
        incoming: b
        outgoing: a
    }
    server := &MuxPipeEnd{
        incoming: a
        outgoing: b
    }
    return client, server
}

// new_test_mux_conn builds a mux conn over the client pipe end with a real
// (non-nil) close_transport, satisfying new_h2_mux_conn's required-closer
// contract. The closer closes the in-memory pipe, mirroring how a real transport
// adapter would close its socket so teardown wakes the blocked reader.
fn new_test_mux_conn(mut cend MuxPipeEnd) &H2MuxConn {
    return new_h2_mux_conn(cend, fn [mut cend] () {
        cend.close_both()
    })
}

// MuxTestPeer is the scripted server side: it parses real frames off the pipe
// (with its own HPACK state) and records what it saw for the test to assert.
@[heap]
struct MuxTestPeer {
mut:
    end     &MuxPipeEnd
    rbuf    []u8
    decoder H2HpackDecoder
    encoder H2HpackEncoder
    mu      &sync.Mutex = sync.new_mutex()
    // stream id -> request :path, in arrival order
    paths               map[u32]string
    stream_ids          []u32
    failure             string
    data_total          map[u32]u64 // DATA bytes received per stream
    rst_streams         []u32
    conn_window_updates u64 // sum of WINDOW_UPDATE increments on stream 0
}

fn (mut p MuxTestPeer) fail(msg string) {
    p.mu.lock()
    if p.failure == '' {
        p.failure = msg
    }
    p.mu.unlock()
}

fn (mut p MuxTestPeer) failure_msg() string {
    p.mu.lock()
    defer {
        p.mu.unlock()
    }
    return p.failure
}

fn (mut p MuxTestPeer) read_exact(n int) ![]u8 {
    for p.rbuf.len < n {
        mut tmp := []u8{len: 4096}
        got := p.end.read(mut tmp)!
        if got <= 0 {
            return error('peer: pipe closed')
        }
        p.rbuf << tmp[..got]
    }
    out := p.rbuf[..n].clone()
    p.rbuf = p.rbuf[n..].clone()
    return out
}

fn (mut p MuxTestPeer) read_preface() ! {
    got := p.read_exact(h2_client_preface.len)!
    if got.bytestr() != h2_client_preface {
        return error('peer: bad connection preface')
    }
}

fn (mut p MuxTestPeer) next_frame() !H2Frame {
    head := p.read_exact(h2_frame_header_len)!
    header := h2_parse_frame_header(head)!
    payload := p.read_exact(int(header.length))!
    return h2_parse_frame(header, payload)!
}

fn (mut p MuxTestPeer) write_frame(f H2Frame) ! {
    p.end.write(f.encode())!
}

// pump consumes one client frame, updating the peer's records. HEADERS blocks
// are decoded (tracking :path); WINDOW_UPDATE / SETTINGS / DATA / RST are
// tallied. Returns the frame.
fn (mut p MuxTestPeer) pump() !H2Frame {
    f := p.next_frame()!
    match f {
        H2SettingsFrame {
            if !f.ack {
                p.write_frame(H2SettingsFrame{
                    ack: true
                })!
            }
        }
        H2HeadersFrame {
            mut fragment := f.fragment.clone()
            if !f.end_headers {
                for {
                    cont := p.next_frame()!
                    if cont is H2ContinuationFrame {
                        fragment << cont.fragment
                        if cont.end_headers {
                            break
                        }
                    } else {
                        return error('peer: expected CONTINUATION')
                    }
                }
            }
            fields := p.decoder.decode(fragment)!
            mut path := ''
            for fld in fields {
                if fld.name == ':path' {
                    path = fld.value
                }
            }
            p.mu.lock()
            p.paths[f.stream_id] = path
            p.stream_ids << f.stream_id
            p.mu.unlock()
        }
        H2DataFrame {
            p.mu.lock()
            p.data_total[f.stream_id] = (p.data_total[f.stream_id] or { u64(0) }) + u64(f.data.len)
            p.mu.unlock()
        }
        H2RstStreamFrame {
            p.mu.lock()
            p.rst_streams << f.stream_id
            p.mu.unlock()
        }
        H2WindowUpdateFrame {
            if f.stream_id == 0 {
                p.mu.lock()
                p.conn_window_updates += u64(f.window_size_increment)
                p.mu.unlock()
            }
        }
        else {}
    }

    return f
}

// wait_for_headers pumps until `n` request header blocks have arrived,
// returning the stream ids in arrival order.
fn (mut p MuxTestPeer) wait_for_headers(n int) ![]u32 {
    for {
        p.mu.lock()
        if p.stream_ids.len >= n {
            ids := p.stream_ids.clone()
            p.mu.unlock()
            return ids
        }
        p.mu.unlock()
        p.pump()!
    }
    return []u32{}
}

// respond_headers sends a 200 response header block for `stream_id`.
fn (mut p MuxTestPeer) respond_headers(stream_id u32, end_stream bool) ! {
    block := p.encoder.encode([H2HeaderField{':status', '200'},
        H2HeaderField{'content-type', 'text/plain'}])
    p.write_frame(H2HeadersFrame{
        stream_id:   stream_id
        fragment:    block
        end_headers: true
        end_stream:  end_stream
    })!
}

// --- worker plumbing ----------------------------------------------------------

@[heap]
struct MuxResults {
mut:
    mu       &sync.Mutex = sync.new_mutex()
    bodies   map[string]string
    statuses map[string]int
    errs     map[string]string
    codes    map[string]int
}

fn (mut r MuxResults) set_ok(path string, resp H2ClientResponse) {
    r.mu.lock()
    r.bodies[path] = resp.body.bytestr()
    r.statuses[path] = resp.status
    r.mu.unlock()
}

fn (mut r MuxResults) set_err(path string, e IError) {
    r.mu.lock()
    r.errs[path] = e.msg()
    r.codes[path] = e.code()
    r.mu.unlock()
}

fn mux_worker(mut c H2MuxConn, req H2ClientRequest, mut out MuxResults) {
    resp := c.do(req) or {
        out.set_err(req.path, err)
        return
    }
    out.set_ok(req.path, resp)
}

// --- tests --------------------------------------------------------------------

// Three concurrent GETs on one connection; the peer interleaves their DATA
// frames. Each requester must receive exactly its own body.
fn test_mux_concurrent_interleaved_streams() {
    mut cend, mut pend := new_mux_pipe()
    mut conn := new_test_mux_conn(mut cend)
    mut peer := &MuxTestPeer{
        end: pend
    }
    mut out := &MuxResults{}
    mut workers := []thread{}
    for i in 0 .. 3 {
        workers << spawn mux_worker(mut conn, H2ClientRequest{ authority: 't', path: '/w${i}' }, mut out)
    }
    peer_thread := spawn fn (mut peer MuxTestPeer) {
        peer.read_preface() or {
            peer.fail('preface: ${err.msg()}')
            return
        }
        ids := peer.wait_for_headers(3) or {
            peer.fail('headers: ${err.msg()}')
            return
        }
        for id in ids {
            peer.respond_headers(id, false) or {
                peer.fail('respond: ${err.msg()}')
                return
            }
        }
        // Interleave the bodies: one chunk per stream per round.
        for round in 0 .. 2 {
            for id in ids {
                peer.mu.lock()
                path := peer.paths[id] or { '' }
                peer.mu.unlock()
                peer.write_frame(H2DataFrame{
                    stream_id:  id
                    data:       '${path}#${round};'.bytes()
                    end_stream: round == 1
                }) or {
                    peer.fail('data: ${err.msg()}')
                    return
                }
            }
        }
    }(mut peer)
    workers.wait()
    peer_thread.wait()
    assert peer.failure_msg() == ''
    assert out.errs.len == 0, 'worker errors: ${out.errs}'
    for i in 0 .. 3 {
        assert out.statuses['/w${i}'] == 200
        assert out.bodies['/w${i}'] == '/w${i}#0;/w${i}#1;'
    }
    cend.close_both()
}

// A large request body must block on the 65535-byte connection/stream send
// windows and resume when the peer grants WINDOW_UPDATEs.
fn test_mux_flow_control_blocks_and_resumes() {
    mut cend, mut pend := new_mux_pipe()
    mut conn := new_test_mux_conn(mut cend)
    mut peer := &MuxTestPeer{
        end: pend
    }
    mut out := &MuxResults{}
    body_len := 100000
    mut workers := []thread{}
    workers << spawn mux_worker(mut conn, H2ClientRequest{
        method:    'POST'
        authority: 't'
        path:      '/up'
        body:      []u8{len: body_len, init: `B`}
    }, mut out)
    peer_thread := spawn fn [body_len] (mut peer MuxTestPeer) {
        peer.read_preface() or {
            peer.fail('preface: ${err.msg()}')
            return
        }
        ids := peer.wait_for_headers(1) or {
            peer.fail('headers: ${err.msg()}')
            return
        }
        id := ids[0]
        // Drain DATA until the client exhausts the 65535-byte windows.
        for {
            peer.mu.lock()
            got := peer.data_total[id] or { u64(0) }
            peer.mu.unlock()
            if got >= u64(h2_default_initial_window) {
                break
            }
            peer.pump() or {
                peer.fail('pump: ${err.msg()}')
                return
            }
        }
        peer.mu.lock()
        at_block := peer.data_total[id] or { u64(0) }
        peer.mu.unlock()
        if at_block != u64(h2_default_initial_window) {
            peer.fail('expected the client to stop at exactly 65535 sent bytes, got ${at_block}')
            return
        }
        // Grant room on both windows; the client must finish the body.
        peer.write_frame(H2WindowUpdateFrame{
            stream_id:             0
            window_size_increment: u32(body_len)
        }) or {
            peer.fail('wu0: ${err.msg()}')
            return
        }
        peer.write_frame(H2WindowUpdateFrame{
            stream_id:             id
            window_size_increment: u32(body_len)
        }) or {
            peer.fail('wu: ${err.msg()}')
            return
        }
        for {
            peer.mu.lock()
            got := peer.data_total[id] or { u64(0) }
            peer.mu.unlock()
            if got >= u64(body_len) {
                break
            }
            peer.pump() or {
                peer.fail('pump2: ${err.msg()}')
                return
            }
        }
        peer.respond_headers(id, true) or {
            peer.fail('respond: ${err.msg()}')
            return
        }
    }(mut peer)
    workers.wait()
    peer_thread.wait()
    assert peer.failure_msg() == ''
    assert out.errs.len == 0, 'worker errors: ${out.errs}'
    assert out.statuses['/up'] == 200
    cend.close_both()
}

// A peer that RST_STREAMs an upload after the client has exhausted its send
// windows must wake the body sender blocked on flow-control credit, so do()
// returns the reset error instead of hanging forever.
fn test_mux_rst_wakes_blocked_body_sender() {
    mut cend, mut pend := new_mux_pipe()
    mut conn := new_test_mux_conn(mut cend)
    mut peer := &MuxTestPeer{
        end: pend
    }
    mut out := &MuxResults{}
    body_len := 100000
    mut workers := []thread{}
    workers << spawn mux_worker(mut conn, H2ClientRequest{
        method:    'POST'
        authority: 't'
        path:      '/up'
        body:      []u8{len: body_len, init: `B`}
    }, mut out)
    peer_thread := spawn fn (mut peer MuxTestPeer) {
        peer.read_preface() or {
            peer.fail('preface: ${err.msg()}')
            return
        }
        ids := peer.wait_for_headers(1) or {
            peer.fail('headers: ${err.msg()}')
            return
        }
        id := ids[0]
        // Let the client send until both windows are exhausted; it then blocks
        // in send_body_on_stream waiting for a WINDOW_UPDATE that never comes.
        for {
            peer.mu.lock()
            got := peer.data_total[id] or { u64(0) }
            peer.mu.unlock()
            if got >= u64(h2_default_initial_window) {
                break
            }
            peer.pump() or {
                peer.fail('pump: ${err.msg()}')
                return
            }
        }
        // Reset the stream instead of granting credit.
        peer.write_frame(H2RstStreamFrame{
            stream_id:  id
            error_code: u32(H2ErrorCode.cancel)
        }) or {
            peer.fail('rst: ${err.msg()}')
            return
        }
    }(mut peer)
    workers.wait()
    peer_thread.wait()
    assert peer.failure_msg() == ''
    upload_err := out.errs['/up'] or { '' }
    assert upload_err != '', 'expected the reset upload to fail, not hang'
    assert upload_err.contains('reset'), 'expected a reset error, got: ${upload_err}'
    cend.close_both()
}

// An early final response that carries a body ends with END_STREAM on the DATA
// frame, not on HEADERS. A body sender blocked on flow control must still be
// woken (by on_response_data), abandon the upload with RST_STREAM, and have the
// response delivered — not hang waiting for a WINDOW_UPDATE that never comes.
fn test_mux_early_final_response_with_body_wakes_blocked_upload() {
    mut cend, mut pend := new_mux_pipe()
    mut conn := new_test_mux_conn(mut cend)
    mut peer := &MuxTestPeer{
        end: pend
    }
    mut out := &MuxResults{}
    body_len := 100000
    mut workers := []thread{}
    workers << spawn mux_worker(mut conn, H2ClientRequest{
        method:    'POST'
        authority: 't'
        path:      '/up'
        body:      []u8{len: body_len, init: `B`}
    }, mut out)
    peer_thread := spawn fn (mut peer MuxTestPeer) {
        peer.read_preface() or {
            peer.fail('preface: ${err.msg()}')
            return
        }
        ids := peer.wait_for_headers(1) or {
            peer.fail('headers: ${err.msg()}')
            return
        }
        id := ids[0]
        // Let the client exhaust both send windows; it then blocks in
        // send_body_on_stream waiting for a WINDOW_UPDATE that never comes.
        for {
            peer.mu.lock()
            got := peer.data_total[id] or { u64(0) }
            peer.mu.unlock()
            if got >= u64(h2_default_initial_window) {
                break
            }
            peer.pump() or {
                peer.fail('pump: ${err.msg()}')
                return
            }
        }
        // Send a complete final response WITH a body, so END_STREAM lands on the
        // DATA frame rather than HEADERS. No WINDOW_UPDATE is ever granted.
        block := peer.encoder.encode([H2HeaderField{':status', '413'},
            H2HeaderField{'content-type', 'text/plain'}])
        peer.write_frame(H2HeadersFrame{
            stream_id:   id
            fragment:    block
            end_headers: true
            end_stream:  false
        }) or {
            peer.fail('resp headers: ${err.msg()}')
            return
        }
        peer.write_frame(H2DataFrame{
            stream_id:  id
            data:       'rejected'.bytes()
            end_stream: true
        }) or {
            peer.fail('resp data: ${err.msg()}')
            return
        }
        // The client must wake, abandon the upload, and close its half with
        // RST_STREAM; pump until we observe it (or the pipe closes).
        for {
            f := peer.pump() or { return }
            if f is H2RstStreamFrame {
                return
            }
        }
    }(mut peer)
    workers.wait()
    peer_thread.wait()
    assert peer.failure_msg() == ''
    upload_err := out.errs['/up'] or { '' }
    assert upload_err == '', 'early final response must not error the request, got: ${upload_err}'
    assert out.statuses['/up'] or { 0 } == 413, 'expected the 413 response to be delivered'
    assert out.bodies['/up'] or { '' } == 'rejected', 'expected the response body to be delivered'
    peer.mu.lock()
    saw_rst := u32(1) in peer.rst_streams
    peer.mu.unlock()
    assert saw_rst, 'client must RST_STREAM to close its abandoned upload half'
    cend.close_both()
}

// RFC 9113 §8.3.1: a response HEADERS block without a valid :status is malformed.
// The client must reset the stream (PROTOCOL_ERROR) and surface an error to the
// requester rather than treating it as a 1xx interim and waiting forever.
fn test_mux_response_missing_status_resets_stream() {
    mut cend, mut pend := new_mux_pipe()
    mut conn := new_test_mux_conn(mut cend)
    mut peer := &MuxTestPeer{
        end: pend
    }
    mut out := &MuxResults{}
    mut workers := []thread{}
    workers << spawn mux_worker(mut conn, H2ClientRequest{
        method:    'GET'
        authority: 't'
        path:      '/get'
    }, mut out)
    peer_thread := spawn fn (mut peer MuxTestPeer) {
        peer.read_preface() or {
            peer.fail('preface: ${err.msg()}')
            return
        }
        ids := peer.wait_for_headers(1) or {
            peer.fail('headers: ${err.msg()}')
            return
        }
        id := ids[0]
        // A response header block with no :status pseudo-header (malformed); the
        // stream is left open so the client cannot fall back to "stream closed".
        block := peer.encoder.encode([H2HeaderField{'content-type', 'text/plain'}])
        peer.write_frame(H2HeadersFrame{
            stream_id:   id
            fragment:    block
            end_headers: true
            end_stream:  false
        }) or {
            peer.fail('resp headers: ${err.msg()}')
            return
        }
        // The client must reset the stream; pump until we observe the RST.
        for {
            f := peer.pump() or { return }
            if f is H2RstStreamFrame {
                return
            }
        }
    }(mut peer)
    workers.wait()
    peer_thread.wait()
    assert peer.failure_msg() == ''
    get_err := out.errs['/get'] or { '' }
    assert get_err != '', 'expected an error for a response missing :status, not a hang'
    assert get_err.contains('status'), 'expected a :status protocol error, got: ${get_err}'
    peer.mu.lock()
    saw_rst := u32(1) in peer.rst_streams
    peer.mu.unlock()
    assert saw_rst, 'client must RST_STREAM a malformed (no :status) response'
    cend.close_both()
}

// string.int() is lenient ('200 OK' -> 200), so a non-three-digit :status must
// be rejected by an explicit digit check rather than accepted as a 200 success.
fn test_mux_response_malformed_status_resets_stream() {
    mut cend, mut pend := new_mux_pipe()
    mut conn := new_test_mux_conn(mut cend)
    mut peer := &MuxTestPeer{
        end: pend
    }
    mut out := &MuxResults{}
    mut workers := []thread{}
    workers << spawn mux_worker(mut conn, H2ClientRequest{
        method:    'GET'
        authority: 't'
        path:      '/get'
    }, mut out)
    peer_thread := spawn fn (mut peer MuxTestPeer) {
        peer.read_preface() or {
            peer.fail('preface: ${err.msg()}')
            return
        }
        ids := peer.wait_for_headers(1) or {
            peer.fail('headers: ${err.msg()}')
            return
        }
        id := ids[0]
        // A :status that int() would parse leniently to 200, but is malformed.
        block := peer.encoder.encode([H2HeaderField{':status', '200 OK'},
            H2HeaderField{'content-type', 'text/plain'}])
        peer.write_frame(H2HeadersFrame{
            stream_id:   id
            fragment:    block
            end_headers: true
            end_stream:  false
        }) or {
            peer.fail('resp headers: ${err.msg()}')
            return
        }
        for {
            f := peer.pump() or { return }
            if f is H2RstStreamFrame {
                return
            }
        }
    }(mut peer)
    workers.wait()
    peer_thread.wait()
    assert peer.failure_msg() == ''
    get_err := out.errs['/get'] or { '' }
    assert get_err != '', 'expected an error for a malformed :status, not a success'
    assert get_err.contains('status'), 'expected a :status protocol error, got: ${get_err}'
    delivered := out.statuses['/get'] or { -1 }
    assert delivered == -1, 'a malformed :status must not be delivered as a success (got ${delivered})'
    cend.close_both()
}

// A stream reset with REFUSED_STREAM means the server did not process the
// request, so it must surface a retryable error (RFC 7540 8.1.4) — even for a
// non-idempotent method — so the pool can replay it on a fresh connection. A
// reset with any other code (e.g. CANCEL) must stay non-retryable.
fn test_mux_refused_stream_reset_is_retryable() {
    mut cend, mut pend := new_mux_pipe()
    mut conn := new_test_mux_conn(mut cend)
    mut peer := &MuxTestPeer{
        end: pend
    }
    mut out := &MuxResults{}
    mut workers := []thread{}
    workers << spawn mux_worker(mut conn, H2ClientRequest{
        method:    'POST'
        authority: 't'
        path:      '/refused'
        body:      'x'.bytes()
    }, mut out)
    peer_thread := spawn fn (mut peer MuxTestPeer) {
        peer.read_preface() or {
            peer.fail('preface: ${err.msg()}')
            return
        }
        ids := peer.wait_for_headers(1) or {
            peer.fail('headers: ${err.msg()}')
            return
        }
        peer.write_frame(H2RstStreamFrame{
            stream_id:  ids[0]
            error_code: u32(H2ErrorCode.refused_stream)
        }) or {
            peer.fail('rst: ${err.msg()}')
            return
        }
    }(mut peer)
    workers.wait()
    peer_thread.wait()
    assert peer.failure_msg() == ''
    assert out.errs['/refused'] or { '' } != '', 'expected the refused POST to fail'
    assert out.codes['/refused'] or { 0 } == h2_err_retryable_code, 'REFUSED_STREAM reset must be retryable, got code ${out.codes['/refused']}'
    cend.close_both()
}

// Padding in a DATA frame counts toward flow control (RFC 7540 6.9.1) even
// though it is never delivered to the app. The client must credit the full
// received payload (pad-length byte + data + padding) back via WINDOW_UPDATE,
// or the connection receive window leaks and eventually stalls.
fn test_mux_padded_data_credits_full_flow_control() {
    mut cend, mut pend := new_mux_pipe()
    mut conn := new_test_mux_conn(mut cend)
    mut peer := &MuxTestPeer{
        end: pend
    }
    mut out := &MuxResults{}
    data := 'hi'.bytes()
    pad := []u8{len: 100}
    payload_len := 1 + data.len + pad.len // pad-length byte + data + padding
    mut workers := []thread{}
    workers << spawn mux_worker(mut conn, H2ClientRequest{ authority: 't', path: '/p' }, mut out)
    peer_thread := spawn fn [data, pad, payload_len] (mut peer MuxTestPeer) {
        peer.read_preface() or {
            peer.fail('preface: ${err.msg()}')
            return
        }
        ids := peer.wait_for_headers(1) or {
            peer.fail('headers: ${err.msg()}')
            return
        }
        peer.respond_headers(ids[0], false) or {
            peer.fail('respond: ${err.msg()}')
            return
        }
        mut payload := [u8(pad.len)]
        payload << data
        payload << pad
        raw := h2_frame_bytes(h2_frame_data, h2_flag_padded | h2_flag_end_stream, ids[0], payload)
        peer.end.write(raw) or {
            peer.fail('data: ${err.msg()}')
            return
        }
        // Pump until the connection-level WINDOW_UPDATEs sum to the full padded
        // payload; with the fix this terminates (data + padding both credited).
        for {
            peer.mu.lock()
            got := peer.conn_window_updates
            peer.mu.unlock()
            if got >= u64(payload_len) {
                break
            }
            peer.pump() or {
                peer.fail('pump: ${err.msg()}')
                return
            }
        }
    }(mut peer)
    workers.wait()
    peer_thread.wait()
    assert peer.failure_msg() == ''
    assert out.statuses['/p'] == 200
    assert out.bodies['/p'] == 'hi'
    assert peer.conn_window_updates == u64(payload_len), 'padding must count toward connection flow control'
    cend.close_both()
}

// A received PUSH_PROMISE must fail the connection: we advertise ENABLE_PUSH=0,
// so it is a PROTOCOL_ERROR (RFC 7540 6.6 / 8.2). The peer also sends a valid
// response — with the guard the request fails; without it the push is ignored
// and the request would wrongly succeed.
fn test_mux_push_promise_fails_connection() {
    mut cend, mut pend := new_mux_pipe()
    mut conn := new_test_mux_conn(mut cend)
    mut peer := &MuxTestPeer{
        end: pend
    }
    mut out := &MuxResults{}
    mut workers := []thread{}
    workers << spawn mux_worker(mut conn, H2ClientRequest{ authority: 't', path: '/pp' }, mut out)
    peer_thread := spawn fn (mut peer MuxTestPeer) {
        peer.read_preface() or {
            peer.fail('preface: ${err.msg()}')
            return
        }
        ids := peer.wait_for_headers(1) or {
            peer.fail('headers: ${err.msg()}')
            return
        }
        block := peer.encoder.encode([H2HeaderField{':status', '200'}])
        peer.write_frame(H2PushPromiseFrame{
            stream_id:          ids[0]
            promised_stream_id: 2
            fragment:           block
            end_headers:        true
        }) or {
            peer.fail('push: ${err.msg()}')
            return
        }
        peer.respond_headers(ids[0], true) or {}
    }(mut peer)
    workers.wait()
    peer_thread.wait()
    assert peer.failure_msg() == ''
    assert out.statuses['/pp'] or { 0 } != 200, 'a PUSH_PROMISE must fail the connection, not be ignored'
    assert out.errs['/pp'] or { '' } != ''
    cend.close_both()
}

// An invalid SETTINGS_ENABLE_PUSH value (not 0/1) must fail the connection
// (RFC 7540 6.5.2). The peer sends ENABLE_PUSH=2 then a valid response; the
// request must fail rather than succeed.
fn test_mux_invalid_enable_push_fails_connection() {
    mut cend, mut pend := new_mux_pipe()
    mut conn := new_test_mux_conn(mut cend)
    mut peer := &MuxTestPeer{
        end: pend
    }
    mut out := &MuxResults{}
    mut workers := []thread{}
    workers << spawn mux_worker(mut conn, H2ClientRequest{ authority: 't', path: '/ep' }, mut out)
    peer_thread := spawn fn (mut peer MuxTestPeer) {
        peer.read_preface() or {
            peer.fail('preface: ${err.msg()}')
            return
        }
        peer.write_frame(H2SettingsFrame{
            settings: [
                H2Setting{
                    id:    h2_settings_enable_push
                    value: 2
                },
            ]
        }) or {
            peer.fail('settings: ${err.msg()}')
            return
        }
        ids := peer.wait_for_headers(1) or {
            peer.fail('headers: ${err.msg()}')
            return
        }
        peer.respond_headers(ids[0], true) or {}
    }(mut peer)
    workers.wait()
    peer_thread.wait()
    assert peer.failure_msg() == ''
    assert out.statuses['/ep'] or { 0 } != 200, 'an invalid ENABLE_PUSH must fail the connection'
    assert out.errs['/ep'] or { '' } != ''
    cend.close_both()
}

// A stream-level WINDOW_UPDATE with a zero increment is a stream error (RFC
// 7540 6.9): that one stream must be RST and failed, but the connection and
// other concurrent streams must survive.
fn test_mux_zero_window_update_resets_only_that_stream() {
    mut cend, mut pend := new_mux_pipe()
    mut conn := new_test_mux_conn(mut cend)
    mut peer := &MuxTestPeer{
        end: pend
    }
    mut out := &MuxResults{}
    mut wa := []thread{}
    wa << spawn mux_worker(mut conn, H2ClientRequest{ authority: 't', path: '/a' }, mut out)
    peer_thread := spawn fn (mut peer MuxTestPeer) {
        peer.read_preface() or {
            peer.fail('preface: ${err.msg()}')
            return
        }
        ids := peer.wait_for_headers(2) or {
            peer.fail('headers: ${err.msg()}')
            return
        }
        mut a := u32(0)
        mut b := u32(0)
        peer.mu.lock()
        for id in ids {
            if peer.paths[id] or { '' } == '/a' {
                a = id
            } else {
                b = id
            }
        }
        peer.mu.unlock()
        // Illegal zero-increment WINDOW_UPDATE on stream A: A must be reset.
        peer.write_frame(H2WindowUpdateFrame{
            stream_id:             a
            window_size_increment: 0
        }) or {
            peer.fail('wu: ${err.msg()}')
            return
        }
        // B is served to completion; the connection must remain usable.
        peer.respond_headers(b, false) or {
            peer.fail('respond b: ${err.msg()}')
            return
        }
        peer.write_frame(H2DataFrame{
            stream_id:  b
            data:       'b-ok'.bytes()
            end_stream: true
        }) or {
            peer.fail('data b: ${err.msg()}')
            return
        }
    }(mut peer)
    time.sleep(50 * time.millisecond)
    mut wb := []thread{}
    wb << spawn mux_worker(mut conn, H2ClientRequest{ authority: 't', path: '/b' }, mut out)
    wa.wait()
    wb.wait()
    peer_thread.wait()
    assert peer.failure_msg() == ''
    assert out.errs['/a'] or { '' } != '', 'the zero-increment stream must fail'
    assert out.statuses['/b'] or { 0 } == 200, 'the other stream and the connection must survive'
    assert out.bodies['/b'] == 'b-ok'
    cend.close_both()
}

// GOAWAY with last_stream_id between two active streams: the lower id
// completes, the higher fails with a retryable error, and new requests are
// refused (also retryable).
fn test_mux_goaway_mid_flight() {
    mut cend, mut pend := new_mux_pipe()
    mut conn := new_test_mux_conn(mut cend)
    mut peer := &MuxTestPeer{
        end: pend
    }
    mut out := &MuxResults{}
    mut w1 := []thread{}
    w1 << spawn mux_worker(mut conn, H2ClientRequest{ authority: 't', path: '/g0' }, mut out)
    peer_thread := spawn fn (mut peer MuxTestPeer) {
        peer.read_preface() or {
            peer.fail('preface: ${err.msg()}')
            return
        }
        // First stream arrives, then wait for the second before answering, so
        // both are unambiguously in flight when GOAWAY is sent.
        ids := peer.wait_for_headers(2) or {
            peer.fail('headers: ${err.msg()}')
            return
        }
        mut sorted := ids.clone()
        sorted.sort()
        low := sorted[0]
        high := sorted[1]
        peer.write_frame(H2GoawayFrame{
            last_stream_id: low
            error_code:     u32(H2ErrorCode.no_error)
        }) or {
            peer.fail('goaway: ${err.msg()}')
            return
        }
        // The processed stream still completes.
        peer.respond_headers(low, false) or {
            peer.fail('respond: ${err.msg()}')
            return
        }
        peer.write_frame(H2DataFrame{
            stream_id:  low
            data:       'survivor'.bytes()
            end_stream: true
        }) or {
            peer.fail('data: ${err.msg()}')
            return
        }
        _ = high
    }(mut peer)
    // Make sure the first stream is registered before starting the second, so
    // the id order is deterministic.
    time.sleep(50 * time.millisecond)
    mut w2 := []thread{}
    w2 << spawn mux_worker(mut conn, H2ClientRequest{ authority: 't', path: '/g1' }, mut out)
    w1.wait()
    w2.wait()
    peer_thread.wait()
    assert peer.failure_msg() == ''
    assert out.statuses['/g0'] == 200
    assert out.bodies['/g0'] == 'survivor'
    assert out.errs['/g1'] or { '' } != ''
    assert out.codes['/g1'] or { 0 } == h2_err_retryable_code
    // New requests on a GOAWAYed connection are refused as retryable.
    conn.do(H2ClientRequest{ authority: 't', path: '/late' }) or {
        assert err.code() == h2_err_retryable_code
        cend.close_both()
        return
    }
    assert false, 'a request on a GOAWAYed connection must fail'
}

// Cancelling one stream (stop_receiving_limit) must not poison the
// connection: the peer's late DATA for the cancelled stream is absorbed, and
// a second stream completes afterwards.
fn test_mux_cancel_one_stream_other_lives() {
    mut cend, mut pend := new_mux_pipe()
    mut conn := new_test_mux_conn(mut cend)
    mut peer := &MuxTestPeer{
        end: pend
    }
    mut out := &MuxResults{}
    mut wa := []thread{}
    wa << spawn mux_worker(mut conn, H2ClientRequest{
        authority:            't'
        path:                 '/cancelme'
        stop_receiving_limit: 5
    }, mut out)
    peer_thread := spawn fn (mut peer MuxTestPeer) {
        peer.read_preface() or {
            peer.fail('preface: ${err.msg()}')
            return
        }
        ids := peer.wait_for_headers(2) or {
            peer.fail('headers: ${err.msg()}')
            return
        }
        mut a := u32(0)
        mut b := u32(0)
        peer.mu.lock()
        for id in ids {
            if peer.paths[id] or { '' } == '/cancelme' {
                a = id
            } else {
                b = id
            }
        }
        peer.mu.unlock()
        // Over-deliver on stream A so the client cancels it.
        peer.respond_headers(a, false) or {
            peer.fail('respond a: ${err.msg()}')
            return
        }
        peer.write_frame(H2DataFrame{
            stream_id: a
            data:      'aaaaaaaaaa'.bytes()
        }) or {
            peer.fail('data a: ${err.msg()}')
            return
        }
        // Wait for the client's RST_STREAM(A).
        for {
            peer.mu.lock()
            rst := a in peer.rst_streams
            peer.mu.unlock()
            if rst {
                break
            }
            peer.pump() or {
                peer.fail('pump: ${err.msg()}')
                return
            }
        }
        // Late DATA for the cancelled stream: the client must absorb it and
        // keep the connection healthy.
        peer.write_frame(H2DataFrame{
            stream_id: a
            data:      'late-data!'.bytes()
        }) or {
            peer.fail('late data: ${err.msg()}')
            return
        }
        // Now serve stream B to completion.
        peer.respond_headers(b, false) or {
            peer.fail('respond b: ${err.msg()}')
            return
        }
        peer.write_frame(H2DataFrame{
            stream_id:  b
            data:       'b-survives'.bytes()
            end_stream: true
        }) or {
            peer.fail('data b: ${err.msg()}')
            return
        }
    }(mut peer)
    time.sleep(50 * time.millisecond)
    mut wb := []thread{}
    wb << spawn mux_worker(mut conn, H2ClientRequest{ authority: 't', path: '/lives' }, mut out)
    wa.wait()
    wb.wait()
    peer_thread.wait()
    assert peer.failure_msg() == ''
    // The cancelled stream returns its truncated body without an error.
    assert out.errs['/cancelme'] or { '' } == ''
    assert out.bodies['/cancelme'] == 'aaaaaaaaaa'
    // The other stream is unaffected — the connection was not poisoned. (This
    // exercises the deregister-before-RST ordering in cancel_stream: the late
    // DATA for the cancelled stream is absorbed by the unknown-stream backstop,
    // which credits the connection window so other streams keep flowing.)
    assert out.errs['/lives'] or { '' } == ''
    assert out.bodies['/lives'] == 'b-survives'
    cend.close_both()
}

// Cancelling a stream (stop_receiving_limit) must credit the connection window
// for ALL DATA received on it — including chunks queued before the cancel that
// are discarded undrained — or the connection window leaks on every early
// cancellation. The peer sends three 10-byte chunks; whatever the client does
// not deliver it must still credit back, so the connection WINDOW_UPDATEs sum
// to the full 30 bytes sent.
fn test_mux_cancel_credits_all_received_data() {
    mut cend, mut pend := new_mux_pipe()
    mut conn := new_test_mux_conn(mut cend)
    mut peer := &MuxTestPeer{
        end: pend
    }
    mut out := &MuxResults{}
    total := 30
    mut workers := []thread{}
    workers << spawn mux_worker(mut conn, H2ClientRequest{
        authority:            't'
        path:                 '/c'
        stop_receiving_limit: 5
    }, mut out)
    peer_thread := spawn fn [total] (mut peer MuxTestPeer) {
        peer.read_preface() or {
            peer.fail('preface: ${err.msg()}')
            return
        }
        ids := peer.wait_for_headers(1) or {
            peer.fail('headers: ${err.msg()}')
            return
        }
        peer.respond_headers(ids[0], false) or {
            peer.fail('respond: ${err.msg()}')
            return
        }
        for _ in 0 .. 3 {
            peer.write_frame(H2DataFrame{
                stream_id: ids[0]
                data:      'xxxxxxxxxx'.bytes()
            }) or {
                peer.fail('data: ${err.msg()}')
                return
            }
        }
        // Pump until every sent DATA byte has been credited back on the
        // connection window; with the fix this terminates regardless of how the
        // chunks raced against the cancel (drained, queued-then-discarded, or
        // late via the backstop).
        for {
            peer.mu.lock()
            got := peer.conn_window_updates
            peer.mu.unlock()
            if got >= u64(total) {
                break
            }
            peer.pump() or {
                peer.fail('pump: ${err.msg()}')
                return
            }
        }
    }(mut peer)
    workers.wait()
    peer_thread.wait()
    assert peer.failure_msg() == ''
    assert peer.conn_window_updates == u64(total), 'cancel must credit every received DATA byte'
    cend.close_both()
}

// A peer SETTINGS_HEADER_TABLE_SIZE only bounds the dynamic table our encoder
// may use for request headers; it must not shrink the table we advertised for
// decoding the server's responses. Applying it to the response decoder would
// break valid dynamic references in later responses, so the decoder limit must
// stay at its advertised default.
fn test_mux_peer_header_table_size_keeps_decoder_limit() {
    mut cend, mut pend := new_mux_pipe()
    mut conn := new_test_mux_conn(mut cend)
    mut peer := &MuxTestPeer{
        end: pend
    }
    mut out := &MuxResults{}
    mut workers := []thread{}
    workers << spawn mux_worker(mut conn, H2ClientRequest{ authority: 't', path: '/h' }, mut out)
    peer_thread := spawn fn (mut peer MuxTestPeer) {
        peer.read_preface() or {
            peer.fail('preface: ${err.msg()}')
            return
        }
        // The peer limits the table it uses to DECODE our request headers to 0.
        peer.write_frame(H2SettingsFrame{
            settings: [
                H2Setting{
                    id:    h2_settings_header_table_size
                    value: 0
                },
            ]
        }) or {
            peer.fail('settings: ${err.msg()}')
            return
        }
        ids := peer.wait_for_headers(1) or {
            peer.fail('headers: ${err.msg()}')
            return
        }
        id := ids[0]
        peer.respond_headers(id, false) or {
            peer.fail('respond: ${err.msg()}')
            return
        }
        peer.write_frame(H2DataFrame{
            stream_id:  id
            data:       'ok'.bytes()
            end_stream: true
        }) or {
            peer.fail('data: ${err.msg()}')
            return
        }
    }(mut peer)
    workers.wait()
    peer_thread.wait()
    assert peer.failure_msg() == ''
    assert out.statuses['/h'] == 200
    // The response round-trip guarantees the reader processed the peer SETTINGS
    // first; our response decoder must still be at the advertised default.
    assert conn.decoder.max_dynamic_size == h2_hpack_default_table_size
    cend.close_both()
}

// A POST whose upload is flow-control blocked when GOAWAY repudiates its stream
// (id above last_stream_id) must surface a retryable error, not a plain one, so
// the pool can replay a request the server never processed.
fn test_mux_goaway_preserves_retryable_for_blocked_upload() {
    mut cend, mut pend := new_mux_pipe()
    mut conn := new_test_mux_conn(mut cend)
    mut peer := &MuxTestPeer{
        end: pend
    }
    mut out := &MuxResults{}
    body_len := 100000
    mut workers := []thread{}
    workers << spawn mux_worker(mut conn, H2ClientRequest{
        method:    'POST'
        authority: 't'
        path:      '/up'
        body:      []u8{len: body_len, init: `B`}
    }, mut out)
    peer_thread := spawn fn (mut peer MuxTestPeer) {
        peer.read_preface() or {
            peer.fail('preface: ${err.msg()}')
            return
        }
        ids := peer.wait_for_headers(1) or {
            peer.fail('headers: ${err.msg()}')
            return
        }
        id := ids[0]
        // Let the upload exhaust both windows so the sender blocks in
        // send_body_on_stream waiting for a WINDOW_UPDATE.
        for {
            peer.mu.lock()
            got := peer.data_total[id] or { u64(0) }
            peer.mu.unlock()
            if got >= u64(h2_default_initial_window) {
                break
            }
            peer.pump() or {
                peer.fail('pump: ${err.msg()}')
                return
            }
        }
        // GOAWAY repudiating the in-flight stream (last_stream_id 0 < id), with
        // no connection error: the stream is retryable and the blocked sender
        // must wake with that classification preserved.
        peer.write_frame(H2GoawayFrame{
            last_stream_id: 0
            error_code:     u32(H2ErrorCode.no_error)
        }) or {
            peer.fail('goaway: ${err.msg()}')
            return
        }
    }(mut peer)
    workers.wait()
    peer_thread.wait()
    assert peer.failure_msg() == ''
    assert out.errs['/up'] or { '' } != '', 'expected the repudiated upload to fail, not hang'
    assert out.codes['/up'] or { 0 } == h2_err_retryable_code, 'upload error was not retryable: ${out.errs['/up']}'
    cend.close_both()
}

// An out-of-range peer SETTINGS_MAX_FRAME_SIZE (here 0) must fail the connection
// (RFC 7540 6.5.2), not be accepted — a zero frame size would make the send
// path's chunk step 0 and hang it in a zero-length-frame loop. The peer also
// sends a valid response afterwards: with the guard the request fails on the
// bad SETTINGS (processed first); without it the request would wrongly succeed.
fn test_mux_invalid_max_frame_size_fails_connection() {
    mut cend, mut pend := new_mux_pipe()
    mut conn := new_test_mux_conn(mut cend)
    mut peer := &MuxTestPeer{
        end: pend
    }
    mut out := &MuxResults{}
    mut workers := []thread{}
    workers << spawn mux_worker(mut conn, H2ClientRequest{ authority: 't', path: '/x' }, mut out)
    peer_thread := spawn fn (mut peer MuxTestPeer) {
        peer.read_preface() or {
            peer.fail('preface: ${err.msg()}')
            return
        }
        peer.write_frame(H2SettingsFrame{
            settings: [
                H2Setting{
                    id:    h2_settings_max_frame_size
                    value: 0
                },
            ]
        }) or {
            peer.fail('settings: ${err.msg()}')
            return
        }
        ids := peer.wait_for_headers(1) or {
            peer.fail('headers: ${err.msg()}')
            return
        }
        // A perfectly valid response — the client must still have failed the
        // connection on the illegal SETTINGS processed before it.
        peer.respond_headers(ids[0], false) or {}
        peer.write_frame(H2DataFrame{
            stream_id:  ids[0]
            data:       'nope'.bytes()
            end_stream: true
        }) or {}
    }(mut peer)
    workers.wait()
    peer_thread.wait()
    assert peer.failure_msg() == ''
    assert out.statuses['/x'] or { 0 } != 200, 'request must not succeed on an illegal SETTINGS_MAX_FRAME_SIZE'
    err_msg := out.errs['/x'] or { '' }
    assert err_msg != '', 'expected the connection to fail, not hang or succeed'
    assert err_msg.contains('MAX_FRAME_SIZE'), 'unexpected error: ${err_msg}'
    cend.close_both()
}

// A response header block split across HEADERS + CONTINUATION is reassembled.
fn test_mux_continuation_assembly() {
    mut cend, mut pend := new_mux_pipe()
    mut conn := new_test_mux_conn(mut cend)
    mut peer := &MuxTestPeer{
        end: pend
    }
    mut out := &MuxResults{}
    mut workers := []thread{}
    workers << spawn mux_worker(mut conn, H2ClientRequest{ authority: 't', path: '/cont' }, mut out)
    peer_thread := spawn fn (mut peer MuxTestPeer) {
        peer.read_preface() or {
            peer.fail('preface: ${err.msg()}')
            return
        }
        ids := peer.wait_for_headers(1) or {
            peer.fail('headers: ${err.msg()}')
            return
        }
        id := ids[0]
        block := peer.encoder.encode([H2HeaderField{':status', '200'},
            H2HeaderField{'x-long-header', 'v'.repeat(64)}])
        half := block.len / 2
        peer.write_frame(H2HeadersFrame{
            stream_id:   id
            fragment:    block[..half]
            end_headers: false
        }) or {
            peer.fail('headers: ${err.msg()}')
            return
        }
        peer.write_frame(H2ContinuationFrame{
            stream_id:   id
            fragment:    block[half..]
            end_headers: true
        }) or {
            peer.fail('cont: ${err.msg()}')
            return
        }
        peer.write_frame(H2DataFrame{
            stream_id:  id
            data:       'ok'.bytes()
            end_stream: true
        }) or {
            peer.fail('data: ${err.msg()}')
            return
        }
    }(mut peer)
    workers.wait()
    peer_thread.wait()
    assert peer.failure_msg() == ''
    assert out.errs.len == 0, 'worker errors: ${out.errs}'
    assert out.statuses['/cont'] == 200
    assert out.bodies['/cont'] == 'ok'
    cend.close_both()
}

// When the connection dies with several streams in flight, every waiter must
// wake with an error (no hangs).
fn test_mux_conn_death_wakes_all_waiters() {
    mut cend, mut pend := new_mux_pipe()
    mut conn := new_test_mux_conn(mut cend)
    mut peer := &MuxTestPeer{
        end: pend
    }
    mut out := &MuxResults{}
    mut workers := []thread{}
    for i in 0 .. 4 {
        workers << spawn mux_worker(mut conn, H2ClientRequest{ authority: 't', path: '/d${i}' }, mut out)
    }
    peer_thread := spawn fn (mut peer MuxTestPeer) {
        peer.read_preface() or {
            peer.fail('preface: ${err.msg()}')
            return
        }
        peer.wait_for_headers(4) or {
            peer.fail('headers: ${err.msg()}')
            return
        }
        // Kill the connection with all four streams mid-flight.
        peer.end.close_both()
    }(mut peer)
    workers.wait()
    peer_thread.wait()
    assert peer.failure_msg() == ''
    assert out.errs.len == 4, 'all four waiters must fail, got: ${out.errs}'
    // A subsequent request on the dead connection fails fast and retryably.
    conn.do(H2ClientRequest{ authority: 't', path: '/postmortem' }) or {
        assert err.code() == h2_err_retryable_code
        return
    }
    assert false, 'a request on a dead connection must fail'
}

// When the connection preface write fails, note_write_failure must be called so
// the connection is torn down and the pool stops admitting new work. Both the
// failing request and any immediately subsequent one must return retryable errors.
fn test_mux_preface_write_failure_tears_down() {
    mut cend, mut pend := new_mux_pipe()
    // Close the write side so the very first transport write (the preface) fails.
    cend.outgoing.close()
    mut conn := new_test_mux_conn(mut cend)
    mut out := &MuxResults{}
    mut w1 := []thread{}
    w1 << spawn mux_worker(mut conn, H2ClientRequest{ authority: 't', path: '/x' }, mut out)
    w1.wait()
    assert out.errs['/x'] or { '' } != '', 'request on broken transport must fail'
    assert out.codes['/x'] or { 0 } == h2_err_retryable_code, 'preface write failure must be retryable'
    // note_write_failure must have set shutting_down; the second request must
    // fail retryably rather than be admitted to the dead connection.
    mut out2 := &MuxResults{}
    mut w2 := []thread{}
    w2 << spawn mux_worker(mut conn, H2ClientRequest{ authority: 't', path: '/y' }, mut out2)
    w2.wait()
    assert out2.errs['/y'] or { '' } != '', 'second request on torn-down connection must fail'
    assert out2.codes['/y'] or { 0 } == h2_err_retryable_code, 'second request must also be retryable'
    cend.incoming.close() // unblock the reader thread so it exits cleanly
    pend.close_both()
}

// Applying a positive SETTINGS_INITIAL_WINDOW_SIZE delta to a stream that
// already has extra credit from a WINDOW_UPDATE can overflow the stream's send
// window above 2^31-1, which is a connection FLOW_CONTROL_ERROR (RFC 7540
// §6.9.2). The connection must be failed, not silently keep the invalid window.
fn test_mux_settings_initial_window_delta_overflow_fails_connection() {
    mut cend, mut pend := new_mux_pipe()
    mut conn := new_test_mux_conn(mut cend)
    mut peer := &MuxTestPeer{
        end: pend
    }
    mut out := &MuxResults{}
    mut workers := []thread{}
    // A simple GET keeps the stream registered in c.streams (waiting for a
    // response that never arrives), so the SETTINGS delta has a stream to hit.
    workers << spawn mux_worker(mut conn, H2ClientRequest{ authority: 't', path: '/ov' }, mut out)
    peer_thread := spawn fn (mut peer MuxTestPeer) {
        peer.read_preface() or {
            peer.fail('preface: ${err.msg()}')
            return
        }
        peer.wait_for_headers(1) or {
            peer.fail('headers: ${err.msg()}')
            return
        }
        // Raise the stream's send window close to 2^31-1 via WINDOW_UPDATE,
        // then raise SETTINGS_INITIAL_WINDOW_SIZE so the delta pushes the window
        // over the limit.  Stream send_window ≈ 65535 + 0x7fff_0000 = 0x7fff_ffff;
        // delta = 0x7fff_ffff - 65535 = 0x7fff_0000 → new window ≈ 0xfffe_0000
        // which exceeds 0x7fff_ffff → FLOW_CONTROL_ERROR.
        peer.write_frame(H2WindowUpdateFrame{
            stream_id:             1
            window_size_increment: u32(0x7fff_0000)
        }) or {
            peer.fail('wu: ${err.msg()}')
            return
        }
        peer.write_frame(H2SettingsFrame{
            settings: [H2Setting{h2_settings_initial_window_size, u32(0x7fff_ffff)}]
        }) or {
            peer.fail('settings: ${err.msg()}')
            return
        }
    }(mut peer)
    workers.wait()
    peer_thread.wait()
    assert peer.failure_msg() == ''
    err_msg := out.errs['/ov'] or { '' }
    assert err_msg != '', 'expected a connection FLOW_CONTROL_ERROR, got success or hang'
    assert err_msg.contains('FLOW_CONTROL_ERROR'), 'unexpected error: ${err_msg}'
    cend.close_both()
}

// A server that ignores our advertised receive window and sends more DATA than
// it allows must trigger a FLOW_CONTROL_ERROR: the connection must be failed,
// not silently buffer unbounded data. on_data blocks the requester from draining
// (and sending WINDOW_UPDATE) while the peer floods the connection window.
fn test_mux_peer_exceeding_recv_window_fails_connection() {
    mut cend, mut pend := new_mux_pipe()
    mut conn := new_test_mux_conn(mut cend)
    mut peer := &MuxTestPeer{
        end: pend
    }
    mut out := &MuxResults{}
    // Deplete the tracked connection-level receive window to 1 byte so the
    // very first DATA frame (2 bytes) immediately overflows it and triggers
    // FLOW_CONTROL_ERROR.  Doing this before any goroutine sends frames means
    // no WINDOW_UPDATE is ever sent that could re-fill the window first, so
    // the result is fully deterministic with no concurrent timing race.
    conn.recv_wmu.lock()
    conn.conn_recv_window = 1
    conn.recv_wmu.unlock()
    mut workers := []thread{}
    workers << spawn mux_worker(mut conn, H2ClientRequest{
        authority: 't'
        path:      '/flood'
    }, mut out)
    peer_thread := spawn fn (mut peer MuxTestPeer) {
        peer.read_preface() or {
            peer.fail('preface: ${err.msg()}')
            return
        }
        ids := peer.wait_for_headers(1) or {
            peer.fail('headers: ${err.msg()}')
            return
        }
        peer.respond_headers(ids[0], false) or {
            peer.fail('respond: ${err.msg()}')
            return
        }
        // 2-byte DATA frame: connection window is 1 byte, so this exceeds it
        // and must trigger FLOW_CONTROL_ERROR on the client.
        peer.write_frame(H2DataFrame{
            stream_id: ids[0]
            data:      [u8(0), 0]
        }) or { peer.fail('data: ${err.msg()}') }
    }(mut peer)
    workers.wait()
    peer_thread.wait()
    assert peer.failure_msg() == ''
    err_msg := out.errs['/flood'] or { '' }
    assert err_msg != '', 'expected FLOW_CONTROL_ERROR, got success or hang'
    assert err_msg.contains('FLOW_CONTROL_ERROR'), 'unexpected error: ${err_msg}'
    cend.close_both()
}

// A STREAM-level receive-window violation (RFC 7540 §6.9.1) must reset only that
// stream, not tear down the whole connection: the offending request fails, but
// the connection stays alive (closed stays false) so other streams are unharmed.
fn test_mux_stream_window_violation_resets_only_that_stream() {
    mut cend, mut pend := new_mux_pipe()
    mut conn := new_test_mux_conn(mut cend)
    mut peer := &MuxTestPeer{
        end: pend
    }
    mut out := &MuxResults{}
    // Inflate the connection window so the oversized DATA trips ONLY the
    // stream-level window check, not the connection-level one.
    conn.recv_wmu.lock()
    conn.conn_recv_window = 10_000_000
    conn.recv_wmu.unlock()
    mut workers := []thread{}
    workers << spawn mux_worker(mut conn, H2ClientRequest{
        authority: 't'
        path:      '/flood'
    }, mut out)
    peer_thread := spawn fn (mut peer MuxTestPeer, mut conn H2MuxConn) {
        peer.read_preface() or {
            peer.fail('preface: ${err.msg()}')
            return
        }
        ids := peer.wait_for_headers(1) or {
            peer.fail('headers: ${err.msg()}')
            return
        }
        id := ids[0]
        peer.respond_headers(id, false) or {
            peer.fail('respond: ${err.msg()}')
            return
        }
        // White-box: shrink just this stream's receive window so the next DATA
        // frame exceeds it while the connection window stays valid. No DATA has
        // arrived yet, so the requester has not credited the window back.
        conn.smu.lock()
        if mut s := conn.streams[id] {
            s.mu.lock()
            s.recv_window = 1
            s.mu.unlock()
        }
        conn.smu.unlock()
        peer.write_frame(H2DataFrame{
            stream_id: id
            data:      [u8(0), 0]
        }) or {
            peer.fail('data: ${err.msg()}')
            return
        }
        // The client must RST only this stream; pump until we see it.
        for {
            f := peer.pump() or { return }
            if f is H2RstStreamFrame {
                return
            }
        }
    }(mut peer, mut conn)
    workers.wait()
    peer_thread.wait()
    assert peer.failure_msg() == ''
    err_msg := out.errs['/flood'] or { '' }
    assert err_msg != '', 'stream-window violation must error the request'
    assert err_msg.contains('FLOW_CONTROL_ERROR') || err_msg.contains('receive window'), 'unexpected error: ${err_msg}'
    // The connection must survive — closed is set only when the reader exits on a
    // connection-level failure, which a stream reset must not cause.
    conn.smu.lock()
    closed := conn.closed
    conn.smu.unlock()
    assert !closed, 'a stream-level flow-control violation must NOT close the connection'
    cend.close_both()
}

// DATA after END_STREAM on a fully "closed" stream (we also sent our END_STREAM)
// is a connection error (RFC 7540 §5.1 MUST).
fn test_mux_data_after_end_stream_on_closed_stream_fails_connection() {
    mut cend, mut pend := new_mux_pipe()
    mut conn := new_test_mux_conn(mut cend)
    mut peer := &MuxTestPeer{
        end: pend
    }
    // White-box: register a stream in the "closed" state — both the peer (ended)
    // and we (send_closed) have finished sending.
    mut s := new_h2_mux_stream()
    s.id = 1
    s.headers_done = true
    s.ended = true
    s.send_closed = true
    conn.smu.lock()
    conn.streams[1] = s
    conn.smu.unlock()
    // A DATA frame on the closed stream must fail the whole connection.
    peer.write_frame(H2DataFrame{ stream_id: 1, data: [u8(0)] }) or {
        assert false, 'write: ${err.msg()}'
    }
    mut closed := false
    for _ in 0 .. 2000 {
        conn.smu.lock()
        closed = conn.closed
        conn.smu.unlock()
        if closed {
            break
        }
        time.sleep(time.millisecond)
    }
    assert closed, 'DATA after END_STREAM on a closed stream must fail the connection'
    cend.close_both()
}

// DATA after END_STREAM on a "half-closed (remote)" stream (the peer ended but
// we have NOT closed our send side) is a STREAM error: RST that stream, keep the
// connection alive (RFC 7540 §5.1).
fn test_mux_data_after_end_stream_half_closed_resets_only_stream() {
    mut cend, mut pend := new_mux_pipe()
    mut conn := new_test_mux_conn(mut cend)
    mut peer := &MuxTestPeer{
        end: pend
    }
    // White-box: register a stream where the peer has ended but our send side is
    // still open (send_closed = false) — "half-closed (remote)".
    mut s := new_h2_mux_stream()
    s.id = 1
    s.headers_done = true
    s.ended = true
    s.send_closed = false
    conn.smu.lock()
    conn.streams[1] = s
    conn.smu.unlock()
    peer.write_frame(H2DataFrame{ stream_id: 1, data: [u8(0)] }) or {
        assert false, 'write: ${err.msg()}'
    }
    // The client must RST just this stream; pump until we see it.
    mut saw_rst := false
    for _ in 0 .. 50 {
        f := peer.pump() or { break }
        if f is H2RstStreamFrame {
            saw_rst = true
            break
        }
    }
    assert saw_rst, 'half-closed-remote DATA-after-END_STREAM must RST the stream'
    conn.smu.lock()
    closed := conn.closed
    conn.smu.unlock()
    assert !closed, 'a stream-level reset must NOT close the connection'
    cend.close_both()
}

// A peer that floods control frames before the client preface (the connection
// stays pre-handshake until the first request) must not grow the deferred-ACK
// buffers without bound — the connection is failed once the cap is exceeded.
fn test_mux_preface_control_flood_fails_connection() {
    mut cend, mut pend := new_mux_pipe()
    mut conn := new_test_mux_conn(mut cend)
    mut peer := &MuxTestPeer{
        end: pend
    }
    // No request is made, so the lazy client preface is never sent; every PING
    // is deferred. Send more than the cap allows.
    for _ in 0 .. (h2_max_pending_preface_acks + 5) {
        peer.write_frame(H2PingFrame{ data: [u8(1), 2, 3, 4, 5, 6, 7, 8] }) or {
            assert false, 'write: ${err.msg()}'
        }
    }
    mut closed := false
    for _ in 0 .. 2000 {
        conn.smu.lock()
        closed = conn.closed
        conn.smu.unlock()
        if closed {
            break
        }
        time.sleep(time.millisecond)
    }
    assert closed, 'a pre-preface control-frame flood must fail the connection'
    cend.close_both()
}

// A malformed Content-Length (string.u64() would leniently parse '12junk' -> 12)
// makes the response malformed; it must be rejected, not accepted as a success.
fn test_mux_malformed_content_length_resets_stream() {
    mut cend, mut pend := new_mux_pipe()
    mut conn := new_test_mux_conn(mut cend)
    mut peer := &MuxTestPeer{
        end: pend
    }
    mut out := &MuxResults{}
    mut workers := []thread{}
    workers << spawn mux_worker(mut conn, H2ClientRequest{ authority: 't', path: '/x' }, mut out)
    peer_thread := spawn fn (mut peer MuxTestPeer) {
        peer.read_preface() or {
            peer.fail('preface: ${err.msg()}')
            return
        }
        ids := peer.wait_for_headers(1) or {
            peer.fail('headers: ${err.msg()}')
            return
        }
        block := peer.encoder.encode([H2HeaderField{':status', '200'},
            H2HeaderField{'content-length', '12junk'}])
        peer.write_frame(H2HeadersFrame{
            stream_id:   ids[0]
            fragment:    block
            end_headers: true
            end_stream:  false
        }) or {
            peer.fail('resp: ${err.msg()}')
            return
        }
        for {
            f := peer.pump() or { return }
            if f is H2RstStreamFrame {
                return
            }
        }
    }(mut peer)
    workers.wait()
    peer_thread.wait()
    assert peer.failure_msg() == ''
    err_msg := out.errs['/x'] or { '' }
    assert err_msg != '', 'malformed Content-Length must error the request, not succeed'
    assert err_msg.to_lower().contains('content-length'), 'unexpected error: ${err_msg}'
    cend.close_both()
}

// RFC 7540 §8.1: a response must begin with HEADERS. A DATA frame before any
// response HEADERS is malformed — reset that stream, but keep the connection.
fn test_mux_data_before_headers_resets_stream() {
    mut cend, mut pend := new_mux_pipe()
    mut conn := new_test_mux_conn(mut cend)
    mut peer := &MuxTestPeer{
        end: pend
    }
    mut out := &MuxResults{}
    mut workers := []thread{}
    workers << spawn mux_worker(mut conn, H2ClientRequest{ authority: 't', path: '/x' }, mut out)
    peer_thread := spawn fn (mut peer MuxTestPeer) {
        peer.read_preface() or {
            peer.fail('preface: ${err.msg()}')
            return
        }
        ids := peer.wait_for_headers(1) or {
            peer.fail('headers: ${err.msg()}')
            return
        }
        // DATA with no preceding response HEADERS (malformed).
        peer.write_frame(H2DataFrame{
            stream_id: ids[0]
            data:      'oops'.bytes()
        }) or {
            peer.fail('data: ${err.msg()}')
            return
        }
        for {
            f := peer.pump() or { return }
            if f is H2RstStreamFrame {
                return
            }
        }
    }(mut peer)
    workers.wait()
    peer_thread.wait()
    assert peer.failure_msg() == ''
    err_msg := out.errs['/x'] or { '' }
    assert err_msg != '', 'DATA before HEADERS must error the request'
    conn.smu.lock()
    closed := conn.closed
    conn.smu.unlock()
    assert !closed, 'DATA before HEADERS is a stream error; the connection must survive'
    cend.close_both()
}

// Releasing the last reference on an idle connection must wake the background
// reader through the required close_transport callback. With no requests and no
// inbound frames the reader is blocked in transport.read(); only closing the
// transport can unblock it (the H2Transport interface has no close()). This
// exercises the teardown -> closer -> reader-exit chain WITHOUT the test closing
// the pipe manually, guarding the Codex P2 fix (discussion_r3475761576): a nil
// closer is now rejected, so teardown can always close the transport. (Note: the
// nil-closer leak itself is reproduced statically by detect_mux_requires_closer.sh,
// since a nil closer now panics and cannot be exercised at runtime.)
fn test_mux_release_wakes_reader_via_closer() {
    mut cend, _ := new_mux_pipe()
    mut conn := new_test_mux_conn(mut cend)
    // No requests, no frames: the reader is blocked in transport.read().
    conn.shutdown_when_idle()
    conn.release() // last ref -> teardown_transport -> close_transport closes the pipe
    // The reader observes the closed transport and exits via fail_conn(closed=true).
    mut woke := false
    for _ in 0 .. 200 {
        conn.smu.lock()
        woke = conn.closed
        conn.smu.unlock()
        if woke {
            break
        }
        time.sleep(5 * time.millisecond)
    }
    assert woke, 'reader did not exit after release(): close_transport failed to wake it'
}

// A trailing HEADERS block (trailers) must surface its non-pseudo fields in the
// response headers, matching the synchronous H2Conn.read_response. Regression for
// Codex P2 (discussion_r3477245531): the mux path HPACK-decoded the trailer block
// but dropped every field, only marking the stream ended — so callers lost
// grpc-status, digest, etc. even though the stream completed normally.
fn test_mux_response_trailers_preserved() {
    mut cend, mut pend := new_mux_pipe()
    mut conn := new_test_mux_conn(mut cend)
    mut peer := &MuxTestPeer{
        end: pend
    }
    peer_thread := spawn fn (mut peer MuxTestPeer) {
        peer.read_preface() or {
            peer.fail('preface: ${err.msg()}')
            return
        }
        ids := peer.wait_for_headers(1) or {
            peer.fail('headers: ${err.msg()}')
            return
        }
        id := ids[0]
        // Final response headers (no END_STREAM: a body and trailers follow).
        peer.respond_headers(id, false) or {
            peer.fail('respond: ${err.msg()}')
            return
        }
        peer.write_frame(H2DataFrame{
            stream_id:  id
            data:       'hi'.bytes()
            end_stream: false
        }) or {
            peer.fail('data: ${err.msg()}')
            return
        }
        // A trailing HEADERS block carrying END_STREAM, with valid (non-pseudo)
        // trailer fields that must be preserved. (Pseudo-headers in trailers are
        // malformed per §8.1 and are covered by the rejection test below.)
        trailer_block := peer.encoder.encode([H2HeaderField{'grpc-status', '0'},
            H2HeaderField{'grpc-message', 'ok'}])
        peer.write_frame(H2HeadersFrame{
            stream_id:   id
            fragment:    trailer_block
            end_headers: true
            end_stream:  true
        }) or {
            peer.fail('trailers: ${err.msg()}')
            return
        }
    }(mut peer)
    resp := conn.do(H2ClientRequest{ authority: 't', path: '/g' }) or {
        assert false, 'request failed: ${err}'
        return
    }
    peer_thread.wait()
    assert peer.failure_msg() == ''
    assert resp.status == 200
    assert resp.body.bytestr() == 'hi'
    mut grpc_status := ''
    mut grpc_message := ''
    mut pseudo_in_headers := false
    for h in resp.headers {
        match h.name {
            'grpc-status' { grpc_status = h.value }
            'grpc-message' { grpc_message = h.value }
            else {}
        }

        if h.name.starts_with(':') {
            pseudo_in_headers = true
        }
    }
    assert grpc_status == '0', 'trailer grpc-status missing from response headers: ${resp.headers}'
    assert grpc_message == 'ok', 'trailer grpc-message missing from response headers: ${resp.headers}'
    assert !pseudo_in_headers, 'pseudo-header field leaked into trailers: ${resp.headers}'
    cend.close_both()
}

// mux_response_rejected runs one request whose response carries `resp_fields` and
// asserts the request fails with an error containing `want` (the stream was reset
// as malformed), rather than delivering the response.
fn mux_response_rejected(resp_fields []H2HeaderField, want string) {
    mut cend, mut pend := new_mux_pipe()
    mut conn := new_test_mux_conn(mut cend)
    mut peer := &MuxTestPeer{
        end: pend
    }
    peer_thread := spawn fn (mut peer MuxTestPeer, resp_fields []H2HeaderField) {
        peer.read_preface() or {
            peer.fail('preface: ${err.msg()}')
            return
        }
        ids := peer.wait_for_headers(1) or {
            peer.fail('headers: ${err.msg()}')
            return
        }
        block := peer.encoder.encode(resp_fields)
        peer.write_frame(H2HeadersFrame{
            stream_id:   ids[0]
            fragment:    block
            end_headers: true
            end_stream:  true
        }) or {
            peer.fail('resp: ${err.msg()}')
            return
        }
        // Drain the client's RST_STREAM until the pipe closes.
        for {
            peer.pump() or { return }
        }
    }(mut peer, resp_fields)
    mut got := ''
    if resp := conn.do(H2ClientRequest{ authority: 't', path: '/x' }) {
        got = '<<accepted: status=${resp.status} headers=${resp.headers}>>'
    } else {
        got = err.msg()
    }
    cend.close_both() // unblock the peer's pump loop so its thread exits
    peer_thread.wait()
    assert peer.failure_msg() == ''
    assert got.contains(want), 'expected "${want}" in the rejection, got: ${got}'
}

// RFC 9113 §8.2: a response carrying a malformed field — a connection-specific
// header (transfer-encoding/connection/...) or an uppercase field name — is
// malformed and must be rejected, not delivered to the caller. Regression for
// the proactive RFC-conformance audit (gap G1): both client paths previously
// appended received fields with only :status / content-length validation.
fn test_mux_response_rejects_malformed_fields() {
    mux_response_rejected([H2HeaderField{':status', '200'},
        H2HeaderField{'transfer-encoding', 'chunked'}], 'transfer-encoding')
    mux_response_rejected([H2HeaderField{':status', '200'},
        H2HeaderField{'Content-Type', 'text/plain'}], 'uppercase')
    // An undefined response pseudo-header is also malformed (§8.3.1).
    mux_response_rejected([H2HeaderField{':status', '200'}, H2HeaderField{':custom', 'x'}],
        'pseudo-header')
}

// RFC 9113 §8.1: a trailing HEADERS block carrying a pseudo-header (or any
// malformed field) is malformed and must reset the stream, not be dropped.
fn test_mux_trailers_reject_pseudo() {
    mut cend, mut pend := new_mux_pipe()
    mut conn := new_test_mux_conn(mut cend)
    mut peer := &MuxTestPeer{
        end: pend
    }
    peer_thread := spawn fn (mut peer MuxTestPeer) {
        peer.read_preface() or {
            peer.fail('preface: ${err.msg()}')
            return
        }
        ids := peer.wait_for_headers(1) or {
            peer.fail('headers: ${err.msg()}')
            return
        }
        id := ids[0]
        peer.respond_headers(id, false) or {
            peer.fail('respond: ${err.msg()}')
            return
        }
        // Trailers must not contain a pseudo-header.
        block := peer.encoder.encode([H2HeaderField{':status', '200'}])
        peer.write_frame(H2HeadersFrame{
            stream_id:   id
            fragment:    block
            end_headers: true
            end_stream:  true
        }) or {
            peer.fail('trailers: ${err.msg()}')
            return
        }
        for {
            peer.pump() or { return }
        }
    }(mut peer)
    mut got := ''
    if resp := conn.do(H2ClientRequest{ authority: 't', path: '/g' }) {
        got = '<<accepted: ${resp.headers}>>'
    } else {
        got = err.msg()
    }
    cend.close_both()
    peer_thread.wait()
    assert peer.failure_msg() == ''
    assert got.contains('malformed trailers'), 'trailers pseudo-header not rejected: ${got}'
}

// RFC 9113 §6.5.2: the client honors the peer's advisory
// SETTINGS_MAX_HEADER_LIST_SIZE and refuses an over-limit request locally rather
// than emitting it. Conformance gap G4. The peer sends no SETTINGS here, so the
// reader never writes peer_max_header_list_size — setting it directly before the
// request is race-free.
fn test_mux_request_respects_peer_max_header_list_size() {
    mut cend, mut pend := new_mux_pipe()
    mut conn := new_test_mux_conn(mut cend)
    conn.peer_max_header_list_size = 40 // tiny: even the pseudo-headers exceed it
    mut peer := &MuxTestPeer{
        end: pend
    }
    peer_thread := spawn fn (mut peer MuxTestPeer) {
        peer.read_preface() or {
            peer.fail('preface: ${err.msg()}')
            return
        }
        for {
            peer.pump() or { return }
        }
    }(mut peer)
    mut got := ''
    if _ := conn.do(H2ClientRequest{
        method:    'GET'
        scheme:    'https'
        authority: 'example.com'
        path:      '/a-fairly-long-path-to-exceed-the-limit'
    })
    {
        got = '<<accepted>>'
    } else {
        got = err.msg()
    }
    cend.close_both()
    peer_thread.wait()
    assert peer.failure_msg() == ''
    assert got.contains('MAX_HEADER_LIST_SIZE'), 'over-limit request not rejected: ${got}'
}