make clean   # when switching tiers or unsure
PROXYSQL31=1 make -j$(nproc) debug
PROXYSQL31=1 make build_tap_test_debug

cd test/tap/tests/unit
PROXYSQL31=1 make -j$(nproc) mysql_resultset_framer_unit-t
./mysql_resultset_framer_unit-t

static void test_mysql_is_eof_payload() {
	unsigned char eof5[] = {0xFE, 0x00, 0x00, 0x02, 0x00}; // warnings=0, status=AUTOCOMMIT
	ok(mysql_is_eof_payload(eof5, sizeof(eof5)) == true, "EOF: 5-byte classic EOF");
	unsigned char not_eof[] = {0xFE, 0x01, 0, 0, 0, 0, 0, 0, 0}; // lenenc 8-byte style length
	ok(mysql_is_eof_payload(not_eof, sizeof(not_eof)) == false, "EOF: 0xFE with len>=9 is not EOF");
	unsigned char okpkt[] = {0x00, 0x00, 0x00, 0x02, 0x00, 0x00, 0x00};
	ok(mysql_is_eof_payload(okpkt, sizeof(okpkt)) == false, "EOF: OK is not EOF");
}

static void test_mysql_parse_ok_payload() {
	// OK: header 0x00, affected=1, last_insert=0, status=0x0002, warnings=0
	unsigned char okp[] = {0x00, 0x01, 0x00, 0x02, 0x00, 0x00, 0x00};
	MySQLOkFields f{};
	ok(mysql_parse_ok_payload(okp, sizeof(okp), &f) == true, "OK parse: success");
	ok(f.affected_rows == 1, "OK parse: affected_rows=1");
	ok(f.status_flags == 0x0002, "OK parse: status AUTOCOMMIT");
	ok((f.status_flags & SERVER_MORE_RESULTS_EXIST) == 0, "OK parse: no MORE_RESULTS");
}

static void test_mysql_parse_ok_more_results() {
	// status = SERVER_MORE_RESULTS_EXIST (8) | AUTOCOMMIT (2) = 10
	unsigned char okp[] = {0x00, 0x00, 0x00, 0x0A, 0x00, 0x00, 0x00};
	MySQLOkFields f{};
	ok(mysql_parse_ok_payload(okp, sizeof(okp), &f) == true, "OK more: parse ok");
	ok((f.status_flags & SERVER_MORE_RESULTS_EXIST) != 0, "OK more: MORE_RESULTS set");
}

static void test_mysql_eof_status_flags() {
	unsigned char eof5[] = {0xFE, 0x00, 0x00, 0x0A, 0x00}; // warnings=0, status=0x000A
	uint16_t st = 0, wr = 0;
	ok(mysql_parse_eof_payload(eof5, sizeof(eof5), &wr, &st) == true, "EOF parse: ok");
	ok(st == 0x000A, "EOF parse: status has MORE_RESULTS|AUTOCOMMIT");
}

#ifndef SERVER_MORE_RESULTS_EXIST
#define SERVER_MORE_RESULTS_EXIST 8
#endif

cd /data/rene/proxysql
PROXYSQL31=1 make -j$(nproc) debug 2>&1 | tail -5
cd test/tap/tests/unit && PROXYSQL31=1 make -j$(nproc) ffto_protocol_unit-t 2>&1 | tail -20

struct MySQLOkFields {
	uint64_t affected_rows;
	uint64_t last_insert_id;
	uint16_t status_flags;
	uint16_t warnings;
};

bool mysql_is_eof_payload(const unsigned char* p, size_t n);
bool mysql_is_ok_header(const unsigned char* p, size_t n); // first byte 0x00 and n>=1
bool mysql_parse_ok_payload(const unsigned char* p, size_t n, MySQLOkFields* out);
bool mysql_parse_eof_payload(const unsigned char* p, size_t n, uint16_t* warnings, uint16_t* status_flags);

