/*
 * test_conn.c -- Connection protocol tests for DeuceSSH.
 *
 * Tier 2 layer tests with mock I/O.  Covers session start/stop,
 * channel open/accept, stream and subsystem channel I/O, signals,
 * window change, close/exit, multiple channels, and thread safety.
 *
 * Uses the new dssh_chan_* API throughout.
 */

#include <stddef.h>
#include <stdint.h>
#include <string.h>
#include <time.h>
#include <threads.h>

#include "dssh_test.h"
#include "deucessh.h"
#include "deucessh-algorithms.h"
#include "deucessh-auth.h"
#include "deucessh-conn.h"
#include "ssh-trans.h"
#include "ssh-internal.h"
#include "dssh_test_internal.h"
#include "dssh_test_alloc.h"
#include "dssh_test_ossl.h"
#include "mock_io.h"
#include "test_dhgex_provider.h"


/* ================================================================
 * Dispatch callbacks for two-sided tests.
 * ================================================================ */

static int
mock_tx_dispatch(uint8_t *buf, size_t bufsz,
    dssh_session sess, void *cbdata)
{
	if (sess->trans.client)
		return mock_tx_client(buf, bufsz, sess, cbdata);
	else
		return mock_tx_server(buf, bufsz, sess, cbdata);
}

static int
mock_rx_dispatch(uint8_t *buf, size_t bufsz,
    dssh_session sess, void *cbdata)
{
	if (sess->trans.client)
		return mock_rx_client(buf, bufsz, sess, cbdata);
	else
		return mock_rx_server(buf, bufsz, sess, cbdata);
}

static int
mock_rxline_dispatch(uint8_t *buf, size_t bufsz,
    size_t *bytes_received, dssh_session sess, void *cbdata)
{
	if (sess->trans.client)
		return mock_rxline_client(buf, bufsz, bytes_received, sess, cbdata);
	else
		return mock_rxline_server(buf, bufsz, bytes_received, sess, cbdata);
}

/* ================================================================
 * Helper: register all algorithms needed for handshake
 * ================================================================ */

static dssh_session
init_server_session(void)
{
	dssh_session s = dssh_session_init(false, 0);
	if (s != NULL)
		test_dhgex_setup(s);
	return s;
}

static int
register_all_algorithms(void)
{
	int res;
	if (test_using_dhgex())
		res = dssh_register_dh_gex_sha256();
	else if (test_using_sntrup())
		res = dssh_register_sntrup761x25519_sha512();
	else if (test_using_mlkem())
		res = dssh_register_mlkem768x25519_sha256();
	else
		res = dssh_register_curve25519_sha256();
	if (res < 0)
		return res;
	res = test_register_key_algos();
	if (res < 0)
		return res;
	res = dssh_register_aes256_ctr();
	if (res < 0)
		return res;
	res = test_register_mac_algos();
	if (res < 0)
		return res;
	res = dssh_register_none_comp();
	if (res < 0)
		return res;
	return 0;
}

/* ================================================================
 * Auth callbacks (accept-all server side)
 * ================================================================ */

static int
auth_password_accept_all(const uint8_t *username, size_t username_len,
    const uint8_t *password, size_t password_len,
    uint8_t **change_prompt, size_t *change_prompt_len,
    void *cbdata)
{
	return DSSH_AUTH_SUCCESS;
}

static struct dssh_auth_server_cbs auth_cbs = {
	.methods_str = "password",
	.password_cb = auth_password_accept_all,
};

/* ================================================================
 * Server accept callbacks (new dssh_chan_accept_cbs)
 * ================================================================ */

static int pty_received;
static char pty_term[64];
static uint32_t pty_cols, pty_rows;
static int wc_received;
static uint32_t wc_cols, wc_rows, wc_wpx, wc_hpx;

static int
accept_all_shell(dssh_channel ch, const struct dssh_chan_params *p,
    struct dssh_chan_accept_result *result, void *cbdata)
{
	(void)ch;
	(void)p;
	(void)result;
	(void)cbdata;
	return 0;
}

static int
accept_all_exec(dssh_channel ch, const struct dssh_chan_params *p,
    struct dssh_chan_accept_result *result, void *cbdata)
{
	(void)ch;
	(void)p;
	(void)result;
	(void)cbdata;
	return 0;
}

static int
accept_all_subsystem(dssh_channel ch, const struct dssh_chan_params *p,
    struct dssh_chan_accept_result *result, void *cbdata)
{
	(void)ch;
	(void)p;
	(void)result;
	(void)cbdata;
	return 0;
}

static int
accept_pty_cb(dssh_channel ch, const struct dssh_chan_params *p,
    void *cbdata)
{
	(void)ch;
	(void)cbdata;
	pty_received = 1;
	size_t n = strlen(p->term);
	if (n >= sizeof(pty_term))
		n = sizeof(pty_term) - 1;
	memcpy(pty_term, p->term, n);
	pty_term[n] = 0;
	pty_cols = p->cols;
	pty_rows = p->rows;
	return 0;
}

static struct dssh_chan_accept_cbs accept_cbs_all = {
	.pty_req = accept_pty_cb,
	.shell = accept_all_shell,
	.exec = accept_all_exec,
	.subsystem = accept_all_subsystem,
};

static void
reset_callbacks(void)
{
	pty_received = 0;
	pty_term[0] = 0;
	pty_cols = 0;
	pty_rows = 0;
	wc_received = 0;
	wc_cols = 0;
	wc_rows = 0;
	wc_wpx = 0;
	wc_hpx = 0;
}

/* ================================================================
 * Connection setup context -- handshake + auth + session_start
 * ================================================================ */

struct conn_ctx {
	struct mock_io_state io;
	dssh_session client;
	dssh_session server;
};

/* ASSERT_* returns directly, so retain the current resource-owning
 * context for the runner's after-each cleanup hook. */
static struct conn_ctx *g_active_ctx;

struct handshake_auth_ctx {
	struct mock_io_state *io;
	dssh_session sess;
	int result;
};

static int
client_handshake_auth_thread(void *arg)
{
	struct handshake_auth_ctx *ctx = arg;
	ctx->result = dssh_transport_handshake(ctx->sess);
	if (ctx->result < 0) {
		mock_io_close_c2s(ctx->io);
		mock_io_close_s2c(ctx->io);
		return 0;
	}
	ctx->result = dssh_auth_password(ctx->sess, "testuser", "testpass",
	    NULL, NULL);
	return 0;
}

static int
server_handshake_auth_thread(void *arg)
{
	struct handshake_auth_ctx *ctx = arg;
	ctx->result = dssh_transport_handshake(ctx->sess);
	if (ctx->result < 0) {
		mock_io_close_c2s(ctx->io);
		mock_io_close_s2c(ctx->io);
		return 0;
	}
	uint8_t username[256];
	size_t username_len = sizeof(username);
	ctx->result = dssh_auth_server(ctx->sess, &auth_cbs,
	    username, &username_len);
	return 0;
}

/*
 * Set up a full connection: handshake + auth + session_start on both sides.
 * Returns 0 on success, -1 on failure.
 */
static int
conn_setup(struct conn_ctx *ctx)
{
	memset(ctx, 0, sizeof(*ctx));
	reset_callbacks();

	dssh_test_reset_global_config();

	if (register_all_algorithms() < 0)
		return -1;
	if (test_generate_host_key() < 0)
		return -1;

	if (mock_io_init(&ctx->io, 0) < 0)
		return -1;

	dssh_transport_set_callbacks(mock_tx_dispatch, mock_rx_dispatch,
	    mock_rxline_dispatch, mock_extra_line_cb);

	ctx->client = dssh_session_init(true, 0);
	if (ctx->client == NULL) {
		mock_io_free(&ctx->io);
		return -1;
	}
	dssh_session_set_cbdata(ctx->client, &ctx->io, &ctx->io,
	    &ctx->io, &ctx->io);
	dssh_session_set_hostkey_verify_cb(ctx->client, dssh_test_accept_hostkey, NULL);

	ctx->server = init_server_session();
	if (ctx->server == NULL) {
		dssh_session_cleanup(ctx->client);
		mock_io_free(&ctx->io);
		return -1;
	}
	dssh_session_set_cbdata(ctx->server, &ctx->io, &ctx->io,
	    &ctx->io, &ctx->io);

	struct handshake_auth_ctx client_ha = {
		.io = &ctx->io,
		.sess = ctx->client,
		.result = -1,
	};
	struct handshake_auth_ctx server_ha = {
		.io = &ctx->io,
		.sess = ctx->server,
		.result = -1,
	};

	thrd_t ct, st;
	if (thrd_create(&ct, client_handshake_auth_thread, &client_ha) != thrd_success) {
		dssh_session_cleanup(ctx->server);
		dssh_session_cleanup(ctx->client);
		mock_io_free(&ctx->io);
		return -1;
	}
	if (thrd_create(&st, server_handshake_auth_thread, &server_ha) != thrd_success) {
		dssh_session_terminate(ctx->client);
		mock_io_close_c2s(&ctx->io);
		mock_io_close_s2c(&ctx->io);
		thrd_join(ct, NULL);
		dssh_session_cleanup(ctx->server);
		dssh_session_cleanup(ctx->client);
		mock_io_free(&ctx->io);
		return -1;
	}

	thrd_join(ct, NULL);
	thrd_join(st, NULL);

	if (client_ha.result != 0 || server_ha.result != 0) {
		dssh_session_cleanup(ctx->server);
		dssh_session_cleanup(ctx->client);
		mock_io_free(&ctx->io);
		return -1;
	}

	if (dssh_session_start(ctx->client) != 0) {
		dssh_session_cleanup(ctx->server);
		dssh_session_cleanup(ctx->client);
		mock_io_free(&ctx->io);
		return -1;
	}
	if (dssh_session_start(ctx->server) != 0) {
		mock_io_close_c2s(&ctx->io);
		mock_io_close_s2c(&ctx->io);
		dssh_session_stop(ctx->client);
		dssh_session_cleanup(ctx->server);
		dssh_session_cleanup(ctx->client);
		mock_io_free(&ctx->io);
		return -1;
	}

	g_active_ctx = ctx;
	return 0;
}

static void
conn_cleanup(struct conn_ctx *ctx)
{
	/* The after-each hook runs just after the test function's stack frame
	 * has been popped.  Snapshot it before cleanup calls reuse that stack. */
	struct conn_ctx snap = *ctx;

	if (g_active_ctx == ctx)
		g_active_ctx = NULL;

	/*
	 * Close the write-ends of both pipes first so that any demux
	 * thread blocked in read() gets EOF and can exit.  With
	 * socketpair-based mock I/O, condvar broadcasts alone cannot
	 * unblock a blocking read() -- only closing the peer fd does.
	 */
	mock_io_close_c2s(&snap.io);
	mock_io_close_s2c(&snap.io);
	dssh_session_stop(snap.server);
	dssh_session_stop(snap.client);
	dssh_session_cleanup(snap.server);
	dssh_session_cleanup(snap.client);
	mock_io_free(&snap.io);
	dssh_test_reset_global_config();
}

/* ================================================================
 * Helper: open an exec channel concurrently (client+server threads)
 * ================================================================ */

/* Channel setup starts with a zero receive window on both sides.  Each side
 * opens its window asynchronously after setup, so a completed open/accept
 * does not by itself guarantee that an immediate non-blocking write works. */
static int
wait_stream_writable(dssh_channel ch)
{
	if (ch == NULL || ch->io_model != DSSH_IO_STREAM)
		return 0;
	int ev = dssh_chan_poll(ch, DSSH_POLL_WRITE, 5000);
	if ((ev & DSSH_POLL_WRITE) == 0)
		return -1;
	return 0;
}

struct open_exec_ctx {
	dssh_session client;
	dssh_session server;
	dssh_channel client_ch;
	dssh_channel server_ch;
	const char *command;
	const struct dssh_chan_accept_cbs *cbs;
	int accept_timeout;
};

static int
client_open_exec_thread(void *arg)
{
	struct open_exec_ctx *ctx = arg;
	struct dssh_chan_params p;
	dssh_chan_params_init(&p, DSSH_CHAN_EXEC);
	dssh_chan_params_set_command(&p, ctx->command);
	dssh_chan_params_set_pty(&p, false);
	ctx->client_ch = dssh_chan_open(ctx->client, &p);
	dssh_chan_params_free(&p);
	return wait_stream_writable(ctx->client_ch);
}

static int
server_accept_exec_thread(void *arg)
{
	struct open_exec_ctx *ctx = arg;
	ctx->server_ch = dssh_chan_accept(ctx->server, ctx->cbs,
	    ctx->accept_timeout);
	return wait_stream_writable(ctx->server_ch);
}

/*
 * Open an exec channel concurrently.  Returns 0 on success, -1 on failure.
 * Populates ctx->client_ch and ctx->server_ch.
 */
static int
open_exec_channel(struct open_exec_ctx *ctx)
{
	thrd_t ct, st;
	int cr, sr;
	if (thrd_create(&ct, client_open_exec_thread, ctx) != thrd_success)
		return -1;
	if (thrd_create(&st, server_accept_exec_thread, ctx) != thrd_success) {
		thrd_join(ct, NULL);
		return -1;
	}
	thrd_join(ct, &cr);
	thrd_join(st, &sr);
	if (cr < 0 || sr < 0)
		return -1;
	if (ctx->client_ch == NULL || ctx->server_ch == NULL)
		return -1;
	return 0;
}

/* ================================================================
 * Helper: open a shell channel concurrently
 * ================================================================ */

struct open_shell_ctx {
	dssh_session client;
	dssh_session server;
	dssh_channel client_ch;
	dssh_channel server_ch;
	const char *term;
	uint32_t cols, rows, wpx, hpx;
	const struct dssh_chan_accept_cbs *cbs;
};

static int
client_open_shell_thread(void *arg)
{
	struct open_shell_ctx *ctx = arg;
	struct dssh_chan_params p;
	dssh_chan_params_init(&p, DSSH_CHAN_SHELL);
	dssh_chan_params_set_term(&p, ctx->term);
	dssh_chan_params_set_size(&p, ctx->cols, ctx->rows, ctx->wpx, ctx->hpx);
	ctx->client_ch = dssh_chan_open(ctx->client, &p);
	dssh_chan_params_free(&p);
	return wait_stream_writable(ctx->client_ch);
}

static int
server_accept_shell_thread(void *arg)
{
	struct open_shell_ctx *ctx = arg;
	ctx->server_ch = dssh_chan_accept(ctx->server, ctx->cbs, 30000);
	return wait_stream_writable(ctx->server_ch);
}

/* ================================================================
 * Helper: open a subsystem channel concurrently
 * ================================================================ */

struct open_subsys_ctx {
	dssh_session client;
	dssh_session server;
	dssh_channel client_ch;
	dssh_channel server_ch;
	const char *subsystem;
};

static int
client_open_subsys_thread(void *arg)
{
	struct open_subsys_ctx *ctx = arg;
	struct dssh_chan_params p;
	dssh_chan_params_init(&p, DSSH_CHAN_SUBSYSTEM);
	dssh_chan_params_set_subsystem(&p, ctx->subsystem);
	dssh_chan_params_set_pty(&p, false);
	ctx->client_ch = dssh_chan_open(ctx->client, &p);
	dssh_chan_params_free(&p);
	return wait_stream_writable(ctx->client_ch);
}

static int
server_accept_subsys_thread(void *arg)
{
	struct open_subsys_ctx *ctx = arg;
	ctx->server_ch = dssh_chan_accept(ctx->server, &accept_cbs_all, 30000);
	return wait_stream_writable(ctx->server_ch);
}


/* ================================================================
 * Session start/stop (~3 tests)
 * ================================================================ */

static int
test_session_start(void)
{
	struct conn_ctx ctx;
	if (conn_setup(&ctx) < 0)
		return TEST_SKIP;

	ASSERT_TRUE(ctx.client->demux_running);
	ASSERT_TRUE(ctx.server->demux_running);

	conn_cleanup(&ctx);
	return TEST_PASS;
}

static int
test_session_stop(void)
{
	struct conn_ctx ctx;
	if (conn_setup(&ctx) < 0)
		return TEST_SKIP;

	/* Close pipes so demux threads' read() returns EOF */
	mock_io_close_c2s(&ctx.io);
	mock_io_close_s2c(&ctx.io);

	dssh_session_stop(ctx.client);
	ASSERT_FALSE(ctx.client->demux_running);

	dssh_session_stop(ctx.server);
	ASSERT_FALSE(ctx.server->demux_running);

	g_active_ctx = NULL;
	dssh_session_cleanup(ctx.server);
	dssh_session_cleanup(ctx.client);
	mock_io_free(&ctx.io);
	dssh_test_reset_global_config();
	return TEST_PASS;
}

static int
test_session_start_stop(void)
{
	struct conn_ctx ctx;
	if (conn_setup(&ctx) < 0)
		return TEST_SKIP;

	ASSERT_TRUE(ctx.client->demux_running);
	ASSERT_TRUE(ctx.server->demux_running);

	mock_io_close_c2s(&ctx.io);
	mock_io_close_s2c(&ctx.io);
	dssh_session_stop(ctx.client);
	dssh_session_stop(ctx.server);

	ASSERT_FALSE(ctx.client->demux_running);
	ASSERT_FALSE(ctx.server->demux_running);

	g_active_ctx = NULL;
	dssh_session_cleanup(ctx.server);
	dssh_session_cleanup(ctx.client);
	mock_io_free(&ctx.io);
	dssh_test_reset_global_config();
	return TEST_PASS;
}

/* ================================================================
 * Channel open -- exec (~5 tests)
 * ================================================================ */

static int
test_open_exec(void)
{
	struct conn_ctx ctx;
	if (conn_setup(&ctx) < 0)
		return TEST_SKIP;

	struct open_exec_ctx oc = {
		.client = ctx.client,
		.server = ctx.server,
		.command = "echo hello",
		.cbs = &accept_cbs_all,
		.accept_timeout = 30000,
	};

	if (open_exec_channel(&oc) < 0) {
		conn_cleanup(&ctx);
		return TEST_FAIL;
	}

	ASSERT_NOT_NULL(oc.client_ch);
	ASSERT_NOT_NULL(oc.server_ch);

	dssh_chan_close(oc.client_ch, 0);
	dssh_chan_close(oc.server_ch, 0);
	conn_cleanup(&ctx);
	return TEST_PASS;
}

static int
test_open_exec_command(void)
{
	struct conn_ctx ctx;
	if (conn_setup(&ctx) < 0)
		return TEST_SKIP;

	struct open_exec_ctx oc = {
		.client = ctx.client,
		.server = ctx.server,
		.command = "ls -la /tmp",
		.cbs = &accept_cbs_all,
		.accept_timeout = 30000,
	};

	if (open_exec_channel(&oc) < 0) {
		conn_cleanup(&ctx);
		return TEST_FAIL;
	}

	/* Verify the command arrived on the server side */
	ASSERT_EQ(dssh_chan_get_type(oc.server_ch), DSSH_CHAN_EXEC);
	ASSERT_NOT_NULL(dssh_chan_get_command(oc.server_ch));
	ASSERT_STR_EQ(dssh_chan_get_command(oc.server_ch), "ls -la /tmp");

	dssh_chan_close(oc.client_ch, 0);
	dssh_chan_close(oc.server_ch, 0);
	conn_cleanup(&ctx);
	return TEST_PASS;
}

/* Reject callback for exec -- returns -1 */
static int
reject_exec_cb(dssh_channel ch, const struct dssh_chan_params *p,
    struct dssh_chan_accept_result *result, void *cbdata)
{
	(void)ch;
	(void)p;
	(void)result;
	(void)cbdata;
	return -1;
}

static struct dssh_chan_accept_cbs reject_exec_cbs = {
	.shell = accept_all_shell,
	.exec = reject_exec_cb,
	.subsystem = accept_all_subsystem,
};

static int
test_open_exec_rejected(void)
{
	struct conn_ctx ctx;
	if (conn_setup(&ctx) < 0)
		return TEST_SKIP;

	struct open_exec_ctx oc = {
		.client = ctx.client,
		.server = ctx.server,
		.command = "rejected_cmd",
		.cbs = &reject_exec_cbs,
		.accept_timeout = 30000,
	};

	thrd_t ct, st;
	ASSERT_THRD_CREATE(&ct, client_open_exec_thread, &oc);
	ASSERT_THRD_CREATE(&st, server_accept_exec_thread, &oc);
	thrd_join(ct, NULL);

	/* After rejection, dssh_chan_accept loops back to wait for the
	 * next CHANNEL_OPEN.  Terminate to unblock it. */
	session_set_terminate(ctx.server);
	thrd_join(st, NULL);

	ASSERT_NULL(oc.client_ch);

	conn_cleanup(&ctx);
	return TEST_PASS;
}

static int
test_open_shell(void)
{
	struct conn_ctx ctx;
	if (conn_setup(&ctx) < 0)
		return TEST_SKIP;

	struct open_shell_ctx oc = {
		.client = ctx.client,
		.server = ctx.server,
		.term = "xterm-256color",
		.cols = 80,
		.rows = 24,
		.wpx = 0,
		.hpx = 0,
		.cbs = &accept_cbs_all,
	};

	thrd_t ct, st;
	ASSERT_THRD_CREATE(&ct, client_open_shell_thread, &oc);
	ASSERT_THRD_CREATE(&st, server_accept_shell_thread, &oc);
	thrd_join(ct, NULL);
	thrd_join(st, NULL);

	ASSERT_NOT_NULL(oc.client_ch);
	ASSERT_NOT_NULL(oc.server_ch);
	ASSERT_EQ(dssh_chan_get_type(oc.server_ch), DSSH_CHAN_SHELL);
	ASSERT_TRUE(pty_received);
	ASSERT_STR_EQ(pty_term, "xterm-256color");
	ASSERT_EQ_U(pty_cols, 80);
	ASSERT_EQ_U(pty_rows, 24);

	dssh_chan_close(oc.client_ch, 0);
	dssh_chan_close(oc.server_ch, 0);
	conn_cleanup(&ctx);
	return TEST_PASS;
}

static int
test_open_subsystem(void)
{
	struct conn_ctx ctx;
	if (conn_setup(&ctx) < 0)
		return TEST_SKIP;

	struct open_subsys_ctx oc = {
		.client = ctx.client,
		.server = ctx.server,
		.subsystem = "sftp",
	};

	thrd_t ct, st;
	ASSERT_THRD_CREATE(&ct, client_open_subsys_thread, &oc);
	ASSERT_THRD_CREATE(&st, server_accept_subsys_thread, &oc);
	thrd_join(ct, NULL);
	thrd_join(st, NULL);

	ASSERT_NOT_NULL(oc.client_ch);
	ASSERT_NOT_NULL(oc.server_ch);

	dssh_chan_close(oc.client_ch, -1);
	dssh_chan_close(oc.server_ch, -1);
	conn_cleanup(&ctx);
	return TEST_PASS;
}

/* ================================================================
 * Session I/O (~8 tests)
 * ================================================================ */

static int
test_session_write_read(void)
{
	struct conn_ctx ctx;
	if (conn_setup(&ctx) < 0)
		return TEST_SKIP;

	struct open_exec_ctx oc = {
		.client = ctx.client,
		.server = ctx.server,
		.command = "cmd",
		.cbs = &accept_cbs_all,
		.accept_timeout = 30000,
	};
	if (open_exec_channel(&oc) < 0) {
		conn_cleanup(&ctx);
		return TEST_SKIP;
	}

	const uint8_t msg[] = "Hello, server!";
	int64_t written = dssh_chan_write(oc.client_ch, 0,
	    msg, sizeof(msg) - 1);
	ASSERT_TRUE(written > 0);

	int ev = dssh_chan_poll(oc.server_ch, DSSH_POLL_READ, 5000);
	ASSERT_TRUE(ev & DSSH_POLL_READ);

	uint8_t buf[256];
	int64_t nread = dssh_chan_read(oc.server_ch, 0,
	    buf, sizeof(buf));
	ASSERT_TRUE(nread > 0);
	ASSERT_EQ(nread, (int64_t)(sizeof(msg) - 1));
	ASSERT_MEM_EQ(buf, msg, sizeof(msg) - 1);

	dssh_chan_close(oc.client_ch, 0);
	dssh_chan_close(oc.server_ch, 0);
	conn_cleanup(&ctx);
	return TEST_PASS;
}

static int
test_session_read_write_reverse(void)
{
	struct conn_ctx ctx;
	if (conn_setup(&ctx) < 0)
		return TEST_SKIP;

	struct open_exec_ctx oc = {
		.client = ctx.client,
		.server = ctx.server,
		.command = "cmd",
		.cbs = &accept_cbs_all,
		.accept_timeout = 30000,
	};
	if (open_exec_channel(&oc) < 0) {
		conn_cleanup(&ctx);
		return TEST_SKIP;
	}

	const uint8_t msg[] = "Hello, client!";
	int64_t written = dssh_chan_write(oc.server_ch, 0,
	    msg, sizeof(msg) - 1);
	ASSERT_TRUE(written > 0);

	int ev = dssh_chan_poll(oc.client_ch, DSSH_POLL_READ, 5000);
	ASSERT_TRUE(ev & DSSH_POLL_READ);

	uint8_t buf[256];
	int64_t nread = dssh_chan_read(oc.client_ch, 0,
	    buf, sizeof(buf));
	ASSERT_TRUE(nread > 0);
	ASSERT_EQ(nread, (int64_t)(sizeof(msg) - 1));
	ASSERT_MEM_EQ(buf, msg, sizeof(msg) - 1);

	dssh_chan_close(oc.client_ch, 0);
	dssh_chan_close(oc.server_ch, 0);
	conn_cleanup(&ctx);
	return TEST_PASS;
}

static int
test_session_write_ext(void)
{
	struct conn_ctx ctx;
	if (conn_setup(&ctx) < 0)
		return TEST_SKIP;

	struct open_exec_ctx oc = {
		.client = ctx.client,
		.server = ctx.server,
		.command = "cmd",
		.cbs = &accept_cbs_all,
		.accept_timeout = 30000,
	};
	if (open_exec_channel(&oc) < 0) {
		conn_cleanup(&ctx);
		return TEST_SKIP;
	}

	const uint8_t stderr_msg[] = "error output";
	int64_t written = dssh_chan_write(oc.server_ch, 1,
	    stderr_msg, sizeof(stderr_msg) - 1);
	ASSERT_TRUE(written > 0);

	int ev = dssh_chan_poll(oc.client_ch, DSSH_POLL_READEXT, 5000);
	ASSERT_TRUE(ev & DSSH_POLL_READEXT);

	uint8_t buf[256];
	int64_t nread = dssh_chan_read(oc.client_ch, 1,
	    buf, sizeof(buf));
	ASSERT_TRUE(nread > 0);
	ASSERT_EQ(nread, (int64_t)(sizeof(stderr_msg) - 1));
	ASSERT_MEM_EQ(buf, stderr_msg, sizeof(stderr_msg) - 1);

	dssh_chan_close(oc.client_ch, 0);
	dssh_chan_close(oc.server_ch, 0);
	conn_cleanup(&ctx);
	return TEST_PASS;
}

static int
test_session_read_ext(void)
{
	struct conn_ctx ctx;
	if (conn_setup(&ctx) < 0)
		return TEST_SKIP;

	struct open_exec_ctx oc = {
		.client = ctx.client,
		.server = ctx.server,
		.command = "cmd",
		.cbs = &accept_cbs_all,
		.accept_timeout = 30000,
	};
	if (open_exec_channel(&oc) < 0) {
		conn_cleanup(&ctx);
		return TEST_SKIP;
	}

	const uint8_t stderr_msg[] = "client stderr";
	int64_t written = dssh_chan_write(oc.client_ch, 1,
	    stderr_msg, sizeof(stderr_msg) - 1);
	ASSERT_TRUE(written > 0);

	int ev = dssh_chan_poll(oc.server_ch, DSSH_POLL_READEXT, 5000);
	ASSERT_TRUE(ev & DSSH_POLL_READEXT);

	uint8_t buf[256];
	int64_t nread = dssh_chan_read(oc.server_ch, 1,
	    buf, sizeof(buf));
	ASSERT_TRUE(nread > 0);
	ASSERT_MEM_EQ(buf, stderr_msg, sizeof(stderr_msg) - 1);

	dssh_chan_close(oc.client_ch, 0);
	dssh_chan_close(oc.server_ch, 0);
	conn_cleanup(&ctx);
	return TEST_PASS;
}

