add MasterConn with master handler registration and dispatch

qkc/cluster/slave/compat_test.go

@@ -4,7 +4,9 @@ package slave
 
 import (
 	"bufio"
+	"bytes"
 	"context"
+	"encoding/binary"
 	"fmt"
 	"net"
 	"os"
@@ -12,12 +14,122 @@ import (
 	"path/filepath"
 	"runtime"
 	"strings"
+	"sync"
 	"testing"
 	"time"
 
 	"github.com/ethereum/go-ethereum/log"
+	"github.com/ethereum/go-ethereum/qkc/cluster/wire"
 )
 
+// startPythonMaster starts the Python master.py subprocess and returns the
+// TCP port it listens on, a function to retrieve captured stdout lines, and a
+// cleanup function. The peer listens on a random port (port=0) and prints
+// "PORT:<port>" to stdout when ready.
+func startPythonMaster(t *testing.T, extraArgs ...string) (int, func() []string, func()) {
+	t.Helper()
+
+	_, filename, _, ok := runtime.Caller(0)
+	if !ok {
+		t.Fatal("cannot get caller path")
+	}
+	pyScript := filepath.Join(filepath.Dir(filename), "testdata", "pyproto", "master.py")
+
+	if _, err := exec.LookPath("python3"); err != nil {
+		t.Skip("python3 not found in PATH")
+	}
+	if _, err := os.Stat(pyScript); err != nil {
+		t.Skipf("master.py not found at %s", pyScript)
+	}
+
+	args := []string{pyScript, "--port", "0", "--id", "py-master", "--shards", "1,2"}
+	args = append(args, extraArgs...)
+
+	cmd := exec.Command("python3", args...)
+	stdout, err := cmd.StdoutPipe()
+	if err != nil {
+		t.Fatalf("stdout pipe: %v", err)
+	}
+	cmd.Stderr = os.Stderr
+
+	if err := cmd.Start(); err != nil {
+		t.Fatalf("start python master: %v", err)
+	}
+
+	portCh := make(chan int, 1)
+	var outputLines []string
+	var outputMu sync.Mutex
+	go func() {
+		scanner := bufio.NewScanner(stdout)
+		for scanner.Scan() {
+			line := scanner.Text()
+			outputMu.Lock()
+			outputLines = append(outputLines, line)
+			outputMu.Unlock()
+			if strings.HasPrefix(line, "PORT:") {
+				var port int
+				if _, err := fmt.Sscanf(line, "PORT:%d", &port); err == nil {
+					portCh <- port
+				}
+			}
+		}
+	}()
+
+	var port int
+	select {
+	case port = <-portCh:
+	case <-time.After(5 * time.Second):
+		cmd.Process.Kill()
+		cmd.Wait()
+		t.Fatal("timeout waiting for python master port")
+	}
+
+	getOutput := func() []string {
+		outputMu.Lock()
+		defer outputMu.Unlock()
+		out := make([]string, len(outputLines))
+		copy(out, outputLines)
+		return out
+	}
+
+	cleanup := func() {
+		cmd.Process.Kill()
+		cmd.Wait()
+	}
+
+	return port, getOutput, cleanup
+}
+
+// dialPythonMaster starts python master.py and dials the port it listens on,
+// wrapping the connection in a MasterConn. Returns the MasterConn, a function
+// to retrieve captured stdout lines, and a cleanup function.
+func dialPythonMaster(t *testing.T) (*MasterConn, func() []string, func()) {
+	t.Helper()
+
+	port, getOutput, cleanupPy := startPythonMaster(t)
+
+	addr := fmt.Sprintf("127.0.0.1:%d", port)
+	mc, err := NewMasterConn(
+		addr,
+		0,
+		[]byte("go-slave"),
+		[]uint32{0x00010001, 0x00020001},
+		log.New(),
+	)
+	if err != nil {
+		cleanupPy()
+		t.Fatalf("create MasterConn: %v", err)
+	}
+	mc.Start()
+
+	cleanup := func() {
+		mc.Close()
+		cleanupPy()
+	}
+
+	return mc, getOutput, cleanup
+}
+
 // startPythonPeer starts a Python protocol peer subprocess and returns the
 // TCP port and a cleanup function. The peer listens on a random port (port=0)
 // and prints "PORT:<port>" to stdout when ready.
@@ -317,3 +429,84 @@ func TestPythonCompat_PoolReconnect(t *testing.T) {
 		t.Fatalf("pool size after reconnect: got %d, want 1", pool.OutboundSize())
 	}
 }
+
+// ---------------------------------------------------------------------------
+// Test: Python Master -> Go Slave full handshake + RPC flow
+//
+// Validates: Python MasterConnection behavior against Go MasterConn.
+// Python sends PING, GetEcoInfoListRequest, AddRootBlockRequest, and
+// DestroyClusterPeerConnectionCommand. Go must decode the 12-byte
+// ClusterMetadata frames, dispatch to the correct handlers, and return
+// protocol-compatible responses.
+// ---------------------------------------------------------------------------
+func TestPythonCompat_MasterFullFlow(t *testing.T) {
+	mc, getOutput, cleanup := dialPythonMaster(t)
+	defer cleanup()
+
+	// Wait for the Python master to finish its scripted exchange.
+	select {
+	case <-mc.WaitUntilClosed():
+	case <-time.After(15 * time.Second):
+		output := getOutput()
+		t.Fatalf("MasterConn did not close after Python master finished; output=%v", output)
+	}
+
+	// Allow a moment for the scanner goroutine to drain the Python stdout pipe.
+	time.Sleep(100 * time.Millisecond)
+
+	output := getOutput()
+	expected := []string{
+		"PONG_OK id=676f2d736c617665", // hex of "go-slave"
+		"ECO_OK error_code=0",
+		"ROOT_OK error_code=0",
+		"DESTROY_OK",
+		"PONG_OK id=676f2d736c617665",
+		"DISCONNECTED",
+	}
+	for _, exp := range expected {
+		found := false
+		for _, line := range output {
+			if line == exp {
+				found = true
+				break
+			}
+		}
+		if !found {
+			t.Fatalf("expected output line %q not found in %v", exp, output)
+		}
+	}
+}
+
+// ---------------------------------------------------------------------------
+// Test: Python-generated ClusterMetadata frame layout
+//
+// Validates: the 12-byte ClusterMetadata encoding (4-byte branch + 8-byte
+// cluster_peer_id) is the same on both sides of the wire.
+// ---------------------------------------------------------------------------
+func TestPythonCompat_MasterFrameLayout(t *testing.T) {
+	// This is a static golden-vector test: we compare Go's wire format against
+	// the documented Python frame layout without requiring a Python subprocess.
+	meta := wire.ClusterMetadata{Branch: 0x01020304, ClusterPeerID: 0x1122334455667788}
+	frame := &wire.Frame{
+		Meta:    meta,
+		Opcode:  byte(wire.ClusterOpPing),
+		RPCID:   1,
+		Payload: []byte{0xAA, 0xBB},
+	}
+
+	var buf bytes.Buffer
+	if err := wire.WriteFrame(&buf, frame); err != nil {
+		t.Fatalf("WriteFrame: %v", err)
+	}
+
+	wireBytes := buf.Bytes()
+	if len(wireBytes) != 4+12+1+8+2 {
+		t.Fatalf("frame length: got %d, want %d", len(wireBytes), 4+12+1+8+2)
+	}
+	if binary.BigEndian.Uint32(wireBytes[4:8]) != meta.Branch {
+		t.Fatalf("branch mismatch")
+	}
+	if binary.BigEndian.Uint64(wireBytes[8:16]) != meta.ClusterPeerID {
+		t.Fatalf("cluster_peer_id mismatch")
+	}
+}

qkc/cluster/slave/master_conn.go