bool mysql_is_eof_payload(const unsigned char* p, size_t n) {
	// Classic EOF: 0xFE and payload length < 9
	return p && n >= 1 && n < 9 && p[0] == 0xFE;
}

bool mysql_is_ok_header(const unsigned char* p, size_t n) {
	return p && n >= 1 && p[0] == 0x00;
}

bool mysql_parse_ok_payload(const unsigned char* p, size_t n, MySQLOkFields* out) {
	if (!p || !out || n < 1 || p[0] != 0x00) return false;
	const unsigned char* cur = p + 1;
	size_t rem = n - 1;
	out->affected_rows = mysql_read_lenenc_int(cur, rem);
	out->last_insert_id = mysql_read_lenenc_int(cur, rem);
	if (rem < 4) return false;
	out->status_flags = (uint16_t)(cur[0] | (cur[1] << 8));
	out->warnings = (uint16_t)(cur[2] | (cur[3] << 8));
	return true;
}

bool mysql_parse_eof_payload(const unsigned char* p, size_t n, uint16_t* warnings, uint16_t* status_flags) {
	if (!mysql_is_eof_payload(p, n) || n < 5) return false;
	if (warnings) *warnings = (uint16_t)(p[1] | (p[2] << 8));
	if (status_flags) *status_flags = (uint16_t)(p[3] | (p[4] << 8));
	return true;
}

PROXYSQL31=1 make -j$(nproc) debug
cd test/tap/tests/unit && PROXYSQL31=1 make -j$(nproc) ffto_protocol_unit-t && ./ffto_protocol_unit-t

git add include/MySQLProtocolUtils.h lib/MySQLProtocolUtils.cpp test/tap/tests/unit/ffto_protocol_unit-t.cpp
git commit -m "feat(ffto): MySQL OK/EOF payload helpers for resultset framer"

#ifndef MYSQL_RESULTSET_FRAMER_H
#define MYSQL_RESULTSET_FRAMER_H

#include <cstdint>
#include <cstddef>

enum class MySQLRSEventKind {
	None = 0,
	Row,
	OKNoResultset,
	ResultsetComplete,
	Error
};

struct MySQLRSEvent {
	MySQLRSEventKind kind { MySQLRSEventKind::None };
	uint64_t rows_sent_total { 0 };
	uint64_t affected_rows { 0 };
	uint16_t status_flags { 0 };
	uint16_t error_code { 0 };
	bool more_results { false };
};

class MySQLResultsetFramer {
public:
	void reset();
	bool active() const;
	void begin(bool binary_protocol, bool deprecate_eof);
	MySQLRSEvent on_first_payload(const unsigned char* p, size_t n);
	MySQLRSEvent on_payload(const unsigned char* p, size_t n);

private:
	enum class State {
		Idle,
		AwaitingFirst,
		ReadingColumns,
		ReadingRows
	};

	State m_state { State::Idle };
	bool m_binary { false };
	bool m_deprecate_eof { false };
	uint64_t m_column_count { 0 };
	uint64_t m_columns_seen { 0 };
	uint64_t m_rows_sent { 0 };
	bool m_saw_intermediate_eof { false };

	MySQLRSEvent make_complete(uint16_t status, uint64_t affected);
	MySQLRSEvent make_error(uint16_t code);
	bool looks_like_ok_terminator(const unsigned char* p, size_t n) const;
};

#endif

#include "MySQLResultsetFramer.h"
#include "MySQLProtocolUtils.h"
#include <cstring>

void MySQLResultsetFramer::reset() {
	m_state = State::Idle;
	m_binary = false;
	m_deprecate_eof = false;
	m_column_count = 0;
	m_columns_seen = 0;
	m_rows_sent = 0;
	m_saw_intermediate_eof = false;
}

bool MySQLResultsetFramer::active() const {
	return m_state != State::Idle;
}

void MySQLResultsetFramer::begin(bool binary_protocol, bool deprecate_eof) {
	reset();
	m_binary = binary_protocol;
	m_deprecate_eof = deprecate_eof;
	m_state = State::AwaitingFirst;
}