static int
test_session_poll_read(void)
{
	struct conn_ctx ctx;
	if (conn_setup(&ctx) < 0)
		return TEST_SKIP;

	struct open_exec_ctx oc = {
		.client = ctx.client,
		.server = ctx.server,
		.command = "cmd",
		.cbs = &accept_cbs_all,
		.accept_timeout = 30000,
	};
	if (open_exec_channel(&oc) < 0) {
		conn_cleanup(&ctx);
		return TEST_SKIP;
	}

	const uint8_t msg[] = "data";
	dssh_chan_write(oc.client_ch, 0, msg, sizeof(msg) - 1);

	int ev = dssh_chan_poll(oc.server_ch, DSSH_POLL_READ, 5000);
	ASSERT_TRUE(ev & DSSH_POLL_READ);

	dssh_chan_close(oc.client_ch, 0);
	dssh_chan_close(oc.server_ch, 0);
	conn_cleanup(&ctx);
	return TEST_PASS;
}

static int
test_session_poll_timeout(void)
{
	struct conn_ctx ctx;
	if (conn_setup(&ctx) < 0)
		return TEST_SKIP;

	struct open_exec_ctx oc = {
		.client = ctx.client,
		.server = ctx.server,
		.command = "cmd",
		.cbs = &accept_cbs_all,
		.accept_timeout = 30000,
	};
	if (open_exec_channel(&oc) < 0) {
		conn_cleanup(&ctx);
		return TEST_SKIP;
	}

	int ev = dssh_chan_poll(oc.server_ch, DSSH_POLL_READ, 100);
	ASSERT_EQ(ev, 0);

	dssh_chan_close(oc.client_ch, 0);
	dssh_chan_close(oc.server_ch, 0);
	conn_cleanup(&ctx);
	return TEST_PASS;
}

static int
test_session_large_write(void)
{
	struct conn_ctx ctx;
	if (conn_setup(&ctx) < 0)
		return TEST_SKIP;

	struct open_exec_ctx oc = {
		.client = ctx.client,
		.server = ctx.server,
		.command = "cmd",
		.cbs = &accept_cbs_all,
		.accept_timeout = 30000,
	};
	if (open_exec_channel(&oc) < 0) {
		conn_cleanup(&ctx);
		return TEST_SKIP;
	}

	size_t total = 32768;
	uint8_t *data = malloc(total);
	if (data == NULL) {
		dssh_chan_close(oc.client_ch, 0);
		dssh_chan_close(oc.server_ch, 0);
		conn_cleanup(&ctx);
		return TEST_SKIP;
	}
	for (size_t i = 0; i < total; i++)
		data[i] = (uint8_t)(i & 0xFF);

	int64_t written = dssh_chan_write(oc.client_ch, 0, data, total);
	ASSERT_TRUE(written > 0);

	uint8_t *readbuf = malloc(total);
	if (readbuf == NULL) {
		free(data);
		dssh_chan_close(oc.client_ch, 0);
		dssh_chan_close(oc.server_ch, 0);
		conn_cleanup(&ctx);
		return TEST_SKIP;
	}

	size_t got = 0;
	while (got < (size_t)written) {
		int ev = dssh_chan_poll(oc.server_ch, DSSH_POLL_READ, 5000);
		if (ev <= 0)
			break;
		int64_t n = dssh_chan_read(oc.server_ch, 0,
		    &readbuf[got], total - got);
		if (n <= 0)
			break;
		got += (size_t)n;
	}

	ASSERT_EQ_U(got, (size_t)written);
	ASSERT_MEM_EQ(readbuf, data, got);

	free(data);
	free(readbuf);
	dssh_chan_close(oc.client_ch, 0);
	dssh_chan_close(oc.server_ch, 0);
	conn_cleanup(&ctx);
	return TEST_PASS;
}

static int
test_session_multiple_writes(void)
{
	struct conn_ctx ctx;
	if (conn_setup(&ctx) < 0)
		return TEST_SKIP;

	struct open_exec_ctx oc = {
		.client = ctx.client,
		.server = ctx.server,
		.command = "cmd",
		.cbs = &accept_cbs_all,
		.accept_timeout = 30000,
	};
	if (open_exec_channel(&oc) < 0) {
		conn_cleanup(&ctx);
		return TEST_SKIP;
	}

	const uint8_t m1[] = "AAA";
	const uint8_t m2[] = "BBB";
	const uint8_t m3[] = "CCC";
	dssh_chan_write(oc.client_ch, 0, m1, 3);
	dssh_chan_write(oc.client_ch, 0, m2, 3);
	dssh_chan_write(oc.client_ch, 0, m3, 3);

	uint8_t buf[256];
	size_t got = 0;
	while (got < 9) {
		int ev = dssh_chan_poll(oc.server_ch, DSSH_POLL_READ, 5000);
		if (ev <= 0)
			break;
		int64_t n = dssh_chan_read(oc.server_ch, 0,
		    &buf[got], sizeof(buf) - got);
		if (n <= 0)
			break;
		got += (size_t)n;
	}
	ASSERT_EQ_U(got, 9);
	ASSERT_MEM_EQ(buf, "AAABBBCCC", 9);

	dssh_chan_close(oc.client_ch, 0);
	dssh_chan_close(oc.server_ch, 0);
	conn_cleanup(&ctx);
	return TEST_PASS;
}

/* ================================================================
 * Subsystem channel I/O (~4 tests)
 * ================================================================ */

static int
test_subsys_write_read(void)
{
	struct conn_ctx ctx;
	if (conn_setup(&ctx) < 0)
		return TEST_SKIP;

	struct open_subsys_ctx oc = {
		.client = ctx.client,
		.server = ctx.server,
		.subsystem = "test-sub",
	};

	thrd_t ct, st;
	ASSERT_THRD_CREATE(&ct, client_open_subsys_thread, &oc);
	ASSERT_THRD_CREATE(&st, server_accept_subsys_thread, &oc);
	thrd_join(ct, NULL);
	thrd_join(st, NULL);

	if (oc.client_ch == NULL || oc.server_ch == NULL) {
		conn_cleanup(&ctx);
		return TEST_SKIP;
	}

	/* Verify subsystem name arrived */
	ASSERT_EQ(dssh_chan_get_type(oc.server_ch), DSSH_CHAN_SUBSYSTEM);

	/* Both sides use stream API now */
	const uint8_t msg[] = "subsystem message";
	int64_t written = dssh_chan_write(oc.client_ch, 0,
	    msg, sizeof(msg) - 1);
	ASSERT_TRUE(written > 0);

	int ev = dssh_chan_poll(oc.server_ch, DSSH_POLL_READ, 5000);
	ASSERT_TRUE(ev & DSSH_POLL_READ);

	uint8_t buf[256];
	int64_t nread = dssh_chan_read(oc.server_ch, 0,
	    buf, sizeof(buf));
	ASSERT_TRUE(nread > 0);
	ASSERT_EQ(nread, (int64_t)(sizeof(msg) - 1));
	ASSERT_MEM_EQ(buf, msg, sizeof(msg) - 1);

	dssh_chan_close(oc.client_ch, -1);
	dssh_chan_close(oc.server_ch, -1);
	conn_cleanup(&ctx);
	return TEST_PASS;
}

static int
test_subsys_multiple_writes(void)
{
	struct conn_ctx ctx;
	if (conn_setup(&ctx) < 0)
		return TEST_SKIP;

	struct open_subsys_ctx oc = {
		.client = ctx.client,
		.server = ctx.server,
		.subsystem = "test-sub",
	};

	thrd_t ct, st;
	ASSERT_THRD_CREATE(&ct, client_open_subsys_thread, &oc);
	ASSERT_THRD_CREATE(&st, server_accept_subsys_thread, &oc);
	thrd_join(ct, NULL);
	thrd_join(st, NULL);

	if (oc.client_ch == NULL || oc.server_ch == NULL) {
		conn_cleanup(&ctx);
		return TEST_SKIP;
	}

	/* Stream API: writes are concatenated, not message-delimited */
	dssh_chan_write(oc.client_ch, 0,
	    (const uint8_t *)"first", 5);
	dssh_chan_write(oc.client_ch, 0,
	    (const uint8_t *)"second", 6);
	dssh_chan_write(oc.client_ch, 0,
	    (const uint8_t *)"third", 5);

	uint8_t buf[256];
	size_t got = 0;

	while (got < 16) {
		int ev = dssh_chan_poll(oc.server_ch, DSSH_POLL_READ, 5000);
		if (ev <= 0)
			break;
		int64_t n = dssh_chan_read(oc.server_ch, 0,
		    &buf[got], sizeof(buf) - got);
		if (n <= 0)
			break;
		got += (size_t)n;
	}
	ASSERT_EQ_U(got, 16);
	ASSERT_MEM_EQ(buf, "firstsecondthird", 16);

	dssh_chan_close(oc.client_ch, -1);
	dssh_chan_close(oc.server_ch, -1);
	conn_cleanup(&ctx);
	return TEST_PASS;
}

static int
test_subsys_poll(void)
{
	struct conn_ctx ctx;
	if (conn_setup(&ctx) < 0)
		return TEST_SKIP;

	struct open_subsys_ctx oc = {
		.client = ctx.client,
		.server = ctx.server,
		.subsystem = "test-sub",
	};

	thrd_t ct, st;
	ASSERT_THRD_CREATE(&ct, client_open_subsys_thread, &oc);
	ASSERT_THRD_CREATE(&st, server_accept_subsys_thread, &oc);
	thrd_join(ct, NULL);
	thrd_join(st, NULL);

	if (oc.client_ch == NULL || oc.server_ch == NULL) {
		conn_cleanup(&ctx);
		return TEST_SKIP;
	}

	/* No data yet -- poll should timeout */
	int ev = dssh_chan_poll(oc.server_ch, DSSH_POLL_READ, 100);
	ASSERT_EQ(ev, 0);

	/* Send data, poll should return readable */
	dssh_chan_write(oc.client_ch, 0,
	    (const uint8_t *)"x", 1);
	ev = dssh_chan_poll(oc.server_ch, DSSH_POLL_READ, 5000);
	ASSERT_TRUE(ev & DSSH_POLL_READ);

	uint8_t buf[8];
	dssh_chan_read(oc.server_ch, 0, buf, sizeof(buf));

	dssh_chan_close(oc.client_ch, -1);
	dssh_chan_close(oc.server_ch, -1);
	conn_cleanup(&ctx);
	return TEST_PASS;
}

static int
test_subsys_large_write(void)
{
	struct conn_ctx ctx;
	if (conn_setup(&ctx) < 0)
		return TEST_SKIP;

	struct open_subsys_ctx oc = {
		.client = ctx.client,
		.server = ctx.server,
		.subsystem = "test-sub",
	};

	thrd_t ct, st;
	ASSERT_THRD_CREATE(&ct, client_open_subsys_thread, &oc);
	ASSERT_THRD_CREATE(&st, server_accept_subsys_thread, &oc);
	thrd_join(ct, NULL);
	thrd_join(st, NULL);

	if (oc.client_ch == NULL || oc.server_ch == NULL) {
		conn_cleanup(&ctx);
		return TEST_SKIP;
	}

	size_t msglen = 16384;
	uint8_t *data = malloc(msglen);
	ASSERT_NOT_NULL(data);
	for (size_t i = 0; i < msglen; i++)
		data[i] = (uint8_t)(i & 0xFF);

	int64_t written = dssh_chan_write(oc.client_ch, 0,
	    data, msglen);
	ASSERT_TRUE(written > 0);

	uint8_t *readbuf = malloc(msglen);
	ASSERT_NOT_NULL(readbuf);

	size_t got = 0;
	while (got < (size_t)written) {
		int ev = dssh_chan_poll(oc.server_ch, DSSH_POLL_READ, 5000);
		if (ev <= 0)
			break;
		int64_t n = dssh_chan_read(oc.server_ch, 0,
		    &readbuf[got], msglen - got);
		if (n <= 0)
			break;
		got += (size_t)n;
	}
	ASSERT_TRUE(got > 0);
	ASSERT_MEM_EQ(readbuf, data, got);

	free(data);
	free(readbuf);
	dssh_chan_close(oc.client_ch, -1);
	dssh_chan_close(oc.server_ch, -1);
	conn_cleanup(&ctx);
	return TEST_PASS;
}

/* ================================================================
 * Signals (~3 tests)
 * ================================================================ */

static int
test_send_signal(void)
{
	struct conn_ctx ctx;
	if (conn_setup(&ctx) < 0)
		return TEST_SKIP;

	struct open_exec_ctx oc = {
		.client = ctx.client,
		.server = ctx.server,
		.command = "cmd",
		.cbs = &accept_cbs_all,
		.accept_timeout = 30000,
	};
	if (open_exec_channel(&oc) < 0) {
		conn_cleanup(&ctx);
		return TEST_SKIP;
	}

	int res = dssh_chan_send_signal(oc.client_ch, "INT");
	ASSERT_OK(res);

	int ev = dssh_chan_poll(oc.server_ch,
	    DSSH_POLL_EVENT, 5000);
	ASSERT_TRUE(ev & DSSH_POLL_EVENT);

	struct dssh_chan_event event;
	res = dssh_chan_read_event(oc.server_ch, &event);
	ASSERT_OK(res);
	ASSERT_EQ(event.type, DSSH_EVENT_SIGNAL);
	ASSERT_NOT_NULL(event.signal.name);
	ASSERT_STR_EQ(event.signal.name, "INT");

	dssh_chan_close(oc.client_ch, 0);
	dssh_chan_close(oc.server_ch, 0);
	conn_cleanup(&ctx);
	return TEST_PASS;
}

static int
test_signal_after_data(void)
{
	struct conn_ctx ctx;
	if (conn_setup(&ctx) < 0)
		return TEST_SKIP;

	struct open_exec_ctx oc = {
		.client = ctx.client,
		.server = ctx.server,
		.command = "cmd",
		.cbs = &accept_cbs_all,
		.accept_timeout = 30000,
	};
	if (open_exec_channel(&oc) < 0) {
		conn_cleanup(&ctx);
		return TEST_SKIP;
	}

	/* Send data, then a signal */
	const uint8_t msg[] = "before signal";
	dssh_chan_write(oc.client_ch, 0, msg, sizeof(msg) - 1);
	dssh_chan_send_signal(oc.client_ch, "TERM");

	/* Drain data first */
	uint8_t buf[256];
	size_t got = 0;
	while (got < sizeof(msg) - 1) {
		int ev = dssh_chan_poll(oc.server_ch,
		    DSSH_POLL_READ, 5000);
		if (ev <= 0)
			break;
		int64_t n = dssh_chan_read(oc.server_ch, 0,
		    &buf[got], sizeof(buf) - got);
		if (n <= 0)
			break;
		got += (size_t)n;
	}
	ASSERT_EQ_U(got, sizeof(msg) - 1);

	/* Now event should be ready */
	int ev = dssh_chan_poll(oc.server_ch,
	    DSSH_POLL_EVENT, 5000);
	ASSERT_TRUE(ev & DSSH_POLL_EVENT);

	struct dssh_chan_event event;
	dssh_chan_read_event(oc.server_ch, &event);
	ASSERT_EQ(event.type, DSSH_EVENT_SIGNAL);
	ASSERT_NOT_NULL(event.signal.name);
	ASSERT_STR_EQ(event.signal.name, "TERM");

	dssh_chan_close(oc.client_ch, 0);
	dssh_chan_close(oc.server_ch, 0);
	conn_cleanup(&ctx);
	return TEST_PASS;
}

static int
test_window_change(void)
{
	struct conn_ctx ctx;
	if (conn_setup(&ctx) < 0)
		return TEST_SKIP;

	struct open_shell_ctx oc = {
		.client = ctx.client,
		.server = ctx.server,
		.term = "vt100",
		.cols = 80,
		.rows = 24,
		.cbs = &accept_cbs_all,
	};

	thrd_t ct, st;
	ASSERT_THRD_CREATE(&ct, client_open_shell_thread, &oc);
	ASSERT_THRD_CREATE(&st, server_accept_shell_thread, &oc);
	thrd_join(ct, NULL);
	thrd_join(st, NULL);

	if (oc.client_ch == NULL || oc.server_ch == NULL) {
		conn_cleanup(&ctx);
		return TEST_SKIP;
	}

	int res = dssh_chan_send_window_change(oc.client_ch,
	    132, 50, 1056, 800);
	ASSERT_OK(res);

	/* Poll for window-change event */
	int ev = dssh_chan_poll(oc.server_ch,
	    DSSH_POLL_EVENT, 5000);
	ASSERT_TRUE(ev & DSSH_POLL_EVENT);

	struct dssh_chan_event event;
	res = dssh_chan_read_event(oc.server_ch, &event);
	ASSERT_OK(res);
	ASSERT_EQ(event.type, DSSH_EVENT_WINDOW_CHANGE);
	ASSERT_EQ_U(event.window_change.cols, 132);
	ASSERT_EQ_U(event.window_change.rows, 50);
	ASSERT_EQ_U(event.window_change.wpx, 1056);
	ASSERT_EQ_U(event.window_change.hpx, 800);

	dssh_chan_close(oc.client_ch, 0);
	dssh_chan_close(oc.server_ch, 0);
	conn_cleanup(&ctx);
	return TEST_PASS;
}

/* ================================================================
 * Close and exit (~5 tests)
 * ================================================================ */

static int
test_session_close_exit_code(void)
{
	struct conn_ctx ctx;
	if (conn_setup(&ctx) < 0)
		return TEST_SKIP;

	struct open_exec_ctx oc = {
		.client = ctx.client,
		.server = ctx.server,
		.command = "cmd",
		.cbs = &accept_cbs_all,
		.accept_timeout = 30000,
	};
	if (open_exec_channel(&oc) < 0) {
		conn_cleanup(&ctx);
		return TEST_SKIP;
	}

	/* Server closes with exit code 42 */
	int res = dssh_chan_close(oc.server_ch, 42);
	ASSERT_OK(res);

	/* Client should see close; give demux time */
	struct timespec ts = { .tv_sec = 0, .tv_nsec = 20000000 };
	thrd_sleep(&ts, NULL);

	/* Client reads should return 0 (EOF) */
	uint8_t buf[64];
	int ev = dssh_chan_poll(oc.client_ch,
	    DSSH_POLL_READ, 1000);
	if (ev & DSSH_POLL_READ) {
		int64_t n = dssh_chan_read(oc.client_ch, 0,
		    buf, sizeof(buf));
		ASSERT_EQ(n, 0);
	}

	dssh_chan_close(oc.client_ch, 0);
	conn_cleanup(&ctx);
	return TEST_PASS;
}

static int
test_channel_close_subsys(void)
{
	struct conn_ctx ctx;
	if (conn_setup(&ctx) < 0)
		return TEST_SKIP;

	struct open_subsys_ctx oc = {
		.client = ctx.client,
		.server = ctx.server,
		.subsystem = "test-raw",
	};

	thrd_t ct, st;
	ASSERT_THRD_CREATE(&ct, client_open_subsys_thread, &oc);
	ASSERT_THRD_CREATE(&st, server_accept_subsys_thread, &oc);
	thrd_join(ct, NULL);
	thrd_join(st, NULL);

	if (oc.client_ch == NULL || oc.server_ch == NULL) {
		conn_cleanup(&ctx);
		return TEST_SKIP;
	}

	int res = dssh_chan_close(oc.client_ch, -1);
	ASSERT_OK(res);

	/* Server should see close */
	struct timespec ts = { .tv_sec = 0, .tv_nsec = 20000000 };
	thrd_sleep(&ts, NULL);

	dssh_chan_close(oc.server_ch, -1);
	conn_cleanup(&ctx);
	return TEST_PASS;
}

static int
test_eof_detection(void)
{
	struct conn_ctx ctx;
	if (conn_setup(&ctx) < 0)
		return TEST_SKIP;

	struct open_exec_ctx oc = {
		.client = ctx.client,
		.server = ctx.server,
		.command = "cmd",
		.cbs = &accept_cbs_all,
		.accept_timeout = 30000,
	};
	if (open_exec_channel(&oc) < 0) {
		conn_cleanup(&ctx);
		return TEST_SKIP;
	}

	/* Server writes data, then closes (sends exit-status + EOF + CLOSE) */
	const uint8_t msg[] = "final output";
	dssh_chan_write(oc.server_ch, 0, msg, sizeof(msg) - 1);
	dssh_chan_close(oc.server_ch, 0);

	/* Client reads data */
	uint8_t buf[256];
	size_t got = 0;
	for (;;) {
		int ev = dssh_chan_poll(oc.client_ch,
		    DSSH_POLL_READ, 5000);
		if (ev <= 0)
			break;
		int64_t n = dssh_chan_read(oc.client_ch, 0,
		    buf + got, sizeof(buf) - got);
		if (n <= 0)
			break;
		got += (size_t)n;
	}
	ASSERT_TRUE(got >= sizeof(msg) - 1);
	ASSERT_MEM_EQ(buf, msg, sizeof(msg) - 1);

	dssh_chan_close(oc.client_ch, 0);
	conn_cleanup(&ctx);
	return TEST_PASS;
}

static int
test_eof_half_close(void)
{
	struct conn_ctx ctx;
	if (conn_setup(&ctx) < 0)
		return TEST_SKIP;

	struct open_exec_ctx oc = {
		.client = ctx.client,
		.server = ctx.server,
		.command = "cmd",
		.cbs = &accept_cbs_all,
		.accept_timeout = 30000,
	};
	if (open_exec_channel(&oc) < 0) {
		conn_cleanup(&ctx);
		return TEST_SKIP;
	}

	/* Server sends data */
	const uint8_t msg[] = "server data after client eof";
	dssh_chan_write(oc.server_ch, 0, msg, sizeof(msg) - 1);

	/* Client should be able to read */
	uint8_t buf[256];
	size_t got = 0;
	while (got < sizeof(msg) - 1) {
		int ev = dssh_chan_poll(oc.client_ch,
		    DSSH_POLL_READ, 1000);
		if (ev <= 0)
			break;
		int64_t n = dssh_chan_read(oc.client_ch, 0,
		    &buf[got], sizeof(buf) - got);
		if (n <= 0)
			break;
		got += (size_t)n;
	}
	ASSERT_EQ_U(got, sizeof(msg) - 1);
	ASSERT_MEM_EQ(buf, msg, sizeof(msg) - 1);

	dssh_chan_close(oc.client_ch, 0);
	dssh_chan_close(oc.server_ch, 0);
	conn_cleanup(&ctx);
	return TEST_PASS;
}

/*
 * Stream I/O follows non-blocking POSIX semantics:
 *   read returns 0 only on real EOF; NOMORE for empty-no-EOF.
 *   write returns 0 only when bufsz == 0; NOMORE for window-full.
 *   Peek (buf == NULL) keeps the legacy "0 = no bytes available".
 */
static int
test_read_nomore_vs_eof(void)
{
	struct conn_ctx ctx;
	if (conn_setup(&ctx) < 0)
		return TEST_SKIP;

	struct open_exec_ctx oc = {
		.client = ctx.client,
		.server = ctx.server,
		.command = "cmd",
		.cbs = &accept_cbs_all,
		.accept_timeout = 30000,
	};
	if (open_exec_channel(&oc) < 0) {
		conn_cleanup(&ctx);
		return TEST_SKIP;
	}

	uint8_t buf[64];

	/* Empty buffer, no EOF: NOMORE. */
	int64_t n = dssh_chan_read(oc.client_ch, 0, buf, sizeof(buf));
	ASSERT_EQ(n, DSSH_ERROR_NOMORE);

	/* Peek never returns NOMORE -- always non-negative count. */
	int64_t avail = dssh_chan_read(oc.client_ch, 0, NULL, 0);
	ASSERT_EQ(avail, 0);

	/* After data arrives, real bytes come back. */
	const uint8_t payload[] = "ping";
	dssh_chan_write(oc.server_ch, 0, payload, sizeof(payload) - 1);
	int ev = dssh_chan_poll(oc.client_ch, DSSH_POLL_READ, 5000);
	ASSERT_TRUE(ev & DSSH_POLL_READ);
	n = dssh_chan_read(oc.client_ch, 0, buf, sizeof(buf));
	ASSERT_EQ(n, (int64_t)(sizeof(payload) - 1));

	/* Drained again with no EOF yet: NOMORE, not 0. */
	n = dssh_chan_read(oc.client_ch, 0, buf, sizeof(buf));
	ASSERT_EQ(n, DSSH_ERROR_NOMORE);

	/* Write side: 0-byte write returns 0; window-full returns
	 * NOMORE; normal write returns the byte count. */
	n = dssh_chan_write(oc.client_ch, 0, payload, 0);
	ASSERT_EQ(n, 0);

	uint32_t saved_window      = oc.client_ch->remote_window;
	oc.client_ch->remote_window = 0;
	n = dssh_chan_write(oc.client_ch, 0, payload, sizeof(payload) - 1);
	ASSERT_EQ(n, DSSH_ERROR_NOMORE);
	oc.client_ch->remote_window = saved_window;

	/* Server closes -> EOF arrives.  Poll wakes; read returns 0. */
	dssh_chan_close(oc.server_ch, 0);
	ev = dssh_chan_poll(oc.client_ch, DSSH_POLL_READ, 5000);
	ASSERT_TRUE(ev & DSSH_POLL_READ);
	n = dssh_chan_read(oc.client_ch, 0, buf, sizeof(buf));
	ASSERT_EQ(n, 0);

	dssh_chan_close(oc.client_ch, 0);
	conn_cleanup(&ctx);
	return TEST_PASS;
}

static int
test_graceful_disconnect(void)
{
	struct conn_ctx ctx;
	if (conn_setup(&ctx) < 0)
		return TEST_SKIP;

	struct open_exec_ctx oc = {
		.client = ctx.client,
		.server = ctx.server,
		.command = "cmd",
		.cbs = &accept_cbs_all,
		.accept_timeout = 30000,
	};
	if (open_exec_channel(&oc) < 0) {
		conn_cleanup(&ctx);
		return TEST_SKIP;
	}

	/* Client terminates the session */
	dssh_session_terminate(ctx.client);

	/* Give demux time to see termination */
	struct timespec ts = { .tv_sec = 0, .tv_nsec = 20000000 };
	thrd_sleep(&ts, NULL);

	/* Server poll should return error or see termination */
	int ev = dssh_chan_poll(oc.server_ch, DSSH_POLL_READ, 500);
	/* Either an error or 0 timeout is acceptable after peer disconnect */
	ASSERT_TRUE(ev <= 0 || (ev & DSSH_POLL_READ));

	/* Clean up without calling chan_close for already-terminated channels */
	conn_cleanup(&ctx);
	return TEST_PASS;
}

/* ================================================================
 * Multiple channels (~3 tests)
 * ================================================================ */

struct two_channel_server_ctx {
	dssh_session server;
	dssh_channel ch1;
	dssh_channel ch2;
	const struct dssh_chan_accept_cbs *cbs;
};

static int
server_accept_two_channels(void *arg)
{
	struct two_channel_server_ctx *sctx = arg;

	sctx->ch1 = dssh_chan_accept(sctx->server, sctx->cbs, 30000);
	sctx->ch2 = dssh_chan_accept(sctx->server, sctx->cbs, 30000);

	return 0;
}

static int
test_two_channels(void)
{
	struct conn_ctx ctx;
	if (conn_setup(&ctx) < 0)
		return TEST_SKIP;

	struct two_channel_server_ctx sctx = {
		.server = ctx.server,
		.cbs = &accept_cbs_all,
	};

	thrd_t st;
	ASSERT_THRD_CREATE(&st, server_accept_two_channels, &sctx);

	struct dssh_chan_params p1;
	dssh_chan_params_init(&p1, DSSH_CHAN_EXEC);
	dssh_chan_params_set_command(&p1, "cmd1");
	dssh_chan_params_set_pty(&p1, false);
	dssh_channel c1 = dssh_chan_open(ctx.client, &p1);
	dssh_chan_params_free(&p1);

	struct dssh_chan_params p2;
	dssh_chan_params_init(&p2, DSSH_CHAN_EXEC);
	dssh_chan_params_set_command(&p2, "cmd2");
	dssh_chan_params_set_pty(&p2, false);
	dssh_channel c2 = dssh_chan_open(ctx.client, &p2);
	dssh_chan_params_free(&p2);

	thrd_join(st, NULL);

	if (c1 == NULL || c2 == NULL || sctx.ch1 == NULL || sctx.ch2 == NULL) {
		if (c1)
			dssh_chan_close(c1, 0);
		if (c2)
			dssh_chan_close(c2, 0);
		if (sctx.ch1)
			dssh_chan_close(sctx.ch1, 0);
		if (sctx.ch2)
			dssh_chan_close(sctx.ch2, 0);
		conn_cleanup(&ctx);
		return TEST_FAIL;
	}

	/* Write on channel 1, read on channel 1 */
	int ev = dssh_chan_poll(c1, DSSH_POLL_WRITE, 5000);
	ASSERT_TRUE(ev & DSSH_POLL_WRITE);
	const uint8_t m1[] = "chan1data";
	dssh_chan_write(c1, 0, m1, 8);

	/* Write on channel 2, read on channel 2 */
	ev = dssh_chan_poll(c2, DSSH_POLL_WRITE, 5000);
	ASSERT_TRUE(ev & DSSH_POLL_WRITE);
	const uint8_t m2[] = "chan2data";
	dssh_chan_write(c2, 0, m2, 8);

	uint8_t buf[256];

	ev = dssh_chan_poll(sctx.ch1, DSSH_POLL_READ, 5000);
	ASSERT_TRUE(ev & DSSH_POLL_READ);
	int64_t n = dssh_chan_read(sctx.ch1, 0, buf, sizeof(buf));
	ASSERT_EQ(n, 8);
	ASSERT_MEM_EQ(buf, "chan1data", 8);

	ev = dssh_chan_poll(sctx.ch2, DSSH_POLL_READ, 5000);
	ASSERT_TRUE(ev & DSSH_POLL_READ);
	n = dssh_chan_read(sctx.ch2, 0, buf, sizeof(buf));
	ASSERT_EQ(n, 8);
	ASSERT_MEM_EQ(buf, "chan2data", 8);

	dssh_chan_close(c1, 0);
	dssh_chan_close(c2, 0);
	dssh_chan_close(sctx.ch1, 0);
	dssh_chan_close(sctx.ch2, 0);
	conn_cleanup(&ctx);
	return TEST_PASS;
}