@@ -0,0 +1,550 @@
+// Copyright 2026-2027, QuarkChain.
+
+package slave
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"net"
+
+	"github.com/ethereum/go-ethereum/log"
+	"github.com/ethereum/go-ethereum/qkc/cluster/wire"
+	"github.com/ethereum/go-ethereum/qkc/serialize"
+)
+
+// MasterConn represents the slave-side TCP connection to the cluster master.
+// It corresponds to Python's quarkchain.cluster.slave.MasterConnection and uses
+// 12-byte ClusterMetadata framing.
+//
+// Architecture:
+//
+//	MasterConn embeds *rpcConn
+//
+// All master→slave ClusterOp handlers are registered during construction.
+// Business handlers that depend on unported components (Shard, StateDB, etc.)
+// are implemented as protocol-compatible stubs that return valid responses.
+type MasterConn struct {
+	*rpcConn
+
+	localID              []byte
+	localFullShardIDList []uint32
+}
+
+// NewMasterConn dials the master at addr and returns a MasterConn.
+// maxPayloadSize controls frame payload size limit; 0 disables the limit.
+// localID and localFullShardIDList identify this slave and are used in PONG.
+func NewMasterConn(addr string, maxPayloadSize uint32, localID []byte, localFullShardIDList []uint32, logger log.Logger) (*MasterConn, error) {
+	conn, err := net.DialTimeout("tcp", addr, defaultDialTimeout)
+	if err != nil {
+		return nil, fmt.Errorf("dial master %s: %w", addr, err)
+	}
+	return newMasterConn(conn, maxPayloadSize, localID, localFullShardIDList, logger), nil
+}
+
+// NewMasterConnFromConn wraps an accepted net.Conn as a MasterConn.
+// maxPayloadSize controls frame payload size limit; 0 disables the limit.
+func NewMasterConnFromConn(conn net.Conn, maxPayloadSize uint32, localID []byte, localFullShardIDList []uint32, logger log.Logger) *MasterConn {
+	return newMasterConn(conn, maxPayloadSize, localID, localFullShardIDList, logger)
+}
+
+func newMasterConn(conn net.Conn, maxPayloadSize uint32, localID []byte, localFullShardIDList []uint32, logger log.Logger) *MasterConn {
+	readFrame := func(r io.Reader) (*wire.Frame, error) {
+		return wire.ReadFrame(r, maxPayloadSize)
+	}
+	mc := &MasterConn{
+		rpcConn:              newRPCConn(conn, readFrame, wire.WriteFrame, logger),
+		localID:              append([]byte(nil), localID...),
+		localFullShardIDList: append([]uint32(nil), localFullShardIDList...),
+	}
+
+	mc.registerOpSerializers()
+	mc.registerHandlers()
+
+	return mc
+}
+
+// registerOpSerializers registers serializers for every opcode in Python's
+// CLUSTER_OP_SERIALIZER_MAP. This covers master→slave, slave→master and
+// slave→slave opcodes so outbound RPC responses can be deserialized if needed.
+func (mc *MasterConn) registerOpSerializers() {
+	mc.rpcConn.RegisterOpSerializers(map[byte]*OpSerializer{
+		// §1 Cluster initialisation
+		byte(wire.ClusterOpPing):                          OpSerializerFor[wire.PingRequest, wire.PongResponse](),
+		byte(wire.ClusterOpPong):                          OpSerializerFor[wire.PongResponse, wire.PingRequest](),
+		byte(wire.ClusterOpConnectToSlavesRequest):        OpSerializerFor[wire.ConnectToSlavesRequest, wire.ConnectToSlavesResponse](),
+		byte(wire.ClusterOpConnectToSlavesResponse):       OpSerializerFor[wire.ConnectToSlavesResponse, wire.ConnectToSlavesRequest](),
+		byte(wire.ClusterOpAddRootBlockRequest):           OpSerializerFor[wire.AddRootBlockRequest, wire.AddRootBlockResponse](),
+		byte(wire.ClusterOpAddRootBlockResponse):          OpSerializerFor[wire.AddRootBlockResponse, wire.AddRootBlockRequest](),
+		byte(wire.ClusterOpGetEcoInfoListRequest):         OpSerializerFor[wire.GetEcoInfoListRequest, wire.GetEcoInfoListResponse](),
+		byte(wire.ClusterOpGetEcoInfoListResponse):        OpSerializerFor[wire.GetEcoInfoListResponse, wire.GetEcoInfoListRequest](),
+		byte(wire.ClusterOpGetNextBlockToMineRequest):     OpSerializerFor[wire.GetNextBlockToMineRequest, wire.GetNextBlockToMineResponse](),
+		byte(wire.ClusterOpGetNextBlockToMineResponse):    OpSerializerFor[wire.GetNextBlockToMineResponse, wire.GetNextBlockToMineRequest](),
+		byte(wire.ClusterOpGetUnconfirmedHeadersRequest):  OpSerializerFor[wire.GetUnconfirmedHeadersRequest, wire.GetUnconfirmedHeadersResponse](),
+		byte(wire.ClusterOpGetUnconfirmedHeadersResponse): OpSerializerFor[wire.GetUnconfirmedHeadersResponse, wire.GetUnconfirmedHeadersRequest](),
+		byte(wire.ClusterOpGetAccountDataRequest):         OpSerializerFor[wire.GetAccountDataRequest, wire.GetAccountDataResponse](),
+		byte(wire.ClusterOpGetAccountDataResponse):        OpSerializerFor[wire.GetAccountDataResponse, wire.GetAccountDataRequest](),
+		byte(wire.ClusterOpAddTransactionRequest):         OpSerializerFor[wire.AddTransactionRequest, wire.AddTransactionResponse](),
+		byte(wire.ClusterOpAddTransactionResponse):        OpSerializerFor[wire.AddTransactionResponse, wire.AddTransactionRequest](),
+
+		// §2 Slave → Master (mining)
+		byte(wire.ClusterOpAddMinorBlockHeaderRequest):  OpSerializerFor[wire.AddMinorBlockHeaderRequest, wire.AddMinorBlockHeaderResponse](),
+		byte(wire.ClusterOpAddMinorBlockHeaderResponse): OpSerializerFor[wire.AddMinorBlockHeaderResponse, wire.AddMinorBlockHeaderRequest](),
+
+		// §3 Slave ↔ Slave (xshard direct)
+		byte(wire.ClusterOpAddXshardTxListRequest):  OpSerializerFor[wire.AddXshardTxListRequest, wire.AddXshardTxListResponse](),
+		byte(wire.ClusterOpAddXshardTxListResponse): OpSerializerFor[wire.AddXshardTxListResponse, wire.AddXshardTxListRequest](),
+
+		// §4 Master → Slave (sync / virtual conns)
+		byte(wire.ClusterOpSyncMinorBlockListRequest):           OpSerializerFor[wire.SyncMinorBlockListRequest, wire.SyncMinorBlockListResponse](),
+		byte(wire.ClusterOpSyncMinorBlockListResponse):          OpSerializerFor[wire.SyncMinorBlockListResponse, wire.SyncMinorBlockListRequest](),
+		byte(wire.ClusterOpAddMinorBlockRequest):                OpSerializerFor[wire.AddMinorBlockRequest, wire.AddMinorBlockResponse](),
+		byte(wire.ClusterOpAddMinorBlockResponse):               OpSerializerFor[wire.AddMinorBlockResponse, wire.AddMinorBlockRequest](),
+		byte(wire.ClusterOpCreateClusterPeerConnectionRequest):  OpSerializerFor[wire.CreateClusterPeerConnectionRequest, wire.CreateClusterPeerConnectionResponse](),
+		byte(wire.ClusterOpCreateClusterPeerConnectionResponse): OpSerializerFor[wire.CreateClusterPeerConnectionResponse, wire.CreateClusterPeerConnectionRequest](),
+		byte(wire.ClusterOpDestroyClusterPeerConnectionCommand): OpSerializerFor[wire.DestroyClusterPeerConnectionCommand, wire.DestroyClusterPeerConnectionCommand](),
+		byte(wire.ClusterOpGetMinorBlockRequest):                OpSerializerFor[wire.GetMinorBlockRequest, wire.GetMinorBlockResponse](),
+		byte(wire.ClusterOpGetMinorBlockResponse):               OpSerializerFor[wire.GetMinorBlockResponse, wire.GetMinorBlockRequest](),
+		byte(wire.ClusterOpGetTransactionRequest):               OpSerializerFor[wire.GetTransactionRequest, wire.GetTransactionResponse](),
+		byte(wire.ClusterOpGetTransactionResponse):              OpSerializerFor[wire.GetTransactionResponse, wire.GetTransactionRequest](),
+
+		// §5 Slave ↔ Slave (xshard batch)
+		byte(wire.ClusterOpBatchAddXshardTxListRequest):  OpSerializerFor[wire.BatchAddXshardTxListRequest, wire.BatchAddXshardTxListResponse](),
+		byte(wire.ClusterOpBatchAddXshardTxListResponse): OpSerializerFor[wire.BatchAddXshardTxListResponse, wire.BatchAddXshardTxListRequest](),
+
+		// §6 Master → Slave (JSON-RPC-like)
+		byte(wire.ClusterOpExecuteTransactionRequest):           OpSerializerFor[wire.ExecuteTransactionRequest, wire.ExecuteTransactionResponse](),
+		byte(wire.ClusterOpExecuteTransactionResponse):          OpSerializerFor[wire.ExecuteTransactionResponse, wire.ExecuteTransactionRequest](),
+		byte(wire.ClusterOpGetTransactionReceiptRequest):        OpSerializerFor[wire.GetTransactionReceiptRequest, wire.GetTransactionReceiptResponse](),
+		byte(wire.ClusterOpGetTransactionReceiptResponse):       OpSerializerFor[wire.GetTransactionReceiptResponse, wire.GetTransactionReceiptRequest](),
+		byte(wire.ClusterOpMineRequest):                         OpSerializerFor[wire.MineRequest, wire.MineResponse](),
+		byte(wire.ClusterOpMineResponse):                        OpSerializerFor[wire.MineResponse, wire.MineRequest](),
+		byte(wire.ClusterOpGenTxRequest):                        OpSerializerFor[wire.GenTxRequest, wire.GenTxResponse](),
+		byte(wire.ClusterOpGenTxResponse):                       OpSerializerFor[wire.GenTxResponse, wire.GenTxRequest](),
+		byte(wire.ClusterOpGetTransactionListByAddressRequest):  OpSerializerFor[wire.GetTransactionListByAddressRequest, wire.GetTransactionListByAddressResponse](),
+		byte(wire.ClusterOpGetTransactionListByAddressResponse): OpSerializerFor[wire.GetTransactionListByAddressResponse, wire.GetTransactionListByAddressRequest](),
+		byte(wire.ClusterOpGetLogRequest):                       OpSerializerFor[wire.GetLogRequest, wire.GetLogResponse](),
+		byte(wire.ClusterOpGetLogResponse):                      OpSerializerFor[wire.GetLogResponse, wire.GetLogRequest](),
+		byte(wire.ClusterOpEstimateGasRequest):                  OpSerializerFor[wire.EstimateGasRequest, wire.EstimateGasResponse](),
+		byte(wire.ClusterOpEstimateGasResponse):                 OpSerializerFor[wire.EstimateGasResponse, wire.EstimateGasRequest](),
+		byte(wire.ClusterOpGetStorageRequest):                   OpSerializerFor[wire.GetStorageRequest, wire.GetStorageResponse](),
+		byte(wire.ClusterOpGetStorageResponse):                  OpSerializerFor[wire.GetStorageResponse, wire.GetStorageRequest](),
+		byte(wire.ClusterOpGetCodeRequest):                      OpSerializerFor[wire.GetCodeRequest, wire.GetCodeResponse](),
+		byte(wire.ClusterOpGetCodeResponse):                     OpSerializerFor[wire.GetCodeResponse, wire.GetCodeRequest](),
+		byte(wire.ClusterOpGasPriceRequest):                     OpSerializerFor[wire.GasPriceRequest, wire.GasPriceResponse](),
+		byte(wire.ClusterOpGasPriceResponse):                    OpSerializerFor[wire.GasPriceResponse, wire.GasPriceRequest](),
+		byte(wire.ClusterOpGetWorkRequest):                      OpSerializerFor[wire.GetWorkRequest, wire.GetWorkResponse](),
+		byte(wire.ClusterOpGetWorkResponse):                     OpSerializerFor[wire.GetWorkResponse, wire.GetWorkRequest](),
+		byte(wire.ClusterOpSubmitWorkRequest):                   OpSerializerFor[wire.SubmitWorkRequest, wire.SubmitWorkResponse](),
+		byte(wire.ClusterOpSubmitWorkResponse):                  OpSerializerFor[wire.SubmitWorkResponse, wire.SubmitWorkRequest](),
+
+		// §7 Slave → Master (block list)
+		byte(wire.ClusterOpAddMinorBlockHeaderListRequest):  OpSerializerFor[wire.AddMinorBlockHeaderListRequest, wire.AddMinorBlockHeaderListResponse](),
+		byte(wire.ClusterOpAddMinorBlockHeaderListResponse): OpSerializerFor[wire.AddMinorBlockHeaderListResponse, wire.AddMinorBlockHeaderListRequest](),
+
+		// §8 Master → Slave (JRPC & staking)
+		byte(wire.ClusterOpCheckMinorBlockRequest):     OpSerializerFor[wire.CheckMinorBlockRequest, wire.CheckMinorBlockResponse](),
+		byte(wire.ClusterOpCheckMinorBlockResponse):    OpSerializerFor[wire.CheckMinorBlockResponse, wire.CheckMinorBlockRequest](),
+		byte(wire.ClusterOpGetAllTransactionsRequest):  OpSerializerFor[wire.GetAllTransactionsRequest, wire.GetAllTransactionsResponse](),
+		byte(wire.ClusterOpGetAllTransactionsResponse): OpSerializerFor[wire.GetAllTransactionsResponse, wire.GetAllTransactionsRequest](),
+		byte(wire.ClusterOpGetRootChainStakesRequest):  OpSerializerFor[wire.GetRootChainStakesRequest, wire.GetRootChainStakesResponse](),
+		byte(wire.ClusterOpGetRootChainStakesResponse): OpSerializerFor[wire.GetRootChainStakesResponse, wire.GetRootChainStakesRequest](),
+		byte(wire.ClusterOpGetTotalBalanceRequest):     OpSerializerFor[wire.GetTotalBalanceRequest, wire.GetTotalBalanceResponse](),
+		byte(wire.ClusterOpGetTotalBalanceResponse):    OpSerializerFor[wire.GetTotalBalanceResponse, wire.GetTotalBalanceRequest](),
+	})
+}
+
+// registerHandlers registers all master→slave RPC handlers and marks the
+// fire-and-forget opcodes as non-RPC.
+func (mc *MasterConn) registerHandlers() {
+	mc.rpcConn.RegisterTypedHandlers(map[byte]TypedHandler{
+		byte(wire.ClusterOpPing):                                mc.handlePing,
+		byte(wire.ClusterOpConnectToSlavesRequest):              mc.handleConnectToSlaves,
+		byte(wire.ClusterOpMineRequest):                         mc.handleMine,
+		byte(wire.ClusterOpGenTxRequest):                        mc.handleGenTx,
+		byte(wire.ClusterOpAddRootBlockRequest):                 mc.handleAddRootBlock,
+		byte(wire.ClusterOpGetEcoInfoListRequest):               mc.handleGetEcoInfoList,
+		byte(wire.ClusterOpGetNextBlockToMineRequest):           mc.handleGetNextBlockToMine,
+		byte(wire.ClusterOpAddMinorBlockRequest):                mc.handleAddMinorBlock,
+		byte(wire.ClusterOpGetUnconfirmedHeadersRequest):        mc.handleGetUnconfirmedHeaders,
+		byte(wire.ClusterOpGetAccountDataRequest):               mc.handleGetAccountData,
+		byte(wire.ClusterOpAddTransactionRequest):               mc.handleAddTransaction,
+		byte(wire.ClusterOpCreateClusterPeerConnectionRequest):  mc.handleCreateClusterPeerConnection,
+		byte(wire.ClusterOpDestroyClusterPeerConnectionCommand): mc.handleDestroyClusterPeerConnection,
+		byte(wire.ClusterOpGetMinorBlockRequest):                mc.handleGetMinorBlock,
+		byte(wire.ClusterOpGetTransactionRequest):               mc.handleGetTransaction,
+		byte(wire.ClusterOpSyncMinorBlockListRequest):           mc.handleSyncMinorBlockList,
+		byte(wire.ClusterOpExecuteTransactionRequest):           mc.handleExecuteTransaction,
+		byte(wire.ClusterOpGetTransactionReceiptRequest):        mc.handleGetTransactionReceipt,
+		byte(wire.ClusterOpGetTransactionListByAddressRequest):  mc.handleGetTransactionListByAddress,
+		byte(wire.ClusterOpGetLogRequest):                       mc.handleGetLogs,
+		byte(wire.ClusterOpEstimateGasRequest):                  mc.handleEstimateGas,
+		byte(wire.ClusterOpGetStorageRequest):                   mc.handleGetStorageAt,
+		byte(wire.ClusterOpGetCodeRequest):                      mc.handleGetCode,
+		byte(wire.ClusterOpGasPriceRequest):                     mc.handleGasPrice,
+		byte(wire.ClusterOpGetWorkRequest):                      mc.handleGetWork,
+		byte(wire.ClusterOpSubmitWorkRequest):                   mc.handleSubmitWork,
+		byte(wire.ClusterOpCheckMinorBlockRequest):              mc.handleCheckMinorBlock,
+		byte(wire.ClusterOpGetAllTransactionsRequest):           mc.handleGetAllTransactions,
+		byte(wire.ClusterOpGetRootChainStakesRequest):           mc.handleGetRootChainStakes,
+		byte(wire.ClusterOpGetTotalBalanceRequest):              mc.handleGetTotalBalance,
+	})
+
+	mc.rpcConn.RegisterNonRPCOps([]byte{
+		byte(wire.ClusterOpDestroyClusterPeerConnectionCommand),
+	})
+}
+
+// rawBytes is a helper that returns a non-nil *wire.RawBytes pointer.
+func rawBytes(b []byte) *wire.RawBytes {
+	rb := wire.RawBytes(b)
+	return &rb
+}
+
+// emptyRawBytes returns a non-nil *wire.RawBytes pointing to an empty slice.
+func emptyRawBytes() *wire.RawBytes {
+	return rawBytes([]byte{})
+}
+
+// LocalID returns this slave's ID used in PONG responses.
+func (mc *MasterConn) LocalID() []byte {
+	return append([]byte(nil), mc.localID...)
+}
+
+// LocalFullShardIDList returns this slave's full shard ID list used in PONG responses.
+func (mc *MasterConn) LocalFullShardIDList() []uint32 {
+	return append([]uint32(nil), mc.localFullShardIDList...)
+}
+
+// handlePing responds to the master's PING with this slave's identity.
+// Python: MasterConnection.handle_ping -> Pong(self.slave_server.id, ...).
+func (mc *MasterConn) handlePing(req any) (any, error) {
+	// TODO: when core.RootBlock is ported, use ping.root_tip to drive shard creation.
+	_ = req.(*wire.PingRequest)
+
+	return &wire.PongResponse{
+		ID:              append([]byte(nil), mc.localID...),
+		FullShardIDList: append([]uint32(nil), mc.localFullShardIDList...),
+	}, nil
+}
+
+// handleConnectToSlaves accepts a list of slaves to connect to.
+// Python: returns ConnectToSlavesResponse with one empty bytes result per slave.
+func (mc *MasterConn) handleConnectToSlaves(req any) (any, error) {
+	r := req.(*wire.ConnectToSlavesRequest)
+
+	// TODO: delegate to SlaveServer.slave_connection_manager.connect_to_slave.
+	resultList := make([]wire.PrependedSizeBytes4, len(r.SlaveInfoList))
+	for i := range resultList {
+		resultList[i] = wire.PrependedSizeBytes4{}
+	}
+	return &wire.ConnectToSlavesResponse{ResultList: resultList}, nil
+}
+
+// handleMine starts or stops mining.
+// Python: MineResponse(error_code=0).
+func (mc *MasterConn) handleMine(req any) (any, error) {
+	_ = req.(*wire.MineRequest)
+	// TODO: delegate to SlaveServer.start_mining / stop_mining.
+	return &wire.MineResponse{ErrorCode: 0}, nil
+}
+
+// handleGenTx generates transactions.
+// Python: GenTxResponse(error_code=0).
+func (mc *MasterConn) handleGenTx(req any) (any, error) {
+	_ = req.(*wire.GenTxRequest)
+	// TODO: delegate to SlaveServer.create_transactions.
+	return &wire.GenTxResponse{ErrorCode: 0}, nil
+}
+
+// handleAddRootBlock processes a root block from the master.
+// Python: returns AddRootBlockResponse(error_code=0, switched=False) on success.
+func (mc *MasterConn) handleAddRootBlock(req any) (any, error) {
+	_ = req.(*wire.AddRootBlockRequest)
+	// TODO: delegate to shard.add_root_block and SlaveServer.create_shards.
+	return &wire.AddRootBlockResponse{ErrorCode: 0, Switched: false}, nil
+}
+
+// handleGetEcoInfoList returns economic info for all initialized shards.
+// Python: returns empty list when no shards are initialized.
+func (mc *MasterConn) handleGetEcoInfoList(req any) (any, error) {
+	_ = req.(*wire.GetEcoInfoListRequest)
+	// TODO: collect real EcoInfo from shard states.
+	return &wire.GetEcoInfoListResponse{ErrorCode: 0, EcoInfoList: []wire.EcoInfo{}}, nil
+}
+
+// handleGetNextBlockToMine returns a block template for the requested branch.
+// Python requires the shard to exist; without shard runtime we return not-found.
+func (mc *MasterConn) handleGetNextBlockToMine(req any) (any, error) {
+	_ = req.(*wire.GetNextBlockToMineRequest)
+	// TODO: delegate to shard.state.create_block_to_mine.
+	return &wire.GetNextBlockToMineResponse{ErrorCode: 1, Block: emptyRawBytes()}, nil
+}
+
+// handleAddMinorBlock adds a JRPC-mined minor block.
+// Python: returns AddMinorBlockResponse(error_code=0) on success.
+func (mc *MasterConn) handleAddMinorBlock(req any) (any, error) {
+	_ = req.(*wire.AddMinorBlockRequest)
+	// TODO: deserialize MinorBlock and delegate to shard.add_block.
+	return &wire.AddMinorBlockResponse{ErrorCode: 0}, nil
+}
+
+// handleGetUnconfirmedHeaders returns unconfirmed headers per shard.
+// Python: returns empty list when no shards are initialized.
+func (mc *MasterConn) handleGetUnconfirmedHeaders(req any) (any, error) {
+	_ = req.(*wire.GetUnconfirmedHeadersRequest)
+	// TODO: collect real HeadersInfo from shard states.
+	return &wire.GetUnconfirmedHeadersResponse{ErrorCode: 0, HeadersInfoList: []wire.HeadersInfo{}}, nil
+}
+
+// handleGetAccountData returns account data across shards.
+// Python: returns empty list when there are no shards for the address.
+func (mc *MasterConn) handleGetAccountData(req any) (any, error) {
+	_ = req.(*wire.GetAccountDataRequest)
+	// TODO: delegate to SlaveServer.get_account_data.
+	return &wire.GetAccountDataResponse{ErrorCode: 0, AccountBranchDataList: []wire.AccountBranchData{}}, nil
+}
+
+// handleAddTransaction adds a transaction to the tx pool.
+// Python: returns AddTransactionResponse(error_code=0) on success.
+func (mc *MasterConn) handleAddTransaction(req any) (any, error) {
+	_ = req.(*wire.AddTransactionRequest)
+	// TODO: delegate to SlaveServer.add_tx.
+	return &wire.AddTransactionResponse{ErrorCode: 0}, nil
+}
+
+// handleCreateClusterPeerConnection creates virtual peer connections for all shards.
+// Python: returns CreateClusterPeerConnectionResponse(error_code=0) on success.
+func (mc *MasterConn) handleCreateClusterPeerConnection(req any) (any, error) {
+	_ = req.(*wire.CreateClusterPeerConnectionRequest)
+	// TODO: create PeerShardConnection instances and wire with the dispatcher (PR6).
+	return &wire.CreateClusterPeerConnectionResponse{ErrorCode: 0}, nil
+}
+
+// handleDestroyClusterPeerConnection is a fire-and-forget command to tear down
+// a virtual peer connection. No response is sent.
+func (mc *MasterConn) handleDestroyClusterPeerConnection(req any) (any, error) {
+	_ = req.(*wire.DestroyClusterPeerConnectionCommand)
+	// TODO: notify dispatcher / close peer shard connections (PR6).
+	return nil, nil
+}
+
+// handleGetMinorBlock fetches a minor block by hash or height.
+// Python returns error_code=1 with an empty block when not found.
+func (mc *MasterConn) handleGetMinorBlock(req any) (any, error) {
+	_ = req.(*wire.GetMinorBlockRequest)
+	// TODO: delegate to SlaveServer.get_minor_block_by_hash / by_height.
+	return &wire.GetMinorBlockResponse{
+		ErrorCode:  1,
+		MinorBlock: emptyRawBytes(),
+		ExtraInfo:  nil,
+	}, nil
+}
+
+// handleGetTransaction fetches a transaction by hash.
+// Python returns error_code=1 with an empty block when not found.
+func (mc *MasterConn) handleGetTransaction(req any) (any, error) {
+	_ = req.(*wire.GetTransactionRequest)
+	// TODO: delegate to SlaveServer.get_transaction_by_hash.
+	return &wire.GetTransactionResponse{
+		ErrorCode:  1,
+		MinorBlock: emptyRawBytes(),
+		Index:      0,
+	}, nil
+}
+
+// handleSyncMinorBlockList downloads and applies a list of minor blocks.
+// Python returns error_code=0 with empty data when the input list is empty.
+func (mc *MasterConn) handleSyncMinorBlockList(req any) (any, error) {
+	r := req.(*wire.SyncMinorBlockListRequest)
+	_ = r
+	// TODO: delegate to SlaveServer.add_block_list_for_sync.
+	return &wire.SyncMinorBlockListResponse{
+		ErrorCode:        0,
+		BlockCoinbaseMap: emptyRawBytes(),
+		ShardStats:       nil,
+	}, nil
+}
+
+// handleExecuteTransaction executes a transaction and returns the result.
+// Python returns error_code=1 when execution fails (e.g. shard missing).
+func (mc *MasterConn) handleExecuteTransaction(req any) (any, error) {
+	_ = req.(*wire.ExecuteTransactionRequest)
+	// TODO: delegate to SlaveServer.execute_tx.
+	return &wire.ExecuteTransactionResponse{ErrorCode: 1, Result: []byte{}}, nil
+}
+
+// handleGetTransactionReceipt fetches a transaction receipt.
+// Python returns error_code=1 with empty block/receipt when not found.
+func (mc *MasterConn) handleGetTransactionReceipt(req any) (any, error) {
+	_ = req.(*wire.GetTransactionReceiptRequest)
+	// TODO: delegate to SlaveServer.get_transaction_receipt.
+	return &wire.GetTransactionReceiptResponse{
+		ErrorCode:  1,
+		MinorBlock: emptyRawBytes(),
+		Index:      0,
+		Receipt:    emptyRawBytes(),
+	}, nil
+}
+
+// handleGetTransactionListByAddress returns transactions for an address.
+// Python returns error_code=1 with empty lists when the shard is missing.
+func (mc *MasterConn) handleGetTransactionListByAddress(req any) (any, error) {
+	_ = req.(*wire.GetTransactionListByAddressRequest)
+	// TODO: delegate to SlaveServer.get_transaction_list_by_address.
+	return &wire.GetTransactionListByAddressResponse{
+		ErrorCode: 1,
+		TxList:    []wire.TransactionDetail{},
+		Next:      []byte{},
+	}, nil
+}
+
+// handleGetLogs returns logs matching the filter.
+// Python returns error_code=1 with empty logs when the shard is missing.
+func (mc *MasterConn) handleGetLogs(req any) (any, error) {
+	_ = req.(*wire.GetLogRequest)
+	// TODO: delegate to SlaveServer.get_logs.
+	return &wire.GetLogResponse{ErrorCode: 1, Logs: []*wire.RawBytes{}}, nil
+}
+
+// handleEstimateGas estimates gas for a transaction.
+// Python returns error_code=1 when estimation fails (e.g. shard missing).
+func (mc *MasterConn) handleEstimateGas(req any) (any, error) {
+	_ = req.(*wire.EstimateGasRequest)
+	// TODO: delegate to SlaveServer.estimate_gas.
+	return &wire.EstimateGasResponse{ErrorCode: 1, Result: 0}, nil
+}
+
+// handleGetStorageAt reads storage at the given address/key.
+// Python returns error_code=1 with a zero result when the shard is missing.
+func (mc *MasterConn) handleGetStorageAt(req any) (any, error) {
+	_ = req.(*wire.GetStorageRequest)
+	// TODO: delegate to SlaveServer.get_storage_at.
+	return &wire.GetStorageResponse{ErrorCode: 1, Result: [wire.HashLength]byte{}}, nil
+}
+
+// handleGetCode reads code at the given address.
+// Python returns error_code=1 with empty bytes when the shard is missing.
+func (mc *MasterConn) handleGetCode(req any) (any, error) {
+	_ = req.(*wire.GetCodeRequest)
+	// TODO: delegate to SlaveServer.get_code.
+	return &wire.GetCodeResponse{ErrorCode: 1, Result: []byte{}}, nil
+}
+
+// handleGasPrice returns the gas price for a token on a branch.
+// Python returns error_code=1 with result 0 when the shard is missing.
+func (mc *MasterConn) handleGasPrice(req any) (any, error) {
+	_ = req.(*wire.GasPriceRequest)
+	// TODO: delegate to SlaveServer.gas_price.
+	return &wire.GasPriceResponse{ErrorCode: 1, Result: 0}, nil
+}
+
+// handleGetWork returns mining work.
+// Python returns error_code=1 when work cannot be produced.
+func (mc *MasterConn) handleGetWork(req any) (any, error) {
+	_ = req.(*wire.GetWorkRequest)
+	// TODO: delegate to SlaveServer.get_work.
+	return &wire.GetWorkResponse{ErrorCode: 1}, nil
+}
+
+// handleSubmitWork submits mining work.
+// Python returns error_code=1, success=False when submission fails.
+func (mc *MasterConn) handleSubmitWork(req any) (any, error) {
+	_ = req.(*wire.SubmitWorkRequest)
+	// TODO: delegate to SlaveServer.submit_work.
+	return &wire.SubmitWorkResponse{ErrorCode: 1, Success: false}, nil
+}
+
+// handleCheckMinorBlock validates a minor block header.
+// Python returns CheckMinorBlockResponse(error_code=0) when the block is valid,
+// and error_code=errno.EBADMSG when the shard is missing or validation fails.
+// This stub returns ErrorCode=1 to signal "not implemented / cannot validate".
+func (mc *MasterConn) handleCheckMinorBlock(req any) (any, error) {
+	_ = req.(*wire.CheckMinorBlockRequest)
+	// TODO: delegate to shard.check_minor_block_by_header.
+	return &wire.CheckMinorBlockResponse{ErrorCode: 1}, nil
+}
+
+// handleGetAllTransactions returns all transactions in the mempool.
+// Python returns error_code=1 with empty lists when the shard is missing.
+func (mc *MasterConn) handleGetAllTransactions(req any) (any, error) {
+	_ = req.(*wire.GetAllTransactionsRequest)
+	// TODO: delegate to SlaveServer.get_all_transactions.
+	return &wire.GetAllTransactionsResponse{
+		ErrorCode: 1,
+		TxList:    []wire.TransactionDetail{},
+		Next:      []byte{},
+	}, nil
+}
+
+// handleGetRootChainStakes reads root-chain stake info.
+// Python returns GetRootChainStakesResponse(0, stakes, signer).
+func (mc *MasterConn) handleGetRootChainStakes(req any) (any, error) {
+	_ = req.(*wire.GetRootChainStakesRequest)
+	// TODO: delegate to SlaveServer.get_root_chain_stakes.
+	return &wire.GetRootChainStakesResponse{
+		ErrorCode: 0,
+		Stakes:    serialize.BigUint{},
+		Signer:    [20]byte{},
+	}, nil
+}
+
+// handleGetTotalBalance returns the total token balance across accounts.
+// Python catches exceptions and returns GetTotalBalanceResponse(1, 0, b"").
+func (mc *MasterConn) handleGetTotalBalance(req any) (any, error) {
+	_ = req.(*wire.GetTotalBalanceRequest)
+	// TODO: delegate to SlaveServer.get_total_balance.
+	return &wire.GetTotalBalanceResponse{
+		ErrorCode:    1,
+		TotalBalance: serialize.BigUint{},
+		Next:         []byte{},
+	}, nil
+}
+
+// SetForwarder installs a raw-frame forwarder hook for peer traffic
+// (cluster_peer_id != 0). This is used by the dispatcher in PR6.
+func (mc *MasterConn) SetForwarder(f func(*wire.Frame) bool) {
+	mc.rpcConn.SetForwarder(f)
+}
+
+// SendRPCMeta sends a request with ClusterMetadata and waits for the response.
+// It is the primitive used by all typed outbound methods.
+func (mc *MasterConn) SendRPCMeta(ctx context.Context, opcode byte, payload []byte, meta wire.ClusterMetadata) (*wire.Frame, error) {
+	return mc.rpcConn.SendRPCMeta(ctx, opcode, payload, meta)
+}
+
+// SendAddMinorBlockHeader sends AddMinorBlockHeaderRequest to the master and
+// returns the parsed response.
+func (mc *MasterConn) SendAddMinorBlockHeader(ctx context.Context, req *wire.AddMinorBlockHeaderRequest) (*wire.AddMinorBlockHeaderResponse, error) {
+	payload, err := serializeBytes(req)
+	if err != nil {
+		return nil, fmt.Errorf("serialize AddMinorBlockHeaderRequest: %w", err)
+	}
+	frame, err := mc.SendRPCMeta(ctx, byte(wire.ClusterOpAddMinorBlockHeaderRequest), payload, wire.ClusterMetadata{})
+	if err != nil {
+		return nil, err
+	}
+	var resp wire.AddMinorBlockHeaderResponse
+	if err := deserializeBytes(frame.Payload, &resp); err != nil {
+		return nil, fmt.Errorf("deserialize AddMinorBlockHeaderResponse: %w", err)
+	}
+	return &resp, nil
+}
+
+// SendAddMinorBlockHeaderList sends AddMinorBlockHeaderListRequest to the master
+// and returns the parsed response.
+func (mc *MasterConn) SendAddMinorBlockHeaderList(ctx context.Context, req *wire.AddMinorBlockHeaderListRequest) (*wire.AddMinorBlockHeaderListResponse, error) {
+	payload, err := serializeBytes(req)
+	if err != nil {
+		return nil, fmt.Errorf("serialize AddMinorBlockHeaderListRequest: %w", err)
+	}
+	frame, err := mc.SendRPCMeta(ctx, byte(wire.ClusterOpAddMinorBlockHeaderListRequest), payload, wire.ClusterMetadata{})
+	if err != nil {
+		return nil, err
+	}
+	var resp wire.AddMinorBlockHeaderListResponse
+	if err := deserializeBytes(frame.Payload, &resp); err != nil {
+		return nil, fmt.Errorf("deserialize AddMinorBlockHeaderListResponse: %w", err)
+	}
+	return &resp, nil
+}