MySQLRSEvent MySQLResultsetFramer::on_first_payload(const unsigned char* p, size_t n) {
	(void)p; (void)n;
	return {};
}

MySQLRSEvent MySQLResultsetFramer::on_payload(const unsigned char* p, size_t n) {
	(void)p; (void)n;
	return {};
}

ifeq ($(PROXYSQLFFTO),1)
_OBJ_CXX += MySQLFFTO.oo PgSQLFFTO.oo MySQLResultsetFramer.oo PgSQLResponseFramer.oo
endif

	mysql_resultset_framer_unit-t \
	ffto_state_machine_unit-t \

#ifdef PROXYSQLFFTO
#include "tap.h"
#include "test_globals.h"
#include "test_init.h"
#include "MySQLResultsetFramer.h"
#include "MySQLProtocolUtils.h"
#include <vector>
#include <cstring>

#ifndef SERVER_MORE_RESULTS_EXIST
#define SERVER_MORE_RESULTS_EXIST 8
#endif

// Helpers to feed column-count + N dummy column defs + terminator
static std::vector<unsigned char> colcount(uint8_t n) { return { n }; }
static std::vector<unsigned char> dummy_coldef() {
	// minimal-ish field packet starting with lenenc "def"
	return { 3,'d','e','f', 0, 0, 0, 0, 0, 0x0c, 0x3f,0x00, 0,0,0,0, 0xfd, 0,0,0,0,0,0 };
}
static std::vector<unsigned char> eof_pkt(uint16_t status = 0x0002) {
	return { 0xFE, 0x00, 0x00, (uint8_t)(status & 0xFF), (uint8_t)(status >> 8) };
}
static std::vector<unsigned char> ok_term(uint16_t status = 0x0002) {
	return { 0x00, 0x00, 0x00, (uint8_t)(status & 0xFF), (uint8_t)(status >> 8), 0x00, 0x00 };
}
static std::vector<unsigned char> text_row_one() {
	// one column value "1" → lenenc 1 + '1'
	return { 0x01, '1' };
}

static void test_text_select_classic_eof() {
	MySQLResultsetFramer f;
	f.begin(false, false);
	auto e = f.on_first_payload(colcount(1).data(), 1);
	ok(e.kind == MySQLRSEventKind::None && f.active(), "classic: after colcount still active");
	e = f.on_payload(dummy_coldef().data(), dummy_coldef().size());
	ok(e.kind == MySQLRSEventKind::None, "classic: coldef");
	e = f.on_payload(eof_pkt().data(), eof_pkt().size());
	ok(e.kind == MySQLRSEventKind::None, "classic: intermediate EOF → rows");
	e = f.on_payload(text_row_one().data(), text_row_one().size());
	ok(e.kind == MySQLRSEventKind::Row && e.rows_sent_total == 1, "classic: row counted");
	e = f.on_payload(eof_pkt().data(), eof_pkt().size());
	ok(e.kind == MySQLRSEventKind::ResultsetComplete && e.rows_sent_total == 1 && !e.more_results,
		"classic: final EOF completes");
	ok(!f.active(), "classic: idle after complete");
}

static void test_text_select_deprecate_eof() {
	MySQLResultsetFramer f;
	f.begin(false, true);
	f.on_first_payload(colcount(1).data(), 1);
	f.on_payload(dummy_coldef().data(), dummy_coldef().size());
	// NO intermediate EOF
	auto e = f.on_payload(text_row_one().data(), text_row_one().size());
	ok(e.kind == MySQLRSEventKind::Row && e.rows_sent_total == 1, "dep_eof: row without intermediate EOF");
	e = f.on_payload(ok_term().data(), ok_term().size());
	ok(e.kind == MySQLRSEventKind::ResultsetComplete && e.rows_sent_total == 1, "dep_eof: OK terminates");
}