/*
 * Accept-reject test: server calls dssh_chan_accept once with
 * reject_exec_cbs.  Client opens exec (rejected by callback),
 * then shell (accepted).  Accept loops internally past the
 * rejected channel and returns the shell channel.
 */
struct accept_reject_server_ctx {
	dssh_session server;
	dssh_channel ch;
};

static int
server_accept_loops_on_reject(void *arg)
{
	struct accept_reject_server_ctx *sctx = arg;

	/* Single accept: loops past rejected exec, returns shell */
	sctx->ch = dssh_chan_accept(sctx->server,
	    &reject_exec_cbs, 30000);

	return 0;
}

static int
test_accept_reject(void)
{
	struct conn_ctx ctx;
	if (conn_setup(&ctx) < 0)
		return TEST_SKIP;

	struct accept_reject_server_ctx sctx = {
		.server = ctx.server,
	};

	thrd_t st;
	ASSERT_THRD_CREATE(&st, server_accept_loops_on_reject, &sctx);

	/* First channel: exec, rejected by server callback */
	struct dssh_chan_params p1;
	dssh_chan_params_init(&p1, DSSH_CHAN_EXEC);
	dssh_chan_params_set_command(&p1, "rejected");
	dssh_chan_params_set_pty(&p1, false);
	dssh_channel c1 = dssh_chan_open(ctx.client, &p1);
	dssh_chan_params_free(&p1);

	/* Second channel: shell, accepted */
	struct dssh_chan_params p2;
	dssh_chan_params_init(&p2, DSSH_CHAN_SHELL);
	dssh_chan_params_set_pty(&p2, false);
	dssh_channel c2 = dssh_chan_open(ctx.client, &p2);
	dssh_chan_params_free(&p2);

	thrd_join(st, NULL);

	ASSERT_NULL(c1);
	ASSERT_NOT_NULL(c2);
	ASSERT_NOT_NULL(sctx.ch);
	ASSERT_EQ(DSSH_CHAN_SHELL, dssh_chan_get_type(sctx.ch));

	dssh_chan_close(c2, 0);
	dssh_chan_close(sctx.ch, 0);
	conn_cleanup(&ctx);
	return TEST_PASS;
}

static int
test_channel_independence(void)
{
	struct conn_ctx ctx;
	if (conn_setup(&ctx) < 0)
		return TEST_SKIP;

	struct two_channel_server_ctx sctx = {
		.server = ctx.server,
		.cbs = &accept_cbs_all,
	};

	thrd_t st;
	ASSERT_THRD_CREATE(&st, server_accept_two_channels, &sctx);

	struct dssh_chan_params p1;
	dssh_chan_params_init(&p1, DSSH_CHAN_EXEC);
	dssh_chan_params_set_command(&p1, "cmd1");
	dssh_chan_params_set_pty(&p1, false);
	dssh_channel c1 = dssh_chan_open(ctx.client, &p1);
	dssh_chan_params_free(&p1);

	struct dssh_chan_params p2;
	dssh_chan_params_init(&p2, DSSH_CHAN_EXEC);
	dssh_chan_params_set_command(&p2, "cmd2");
	dssh_chan_params_set_pty(&p2, false);
	dssh_channel c2 = dssh_chan_open(ctx.client, &p2);
	dssh_chan_params_free(&p2);

	thrd_join(st, NULL);

	if (c1 == NULL || c2 == NULL || sctx.ch1 == NULL || sctx.ch2 == NULL) {
		if (c1)
			dssh_chan_close(c1, 0);
		if (c2)
			dssh_chan_close(c2, 0);
		if (sctx.ch1)
			dssh_chan_close(sctx.ch1, 0);
		if (sctx.ch2)
			dssh_chan_close(sctx.ch2, 0);
		conn_cleanup(&ctx);
		return TEST_FAIL;
	}

	/* Close channel 1 */
	dssh_chan_close(c1, 0);
	dssh_chan_close(sctx.ch1, 0);

	/* Channel 2 should still work */
	const uint8_t msg[] = "still alive";
	int write_ev = dssh_chan_poll(c2, DSSH_POLL_WRITE, 5000);
	ASSERT_TRUE(write_ev & DSSH_POLL_WRITE);
	ASSERT_EQ(dssh_chan_write(c2, 0, msg, sizeof(msg) - 1), sizeof(msg) - 1);

	uint8_t buf[256];
	int ev = dssh_chan_poll(sctx.ch2, DSSH_POLL_READ, 5000);
	ASSERT_TRUE(ev & DSSH_POLL_READ);
	int64_t n = dssh_chan_read(sctx.ch2, 0, buf, sizeof(buf));
	ASSERT_TRUE(n > 0);
	ASSERT_MEM_EQ(buf, "still alive", (size_t)n);

	dssh_chan_close(c2, 0);
	dssh_chan_close(sctx.ch2, 0);
	conn_cleanup(&ctx);
	return TEST_PASS;
}

/* ================================================================
 * Thread safety (~2 tests)
 * ================================================================ */

struct writer_ctx {
	dssh_channel ch;
	size_t total_written;
	int result;
};

static int
writer_thread_func(void *arg)
{
	struct writer_ctx *wc = arg;
	uint8_t buf[128];
	memset(buf, 'W', sizeof(buf));
	wc->total_written = 0;
	wc->result = 0;

	for (int i = 0; i < 100; i++) {
		int ev = dssh_chan_poll(wc->ch, DSSH_POLL_WRITE, 5000);
		if (ev < 0) {
			wc->result = ev;
			break;
		}
		if (!(ev & DSSH_POLL_WRITE))
			continue;
		int64_t w = dssh_chan_write(wc->ch, 0, buf, sizeof(buf));
		if (w < 0) {
			wc->result = (int)w;
			break;
		}
		wc->total_written += (size_t)w;
	}
	return 0;
}

struct reader_ctx {
	dssh_channel ch;
	size_t total_read;
	int result;
};

static int
reader_thread_func(void *arg)
{
	struct reader_ctx *rc = arg;
	uint8_t buf[256];
	rc->total_read = 0;
	rc->result = 0;

	for (;;) {
		int ev = dssh_chan_poll(rc->ch, DSSH_POLL_READ, 2000);
		if (ev <= 0)
			break;
		if (!(ev & DSSH_POLL_READ))
			continue;
		int64_t n = dssh_chan_read(rc->ch, 0, buf, sizeof(buf));
		if (n <= 0)
			break;
		rc->total_read += (size_t)n;
	}
	return 0;
}

static int
test_concurrent_write_read(void)
{
	struct conn_ctx ctx;
	if (conn_setup(&ctx) < 0)
		return TEST_SKIP;

	struct open_exec_ctx oc = {
		.client = ctx.client,
		.server = ctx.server,
		.command = "cmd",
		.cbs = &accept_cbs_all,
		.accept_timeout = 30000,
	};
	if (open_exec_channel(&oc) < 0) {
		conn_cleanup(&ctx);
		return TEST_SKIP;
	}

	struct writer_ctx wc = {
		.ch = oc.client_ch,
	};
	struct reader_ctx rc = {
		.ch = oc.server_ch,
	};

	thrd_t wt, rt;
	ASSERT_THRD_CREATE(&rt, reader_thread_func, &rc);
	ASSERT_THRD_CREATE(&wt, writer_thread_func, &wc);

	thrd_join(wt, NULL);

	/* After writer finishes, close the channel so reader sees EOF */
	dssh_chan_close(oc.client_ch, 0);

	thrd_join(rt, NULL);

	ASSERT_TRUE(wc.total_written > 0);
	ASSERT_TRUE(rc.total_read > 0);
	/* Reader should have received at least some of what was written */
	ASSERT_TRUE(rc.total_read <= wc.total_written);

	dssh_chan_close(oc.server_ch, 0);
	conn_cleanup(&ctx);
	return TEST_PASS;
}

static int
test_terminate_during_io(void)
{
	struct conn_ctx ctx;
	if (conn_setup(&ctx) < 0)
		return TEST_SKIP;

	struct open_exec_ctx oc = {
		.client = ctx.client,
		.server = ctx.server,
		.command = "cmd",
		.cbs = &accept_cbs_all,
		.accept_timeout = 30000,
	};
	if (open_exec_channel(&oc) < 0) {
		conn_cleanup(&ctx);
		return TEST_SKIP;
	}

	/* Start a reader that will block on poll */
	struct reader_ctx rc = {
		.ch = oc.server_ch,
	};
	thrd_t rt;
	ASSERT_THRD_CREATE(&rt, reader_thread_func, &rc);

	/* Brief delay, then terminate */
	struct timespec ts = { .tv_sec = 0, .tv_nsec = 10000000 };
	thrd_sleep(&ts, NULL);

	dssh_session_terminate(ctx.server);

	thrd_join(rt, NULL);

	/* Reader should have exited cleanly (no crash) */
	ASSERT_EQ_U(rc.total_read, 0);

	conn_cleanup(&ctx);
	return TEST_PASS;
}


/* ================================================================
 * Unexpected connection drop
 * ================================================================ */

static int
test_connection_drop_during_read(void)
{
	struct conn_ctx ctx;
	if (conn_setup(&ctx) < 0)
		return TEST_SKIP;

	struct open_exec_ctx oc = {
		.client = ctx.client,
		.server = ctx.server,
		.command = "cmd",
		.cbs = &accept_cbs_all,
		.accept_timeout = 30000,
	};
	if (open_exec_channel(&oc) < 0) {
		conn_cleanup(&ctx);
		return TEST_SKIP;
	}

	/* Start a reader blocked on poll */
	struct reader_ctx rc = {
		.ch = oc.server_ch,
	};
	thrd_t rt;
	ASSERT_THRD_CREATE(&rt, reader_thread_func, &rc);

	/* Brief delay to let reader block */
	struct timespec ts = { .tv_sec = 0, .tv_nsec = 10000000 };
	thrd_sleep(&ts, NULL);

	/* Kill the connection */
	mock_io_close_c2s(&ctx.io);
	mock_io_close_s2c(&ctx.io);

	thrd_join(rt, NULL);

	/* Session should be terminated */
	ASSERT_TRUE(dssh_session_is_terminated(ctx.server));

	/* Cleanup must not crash */
	conn_cleanup(&ctx);
	return TEST_PASS;
}

static int
test_connection_drop_during_write(void)
{
	struct conn_ctx ctx;
	if (conn_setup(&ctx) < 0)
		return TEST_SKIP;

	struct open_exec_ctx oc = {
		.client = ctx.client,
		.server = ctx.server,
		.command = "cmd",
		.cbs = &accept_cbs_all,
		.accept_timeout = 30000,
	};
	if (open_exec_channel(&oc) < 0) {
		conn_cleanup(&ctx);
		return TEST_SKIP;
	}

	/* Kill the connection */
	mock_io_close_c2s(&ctx.io);
	mock_io_close_s2c(&ctx.io);

	/* Brief delay for demux to notice I/O error */
	struct timespec ts = { .tv_sec = 0, .tv_nsec = 100000000 };
	thrd_sleep(&ts, NULL);

	/* Write should fail (not hang, not crash) */
	const uint8_t data[] = "after drop";
	int64_t w = dssh_chan_write(oc.client_ch, 0,
	    data, sizeof(data) - 1);
	ASSERT_TRUE(w <= 0);

	/* Poll should indicate error or return immediately */
	int ev = dssh_chan_poll(oc.client_ch, DSSH_POLL_READ, 100);
	(void)ev;  /* any result is OK as long as it doesn't hang */

	conn_cleanup(&ctx);
	return TEST_PASS;
}

static int
test_cleanup_after_drop(void)
{
	struct conn_ctx ctx;
	if (conn_setup(&ctx) < 0)
		return TEST_SKIP;

	struct open_exec_ctx oc = {
		.client = ctx.client,
		.server = ctx.server,
		.command = "cmd",
		.cbs = &accept_cbs_all,
		.accept_timeout = 30000,
	};
	if (open_exec_channel(&oc) < 0) {
		conn_cleanup(&ctx);
		return TEST_SKIP;
	}

	mock_io_close_c2s(&ctx.io);
	mock_io_close_s2c(&ctx.io);

	/* This must not hang or crash */
	g_active_ctx = NULL;
	dssh_session_cleanup(ctx.server);
	dssh_session_cleanup(ctx.client);
	ctx.server = NULL;
	ctx.client = NULL;
	mock_io_free(&ctx.io);
	dssh_test_reset_global_config();
	return TEST_PASS;
}

/* ================================================================
 * Coverage: session_start twice, accept timeout
 * ================================================================ */

static int
test_session_start_twice(void)
{
	struct conn_ctx ctx;
	if (conn_setup(&ctx) < 0)
		return TEST_FAIL;

	/* Already started in conn_setup. Start again should fail. */
	int res = dssh_session_start(ctx.client);
	ASSERT_TRUE(res < 0);

	conn_cleanup(&ctx);
	return TEST_PASS;
}

static int
test_accept_timeout_zero(void)
{
	struct conn_ctx ctx;
	if (conn_setup(&ctx) < 0)
		return TEST_FAIL;

	/* accept with timeout_ms=0 -- non-blocking, should return immediately */
	dssh_channel ch = dssh_chan_accept(ctx.server, &accept_cbs_all, 0);
	ASSERT_NULL(ch);

	conn_cleanup(&ctx);
	return TEST_PASS;
}

static int
test_accept_timeout_short(void)
{
	struct conn_ctx ctx;
	if (conn_setup(&ctx) < 0)
		return TEST_FAIL;

	dssh_channel ch = dssh_chan_accept(ctx.server, &accept_cbs_all, 50);
	ASSERT_NULL(ch);

	conn_cleanup(&ctx);
	return TEST_PASS;
}

static int
test_session_poll_timeout_zero(void)
{
	struct conn_ctx ctx;
	if (conn_setup(&ctx) < 0)
		return TEST_FAIL;

	struct open_exec_ctx oc = {
		.client = ctx.client,
		.server = ctx.server,
		.command = "echo polltest",
		.cbs = &accept_cbs_all,
		.accept_timeout = 30000,
	};
	if (open_exec_channel(&oc) < 0) {
		conn_cleanup(&ctx);
		return TEST_FAIL;
	}
	if (oc.client_ch == NULL || oc.server_ch == NULL) {
		conn_cleanup(&ctx);
		return TEST_FAIL;
	}

	/* Poll with timeout=0 on client -- no data pending */
	int events = dssh_chan_poll(oc.client_ch,
	    DSSH_POLL_READ | DSSH_POLL_WRITE, 0);
	/* Should return 0 (no events) or just DSSH_POLL_WRITE */
	ASSERT_TRUE(events >= 0);

	conn_cleanup(&ctx);
	return TEST_PASS;
}

/* ================================================================
 * Auto-reject channel types -- covers demux_channel_open branches.
 * ================================================================ */

static size_t
build_channel_open(uint8_t *buf, size_t bufsz, const char *type,
    uint32_t sender_ch, uint32_t window, uint32_t max_packet)
{
	size_t pos = 0;
	buf[pos++] = SSH_MSG_CHANNEL_OPEN;
	size_t tlen = strlen(type);
	dssh_serialize_uint32((uint32_t)tlen, buf, bufsz, &pos);
	memcpy(&buf[pos], type, tlen);
	pos += tlen;
	dssh_serialize_uint32(sender_ch, buf, bufsz, &pos);
	dssh_serialize_uint32(window, buf, bufsz, &pos);
	dssh_serialize_uint32(max_packet, buf, bufsz, &pos);
	return pos;
}

static int
test_auto_reject_x11(void)
{
	struct conn_ctx ctx;
	if (conn_setup(&ctx) < 0)
		return TEST_FAIL;

	uint8_t msg[64];
	size_t len = build_channel_open(msg, sizeof(msg), "x11", 0, 65536, 32768);
	ASSERT_OK(send_packet(ctx.server, msg, len, NULL));

	dssh_channel ch = dssh_chan_accept(ctx.client, &accept_cbs_all, 0);
	ASSERT_NULL(ch);

	conn_cleanup(&ctx);
	return TEST_PASS;
}

static int
test_auto_reject_forwarded_tcpip(void)
{
	struct conn_ctx ctx;
	if (conn_setup(&ctx) < 0)
		return TEST_FAIL;

	uint8_t msg[64];
	size_t len = build_channel_open(msg, sizeof(msg), "forwarded-tcpip",
	    0, 65536, 32768);
	ASSERT_OK(send_packet(ctx.server, msg, len, NULL));

	dssh_channel ch = dssh_chan_accept(ctx.client, &accept_cbs_all, 0);
	ASSERT_NULL(ch);

	conn_cleanup(&ctx);
	return TEST_PASS;
}

static int
test_auto_reject_direct_tcpip(void)
{
	struct conn_ctx ctx;
	if (conn_setup(&ctx) < 0)
		return TEST_FAIL;

	uint8_t msg[64];
	size_t len = build_channel_open(msg, sizeof(msg), "direct-tcpip",
	    0, 65536, 32768);
	ASSERT_OK(send_packet(ctx.server, msg, len, NULL));

	dssh_channel ch = dssh_chan_accept(ctx.client, &accept_cbs_all, 0);
	ASSERT_NULL(ch);

	conn_cleanup(&ctx);
	return TEST_PASS;
}

static int
test_auto_reject_session_from_server(void)
{
	struct conn_ctx ctx;
	if (conn_setup(&ctx) < 0)
		return TEST_FAIL;

	uint8_t msg[64];
	size_t len = build_channel_open(msg, sizeof(msg), "session",
	    0, 65536, 32768);
	ASSERT_OK(send_packet(ctx.server, msg, len, NULL));

	dssh_channel ch = dssh_chan_accept(ctx.client, &accept_cbs_all, 0);
	ASSERT_NULL(ch);

	conn_cleanup(&ctx);
	return TEST_PASS;
}

/* ================================================================
 * WINDOW_ADJUST from peer
 * ================================================================ */

static int
test_window_adjust_from_peer(void)
{
	struct conn_ctx ctx;
	if (conn_setup(&ctx) < 0)
		return TEST_FAIL;

	struct open_exec_ctx oc = {
		.client = ctx.client,
		.server = ctx.server,
		.command = "echo watest",
		.cbs = &accept_cbs_all,
		.accept_timeout = 30000,
	};
	if (open_exec_channel(&oc) < 0 || oc.client_ch == NULL || oc.server_ch == NULL) {
		conn_cleanup(&ctx);
		return TEST_FAIL;
	}

	uint8_t msg[16];
	size_t pos = 0;
	msg[pos++] = SSH_MSG_CHANNEL_WINDOW_ADJUST;
	dssh_serialize_uint32(oc.client_ch->local_id, msg, sizeof(msg), &pos);
	dssh_serialize_uint32(1000, msg, sizeof(msg), &pos);
	ASSERT_OK(send_packet(ctx.server, msg, pos, NULL));

	/* Verify session still works */
	uint8_t data[] = "after window adjust";
	int64_t res = dssh_chan_write(oc.server_ch, 0,
	    data, sizeof(data));
	ASSERT_TRUE(res > 0);

	conn_cleanup(&ctx);
	return TEST_PASS;
}

/* ================================================================
 * Data after EOF
 * ================================================================ */

static int
test_data_after_eof(void)
{
	struct conn_ctx ctx;
	if (conn_setup(&ctx) < 0)
		return TEST_FAIL;

	struct open_exec_ctx oc = {
		.client = ctx.client,
		.server = ctx.server,
		.command = "echo eoftest",
		.cbs = &accept_cbs_all,
		.accept_timeout = 30000,
	};
	if (open_exec_channel(&oc) < 0 || oc.client_ch == NULL || oc.server_ch == NULL) {
		conn_cleanup(&ctx);
		return TEST_FAIL;
	}

	/* Server sends EOF to client */
	{
		uint8_t msg[8];
		size_t pos = 0;
		msg[pos++] = SSH_MSG_CHANNEL_EOF;
		dssh_serialize_uint32(oc.client_ch->local_id, msg, sizeof(msg), &pos);
		ASSERT_OK(send_packet(ctx.server, msg, pos, NULL));
	}

	/* Now server sends CHANNEL_DATA after EOF -- should be discarded */
	{
		uint8_t msg[32];
		size_t pos = 0;
		msg[pos++] = SSH_MSG_CHANNEL_DATA;
		dssh_serialize_uint32(oc.client_ch->local_id, msg, sizeof(msg), &pos);
		dssh_serialize_uint32(5, msg, sizeof(msg), &pos);
		memcpy(&msg[pos], "stale", 5);
		pos += 5;
		ASSERT_OK(send_packet(ctx.server, msg, pos, NULL));
	}

	/* Client should observe EOF (poll surfaces READ-ready, read
	 * returns 0).  Polling here also guarantees the demux has
	 * processed both the EOF and the stale-data drop before we
	 * sample the bytebuf. */
	int ev = dssh_chan_poll(oc.client_ch, DSSH_POLL_READ, 5000);
	ASSERT_TRUE(ev & DSSH_POLL_READ);
	uint8_t buf[32];
	int64_t got = dssh_chan_read(oc.client_ch, 0, buf, sizeof(buf));
	ASSERT_EQ(got, 0);

	conn_cleanup(&ctx);
	return TEST_PASS;
}

/* ================================================================
 * Truncated channel messages
 * ================================================================ */

static int
test_truncated_channel_data(void)
{
	struct conn_ctx ctx;
	if (conn_setup(&ctx) < 0)
		return TEST_FAIL;

	struct open_exec_ctx oc = {
		.client = ctx.client,
		.server = ctx.server,
		.command = "echo trunctest",
		.cbs = &accept_cbs_all,
		.accept_timeout = 30000,
	};
	if (open_exec_channel(&oc) < 0 || oc.client_ch == NULL || oc.server_ch == NULL) {
		conn_cleanup(&ctx);
		return TEST_FAIL;
	}

	uint8_t msg[8];
	size_t pos = 0;
	msg[pos++] = SSH_MSG_CHANNEL_DATA;
	dssh_serialize_uint32(oc.client_ch->local_id, msg, sizeof(msg), &pos);
	msg[pos++] = 0;
	ASSERT_OK(send_packet(ctx.server, msg, pos, NULL));

	/* Session should still be functional */
	uint8_t data[] = "still works";
	int64_t res = dssh_chan_write(oc.server_ch, 0,
	    data, sizeof(data));
	ASSERT_TRUE(res > 0);

	conn_cleanup(&ctx);
	return TEST_PASS;
}

/* ================================================================
 * Extended data after EOF
 * ================================================================ */

static int
test_ext_data_after_eof(void)
{
	struct conn_ctx ctx;
	if (conn_setup(&ctx) < 0)
		return TEST_FAIL;

	struct open_exec_ctx oc = {
		.client = ctx.client,
		.server = ctx.server,
		.command = "echo exteof",
		.cbs = &accept_cbs_all,
		.accept_timeout = 30000,
	};
	if (open_exec_channel(&oc) < 0 || oc.client_ch == NULL || oc.server_ch == NULL) {
		conn_cleanup(&ctx);
		return TEST_FAIL;
	}

	{
		uint8_t msg[8];
		size_t pos = 0;
		msg[pos++] = SSH_MSG_CHANNEL_EOF;
		dssh_serialize_uint32(oc.client_ch->local_id, msg, sizeof(msg), &pos);
		ASSERT_OK(send_packet(ctx.server, msg, pos, NULL));
	}

	{
		uint8_t msg[32];
		size_t pos = 0;
		msg[pos++] = SSH_MSG_CHANNEL_EXTENDED_DATA;
		dssh_serialize_uint32(oc.client_ch->local_id, msg, sizeof(msg), &pos);
		dssh_serialize_uint32(1, msg, sizeof(msg), &pos);
		dssh_serialize_uint32(5, msg, sizeof(msg), &pos);
		memcpy(&msg[pos], "stale", 5);
		pos += 5;
		ASSERT_OK(send_packet(ctx.server, msg, pos, NULL));
	}

	conn_cleanup(&ctx);
	return TEST_PASS;
}

/* ================================================================
 * Truncated CHANNEL_EXTENDED_DATA
 * ================================================================ */

static int
test_truncated_ext_data(void)
{
	struct conn_ctx ctx;
	if (conn_setup(&ctx) < 0)
		return TEST_FAIL;

	struct open_exec_ctx oc = {
		.client = ctx.client,
		.server = ctx.server,
		.command = "echo truncext",
		.cbs = &accept_cbs_all,
		.accept_timeout = 30000,
	};
	if (open_exec_channel(&oc) < 0 || oc.client_ch == NULL || oc.server_ch == NULL) {
		conn_cleanup(&ctx);
		return TEST_FAIL;
	}

	uint8_t msg[12];
	size_t pos = 0;
	msg[pos++] = SSH_MSG_CHANNEL_EXTENDED_DATA;
	dssh_serialize_uint32(oc.client_ch->local_id, msg, sizeof(msg), &pos);
	dssh_serialize_uint32(1, msg, sizeof(msg), &pos);
	ASSERT_OK(send_packet(ctx.server, msg, pos, NULL));

	uint8_t data[] = "ok";
	int64_t res = dssh_chan_write(oc.server_ch, 0, data, sizeof(data));
	ASSERT_TRUE(res > 0);

	conn_cleanup(&ctx);
	return TEST_PASS;
}

/* ================================================================
 * Channel request with want_reply=true
 * ================================================================ */

static int
test_channel_request_want_reply(void)
{
	struct conn_ctx ctx;
	if (conn_setup(&ctx) < 0)
		return TEST_FAIL;

	struct open_exec_ctx oc = {
		.client = ctx.client,
		.server = ctx.server,
		.command = "echo reqtest",
		.cbs = &accept_cbs_all,
		.accept_timeout = 30000,
	};
	if (open_exec_channel(&oc) < 0 || oc.client_ch == NULL || oc.server_ch == NULL) {
		conn_cleanup(&ctx);
		return TEST_FAIL;
	}

	{
		uint8_t msg[64];
		size_t pos = 0;
		msg[pos++] = SSH_MSG_CHANNEL_REQUEST;
		dssh_serialize_uint32(oc.client_ch->local_id, msg, sizeof(msg), &pos);
		dssh_serialize_uint32(7, msg, sizeof(msg), &pos);
		memcpy(&msg[pos], "unknown", 7);
		pos += 7;
		msg[pos++] = 1;
		ASSERT_OK(send_packet(ctx.server, msg, pos, NULL));
	}

	uint8_t data[] = "after request";
	int64_t res = dssh_chan_write(oc.server_ch, 0,
	    data, sizeof(data));
	ASSERT_TRUE(res > 0);

	conn_cleanup(&ctx);
	return TEST_PASS;
}

/* ================================================================
 * Data with dlen > local_window
 * ================================================================ */

static int
test_data_exceeds_window(void)
{
	struct conn_ctx ctx;
	if (conn_setup(&ctx) < 0)
		return TEST_FAIL;

	struct open_exec_ctx oc = {
		.client = ctx.client,
		.server = ctx.server,
		.command = "echo wintest",
		.cbs = &accept_cbs_all,
		.accept_timeout = 30000,
	};
	if (open_exec_channel(&oc) < 0 || oc.client_ch == NULL || oc.server_ch == NULL) {
		conn_cleanup(&ctx);
		return TEST_FAIL;
	}

	{
		uint8_t msg[32];
		size_t pos = 0;
		msg[pos++] = SSH_MSG_CHANNEL_DATA;
		dssh_serialize_uint32(oc.client_ch->local_id, msg, sizeof(msg), &pos);
		dssh_serialize_uint32(UINT32_MAX, msg, sizeof(msg), &pos);
		memcpy(&msg[pos], "x", 1);
		pos += 1;
		ASSERT_OK(send_packet(ctx.server, msg, pos, NULL));
	}

	/* The extended-data parser has a different fixed header size and
	 * must reject the same overflow attempt independently. */
	{
		uint8_t msg[32];
		size_t pos = 0;
		msg[pos++] = SSH_MSG_CHANNEL_EXTENDED_DATA;
		dssh_serialize_uint32(oc.client_ch->local_id, msg, sizeof(msg), &pos);
		dssh_serialize_uint32(1, msg, sizeof(msg), &pos);
		dssh_serialize_uint32(UINT32_MAX, msg, sizeof(msg), &pos);
		msg[pos++] = 'x';
		ASSERT_OK(send_packet(ctx.server, msg, pos, NULL));
	}

	/* Malformed packets must not poison the channel. */
	uint8_t data[] = "still works";
	int64_t res = dssh_chan_write(oc.server_ch, 0, data, sizeof(data));
	ASSERT_TRUE(res > 0);

	conn_cleanup(&ctx);
	return TEST_PASS;
}