qkc/cluster/slave/master_conn_test.go

@@ -0,0 +1,854 @@
+// Copyright 2026-2027, QuarkChain.
+
+package slave
+
+import (
+	"bytes"
+	"context"
+	"encoding/binary"
+	"net"
+	"reflect"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/ethereum/go-ethereum/log"
+	"github.com/ethereum/go-ethereum/qkc/account"
+	"github.com/ethereum/go-ethereum/qkc/cluster/wire"
+	"github.com/ethereum/go-ethereum/qkc/serialize"
+)
+
+// newMasterTestConnPair creates a pair of MasterConns connected over a local TCP
+// socket. The caller is responsible for calling cleanup.
+func newMasterTestConnPair(t *testing.T) (client, server *MasterConn, cleanup func()) {
+	t.Helper()
+	return newMasterTestConnPairWithIdentity(
+		t,
+		[]byte("go-slave-client"), []uint32{0x00010001},
+		[]byte("go-slave-server"), []uint32{0x00010001, 0x00020001},
+	)
+}
+
+func newMasterTestConnPairWithIdentity(
+	t *testing.T,
+	clientID []byte, clientShards []uint32,
+	serverID []byte, serverShards []uint32,
+) (client, server *MasterConn, cleanup func()) {
+	t.Helper()
+
+	ln, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatalf("listen: %v", err)
+	}
+
+	var serverConn net.Conn
+	var acceptErr error
+	accepted := make(chan struct{})
+	go func() {
+		defer close(accepted)
+		serverConn, acceptErr = ln.Accept()
+		ln.Close()
+	}()
+
+	clientConn, err := net.Dial("tcp", ln.Addr().String())
+	if err != nil {
+		t.Fatalf("dial: %v", err)
+	}
+	<-accepted
+	if acceptErr != nil {
+		t.Fatalf("accept: %v", acceptErr)
+	}
+
+	logger := log.New()
+	client = NewMasterConnFromConn(clientConn, 0, clientID, clientShards, logger)
+	server = NewMasterConnFromConn(serverConn, 0, serverID, serverShards, logger)
+	cleanup = func() {
+		client.Close()
+		server.Close()
+	}
+	return
+}
+
+// writeRawMasterFrame writes a raw ClusterMetadata frame directly to the
+// underlying TCP connection, bypassing the connection's frame writer.
+func writeRawMasterFrame(t *testing.T, conn net.Conn, frame *wire.Frame) {
+	t.Helper()
+	if err := wire.WriteFrame(conn, frame); err != nil {
+		t.Fatalf("write raw frame: %v", err)
+	}
+}
+
+// hasHandler reports whether the connection has a typed handler for opcode.
+func hasHandler(c *MasterConn, opcode byte) bool {
+	rv := reflect.ValueOf(c.rpcConn).Elem()
+	handlers := rv.FieldByName("typedHandlers").MapKeys()
+	for _, k := range handlers {
+		if k.Uint() == uint64(opcode) {
+			return true
+		}
+	}
+	return false
+}
+
+// hasSerializer reports whether the connection has an OpSerializer for opcode.
+func hasSerializer(c *MasterConn, opcode byte) bool {
+	rv := reflect.ValueOf(c.rpcConn).Elem()
+	serializers := rv.FieldByName("serializers").MapKeys()
+	for _, k := range serializers {
+		if k.Uint() == uint64(opcode) {
+			return true
+		}
+	}
+	return false
+}
+
+// TestMasterConn_AllMasterHandlersRegistered verifies that every master→slave
+// request opcode has a handler registered and that the fire-and-forget opcode
+// is marked as non-RPC.
+func TestMasterConn_AllMasterHandlersRegistered(t *testing.T) {
+	_, server, cleanup := newMasterTestConnPair(t)
+	defer cleanup()
+
+	masterRPCOps := []wire.ClusterOp{
+		wire.ClusterOpPing,
+		wire.ClusterOpConnectToSlavesRequest,
+		wire.ClusterOpMineRequest,
+		wire.ClusterOpGenTxRequest,
+		wire.ClusterOpAddRootBlockRequest,
+		wire.ClusterOpGetEcoInfoListRequest,
+		wire.ClusterOpGetNextBlockToMineRequest,
+		wire.ClusterOpAddMinorBlockRequest,
+		wire.ClusterOpGetUnconfirmedHeadersRequest,
+		wire.ClusterOpGetAccountDataRequest,
+		wire.ClusterOpAddTransactionRequest,
+		wire.ClusterOpCreateClusterPeerConnectionRequest,
+		wire.ClusterOpGetMinorBlockRequest,
+		wire.ClusterOpGetTransactionRequest,
+		wire.ClusterOpSyncMinorBlockListRequest,
+		wire.ClusterOpExecuteTransactionRequest,
+		wire.ClusterOpGetTransactionReceiptRequest,
+		wire.ClusterOpGetTransactionListByAddressRequest,
+		wire.ClusterOpGetLogRequest,
+		wire.ClusterOpEstimateGasRequest,
+		wire.ClusterOpGetStorageRequest,
+		wire.ClusterOpGetCodeRequest,
+		wire.ClusterOpGasPriceRequest,
+		wire.ClusterOpGetWorkRequest,
+		wire.ClusterOpSubmitWorkRequest,
+		wire.ClusterOpCheckMinorBlockRequest,
+		wire.ClusterOpGetAllTransactionsRequest,
+		wire.ClusterOpGetRootChainStakesRequest,
+		wire.ClusterOpGetTotalBalanceRequest,
+	}
+
+	for _, op := range masterRPCOps {
+		if !hasHandler(server, byte(op)) {
+			t.Fatalf("missing handler for opcode 0x%02x (%v)", op, op)
+		}
+	}
+
+	if !isNonRPC(server, byte(wire.ClusterOpDestroyClusterPeerConnectionCommand)) {
+		t.Fatalf("DESTROY_CLUSTER_PEER_CONNECTION_COMMAND is not marked as non-RPC")
+	}
+}
+
+// isNonRPC reports whether opcode is registered as fire-and-forget.
+func isNonRPC(c *MasterConn, opcode byte) bool {
+	rv := reflect.ValueOf(c.rpcConn).Elem()
+	nonRPCOps := rv.FieldByName("nonRPCOps").MapKeys()
+	for _, k := range nonRPCOps {
+		if k.Uint() == uint64(opcode) {
+			return true
+		}
+	}
+	return false
+}
+
+// TestMasterConn_AllSerializersRegistered verifies that every ClusterOp defined
+// in wire/opcode.go has a registered OpSerializer.
+func TestMasterConn_AllSerializersRegistered(t *testing.T) {
+	_, server, cleanup := newMasterTestConnPair(t)
+	defer cleanup()
+
+	for op := wire.ClusterOpPing; op <= wire.ClusterOpGetTotalBalanceResponse; op++ {
+		if op == 0x9C { // 28 is intentionally skipped in Python
+			continue
+		}
+		if !hasSerializer(server, byte(op)) {
+			t.Fatalf("missing serializer for opcode 0x%02x (%v)", op, op)
+		}
+	}
+}
+
+// TestMasterConn_Ping verifies the master→slave PING handshake.
+func TestMasterConn_Ping(t *testing.T) {
+	clientID := []byte("go-slave-client")
+	clientShards := []uint32{0x00010001}
+	serverID := []byte("go-slave-server")
+	serverShards := []uint32{0x00010001, 0x00020001}
+
+	client, server, cleanup := newMasterTestConnPairWithIdentity(t, clientID, clientShards, serverID, serverShards)
+	defer cleanup()
+
+	server.Start()
+	client.Start()
+
+	pingPayload, err := serialize.SerializeToBytes(&wire.PingRequest{
+		ID:              []byte("master"),
+		FullShardIDList: []uint32{0x00010001},
+		RootTip:         nil,
+	})
+	if err != nil {
+		t.Fatalf("serialize ping: %v", err)
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	resp, err := client.SendRPCMeta(ctx, byte(wire.ClusterOpPing), pingPayload, wire.ClusterMetadata{})
+	if err != nil {
+		t.Fatalf("send ping: %v", err)
+	}
+	if resp.Opcode != byte(wire.ClusterOpPong) {
+		t.Fatalf("expected opcode 0x%x, got 0x%x", wire.ClusterOpPong, resp.Opcode)
+	}
+
+	var pong wire.PongResponse
+	if err := serialize.Deserialize(serialize.NewByteBuffer(resp.Payload), &pong); err != nil {
+		t.Fatalf("deserialize pong: %v", err)
+	}
+	if string(pong.ID) != string(serverID) {
+		t.Fatalf("pong id mismatch: got %s, expected %s", pong.ID, serverID)
+	}
+	if len(pong.FullShardIDList) != len(serverShards) {
+		t.Fatalf("pong shard list mismatch: got %v", pong.FullShardIDList)
+	}
+}
+
+// TestMasterConn_RPCRoundTrip verifies request/response dispatch for a
+// representative set of master→slave RPCs.
+func TestMasterConn_RPCRoundTrip(t *testing.T) {
+	client, server, cleanup := newMasterTestConnPair(t)
+	defer cleanup()
+
+	server.Start()
+	client.Start()
+
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	cases := []struct {
+		name       string
+		opcode     wire.ClusterOp
+		req        any
+		resp       any
+		respOpcode wire.ClusterOp
+	}{
+		{
+			name:       "add_root_block",
+			opcode:     wire.ClusterOpAddRootBlockRequest,
+			req:        &wire.AddRootBlockRequest{RootBlock: emptyRawBytes(), ExpectSwitch: false},
+			resp:       &wire.AddRootBlockResponse{},
+			respOpcode: wire.ClusterOpAddRootBlockResponse,
+		},
+		{
+			name:       "get_eco_info_list",
+			opcode:     wire.ClusterOpGetEcoInfoListRequest,
+			req:        &wire.GetEcoInfoListRequest{},
+			resp:       &wire.GetEcoInfoListResponse{},
+			respOpcode: wire.ClusterOpGetEcoInfoListResponse,
+		},
+		{
+			name:       "add_transaction",
+			opcode:     wire.ClusterOpAddTransactionRequest,
+			req:        &wire.AddTransactionRequest{Tx: emptyRawBytes()},
+			resp:       &wire.AddTransactionResponse{},
+			respOpcode: wire.ClusterOpAddTransactionResponse,
+		},
+		{
+			name:       "get_minor_block",
+			opcode:     wire.ClusterOpGetMinorBlockRequest,
+			req:        &wire.GetMinorBlockRequest{Branch: 0x00010001, Height: 1, NeedExtraInfo: false},
+			resp:       &wire.GetMinorBlockResponse{},
+			respOpcode: wire.ClusterOpGetMinorBlockResponse,
+		},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			payload, err := serialize.SerializeToBytes(tc.req)
+			if err != nil {
+				t.Fatalf("serialize request: %v", err)
+			}
+
+			frame, err := client.SendRPCMeta(ctx, byte(tc.opcode), payload, wire.ClusterMetadata{Branch: 0x00010001})
+			if err != nil {
+				t.Fatalf("send rpc: %v", err)
+			}
+			if frame.Opcode != byte(tc.respOpcode) {
+				t.Fatalf("expected response opcode 0x%x, got 0x%x", tc.respOpcode, frame.Opcode)
+			}
+			if frame.Meta.Branch != 0x00010001 {
+				t.Fatalf("metadata branch not preserved: got %d", frame.Meta.Branch)
+			}
+
+			if err := serialize.Deserialize(serialize.NewByteBuffer(frame.Payload), tc.resp); err != nil {
+				t.Fatalf("deserialize response: %v", err)
+			}
+		})
+	}
+}
+
+// TestMasterConn_NonRPCDispatch verifies that the fire-and-forget
+// DESTROY_CLUSTER_PEER_CONNECTION_COMMAND is accepted with rpc_id == 0 and does
+// not produce a response or close the connection.
+func TestMasterConn_NonRPCDispatch(t *testing.T) {
+	client, server, cleanup := newMasterTestConnPair(t)
+	defer cleanup()
+
+	server.Start()
+	client.Start()
+
+	payload, err := serialize.SerializeToBytes(&wire.DestroyClusterPeerConnectionCommand{ClusterPeerID: 42})
+	if err != nil {
+		t.Fatalf("serialize command: %v", err)
+	}
+
+	// Write a non-RPC frame directly; no response should come back, but the
+	// connection must remain usable for a subsequent RPC.
+	writeRawMasterFrame(t, client.conn, &wire.Frame{
+		Meta:    wire.ClusterMetadata{Branch: 0x00010001},
+		Opcode:  byte(wire.ClusterOpDestroyClusterPeerConnectionCommand),
+		RPCID:   0,
+		Payload: payload,
+	})
+
+	// Give the server a moment to process the command.
+	time.Sleep(50 * time.Millisecond)
+
+	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
+	defer cancel()
+
+	pingPayload, _ := serialize.SerializeToBytes(&wire.PingRequest{
+		ID:              []byte("master"),
+		FullShardIDList: []uint32{0x00010001},
+	})
+	resp, err := client.SendRPCMeta(ctx, byte(wire.ClusterOpPing), pingPayload, wire.ClusterMetadata{})
+	if err != nil {
+		t.Fatalf("ping after non-rpc command failed: %v", err)
+	}
+	if resp.Opcode != byte(wire.ClusterOpPong) {
+		t.Fatalf("expected pong, got opcode 0x%x", resp.Opcode)
+	}
+}
+
+// TestMasterConn_NonRPCWithNonZeroRPCID verifies that a non-RPC command with a
+// non-zero rpc_id causes the server to close the connection.
+func TestMasterConn_NonRPCWithNonZeroRPCID(t *testing.T) {
+	client, server, cleanup := newMasterTestConnPair(t)
+	defer cleanup()
+
+	server.Start()
+	client.Start()
+
+	payload, _ := serialize.SerializeToBytes(&wire.DestroyClusterPeerConnectionCommand{ClusterPeerID: 42})
+
+	writeRawMasterFrame(t, client.conn, &wire.Frame{
+		Meta:    wire.ClusterMetadata{Branch: 0x00010001},
+		Opcode:  byte(wire.ClusterOpDestroyClusterPeerConnectionCommand),
+		RPCID:   1, // non-RPC must have rpc_id == 0
+		Payload: payload,
+	})
+
+	select {
+	case <-server.WaitUntilClosed():
+	case <-time.After(2 * time.Second):
+		t.Fatal("server did not close after non-rpc with non-zero rpc_id")
+	}
+}
+
+// TestMasterConn_Forwarder verifies that frames with cluster_peer_id != 0 are
+// routed through the forwarder hook and are not dispatched locally.
+func TestMasterConn_Forwarder(t *testing.T) {
+	client, server, cleanup := newMasterTestConnPair(t)
+	defer cleanup()
+
+	var forwardedMu sync.Mutex
+	var forwarded []*wire.Frame
+	server.SetForwarder(func(frame *wire.Frame) bool {
+		if frame.Meta.ClusterPeerID == 0 {
+			return false
+		}
+		forwardedMu.Lock()
+		forwarded = append(forwarded, frame)
+		forwardedMu.Unlock()
+		return true
+	})
+
+	server.Start()
+	client.Start()
+
+	payload, _ := serialize.SerializeToBytes(&wire.GetMinorBlockRequest{Branch: 0x00010001, Height: 1})
+
+	// Peer-originated frame: cluster_peer_id != 0.
+	writeRawMasterFrame(t, client.conn, &wire.Frame{
+		Meta:    wire.ClusterMetadata{Branch: 0x00010001, ClusterPeerID: 123},
+		Opcode:  byte(wire.ClusterOpGetMinorBlockRequest),
+		RPCID:   7,
+		Payload: payload,
+	})
+
+	time.Sleep(100 * time.Millisecond)
+
+	forwardedMu.Lock()
+	count := len(forwarded)
+	forwardedMu.Unlock()
+	if count != 1 {
+		t.Fatalf("expected 1 forwarded frame, got %d", count)
+	}
+
+	// Connection should still be open; a subsequent master RPC works.
+	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
+	defer cancel()
+
+	pingPayload, _ := serialize.SerializeToBytes(&wire.PingRequest{ID: []byte("m"), FullShardIDList: []uint32{1}})
+	resp, err := client.SendRPCMeta(ctx, byte(wire.ClusterOpPing), pingPayload, wire.ClusterMetadata{})
+	if err != nil {
+		t.Fatalf("ping after forwarded frame failed: %v", err)
+	}
+	if resp.Opcode != byte(wire.ClusterOpPong) {
+		t.Fatalf("expected pong, got 0x%x", resp.Opcode)
+	}
+}
+
+// TestMasterConn_UnsupportedOpcodeClosesConnection verifies that an opcode
+// without a registered handler and rpc_id == 0 causes the server to close the
+// connection.
+func TestMasterConn_UnsupportedOpcodeClosesConnection(t *testing.T) {
+	client, server, cleanup := newMasterTestConnPair(t)
+	defer cleanup()
+
+	server.Start()
+	client.Start()
+
+	// 0x01 is a CommandOp with no handler registered on the master side.
+	// rpc_id must be 0 for the server to treat it as a non-RPC unsupported
+	// command and close the connection.
+	writeRawMasterFrame(t, client.conn, &wire.Frame{
+		Meta:    wire.ClusterMetadata{},
+		Opcode:  0x01,
+		RPCID:   0,
+		Payload: []byte("payload"),
+	})
+
+	select {
+	case <-server.WaitUntilClosed():
+	case <-time.After(2 * time.Second):
+		t.Fatal("server did not close after unsupported opcode")
+	}
+}
+
+// TestMasterConn_RPCIDMonotonic verifies that duplicate RPC IDs cause the
+// server to close the connection.
+func TestMasterConn_RPCIDMonotonic(t *testing.T) {
+	client, server, cleanup := newMasterTestConnPair(t)
+	defer cleanup()
+
+	server.Start()
+	client.Start()
+
+	pingPayload, _ := serialize.SerializeToBytes(&wire.PingRequest{
+		ID:              []byte("master"),
+		FullShardIDList: []uint32{0x00010001},
+	})
+
+	// Manually send two PING frames with the same RPC ID.
+	writeRawMasterFrame(t, client.conn, &wire.Frame{
+		Meta:    wire.ClusterMetadata{},
+		Opcode:  byte(wire.ClusterOpPing),
+		RPCID:   1,
+		Payload: pingPayload,
+	})
+	writeRawMasterFrame(t, client.conn, &wire.Frame{
+		Meta:    wire.ClusterMetadata{},
+		Opcode:  byte(wire.ClusterOpPing),
+		RPCID:   1, // duplicate
+		Payload: pingPayload,
+	})
+
+	select {
+	case <-server.WaitUntilClosed():
+	case <-time.After(2 * time.Second):
+		t.Fatal("server did not close after duplicate rpc_id")
+	}
+}
+
+// TestMasterConn_RPCIDDecreasing verifies that a decreasing RPC ID causes the
+// server to close the connection.
+func TestMasterConn_RPCIDDecreasing(t *testing.T) {
+	client, server, cleanup := newMasterTestConnPair(t)
+	defer cleanup()
+
+	server.Start()
+	client.Start()
+
+	pingPayload, _ := serialize.SerializeToBytes(&wire.PingRequest{
+		ID:              []byte("master"),
+		FullShardIDList: []uint32{0x00010001},
+	})
+
+	writeRawMasterFrame(t, client.conn, &wire.Frame{
+		Meta:    wire.ClusterMetadata{},
+		Opcode:  byte(wire.ClusterOpPing),
+		RPCID:   2,
+		Payload: pingPayload,
+	})
+	writeRawMasterFrame(t, client.conn, &wire.Frame{
+		Meta:    wire.ClusterMetadata{},
+		Opcode:  byte(wire.ClusterOpPing),
+		RPCID:   1, // decreasing
+		Payload: pingPayload,
+	})
+
+	select {
+	case <-server.WaitUntilClosed():
+	case <-time.After(2 * time.Second):
+		t.Fatal("server did not close after decreasing rpc_id")
+	}
+}
+
+// TestMasterConn_CloseWakesPendingRPC verifies that Close wakes all pending
+// outbound RPCs with ErrConnectionClosed.
+func TestMasterConn_CloseWakesPendingRPC(t *testing.T) {
+	client, _, cleanup := newMasterTestConnPair(t)
+	defer cleanup()
+
+	// Server intentionally left unstarted so it never replies.
+	client.Start()
+
+	var wg sync.WaitGroup
+	wg.Add(1)
+	errChan := make(chan error, 1)
+	go func() {
+		wg.Done()
+		_, err := client.SendRPCMeta(context.Background(), byte(wire.ClusterOpPing), []byte("ping"), wire.ClusterMetadata{})
+		errChan <- err
+	}()
+
+	wg.Wait()
+	client.Close()
+
+	select {
+	case err := <-errChan:
+		if err != ErrConnectionClosed {
+			t.Fatalf("expected ErrConnectionClosed, got %v", err)
+		}
+	case <-time.After(2 * time.Second):
+		t.Fatal("pending RPC was not woken by Close")
+	}
+}
+
+// TestMasterConn_OutboundRPCMeta verifies that outbound RPCs from the slave
+// encode ClusterMetadata correctly on the wire.
+func TestMasterConn_OutboundRPCMeta(t *testing.T) {
+	client, server, cleanup := newMasterTestConnPair(t)
+	defer cleanup()
+
+	// Server echoes the request opcode + 1 and preserves metadata.
+	server.RegisterTypedHandlers(map[byte]TypedHandler{
+		byte(wire.ClusterOpGetEcoInfoListRequest): func(req any) (any, error) {
+			_ = req.(*wire.GetEcoInfoListRequest)
+			return &wire.GetEcoInfoListResponse{ErrorCode: 0}, nil
+		},
+	})
+
+	server.Start()
+	client.Start()
+
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	meta := wire.ClusterMetadata{Branch: 0x00010001, ClusterPeerID: 99}
+	payload, _ := serialize.SerializeToBytes(&wire.GetEcoInfoListRequest{})
+	resp, err := client.SendRPCMeta(ctx, byte(wire.ClusterOpGetEcoInfoListRequest), payload, meta)
+	if err != nil {
+		t.Fatalf("send rpc: %v", err)
+	}
+	if resp.Opcode != byte(wire.ClusterOpGetEcoInfoListResponse) {
+		t.Fatalf("unexpected response opcode 0x%x", resp.Opcode)
+	}
+	if resp.Meta.Branch != meta.Branch || resp.Meta.ClusterPeerID != meta.ClusterPeerID {
+		t.Fatalf("response metadata mismatch: got %+v, want %+v", resp.Meta, meta)
+	}
+}
+
+// TestMasterConn_SendAddMinorBlockHeader verifies the typed outbound helper.
+func TestMasterConn_SendAddMinorBlockHeader(t *testing.T) {
+	client, server, cleanup := newMasterTestConnPair(t)
+	defer cleanup()
+
+	server.RegisterTypedHandlers(map[byte]TypedHandler{
+		byte(wire.ClusterOpAddMinorBlockHeaderRequest): func(req any) (any, error) {
+			r := req.(*wire.AddMinorBlockHeaderRequest)
+			if r.TxCount != 5 {
+				t.Fatalf("unexpected tx_count: %d", r.TxCount)
+			}
+			return &wire.AddMinorBlockHeaderResponse{ErrorCode: 0}, nil
+		},
+	})
+
+	server.Start()
+	client.Start()
+
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	req := &wire.AddMinorBlockHeaderRequest{
+		MinorBlockHeader:  emptyRawBytes(),
+		TxCount:           5,
+		XShardTxCount:     0,
+		CoinbaseAmountMap: emptyRawBytes(),
+		ShardStats:        wire.ShardStats{Branch: 0x00010001},
+	}
+	resp, err := client.SendAddMinorBlockHeader(ctx, req)
+	if err != nil {
+		t.Fatalf("SendAddMinorBlockHeader: %v", err)
+	}
+	if resp.ErrorCode != 0 {
+		t.Fatalf("unexpected error_code: %d", resp.ErrorCode)
+	}
+}
+
+// TestMasterConn_12ByteMetadata verifies that ClusterMetadata is encoded as
+// 4-byte branch followed by 8-byte cluster_peer_id.
+func TestMasterConn_12ByteMetadata(t *testing.T) {
+	ln, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatalf("listen: %v", err)
+	}
+
+	var serverConn net.Conn
+	var acceptErr error
+	accepted := make(chan struct{})
+	go func() {
+		defer close(accepted)
+		serverConn, acceptErr = ln.Accept()
+		ln.Close()
+	}()
+
+	clientConn, err := net.Dial("tcp", ln.Addr().String())
+	if err != nil {
+		t.Fatalf("dial: %v", err)
+	}
+	<-accepted
+	if acceptErr != nil {
+		t.Fatalf("accept: %v", acceptErr)
+	}
+	defer clientConn.Close()
+	defer serverConn.Close()
+
+	mc := NewMasterConnFromConn(clientConn, 0, []byte("s"), []uint32{1}, log.New())
+	defer mc.Close()
+
+	// Read the raw first frame written by the client to inspect metadata layout.
+	go func() {
+		// Accept but do not respond; we only need the wire bytes.
+		buf := make([]byte, 1024)
+		_, _ = serverConn.Read(buf)
+	}()
+
+	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
+	defer cancel()
+
+	payload, _ := serialize.SerializeToBytes(&wire.GetEcoInfoListRequest{})
+	// This RPC will time out because the fake server does not reply, but the
+	// request bytes are still written to the wire before the timeout.
+	_, _ = mc.SendRPCMeta(ctx, byte(wire.ClusterOpGetEcoInfoListRequest), payload, wire.ClusterMetadata{Branch: 0x01020304, ClusterPeerID: 0x1122334455667788})
+
+	// Read back what the client wrote from serverConn using a fresh connection
+	// is not straightforward; instead verify the metadata marshal helper.
+	meta := wire.ClusterMetadata{Branch: 0x01020304, ClusterPeerID: 0x1122334455667788}
+	b := wire.MarshalClusterMetadata(meta)
+	if len(b) != 12 {
+		t.Fatalf("metadata length: got %d, want 12", len(b))
+	}
+	if binary.BigEndian.Uint32(b[0:4]) != meta.Branch {
+		t.Fatalf("branch mismatch")
+	}
+	if binary.BigEndian.Uint64(b[4:12]) != meta.ClusterPeerID {
+		t.Fatalf("cluster_peer_id mismatch")
+	}
+}
+
+// TestMasterConn_StubResponsesAreValidBytes verifies that every master handler
+// stub returns a response that can be serialized.
+func TestMasterConn_StubResponsesAreValidBytes(t *testing.T) {
+	_, server, cleanup := newMasterTestConnPair(t)
+	defer cleanup()
+
+	server.Start()
+
+	cases := []struct {
+		opcode wire.ClusterOp
+		req    any
+		resp   any
+	}{
+		{wire.ClusterOpPing, &wire.PingRequest{ID: []byte("m"), FullShardIDList: []uint32{1}}, &wire.PongResponse{}},
+		{wire.ClusterOpConnectToSlavesRequest, &wire.ConnectToSlavesRequest{SlaveInfoList: []wire.SlaveInfo{}}, &wire.ConnectToSlavesResponse{}},
+		{wire.ClusterOpMineRequest, &wire.MineRequest{}, &wire.MineResponse{}},
+		{wire.ClusterOpGenTxRequest, &wire.GenTxRequest{Tx: emptyRawBytes()}, &wire.GenTxResponse{}},
+		{wire.ClusterOpAddRootBlockRequest, &wire.AddRootBlockRequest{RootBlock: emptyRawBytes()}, &wire.AddRootBlockResponse{}},
+		{wire.ClusterOpGetEcoInfoListRequest, &wire.GetEcoInfoListRequest{}, &wire.GetEcoInfoListResponse{}},
+		{wire.ClusterOpGetNextBlockToMineRequest, &wire.GetNextBlockToMineRequest{Address: account.Address{}}, &wire.GetNextBlockToMineResponse{}},
+		{wire.ClusterOpAddMinorBlockRequest, &wire.AddMinorBlockRequest{MinorBlockData: []byte{}}, &wire.AddMinorBlockResponse{}},
+		{wire.ClusterOpGetUnconfirmedHeadersRequest, &wire.GetUnconfirmedHeadersRequest{}, &wire.GetUnconfirmedHeadersResponse{}},
+		{wire.ClusterOpGetAccountDataRequest, &wire.GetAccountDataRequest{}, &wire.GetAccountDataResponse{}},
+		{wire.ClusterOpAddTransactionRequest, &wire.AddTransactionRequest{Tx: emptyRawBytes()}, &wire.AddTransactionResponse{}},
+		{wire.ClusterOpCreateClusterPeerConnectionRequest, &wire.CreateClusterPeerConnectionRequest{ClusterPeerID: 1}, &wire.CreateClusterPeerConnectionResponse{}},
+		{wire.ClusterOpGetMinorBlockRequest, &wire.GetMinorBlockRequest{}, &wire.GetMinorBlockResponse{}},
+		{wire.ClusterOpGetTransactionRequest, &wire.GetTransactionRequest{}, &wire.GetTransactionResponse{}},
+		{wire.ClusterOpSyncMinorBlockListRequest, &wire.SyncMinorBlockListRequest{MinorBlockHashList: [][wire.HashLength]byte{}}, &wire.SyncMinorBlockListResponse{}},
+		{wire.ClusterOpExecuteTransactionRequest, &wire.ExecuteTransactionRequest{Tx: emptyRawBytes()}, &wire.ExecuteTransactionResponse{}},
+		{wire.ClusterOpGetTransactionReceiptRequest, &wire.GetTransactionReceiptRequest{}, &wire.GetTransactionReceiptResponse{}},
+		{wire.ClusterOpGetTransactionListByAddressRequest, &wire.GetTransactionListByAddressRequest{}, &wire.GetTransactionListByAddressResponse{}},
+		{wire.ClusterOpGetLogRequest, &wire.GetLogRequest{}, &wire.GetLogResponse{}},
+		{wire.ClusterOpEstimateGasRequest, &wire.EstimateGasRequest{Tx: emptyRawBytes()}, &wire.EstimateGasResponse{}},
+		{wire.ClusterOpGetStorageRequest, &wire.GetStorageRequest{}, &wire.GetStorageResponse{}},
+		{wire.ClusterOpGetCodeRequest, &wire.GetCodeRequest{}, &wire.GetCodeResponse{}},
+		{wire.ClusterOpGasPriceRequest, &wire.GasPriceRequest{}, &wire.GasPriceResponse{}},
+		{wire.ClusterOpGetWorkRequest, &wire.GetWorkRequest{}, &wire.GetWorkResponse{}},
+		{wire.ClusterOpSubmitWorkRequest, &wire.SubmitWorkRequest{}, &wire.SubmitWorkResponse{}},
+		{wire.ClusterOpCheckMinorBlockRequest, &wire.CheckMinorBlockRequest{MinorBlockHeader: emptyRawBytes()}, &wire.CheckMinorBlockResponse{}},
+		{wire.ClusterOpGetAllTransactionsRequest, &wire.GetAllTransactionsRequest{}, &wire.GetAllTransactionsResponse{}},
+		{wire.ClusterOpGetRootChainStakesRequest, &wire.GetRootChainStakesRequest{}, &wire.GetRootChainStakesResponse{}},
+		{wire.ClusterOpGetTotalBalanceRequest, &wire.GetTotalBalanceRequest{}, &wire.GetTotalBalanceResponse{}},
+	}
+
+	for _, tc := range cases {
+		// Serialize the request bytes.
+		reqBytes, err := serialize.SerializeToBytes(tc.req)
+		if err != nil {
+			t.Fatalf("serialize request for opcode 0x%x: %v", tc.opcode, err)
+		}
+
+		// Ask the server to process the request by writing a raw frame.
+		// We use a fresh connection per case to avoid ordering issues.
+		client, srv, cleanupPair := newMasterTestConnPair(t)
+		srv.Start()
+
+		writeRawMasterFrame(t, client.conn, &wire.Frame{
+			Meta:    wire.ClusterMetadata{Branch: 0x00010001},
+			Opcode:  byte(tc.opcode),
+			RPCID:   1,
+			Payload: reqBytes,
+		})
+
+		// Read the raw response from the connection.
+		clientConn := client.conn
+		clientConn.SetReadDeadline(time.Now().Add(2 * time.Second))
+		frame, err := wire.ReadFrame(clientConn, 0)
+		if err != nil {
+			t.Fatalf("read response for opcode 0x%x: %v", tc.opcode, err)
+		}
+		if frame.Opcode != byte(tc.opcode)+1 {
+			t.Fatalf("opcode 0x%x: expected response opcode 0x%x, got 0x%x", tc.opcode, byte(tc.opcode)+1, frame.Opcode)
+		}
+		if err := serialize.Deserialize(serialize.NewByteBuffer(frame.Payload), tc.resp); err != nil {
+			t.Fatalf("deserialize response for opcode 0x%x: %v", tc.opcode, err)
+		}
+
+		cleanupPair()
+	}
+}
+
+// TestMasterConn_EmptyPayloadDeserialization verifies that request types with
+// empty bodies deserialize correctly.
+func TestMasterConn_EmptyPayloadDeserialization(t *testing.T) {
+	client, server, cleanup := newMasterTestConnPair(t)
+	defer cleanup()
+
+	// Only start the server; the client connection is used as a bare socket so
+	// we can observe the raw response frame without the client's readLoop
+	// competing for bytes.
+	server.Start()
+
+	// Empty payload should deserialize to an empty GetEcoInfoListRequest.
+	writeRawMasterFrame(t, client.conn, &wire.Frame{
+		Meta:    wire.ClusterMetadata{},
+		Opcode:  byte(wire.ClusterOpGetEcoInfoListRequest),
+		RPCID:   1,
+		Payload: []byte{},
+	})
+
+	// Read the response from the same connection the client wrote on.
+	client.conn.SetReadDeadline(time.Now().Add(2 * time.Second))
+	frame, err := wire.ReadFrame(client.conn, 0)
+	if err != nil {
+		t.Fatalf("read response: %v", err)
+	}
+	if frame.Opcode != byte(wire.ClusterOpGetEcoInfoListResponse) {
+		t.Fatalf("expected GetEcoInfoListResponse, got 0x%x", frame.Opcode)
+	}
+}
+
+// TestMasterConn_MetadataPreserved verifies that request metadata is echoed
+// back in the response.
+func TestMasterConn_MetadataPreserved(t *testing.T) {
+	client, server, cleanup := newMasterTestConnPair(t)
+	defer cleanup()
+
+	server.Start()
+	client.Start()
+
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	meta := wire.ClusterMetadata{Branch: 0xDEADBEEF, ClusterPeerID: 0xCAFEBABECAFEBABE}
+	payload, _ := serialize.SerializeToBytes(&wire.GetEcoInfoListRequest{})
+	resp, err := client.SendRPCMeta(ctx, byte(wire.ClusterOpGetEcoInfoListRequest), payload, meta)
+	if err != nil {
+		t.Fatalf("send rpc: %v", err)
+	}
+	if resp.Meta != meta {
+		t.Fatalf("metadata not preserved: got %+v, want %+v", resp.Meta, meta)
+	}
+}
+
+// TestMasterConn_FrameWireLayout verifies the full ClusterMetadata frame layout
+// written by MasterConn matches the Python protocol.
+func TestMasterConn_FrameWireLayout(t *testing.T) {
+	var buf bytes.Buffer
+	frame := &wire.Frame{
+		Meta:    wire.ClusterMetadata{Branch: 0x01020304, ClusterPeerID: 0x1122334455667788},
+		Opcode:  byte(wire.ClusterOpPing),
+		RPCID:   0xAABBCCDDEEFF0011,
+		Payload: []byte{0xAA, 0xBB},
+	}
+	if err := wire.WriteFrame(&buf, frame); err != nil {
+		t.Fatalf("WriteFrame: %v", err)
+	}
+
+	wireBytes := buf.Bytes()
+	if len(wireBytes) != 4+12+1+8+2 {
+		t.Fatalf("frame length: got %d, want %d", len(wireBytes), 4+12+1+8+2)
+	}
+
+	if got := binary.BigEndian.Uint32(wireBytes[0:4]); got != 2 {
+		t.Fatalf("payload_len: got %d, want 2", got)
+	}
+	if got := binary.BigEndian.Uint32(wireBytes[4:8]); got != frame.Meta.Branch {
+		t.Fatalf("branch mismatch: got 0x%x", got)
+	}
+	if got := binary.BigEndian.Uint64(wireBytes[8:16]); got != frame.Meta.ClusterPeerID {
+		t.Fatalf("cluster_peer_id mismatch: got 0x%x", got)
+	}
+	if wireBytes[16] != frame.Opcode {
+		t.Fatalf("opcode mismatch: got 0x%x", wireBytes[16])
+	}
+	if got := binary.BigEndian.Uint64(wireBytes[17:25]); got != frame.RPCID {
+		t.Fatalf("rpc_id mismatch: got 0x%x", got)
+	}
+	if !bytes.Equal(wireBytes[25:], frame.Payload) {
+		t.Fatalf("payload mismatch: got %x", wireBytes[25:])
+	}
+}