static void test_ok_dml() {
	MySQLResultsetFramer f;
	f.begin(false, true);
	unsigned char okp[] = {0x00, 0x05, 0x00, 0x02, 0x00, 0x00, 0x00}; // affected=5
	auto e = f.on_first_payload(okp, sizeof(okp));
	ok(e.kind == MySQLRSEventKind::OKNoResultset && e.affected_rows == 5, "DML OK affected=5");
	ok(!f.active(), "DML: idle");
}

int main() {
	plan(10);
	int rc = test_init_minimal();
	ok(rc == 0, "test_init_minimal");
	test_text_select_classic_eof();
	test_text_select_deprecate_eof();
	test_ok_dml();
	test_cleanup_minimal();
	return exit_status();
}
#else
#include "tap.h"
int main() { plan(1); ok(1, "skipped without PROXYSQLFFTO"); return exit_status(); }
#endif

PROXYSQL31=1 make -j$(nproc) debug
cd test/tap/tests/unit && PROXYSQL31=1 make -j$(nproc) mysql_resultset_framer_unit-t && ./mysql_resultset_framer_unit-t

git add include/MySQLResultsetFramer.h lib/MySQLResultsetFramer.cpp lib/Makefile \
  test/tap/tests/unit/mysql_resultset_framer_unit-t.cpp test/tap/tests/unit/Makefile
git commit -m "test(ffto): MySQLResultsetFramer scaffold and failing unit tests"

MySQLRSEvent MySQLResultsetFramer::on_first_payload(const unsigned char* p, size_t n) {
	if (m_state != State::AwaitingFirst || !p || n == 0) return {};
	if (p[0] == 0xFF) {
		uint16_t code = (n >= 3) ? (uint16_t)(p[1] | (p[2] << 8)) : 0;
		reset();
		return make_error(code);
	}
	if (p[0] == 0x00) {
		MySQLOkFields okf{};
		if (!mysql_parse_ok_payload(p, n, &okf)) { reset(); return {}; }
		bool more = (okf.status_flags & SERVER_MORE_RESULTS_EXIST) != 0;
		MySQLRSEvent ev = make_complete(okf.status_flags, okf.affected_rows);
		ev.kind = MySQLRSEventKind::OKNoResultset;
		ev.more_results = more;
		if (more) {
			// stay in command: prepare for next resultset
			m_rows_sent = 0;
			m_columns_seen = 0;
			m_column_count = 0;
			m_state = State::AwaitingFirst;
		} else {
			reset();
		}
		return ev;
	}
	// column count (lenenc). For small counts first byte is the count.
	const unsigned char* cur = p;
	size_t rem = n;
	m_column_count = mysql_read_lenenc_int(cur, rem);
	m_columns_seen = 0;
	m_rows_sent = 0;
	if (m_column_count == 0) {
		// empty column count edge: treat subsequent as rows phase
		m_state = m_deprecate_eof ? State::ReadingRows : State::ReadingColumns;
		// if !deprecate_eof, still expect EOF before rows; with 0 cols servers send EOF immediately
	}
	m_state = State::ReadingColumns;
	return {};
}

MySQLRSEvent MySQLResultsetFramer::on_payload(const unsigned char* p, size_t n) {
	if (!active() || m_state == State::AwaitingFirst) {
		return on_first_payload(p, n);
	}
	if (!p || n == 0) return {};

	if (m_state == State::ReadingColumns) {
		if (p[0] == 0xFF) {
			uint16_t code = (n >= 3) ? (uint16_t)(p[1] | (p[2] << 8)) : 0;
			reset();
			return make_error(code);
		}
		if (!m_deprecate_eof && mysql_is_eof_payload(p, n)) {
			m_state = State::ReadingRows;
			return {};
		}
		// column definition packet
		m_columns_seen++;
		if (m_columns_seen >= m_column_count) {
			if (m_deprecate_eof) {
				m_state = State::ReadingRows;
			}
			// else wait for intermediate EOF
		}
		return {};
	}

	if (m_state == State::ReadingRows) {
		if (p[0] == 0xFF) {
			uint16_t code = (n >= 3) ? (uint16_t)(p[1] | (p[2] << 8)) : 0;
			MySQLRSEvent ev = make_error(code);
			ev.rows_sent_total = m_rows_sent;
			reset();
			return ev;
		}
		if (!m_deprecate_eof && mysql_is_eof_payload(p, n)) {
			uint16_t st = 0, wr = 0;
			mysql_parse_eof_payload(p, n, &wr, &st);
			return finish_resultset(st, 0);
		}
		if (m_deprecate_eof && looks_like_ok_terminator(p, n)) {
			MySQLOkFields okf{};
			mysql_parse_ok_payload(p, n, &okf);
			return finish_resultset(okf.status_flags, okf.affected_rows);
		}
		// row
		m_rows_sent++;
		MySQLRSEvent ev;
		ev.kind = MySQLRSEventKind::Row;
		ev.rows_sent_total = m_rows_sent;
		return ev;
	}
	return {};
}