/* ================================================================
 * Coverage: poll WRITE readiness
 * ================================================================ */

static int
test_session_poll_write(void)
{
	struct conn_ctx ctx;
	if (conn_setup(&ctx) < 0)
		return TEST_SKIP;

	struct open_exec_ctx oc = {
		.client = ctx.client,
		.server = ctx.server,
		.command = "echo poll_write",
		.cbs = &accept_cbs_all,
		.accept_timeout = 30000,
	};
	if (open_exec_channel(&oc) < 0 || oc.client_ch == NULL || oc.server_ch == NULL) {
		conn_cleanup(&ctx);
		return TEST_FAIL;
	}

	int ev = dssh_chan_poll(oc.client_ch, DSSH_POLL_WRITE, 0);
	ASSERT_TRUE(ev & DSSH_POLL_WRITE);

	ev = dssh_chan_poll(oc.client_ch, DSSH_POLL_WRITE, 100);
	ASSERT_TRUE(ev & DSSH_POLL_WRITE);

	dssh_chan_close(oc.server_ch, 0);
	dssh_chan_close(oc.client_ch, 0);
	conn_cleanup(&ctx);
	return TEST_PASS;
}

/* ================================================================
 * Coverage: write edge cases
 * ================================================================ */

static int
test_session_write_after_eof(void)
{
	struct conn_ctx ctx;
	if (conn_setup(&ctx) < 0)
		return TEST_SKIP;

	struct open_exec_ctx oc = {
		.client = ctx.client,
		.server = ctx.server,
		.command = "echo eof_write",
		.cbs = &accept_cbs_all,
		.accept_timeout = 30000,
	};
	if (open_exec_channel(&oc) < 0 || oc.client_ch == NULL || oc.server_ch == NULL) {
		conn_cleanup(&ctx);
		return TEST_FAIL;
	}

	oc.client_ch->eof_sent = true;
	int64_t w = dssh_chan_write(oc.client_ch, 0,
	    (const uint8_t *)"after", 5);
	ASSERT_TRUE(w < 0);

	oc.client_ch->eof_sent = false;
	dssh_chan_close(oc.server_ch, 0);
	dssh_chan_close(oc.client_ch, 0);
	conn_cleanup(&ctx);
	return TEST_PASS;
}

static int
test_session_write_ext_after_eof(void)
{
	struct conn_ctx ctx;
	if (conn_setup(&ctx) < 0)
		return TEST_SKIP;

	struct open_exec_ctx oc = {
		.client = ctx.client,
		.server = ctx.server,
		.command = "echo ext_eof",
		.cbs = &accept_cbs_all,
		.accept_timeout = 30000,
	};
	if (open_exec_channel(&oc) < 0 || oc.client_ch == NULL || oc.server_ch == NULL) {
		conn_cleanup(&ctx);
		return TEST_FAIL;
	}

	oc.server_ch->eof_sent = true;
	int64_t w = dssh_chan_write(oc.server_ch, 1,
	    (const uint8_t *)"after", 5);
	ASSERT_TRUE(w < 0);

	oc.server_ch->eof_sent = false;
	dssh_chan_close(oc.server_ch, 0);
	dssh_chan_close(oc.client_ch, 0);
	conn_cleanup(&ctx);
	return TEST_PASS;
}

/* ================================================================
 * Coverage: subsystem write edge cases
 * ================================================================ */

static int
test_subsys_write_after_close(void)
{
	struct conn_ctx ctx;
	if (conn_setup(&ctx) < 0)
		return TEST_SKIP;

	struct open_subsys_ctx oc = {
		.client = ctx.client,
		.server = ctx.server,
		.subsystem = "test-sub",
	};
	thrd_t ct, st;
	ASSERT_THRD_CREATE(&ct, client_open_subsys_thread, &oc);
	ASSERT_THRD_CREATE(&st, server_accept_subsys_thread, &oc);
	thrd_join(ct, NULL);
	thrd_join(st, NULL);

	if (oc.client_ch == NULL || oc.server_ch == NULL) {
		conn_cleanup(&ctx);
		return TEST_SKIP;
	}

	oc.server_ch->eof_sent = true;
	int64_t w = dssh_chan_write(oc.server_ch, 0,
	    (const uint8_t *)"after", 5);
	ASSERT_TRUE(w < 0);

	oc.server_ch->eof_sent = false;
	dssh_chan_close(oc.client_ch, -1);
	dssh_chan_close(oc.server_ch, -1);
	conn_cleanup(&ctx);
	return TEST_PASS;
}

/* ================================================================
 * Coverage: subsystem poll edge cases
 * ================================================================ */

static int
test_subsys_poll_write(void)
{
	struct conn_ctx ctx;
	if (conn_setup(&ctx) < 0)
		return TEST_SKIP;

	struct open_subsys_ctx oc = {
		.client = ctx.client,
		.server = ctx.server,
		.subsystem = "test-sub",
	};
	thrd_t ct, st;
	ASSERT_THRD_CREATE(&ct, client_open_subsys_thread, &oc);
	ASSERT_THRD_CREATE(&st, server_accept_subsys_thread, &oc);
	thrd_join(ct, NULL);
	thrd_join(st, NULL);

	if (oc.client_ch == NULL || oc.server_ch == NULL) {
		conn_cleanup(&ctx);
		return TEST_SKIP;
	}

	int ev = dssh_chan_poll(oc.client_ch, DSSH_POLL_WRITE, 0);
	ASSERT_TRUE(ev & DSSH_POLL_WRITE);

	ev = dssh_chan_poll(oc.client_ch, DSSH_POLL_WRITE, 100);
	ASSERT_TRUE(ev & DSSH_POLL_WRITE);

	dssh_chan_close(oc.client_ch, -1);
	dssh_chan_close(oc.server_ch, -1);
	conn_cleanup(&ctx);
	return TEST_PASS;
}

static int
test_subsys_poll_timeout_zero(void)
{
	struct conn_ctx ctx;
	if (conn_setup(&ctx) < 0)
		return TEST_SKIP;

	struct open_subsys_ctx oc = {
		.client = ctx.client,
		.server = ctx.server,
		.subsystem = "test-sub",
	};
	thrd_t ct, st;
	ASSERT_THRD_CREATE(&ct, client_open_subsys_thread, &oc);
	ASSERT_THRD_CREATE(&st, server_accept_subsys_thread, &oc);
	thrd_join(ct, NULL);
	thrd_join(st, NULL);

	if (oc.client_ch == NULL || oc.server_ch == NULL) {
		conn_cleanup(&ctx);
		return TEST_SKIP;
	}

	int ev = dssh_chan_poll(oc.server_ch, DSSH_POLL_READ, 0);
	ASSERT_EQ(ev, 0);

	dssh_chan_close(oc.client_ch, -1);
	dssh_chan_close(oc.server_ch, -1);
	conn_cleanup(&ctx);
	return TEST_PASS;
}

/* ================================================================
 * Coverage: signal read when empty, interleave tests
 * ================================================================ */

static int
test_read_event_empty(void)
{
	struct conn_ctx ctx;
	if (conn_setup(&ctx) < 0)
		return TEST_SKIP;

	struct open_exec_ctx oc = {
		.client = ctx.client,
		.server = ctx.server,
		.command = "echo sig_empty",
		.cbs = &accept_cbs_all,
		.accept_timeout = 30000,
	};
	if (open_exec_channel(&oc) < 0 || oc.client_ch == NULL || oc.server_ch == NULL) {
		conn_cleanup(&ctx);
		return TEST_FAIL;
	}

	/* No event pending -- should return NOMORE */
	struct dssh_chan_event event;
	int res = dssh_chan_read_event(oc.server_ch, &event);
	ASSERT_EQ(res, DSSH_ERROR_NOMORE);

	dssh_chan_close(oc.server_ch, 0);
	dssh_chan_close(oc.client_ch, 0);
	conn_cleanup(&ctx);
	return TEST_PASS;
}

static int
test_signal_interleave_read(void)
{
	struct conn_ctx ctx;
	if (conn_setup(&ctx) < 0)
		return TEST_SKIP;

	struct open_exec_ctx oc = {
		.client = ctx.client,
		.server = ctx.server,
		.command = "echo siginterleave",
		.cbs = &accept_cbs_all,
		.accept_timeout = 30000,
	};
	if (open_exec_channel(&oc) < 0 || oc.client_ch == NULL || oc.server_ch == NULL) {
		conn_cleanup(&ctx);
		return TEST_FAIL;
	}

	/* Client sends data, then a signal, then more data */
	const uint8_t before[] = "BEFORE";
	int64_t w = dssh_chan_write(oc.client_ch, 0,
	    before, sizeof(before) - 1);
	ASSERT_TRUE(w > 0);

	int res = dssh_chan_send_signal(oc.client_ch, "INT");
	ASSERT_OK(res);

	const uint8_t after[] = "AFTER";
	w = dssh_chan_write(oc.client_ch, 0,
	    after, sizeof(after) - 1);
	ASSERT_TRUE(w > 0);

	int ev = dssh_chan_poll(oc.server_ch, DSSH_POLL_READ, 5000);
	ASSERT_TRUE(ev & DSSH_POLL_READ);

	/* Brief delay so signal + second data arrive through demux */
	struct timespec ts = { .tv_sec = 0, .tv_nsec = 10000000 };
	thrd_sleep(&ts, NULL);

	/* New API: data flows freely, events are separate.
	 * Read all data (BEFORE + AFTER), then read signal event. */
	uint8_t buf[64];
	size_t total_read = 0;

	for (int i = 0; i < 50; i++) {
		ev = dssh_chan_poll(oc.server_ch, DSSH_POLL_READ, 200);

		if (ev <= 0)
			break;
		if (!(ev & DSSH_POLL_READ))
			break;

		int64_t n = dssh_chan_read(oc.server_ch, 0,
		    buf + total_read, sizeof(buf) - total_read);

		if (n <= 0)
			break;
		total_read += (size_t)n;
		if (total_read >= sizeof(before) - 1 + sizeof(after) - 1)
			break;
	}
	ASSERT_EQ_U(total_read, sizeof(before) - 1 + sizeof(after) - 1);
	ASSERT_MEM_EQ(buf, "BEFOREAFTER", 11);

	/* Signal event available separately */
	ev = dssh_chan_poll(oc.server_ch, DSSH_POLL_EVENT, 5000);
	ASSERT_TRUE(ev & DSSH_POLL_EVENT);

	struct dssh_chan_event event;

	res = dssh_chan_read_event(oc.server_ch, &event);
	ASSERT_OK(res);
	ASSERT_EQ(event.type, DSSH_EVENT_SIGNAL);
	ASSERT_STR_EQ(event.signal.name, "INT");

	dssh_chan_close(oc.server_ch, 0);
	dssh_chan_close(oc.client_ch, 0);
	conn_cleanup(&ctx);
	return TEST_PASS;
}

static int
test_signal_interleave_read_ext(void)
{
	struct conn_ctx ctx;
	if (conn_setup(&ctx) < 0)
		return TEST_SKIP;

	struct open_exec_ctx oc = {
		.client = ctx.client,
		.server = ctx.server,
		.command = "echo sigext",
		.cbs = &accept_cbs_all,
		.accept_timeout = 30000,
	};
	if (open_exec_channel(&oc) < 0 || oc.client_ch == NULL || oc.server_ch == NULL) {
		conn_cleanup(&ctx);
		return TEST_FAIL;
	}

	/* Server sends stderr data, then a signal, then more stderr data */
	const uint8_t before[] = "ERR1";
	int64_t w = dssh_chan_write(oc.server_ch, 1,
	    before, sizeof(before) - 1);
	ASSERT_TRUE(w > 0);

	int res = dssh_chan_send_signal(oc.server_ch, "TERM");
	ASSERT_OK(res);

	const uint8_t after[] = "ERR2";
	w = dssh_chan_write(oc.server_ch, 1,
	    after, sizeof(after) - 1);
	ASSERT_TRUE(w > 0);

	int ev = dssh_chan_poll(oc.client_ch, DSSH_POLL_READEXT, 5000);
	ASSERT_TRUE(ev & DSSH_POLL_READEXT);

	struct timespec ts = { .tv_sec = 0, .tv_nsec = 10000000 };
	thrd_sleep(&ts, NULL);

	/* New API: data flows freely, events separate.
	 * Read all stderr data, then read signal event. */
	uint8_t buf[64];
	size_t total_read = 0;

	for (int i = 0; i < 50; i++) {
		ev = dssh_chan_poll(oc.client_ch, DSSH_POLL_READEXT, 200);

		if (ev <= 0)
			break;
		if (!(ev & DSSH_POLL_READEXT))
			break;

		int64_t n = dssh_chan_read(oc.client_ch, 1,
		    buf + total_read, sizeof(buf) - total_read);

		if (n <= 0)
			break;
		total_read += (size_t)n;
		if (total_read >= sizeof(before) - 1 + sizeof(after) - 1)
			break;
	}
	ASSERT_EQ_U(total_read, sizeof(before) - 1 + sizeof(after) - 1);
	ASSERT_MEM_EQ(buf, "ERR1ERR2", 8);

	ev = dssh_chan_poll(oc.client_ch, DSSH_POLL_EVENT, 5000);
	ASSERT_TRUE(ev & DSSH_POLL_EVENT);

	struct dssh_chan_event event;

	res = dssh_chan_read_event(oc.client_ch, &event);
	ASSERT_OK(res);
	ASSERT_EQ(event.type, DSSH_EVENT_SIGNAL);
	ASSERT_STR_EQ(event.signal.name, "TERM");

	dssh_chan_close(oc.server_ch, 0);
	dssh_chan_close(oc.client_ch, 0);
	conn_cleanup(&ctx);
	return TEST_PASS;
}

/* ================================================================
 * Coverage: truncated CHANNEL_OPEN
 * ================================================================ */

static int
test_truncated_channel_open(void)
{
	/*
	 * Malformed CHANNEL_OPEN now sends OPEN_FAILURE (when
	 * sender-channel is extractable) then disconnects.
	 * Test each truncation point individually.
	 */
	/* Case 1: too short to read type_len -- disconnect only */
	{
		struct conn_ctx ctx;
		if (conn_setup(&ctx) < 0)
			return TEST_FAIL;

		uint8_t msg[2];
		msg[0] = SSH_MSG_CHANNEL_OPEN;
		msg[1] = 0;
		ASSERT_OK(send_packet(ctx.client, msg, 2, NULL));

		struct timespec ts = { .tv_sec = 0, .tv_nsec = 50000000 };
		thrd_sleep(&ts, NULL);

		ASSERT_TRUE(!ctx.server->demux_running);
		conn_cleanup(&ctx);
	}

	/* Case 2: type_len exceeds payload -- disconnect only */
	{
		struct conn_ctx ctx;
		if (conn_setup(&ctx) < 0)
			return TEST_FAIL;

		uint8_t msg[8];
		size_t pos = 0;
		msg[pos++] = SSH_MSG_CHANNEL_OPEN;
		dssh_serialize_uint32(100, msg, sizeof(msg), &pos);
		ASSERT_OK(send_packet(ctx.client, msg, pos, NULL));

		struct timespec ts = { .tv_sec = 0, .tv_nsec = 50000000 };
		thrd_sleep(&ts, NULL);

		ASSERT_TRUE(!ctx.server->demux_running);
		conn_cleanup(&ctx);
	}

	/* Case 3: type OK but missing sender/window/maxpacket --
	 * sends OPEN_FAILURE (sender-channel extractable) then disconnects */
	{
		struct conn_ctx ctx;
		if (conn_setup(&ctx) < 0)
			return TEST_FAIL;

		uint8_t msg[16];
		size_t pos = 0;
		msg[pos++] = SSH_MSG_CHANNEL_OPEN;
		dssh_serialize_uint32(4, msg, sizeof(msg), &pos);
		memcpy(&msg[pos], "test", 4);
		pos += 4;
		/* Include sender-channel so OPEN_FAILURE can be sent */
		dssh_serialize_uint32(42, msg, sizeof(msg), &pos);
		ASSERT_OK(send_packet(ctx.client, msg, pos, NULL));

		struct timespec ts = { .tv_sec = 0, .tv_nsec = 50000000 };
		thrd_sleep(&ts, NULL);

		ASSERT_TRUE(!ctx.server->demux_running);
		conn_cleanup(&ctx);
	}

	return TEST_PASS;
}

/* ================================================================
 * Coverage: truncated CHANNEL_REQUEST
 * ================================================================ */

static int
test_truncated_channel_request(void)
{
	/*
	 * Malformed CHANNEL_REQUEST now sends CHANNEL_FAILURE
	 * then disconnects.  Test truncated type-length.
	 */
	struct conn_ctx ctx;
	if (conn_setup(&ctx) < 0)
		return TEST_FAIL;

	struct open_exec_ctx oc = {
		.client = ctx.client,
		.server = ctx.server,
		.command = "echo chanreq",
		.cbs = &accept_cbs_all,
		.accept_timeout = 30000,
	};
	if (open_exec_channel(&oc) < 0 || oc.client_ch == NULL || oc.server_ch == NULL) {
		conn_cleanup(&ctx);
		return TEST_FAIL;
	}

	{
		uint8_t msg[8];
		size_t pos = 0;
		msg[pos++] = SSH_MSG_CHANNEL_REQUEST;
		dssh_serialize_uint32(oc.server_ch->local_id, msg, sizeof(msg), &pos);
		ASSERT_OK(send_packet(ctx.client, msg, pos, NULL));
	}

	struct timespec ts = { .tv_sec = 0, .tv_nsec = 50000000 };
	thrd_sleep(&ts, NULL);

	ASSERT_TRUE(!ctx.server->demux_running);

	conn_cleanup(&ctx);
	return TEST_PASS;
}

/* ================================================================
 * Coverage: short payload
 * ================================================================ */

static int
test_demux_short_payload(void)
{
	struct conn_ctx ctx;
	if (conn_setup(&ctx) < 0)
		return TEST_FAIL;

	{
		uint8_t msg[4];
		size_t pos = 0;
		msg[pos++] = SSH_MSG_CHANNEL_DATA;
		msg[pos++] = 0;
		msg[pos++] = 0;
		ASSERT_OK(send_packet(ctx.client, msg, pos, NULL));
	}

	{
		uint8_t msg[2];
		msg[0] = SSH_MSG_CHANNEL_CLOSE;
		msg[1] = 0;
		ASSERT_OK(send_packet(ctx.client, msg, 2, NULL));
	}

	struct timespec ts = { .tv_sec = 0, .tv_nsec = 10000000 };
	thrd_sleep(&ts, NULL);

	ASSERT_TRUE(ctx.server->demux_running);

	conn_cleanup(&ctx);
	return TEST_PASS;
}

/* ================================================================
 * Coverage: poll READ when empty (timeout=0)
 * ================================================================ */

static int
test_session_poll_read_empty(void)
{
	struct conn_ctx ctx;
	if (conn_setup(&ctx) < 0)
		return TEST_SKIP;

	struct open_exec_ctx oc = {
		.client = ctx.client,
		.server = ctx.server,
		.command = "echo poll_empty",
		.cbs = &accept_cbs_all,
		.accept_timeout = 30000,
	};
	if (open_exec_channel(&oc) < 0 || oc.client_ch == NULL || oc.server_ch == NULL) {
		conn_cleanup(&ctx);
		return TEST_FAIL;
	}

	int ev = dssh_chan_poll(oc.server_ch, DSSH_POLL_READ, 0);
	ASSERT_EQ(ev, 0);

	ev = dssh_chan_poll(oc.server_ch, DSSH_POLL_READEXT, 0);
	ASSERT_EQ(ev, 0);

	ev = dssh_chan_poll(oc.server_ch, DSSH_POLL_EVENT, 0);
	ASSERT_EQ(ev, 0);

	dssh_chan_close(oc.server_ch, 0);
	dssh_chan_close(oc.client_ch, 0);
	conn_cleanup(&ctx);
	return TEST_PASS;
}

/* ================================================================
 * Coverage: write with window=0 and max_packet clamping
 * ================================================================ */

static int
test_session_write_window_zero(void)
{
	struct conn_ctx ctx;
	if (conn_setup(&ctx) < 0)
		return TEST_SKIP;

	struct open_exec_ctx oc = {
		.client = ctx.client,
		.server = ctx.server,
		.command = "echo winzero",
		.cbs = &accept_cbs_all,
		.accept_timeout = 30000,
	};
	if (open_exec_channel(&oc) < 0 || oc.client_ch == NULL || oc.server_ch == NULL) {
		conn_cleanup(&ctx);
		return TEST_FAIL;
	}

	oc.client_ch->remote_window = 0;
	int64_t w = dssh_chan_write(oc.client_ch, 0,
	    (const uint8_t *)"data", 4);
	ASSERT_EQ(w, DSSH_ERROR_NOMORE);

	oc.client_ch->remote_window = 2;
	w = dssh_chan_write(oc.client_ch, 0,
	    (const uint8_t *)"data", 4);
	ASSERT_EQ(w, 2);

	oc.client_ch->remote_window = 1000;
	uint32_t orig_max = oc.client_ch->remote_max_packet;
	oc.client_ch->remote_max_packet = 3;
	w = dssh_chan_write(oc.client_ch, 0,
	    (const uint8_t *)"data", 4);
	ASSERT_EQ(w, 3);
	oc.client_ch->remote_max_packet = orig_max;

	dssh_chan_close(oc.server_ch, 0);
	dssh_chan_close(oc.client_ch, 0);
	conn_cleanup(&ctx);
	return TEST_PASS;
}

static int
test_session_write_ext_window_zero(void)
{
	struct conn_ctx ctx;
	if (conn_setup(&ctx) < 0)
		return TEST_SKIP;

	struct open_exec_ctx oc = {
		.client = ctx.client,
		.server = ctx.server,
		.command = "echo extwinzero",
		.cbs = &accept_cbs_all,
		.accept_timeout = 30000,
	};
	if (open_exec_channel(&oc) < 0 || oc.client_ch == NULL || oc.server_ch == NULL) {
		conn_cleanup(&ctx);
		return TEST_FAIL;
	}

	oc.server_ch->remote_window = 0;
	int64_t w = dssh_chan_write(oc.server_ch, 1,
	    (const uint8_t *)"data", 4);
	ASSERT_EQ(w, DSSH_ERROR_NOMORE);

	oc.server_ch->remote_window = 2;
	w = dssh_chan_write(oc.server_ch, 1,
	    (const uint8_t *)"data", 4);
	ASSERT_EQ(w, 2);

	oc.server_ch->remote_window = 1000;
	uint32_t orig_max = oc.server_ch->remote_max_packet;
	oc.server_ch->remote_max_packet = 3;
	w = dssh_chan_write(oc.server_ch, 1,
	    (const uint8_t *)"data", 4);
	ASSERT_EQ(w, 3);
	oc.server_ch->remote_max_packet = orig_max;

	dssh_chan_close(oc.server_ch, 0);
	dssh_chan_close(oc.client_ch, 0);
	conn_cleanup(&ctx);
	return TEST_PASS;
}

/* ================================================================
 * Coverage: accept with negative timeout (blocking)
 * ================================================================ */

struct accept_blocking_ctx {
	struct conn_ctx *ctx;
	_Atomic bool entered;
};

static int
accept_blocking_thread(void *arg)
{
	struct accept_blocking_ctx *bc = arg;
	atomic_store(&bc->entered, true);
	dssh_chan_accept(bc->ctx->server, &accept_cbs_all, -1);
	return 0;
}

static int
test_accept_timeout_negative(void)
{
	struct conn_ctx ctx;
	if (conn_setup(&ctx) < 0)
		return TEST_SKIP;

	struct accept_blocking_ctx bc = {
		.ctx = &ctx,
	};
	atomic_init(&bc.entered, false);

	thrd_t at;
	ASSERT_THRD_CREATE(&at, accept_blocking_thread, &bc);

	/* Wait for the accept thread to enter dssh_chan_accept */
	while (!atomic_load(&bc.entered))
		thrd_yield();

	conn_cleanup(&ctx);
	thrd_join(at, NULL);
	return TEST_PASS;
}

/* ================================================================
 * Coverage: server rejects exec request via callback
 * ================================================================ */

static int
test_setup_exec_rejected_by_callback(void)
{
	struct conn_ctx ctx;
	if (conn_setup(&ctx) < 0)
		return TEST_SKIP;

	struct open_exec_ctx oc = {
		.client = ctx.client,
		.server = ctx.server,
		.command = "rejected_exec",
		.cbs = &reject_exec_cbs,
		.accept_timeout = 30000,
	};

	thrd_t ct, st;
	ASSERT_THRD_CREATE(&ct, client_open_exec_thread, &oc);
	ASSERT_THRD_CREATE(&st, server_accept_exec_thread, &oc);
	thrd_join(ct, NULL);

	/* After rejection, dssh_chan_accept loops back to wait for the
	 * next CHANNEL_OPEN.  Terminate to unblock it. */
	session_set_terminate(ctx.server);
	thrd_join(st, NULL);

	ASSERT_NULL(oc.client_ch);
	ASSERT_NULL(oc.server_ch);

	conn_cleanup(&ctx);
	return TEST_PASS;
}

/* ================================================================
 * Coverage: truncated OPEN_CONFIRMATION
 * ================================================================ */

static int
test_truncated_open_confirmation(void)
{
	struct conn_ctx ctx;
	if (conn_setup(&ctx) < 0)
		return TEST_FAIL;

	{
		uint8_t msg[8];
		size_t pos = 0;
		msg[pos++] = SSH_MSG_CHANNEL_OPEN_CONFIRMATION;
		dssh_serialize_uint32(0, msg, sizeof(msg), &pos);
		ASSERT_OK(send_packet(ctx.client, msg, pos, NULL));
	}

	struct timespec ts = { .tv_sec = 0, .tv_nsec = 10000000 };
	thrd_sleep(&ts, NULL);
	ASSERT_TRUE(ctx.server->demux_running);

	conn_cleanup(&ctx);
	return TEST_PASS;
}

/* ================================================================
 * Coverage: OPEN_CONFIRMATION for unknown channel
 * ================================================================ */

static int
test_open_confirmation_unknown_channel(void)
{
	struct conn_ctx ctx;
	if (conn_setup(&ctx) < 0)
		return TEST_FAIL;

	{
		uint8_t msg[20];
		size_t pos = 0;
		msg[pos++] = SSH_MSG_CHANNEL_OPEN_CONFIRMATION;
		dssh_serialize_uint32(9999, msg, sizeof(msg), &pos);
		dssh_serialize_uint32(0, msg, sizeof(msg), &pos);
		dssh_serialize_uint32(0x200000, msg, sizeof(msg), &pos);
		dssh_serialize_uint32(0x8000, msg, sizeof(msg), &pos);
		ASSERT_OK(send_packet(ctx.client, msg, pos, NULL));
	}

	struct timespec ts = { .tv_sec = 0, .tv_nsec = 10000000 };
	thrd_sleep(&ts, NULL);
	ASSERT_TRUE(ctx.server->demux_running);

	conn_cleanup(&ctx);
	return TEST_PASS;
}

/* ================================================================
 * Coverage: CHANNEL_SUCCESS/FAILURE when no request pending
 * ================================================================ */

static int
test_channel_success_no_request(void)
{
	struct conn_ctx ctx;
	if (conn_setup(&ctx) < 0)
		return TEST_FAIL;

	struct open_exec_ctx oc = {
		.client = ctx.client,
		.server = ctx.server,
		.command = "echo success",
		.cbs = &accept_cbs_all,
		.accept_timeout = 30000,
	};
	if (open_exec_channel(&oc) < 0 || oc.client_ch == NULL || oc.server_ch == NULL) {
		conn_cleanup(&ctx);
		return TEST_FAIL;
	}

	{
		uint8_t msg[8];
		size_t pos = 0;
		msg[pos++] = SSH_MSG_CHANNEL_SUCCESS;
		dssh_serialize_uint32(oc.server_ch->local_id, msg, sizeof(msg), &pos);
		ASSERT_OK(send_packet(ctx.client, msg, pos, NULL));
	}
	{
		uint8_t msg[8];
		size_t pos = 0;
		msg[pos++] = SSH_MSG_CHANNEL_FAILURE;
		dssh_serialize_uint32(oc.server_ch->local_id, msg, sizeof(msg), &pos);
		ASSERT_OK(send_packet(ctx.client, msg, pos, NULL));
	}

	struct timespec ts = { .tv_sec = 0, .tv_nsec = 10000000 };
	thrd_sleep(&ts, NULL);
	ASSERT_TRUE(ctx.server->demux_running);

	dssh_chan_close(oc.server_ch, 0);
	dssh_chan_close(oc.client_ch, 0);
	conn_cleanup(&ctx);
	return TEST_PASS;
}