qkc/cluster/slave/testdata/pyproto/master.py

@@ -0,0 +1,163 @@
+#!/usr/bin/env python3
+"""Minimal MasterConnection protocol peer for Go compatibility tests.
+
+This peer implements the master-to-slave protocol that MasterConn needs:
+  - Frame read/write (12-byte ClusterMetadata, matching ReadFrame/WriteFrame)
+  - PING/PONG identity exchange (ClusterOp 0x81/0x82)
+  - RPC request/response for a representative set of master->slave opcodes
+  - Fire-and-forget command dispatch
+
+Usage:
+    python3 master.py --port 0 --id "master" --shards "1,2"
+
+Output:
+    PORT:<port>                   Printed when listening
+    PONG_OK id=<hex>              Printed when PONG is received
+    ECO_OK error_code=<n>         Printed when GetEcoInfoListResponse is received
+    ROOT_OK error_code=<n>        Printed when AddRootBlockResponse is received
+    DESTROY_OK                    Printed after DESTROY command (no response expected)
+    DISCONNECTED                  Printed when connection closes
+
+Behavior:
+  - Listens on TCP, accepts one connection
+  - Sends PING (rpc_id=1), waits for PONG
+  - Sends GetEcoInfoListRequest (rpc_id=2), waits for GetEcoInfoListResponse
+  - Sends AddRootBlockRequest (rpc_id=3), waits for AddRootBlockResponse
+  - Sends DestroyClusterPeerConnectionCommand (rpc_id=0)
+  - Sends a second PING (rpc_id=4) to verify the connection is still alive
+  - Closes connection and exits
+"""
+import argparse
+import socket
+import struct
+import sys
+
+from master_frame import read_master_frame, write_master_frame
+from messages import serialize_ping_request, parse_pong_response
+
+CLUSTER_OP_BASE = 0x80
+
+CLUSTER_OP_PING = 1 + CLUSTER_OP_BASE
+CLUSTER_OP_PONG = 2 + CLUSTER_OP_BASE
+
+# Master -> Slave opcodes
+CLUSTER_OP_GET_ECO_INFO_LIST_REQUEST = 7 + CLUSTER_OP_BASE
+CLUSTER_OP_GET_ECO_INFO_LIST_RESPONSE = 8 + CLUSTER_OP_BASE
+CLUSTER_OP_ADD_ROOT_BLOCK_REQUEST = 5 + CLUSTER_OP_BASE
+CLUSTER_OP_ADD_ROOT_BLOCK_RESPONSE = 6 + CLUSTER_OP_BASE
+CLUSTER_OP_DESTROY_CLUSTER_PEER_CONNECTION_COMMAND = 27 + CLUSTER_OP_BASE
+
+
+def main():
+    parser = argparse.ArgumentParser()
+    parser.add_argument('--port', type=int, required=True)
+    parser.add_argument('--id', type=str, required=True)
+    parser.add_argument('--shards', type=str, required=True)
+    args = parser.parse_args()
+
+    master_id = args.id.encode('utf-8')
+    shard_list = [int(s) for s in args.shards.split(',')]
+
+    server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+    server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+    server.bind(('127.0.0.1', args.port))
+    server.listen(1)
+
+    actual_port = server.getsockname()[1]
+    print(f"PORT:{actual_port}", flush=True)
+
+    conn, _ = server.accept()
+
+    try:
+        # 1. PING -> PONG
+        _send_ping(conn, master_id, shard_list)
+
+        # 2. GetEcoInfoListRequest -> GetEcoInfoListResponse
+        _send_rpc(
+            conn,
+            CLUSTER_OP_GET_ECO_INFO_LIST_REQUEST,
+            CLUSTER_OP_GET_ECO_INFO_LIST_RESPONSE,
+            2,
+            b'',
+            'ECO_OK',
+        )
+
+        # 3. AddRootBlockRequest -> AddRootBlockResponse
+        add_root_block_payload = b'\x00\x00\x00\x00' + b'\x00'  # empty root block + expect_switch=False
+        _send_rpc(
+            conn,
+            CLUSTER_OP_ADD_ROOT_BLOCK_REQUEST,
+            CLUSTER_OP_ADD_ROOT_BLOCK_RESPONSE,
+            3,
+            add_root_block_payload,
+            'ROOT_OK',
+        )
+
+        # 4. Fire-and-forget DestroyClusterPeerConnectionCommand
+        write_master_frame(
+            conn,
+            CLUSTER_OP_DESTROY_CLUSTER_PEER_CONNECTION_COMMAND,
+            0,
+            struct.pack('>Q', 42),
+            branch=0x00010001,
+        )
+        print("DESTROY_OK", flush=True)
+
+        # 5. Second PING to confirm the connection is still alive after the
+        #    fire-and-forget command.
+        _send_ping(conn, master_id, shard_list)
+
+    except (ConnectionError, BrokenPipeError, OSError) as e:
+        print(f"ERROR: {e}", flush=True)
+        sys.exit(1)
+    finally:
+        conn.close()
+        server.close()
+        print("DISCONNECTED", flush=True)
+
+
+def _send_ping(conn, master_id, shard_list):
+    ping_payload = serialize_ping_request(master_id, shard_list)
+    write_master_frame(conn, CLUSTER_OP_PING, 1, ping_payload)
+
+    frame = read_master_frame(conn)
+    if frame is None:
+        print("ERROR: no pong received", flush=True)
+        sys.exit(1)
+
+    if frame['opcode'] != CLUSTER_OP_PONG:
+        print(f"ERROR: expected PONG(0x{CLUSTER_OP_PONG:02x}), got 0x{frame['opcode']:02x}", flush=True)
+        sys.exit(1)
+    if frame['rpc_id'] != 1:
+        print(f"ERROR: expected rpc_id 1, got {frame['rpc_id']}", flush=True)
+        sys.exit(1)
+
+    peer_id_recv, _ = parse_pong_response(frame['payload'])
+    print(f"PONG_OK id={peer_id_recv.hex()}", flush=True)
+
+
+def _send_rpc(conn, req_opcode, resp_opcode, rpc_id, payload, ok_label):
+    write_master_frame(conn, req_opcode, rpc_id, payload, branch=0x00010001)
+
+    frame = read_master_frame(conn)
+    if frame is None:
+        print(f"ERROR: no response for opcode 0x{req_opcode:02x}", flush=True)
+        sys.exit(1)
+
+    if frame['opcode'] != resp_opcode:
+        print(f"ERROR: expected 0x{resp_opcode:02x}, got 0x{frame['opcode']:02x}", flush=True)
+        sys.exit(1)
+    if frame['rpc_id'] != rpc_id:
+        print(f"ERROR: expected rpc_id {rpc_id}, got {frame['rpc_id']}", flush=True)
+        sys.exit(1)
+
+    # First field of these responses is a uint32 error_code.
+    if len(frame['payload']) < 4:
+        print(f"ERROR: response payload too short for {ok_label}", flush=True)
+        sys.exit(1)
+    error_code = struct.unpack('>I', frame['payload'][0:4])[0]
+    print(f"{ok_label} error_code={error_code}", flush=True)
+
+
+if __name__ == '__main__':
+    main()