// finish_resultset: emit ResultsetComplete; if MORE_RESULTS keep AwaitingFirst else reset
// looks_like_ok_terminator: mysql_parse_ok_payload succeeds; if m_binary, also reject packets
// that match binary row layout (null bitmap length (m_column_count+9)/8 present and n larger than minimal OK in a row-like way).
// Prefer: if parse OK succeeds AND ( !m_binary OR n is consistent with OK and not with mandatory null-bitmap+values ), treat as OK.

static constexpr uint16_t kServerMoreResultsExist = 8;

PROXYSQL31=1 make -j$(nproc) debug
cd test/tap/tests/unit && PROXYSQL31=1 make -j$(nproc) mysql_resultset_framer_unit-t && ./mysql_resultset_framer_unit-t

git add lib/MySQLResultsetFramer.cpp include/MySQLResultsetFramer.h
git commit -m "feat(ffto): MySQLResultsetFramer handles classic and deprecate_eof resultsets"

static void test_multi_result_two_selects_dep_eof() {
	MySQLResultsetFramer f;
	f.begin(false, true);
	// RS1: 1 col, 1 row, OK with MORE_RESULTS
	f.on_first_payload(colcount(1).data(), 1);
	f.on_payload(dummy_coldef().data(), dummy_coldef().size());
	f.on_payload(text_row_one().data(), text_row_one().size());
	auto e = f.on_payload(ok_term(0x0002 | SERVER_MORE_RESULTS_EXIST).data(), 7);
	ok(e.kind == MySQLRSEventKind::ResultsetComplete && e.more_results && e.rows_sent_total == 1,
		"multi: first RS complete more=1");
	ok(f.active(), "multi: still active for next RS");
	// RS2
	f.on_payload(colcount(1).data(), 1); // or on_first via on_payload idle-first path
	// Use on_payload which delegates to on_first when AwaitingFirst
	f.on_payload(dummy_coldef().data(), dummy_coldef().size());
	f.on_payload(text_row_one().data(), text_row_one().size());
	e = f.on_payload(ok_term(0x0002).data(), 7);
	ok(e.kind == MySQLRSEventKind::ResultsetComplete && !e.more_results && e.rows_sent_total == 1,
		"multi: second RS complete more=0");
}

static void test_binary_row_not_ok() {
	MySQLResultsetFramer f;
	f.begin(true, true);
	f.on_first_payload(colcount(1).data(), 1);
	f.on_payload(dummy_coldef().data(), dummy_coldef().size());
	// binary row: 0x00 + null_bitmap (1 byte for 1 col) + int32 value 7
	unsigned char brow[] = { 0x00, 0x00, 0x07, 0x00, 0x00, 0x00 };
	auto e = f.on_payload(brow, sizeof(brow));
	ok(e.kind == MySQLRSEventKind::Row, "binary: row not mistaken for OK");
	e = f.on_payload(ok_term().data(), ok_term().size());
	ok(e.kind == MySQLRSEventKind::ResultsetComplete, "binary: OK ends RS");
}