/* ================================================================
 * Coverage: double EOF/CLOSE
 * ================================================================ */

static int
test_double_eof_close(void)
{
	struct conn_ctx ctx;
	if (conn_setup(&ctx) < 0)
		return TEST_SKIP;

	struct open_exec_ctx oc = {
		.client = ctx.client,
		.server = ctx.server,
		.command = "echo double",
		.cbs = &accept_cbs_all,
		.accept_timeout = 30000,
	};
	if (open_exec_channel(&oc) < 0 || oc.client_ch == NULL || oc.server_ch == NULL) {
		conn_cleanup(&ctx);
		return TEST_FAIL;
	}

	dssh_chan_close(oc.server_ch, 0);
	dssh_chan_close(oc.client_ch, 0);

	conn_cleanup(&ctx);
	return TEST_PASS;
}

/* ================================================================
 * Coverage: DATA with dlen > actual payload bytes
 * ================================================================ */

static int
test_data_dlen_exceeds_payload(void)
{
	struct conn_ctx ctx;
	if (conn_setup(&ctx) < 0)
		return TEST_FAIL;

	struct open_exec_ctx oc = {
		.client = ctx.client,
		.server = ctx.server,
		.command = "echo dlen",
		.cbs = &accept_cbs_all,
		.accept_timeout = 30000,
	};
	if (open_exec_channel(&oc) < 0 || oc.client_ch == NULL || oc.server_ch == NULL) {
		conn_cleanup(&ctx);
		return TEST_FAIL;
	}

	{
		uint8_t msg[32];
		size_t pos = 0;
		msg[pos++] = SSH_MSG_CHANNEL_DATA;
		dssh_serialize_uint32(oc.server_ch->local_id, msg, sizeof(msg), &pos);
		dssh_serialize_uint32(100, msg, sizeof(msg), &pos);
		memcpy(&msg[pos], "abc", 3);
		pos += 3;
		ASSERT_OK(send_packet(ctx.client, msg, pos, NULL));
	}

	{
		uint8_t msg[32];
		size_t pos = 0;
		msg[pos++] = SSH_MSG_CHANNEL_EXTENDED_DATA;
		dssh_serialize_uint32(oc.server_ch->local_id, msg, sizeof(msg), &pos);
		dssh_serialize_uint32(1, msg, sizeof(msg), &pos);
		dssh_serialize_uint32(100, msg, sizeof(msg), &pos);
		memcpy(&msg[pos], "xy", 2);
		pos += 2;
		ASSERT_OK(send_packet(ctx.client, msg, pos, NULL));
	}

	struct timespec ts = { .tv_sec = 0, .tv_nsec = 10000000 };
	thrd_sleep(&ts, NULL);

	ASSERT_TRUE(ctx.server->demux_running);

	/* Both packets had bogus dlen and were dropped by the demux,
	 * so no bytes are buffered and no EOF arrived.  The new read
	 * contract returns NOMORE for empty-no-EOF; old contract
	 * returned 0.  Accept either to validate "didn't tear down". */
	uint8_t buf[256];
	int64_t n = dssh_chan_read(oc.server_ch, 0, buf, sizeof(buf));
	ASSERT_TRUE(n >= 0 || n == DSSH_ERROR_NOMORE);

	dssh_chan_close(oc.server_ch, 0);
	dssh_chan_close(oc.client_ch, 0);
	conn_cleanup(&ctx);
	return TEST_PASS;
}


static int
test_demux_dispatch_chan_type_zero(void)
{
	struct conn_ctx ctx;
	if (conn_setup(&ctx) < 0)
		return TEST_SKIP;

	struct open_exec_ctx oc = {
		.client = ctx.client,
		.server = ctx.server,
		.command = "echo chantype",
		.cbs = &accept_cbs_all,
		.accept_timeout = 30000,
	};
	if (open_exec_channel(&oc) < 0 || oc.client_ch == NULL || oc.server_ch == NULL) {
		conn_cleanup(&ctx);
		return TEST_FAIL;
	}

	int saved_type = oc.server_ch->chan_type;
	oc.server_ch->chan_type = 0;

	uint8_t payload[16];
	size_t pos = 0;
	payload[pos++] = SSH_MSG_CHANNEL_DATA;
	dssh_serialize_uint32(oc.server_ch->local_id, payload, sizeof(payload), &pos);
	dssh_serialize_uint32(1, payload, sizeof(payload), &pos);
	payload[pos++] = 'x';

	dssh_test_demux_dispatch(ctx.server, SSH_MSG_CHANNEL_DATA, payload, pos);

	oc.server_ch->chan_type = saved_type;

	dssh_chan_close(oc.server_ch, 0);
	dssh_chan_close(oc.client_ch, 0);
	conn_cleanup(&ctx);
	return TEST_PASS;
}

/* ================================================================
 * Coverage: send_eof/close already-sent, window replenish
 * ================================================================ */

static int
test_send_eof_already_sent(void)
{
	struct conn_ctx ctx;
	if (conn_setup(&ctx) < 0)
		return TEST_SKIP;

	struct open_exec_ctx oc = {
		.client = ctx.client, .server = ctx.server,
		.command = "echo eof2", .cbs = &accept_cbs_all,
		.accept_timeout = 30000,
	};
	if (open_exec_channel(&oc) < 0 || !oc.client_ch || !oc.server_ch) {
		conn_cleanup(&ctx);
		return TEST_FAIL;
	}

	ASSERT_OK(send_eof(ctx.client, oc.client_ch));
	ASSERT_TRUE(oc.client_ch->eof_sent);
	ASSERT_EQ(send_eof(ctx.client, oc.client_ch), 0);

	dssh_chan_close(oc.server_ch, 0);
	conn_cleanup(&ctx);
	return TEST_PASS;
}

static int
test_send_close_already_sent(void)
{
	struct conn_ctx ctx;
	if (conn_setup(&ctx) < 0)
		return TEST_SKIP;

	struct open_exec_ctx oc = {
		.client = ctx.client, .server = ctx.server,
		.command = "echo close2", .cbs = &accept_cbs_all,
		.accept_timeout = 30000,
	};
	if (open_exec_channel(&oc) < 0 || !oc.client_ch || !oc.server_ch) {
		conn_cleanup(&ctx);
		return TEST_FAIL;
	}

	ASSERT_OK(conn_close(ctx.client, oc.client_ch));
	ASSERT_TRUE(oc.client_ch->close_sent);
	ASSERT_EQ(conn_close(ctx.client, oc.client_ch), 0);

	dssh_chan_close(oc.server_ch, 0);
	conn_cleanup(&ctx);
	return TEST_PASS;
}

static int
test_maybe_replenish_after_eof(void)
{
	struct conn_ctx ctx;
	if (conn_setup(&ctx) < 0)
		return TEST_SKIP;

	struct open_exec_ctx oc = {
		.client = ctx.client, .server = ctx.server,
		.command = "echo replenish", .cbs = &accept_cbs_all,
		.accept_timeout = 30000,
	};
	if (open_exec_channel(&oc) < 0 || !oc.client_ch || !oc.server_ch) {
		conn_cleanup(&ctx);
		return TEST_FAIL;
	}

	oc.server_ch->local_window = 0;
	oc.server_ch->eof_received = true;
	maybe_replenish_window(ctx.server, oc.server_ch);
	ASSERT_EQ(oc.server_ch->local_window, (uint32_t)0);
	oc.server_ch->eof_received = false;

	dssh_chan_close(oc.server_ch, 0);
	dssh_chan_close(oc.client_ch, 0);
	conn_cleanup(&ctx);
	return TEST_PASS;
}

static int
test_maybe_replenish_low_window(void)
{
	struct conn_ctx ctx;
	if (conn_setup(&ctx) < 0)
		return TEST_SKIP;

	struct open_exec_ctx oc = {
		.client = ctx.client, .server = ctx.server,
		.command = "echo replow", .cbs = &accept_cbs_all,
		.accept_timeout = 30000,
	};
	if (open_exec_channel(&oc) < 0 || !oc.client_ch || !oc.server_ch) {
		conn_cleanup(&ctx);
		return TEST_FAIL;
	}

	oc.server_ch->local_window = 1;
	maybe_replenish_window(ctx.server, oc.server_ch);
	ASSERT_TRUE(oc.server_ch->local_window > 1);

	dssh_chan_close(oc.server_ch, 0);
	dssh_chan_close(oc.client_ch, 0);
	conn_cleanup(&ctx);
	return TEST_PASS;
}

struct tx_hold_state {
	dssh_session sess;
	mtx_t        mtx;
	cnd_t        cnd;
	bool         locked;
	bool         release;
};

static int
hold_tx_until_released(void *arg)
{
	struct tx_hold_state *st = arg;

	if (tx_acquire(st->sess, SSH_MSG_CHANNEL_DATA) < 0)
		return -1;
	mtx_lock(&st->mtx);
	st->locked = true;
	cnd_broadcast(&st->cnd);
	while (!st->release)
		cnd_wait(&st->cnd, &st->mtx);
	mtx_unlock(&st->mtx);
	tx_unlock_with_slots(st->sess);
	return 0;
}

static int
test_replenish_queued_accounting(void)
{
	struct conn_ctx ctx;
	if (conn_setup(&ctx) < 0)
		return TEST_SKIP;

	struct open_exec_ctx oc = {
		.client = ctx.client, .server = ctx.server,
		.command = "echo repqueue", .cbs = &accept_cbs_all,
		.accept_timeout = 30000,
	};
	if (open_exec_channel(&oc) < 0 || !oc.client_ch || !oc.server_ch) {
		conn_cleanup(&ctx);
		return TEST_FAIL;
	}

	struct tx_hold_state st = { .sess = ctx.server };
	ASSERT_EQ(mtx_init(&st.mtx, mtx_plain), thrd_success);
	ASSERT_EQ(cnd_init(&st.cnd), thrd_success);
	thrd_t holder;
	ASSERT_THRD_CREATE(&holder, hold_tx_until_released, &st);

	mtx_lock(&st.mtx);
	while (!st.locked)
		cnd_wait(&st.cnd, &st.mtx);
	mtx_unlock(&st.mtx);

	mtx_lock(&oc.server_ch->buf_mtx);
	oc.server_ch->window_max = 32;
	oc.server_ch->local_window = 1;
	oc.server_ch->window_adjust_pending = 0;
	oc.server_ch->window_adjust_inflight = 0;
	mtx_unlock(&oc.server_ch->buf_mtx);

	ASSERT_OK(maybe_replenish_window(ctx.server, oc.server_ch));
	uint8_t fail[8];
	size_t fail_pos = 0;

	fail[fail_pos++] = SSH_MSG_CHANNEL_FAILURE;
	DSSH_PUT_U32(oc.server_ch->remote_id, fail, &fail_pos);
	ASSERT_OK(send_to_slot(ctx.server, &oc.server_ch->chan_fail_slot, fail, fail_pos));
	mtx_lock(&oc.server_ch->buf_mtx);
	uint32_t local = oc.server_ch->local_window;
	uint32_t pending = oc.server_ch->window_adjust_pending;
	uint32_t inflight = oc.server_ch->window_adjust_inflight;
	bool wa_ready = atomic_load(&oc.server_ch->wa_slot.ready);
	bool fail_ready = atomic_load(&oc.server_ch->chan_fail_slot.ready);
	mtx_unlock(&oc.server_ch->buf_mtx);
	ASSERT_EQ_U(local, 1);
	ASSERT_EQ_U(pending, 31);
	ASSERT_EQ_U(inflight, 0);
	ASSERT_TRUE(wa_ready);
	ASSERT_TRUE(fail_ready);

	/* Pending credit participates in the threshold calculation, so a
	 * second reader cannot queue a duplicate grant. */
	ASSERT_OK(maybe_replenish_window(ctx.server, oc.server_ch));
	mtx_lock(&oc.server_ch->buf_mtx);
	pending = oc.server_ch->window_adjust_pending;
	mtx_unlock(&oc.server_ch->buf_mtx);
	ASSERT_EQ_U(pending, 31);

	mtx_lock(&st.mtx);
	st.release = true;
	cnd_broadcast(&st.cnd);
	mtx_unlock(&st.mtx);
	thrd_join(holder, NULL);

	mtx_lock(&oc.server_ch->buf_mtx);
	local = oc.server_ch->local_window;
	pending = oc.server_ch->window_adjust_pending;
	inflight = oc.server_ch->window_adjust_inflight;
	wa_ready = atomic_load(&oc.server_ch->wa_slot.ready);
	fail_ready = atomic_load(&oc.server_ch->chan_fail_slot.ready);
	mtx_unlock(&oc.server_ch->buf_mtx);
	ASSERT_EQ_U(local, 32);
	ASSERT_EQ_U(pending, 0);
	ASSERT_EQ_U(inflight, 0);
	ASSERT_FALSE(wa_ready);
	ASSERT_FALSE(fail_ready);

	cnd_destroy(&st.cnd);
	mtx_destroy(&st.mtx);
	dssh_chan_close(oc.server_ch, 0);
	dssh_chan_close(oc.client_ch, 0);
	conn_cleanup(&ctx);
	return TEST_PASS;
}

static uint32_t
zc_accounting_cb(dssh_channel ch, int stream, const uint8_t *data, size_t len, void *cbdata)
{
	(void)ch;
	(void)stream;
	(void)data;
	(void)len;
	return *(uint32_t *)cbdata;
}

static int
test_zc_consumption_clamped(void)
{
	struct conn_ctx ctx;
	if (conn_setup(&ctx) < 0)
		return TEST_SKIP;

	struct open_exec_ctx oc = {
		.client = ctx.client, .server = ctx.server,
		.command = "echo zcclamp", .cbs = &accept_cbs_all,
		.accept_timeout = 30000,
	};
	if (open_exec_channel(&oc) < 0 || !oc.client_ch || !oc.server_ch) {
		conn_cleanup(&ctx);
		return TEST_FAIL;
	}

	mtx_lock(&oc.server_ch->buf_mtx);
	uint32_t consumed = UINT32_MAX;

	oc.server_ch->io_model = DSSH_IO_ZC;
	oc.server_ch->zc_cb = zc_accounting_cb;
	oc.server_ch->zc_cbdata = &consumed;
	oc.server_ch->local_window = 4;
	oc.server_ch->window_max = 4;
	oc.server_ch->setup_mode = false;
	mtx_unlock(&oc.server_ch->buf_mtx);

	uint8_t payload[16];
	size_t pos = 0;
	payload[pos++] = SSH_MSG_CHANNEL_DATA;
	DSSH_PUT_U32(oc.server_ch->local_id, payload, &pos);
	DSSH_PUT_U32(4, payload, &pos);
	memcpy(&payload[pos], "four", 4);
	pos += 4;

	ASSERT_OK(dssh_test_demux_dispatch(ctx.server, SSH_MSG_CHANNEL_DATA, payload, pos));
	mtx_lock(&oc.server_ch->buf_mtx);
	uint32_t local = oc.server_ch->local_window;
	uint32_t pending = oc.server_ch->window_adjust_pending;
	uint32_t inflight = oc.server_ch->window_adjust_inflight;
	mtx_unlock(&oc.server_ch->buf_mtx);
	ASSERT_EQ_U(local, 4);
	ASSERT_EQ_U(pending, 0);
	ASSERT_EQ_U(inflight, 0);

	/* The full packet consumes peer credit even if the callback consumes
	 * only part of it; only the consumed bytes are granted back. */
	consumed = 2;
	ASSERT_OK(dssh_test_demux_dispatch(ctx.server, SSH_MSG_CHANNEL_DATA, payload, pos));
	mtx_lock(&oc.server_ch->buf_mtx);
	local = oc.server_ch->local_window;
	pending = oc.server_ch->window_adjust_pending;
	inflight = oc.server_ch->window_adjust_inflight;
	mtx_unlock(&oc.server_ch->buf_mtx);
	ASSERT_EQ_U(local, 2);
	ASSERT_EQ_U(pending, 0);
	ASSERT_EQ_U(inflight, 0);

	dssh_chan_close(oc.server_ch, 0);
	dssh_chan_close(oc.client_ch, 0);
	conn_cleanup(&ctx);
	return TEST_PASS;
}

static int
test_demux_window_underflow_to_zero(void)
{
	struct conn_ctx ctx;
	if (conn_setup(&ctx) < 0)
		return TEST_FAIL;

	struct open_exec_ctx oc = {
		.client = ctx.client, .server = ctx.server,
		.command = "echo wunder", .cbs = &accept_cbs_all,
		.accept_timeout = 30000,
	};
	if (open_exec_channel(&oc) < 0 || !oc.client_ch || !oc.server_ch) {
		conn_cleanup(&ctx);
		return TEST_FAIL;
	}

	oc.server_ch->local_window = 5;
	oc.server_ch->window_max = 5;

	const uint8_t data[] = "hello";
	dssh_chan_write(oc.client_ch, 0, data, 5);

	int ev = dssh_chan_poll(oc.server_ch, DSSH_POLL_READ, 5000);
	ASSERT_TRUE(ev & DSSH_POLL_READ);

	{
		uint8_t msg[32];
		size_t pos = 0;
		msg[pos++] = SSH_MSG_CHANNEL_DATA;
		dssh_serialize_uint32(oc.server_ch->local_id, msg, sizeof(msg), &pos);
		dssh_serialize_uint32(3, msg, sizeof(msg), &pos);
		memcpy(&msg[pos], "abc", 3);
		pos += 3;
		ASSERT_OK(send_packet(ctx.client, msg, pos, NULL));
	}

	struct timespec ts = { .tv_sec = 0, .tv_nsec = 10000000 };
	thrd_sleep(&ts, NULL);

	dssh_chan_close(oc.server_ch, 0);
	dssh_chan_close(oc.client_ch, 0);
	conn_cleanup(&ctx);
	return TEST_PASS;
}

/* ================================================================
 * Coverage: window_add overflow via send_window_adjust
 * ================================================================ */

static int
test_window_add_overflow(void)
{
	struct conn_ctx ctx;
	if (conn_setup(&ctx) < 0)
		return TEST_SKIP;

	struct open_exec_ctx oc = {
		.client = ctx.client, .server = ctx.server,
		.command = "echo woverflow", .cbs = &accept_cbs_all,
		.accept_timeout = 30000,
	};
	if (open_exec_channel(&oc) < 0 || !oc.client_ch || !oc.server_ch) {
		conn_cleanup(&ctx);
		return TEST_FAIL;
	}

	oc.server_ch->local_window = UINT32_MAX - 10;
	send_window_adjust(ctx.server, oc.server_ch, 20);
	ASSERT_EQ(oc.server_ch->local_window, UINT32_MAX);

	dssh_chan_close(oc.server_ch, 0);
	dssh_chan_close(oc.client_ch, 0);
	conn_cleanup(&ctx);
	return TEST_PASS;
}

/* ================================================================
 * Coverage: write not_open/close_received guards
 * ================================================================ */

static int
test_chan_write_not_open(void)
{
	dssh_test_reset_global_config();
	if (register_all_algorithms() < 0)
		return TEST_SKIP;
	dssh_session s = dssh_session_init(false, 0);
	if (s == NULL)
		return TEST_SKIP;

	struct dssh_channel_s ch = {0};
	ch.sess = s;
	ch.io_model = DSSH_IO_STREAM;
	ch.open = false;
	ch.eof_sent = false;
	ch.close_received = false;
	mtx_init(&ch.buf_mtx, mtx_plain);

	int64_t r = dssh_chan_write(&ch, 0, (const uint8_t *)"x", 1);
	ASSERT_EQ(r, DSSH_ERROR_TERMINATED);

	mtx_destroy(&ch.buf_mtx);
	dssh_session_cleanup(s);
	dssh_test_reset_global_config();
	return TEST_PASS;
}

static int
test_chan_write_close_received(void)
{
	dssh_test_reset_global_config();
	if (register_all_algorithms() < 0)
		return TEST_SKIP;
	dssh_session s = dssh_session_init(false, 0);
	if (s == NULL)
		return TEST_SKIP;

	struct dssh_channel_s ch = {0};
	ch.sess = s;
	ch.io_model = DSSH_IO_STREAM;
	ch.open = true;
	ch.eof_sent = false;
	ch.close_received = true;
	mtx_init(&ch.buf_mtx, mtx_plain);

	int64_t r = dssh_chan_write(&ch, 0, (const uint8_t *)"x", 1);
	ASSERT_EQ(r, DSSH_ERROR_TERMINATED);

	mtx_destroy(&ch.buf_mtx);
	dssh_session_cleanup(s);
	dssh_test_reset_global_config();
	return TEST_PASS;
}

/* ================================================================
 * Coverage: nsec overflow tests
 * ================================================================ */

static int
test_session_poll_nsec_overflow(void)
{
	struct conn_ctx ctx;
	if (conn_setup(&ctx) < 0)
		return TEST_SKIP;

	struct open_exec_ctx oc = {
		.client = ctx.client,
		.server = ctx.server,
		.command = "echo nsec",
		.cbs = &accept_cbs_all,
		.accept_timeout = 30000,
	};
	if (open_exec_channel(&oc) < 0 || oc.client_ch == NULL || oc.server_ch == NULL) {
		conn_cleanup(&ctx);
		return TEST_FAIL;
	}

	for (int i = 0; i < 3; i++) {
		int r = dssh_chan_poll(oc.server_ch, DSSH_POLL_READ, 1);
		(void)r;
	}
	int r = dssh_chan_poll(oc.server_ch, DSSH_POLL_READ, 999);
	ASSERT_EQ(r, 0);

	dssh_chan_close(oc.server_ch, 0);
	dssh_chan_close(oc.client_ch, 0);
	conn_cleanup(&ctx);
	return TEST_PASS;
}

static int
test_accept_nsec_overflow(void)
{
	struct conn_ctx ctx;
	if (conn_setup(&ctx) < 0)
		return TEST_FAIL;

	for (int i = 0; i < 3; i++) {
		dssh_channel ch = dssh_chan_accept(ctx.server, &accept_cbs_all, 1);
		(void)ch;
	}
	dssh_channel ch = dssh_chan_accept(ctx.server, &accept_cbs_all, 999);
	ASSERT_NULL(ch);

	conn_cleanup(&ctx);
	return TEST_PASS;
}

/* ================================================================
 * Subsystem roundtrip
 * ================================================================ */

static int
test_subsystem_roundtrip(void)
{
	struct conn_ctx ctx;

	if (conn_setup(&ctx) < 0)
		return TEST_SKIP;

	struct open_subsys_ctx oc = {
		.client = ctx.client,
		.server = ctx.server,
		.subsystem = "test-echo",
	};

	thrd_t ct, st;

	ASSERT_THRD_CREATE(&ct, client_open_subsys_thread, &oc);
	ASSERT_THRD_CREATE(&st, server_accept_subsys_thread, &oc);
	thrd_join(ct, NULL);
	thrd_join(st, NULL);

	if (oc.client_ch == NULL || oc.server_ch == NULL) {
		if (oc.client_ch != NULL)
			dssh_chan_close(oc.client_ch, -1);
		if (oc.server_ch != NULL)
			dssh_chan_close(oc.server_ch, -1);
		conn_cleanup(&ctx);
		return TEST_SKIP;
	}

	ASSERT_EQ(dssh_chan_get_type(oc.server_ch), DSSH_CHAN_SUBSYSTEM);
	ASSERT_NOT_NULL(dssh_chan_get_subsystem(oc.server_ch));
	ASSERT_STR_EQ(dssh_chan_get_subsystem(oc.server_ch), "test-echo");

	/* Client sends a message to server */
	static const uint8_t msg1[] = "hello from client";
	int64_t w = dssh_chan_write(oc.client_ch, 0,
	    msg1, sizeof(msg1));
	ASSERT_TRUE(w > 0);

	/* Server polls and reads */
	int ev = dssh_chan_poll(oc.server_ch, DSSH_POLL_READ, 5000);
	ASSERT_TRUE(ev & DSSH_POLL_READ);

	uint8_t buf[256];
	size_t got = 0;
	while (got < sizeof(msg1)) {
		int64_t r = dssh_chan_read(oc.server_ch, 0,
		    &buf[got], sizeof(buf) - got);
		if (r <= 0)
			break;
		got += (size_t)r;
	}
	ASSERT_EQ_U(got, sizeof(msg1));
	ASSERT_TRUE(memcmp(buf, msg1, sizeof(msg1)) == 0);

	/* Server sends reply back */
	static const uint8_t msg2[] = "hello from server";
	w = dssh_chan_write(oc.server_ch, 0,
	    msg2, sizeof(msg2));
	ASSERT_TRUE(w > 0);

	/* Client polls and reads */
	ev = dssh_chan_poll(oc.client_ch, DSSH_POLL_READ, 5000);
	ASSERT_TRUE(ev & DSSH_POLL_READ);

	got = 0;
	while (got < sizeof(msg2)) {
		int64_t r = dssh_chan_read(oc.client_ch, 0,
		    &buf[got], sizeof(buf) - got);
		if (r <= 0)
			break;
		got += (size_t)r;
	}
	ASSERT_EQ_U(got, sizeof(msg2));
	ASSERT_TRUE(memcmp(buf, msg2, sizeof(msg2)) == 0);

	/* Clean close */
	dssh_chan_close(oc.client_ch, -1);
	dssh_chan_close(oc.server_ch, -1);
	conn_cleanup(&ctx);
	return TEST_PASS;
}

/* ================================================================
 * Audit item 13: ZC accept — server accepts a zero-copy channel
 * ================================================================ */

struct zc_rx_state {
	mtx_t    mtx;
	cnd_t    cnd;
	uint8_t  buf[256];
	size_t   len;
	bool     done;
};

static uint32_t
test_zc_data_cb(dssh_channel ch, int stream,
    const uint8_t *data, size_t len, void *cbdata)
{
	(void)ch;
	(void)stream;
	struct zc_rx_state *st = cbdata;
	mtx_lock(&st->mtx);
	size_t avail = sizeof(st->buf) - st->len;
	size_t n = len < avail ? len : avail;
	memcpy(&st->buf[st->len], data, n);
	st->len += n;
	st->done = true;
	cnd_signal(&st->cnd);
	mtx_unlock(&st->mtx);
	uint32_t n_u32 = (uint32_t)n;
	return n_u32;
}

static int
accept_subsys_zc(dssh_channel ch, const struct dssh_chan_params *p,
    struct dssh_chan_accept_result *result, void *cbdata)
{
	(void)ch;
	(void)p;
	result->zc_cb = test_zc_data_cb;
	result->zc_cbdata = cbdata;
	return 0;
}

struct zc_accept_ctx {
	dssh_session server;
	const struct dssh_chan_accept_cbs *cbs;
	dssh_channel server_ch;
};

static int
server_accept_zc_thread(void *arg)
{
	struct zc_accept_ctx *ctx = arg;
	ctx->server_ch = dssh_chan_accept(ctx->server, ctx->cbs, 30000);
	return 0;
}