qkc/cluster/slave/testdata/pyproto/master_frame.py

@@ -0,0 +1,53 @@
+"""Frame read/write for master-slave protocol (12-byte ClusterMetadata).
+
+Wire format: [4B payload_len][4B branch][8B cluster_peer_id][1B opcode][8B rpc_id][payload]
+
+This matches Go's qkc/cluster/wire ReadFrame/WriteFrame with ClusterMetadata.
+"""
+import struct
+
+
+def read_master_frame(conn):
+    """Read one master frame from conn.
+
+    Returns a dict with keys: branch, cluster_peer_id, opcode, rpc_id, payload.
+    Returns None on EOF.
+    """
+    header = conn.recv(25)  # 4 + 12 + 1 + 8
+    if not header:
+        return None
+    if len(header) < 25:
+        raise ConnectionError("truncated master frame header")
+
+    payload_len = struct.unpack('>I', header[0:4])[0]
+    branch = struct.unpack('>I', header[4:8])[0]
+    cluster_peer_id = struct.unpack('>Q', header[8:16])[0]
+    opcode = header[16]
+    rpc_id = struct.unpack('>Q', header[17:25])[0]
+
+    payload = b''
+    while len(payload) < payload_len:
+        chunk = conn.recv(payload_len - len(payload))
+        if not chunk:
+            raise ConnectionError("truncated master frame payload")
+        payload += chunk
+
+    return {
+        'branch': branch,
+        'cluster_peer_id': cluster_peer_id,
+        'opcode': opcode,
+        'rpc_id': rpc_id,
+        'payload': payload,
+    }
+
+
+def write_master_frame(conn, opcode, rpc_id, payload, branch=0, cluster_peer_id=0):
+    """Write one master frame to conn."""
+    header = (
+        struct.pack('>I', len(payload))
+        + struct.pack('>I', branch)
+        + struct.pack('>Q', cluster_peer_id)
+        + bytes([opcode])
+        + struct.pack('>Q', rpc_id)
+    )
+    conn.sendall(header + payload)