static void test_err_mid_rows() {
	MySQLResultsetFramer f;
	f.begin(false, true);
	f.on_first_payload(colcount(1).data(), 1);
	f.on_payload(dummy_coldef().data(), dummy_coldef().size());
	unsigned char err[] = { 0xFF, 0x15, 0x04, '#', 'H','Y','0','0','0', 'x' };
	auto e = f.on_payload(err, sizeof(err));
	ok(e.kind == MySQLRSEventKind::Error && e.error_code == 1045, "ERR mid-rows");
	ok(!f.active(), "idle after ERR");
}

PROXYSQL31=1 make -j$(nproc) debug
cd test/tap/tests/unit && PROXYSQL31=1 make mysql_resultset_framer_unit-t && ./mysql_resultset_framer_unit-t

git add lib/MySQLResultsetFramer.cpp test/tap/tests/unit/mysql_resultset_framer_unit-t.cpp
git commit -m "test(ffto): multi-result and binary-vs-OK coverage for MySQL framer"

} else if (command == _MYSQL_COM_STMT_FETCH) {
	if (len >= 5) {
		uint32_t stmt_id; memcpy(&stmt_id, data + 1, 4);
		auto it = m_statements.find(stmt_id);
		if (it != m_statements.end()) {
			// optional: finalize previous in-flight
			m_current_query = it->second;
			m_query_start_time = monotonic_time();
			m_affected_rows = 0; m_rows_sent = 0;
			m_state = AWAITING_RESPONSE;
			bool dep = /* client_flag & CLIENT_DEPRECATE_EOF */ false;
			if (m_session && m_session->client_myds && m_session->client_myds->myconn)
				dep = (m_session->client_myds->myconn->options.client_flag & CLIENT_DEPRECATE_EOF);
			m_rs.begin(/*binary*/true, dep);
		}
	}
}

if (m_state == AWAITING_PREPARE_OK) { /* existing prepare OK handling */ return; }

if (m_state == AWAITING_RESPONSE || m_rs.active()) {
	MySQLRSEvent ev = m_rs.active() && m_rs /* already begun */
		? (/* if first packet of command already consumed use on_payload */ m_rs.on_payload(data, len))
		: m_rs.on_first_payload(data, len);
	// Simpler approach: always begin() on client command; always on_payload on server which
	// routes AwaitingFirst internally via on_first when state is AwaitingFirst.
	// So server only calls: MySQLRSEvent ev = m_rs.on_payload(data, len);
	// and on_payload delegates to on_first when AwaitingFirst.

	switch (ev.kind) {
	case MySQLRSEventKind::Row:
		m_rows_sent = ev.rows_sent_total; // or accumulate across multi RS in FFTO
		break;
	case MySQLRSEventKind::OKNoResultset:
		m_affected_rows += ev.affected_rows;
		if (!ev.more_results) {
			report_query_stats(m_current_query, monotonic_time() - m_query_start_time, m_affected_rows, m_rows_sent);
			m_state = IDLE; clear_active_query(); m_rs.reset();
		}
		break;
	case MySQLRSEventKind::ResultsetComplete:
		m_rows_sent += /* delta: track last */ // easiest: keep m_rows_sent as sum; framer returns per-RS total
		// Spec: accumulate across multi-result
		// Implementation: on each Complete, m_rows_sent += ev.rows_sent_total (framer resets per RS)
		if (!ev.more_results) {
			report_query_stats(...);
			m_state = IDLE; clear_active_query(); m_rs.reset();
		}
		break;
	case MySQLRSEventKind::Error:
		report_query_stats(...);
		report_error(data, len);
		m_state = IDLE; clear_active_query(); m_rs.reset();
		break;
	default: break;
	}
}

PROXYSQL31=1 make -j$(nproc) debug
cd test/tap/tests/unit && PROXYSQL31=1 make ffto_state_machine_unit-t mysql_resultset_framer_unit-t && ./ffto_state_machine_unit-t && ./mysql_resultset_framer_unit-t