static int
test_accept_zc(void)
{
	struct conn_ctx ctx;

	if (conn_setup(&ctx) < 0)
		return TEST_SKIP;

	/* Set up ZC receive state for the server */
	struct zc_rx_state zc_st;
	memset(&zc_st, 0, sizeof(zc_st));
	if (mtx_init(&zc_st.mtx, mtx_plain) != thrd_success) {
		conn_cleanup(&ctx);
		return TEST_SKIP;
	}
	if (cnd_init(&zc_st.cnd) != thrd_success) {
		mtx_destroy(&zc_st.mtx);
		conn_cleanup(&ctx);
		return TEST_SKIP;
	}

	struct dssh_chan_accept_cbs zc_cbs = {
		.subsystem = accept_subsys_zc,
		.cbdata = &zc_st,
	};

	struct zc_accept_ctx sa = {
		.server = ctx.server,
		.cbs = &zc_cbs,
		.server_ch = NULL,
	};

	/* Client opens a subsystem channel (stream mode) */
	struct open_subsys_ctx oc = {
		.client = ctx.client,
		.server = ctx.server,
		.subsystem = "test-zc",
	};

	thrd_t ct, st;
	ASSERT_THRD_CREATE(&ct, client_open_subsys_thread, &oc);
	ASSERT_THRD_CREATE(&st, server_accept_zc_thread, &sa);
	thrd_join(ct, NULL);
	thrd_join(st, NULL);

	if (oc.client_ch == NULL || sa.server_ch == NULL) {
		if (oc.client_ch != NULL)
			dssh_chan_close(oc.client_ch, -1);
		if (sa.server_ch != NULL)
			dssh_chan_close(sa.server_ch, -1);
		cnd_destroy(&zc_st.cnd);
		mtx_destroy(&zc_st.mtx);
		conn_cleanup(&ctx);
		return TEST_SKIP;
	}

	/* Verify server channel is ZC mode */
	ASSERT_EQ(sa.server_ch->io_model, DSSH_IO_ZC);
	ASSERT_EQ(dssh_chan_get_type(sa.server_ch), DSSH_CHAN_SUBSYSTEM);
	ASSERT_STR_EQ(dssh_chan_get_subsystem(sa.server_ch), "test-zc");

	/* Client sends data (stream mode) */
	static const uint8_t msg1[] = "hello from client to zc";
	int64_t w = dssh_chan_write(oc.client_ch, 0,
	    msg1, sizeof(msg1));
	ASSERT_TRUE(w > 0);

	/* Wait for ZC callback to fire on server */
	mtx_lock(&zc_st.mtx);
	struct timespec deadline;
	timespec_get(&deadline, TIME_UTC);
	deadline.tv_sec += 1;
	while (!zc_st.done)
		cnd_timedwait(&zc_st.cnd, &zc_st.mtx, &deadline);
	ASSERT_TRUE(zc_st.done);
	ASSERT_EQ_U(zc_st.len, sizeof(msg1));
	ASSERT_TRUE(memcmp(zc_st.buf, msg1, sizeof(msg1)) == 0);
	mtx_unlock(&zc_st.mtx);

	/* Server sends reply via ZC getbuf/send */
	static const uint8_t msg2[] = "zc reply from server";
	uint8_t *txbuf;
	size_t max_len;
	int res = dssh_chan_zc_getbuf(sa.server_ch, 0, &txbuf, &max_len);
	ASSERT_OK(res);
	ASSERT_TRUE(max_len >= sizeof(msg2));
	memcpy(txbuf, msg2, sizeof(msg2));
	res = dssh_chan_zc_send(sa.server_ch, sizeof(msg2));
	ASSERT_OK(res);

	/* Client reads reply (stream mode) */
	int ev = dssh_chan_poll(oc.client_ch, DSSH_POLL_READ, 1000);
	ASSERT_TRUE(ev & DSSH_POLL_READ);

	uint8_t rbuf[256];
	size_t got = 0;
	while (got < sizeof(msg2)) {
		int64_t r = dssh_chan_read(oc.client_ch, 0,
		    &rbuf[got], sizeof(rbuf) - got);
		if (r <= 0)
			break;
		got += (size_t)r;
	}
	ASSERT_EQ_U(got, sizeof(msg2));
	ASSERT_TRUE(memcmp(rbuf, msg2, sizeof(msg2)) == 0);

	/* Clean close */
	dssh_chan_close(oc.client_ch, -1);
	dssh_chan_close(sa.server_ch, -1);
	cnd_destroy(&zc_st.cnd);
	mtx_destroy(&zc_st.mtx);
	conn_cleanup(&ctx);
	return TEST_PASS;
}

/* A channel close on another thread must wait while demux has dropped
 * buf_mtx to call a ZC consumer. */
struct zc_close_race_state {
	mtx_t mtx;
	cnd_t cnd;
	bool  entered;
	bool  release;
};

static uint32_t
blocking_zc_data_cb(dssh_channel ch, int stream, const uint8_t *data, size_t len, void *cbdata)
{
	(void)ch;
	(void)stream;
	(void)data;
	struct zc_close_race_state *st = cbdata;

	mtx_lock(&st->mtx);
	st->entered = true;
	cnd_broadcast(&st->cnd);
	while (!st->release)
		cnd_wait(&st->cnd, &st->mtx);
	mtx_unlock(&st->mtx);
	uint32_t consumed = len > UINT32_MAX ? UINT32_MAX : (uint32_t)len;

	return consumed;
}

static int
accept_subsys_blocking_zc(dssh_channel ch, const struct dssh_chan_params *p,
	struct dssh_chan_accept_result *result, void *cbdata)
{
	(void)ch;
	(void)p;
	result->zc_cb = blocking_zc_data_cb;
	result->zc_cbdata = cbdata;
	return 0;
}

struct close_channel_ctx {
	dssh_channel ch;
	atomic_bool  started;
	atomic_bool  done;
	int          result;
};

static int
close_channel_thread(void *arg)
{
	struct close_channel_ctx *ctx = arg;

	atomic_store(&ctx->started, true);
	ctx->result = dssh_chan_close(ctx->ch, -1);
	atomic_store(&ctx->done, true);
	return 0;
}

static int
test_close_during_zc_callback(void)
{
	struct conn_ctx ctx;
	if (conn_setup(&ctx) < 0)
		return TEST_SKIP;

	struct zc_close_race_state zc_st;
	memset(&zc_st, 0, sizeof(zc_st));
	if (mtx_init(&zc_st.mtx, mtx_plain) != thrd_success) {
		conn_cleanup(&ctx);
		return TEST_SKIP;
	}
	if (cnd_init(&zc_st.cnd) != thrd_success) {
		mtx_destroy(&zc_st.mtx);
		conn_cleanup(&ctx);
		return TEST_SKIP;
	}

	struct dssh_chan_accept_cbs zc_cbs = {
		.subsystem = accept_subsys_blocking_zc,
		.cbdata = &zc_st,
	};
	struct zc_accept_ctx sa = {
		.server = ctx.server,
		.cbs = &zc_cbs,
	};
	struct open_subsys_ctx oc = {
		.client = ctx.client,
		.server = ctx.server,
		.subsystem = "close-during-zc",
	};

	thrd_t ct, st;
	ASSERT_THRD_CREATE(&ct, client_open_subsys_thread, &oc);
	ASSERT_THRD_CREATE(&st, server_accept_zc_thread, &sa);
	thrd_join(ct, NULL);
	thrd_join(st, NULL);
	if (oc.client_ch == NULL || sa.server_ch == NULL) {
		if (oc.client_ch != NULL)
			dssh_chan_close(oc.client_ch, -1);
		if (sa.server_ch != NULL)
			dssh_chan_close(sa.server_ch, -1);
		cnd_destroy(&zc_st.cnd);
		mtx_destroy(&zc_st.mtx);
		conn_cleanup(&ctx);
		return TEST_SKIP;
	}

	static const uint8_t msg[] = "hold callback";
	ASSERT_TRUE(dssh_chan_write(oc.client_ch, 0, msg, sizeof(msg)) > 0);

	mtx_lock(&zc_st.mtx);
	while (!zc_st.entered)
		cnd_wait(&zc_st.cnd, &zc_st.mtx);
	mtx_unlock(&zc_st.mtx);

	struct close_channel_ctx close_ctx = {
		.ch = sa.server_ch,
		.result = DSSH_ERROR_INVALID,
	};
	atomic_init(&close_ctx.started, false);
	atomic_init(&close_ctx.done, false);
	thrd_t close_thread;
	ASSERT_THRD_CREATE(&close_thread, close_channel_thread, &close_ctx);
	while (!atomic_load(&close_ctx.started))
		thrd_yield();
	while (!atomic_load(&close_ctx.done) && !atomic_load(&sa.server_ch->closing))
		thrd_yield();
	bool closed_early = atomic_load(&close_ctx.done);

	mtx_lock(&zc_st.mtx);
	zc_st.release = true;
	cnd_broadcast(&zc_st.cnd);
	mtx_unlock(&zc_st.mtx);
	thrd_join(close_thread, NULL);

	ASSERT_FALSE(closed_early);
	ASSERT_OK(close_ctx.result);
	ASSERT_FALSE(dssh_session_is_terminated(ctx.server));
	dssh_chan_close(oc.client_ch, -1);
	cnd_destroy(&zc_st.cnd);
	mtx_destroy(&zc_st.mtx);
	conn_cleanup(&ctx);
	return TEST_PASS;
}

static int
test_dispatch_sync_failures(void)
{
	for (int n = 0; n < 5; n++) {
		struct conn_ctx ctx;
		if (conn_setup(&ctx) < 0)
			return TEST_SKIP;

		struct open_exec_ctx oc = {
			.client = ctx.client,
			.server = ctx.server,
			.command = "dispatch-sync-failure",
			.cbs = &accept_cbs_all,
			.accept_timeout = 30000,
		};
		if (open_exec_channel(&oc) < 0 || oc.client_ch == NULL || oc.server_ch == NULL) {
			conn_cleanup(&ctx);
			return TEST_FAIL;
		}

		uint8_t payload[9];
		size_t pos = 0;

		payload[pos++] = SSH_MSG_CHANNEL_WINDOW_ADJUST;
		DSSH_PUT_U32(oc.server_ch->local_id, payload, &pos);
		DSSH_PUT_U32(1, payload, &pos);
		dssh_test_thrd_fail_after(n);
		int ret = dssh_test_demux_dispatch(ctx.server,
		    SSH_MSG_CHANNEL_WINDOW_ADJUST, payload, pos);

		dssh_test_thrd_reset();
		ASSERT_ERR(ret, DSSH_ERROR_TERMINATED);
		ASSERT_TRUE(dssh_session_is_terminated(ctx.server));
		/* conn_cleanup() joins the demux before freeing registered channels. */
		conn_cleanup(&ctx);
	}
	return TEST_PASS;
}

static int
test_detach_sync_failures(void)
{
	for (int n = 0; n < 2; n++) {
		struct conn_ctx ctx;
		if (conn_setup(&ctx) < 0)
			return TEST_SKIP;

		struct open_exec_ctx oc = {
			.client = ctx.client,
			.server = ctx.server,
			.command = "detach-sync-failure",
			.cbs = &accept_cbs_all,
			.accept_timeout = 30000,
		};
		if (open_exec_channel(&oc) < 0 || oc.client_ch == NULL || oc.server_ch == NULL) {
			conn_cleanup(&ctx);
			return TEST_FAIL;
		}

		dssh_test_thrd_fail_after(n);
		int ret = detach_channel(ctx.server, oc.server_ch);

		dssh_test_thrd_reset();
		if (n == 0)
			ASSERT_ERR(ret, DSSH_ERROR_TERMINATED);
		else
			ASSERT_OK(ret);
		ASSERT_TRUE(dssh_session_is_terminated(ctx.server));
		/* A successful detach made this channel unreachable by demux, so
		 * close the detached handle before session cleanup.  A failed detach
		 * leaves it registered for session_stop() to free after joining demux. */
		if (n == 1)
			ASSERT_OK(dssh_chan_close(oc.server_ch, -1));
		conn_cleanup(&ctx);
	}
	return TEST_PASS;
}

/* ================================================================
 * Item 74: bytebuf write truncation + window accounting
 * ================================================================ */

static int
test_demux_data_truncation_window(void)
{
	struct conn_ctx ctx;
	if (conn_setup(&ctx) < 0)
		return TEST_SKIP;

	struct open_exec_ctx oc = {
		.client = ctx.client, .server = ctx.server,
		.command = "echo trunc", .cbs = &accept_cbs_all,
		.accept_timeout = 30000,
	};
	if (open_exec_channel(&oc) < 0 || !oc.client_ch || !oc.server_ch) {
		conn_cleanup(&ctx);
		return TEST_FAIL;
	}

	bytebuf_free(&oc.server_ch->buf.stdout_buf);
	ASSERT_OK(bytebuf_init(&oc.server_ch->buf.stdout_buf, 16));
	oc.server_ch->window_max = 16;
	oc.server_ch->local_window = 16;

	uint8_t fill[12];
	memset(fill, 'A', sizeof(fill));
	bytebuf_write(&oc.server_ch->buf.stdout_buf, fill, 12);
	oc.server_ch->local_window = 16;

	uint8_t payload[32];
	size_t pos = 0;
	payload[pos++] = SSH_MSG_CHANNEL_DATA;
	dssh_serialize_uint32(oc.server_ch->local_id, payload, sizeof(payload), &pos);
	dssh_serialize_uint32(10, payload, sizeof(payload), &pos);
	memset(&payload[pos], 'B', 10);
	pos += 10;

	mtx_lock(&oc.server_ch->buf_mtx);
	oc.server_ch->setup_mode = false;
	mtx_unlock(&oc.server_ch->buf_mtx);

	dssh_test_demux_dispatch(ctx.server, SSH_MSG_CHANNEL_DATA, payload, pos);

	/* The peer consumed ten bytes of window even though only four fit
	 * in the stream buffer; dropped bytes are not granted back. */
	ASSERT_EQ(oc.server_ch->local_window, (uint32_t)6);

	dssh_chan_close(oc.server_ch, 0);
	dssh_chan_close(oc.client_ch, 0);
	conn_cleanup(&ctx);
	return TEST_PASS;
}

static int
test_replenish_caps_to_buffer_space(void)
{
	struct conn_ctx ctx;
	if (conn_setup(&ctx) < 0)
		return TEST_SKIP;

	struct open_exec_ctx oc = {
		.client = ctx.client, .server = ctx.server,
		.command = "echo repcap", .cbs = &accept_cbs_all,
		.accept_timeout = 30000,
	};
	if (open_exec_channel(&oc) < 0 || !oc.client_ch || !oc.server_ch) {
		conn_cleanup(&ctx);
		return TEST_FAIL;
	}

	bytebuf_free(&oc.server_ch->buf.stdout_buf);
	ASSERT_OK(bytebuf_init(&oc.server_ch->buf.stdout_buf, 32));
	bytebuf_free(&oc.server_ch->buf.stderr_buf);
	ASSERT_OK(bytebuf_init(&oc.server_ch->buf.stderr_buf, 32));
	oc.server_ch->window_max = 32;

	uint8_t fill[28];
	memset(fill, 'X', sizeof(fill));
	bytebuf_write(&oc.server_ch->buf.stdout_buf, fill, 28);

	oc.server_ch->local_window = 0;

	maybe_replenish_window(ctx.server, oc.server_ch);

	ASSERT_TRUE(oc.server_ch->local_window <= 4);
	ASSERT_TRUE(oc.server_ch->local_window > 0);

	dssh_chan_close(oc.server_ch, 0);
	dssh_chan_close(oc.client_ch, 0);
	conn_cleanup(&ctx);
	return TEST_PASS;
}

/* ================================================================
 * Universal rx-window truncation + DSSH_PARAM_ACCEPT_EARLY_DATA bypass
 * ================================================================ */

/*
 * Helper: open an exec channel and set up the server-side channel so
 * we can poke a synthetic CHANNEL_DATA at it via demux_dispatch.  The
 * setup_complete and accept_pre_window_data fields are set per-test
 * by the caller after this returns.
 */
static int
rx_window_setup(struct conn_ctx *ctx, struct open_exec_ctx *oc, uint32_t bufsz)
{
	oc->client = ctx->client;
	oc->server = ctx->server;
	oc->command = "rx-window";
	oc->cbs = &accept_cbs_all;
	oc->accept_timeout = 30000;
	if (open_exec_channel(oc) < 0 || !oc->client_ch || !oc->server_ch)
		return -1;

	bytebuf_free(&oc->server_ch->buf.stdout_buf);
	if (bytebuf_init(&oc->server_ch->buf.stdout_buf, bufsz) < 0)
		return -1;
	oc->server_ch->window_max = bufsz;
	mtx_lock(&oc->server_ch->buf_mtx);
	oc->server_ch->setup_mode = false;
	mtx_unlock(&oc->server_ch->buf_mtx);
	return 0;
}

/* Build a CHANNEL_DATA payload with N bytes of 'B' for the given channel. */
static size_t
build_channel_data(uint8_t *buf, size_t bufsz, uint32_t local_id, uint32_t dlen)
{
	size_t pos = 0;
	buf[pos++] = SSH_MSG_CHANNEL_DATA;
	dssh_serialize_uint32(local_id, buf, bufsz, &pos);
	dssh_serialize_uint32(dlen, buf, bufsz, &pos);
	memset(&buf[pos], 'B', dlen);
	pos += dlen;
	return pos;
}

/* Without the flag and with a closed window, all early bytes are
 * clipped to len=0 -- the bytebuf stays empty and local_window stays 0. */
static int
test_rx_truncation_default(void)
{
	struct conn_ctx ctx;
	if (conn_setup(&ctx) < 0)
		return TEST_SKIP;

	struct open_exec_ctx oc = { 0 };
	if (rx_window_setup(&ctx, &oc, 64) < 0) {
		conn_cleanup(&ctx);
		return TEST_FAIL;
	}

	/* No flag, no setup_complete, but local_window=0.  In the existing
	 * universal path (bypass==false), dlen gets clipped to 0. */
	oc.server_ch->accept_pre_window_data = false;
	atomic_store(&oc.server_ch->setup_complete, false);
	oc.server_ch->local_window = 0;

	uint8_t payload[64];
	size_t pos = build_channel_data(payload, sizeof(payload),
	    oc.server_ch->local_id, 10);

	dssh_test_demux_dispatch(ctx.server, SSH_MSG_CHANNEL_DATA, payload, pos);

	ASSERT_EQ(bytebuf_available(&oc.server_ch->buf.stdout_buf), (size_t)0);
	ASSERT_EQ(oc.server_ch->local_window, (uint32_t)0);

	dssh_chan_close(oc.server_ch, 0);
	dssh_chan_close(oc.client_ch, 0);
	conn_cleanup(&ctx);
	return TEST_PASS;
}

/* Universal rx-truncation: dlen exceeds local_window, bytes are clipped
 * to local_window before stream_zc_cb runs. */
static int
test_rx_truncation_clips_to_window(void)
{
	struct conn_ctx ctx;
	if (conn_setup(&ctx) < 0)
		return TEST_SKIP;

	struct open_exec_ctx oc = { 0 };
	if (rx_window_setup(&ctx, &oc, 64) < 0) {
		conn_cleanup(&ctx);
		return TEST_FAIL;
	}

	oc.server_ch->accept_pre_window_data = false;
	atomic_store(&oc.server_ch->setup_complete, true);
	oc.server_ch->local_window = 4;

	uint8_t payload[64];
	size_t pos = build_channel_data(payload, sizeof(payload),
	    oc.server_ch->local_id, 10);

	dssh_test_demux_dispatch(ctx.server, SSH_MSG_CHANNEL_DATA, payload, pos);

	/* cb saw len=4, wrote 4 bytes; local_window depleted. */
	ASSERT_EQ(bytebuf_available(&oc.server_ch->buf.stdout_buf), (size_t)4);
	ASSERT_EQ(oc.server_ch->local_window, (uint32_t)0);

	dssh_chan_close(oc.server_ch, 0);
	dssh_chan_close(oc.client_ch, 0);
	conn_cleanup(&ctx);
	return TEST_PASS;
}

/* With the flag and !setup_complete, full dlen flows through despite
 * a closed window -- and local_window is NOT decremented (no credit). */
static int
test_rx_bypass_pre_setup(void)
{
	struct conn_ctx ctx;
	if (conn_setup(&ctx) < 0)
		return TEST_SKIP;

	struct open_exec_ctx oc = { 0 };
	if (rx_window_setup(&ctx, &oc, 64) < 0) {
		conn_cleanup(&ctx);
		return TEST_FAIL;
	}

	oc.server_ch->accept_pre_window_data = true;
	atomic_store(&oc.server_ch->setup_complete, false);
	oc.server_ch->local_window = 0;

	uint8_t payload[64];
	size_t pos = build_channel_data(payload, sizeof(payload),
	    oc.server_ch->local_id, 10);

	dssh_test_demux_dispatch(ctx.server, SSH_MSG_CHANNEL_DATA, payload, pos);

	ASSERT_EQ(bytebuf_available(&oc.server_ch->buf.stdout_buf), (size_t)10);
	/* Window was not credited for these out-of-budget bytes. */
	ASSERT_EQ(oc.server_ch->local_window, (uint32_t)0);

	dssh_chan_close(oc.server_ch, 0);
	dssh_chan_close(oc.client_ch, 0);
	conn_cleanup(&ctx);
	return TEST_PASS;
}

/* Once setup_complete is set, bypass disengages even with the flag --
 * truncation kicks back in and a closed window clips everything. */
static int
test_rx_bypass_disengages_post_setup(void)
{
	struct conn_ctx ctx;
	if (conn_setup(&ctx) < 0)
		return TEST_SKIP;

	struct open_exec_ctx oc = { 0 };
	if (rx_window_setup(&ctx, &oc, 64) < 0) {
		conn_cleanup(&ctx);
		return TEST_FAIL;
	}

	oc.server_ch->accept_pre_window_data = true;
	atomic_store(&oc.server_ch->setup_complete, true);
	oc.server_ch->local_window = 0;

	uint8_t payload[64];
	size_t pos = build_channel_data(payload, sizeof(payload),
	    oc.server_ch->local_id, 10);

	dssh_test_demux_dispatch(ctx.server, SSH_MSG_CHANNEL_DATA, payload, pos);

	ASSERT_EQ(bytebuf_available(&oc.server_ch->buf.stdout_buf), (size_t)0);
	ASSERT_EQ(oc.server_ch->local_window, (uint32_t)0);

	dssh_chan_close(oc.server_ch, 0);
	dssh_chan_close(oc.client_ch, 0);
	conn_cleanup(&ctx);
	return TEST_PASS;
}

/* Type-lock: dssh_chan_open with the flag set on a SUBSYSTEM channel
 * must return NULL without any wire activity. */
static int
test_chan_open_type_lock_subsystem(void)
{
	struct conn_ctx ctx;
	if (conn_setup(&ctx) < 0)
		return TEST_SKIP;

	struct dssh_chan_params p;
	ASSERT_OK(dssh_chan_params_init(&p, DSSH_CHAN_SUBSYSTEM));
	ASSERT_OK(dssh_chan_params_set_subsystem(&p, "sftp"));
	ASSERT_OK(dssh_chan_params_set_accept_early_data(&p, true));

	dssh_channel ch = dssh_chan_open(ctx.client, &p);
	ASSERT_NULL(ch);

	dssh_chan_params_free(&p);
	conn_cleanup(&ctx);
	return TEST_PASS;
}

/* ================================================================
 * dssh_chan_poll — POSIX POLLHUP-style wake-up on session terminate
 * ================================================================ */

/* Terminate before poll: poll returns the requested data flags
 * immediately, even with no data and a long timeout. */
static int
test_poll_terminate_surfaces_data_flags(void)
{
	struct conn_ctx ctx;
	if (conn_setup(&ctx) < 0)
		return TEST_SKIP;

	struct open_exec_ctx oc = {
		.client = ctx.client, .server = ctx.server,
		.command = "echo poll-terminate",
		.cbs = &accept_cbs_all, .accept_timeout = 30000,
	};
	if (open_exec_channel(&oc) < 0 || !oc.client_ch || !oc.server_ch) {
		conn_cleanup(&ctx);
		return TEST_FAIL;
	}

	/* Drain any data the echo command produced, so READ would
	 * otherwise wait. */
	uint8_t scratch[64];
	while (dssh_chan_read(oc.client_ch, 0, scratch, sizeof(scratch)) > 0)
		;

	dssh_session_terminate(ctx.client);

	/* timeout=5000 ms; should return immediately. */
	int ev = dssh_chan_poll(oc.client_ch,
	    DSSH_POLL_READ | DSSH_POLL_READEXT | DSSH_POLL_WRITE, 5000);
	ASSERT_TRUE(ev & DSSH_POLL_READ);
	ASSERT_TRUE(ev & DSSH_POLL_READEXT);
	ASSERT_TRUE(ev & DSSH_POLL_WRITE);

	/* chan_read after terminate-induced wake returns 0 (EOF) once
	 * the bytebuf is drained. */
	int64_t r = dssh_chan_read(oc.client_ch, 0, scratch, sizeof(scratch));
	ASSERT_EQ(r, (int64_t)0);

	conn_cleanup(&ctx);
	return TEST_PASS;
}

/* DSSH_POLL_EVENT alone is NOT surfaced by terminate (no real event
 * is queued; caller should rely on data flags or the terminate cb). */
static int
test_poll_terminate_no_event_bit(void)
{
	struct conn_ctx ctx;
	if (conn_setup(&ctx) < 0)
		return TEST_SKIP;

	struct open_exec_ctx oc = {
		.client = ctx.client, .server = ctx.server,
		.command = "echo poll-evt",
		.cbs = &accept_cbs_all, .accept_timeout = 30000,
	};
	if (open_exec_channel(&oc) < 0 || !oc.client_ch || !oc.server_ch) {
		conn_cleanup(&ctx);
		return TEST_FAIL;
	}

	/* Drain any pending events queued by the echo. */
	struct dssh_chan_event tmp;
	while (dssh_chan_read_event(oc.client_ch, &tmp) == 0)
		;

	dssh_session_terminate(ctx.client);

	/* timeout=0 (peek): EVENT alone returns 0 because no real event
	 * is queued.  Using DSSH_POLL_EVENT plus DSSH_POLL_READ should
	 * surface only READ, not EVENT. */
	int ev = dssh_chan_poll(oc.client_ch,
	    DSSH_POLL_EVENT | DSSH_POLL_READ, 0);
	ASSERT_TRUE(ev & DSSH_POLL_READ);
	ASSERT_FALSE(ev & DSSH_POLL_EVENT);

	conn_cleanup(&ctx);
	return TEST_PASS;
}

/* ================================================================
 * Data/ext after CLOSE
 * ================================================================ */

static int
test_data_after_close(void)
{
	struct conn_ctx ctx;
	if (conn_setup(&ctx) < 0)
		return TEST_FAIL;

	struct open_exec_ctx oc = {
		.client = ctx.client,
		.server = ctx.server,
		.command = "echo closetest",
		.cbs = &accept_cbs_all,
		.accept_timeout = 30000,
	};
	if (open_exec_channel(&oc) < 0 || oc.client_ch == NULL || oc.server_ch == NULL) {
		conn_cleanup(&ctx);
		return TEST_FAIL;
	}

	{
		uint8_t msg[8];
		size_t pos = 0;
		msg[pos++] = SSH_MSG_CHANNEL_CLOSE;
		DSSH_PUT_U32(oc.client_ch->local_id, msg, &pos);
		ASSERT_OK(send_packet(ctx.server, msg, pos, NULL));
	}

	{
		uint8_t msg[32];
		size_t pos = 0;
		msg[pos++] = SSH_MSG_CHANNEL_DATA;
		DSSH_PUT_U32(oc.client_ch->local_id, msg, &pos);
		DSSH_PUT_U32(5, msg, &pos);
		memcpy(&msg[pos], "stale", 5);
		pos += 5;
		ASSERT_OK(send_packet(ctx.server, msg, pos, NULL));
	}

	struct timespec ts = { .tv_sec = 0, .tv_nsec = 20000000 };
	thrd_sleep(&ts, NULL);

	uint8_t buf[32];
	int64_t got = dssh_chan_read(oc.client_ch, 0, buf, sizeof(buf));
	ASSERT_EQ(got, 0);

	conn_cleanup(&ctx);
	return TEST_PASS;
}

static int
test_ext_data_after_close(void)
{
	struct conn_ctx ctx;
	if (conn_setup(&ctx) < 0)
		return TEST_FAIL;

	struct open_exec_ctx oc = {
		.client = ctx.client,
		.server = ctx.server,
		.command = "echo extclose",
		.cbs = &accept_cbs_all,
		.accept_timeout = 30000,
	};
	if (open_exec_channel(&oc) < 0 || oc.client_ch == NULL || oc.server_ch == NULL) {
		conn_cleanup(&ctx);
		return TEST_FAIL;
	}

	{
		uint8_t msg[8];
		size_t pos = 0;
		msg[pos++] = SSH_MSG_CHANNEL_CLOSE;
		DSSH_PUT_U32(oc.client_ch->local_id, msg, &pos);
		ASSERT_OK(send_packet(ctx.server, msg, pos, NULL));
	}

	{
		uint8_t msg[32];
		size_t pos = 0;
		msg[pos++] = SSH_MSG_CHANNEL_EXTENDED_DATA;
		DSSH_PUT_U32(oc.client_ch->local_id, msg, &pos);
		DSSH_PUT_U32(1, msg, &pos);
		DSSH_PUT_U32(5, msg, &pos);
		memcpy(&msg[pos], "stale", 5);
		pos += 5;
		ASSERT_OK(send_packet(ctx.server, msg, pos, NULL));
	}

	struct timespec ts = { .tv_sec = 0, .tv_nsec = 20000000 };
	thrd_sleep(&ts, NULL);

	uint8_t buf[32];
	int64_t got = dssh_chan_read(oc.client_ch, 1, buf, sizeof(buf));
	ASSERT_EQ(got, 0);

	conn_cleanup(&ctx);
	return TEST_PASS;
}

/* ================================================================
 * Coverage: replenish window with close_received
 * ================================================================ */