git add include/MySQLFFTO.hpp lib/MySQLFFTO.cpp
git commit -m "feat(ffto): wire MySQLFFTO to ResultsetFramer; track COM_STMT_FETCH"

#ifndef PGSQL_RESPONSE_FRAMER_H
#define PGSQL_RESPONSE_FRAMER_H

#include <cstdint>
#include <cstddef>

enum class PgSQLRSEventKind {
	None = 0,
	CommandComplete,
	EmptyQuery,
	PortalSuspended,
	ReadyForQuery,
	Error
};

struct PgSQLRSEvent {
	PgSQLRSEventKind kind { PgSQLRSEventKind::None };
	uint64_t rows { 0 };
	bool is_select { false };
};

class PgSQLResponseFramer {
public:
	void reset();
	void set_finalize_on_sync(bool v);
	bool response_seen() const { return m_response_seen; }
	bool finalize_on_sync() const { return m_finalize_on_sync; }

	// Returns event; sets m_response_seen on C/I/s; does not clear on Z (caller decides)
	PgSQLRSEvent on_message(char type, const unsigned char* payload, size_t len);

	// Call when FFTO activates a new current query
	void on_query_activated(bool finalize_on_sync);

private:
	bool m_finalize_on_sync { false };
	bool m_response_seen { false };
};

#endif

#include "PgSQLResponseFramer.h"
#include "PgSQLCommandComplete.h"

void PgSQLResponseFramer::reset() {
	m_finalize_on_sync = false;
	m_response_seen = false;
}

void PgSQLResponseFramer::on_query_activated(bool finalize_on_sync) {
	m_finalize_on_sync = finalize_on_sync;
	m_response_seen = false;
}

void PgSQLResponseFramer::set_finalize_on_sync(bool v) { m_finalize_on_sync = v; }

PgSQLRSEvent PgSQLResponseFramer::on_message(char type, const unsigned char* payload, size_t len) {
	PgSQLRSEvent ev;
	switch (type) {
	case 'C': {
		auto r = parse_pgsql_command_complete(payload, len);
		ev.kind = PgSQLRSEventKind::CommandComplete;
		ev.rows = r.rows;
		ev.is_select = r.is_select;
		m_response_seen = true;
		return ev;
	}
	case 'I':
		ev.kind = PgSQLRSEventKind::EmptyQuery;
		m_response_seen = true;
		return ev;
	case 's':
		ev.kind = PgSQLRSEventKind::PortalSuspended;
		m_response_seen = true;
		return ev;
	case 'Z':
		ev.kind = PgSQLRSEventKind::ReadyForQuery;
		return ev;
	case 'E':
		ev.kind = PgSQLRSEventKind::Error;
		return ev;
	default:
		return ev;
	}
}

static void test_command_complete_select() {
	PgSQLResponseFramer f;
	f.on_query_activated(true);
	const char* tag = "SELECT 3";
	auto e = f.on_message('C', (const unsigned char*)tag, 8);
	ok(e.kind == PgSQLRSEventKind::CommandComplete && e.rows == 3 && e.is_select, "C SELECT");
	ok(f.response_seen(), "response_seen after C");
}

static void test_portal_suspended() {
	PgSQLResponseFramer f;
	f.on_query_activated(false);
	auto e = f.on_message('s', nullptr, 0);
	ok(e.kind == PgSQLRSEventKind::PortalSuspended && f.response_seen(), "s sets seen");
}

static void test_ready_for_query_no_side_effect() {
	PgSQLResponseFramer f;
	f.on_query_activated(true);
	auto e = f.on_message('Z', (const unsigned char*)"I", 1);
	ok(e.kind == PgSQLRSEventKind::ReadyForQuery, "Z event");
	ok(!f.response_seen(), "Z alone does not set response_seen");
}

PROXYSQL31=1 make -j$(nproc) debug
cd test/tap/tests/unit && PROXYSQL31=1 make pgsql_response_framer_unit-t && ./pgsql_response_framer_unit-t