static int
test_replenish_after_close(void)
{
	dssh_test_reset_global_config();
	if (register_all_algorithms() < 0)
		return TEST_SKIP;
	dssh_session s = dssh_session_init(false, 0);
	if (s == NULL)
		return TEST_SKIP;

	struct dssh_channel_s ch = {0};
	ch.chan_type = DSSH_CHAN_SESSION;
	mtx_init(&ch.buf_mtx, mtx_plain);
	cnd_init(&ch.poll_cnd);
	bytebuf_init(&ch.buf.stdout_buf, 256);
	bytebuf_init(&ch.buf.stderr_buf, 256);
	ch.close_received = true;
	ch.window_max = 256;
	ch.local_window = 0;

	int r = maybe_replenish_window(s, &ch);
	ASSERT_EQ(r, 0);
	ASSERT_EQ_U(ch.local_window, 0);

	bytebuf_free(&ch.buf.stdout_buf);
	bytebuf_free(&ch.buf.stderr_buf);
	cnd_destroy(&ch.poll_cnd);
	mtx_destroy(&ch.buf_mtx);
	dssh_session_cleanup(s);
	dssh_test_reset_global_config();
	return TEST_PASS;
}

/* ================================================================
 * Coverage: truncated WINDOW_ADJUST
 * ================================================================ */

static int
test_truncated_window_adjust(void)
{
	struct conn_ctx ctx;
	if (conn_setup(&ctx) < 0)
		return TEST_FAIL;

	struct open_exec_ctx oc = {
		.client = ctx.client,
		.server = ctx.server,
		.command = "echo winadj",
		.cbs = &accept_cbs_all,
		.accept_timeout = 30000,
	};
	if (open_exec_channel(&oc) < 0 || oc.client_ch == NULL || oc.server_ch == NULL) {
		conn_cleanup(&ctx);
		return TEST_FAIL;
	}

	{
		uint8_t msg[8];
		size_t pos = 0;
		msg[pos++] = SSH_MSG_CHANNEL_WINDOW_ADJUST;
		DSSH_PUT_U32(oc.client_ch->local_id, msg, &pos);
		msg[pos++] = 0;
		ASSERT_OK(send_packet(ctx.server, msg, pos, NULL));
	}

	struct timespec ts = { .tv_sec = 0, .tv_nsec = 10000000 };
	thrd_sleep(&ts, NULL);

	ASSERT_TRUE(ctx.client->demux_running);

	dssh_chan_close(oc.server_ch, 0);
	dssh_chan_close(oc.client_ch, 0);
	conn_cleanup(&ctx);
	return TEST_PASS;
}

/* ================================================================
 * Allocation failure injection during channel open
 * ================================================================ */

static int
server_accept_alloc_thread(void *arg)
{
	struct open_exec_ctx *ctx = arg;

	dssh_test_alloc_exclude_thread();

	ctx->server_ch = dssh_chan_accept(ctx->server, ctx->cbs, 1000);
	return 0;
}

static int
test_open_exec_alloc_sweep(void)
{
	int any_failed = 0;

	for (int n = 0; n < 50; n++) {
		struct conn_ctx ctx;

		dssh_test_alloc_exclude_new_threads();

		if (conn_setup(&ctx) < 0)
			return TEST_FAIL;

		struct open_exec_ctx oc = {
			.client = ctx.client,
			.server = ctx.server,
			.client_ch = NULL,
			.server_ch = NULL,
			.command = "echo alloc",
			.cbs = &accept_cbs_all,
		};

		dssh_test_alloc_fail_after(n);

		thrd_t ct, st;
		ASSERT_THRD_CREATE(&ct, client_open_exec_thread, &oc);
		ASSERT_THRD_CREATE(&st, server_accept_alloc_thread, &oc);
		thrd_join(ct, NULL);

		int count = dssh_test_alloc_count();
		dssh_test_alloc_reset();

		if (oc.client_ch != NULL) {
			thrd_join(st, NULL);
			if (oc.server_ch != NULL)
				dssh_chan_close(oc.server_ch, 0);
			dssh_chan_close(oc.client_ch, 0);
			conn_cleanup(&ctx);

			ASSERT_TRUE(any_failed);
			return TEST_PASS;
		}

		any_failed = 1;
		session_set_terminate(ctx.server);
		thrd_join(st, NULL);

		if (oc.server_ch != NULL)
			dssh_chan_close(oc.server_ch, 0);

		conn_cleanup(&ctx);

		if (n >= count)
			break;
	}

	return any_failed ? TEST_PASS : TEST_FAIL;
}

static int
client_open_shell_alloc_thread(void *arg)
{
	struct open_shell_ctx *ctx = arg;
	struct dssh_chan_params p;
	dssh_chan_params_init(&p, DSSH_CHAN_SHELL);
	dssh_chan_params_set_term(&p, ctx->term);
	dssh_chan_params_set_size(&p, ctx->cols, ctx->rows, ctx->wpx, ctx->hpx);
	ctx->client_ch = dssh_chan_open(ctx->client, &p);
	dssh_chan_params_free(&p);
	return 0;
}

static int
server_accept_shell_alloc_thread(void *arg)
{
	struct open_shell_ctx *ctx = arg;

	dssh_test_alloc_exclude_thread();

	ctx->server_ch = dssh_chan_accept(ctx->server, ctx->cbs, 1000);
	return 0;
}

static int
test_open_shell_alloc_sweep(void)
{
	int any_failed = 0;

	for (int n = 0; n < 50; n++) {
		struct conn_ctx ctx;

		dssh_test_alloc_exclude_new_threads();

		if (conn_setup(&ctx) < 0)
			return TEST_FAIL;

		struct open_shell_ctx oc = {
			.client = ctx.client,
			.server = ctx.server,
			.client_ch = NULL,
			.server_ch = NULL,
			.term = "xterm",
			.cols = 80,
			.rows = 24,
			.cbs = &accept_cbs_all,
		};

		dssh_test_alloc_fail_after(n);

		thrd_t ct, st;
		ASSERT_THRD_CREATE(&ct, client_open_shell_alloc_thread, &oc);
		ASSERT_THRD_CREATE(&st, server_accept_shell_alloc_thread, &oc);
		thrd_join(ct, NULL);

		int count = dssh_test_alloc_count();
		dssh_test_alloc_reset();

		if (oc.client_ch != NULL) {
			thrd_join(st, NULL);
			if (oc.server_ch != NULL)
				dssh_chan_close(oc.server_ch, 0);
			dssh_chan_close(oc.client_ch, 0);
			conn_cleanup(&ctx);
			ASSERT_TRUE(any_failed);
			return TEST_PASS;
		}

		any_failed = 1;
		session_set_terminate(ctx.server);
		thrd_join(st, NULL);
		if (oc.server_ch != NULL)
			dssh_chan_close(oc.server_ch, 0);
		conn_cleanup(&ctx);

		if (n >= count)
			break;
	}

	return any_failed ? TEST_PASS : TEST_FAIL;
}

/* ================================================================
 * Subsystem alloc failure sweep
 * ================================================================ */

static int
client_open_subsys_alloc_thread(void *arg)
{
	struct open_subsys_ctx *ctx = arg;
	struct dssh_chan_params p;
	dssh_chan_params_init(&p, DSSH_CHAN_SUBSYSTEM);
	dssh_chan_params_set_subsystem(&p, ctx->subsystem);
	dssh_chan_params_set_pty(&p, false);
	ctx->client_ch = dssh_chan_open(ctx->client, &p);
	dssh_chan_params_free(&p);
	return 0;
}

static int
server_accept_subsys_alloc_thread(void *arg)
{
	struct open_subsys_ctx *ctx = arg;

	dssh_test_alloc_exclude_thread();

	ctx->server_ch = dssh_chan_accept(ctx->server, &accept_cbs_all, 1000);
	return 0;
}

static int
test_open_subsystem_alloc_sweep(void)
{
	int any_failed = 0;

	for (int n = 0; n < 50; n++) {
		struct conn_ctx ctx;

		dssh_test_alloc_exclude_new_threads();

		if (conn_setup(&ctx) < 0)
			return TEST_FAIL;

		struct open_subsys_ctx oc = {
			.client = ctx.client,
			.server = ctx.server,
			.client_ch = NULL,
			.server_ch = NULL,
			.subsystem = "sftp",
		};

		dssh_test_alloc_fail_after(n);

		thrd_t ct, st;
		ASSERT_THRD_CREATE(&ct, client_open_subsys_alloc_thread, &oc);
		ASSERT_THRD_CREATE(&st, server_accept_subsys_alloc_thread, &oc);
		thrd_join(ct, NULL);

		int count = dssh_test_alloc_count();
		dssh_test_alloc_reset();

		if (oc.client_ch != NULL) {
			thrd_join(st, NULL);
			if (oc.server_ch != NULL)
				dssh_chan_close(oc.server_ch, -1);
			dssh_chan_close(oc.client_ch, -1);
			conn_cleanup(&ctx);
			ASSERT_TRUE(any_failed);
			return TEST_PASS;
		}

		any_failed = 1;
		session_set_terminate(ctx.server);
		thrd_join(st, NULL);

		if (oc.server_ch != NULL)
			dssh_chan_close(oc.server_ch, -1);

		conn_cleanup(&ctx);

		if (n >= count)
			break;
	}

	return any_failed ? TEST_PASS : TEST_FAIL;
}

/* ================================================================
 * Server-side accept alloc sweep
 * ================================================================ */

static int
client_open_exec_nofail_thread(void *arg)
{
	struct open_exec_ctx *ctx = arg;
	dssh_test_alloc_exclude_thread();
	struct dssh_chan_params p;
	dssh_chan_params_init(&p, DSSH_CHAN_EXEC);
	dssh_chan_params_set_command(&p, ctx->command);
	dssh_chan_params_set_pty(&p, false);
	ctx->client_ch = dssh_chan_open(ctx->client, &p);
	dssh_chan_params_free(&p);
	return 0;
}

static int
server_accept_alloc_inject_thread(void *arg)
{
	struct open_exec_ctx *ctx = arg;
	ctx->server_ch = dssh_chan_accept(ctx->server, ctx->cbs, 1000);
	return 0;
}

static int
test_server_accept_alloc_sweep(void)
{
	int any_failed = 0;

	for (int n = 0; n < 50; n++) {
		struct conn_ctx ctx;

		dssh_test_alloc_exclude_new_threads();

		if (conn_setup(&ctx) < 0)
			return TEST_FAIL;

		struct open_exec_ctx oc = {
			.client = ctx.client,
			.server = ctx.server,
			.client_ch = NULL,
			.server_ch = NULL,
			.command = "echo srvalloc",
			.cbs = &accept_cbs_all,
		};

		thrd_t ct, st;
		ASSERT_THRD_CREATE(&ct, client_open_exec_nofail_thread, &oc);

		dssh_test_alloc_fail_after(n);

		ASSERT_THRD_CREATE(&st, server_accept_alloc_inject_thread, &oc);
		thrd_join(st, NULL);

		int count = dssh_test_alloc_count();
		dssh_test_alloc_reset();

		if (oc.server_ch != NULL) {
			thrd_join(ct, NULL);
			dssh_chan_close(oc.server_ch, 0);
			if (oc.client_ch != NULL)
				dssh_chan_close(oc.client_ch, 0);
			conn_cleanup(&ctx);
			ASSERT_TRUE(any_failed);
			return TEST_PASS;
		}

		any_failed = 1;
		session_set_terminate(ctx.client);
		thrd_join(ct, NULL);

		if (oc.client_ch != NULL)
			dssh_chan_close(oc.client_ch, 0);

		conn_cleanup(&ctx);

		if (n >= count)
			break;
	}

	return any_failed ? TEST_PASS : TEST_FAIL;
}

/* ================================================================
 * Subsystem: write zero bytes
 * ================================================================ */

static int
test_subsys_write_zero(void)
{
	struct conn_ctx ctx;
	if (conn_setup(&ctx) < 0)
		return TEST_SKIP;

	struct open_subsys_ctx oc = {
		.client = ctx.client,
		.server = ctx.server,
		.subsystem = "test-sub",
	};
	thrd_t ct, st;
	ASSERT_THRD_CREATE(&ct, client_open_subsys_thread, &oc);
	ASSERT_THRD_CREATE(&st, server_accept_subsys_thread, &oc);
	thrd_join(ct, NULL);
	thrd_join(st, NULL);

	if (oc.server_ch == NULL || oc.client_ch == NULL) {
		conn_cleanup(&ctx);
		return TEST_SKIP;
	}

	/* Write zero bytes should return 0 (no-op) */
	uint8_t dummy = 0;
	int64_t w = dssh_chan_write(oc.server_ch, 0, &dummy, 0);
	ASSERT_EQ(w, 0);

	int ev = dssh_chan_poll(oc.server_ch, DSSH_POLL_READ, 0);
	ASSERT_TRUE(ev >= 0);

	dssh_chan_close(oc.client_ch, -1);
	dssh_chan_close(oc.server_ch, -1);
	conn_cleanup(&ctx);
	return TEST_PASS;
}

/* ================================================================
 * Item 64: poll deadline must not reset on spurious wakeup
 * ================================================================ */

struct spam_cnd_arg {
	cnd_t *cnd;
	mtx_t *mtx;
	_Atomic bool stop;
};

static int
spam_cnd_thread(void *arg)
{
	struct spam_cnd_arg *sa = arg;
	const struct timespec dur = { .tv_sec = 0, .tv_nsec = 50000000L };

	while (!sa->stop) {
		thrd_sleep(&dur, NULL);

		mtx_lock(sa->mtx);
		cnd_broadcast(sa->cnd);
		mtx_unlock(sa->mtx);
	}

	return 0;
}

static int
test_poll_deadline_not_reset(void)
{
	struct conn_ctx ctx;

	if (conn_setup(&ctx) < 0)
		return TEST_SKIP;

	struct open_exec_ctx oc = {
		.client = ctx.client,
		.server = ctx.server,
		.command = "cmd",
		.cbs = &accept_cbs_all,
		.accept_timeout = 30000,
	};
	if (open_exec_channel(&oc) < 0) {
		conn_cleanup(&ctx);
		return TEST_SKIP;
	}

	struct spam_cnd_arg sa = {
		.cnd = &oc.server_ch->poll_cnd,
		.mtx = &oc.server_ch->buf_mtx,
		.stop = false,
	};
	thrd_t spammer;

	ASSERT_THRD_CREATE(&spammer, spam_cnd_thread, &sa);

	struct timespec t0;

	timespec_get(&t0, TIME_UTC);

	int ev = dssh_chan_poll(oc.server_ch, DSSH_POLL_READ, 500);
	ASSERT_EQ(ev, 0);

	struct timespec t1;

	timespec_get(&t1, TIME_UTC);
	sa.stop = true;
	thrd_join(spammer, NULL);

	long elapsed_ms = (t1.tv_sec - t0.tv_sec) * 1000L
	    + (t1.tv_nsec - t0.tv_nsec) / 1000000L;

	if (elapsed_ms > 1200) {
		fprintf(stderr, "  poll took %ldms (expected ~500ms)\n",
		    elapsed_ms);
		dssh_chan_close(oc.client_ch, 0);
		dssh_chan_close(oc.server_ch, 0);
		conn_cleanup(&ctx);
		return TEST_FAIL;
	}

	dssh_chan_close(oc.client_ch, 0);
	dssh_chan_close(oc.server_ch, 0);
	conn_cleanup(&ctx);
	return TEST_PASS;
}

/* ================================================================
 * Channel request parser unit tests
 * ================================================================ */

static size_t
build_channel_request(uint8_t *buf, size_t bufsz,
    uint32_t channel, const char *rtype, bool want_reply,
    const uint8_t *extra, size_t extra_len)
{
	size_t pos = 0;

	buf[pos++] = SSH_MSG_CHANNEL_REQUEST;
	DSSH_PUT_U32(channel, buf, &pos);
	uint32_t rtlen = (uint32_t)strlen(rtype);
	DSSH_PUT_U32(rtlen, buf, &pos);
	memcpy(&buf[pos], rtype, rtlen);
	pos += rtlen;
	buf[pos++] = want_reply ? 1 : 0;
	if (extra != NULL && extra_len > 0) {
		memcpy(&buf[pos], extra, extra_len);
		pos += extra_len;
	}
	(void)bufsz;
	return pos;
}

static int
test_chanreq_parse_valid(void)
{
	uint8_t buf[128];
	uint8_t extra[] = { 0x00, 0x00, 0x00, 0x04, 'b', 'a', 's', 'h' };
	size_t len = build_channel_request(buf, sizeof(buf),
	    42, "exec", true, extra, sizeof(extra));

	const uint8_t *rtype;
	uint32_t rtype_len;
	bool want_reply;
	const uint8_t *rdata;
	size_t rdata_len;

	ASSERT_OK(parse_channel_request(buf, len,
	    &rtype, &rtype_len, &want_reply, &rdata, &rdata_len));
	ASSERT_EQ(rtype_len, 4U);
	ASSERT_MEM_EQ(rtype, "exec", 4);
	ASSERT_TRUE(want_reply);
	ASSERT_EQ(rdata_len, sizeof(extra));
	ASSERT_MEM_EQ(rdata, extra, sizeof(extra));

	return TEST_PASS;
}

static int
test_chanreq_parse_empty_type(void)
{
	uint8_t buf[32];
	size_t len = build_channel_request(buf, sizeof(buf),
	    0, "", false, NULL, 0);

	const uint8_t *rtype;
	uint32_t rtype_len;
	bool want_reply;
	const uint8_t *rdata;
	size_t rdata_len;

	ASSERT_OK(parse_channel_request(buf, len,
	    &rtype, &rtype_len, &want_reply, &rdata, &rdata_len));
	ASSERT_EQ(rtype_len, 0U);
	ASSERT_FALSE(want_reply);
	ASSERT_EQ(rdata_len, 0U);

	return TEST_PASS;
}

static int
test_chanreq_parse_no_data(void)
{
	uint8_t buf[32];
	size_t len = build_channel_request(buf, sizeof(buf),
	    7, "shell", true, NULL, 0);

	const uint8_t *rtype;
	uint32_t rtype_len;
	bool want_reply;
	const uint8_t *rdata;
	size_t rdata_len;

	ASSERT_OK(parse_channel_request(buf, len,
	    &rtype, &rtype_len, &want_reply, &rdata, &rdata_len));
	ASSERT_EQ(rtype_len, 5U);
	ASSERT_MEM_EQ(rtype, "shell", 5);
	ASSERT_TRUE(want_reply);
	ASSERT_EQ(rdata_len, 0U);

	return TEST_PASS;
}

static int
test_chanreq_parse_truncated_type_len(void)
{
	uint8_t buf[5];

	buf[0] = SSH_MSG_CHANNEL_REQUEST;
	size_t pos = 1;
	DSSH_PUT_U32(0, buf, &pos);

	const uint8_t *rtype;
	uint32_t rtype_len;
	bool want_reply;
	const uint8_t *rdata;
	size_t rdata_len;

	ASSERT_ERR(parse_channel_request(buf, pos,
	    &rtype, &rtype_len, &want_reply, &rdata, &rdata_len),
	    DSSH_ERROR_PARSE);

	return TEST_PASS;
}

static int
test_chanreq_parse_truncated_type(void)
{
	uint8_t buf[16];
	size_t pos = 0;

	buf[pos++] = SSH_MSG_CHANNEL_REQUEST;
	DSSH_PUT_U32(0, buf, &pos);
	DSSH_PUT_U32(10, buf, &pos);
	buf[pos++] = 'a';
	buf[pos++] = 'b';
	buf[pos++] = 'c';

	const uint8_t *rtype;
	uint32_t rtype_len;
	bool want_reply;
	const uint8_t *rdata;
	size_t rdata_len;

	ASSERT_ERR(parse_channel_request(buf, pos,
	    &rtype, &rtype_len, &want_reply, &rdata, &rdata_len),
	    DSSH_ERROR_PARSE);

	return TEST_PASS;
}

static int
test_chanreq_parse_truncated_want_reply(void)
{
	uint8_t buf[16];
	size_t pos = 0;

	buf[pos++] = SSH_MSG_CHANNEL_REQUEST;
	DSSH_PUT_U32(0, buf, &pos);
	DSSH_PUT_U32(2, buf, &pos);
	buf[pos++] = 'o';
	buf[pos++] = 'k';

	const uint8_t *rtype;
	uint32_t rtype_len;
	bool want_reply;
	const uint8_t *rdata;
	size_t rdata_len;

	ASSERT_ERR(parse_channel_request(buf, pos,
	    &rtype, &rtype_len, &want_reply, &rdata, &rdata_len),
	    DSSH_ERROR_PARSE);

	return TEST_PASS;
}

/* ================================================================
 * NULL parameter validation
 * ================================================================ */

static int
test_chan_read_null(void)
{
	uint8_t buf[16];

	ASSERT_EQ(dssh_chan_read(NULL, 0, buf, sizeof(buf)),
	    DSSH_ERROR_INVALID);
	return TEST_PASS;
}

static int
test_chan_write_null(void)
{
	uint8_t buf[16] = {0};

	ASSERT_EQ(dssh_chan_write(NULL, 0, buf, sizeof(buf)),
	    DSSH_ERROR_INVALID);
	return TEST_PASS;
}

static int
test_chan_poll_null(void)
{
	ASSERT_EQ(dssh_chan_poll(NULL, DSSH_POLL_READ, 0),
	    DSSH_ERROR_INVALID);
	return TEST_PASS;
}

static int
test_chan_close_null(void)
{
	ASSERT_EQ(dssh_chan_close(NULL, 0),
	    DSSH_ERROR_INVALID);
	return TEST_PASS;
}

static int
test_chan_send_signal_null(void)
{
	ASSERT_EQ(dssh_chan_send_signal(NULL, "HUP"),
	    DSSH_ERROR_INVALID);
	return TEST_PASS;
}

static int
test_chan_send_window_change_null(void)
{
	ASSERT_EQ(dssh_chan_send_window_change(NULL, 80, 24, 0, 0),
	    DSSH_ERROR_INVALID);
	return TEST_PASS;
}

/* ================================================================
 * conn_setup_no_start: handshake + auth but no session_start.
 * Allows setting session-level callbacks before demux thread runs.
 * ================================================================ */

static int
conn_setup_no_start(struct conn_ctx *ctx)
{
	memset(ctx, 0, sizeof(*ctx));
	reset_callbacks();

	dssh_test_reset_global_config();

	if (register_all_algorithms() < 0)
		return -1;
	if (test_generate_host_key() < 0)
		return -1;

	if (mock_io_init(&ctx->io, 0) < 0)
		return -1;

	dssh_transport_set_callbacks(mock_tx_dispatch, mock_rx_dispatch,
	    mock_rxline_dispatch, mock_extra_line_cb);

	ctx->client = dssh_session_init(true, 0);
	if (ctx->client == NULL) {
		mock_io_free(&ctx->io);
		return -1;
	}
	dssh_session_set_cbdata(ctx->client, &ctx->io, &ctx->io,
	    &ctx->io, &ctx->io);
	dssh_session_set_hostkey_verify_cb(ctx->client, dssh_test_accept_hostkey, NULL);

	ctx->server = init_server_session();
	if (ctx->server == NULL) {
		dssh_session_cleanup(ctx->client);
		mock_io_free(&ctx->io);
		return -1;
	}
	dssh_session_set_cbdata(ctx->server, &ctx->io, &ctx->io,
	    &ctx->io, &ctx->io);

	struct handshake_auth_ctx client_ha = {
		.io = &ctx->io,
		.sess = ctx->client,
		.result = -1,
	};
	struct handshake_auth_ctx server_ha = {
		.io = &ctx->io,
		.sess = ctx->server,
		.result = -1,
	};

	thrd_t ct, st;
	if (thrd_create(&ct, client_handshake_auth_thread, &client_ha) != thrd_success) {
		dssh_session_cleanup(ctx->server);
		dssh_session_cleanup(ctx->client);
		mock_io_free(&ctx->io);
		return -1;
	}
	if (thrd_create(&st, server_handshake_auth_thread, &server_ha) != thrd_success) {
		dssh_session_terminate(ctx->client);
		mock_io_close_c2s(&ctx->io);
		mock_io_close_s2c(&ctx->io);
		thrd_join(ct, NULL);
		dssh_session_cleanup(ctx->server);
		dssh_session_cleanup(ctx->client);
		mock_io_free(&ctx->io);
		return -1;
	}

	thrd_join(ct, NULL);
	thrd_join(st, NULL);

	if (client_ha.result != 0 || server_ha.result != 0) {
		dssh_session_cleanup(ctx->server);
		dssh_session_cleanup(ctx->client);
		mock_io_free(&ctx->io);
		return -1;
	}

	g_active_ctx = ctx;
	return 0;
}

static void
conn_cleanup_no_start(struct conn_ctx *ctx)
{
	if (g_active_ctx == ctx)
		g_active_ctx = NULL;
	dssh_session_cleanup(ctx->server);
	dssh_session_cleanup(ctx->client);
	mock_io_free(&ctx->io);
}

/* ================================================================
 * Zero-copy client-side open (dssh_chan_zc_open)
 * ================================================================ */

/* ZC RX state for client-side receive */
struct client_zc_rx_state {
	mtx_t    mtx;
	cnd_t    cnd;
	uint8_t  buf[4096];
	size_t   len;
	bool     done;
};

static uint32_t
client_zc_data_cb(dssh_channel ch, int stream,
    const uint8_t *data, size_t len, void *cbdata)
{
	(void)ch;
	(void)stream;
	struct client_zc_rx_state *st = cbdata;
	mtx_lock(&st->mtx);
	size_t avail = sizeof(st->buf) - st->len;
	size_t n = len < avail ? len : avail;
	memcpy(&st->buf[st->len], data, n);
	st->len += n;
	st->done = true;
	cnd_signal(&st->cnd);
	mtx_unlock(&st->mtx);
	uint32_t n_u32 = (uint32_t)n;
	return n_u32;
}

struct client_zc_open_ctx {
	dssh_session client;
	dssh_channel client_ch;
	dssh_chan_zc_cb zc_cb;
	void *zc_cbdata;
};

static int
client_zc_open_thread(void *arg)
{
	struct client_zc_open_ctx *ctx = arg;
	struct dssh_chan_params p;
	dssh_chan_params_init(&p, DSSH_CHAN_SUBSYSTEM);
	dssh_chan_params_set_subsystem(&p, "test-zc-client");
	dssh_chan_params_set_pty(&p, false);
	ctx->client_ch = dssh_chan_zc_open(ctx->client, &p,
	    ctx->zc_cb, ctx->zc_cbdata);
	dssh_chan_params_free(&p);
	return 0;
}

static int
test_zc_open_client(void)
{
	struct conn_ctx ctx;
	if (conn_setup(&ctx) < 0)
		return TEST_SKIP;

	/* Set up ZC receive state for the client */
	struct client_zc_rx_state zc_st;
	memset(&zc_st, 0, sizeof(zc_st));
	if (mtx_init(&zc_st.mtx, mtx_plain) != thrd_success) {
		conn_cleanup(&ctx);
		return TEST_SKIP;
	}
	if (cnd_init(&zc_st.cnd) != thrd_success) {
		mtx_destroy(&zc_st.mtx);
		conn_cleanup(&ctx);
		return TEST_SKIP;
	}

	/* Client opens ZC channel */
	struct client_zc_open_ctx oc = {
		.client = ctx.client,
		.client_ch = NULL,
		.zc_cb = client_zc_data_cb,
		.zc_cbdata = &zc_st,
	};

	/* Server accepts with stream mode */
	struct open_subsys_ctx sc = {
		.client = ctx.client,
		.server = ctx.server,
		.subsystem = "test-zc-client",
	};

	thrd_t ct, st;
	ASSERT_THRD_CREATE(&ct, client_zc_open_thread, &oc);
	ASSERT_THRD_CREATE(&st, server_accept_subsys_thread, &sc);
	thrd_join(ct, NULL);
	thrd_join(st, NULL);

	if (oc.client_ch == NULL || sc.server_ch == NULL) {
		if (oc.client_ch != NULL)
			dssh_chan_close(oc.client_ch, -1);
		if (sc.server_ch != NULL)
			dssh_chan_close(sc.server_ch, -1);
		cnd_destroy(&zc_st.cnd);
		mtx_destroy(&zc_st.mtx);
		conn_cleanup(&ctx);
		return TEST_SKIP;
	}

	/* Verify client channel is ZC mode */
	ASSERT_EQ(oc.client_ch->io_model, DSSH_IO_ZC);

	/* Server sends data (stream mode) -- should trigger client ZC callback */
	int ev = dssh_chan_poll(sc.server_ch, DSSH_POLL_WRITE, 5000);
	ASSERT_TRUE(ev & DSSH_POLL_WRITE);
	static const uint8_t msg1[] = "hello to zc client";
	int64_t w = dssh_chan_write(sc.server_ch, 0, msg1, sizeof(msg1));
	ASSERT_TRUE(w > 0);

	/* Wait for ZC callback to fire on client */
	mtx_lock(&zc_st.mtx);
	struct timespec deadline;
	timespec_get(&deadline, TIME_UTC);
	deadline.tv_sec += 5;
	while (!zc_st.done) {
		if (cnd_timedwait(&zc_st.cnd, &zc_st.mtx, &deadline)
		    == thrd_timedout)
			break;
	}
	ASSERT_TRUE(zc_st.done);
	ASSERT_TRUE(zc_st.len >= sizeof(msg1));
	ASSERT_TRUE(memcmp(zc_st.buf, msg1, sizeof(msg1)) == 0);
	mtx_unlock(&zc_st.mtx);

	/* Client sends data via ZC getbuf/send */
	static const uint8_t msg2[] = "zc reply from client";
	uint8_t *txbuf;
	size_t max_len;
	int res = dssh_chan_zc_getbuf(oc.client_ch, 0, &txbuf, &max_len);
	ASSERT_OK(res);
	ASSERT_TRUE(max_len >= sizeof(msg2));
	memcpy(txbuf, msg2, sizeof(msg2));
	res = dssh_chan_zc_send(oc.client_ch, sizeof(msg2));
	ASSERT_OK(res);

	/* Server reads reply (stream mode) */
	ev = dssh_chan_poll(sc.server_ch, DSSH_POLL_READ, 5000);
	ASSERT_TRUE(ev & DSSH_POLL_READ);

	uint8_t rbuf[256];
	size_t got = 0;
	while (got < sizeof(msg2)) {
		int64_t r = dssh_chan_read(sc.server_ch, 0,
		    &rbuf[got], sizeof(rbuf) - got);
		if (r <= 0)
			break;
		got += (size_t)r;
	}
	ASSERT_EQ_U(got, sizeof(msg2));
	ASSERT_TRUE(memcmp(rbuf, msg2, sizeof(msg2)) == 0);

	/* Clean close */
	dssh_chan_close(oc.client_ch, -1);
	dssh_chan_close(sc.server_ch, -1);
	cnd_destroy(&zc_st.cnd);
	mtx_destroy(&zc_st.mtx);
	conn_cleanup(&ctx);
	return TEST_PASS;
}