git add include/PgSQLResponseFramer.h lib/PgSQLResponseFramer.cpp lib/Makefile \
  test/tap/tests/unit/pgsql_response_framer_unit-t.cpp test/tap/tests/unit/Makefile
git commit -m "feat(ffto): extract PgSQLResponseFramer with unit tests"

void PgSQLFFTO::process_server_message(char type, const unsigned char* payload, size_t len) {
	if (m_state == IDLE) return;
	PgSQLRSEvent ev = m_rs.on_message(type, payload, len);
	switch (ev.kind) {
	case PgSQLRSEventKind::CommandComplete:
		if (ev.is_select) m_rows_sent += ev.rows;
		else m_affected_rows += ev.rows;
		m_response_seen = m_rs.response_seen();
		if (!m_current_finalize_on_sync) finalize_current_query();
		break;
	case PgSQLRSEventKind::EmptyQuery:
	case PgSQLRSEventKind::PortalSuspended:
		m_response_seen = true;
		if (!m_current_finalize_on_sync) finalize_current_query();
		break;
	case PgSQLRSEventKind::ReadyForQuery:
		if (m_rs.response_seen() || m_response_seen) finalize_current_query();
		break;
	case PgSQLRSEventKind::Error:
		// existing report path
		...
		m_rs.reset();
		break;
	default: break;
	}
}

PROXYSQL31=1 make -j$(nproc) debug
cd test/tap/tests/unit && PROXYSQL31=1 make pgsql_response_framer_unit-t ffto_state_machine_unit-t && ./pgsql_response_framer_unit-t && ./ffto_state_machine_unit-t

git add include/PgSQLFFTO.hpp lib/PgSQLFFTO.cpp
git commit -m "feat(ffto): wire PgSQLFFTO to PgSQLResponseFramer (behavior-preserving)"

SET mysql-enable_client_deprecate_eof='true'|'false';
SET mysql-enable_server_deprecate_eof='true'|'false';
LOAD MYSQL VARIABLES TO RUNTIME;

SELECT 1; SELECT 2,3;

PROXYSQL31=1 make -j$(nproc) debug
PROXYSQL31=1 make build_tap_test_debug
# determine TAP_GROUP from groups.json for the new tests
WORKSPACE=$(pwd) INFRA_ID=dev-$USER TAP_GROUP=<group> \
  TEST_PY_TAP_INCL='ffto_mysql_deprecate_eof|ffto_mysql_multi_result|ffto_mysql_stmt_fetch' \
  test/infra/control/run-tests-isolated.bash

git add test/tap/tests/test_ffto_mysql_deprecate_eof-t.cpp \
  test/tap/tests/test_ffto_mysql_multi_result-t.cpp \
  test/tap/tests/test_ffto_mysql_stmt_fetch-t.cpp \
  test/tap/groups/groups.json
git commit -m "test(ffto): TAP coverage for deprecate_eof, multi-result, COM_STMT_FETCH"

WORKSPACE=$(pwd) INFRA_ID=dev-$USER TAP_GROUP=<mysql-ffto-group> \
  TEST_PY_TAP_INCL='ffto_mysql' \
  test/infra/control/run-tests-isolated.bash

WORKSPACE=$(pwd) INFRA_ID=dev-$USER TAP_GROUP=<pgsql-ffto-group> \
  TEST_PY_TAP_INCL='ffto_pgsql' \
  test/infra/control/run-tests-isolated.bash

cd test/tap/tests/unit
PROXYSQL31=1 make -j$(nproc) \
  mysql_resultset_framer_unit-t pgsql_response_framer_unit-t \
  ffto_protocol_unit-t ffto_state_machine_unit-t pgsql_command_complete_unit-t
./mysql_resultset_framer_unit-t
./pgsql_response_framer_unit-t
./ffto_protocol_unit-t
./ffto_state_machine_unit-t
./pgsql_command_complete_unit-t

git status
# optional: short note in CHANGELOG if project requires it for the feature PR