/* ================================================================
 * ZC cancel
 * ================================================================ */

static int
test_zc_cancel(void)
{
	struct conn_ctx ctx;
	if (conn_setup(&ctx) < 0)
		return TEST_SKIP;

	struct client_zc_rx_state zc_st;
	memset(&zc_st, 0, sizeof(zc_st));
	if (mtx_init(&zc_st.mtx, mtx_plain) != thrd_success) {
		conn_cleanup(&ctx);
		return TEST_SKIP;
	}
	if (cnd_init(&zc_st.cnd) != thrd_success) {
		mtx_destroy(&zc_st.mtx);
		conn_cleanup(&ctx);
		return TEST_SKIP;
	}

	struct client_zc_open_ctx oc = {
		.client = ctx.client,
		.client_ch = NULL,
		.zc_cb = client_zc_data_cb,
		.zc_cbdata = &zc_st,
	};

	struct open_subsys_ctx sc = {
		.client = ctx.client,
		.server = ctx.server,
		.subsystem = "test-zc-cancel",
	};

	thrd_t ct, st;
	ASSERT_THRD_CREATE(&ct, client_zc_open_thread, &oc);
	ASSERT_THRD_CREATE(&st, server_accept_subsys_thread, &sc);
	thrd_join(ct, NULL);
	thrd_join(st, NULL);

	if (oc.client_ch == NULL || sc.server_ch == NULL) {
		if (oc.client_ch != NULL)
			dssh_chan_close(oc.client_ch, -1);
		if (sc.server_ch != NULL)
			dssh_chan_close(sc.server_ch, -1);
		cnd_destroy(&zc_st.cnd);
		mtx_destroy(&zc_st.mtx);
		conn_cleanup(&ctx);
		return TEST_SKIP;
	}

	/* Get a TX buffer, then cancel without sending */
	uint8_t *txbuf;
	size_t max_len;
	int res = dssh_chan_zc_getbuf(oc.client_ch, 0, &txbuf, &max_len);
	ASSERT_OK(res);
	ASSERT_NOT_NULL(txbuf);
	ASSERT_TRUE(max_len > 0);

	/* Cancel -- should release packet-engine ownership without sending */
	res = dssh_chan_zc_cancel(oc.client_ch);
	ASSERT_OK(res);

	/* Verify channel is still usable: getbuf/send should work */
	static const uint8_t msg[] = "after cancel";
	res = dssh_chan_zc_getbuf(oc.client_ch, 0, &txbuf, &max_len);
	ASSERT_OK(res);
	ASSERT_TRUE(max_len >= sizeof(msg));
	memcpy(txbuf, msg, sizeof(msg));
	res = dssh_chan_zc_send(oc.client_ch, sizeof(msg));
	ASSERT_OK(res);

	/* Server reads the data to confirm it arrived */
	int ev = dssh_chan_poll(sc.server_ch, DSSH_POLL_READ, 5000);
	ASSERT_TRUE(ev & DSSH_POLL_READ);

	uint8_t rbuf[256];
	size_t got = 0;
	while (got < sizeof(msg)) {
		int64_t r = dssh_chan_read(sc.server_ch, 0,
		    &rbuf[got], sizeof(rbuf) - got);
		if (r <= 0)
			break;
		got += (size_t)r;
	}
	ASSERT_EQ_U(got, sizeof(msg));
	ASSERT_TRUE(memcmp(rbuf, msg, sizeof(msg)) == 0);

	dssh_chan_close(oc.client_ch, -1);
	dssh_chan_close(sc.server_ch, -1);
	cnd_destroy(&zc_st.cnd);
	mtx_destroy(&zc_st.mtx);
	conn_cleanup(&ctx);
	return TEST_PASS;
}

/* ================================================================
 * ZC NULL parameter guards
 * ================================================================ */

static int
test_zc_open_null(void)
{
	struct dssh_chan_params p;
	dssh_chan_params_init(&p, DSSH_CHAN_SUBSYSTEM);
	dssh_chan_params_set_subsystem(&p, "test");

	/* NULL session */
	ASSERT_NULL(dssh_chan_zc_open(NULL, &p, client_zc_data_cb, NULL));
	/* NULL params (sess=NULL triggers first, so this is a combined check) */
	ASSERT_NULL(dssh_chan_zc_open(NULL, NULL, client_zc_data_cb, NULL));
	/* NULL callback */
	ASSERT_NULL(dssh_chan_zc_open(NULL, &p, NULL, NULL));

	/* ZC getbuf/send/cancel NULL guards */
	uint8_t *buf;
	size_t max_len;
	ASSERT_ERR(dssh_chan_zc_getbuf(NULL, 0, &buf, &max_len),
	    DSSH_ERROR_INVALID);
	ASSERT_ERR(dssh_chan_zc_send(NULL, 0),
	    DSSH_ERROR_INVALID);
	ASSERT_ERR(dssh_chan_zc_cancel(NULL),
	    DSSH_ERROR_INVALID);

	dssh_chan_params_free(&p);
	return TEST_PASS;
}

/* ================================================================
 * Event callback coverage
 * ================================================================ */

/* Dummy event callback for verifying setter stores the pointer */
static void
test_event_cb(dssh_channel ch, const struct dssh_chan_event *event,
    void *cbdata)
{
	(void)ch;
	(void)event;
	(void)cbdata;
}

/* Dummy state for verifying cbdata storage */
struct event_cb_state {
	int dummy;
};

static int
test_session_set_event_cb(void)
{
	struct conn_ctx ctx;
	if (conn_setup_no_start(&ctx) < 0)
		return TEST_SKIP;

	/* NULL sess returns error */
	ASSERT_ERR(dssh_session_set_event_cb(NULL, test_event_cb, NULL),
	    DSSH_ERROR_INVALID);

	/* Set event callback before session_start -- should succeed */
	struct event_cb_state es;
	memset(&es, 0, sizeof(es));
	ASSERT_OK(dssh_session_set_event_cb(ctx.client, test_event_cb, &es));

	/* Verify the callback is stored on the session */
	ASSERT_TRUE(ctx.client->default_event_cb == test_event_cb);
	ASSERT_TRUE(ctx.client->default_event_cbdata == &es);

	/* Set to NULL (disable) */
	ASSERT_OK(dssh_session_set_event_cb(ctx.client, NULL, NULL));
	ASSERT_NULL(ctx.client->default_event_cb);

	conn_cleanup_no_start(&ctx);
	return TEST_PASS;
}

static int
test_session_set_event_cb_toolate(void)
{
	struct conn_ctx ctx;
	if (conn_setup(&ctx) < 0)
		return TEST_SKIP;

	/* After session_start, setting event callback should be refused */
	ASSERT_ERR(dssh_session_set_event_cb(ctx.client, test_event_cb, NULL),
	    DSSH_ERROR_TOOLATE);

	conn_cleanup(&ctx);
	return TEST_PASS;
}

static int
test_chan_set_event_cb(void)
{
	struct conn_ctx ctx;
	if (conn_setup(&ctx) < 0)
		return TEST_SKIP;

	/* NULL guard */
	ASSERT_ERR(dssh_chan_set_event_cb(NULL, test_event_cb, NULL),
	    DSSH_ERROR_INVALID);

	/* Open a channel and set per-channel event callback */
	struct open_exec_ctx oc = {
		.client = ctx.client,
		.server = ctx.server,
		.command = "cmd",
		.cbs = &accept_cbs_all,
		.accept_timeout = 30000,
	};
	if (open_exec_channel(&oc) < 0) {
		conn_cleanup(&ctx);
		return TEST_SKIP;
	}

	struct event_cb_state es;
	memset(&es, 0, sizeof(es));

	/* Set per-channel event callback -- should succeed */
	ASSERT_OK(dssh_chan_set_event_cb(oc.client_ch, test_event_cb, &es));

	/* Verify it's stored on the channel */
	ASSERT_TRUE(oc.client_ch->event_cb == test_event_cb);
	ASSERT_TRUE(oc.client_ch->event_cbdata == &es);

	/* Set to NULL (disable) */
	ASSERT_OK(dssh_chan_set_event_cb(oc.client_ch, NULL, NULL));
	ASSERT_NULL(oc.client_ch->event_cb);

	dssh_chan_close(oc.client_ch, 0);
	dssh_chan_close(oc.server_ch, 0);
	conn_cleanup(&ctx);
	return TEST_PASS;
}

/* ================================================================
 * Event queue limits coverage
 * ================================================================ */

static int
test_session_set_max_events(void)
{
	struct conn_ctx ctx;
	if (conn_setup_no_start(&ctx) < 0)
		return TEST_SKIP;

	/* NULL guard */
	ASSERT_ERR(dssh_session_set_max_events(NULL, 128),
	    DSSH_ERROR_INVALID);

	/* Set custom limit */
	ASSERT_OK(dssh_session_set_max_events(ctx.client, 128));

	/* Set to 0 (unlimited) */
	ASSERT_OK(dssh_session_set_max_events(ctx.client, 0));

	conn_cleanup_no_start(&ctx);
	return TEST_PASS;
}

static int
test_session_set_max_events_toolate(void)
{
	struct conn_ctx ctx;
	if (conn_setup(&ctx) < 0)
		return TEST_SKIP;

	ASSERT_ERR(dssh_session_set_max_events(ctx.client, 128),
	    DSSH_ERROR_TOOLATE);

	conn_cleanup(&ctx);
	return TEST_PASS;
}

static int
test_chan_set_max_events(void)
{
	struct conn_ctx ctx;
	if (conn_setup(&ctx) < 0)
		return TEST_SKIP;

	/* NULL guard */
	ASSERT_ERR(dssh_chan_set_max_events(NULL, 32),
	    DSSH_ERROR_INVALID);

	struct open_exec_ctx oc = {
		.client = ctx.client,
		.server = ctx.server,
		.command = "cmd",
		.cbs = &accept_cbs_all,
		.accept_timeout = 30000,
	};
	if (open_exec_channel(&oc) < 0) {
		conn_cleanup(&ctx);
		return TEST_SKIP;
	}

	/* Set per-channel limit */
	ASSERT_OK(dssh_chan_set_max_events(oc.client_ch, 32));

	/* Set to 0 (unlimited) */
	ASSERT_OK(dssh_chan_set_max_events(oc.client_ch, 0));

	dssh_chan_close(oc.client_ch, 0);
	dssh_chan_close(oc.server_ch, 0);
	conn_cleanup(&ctx);
	return TEST_PASS;
}

/* ================================================================
 * dssh_chan_get_pty coverage
 * ================================================================ */

static int
test_chan_get_pty(void)
{
	struct conn_ctx ctx;
	if (conn_setup(&ctx) < 0)
		return TEST_SKIP;

	/* Open shell with PTY */
	struct open_shell_ctx oc = {
		.client = ctx.client,
		.server = ctx.server,
		.term = "vt100",
		.cols = 132,
		.rows = 43,
		.wpx = 800,
		.hpx = 600,
		.cbs = &accept_cbs_all,
	};

	thrd_t ct, st;
	ASSERT_THRD_CREATE(&ct, client_open_shell_thread, &oc);
	ASSERT_THRD_CREATE(&st, server_accept_shell_thread, &oc);
	thrd_join(ct, NULL);
	thrd_join(st, NULL);

	if (oc.client_ch == NULL || oc.server_ch == NULL) {
		if (oc.client_ch != NULL)
			dssh_chan_close(oc.client_ch, -1);
		if (oc.server_ch != NULL)
			dssh_chan_close(oc.server_ch, -1);
		conn_cleanup(&ctx);
		return TEST_SKIP;
	}

	/* Server channel should have PTY info */
	ASSERT_TRUE(dssh_chan_has_pty(oc.server_ch));
	const struct dssh_chan_params *pty = dssh_chan_get_pty(oc.server_ch);
	ASSERT_NOT_NULL(pty);
	ASSERT_STR_EQ(pty->term, "vt100");
	ASSERT_EQ_U(pty->cols, 132);
	ASSERT_EQ_U(pty->rows, 43);
	ASSERT_EQ_U(pty->wpx, 800);
	ASSERT_EQ_U(pty->hpx, 600);

	/* Client channel should also have PTY (it requested it) */
	ASSERT_TRUE(dssh_chan_has_pty(oc.client_ch));
	const struct dssh_chan_params *cpty = dssh_chan_get_pty(oc.client_ch);
	ASSERT_NOT_NULL(cpty);

	/* Subsystem channel without PTY should return NULL */
	struct open_subsys_ctx sc = {
		.client = ctx.client,
		.server = ctx.server,
		.subsystem = "test-no-pty",
	};
	ASSERT_THRD_CREATE(&ct, client_open_subsys_thread, &sc);
	ASSERT_THRD_CREATE(&st, server_accept_subsys_thread, &sc);
	thrd_join(ct, NULL);
	thrd_join(st, NULL);

	if (sc.client_ch != NULL && sc.server_ch != NULL) {
		ASSERT_FALSE(dssh_chan_has_pty(sc.server_ch));
		ASSERT_NULL(dssh_chan_get_pty(sc.server_ch));
		dssh_chan_close(sc.client_ch, -1);
		dssh_chan_close(sc.server_ch, -1);
	}

	dssh_chan_close(oc.client_ch, 0);
	dssh_chan_close(oc.server_ch, 0);
	conn_cleanup(&ctx);
	return TEST_PASS;
}

/* ================================================================
 * Environment variable roundtrip (exercises accept_parse_env +
 * parse_env_data on the server side)
 * ================================================================ */

static int
accept_env_cb(dssh_channel ch, const struct dssh_chan_params *p, void *cbdata)
{
	(void)ch;
	(void)p;
	(void)cbdata;
	return 0;
}

static int
client_open_exec_with_env_thread(void *arg)
{
	struct open_exec_ctx *ctx = arg;
	struct dssh_chan_params p;
	dssh_chan_params_init(&p, DSSH_CHAN_EXEC);
	dssh_chan_params_set_command(&p, ctx->command);
	dssh_chan_params_set_pty(&p, false);
	dssh_chan_params_add_env(&p, "LANG", "en_US.UTF-8");
	dssh_chan_params_add_env(&p, "TERM", "xterm");
	ctx->client_ch = dssh_chan_open(ctx->client, &p);
	dssh_chan_params_free(&p);
	return 0;
}

static struct dssh_chan_accept_cbs accept_cbs_with_env = {
	.pty_req = accept_pty_cb,
	.env = accept_env_cb,
	.shell = accept_all_shell,
	.exec = accept_all_exec,
	.subsystem = accept_all_subsystem,
};

static int
test_env_roundtrip(void)
{
	struct conn_ctx ctx;
	if (conn_setup(&ctx) < 0)
		return TEST_SKIP;

	struct open_exec_ctx oc = {
		.client = ctx.client,
		.server = ctx.server,
		.command = "env",
		.cbs = &accept_cbs_with_env,
		.accept_timeout = 30000,
	};

	thrd_t ct, st;
	ASSERT_THRD_CREATE(&ct, client_open_exec_with_env_thread, &oc);
	ASSERT_THRD_CREATE(&st, server_accept_exec_thread, &oc);
	thrd_join(ct, NULL);
	thrd_join(st, NULL);

	if (oc.client_ch == NULL || oc.server_ch == NULL) {
		if (oc.client_ch != NULL)
			dssh_chan_close(oc.client_ch, -1);
		if (oc.server_ch != NULL)
			dssh_chan_close(oc.server_ch, -1);
		conn_cleanup(&ctx);
		return TEST_SKIP;
	}

	/* Server channel should have received the env vars */
	ASSERT_EQ_U(oc.server_ch->params.env_count, 2);
	ASSERT_STR_EQ(oc.server_ch->params.env[0].name, "LANG");
	ASSERT_STR_EQ(oc.server_ch->params.env[0].value, "en_US.UTF-8");
	ASSERT_STR_EQ(oc.server_ch->params.env[1].name, "TERM");
	ASSERT_STR_EQ(oc.server_ch->params.env[1].value, "xterm");

	dssh_chan_close(oc.client_ch, 0);
	dssh_chan_close(oc.server_ch, 0);
	conn_cleanup(&ctx);
	return TEST_PASS;
}

/* ================================================================
 * Test table
 * ================================================================ */

static struct dssh_test_entry tests[] = {
	/* Parse helpers */

	/* Session start/stop */
	{ "test_session_start",              test_session_start },
	{ "test_session_stop",               test_session_stop },
	{ "test_session_start_stop",         test_session_start_stop },

	/* Channel open */
	{ "test_open_exec",                  test_open_exec },
	{ "test_open_exec_command",          test_open_exec_command },
	{ "test_open_exec_rejected",         test_open_exec_rejected },
	{ "test_open_shell",                 test_open_shell },
	{ "test_open_subsystem",             test_open_subsystem },

	/* Session I/O */
	{ "test_session_write_read",         test_session_write_read },
	{ "test_session_read_write_reverse", test_session_read_write_reverse },
	{ "test_session_write_ext",          test_session_write_ext },
	{ "test_session_read_ext",           test_session_read_ext },
	{ "test_session_poll_read",          test_session_poll_read },
	{ "test_session_poll_timeout",       test_session_poll_timeout },
	{ "test_session_large_write",        test_session_large_write },
	{ "test_session_multiple_writes",    test_session_multiple_writes },

	/* Subsystem channel I/O */
	{ "test_subsys_write_read",          test_subsys_write_read },
	{ "test_subsys_multiple_writes",     test_subsys_multiple_writes },
	{ "test_subsys_poll",                test_subsys_poll },
	{ "test_subsys_large_write",         test_subsys_large_write },

	/* Signals */
	{ "test_send_signal",                test_send_signal },
	{ "test_signal_after_data",          test_signal_after_data },
	{ "test_window_change",              test_window_change },

	/* Close and exit */
	{ "test_session_close_exit_code",    test_session_close_exit_code },
	{ "test_channel_close_subsys",       test_channel_close_subsys },
	{ "test_eof_detection",              test_eof_detection },
	{ "test_eof_half_close",             test_eof_half_close },
	{ "test_read_nomore_vs_eof",         test_read_nomore_vs_eof },
	{ "test_graceful_disconnect",        test_graceful_disconnect },

	/* Multiple channels */
	{ "test_two_channels",               test_two_channels },
	{ "test_accept_reject",              test_accept_reject },
	{ "test_channel_independence",        test_channel_independence },

	/* Thread safety */
	{ "test_concurrent_write_read",      test_concurrent_write_read },
	{ "test_terminate_during_io",        test_terminate_during_io },

	/* Connection drop */
	{ "test_connection_drop_during_read",  test_connection_drop_during_read },
	{ "test_connection_drop_during_write", test_connection_drop_during_write },
	{ "test_cleanup_after_drop",           test_cleanup_after_drop },

	/* Coverage: edge cases */
	{ "test_session_start_twice",          test_session_start_twice },
	{ "test_accept_timeout_zero",          test_accept_timeout_zero },
	{ "test_accept_timeout_short",         test_accept_timeout_short },
	{ "test_session_poll_timeout_zero",    test_session_poll_timeout_zero },

	/* Auto-reject channel types */
	{ "test_auto_reject_x11",              test_auto_reject_x11 },
	{ "test_auto_reject_forwarded_tcpip",  test_auto_reject_forwarded_tcpip },
	{ "test_auto_reject_direct_tcpip",     test_auto_reject_direct_tcpip },
	{ "test_auto_reject_session_server",   test_auto_reject_session_from_server },

	/* Demux edge cases */
	{ "test_window_adjust_from_peer",      test_window_adjust_from_peer },
	{ "test_data_after_eof",               test_data_after_eof },
	{ "test_truncated_channel_data",       test_truncated_channel_data },
	{ "test_ext_data_after_eof",           test_ext_data_after_eof },
	{ "test_truncated_ext_data",           test_truncated_ext_data },
	{ "test_channel_request_want_reply",   test_channel_request_want_reply },
	{ "test_data_exceeds_window",          test_data_exceeds_window },

	/* Coverage: additional edge cases */
	{ "test_session_poll_write",            test_session_poll_write },
	{ "test_session_write_after_eof",       test_session_write_after_eof },
	{ "test_session_write_ext_after_eof",   test_session_write_ext_after_eof },
	{ "test_subsys_write_after_close",      test_subsys_write_after_close },
	{ "test_subsys_poll_write",             test_subsys_poll_write },
	{ "test_subsys_poll_timeout_zero",      test_subsys_poll_timeout_zero },
	{ "test_read_event_empty",              test_read_event_empty },
	{ "test_signal_interleave_read",        test_signal_interleave_read },
	{ "test_signal_interleave_read_ext",    test_signal_interleave_read_ext },
	{ "test_truncated_channel_open",        test_truncated_channel_open },
	{ "test_truncated_channel_request",     test_truncated_channel_request },
	{ "test_demux_short_payload",           test_demux_short_payload },
	{ "test_session_poll_read_empty",       test_session_poll_read_empty },
	{ "test_session_write_window_zero",     test_session_write_window_zero },
	{ "test_session_write_ext_window_zero", test_session_write_ext_window_zero },
	{ "test_accept_timeout_negative",       test_accept_timeout_negative },
	{ "test_setup_exec_rejected_by_callback", test_setup_exec_rejected_by_callback },
	{ "test_truncated_open_confirmation",   test_truncated_open_confirmation },
	{ "test_open_conf_unknown_channel",    test_open_confirmation_unknown_channel },
	{ "test_channel_success_no_request",   test_channel_success_no_request },
	{ "test_double_eof_close",              test_double_eof_close },
	{ "test_data_dlen_exceeds_payload",     test_data_dlen_exceeds_payload },
	{ "test_demux_chan_type_zero",          test_demux_dispatch_chan_type_zero },
	{ "test_send_eof_already_sent",        test_send_eof_already_sent },
	{ "test_send_close_already_sent",      test_send_close_already_sent },
	{ "test_replenish_after_eof",          test_maybe_replenish_after_eof },
	{ "test_replenish_low_window",         test_maybe_replenish_low_window },
	{ "test_replenish_queued_accounting",  test_replenish_queued_accounting },
	{ "test_zc_consumption_clamped",        test_zc_consumption_clamped },
	{ "test_window_underflow_to_zero",     test_demux_window_underflow_to_zero },
	{ "test_window_add_overflow",          test_window_add_overflow },

	/* Write guard coverage */
	{ "test_chan_write_not_open",           test_chan_write_not_open },
	{ "test_chan_write_close_recv",         test_chan_write_close_received },

	{ "test_subsystem_roundtrip",          test_subsystem_roundtrip },
	{ "test_accept_zc",                    test_accept_zc },
	{ "test_close_during_zc_callback",     test_close_during_zc_callback },
	{ "test_dispatch_sync_failures",       test_dispatch_sync_failures },
	{ "test_detach_sync_failures",         test_detach_sync_failures },

	/* Deterministic branch coverage -- profiling-unstable */
	{ "test_session_poll_nsec_overflow",   test_session_poll_nsec_overflow },
	{ "test_accept_nsec_overflow",         test_accept_nsec_overflow },

	/* Item 74: bytebuf truncation + window accounting */
	{ "test_demux_data_truncation",        test_demux_data_truncation_window },
	{ "test_replenish_caps_buf_space",     test_replenish_caps_to_buffer_space },

	/* Universal rx-window truncation + ACCEPT_EARLY_DATA bypass */
	{ "rx_truncation/default",             test_rx_truncation_default },
	{ "rx_truncation/clips_to_window",     test_rx_truncation_clips_to_window },
	{ "rx_bypass/pre_setup",               test_rx_bypass_pre_setup },
	{ "rx_bypass/disengages_post_setup",   test_rx_bypass_disengages_post_setup },
	{ "early_data/type_lock_subsystem",    test_chan_open_type_lock_subsystem },
	{ "poll/terminate_surfaces_data",      test_poll_terminate_surfaces_data_flags },
	{ "poll/terminate_no_event_bit",       test_poll_terminate_no_event_bit },

	/* NULL parameter validation */
	{ "null/chan_read",                    test_chan_read_null },
	{ "null/chan_write",                   test_chan_write_null },
	{ "null/chan_poll",                    test_chan_poll_null },
	{ "null/chan_close",                   test_chan_close_null },
	{ "null/chan_send_signal",             test_chan_send_signal_null },
	{ "null/chan_send_window_change",      test_chan_send_window_change_null },

	/* Allocation failure injection */
	{ "alloc/open_exec_sweep",             test_open_exec_alloc_sweep },
	{ "alloc/open_shell_sweep",            test_open_shell_alloc_sweep },
	{ "alloc/open_subsystem_sweep",        test_open_subsystem_alloc_sweep },
	{ "alloc/server_accept_sweep",         test_server_accept_alloc_sweep },
	{ "subsys/write_zero",                 test_subsys_write_zero },

	/* Additional branch coverage */
	{ "test_data_after_close",             test_data_after_close },
	{ "test_ext_data_after_close",         test_ext_data_after_close },
	{ "test_replenish_after_close",        test_replenish_after_close },
	{ "test_truncated_window_adjust",      test_truncated_window_adjust },

	/* Item 64: deadline must not reset on spurious wakeup */
	{ "test_poll_deadline_not_reset",      test_poll_deadline_not_reset },

	/* Channel request parser unit tests */
	{ "chanreq/parse_valid",               test_chanreq_parse_valid },
	{ "chanreq/parse_empty_type",          test_chanreq_parse_empty_type },
	{ "chanreq/parse_no_data",             test_chanreq_parse_no_data },
	{ "chanreq/parse_truncated_type_len",  test_chanreq_parse_truncated_type_len },
	{ "chanreq/parse_truncated_type",      test_chanreq_parse_truncated_type },
	{ "chanreq/parse_truncated_want_reply", test_chanreq_parse_truncated_want_reply },

	/* Zero-copy client-side open */
	{ "zc/open_client",                    test_zc_open_client },
	{ "zc/cancel",                         test_zc_cancel },
	{ "null/zc_open",                      test_zc_open_null },

	/* Event callback coverage */
	{ "event/session_set_cb",              test_session_set_event_cb },
	{ "event/session_set_cb_toolate",      test_session_set_event_cb_toolate },
	{ "event/chan_set_cb",                 test_chan_set_event_cb },

	/* Event queue limits */
	{ "event/session_set_max",             test_session_set_max_events },
	{ "event/session_set_max_toolate",     test_session_set_max_events_toolate },
	{ "event/chan_set_max",                test_chan_set_max_events },

	/* PTY getter and env roundtrip */
	{ "pty/get_pty",                       test_chan_get_pty },
	{ "env/roundtrip",                     test_env_roundtrip },
};

static void
dssh_test_after_each(int result)
{
	(void)result;
	if (g_active_ctx != NULL)
		conn_cleanup(g_active_ctx);
}

DSSH_TEST_MAIN(tests)
