From 5ac4bca3029ab592ad79cf21b51c14a6162f5d09 Mon Sep 17 00:00:00 2001
From: iteye <iteyelmp@gmail.com>
Date: Wed, 15 Jul 2026 10:06:46 +0800
Subject: [PATCH 1/4] add PeerConn, Dispatcher, and peer routing

---
 qkc/cluster/slave/compat_test.go              | 174 +++++
 qkc/cluster/slave/dispatcher.go               | 169 +++++
 qkc/cluster/slave/master_conn.go              |  86 ++-
 qkc/cluster/slave/peer_conn.go                | 228 ++++++
 qkc/cluster/slave/peer_conn_test.go           | 669 ++++++++++++++++++
 .../slave/testdata/pyproto/peer_master.py     | 245 +++++++
 6 files changed, 1566 insertions(+), 5 deletions(-)
 create mode 100644 qkc/cluster/slave/dispatcher.go
 create mode 100644 qkc/cluster/slave/peer_conn.go
 create mode 100644 qkc/cluster/slave/peer_conn_test.go
 create mode 100644 qkc/cluster/slave/testdata/pyproto/peer_master.py

diff --git a/qkc/cluster/slave/compat_test.go b/qkc/cluster/slave/compat_test.go
index 79e97c21e2f8..d96d1e8f266a 100644
--- a/qkc/cluster/slave/compat_test.go
+++ b/qkc/cluster/slave/compat_test.go
@@ -228,6 +228,117 @@ func dialPythonPeer(t *testing.T, extraArgs ...string) (*XshardConn, func()) {
 	return xc, cleanup
 }
 
+// startPythonPeerMaster starts a Python peer_master.py subprocess that simulates
+// a Master creating PeerConn and sending peer traffic. Returns the TCP port,
+// a function to retrieve captured stdout lines, and a cleanup function.
+func startPythonPeerMaster(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", "peer_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("peer_master.py not found at %s", pyScript)
+	}
+
+	args := []string{pyScript, "--port", "0", "--id", "py-master", "--shards", "1,2", "--cluster-peer-id", "42"}
+	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 peer 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 peer 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
+}
+
+// dialPythonPeerMaster starts python peer_master.py and dials the port it listens
+// on, wrapping the connection in a MasterConn with a Dispatcher. Returns the
+// MasterConn, Dispatcher, a function to retrieve captured stdout lines, and a
+// cleanup function.
+func dialPythonPeerMaster(t *testing.T) (*MasterConn, *Dispatcher, func() []string, func()) {
+	t.Helper()
+
+	port, getOutput, cleanupPy := startPythonPeerMaster(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)
+	}
+
+	dispatcher := NewDispatcher(log.New())
+	mc.SetDispatcher(dispatcher)
+	mc.Start()
+
+	cleanup := func() {
+		mc.Close()
+		cleanupPy()
+	}
+
+	return mc, dispatcher, getOutput, cleanup
+}
+
 // ---------------------------------------------------------------------------
 // Test: Python → Go PING/PONG
 //
@@ -477,6 +588,69 @@ func TestPythonCompat_MasterFullFlow(t *testing.T) {
 	}
 }
 
+// ---------------------------------------------------------------------------
+// Test: Python Master → Go Slave PeerConn integration flow
+//
+// Validates: Python Master creates PeerConn via CreateClusterPeerConnectionRequest,
+// sends peer traffic (CommandOp frames with cluster_peer_id != 0), and destroys
+// PeerConn via DestroyClusterPeerConnectionCommand. Go must correctly route frames
+// through Dispatcher, handle peer RPC and non-RPC traffic, and maintain MasterConn
+// alive after PeerConn lifecycle events.
+//
+// drives the entire flow, and Go must respond with protocol-compatible behavior.
+// ---------------------------------------------------------------------------
+func TestPythonCompat_PeerConnIntegrationFlow(t *testing.T) {
+	mc, dispatcher, getOutput, cleanup := dialPythonPeerMaster(t)
+	defer cleanup()
+
+	// Wait for the Python peer 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 peer 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()
+
+	// Verify all expected Python outputs are present.
+	expected := []string{
+		"PONG_OK id=676f2d736c617665", // hex of "go-slave"
+		"CREATE_OK error_code=0",
+		"PEER_RPC_OK opcode=0x0a rpc_id=100",
+		"PEER_NONRPC_OK",
+		"DESTROY_OK",
+		"POST_DESTROY_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)
+		}
+	}
+
+	// Verify that the Dispatcher created and destroyed the PeerConn.
+	// After the Python script finishes, the PeerConn should have been destroyed.
+	dispatcher.mu.RLock()
+	peerCount := len(dispatcher.peers)
+	dispatcher.mu.RUnlock()
+
+	if peerCount != 0 {
+		t.Fatalf("expected 0 peers after destroy, got %d", peerCount)
+	}
+}
+
 // ---------------------------------------------------------------------------
 // Test: Python-generated ClusterMetadata frame layout
 //
diff --git a/qkc/cluster/slave/dispatcher.go b/qkc/cluster/slave/dispatcher.go
new file mode 100644
index 000000000000..d2bf05386661
--- /dev/null
+++ b/qkc/cluster/slave/dispatcher.go
@@ -0,0 +1,169 @@
+// Copyright 2026-2027, QuarkChain.
+
+package slave
+
+import (
+	"fmt"
+	"sync"
+
+	"github.com/ethereum/go-ethereum/log"
+	"github.com/ethereum/go-ethereum/qkc/cluster/wire"
+)
+
+// Dispatcher routes intra-cluster frames that carry a non-zero cluster_peer_id
+// to the corresponding virtual PeerConn. Frames with cluster_peer_id == 0 are
+// left for MasterConn to handle.
+//
+// It owns the registry of PeerConns as a two-layer map:
+//
+//	cluster_peer_id -> branch -> *PeerConn
+//
+// This matches Python's MasterConnection.v_conn_map and shard.peers layout.
+type Dispatcher struct {
+	mu    sync.RWMutex
+	peers map[uint64]map[uint32]*PeerConn
+	log   log.Logger
+}
+
+// NewDispatcher creates an empty dispatcher.
+func NewDispatcher(logger log.Logger) *Dispatcher {
+	if logger == nil {
+		logger = log.Root()
+	}
+	return &Dispatcher{
+		peers: make(map[uint64]map[uint32]*PeerConn),
+		log:   logger,
+	}
+}
+
+// Register adds an already-created PeerConn to the registry. It returns an
+// error if a PeerConn for the same cluster_peer_id and branch already exists.
+func (d *Dispatcher) Register(pc *PeerConn) error {
+	d.mu.Lock()
+	defer d.mu.Unlock()
+
+	branchMap, ok := d.peers[pc.ClusterPeerID()]
+	if !ok {
+		branchMap = make(map[uint32]*PeerConn)
+		d.peers[pc.ClusterPeerID()] = branchMap
+	}
+	if _, exists := branchMap[pc.Branch()]; exists {
+		return fmt.Errorf("peer connection already exists for cluster_peer_id %d branch %d", pc.ClusterPeerID(), pc.Branch())
+	}
+	branchMap[pc.Branch()] = pc
+	return nil
+}
+
+// Unregister removes a single PeerConn from the registry. It returns the
+// removed PeerConn (if any) without closing it.
+func (d *Dispatcher) Unregister(clusterPeerID uint64, branch uint32) *PeerConn {
+	d.mu.Lock()
+	defer d.mu.Unlock()
+
+	branchMap, ok := d.peers[clusterPeerID]
+	if !ok {
+		return nil
+	}
+	pc := branchMap[branch]
+	delete(branchMap, branch)
+	if len(branchMap) == 0 {
+		delete(d.peers, clusterPeerID)
+	}
+	return pc
+}
+
+// CreatePeerConns creates and starts one PeerConn per branch for the given
+// cluster_peer_id, using masterConn as the transport. Existing branch entries
+// are skipped (logged as duplicates), matching Python's behavior.
+func (d *Dispatcher) CreatePeerConns(clusterPeerID uint64, branches []uint32, masterConn *MasterConn, logger log.Logger) {
+	if clusterPeerID == ReservedClusterPeerID {
+		d.log.Error("refusing to create peer connection with reserved cluster_peer_id", "cluster_peer_id", clusterPeerID)
+		return
+	}
+
+	d.mu.Lock()
+	defer d.mu.Unlock()
+
+	branchMap, ok := d.peers[clusterPeerID]
+	if !ok {
+		branchMap = make(map[uint32]*PeerConn)
+		d.peers[clusterPeerID] = branchMap
+	}
+
+	for _, branch := range branches {
+		if _, exists := branchMap[branch]; exists {
+			d.log.Warn("duplicate create cluster peer connection", "cluster_peer_id", clusterPeerID, "branch", branch)
+			continue
+		}
+		pc := NewPeerConn(clusterPeerID, branch, masterConn, logger)
+		pc.Start()
+		branchMap[branch] = pc
+	}
+}
+
+// DestroyPeerConns removes all PeerConns for clusterPeerID from the registry
+// and closes them. Missing entries are silently ignored.
+func (d *Dispatcher) DestroyPeerConns(clusterPeerID uint64) {
+	d.mu.Lock()
+	branchMap, ok := d.peers[clusterPeerID]
+	if ok {
+		delete(d.peers, clusterPeerID)
+	}
+	d.mu.Unlock()
+
+	if !ok {
+		return
+	}
+	for _, pc := range branchMap {
+		pc.Close()
+	}
+}
+
+// RouteFrame is the forwarder callback installed on MasterConn. It returns
+// false for master-local traffic (cluster_peer_id == 0) so MasterConn handles
+// the frame normally. For peer traffic it looks up the PeerConn, enqueues the
+// frame if found, or drops it (matching Python's NULL_CONNECTION) and logs a
+// warning if not found.
+func (d *Dispatcher) RouteFrame(frame *wire.Frame) bool {
+	if frame.Meta.ClusterPeerID == 0 {
+		return false
+	}
+
+	d.mu.RLock()
+	branchMap, ok := d.peers[frame.Meta.ClusterPeerID]
+	if !ok {
+		d.mu.RUnlock()
+		d.log.Warn("no peer connection for cluster_peer_id", "cluster_peer_id", frame.Meta.ClusterPeerID)
+		return true
+	}
+	pc, ok := branchMap[frame.Meta.Branch]
+	d.mu.RUnlock()
+
+	if !ok {
+		d.log.Warn("no peer connection for branch", "cluster_peer_id", frame.Meta.ClusterPeerID, "branch", frame.Meta.Branch)
+		return true
+	}
+
+	if err := pc.HandleFrame(frame); err != nil {
+		d.log.Warn("failed to deliver frame to peer connection", "cluster_peer_id", frame.Meta.ClusterPeerID, "branch", frame.Meta.Branch, "err", err)
+	}
+	return true
+}
+
+// Close closes all registered PeerConns and clears the registry.
+func (d *Dispatcher) Close() error {
+	d.mu.Lock()
+	all := make([]*PeerConn, 0)
+	for _, branchMap := range d.peers {
+		for _, pc := range branchMap {
+			all = append(all, pc)
+		}
+	}
+	d.peers = make(map[uint64]map[uint32]*PeerConn)
+	d.mu.Unlock()
+
+	for _, pc := range all {
+		pc.Close()
+	}
+	return nil
+}
diff --git a/qkc/cluster/slave/master_conn.go b/qkc/cluster/slave/master_conn.go
index 7b14038a990b..113ec02c774e 100644
--- a/qkc/cluster/slave/master_conn.go
+++ b/qkc/cluster/slave/master_conn.go
@@ -7,6 +7,7 @@ import (
 	"fmt"
 	"io"
 	"net"
+	"sync"
 
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/qkc/cluster/wire"
@@ -29,6 +30,18 @@ type MasterConn struct {
 
 	localID              []byte
 	localFullShardIDList []uint32
+
+	// dispatcher routes peer traffic (cluster_peer_id != 0) to virtual PeerConns.
+	// It is nil until wired by SetDispatcher.
+	dispatcher   *Dispatcher
+	dispatcherMu sync.RWMutex
+
+	// peerRPCIDs tracks the most recent inbound RPC ID per cluster_peer_id.
+	// cluster_peer_id == 0 is the master itself; each non-zero peer has its
+	// own independent monotonic sequence so PeerConns sharing this MasterConn
+	// do not collide on rpc_id.
+	peerRPCIDs   map[uint64]int64
+	peerRPCIDsMu sync.Mutex
 }
 
 // NewMasterConn dials the master at addr and returns a MasterConn.
@@ -56,8 +69,11 @@ func newMasterConn(conn net.Conn, maxPayloadSize uint32, localID []byte, localFu
 		rpcConn:              newRPCConnFromConn(conn, readFrame, wire.WriteFrame, logger),
 		localID:              append([]byte(nil), localID...),
 		localFullShardIDList: append([]uint32(nil), localFullShardIDList...),
+		peerRPCIDs:           make(map[uint64]int64),
 	}
 
+	mc.rpcConn.validateRPCID = mc.validatePeerRPCID
+
 	mc.registerOpSerializers()
 	mc.registerHandlers()
 
@@ -206,6 +222,24 @@ func emptyRawBytes() *wire.RawBytes {
 	return rawBytes([]byte{})
 }
 
+// validatePeerRPCID validates inbound RPC IDs independently for each
+// cluster_peer_id. This lets multiple PeerConns share one MasterConn without
+// colliding on rpc_id.
+func (mc *MasterConn) validatePeerRPCID(clusterPeerID uint64, rpcID uint64) bool {
+	mc.peerRPCIDsMu.Lock()
+	defer mc.peerRPCIDsMu.Unlock()
+
+	last, ok := mc.peerRPCIDs[clusterPeerID]
+	if !ok {
+		last = -1
+	}
+	if int64(rpcID) <= last {
+		return false
+	}
+	mc.peerRPCIDs[clusterPeerID] = int64(rpcID)
+	return true
+}
+
 // LocalID returns this slave's ID used in PONG responses.
 func (mc *MasterConn) LocalID() []byte {
 	return append([]byte(nil), mc.localID...)
@@ -313,19 +347,35 @@ func (mc *MasterConn) handleAddTransaction(req any) (any, error) {
 	return &wire.AddTransactionResponse{ErrorCode: 0}, nil
 }
 
-// handleCreateClusterPeerConnection creates virtual peer connections for all shards.
+// handleCreateClusterPeerConnection creates virtual peer connections for all
+// shards. Until PR7's Shard Registry is available, localFullShardIDList is used
+// as the shard list.
 // 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).
+	r := req.(*wire.CreateClusterPeerConnectionRequest)
+
+	mc.dispatcherMu.RLock()
+	d := mc.dispatcher
+	mc.dispatcherMu.RUnlock()
+	if d != nil {
+		d.CreatePeerConns(r.ClusterPeerID, mc.localFullShardIDList, mc, mc.rpcConn.log)
+	}
+
 	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).
+	r := req.(*wire.DestroyClusterPeerConnectionCommand)
+
+	mc.dispatcherMu.RLock()
+	d := mc.dispatcher
+	mc.dispatcherMu.RUnlock()
+	if d != nil {
+		d.DestroyPeerConns(r.ClusterPeerID)
+	}
+
 	return nil, nil
 }
 
@@ -507,6 +557,32 @@ func (mc *MasterConn) SetForwarder(f func(*wire.Frame) bool) {
 	mc.rpcConn.SetForwarder(f)
 }
 
+// SetDispatcher wires the dispatcher that routes peer traffic. It also installs
+// the dispatcher as the raw-frame forwarder on this MasterConn.
+func (mc *MasterConn) SetDispatcher(d *Dispatcher) {
+	mc.dispatcherMu.Lock()
+	mc.dispatcher = d
+	mc.dispatcherMu.Unlock()
+	mc.SetForwarder(d.RouteFrame)
+}
+
+// ForwardFrame writes a raw frame to the underlying TCP transport. It is used
+// by virtual PeerConns to send responses back to the master.
+func (mc *MasterConn) ForwardFrame(f *wire.Frame) error {
+	return mc.rpcConn.writeFrame(f)
+}
+
+// Close closes the master connection and all associated peer connections.
+func (mc *MasterConn) Close() error {
+	mc.dispatcherMu.RLock()
+	d := mc.dispatcher
+	mc.dispatcherMu.RUnlock()
+	if d != nil {
+		d.Close()
+	}
+	return mc.rpcConn.Close()
+}
+
 // 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) {
diff --git a/qkc/cluster/slave/peer_conn.go b/qkc/cluster/slave/peer_conn.go
new file mode 100644
index 000000000000..c224ceb9bff9
--- /dev/null
+++ b/qkc/cluster/slave/peer_conn.go
@@ -0,0 +1,228 @@
+// Copyright 2026-2027, QuarkChain.
+
+package slave
+
+import (
+	"fmt"
+	"sync"
+
+	"github.com/ethereum/go-ethereum/log"
+	"github.com/ethereum/go-ethereum/qkc/cluster/wire"
+)
+
+// virtualTransport implements frameTransport for PeerConn. It has no TCP
+// socket; inbound frames are pushed by the Dispatcher via receive(), and
+// outbound frames are forwarded through the associated MasterConn.
+type virtualTransport struct {
+	clusterPeerID uint64
+	branch        uint32
+	masterConn    *MasterConn
+
+	inbound    chan *wire.Frame
+	closedChan chan struct{}
+	closeOnce  sync.Once
+	remoteAddr string
+}
+
+func newVirtualTransport(clusterPeerID uint64, branch uint32, masterConn *MasterConn) *virtualTransport {
+	return &virtualTransport{
+		clusterPeerID: clusterPeerID,
+		branch:        branch,
+		masterConn:    masterConn,
+		inbound:       make(chan *wire.Frame, 64),
+		closedChan:    make(chan struct{}),
+		remoteAddr:    fmt.Sprintf("virtual://peer/%d/%d", clusterPeerID, branch),
+	}
+}
+
+func (vt *virtualTransport) readFrame() (*wire.Frame, error) {
+	select {
+	case frame := <-vt.inbound:
+		return frame, nil
+	case <-vt.closedChan:
+		return nil, ErrConnectionClosed
+	}
+}
+
+func (vt *virtualTransport) writeFrame(f *wire.Frame) error {
+	// PeerShardConnection in Python always writes with the shard branch and its
+	// own cluster_peer_id so the master can route the frame back to the peer.
+	f.Meta = wire.ClusterMetadata{
+		Branch:        vt.branch,
+		ClusterPeerID: vt.clusterPeerID,
+	}
+	return vt.masterConn.ForwardFrame(f)
+}
+
+func (vt *virtualTransport) close() error {
+	vt.closeOnce.Do(func() { close(vt.closedChan) })
+	return nil
+}
+
+func (vt *virtualTransport) RemoteAddr() string {
+	return vt.remoteAddr
+}
+
+// receive pushes a frame into the inbound queue. It returns false if the
+// transport is already closed.
+func (vt *virtualTransport) receive(frame *wire.Frame) bool {
+	select {
+	case vt.inbound <- frame:
+		return true
+	case <-vt.closedChan:
+		return false
+	}
+}
+
+// PeerConn is a virtual RPC channel representing the slave-side endpoint of a
+// forwarded external peer connection. It does not own a TCP socket; all wire
+// traffic is tunneled through the slave's MasterConn.
+//
+// It corresponds to Python's PeerShardConnection and shares the same
+// responsibilities: independent RPC ID namespace, CommandOp handler dispatch,
+// and lifecycle tied to master commands.
+type PeerConn struct {
+	*rpcConn
+
+	clusterPeerID uint64
+	branch        uint32
+	vt            *virtualTransport
+}
+
+// NewPeerConn creates a virtual peer connection for the given cluster_peer_id
+// and branch, tunneling outbound frames through masterConn.
+func NewPeerConn(clusterPeerID uint64, branch uint32, masterConn *MasterConn, logger log.Logger) *PeerConn {
+	vt := newVirtualTransport(clusterPeerID, branch, masterConn)
+	pc := &PeerConn{
+		rpcConn:       newRPCConn(vt, logger),
+		clusterPeerID: clusterPeerID,
+		branch:        branch,
+		vt:            vt,
+	}
+	pc.registerOpSerializers()
+	pc.registerHandlers()
+	return pc
+}
+
+// ReservedClusterPeerID is the reserved cluster_peer_id used by the master for
+// its own control traffic. PeerConn must not use this value.
+const ReservedClusterPeerID = 0
+
+// registerOpSerializers registers serializers for every CommandOp so that both
+// inbound requests and outbound responses can be (de)serialized.
+func (pc *PeerConn) registerOpSerializers() {
+	pc.rpcConn.RegisterOpSerializers(map[byte]*OpSerializer{
+		// §1 Hello / master-only
+		byte(wire.CommandOpHello):                          OpSerializerFor[wire.HelloCommand, wire.HelloCommand](),
+		byte(wire.CommandOpNewMinorBlockHeaderList):        OpSerializerFor[wire.NewMinorBlockHeaderListCommand, wire.NewMinorBlockHeaderListCommand](),
+		byte(wire.CommandOpNewTransactionList):             OpSerializerFor[wire.NewTransactionListCommand, wire.NewTransactionListCommand](),
+		byte(wire.CommandOpGetPeerListRequest):             OpSerializerFor[wire.GetPeerListRequest, wire.GetPeerListResponse](),
+		byte(wire.CommandOpGetPeerListResponse):            OpSerializerFor[wire.GetPeerListResponse, wire.GetPeerListRequest](),
+		byte(wire.CommandOpGetRootBlockHeaderListRequest):  OpSerializerFor[wire.GetRootBlockHeaderListRequest, wire.GetRootBlockHeaderListResponse](),
+		byte(wire.CommandOpGetRootBlockHeaderListResponse): OpSerializerFor[wire.GetRootBlockHeaderListResponse, wire.GetRootBlockHeaderListRequest](),
+		byte(wire.CommandOpGetRootBlockListRequest):        OpSerializerFor[wire.GetRootBlockListRequest, wire.GetRootBlockListResponse](),
+		byte(wire.CommandOpGetRootBlockListResponse):       OpSerializerFor[wire.GetRootBlockListResponse, wire.GetRootBlockListRequest](),
+
+		// §2 Slave RPC request/response pairs
+		byte(wire.CommandOpGetMinorBlockListRequest):        OpSerializerFor[wire.GetMinorBlockListRequest, wire.GetMinorBlockListResponse](),
+		byte(wire.CommandOpGetMinorBlockListResponse):       OpSerializerFor[wire.GetMinorBlockListResponse, wire.GetMinorBlockListRequest](),
+		byte(wire.CommandOpGetMinorBlockHeaderListRequest):  OpSerializerFor[wire.GetMinorBlockHeaderListRequest, wire.GetMinorBlockHeaderListResponse](),
+		byte(wire.CommandOpGetMinorBlockHeaderListResponse): OpSerializerFor[wire.GetMinorBlockHeaderListResponse, wire.GetMinorBlockHeaderListRequest](),
+
+		// §3 More master-only / root-chain peer opcodes
+		byte(wire.CommandOpNewBlockMinor):                           OpSerializerFor[wire.NewBlockMinorCommand, wire.NewBlockMinorCommand](),
+		byte(wire.CommandOpPing):                                    OpSerializerFor[wire.PingPongCommand, wire.PingPongCommand](),
+		byte(wire.CommandOpPong):                                    OpSerializerFor[wire.PingPongCommand, wire.PingPongCommand](),
+		byte(wire.CommandOpGetRootBlockHeaderListWithSkipRequest):   OpSerializerFor[wire.GetRootBlockHeaderListWithSkipRequest, wire.GetRootBlockHeaderListResponse](),
+		byte(wire.CommandOpGetRootBlockHeaderListWithSkipResponse):  OpSerializerFor[wire.GetRootBlockHeaderListResponse, wire.GetRootBlockHeaderListWithSkipRequest](),
+		byte(wire.CommandOpNewRootBlock):                            OpSerializerFor[wire.NewRootBlockCommand, wire.NewRootBlockCommand](),
+		byte(wire.CommandOpGetMinorBlockHeaderListWithSkipRequest):  OpSerializerFor[wire.GetMinorBlockHeaderListWithSkipRequest, wire.GetMinorBlockHeaderListResponse](),
+		byte(wire.CommandOpGetMinorBlockHeaderListWithSkipResponse): OpSerializerFor[wire.GetMinorBlockHeaderListResponse, wire.GetMinorBlockHeaderListWithSkipRequest](),
+	})
+}
+
+// registerHandlers registers the shard-level peer handlers. These are stubs
+// because PR6 does not implement shard runtime / block processing.
+func (pc *PeerConn) registerHandlers() {
+	pc.rpcConn.RegisterTypedHandlers(map[byte]TypedHandler{
+		// Non-RPC commands (fire-and-forget).
+		byte(wire.CommandOpNewMinorBlockHeaderList): pc.handleNewMinorBlockHeaderList,
+		byte(wire.CommandOpNewTransactionList):      pc.handleNewTransactionList,
+		byte(wire.CommandOpNewBlockMinor):           pc.handleNewBlockMinor,
+
+		// RPC requests; responses use opcode+1.
+		byte(wire.CommandOpGetMinorBlockListRequest):               pc.handleGetMinorBlockListRequest,
+		byte(wire.CommandOpGetMinorBlockHeaderListRequest):         pc.handleGetMinorBlockHeaderListRequest,
+		byte(wire.CommandOpGetMinorBlockHeaderListWithSkipRequest): pc.handleGetMinorBlockHeaderListWithSkipRequest,
+	})
+
+	pc.rpcConn.RegisterNonRPCOps([]byte{
+		byte(wire.CommandOpNewMinorBlockHeaderList),
+		byte(wire.CommandOpNewTransactionList),
+		byte(wire.CommandOpNewBlockMinor),
+	})
+}
+
+// HandleFrame receives a frame routed by the Dispatcher. It enqueues the frame
+// for the PeerConn read loop. Frames received after close are dropped.
+func (pc *PeerConn) HandleFrame(frame *wire.Frame) error {
+	if pc.Closed() {
+		return ErrConnectionClosed
+	}
+	if !pc.vt.receive(frame) {
+		return ErrConnectionClosed
+	}
+	return nil
+}
+
+// ClusterPeerID returns the peer's cluster-scoped identifier.
+func (pc *PeerConn) ClusterPeerID() uint64 { return pc.clusterPeerID }
+
+// Branch returns the shard branch this virtual connection serves.
+func (pc *PeerConn) Branch() uint32 { return pc.branch }
+
+// ── stub handlers ────────────────────────────────────────────────────────────
+
+func (pc *PeerConn) handleNewMinorBlockHeaderList(req any) (any, error) {
+	_ = req.(*wire.NewMinorBlockHeaderListCommand)
+	// TODO: delegate to shard synchronizer once Shard Runtime is ported.
+	return nil, nil
+}
+
+func (pc *PeerConn) handleNewTransactionList(req any) (any, error) {
+	_ = req.(*wire.NewTransactionListCommand)
+	// TODO: delegate to shard tx pool once Shard Runtime is ported.
+	return nil, nil
+}
+
+func (pc *PeerConn) handleNewBlockMinor(req any) (any, error) {
+	_ = req.(*wire.NewBlockMinorCommand)
+	// TODO: delegate to shard block processing once Shard Runtime is ported.
+	return nil, nil
+}
+
+func (pc *PeerConn) handleGetMinorBlockListRequest(req any) (any, error) {
+	_ = req.(*wire.GetMinorBlockListRequest)
+	// TODO: fetch blocks from shard state db once Shard Runtime is ported.
+	return &wire.GetMinorBlockListResponse{MinorBlockList: []*wire.RawBytes{}}, nil
+}
+
+func (pc *PeerConn) handleGetMinorBlockHeaderListRequest(req any) (any, error) {
+	_ = req.(*wire.GetMinorBlockHeaderListRequest)
+	// TODO: fetch headers from shard state db once Shard Runtime is ported.
+	return &wire.GetMinorBlockHeaderListResponse{
+		RootTip:         nil,
+		ShardTip:        nil,
+		BlockHeaderList: []*wire.RawBytes{},
+	}, nil
+}
+
+func (pc *PeerConn) handleGetMinorBlockHeaderListWithSkipRequest(req any) (any, error) {
+	_ = req.(*wire.GetMinorBlockHeaderListWithSkipRequest)
+	// TODO: fetch headers from shard state db once Shard Runtime is ported.
+	return &wire.GetMinorBlockHeaderListResponse{
+		RootTip:         nil,
+		ShardTip:        nil,
+		BlockHeaderList: []*wire.RawBytes{},
+	}, nil
+}
diff --git a/qkc/cluster/slave/peer_conn_test.go b/qkc/cluster/slave/peer_conn_test.go
new file mode 100644
index 000000000000..d2224e9034f2
--- /dev/null
+++ b/qkc/cluster/slave/peer_conn_test.go
@@ -0,0 +1,669 @@
+// Copyright 2026-2027, QuarkChain.
+
+package slave
+
+import (
+	"context"
+	"net"
+	"testing"
+	"time"
+
+	"github.com/ethereum/go-ethereum/log"
+	"github.com/ethereum/go-ethereum/qkc/cluster/wire"
+	"github.com/ethereum/go-ethereum/qkc/serialize"
+)
+
+// newMasterConnWithDispatcher creates a MasterConn over a local TCP pair and
+// wires a Dispatcher. The caller gets the raw server-side net.Conn so it can
+// act as the fake master, plus the client MasterConn and cleanup.
+func newMasterConnWithDispatcher(t *testing.T) (client *MasterConn, serverConn net.Conn, cleanup func()) {
+	t.Helper()
+
+	ln, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatalf("listen: %v", err)
+	}
+
+	var srvConn net.Conn
+	var acceptErr error
+	accepted := make(chan struct{})
+	go func() {
+		defer close(accepted)
+		srvConn, 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, []byte("go-slave"), []uint32{0x00010001, 0x00020001}, logger)
+	dispatcher := NewDispatcher(logger)
+	client.SetDispatcher(dispatcher)
+	client.Start()
+	serverConn = srvConn
+
+	cleanup = func() {
+		client.Close()
+		if serverConn != nil {
+			serverConn.Close()
+		}
+	}
+	return
+}
+
+// readMasterFrame reads a 12-byte metadata frame from the fake master side.
+func readMasterFrame(t *testing.T, conn net.Conn) *wire.Frame {
+	t.Helper()
+	if err := conn.SetReadDeadline(time.Now().Add(5 * time.Second)); err != nil {
+		t.Fatalf("set read deadline: %v", err)
+	}
+	frame, err := wire.ReadFrame(conn, 0)
+	if err != nil {
+		t.Fatalf("read frame: %v", err)
+	}
+	return frame
+}
+
+// writeMasterFrame writes a 12-byte metadata frame from the fake master side.
+func writeMasterFrame(t *testing.T, conn net.Conn, frame *wire.Frame) {
+	t.Helper()
+	if err := conn.SetWriteDeadline(time.Now().Add(5 * time.Second)); err != nil {
+		t.Fatalf("set write deadline: %v", err)
+	}
+	if err := wire.WriteFrame(conn, frame); err != nil {
+		t.Fatalf("write frame: %v", err)
+	}
+}
+
+// TestDispatcher_RouteToMasterConn verifies that frames with cluster_peer_id == 0
+// are handled by MasterConn itself (PING -> PONG).
+func TestDispatcher_RouteToMasterConn(t *testing.T) {
+	client, serverConn, cleanup := newMasterConnWithDispatcher(t)
+	defer cleanup()
+
+	pingPayload, err := serialize.SerializeToBytes(&wire.PingRequest{
+		ID:              []byte("master"),
+		FullShardIDList: []uint32{0x00010001},
+	})
+	if err != nil {
+		t.Fatalf("serialize ping: %v", err)
+	}
+
+	writeMasterFrame(t, serverConn, &wire.Frame{
+		Meta:    wire.ClusterMetadata{Branch: 0x00010001},
+		Opcode:  byte(wire.ClusterOpPing),
+		RPCID:   1,
+		Payload: pingPayload,
+	})
+
+	resp := readMasterFrame(t, serverConn)
+	if resp.Opcode != byte(wire.ClusterOpPong) {
+		t.Fatalf("expected PONG opcode 0x%x, got 0x%x", wire.ClusterOpPong, resp.Opcode)
+	}
+	if resp.RPCID != 1 {
+		t.Fatalf("expected rpc_id 1, got %d", resp.RPCID)
+	}
+	if resp.Meta.ClusterPeerID != 0 {
+		t.Fatalf("expected cluster_peer_id 0 for master-local response, got %d", resp.Meta.ClusterPeerID)
+	}
+
+	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) != "go-slave" {
+		t.Fatalf("pong id mismatch: got %q", pong.ID)
+	}
+
+	_ = client
+}
+
+// TestDispatcher_RouteToPeerConn verifies that frames with cluster_peer_id != 0
+// are forwarded to the matching virtual PeerConn and the stub response is sent
+// back through MasterConn.
+func TestDispatcher_RouteToPeerConn(t *testing.T) {
+	client, serverConn, cleanup := newMasterConnWithDispatcher(t)
+	defer cleanup()
+
+	const clusterPeerID uint64 = 7
+	const branch uint32 = 0x00010001
+
+	client.dispatcher.CreatePeerConns(clusterPeerID, []uint32{branch}, client, log.New())
+
+	reqPayload, err := serialize.SerializeToBytes(&wire.GetMinorBlockListRequest{
+		MinorBlockHashList: [][wire.HashLength]byte{},
+	})
+	if err != nil {
+		t.Fatalf("serialize request: %v", err)
+	}
+
+	writeMasterFrame(t, serverConn, &wire.Frame{
+		Meta:    wire.ClusterMetadata{Branch: branch, ClusterPeerID: clusterPeerID},
+		Opcode:  byte(wire.CommandOpGetMinorBlockListRequest),
+		RPCID:   3,
+		Payload: reqPayload,
+	})
+
+	resp := readMasterFrame(t, serverConn)
+	if resp.Opcode != byte(wire.CommandOpGetMinorBlockListResponse) {
+		t.Fatalf("expected response opcode 0x%x, got 0x%x", wire.CommandOpGetMinorBlockListResponse, resp.Opcode)
+	}
+	if resp.RPCID != 3 {
+		t.Fatalf("expected rpc_id 3, got %d", resp.RPCID)
+	}
+	if resp.Meta.Branch != branch || resp.Meta.ClusterPeerID != clusterPeerID {
+		t.Fatalf("metadata mismatch: got %+v, want branch=%d cluster_peer_id=%d", resp.Meta, branch, clusterPeerID)
+	}
+
+	var listResp wire.GetMinorBlockListResponse
+	if err := serialize.Deserialize(serialize.NewByteBuffer(resp.Payload), &listResp); err != nil {
+		t.Fatalf("deserialize response: %v", err)
+	}
+}
+
+// TestDispatcher_UnknownPeerDropped verifies that frames for an unregistered
+// cluster_peer_id are dropped and do not close MasterConn.
+func TestDispatcher_UnknownPeerDropped(t *testing.T) {
+	_, serverConn, cleanup := newMasterConnWithDispatcher(t)
+	defer cleanup()
+
+	reqPayload, err := serialize.SerializeToBytes(&wire.GetMinorBlockListRequest{
+		MinorBlockHashList: [][wire.HashLength]byte{},
+	})
+	if err != nil {
+		t.Fatalf("serialize request: %v", err)
+	}
+
+	// Unknown peer frame should be consumed and dropped.
+	writeMasterFrame(t, serverConn, &wire.Frame{
+		Meta:    wire.ClusterMetadata{Branch: 0x00010001, ClusterPeerID: 999},
+		Opcode:  byte(wire.CommandOpGetMinorBlockListRequest),
+		RPCID:   1,
+		Payload: reqPayload,
+	})
+
+	// The fake master should see no response for the dropped frame, but a
+	// subsequent master-local PING must still work.
+	if err := serverConn.SetReadDeadline(time.Now().Add(200 * time.Millisecond)); err != nil {
+		t.Fatalf("set read deadline: %v", err)
+	}
+	if _, err := wire.ReadFrame(serverConn, 0); err == nil {
+		t.Fatal("expected no response for unknown peer frame")
+	}
+
+	pingPayload, _ := serialize.SerializeToBytes(&wire.PingRequest{
+		ID:              []byte("master"),
+		FullShardIDList: []uint32{0x00010001},
+	})
+	writeMasterFrame(t, serverConn, &wire.Frame{
+		Meta:    wire.ClusterMetadata{Branch: 0x00010001},
+		Opcode:  byte(wire.ClusterOpPing),
+		RPCID:   2,
+		Payload: pingPayload,
+	})
+
+	resp := readMasterFrame(t, serverConn)
+	if resp.Opcode != byte(wire.ClusterOpPong) {
+		t.Fatalf("expected PONG after unknown peer drop, got opcode 0x%x", resp.Opcode)
+	}
+	if resp.RPCID != 2 {
+		t.Fatalf("expected rpc_id 2 in pong, got %d", resp.RPCID)
+	}
+}
+
+// TestPeerConn_RPCIDIsolation verifies that two PeerConns sharing a MasterConn
+// can use the same RPC ID without collision; responses are routed back to the
+// correct peer via metadata.
+func TestPeerConn_RPCIDIsolation(t *testing.T) {
+	client, serverConn, cleanup := newMasterConnWithDispatcher(t)
+	defer cleanup()
+
+	client.dispatcher.CreatePeerConns(7, []uint32{0x00010001}, client, log.New())
+	client.dispatcher.CreatePeerConns(9, []uint32{0x00020001}, client, log.New())
+
+	reqPayload, err := serialize.SerializeToBytes(&wire.GetMinorBlockListRequest{
+		MinorBlockHashList: [][wire.HashLength]byte{},
+	})
+	if err != nil {
+		t.Fatalf("serialize request: %v", err)
+	}
+
+	// Both peers use rpc_id=5.
+	writeMasterFrame(t, serverConn, &wire.Frame{
+		Meta:    wire.ClusterMetadata{Branch: 0x00010001, ClusterPeerID: 7},
+		Opcode:  byte(wire.CommandOpGetMinorBlockListRequest),
+		RPCID:   5,
+		Payload: reqPayload,
+	})
+	writeMasterFrame(t, serverConn, &wire.Frame{
+		Meta:    wire.ClusterMetadata{Branch: 0x00020001, ClusterPeerID: 9},
+		Opcode:  byte(wire.CommandOpGetMinorBlockListRequest),
+		RPCID:   5,
+		Payload: reqPayload,
+	})
+
+	resp1 := readMasterFrame(t, serverConn)
+	resp2 := readMasterFrame(t, serverConn)
+
+	if resp1.RPCID != 5 || resp2.RPCID != 5 {
+		t.Fatalf("expected both responses to have rpc_id 5, got %d and %d", resp1.RPCID, resp2.RPCID)
+	}
+
+	// Each response must belong to a distinct peer/branch pair.
+	peers := map[uint64]uint32{
+		resp1.Meta.ClusterPeerID: resp1.Meta.Branch,
+		resp2.Meta.ClusterPeerID: resp2.Meta.Branch,
+	}
+	if len(peers) != 2 {
+		t.Fatalf("responses were not routed to distinct peers: %+v", peers)
+	}
+	if peers[7] != 0x00010001 {
+		t.Fatalf("peer 7 response routed to wrong branch: got 0x%x", peers[7])
+	}
+	if peers[9] != 0x00020001 {
+		t.Fatalf("peer 9 response routed to wrong branch: got 0x%x", peers[9])
+	}
+
+	for _, resp := range []*wire.Frame{resp1, resp2} {
+		if resp.Opcode != byte(wire.CommandOpGetMinorBlockListResponse) {
+			t.Fatalf("expected GetMinorBlockListResponse, got opcode 0x%x", resp.Opcode)
+		}
+		var listResp wire.GetMinorBlockListResponse
+		if err := serialize.Deserialize(serialize.NewByteBuffer(resp.Payload), &listResp); err != nil {
+			t.Fatalf("deserialize response: %v", err)
+		}
+	}
+}
+
+// TestPeerConn_PeerHandlerRPCRoundTrip sends a CommandOp RPC through a virtual
+// PeerConn and verifies the stub response deserializes correctly.
+func TestPeerConn_PeerHandlerRPCRoundTrip(t *testing.T) {
+	client, serverConn, cleanup := newMasterConnWithDispatcher(t)
+	defer cleanup()
+
+	const clusterPeerID uint64 = 11
+	const branch uint32 = 0x00010001
+
+	client.dispatcher.CreatePeerConns(clusterPeerID, []uint32{branch}, client, log.New())
+
+	reqPayload, err := serialize.SerializeToBytes(&wire.GetMinorBlockHeaderListRequest{
+		Branch:    branch,
+		BlockHash: [wire.HashLength]byte{},
+		Limit:     10,
+		Direction: 0,
+	})
+	if err != nil {
+		t.Fatalf("serialize request: %v", err)
+	}
+
+	writeMasterFrame(t, serverConn, &wire.Frame{
+		Meta:    wire.ClusterMetadata{Branch: branch, ClusterPeerID: clusterPeerID},
+		Opcode:  byte(wire.CommandOpGetMinorBlockHeaderListRequest),
+		RPCID:   1,
+		Payload: reqPayload,
+	})
+
+	resp := readMasterFrame(t, serverConn)
+	if resp.Opcode != byte(wire.CommandOpGetMinorBlockHeaderListResponse) {
+		t.Fatalf("expected response opcode 0x%x, got 0x%x", wire.CommandOpGetMinorBlockHeaderListResponse, resp.Opcode)
+	}
+	if resp.RPCID != 1 {
+		t.Fatalf("expected rpc_id 1, got %d", resp.RPCID)
+	}
+
+	var headerResp wire.GetMinorBlockHeaderListResponse
+	if err := serialize.Deserialize(serialize.NewByteBuffer(resp.Payload), &headerResp); err != nil {
+		t.Fatalf("deserialize response: %v", err)
+	}
+}
+
+// TestMasterConn_CreateDestroyPeerConnection verifies that the master commands
+// create and destroy virtual peer connections through the dispatcher.
+func TestMasterConn_CreateDestroyPeerConnection(t *testing.T) {
+	client, serverConn, cleanup := newMasterConnWithDispatcher(t)
+	defer cleanup()
+
+	const clusterPeerID uint64 = 21
+
+	createPayload, err := serialize.SerializeToBytes(&wire.CreateClusterPeerConnectionRequest{ClusterPeerID: clusterPeerID})
+	if err != nil {
+		t.Fatalf("serialize create request: %v", err)
+	}
+
+	writeMasterFrame(t, serverConn, &wire.Frame{
+		Meta:    wire.ClusterMetadata{Branch: 0x00010001},
+		Opcode:  byte(wire.ClusterOpCreateClusterPeerConnectionRequest),
+		RPCID:   1,
+		Payload: createPayload,
+	})
+
+	resp := readMasterFrame(t, serverConn)
+	if resp.Opcode != byte(wire.ClusterOpCreateClusterPeerConnectionResponse) {
+		t.Fatalf("expected create response opcode 0x%x, got 0x%x", wire.ClusterOpCreateClusterPeerConnectionResponse, resp.Opcode)
+	}
+	var createResp wire.CreateClusterPeerConnectionResponse
+	if err := serialize.Deserialize(serialize.NewByteBuffer(resp.Payload), &createResp); err != nil {
+		t.Fatalf("deserialize create response: %v", err)
+	}
+	if createResp.ErrorCode != 0 {
+		t.Fatalf("expected error_code 0, got %d", createResp.ErrorCode)
+	}
+
+	// The dispatcher should now hold one PeerConn per local shard.
+	branchMap := client.dispatcher.peers[clusterPeerID]
+	if len(branchMap) != len(client.localFullShardIDList) {
+		t.Fatalf("expected %d peer conns, got %d", len(client.localFullShardIDList), len(branchMap))
+	}
+	for _, branch := range client.localFullShardIDList {
+		if branchMap[branch] == nil {
+			t.Fatalf("missing peer conn for branch 0x%x", branch)
+		}
+		if branchMap[branch].IsClosed() {
+			t.Fatalf("peer conn for branch 0x%x is already closed", branch)
+		}
+	}
+
+	// Destroy the peer connections.
+	destroyPayload, err := serialize.SerializeToBytes(&wire.DestroyClusterPeerConnectionCommand{ClusterPeerID: clusterPeerID})
+	if err != nil {
+		t.Fatalf("serialize destroy command: %v", err)
+	}
+	writeMasterFrame(t, serverConn, &wire.Frame{
+		Meta:    wire.ClusterMetadata{Branch: 0x00010001},
+		Opcode:  byte(wire.ClusterOpDestroyClusterPeerConnectionCommand),
+		RPCID:   0, // non-RPC
+		Payload: destroyPayload,
+	})
+
+	// Give the close goroutines a moment to finish.
+	time.Sleep(50 * time.Millisecond)
+
+	for _, branch := range client.localFullShardIDList {
+		if pc := client.dispatcher.peers[clusterPeerID][branch]; pc != nil && !pc.IsClosed() {
+			t.Fatalf("peer conn for branch 0x%x was not closed after destroy", branch)
+		}
+	}
+
+	// MasterConn must still be alive for a follow-up PING.
+	pingPayload, _ := serialize.SerializeToBytes(&wire.PingRequest{
+		ID:              []byte("master"),
+		FullShardIDList: []uint32{0x00010001},
+	})
+	writeMasterFrame(t, serverConn, &wire.Frame{
+		Meta:    wire.ClusterMetadata{Branch: 0x00010001},
+		Opcode:  byte(wire.ClusterOpPing),
+		RPCID:   2,
+		Payload: pingPayload,
+	})
+
+	pingResp := readMasterFrame(t, serverConn)
+	if pingResp.Opcode != byte(wire.ClusterOpPong) {
+		t.Fatalf("expected PONG after destroy, got opcode 0x%x", pingResp.Opcode)
+	}
+}
+
+// TestMasterConn_CloseClosesPeerConns verifies that closing MasterConn closes
+// all associated PeerConns and clears the dispatcher registry.
+func TestMasterConn_CloseClosesPeerConns(t *testing.T) {
+	client, _, cleanup := newMasterConnWithDispatcher(t)
+	defer cleanup()
+
+	client.dispatcher.CreatePeerConns(7, []uint32{0x00010001, 0x00020001}, client, log.New())
+	client.dispatcher.CreatePeerConns(9, []uint32{0x00010001}, client, log.New())
+
+	// Keep references before Close clears the map.
+	var peerConns []*PeerConn
+	for _, branchMap := range client.dispatcher.peers {
+		for _, pc := range branchMap {
+			peerConns = append(peerConns, pc)
+		}
+	}
+	if len(peerConns) != 3 {
+		t.Fatalf("expected 3 peer conns, got %d", len(peerConns))
+	}
+
+	client.Close()
+
+	for _, pc := range peerConns {
+		if !pc.IsClosed() {
+			t.Fatalf("peer conn %d/%d was not closed by MasterConn.Close", pc.ClusterPeerID(), pc.Branch())
+		}
+	}
+
+	if len(client.dispatcher.peers) != 0 {
+		t.Fatalf("dispatcher registry not cleared: got %d cluster_peer_id entries", len(client.dispatcher.peers))
+	}
+}
+
+// TestPeerConn_OutboundRPCThroughMasterConn verifies that a PeerConn can issue
+// an outbound RPC and the request is written to the underlying MasterConn with
+// the correct cluster_peer_id metadata.
+func TestPeerConn_OutboundRPCThroughMasterConn(t *testing.T) {
+	client, serverConn, cleanup := newMasterConnWithDispatcher(t)
+	defer cleanup()
+
+	const clusterPeerID uint64 = 31
+	const branch uint32 = 0x00010001
+
+	client.dispatcher.CreatePeerConns(clusterPeerID, []uint32{branch}, client, log.New())
+	pc := client.dispatcher.peers[clusterPeerID][branch]
+
+	req := &wire.GetMinorBlockListRequest{MinorBlockHashList: [][wire.HashLength]byte{}}
+	reqPayload, err := serialize.SerializeToBytes(req)
+	if err != nil {
+		t.Fatalf("serialize request: %v", err)
+	}
+
+	// Echo the request back as a response from the fake master.
+	go func() {
+		frame := readMasterFrame(t, serverConn)
+		if frame.Meta.ClusterPeerID != clusterPeerID {
+			t.Errorf("outbound request cluster_peer_id mismatch: got %d, want %d", frame.Meta.ClusterPeerID, clusterPeerID)
+		}
+		if frame.Meta.Branch != branch {
+			t.Errorf("outbound request branch mismatch: got 0x%x, want 0x%x", frame.Meta.Branch, branch)
+		}
+		writeMasterFrame(t, serverConn, &wire.Frame{
+			Meta:    frame.Meta,
+			Opcode:  frame.Opcode + 1,
+			RPCID:   frame.RPCID,
+			Payload: frame.Payload,
+		})
+	}()
+
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	resp, err := pc.SendRPCMeta(ctx, byte(wire.CommandOpGetMinorBlockListRequest), reqPayload, wire.ClusterMetadata{})
+	if err != nil {
+		t.Fatalf("peer conn SendRPCMeta: %v", err)
+	}
+	if resp.Opcode != byte(wire.CommandOpGetMinorBlockListResponse) {
+		t.Fatalf("expected response opcode 0x%x, got 0x%x", wire.CommandOpGetMinorBlockListResponse, resp.Opcode)
+	}
+}
+
+// ── Additional tests ─────────────────────────────────────────────────────────
+
+// TestDispatcher_DuplicateCreatePeerConn verifies that a duplicate create
+// request for the same cluster_peer_id and branch does not replace the existing
+// PeerConn. This matches Python's behavior of logging an error and skipping.
+// Python: slave.py#L335-L341
+func TestDispatcher_DuplicateCreatePeerConn(t *testing.T) {
+	client, serverConn, cleanup := newMasterConnWithDispatcher(t)
+	defer cleanup()
+
+	const clusterPeerID uint64 = 41
+	const branch uint32 = 0x00010001
+
+	// First create.
+	client.dispatcher.CreatePeerConns(clusterPeerID, []uint32{branch}, client, log.New())
+	original := client.dispatcher.peers[clusterPeerID][branch]
+	if original == nil {
+		t.Fatal("expected peer conn after first create")
+	}
+
+	// Duplicate create — should not replace the existing PeerConn.
+	client.dispatcher.CreatePeerConns(clusterPeerID, []uint32{branch}, client, log.New())
+	afterDup := client.dispatcher.peers[clusterPeerID][branch]
+	if afterDup != original {
+		t.Fatal("duplicate create replaced the existing PeerConn")
+	}
+
+	// The branch map should still have exactly one entry.
+	if len(client.dispatcher.peers[clusterPeerID]) != 1 {
+		t.Fatalf("expected 1 branch entry, got %d", len(client.dispatcher.peers[clusterPeerID]))
+	}
+
+	// MasterConn must still be alive.
+	pingPayload, _ := serialize.SerializeToBytes(&wire.PingRequest{
+		ID:              []byte("master"),
+		FullShardIDList: []uint32{0x00010001},
+	})
+	writeMasterFrame(t, serverConn, &wire.Frame{
+		Meta:    wire.ClusterMetadata{Branch: 0x00010001},
+		Opcode:  byte(wire.ClusterOpPing),
+		RPCID:   1,
+		Payload: pingPayload,
+	})
+	resp := readMasterFrame(t, serverConn)
+	if resp.Opcode != byte(wire.ClusterOpPong) {
+		t.Fatalf("expected PONG, got opcode 0x%x", resp.Opcode)
+	}
+}
+
+// TestDispatcher_NonRPCCommandRouted verifies that fire-and-forget (non-RPC)
+// commands are routed through the Dispatcher to the correct PeerConn.
+// Python: shard.py OP_NONRPC_MAP (L275-L279)
+func TestDispatcher_NonRPCCommandRouted(t *testing.T) {
+	client, serverConn, cleanup := newMasterConnWithDispatcher(t)
+	defer cleanup()
+
+	const clusterPeerID uint64 = 42
+	const branch uint32 = 0x00010001
+
+	client.dispatcher.CreatePeerConns(clusterPeerID, []uint32{branch}, client, log.New())
+
+	cmdPayload, err := serialize.SerializeToBytes(&wire.NewMinorBlockHeaderListCommand{
+		RootBlockHeader:      nil,
+		MinorBlockHeaderList: nil,
+	})
+	if err != nil {
+		t.Fatalf("serialize command: %v", err)
+	}
+
+	// Send a non-RPC command (rpc_id must be 0).
+	writeMasterFrame(t, serverConn, &wire.Frame{
+		Meta:    wire.ClusterMetadata{Branch: branch, ClusterPeerID: clusterPeerID},
+		Opcode:  byte(wire.CommandOpNewMinorBlockHeaderList),
+		RPCID:   0,
+		Payload: cmdPayload,
+	})
+
+	// Non-RPC commands produce no response. Verify no frame is sent back.
+	if err := serverConn.SetReadDeadline(time.Now().Add(200 * time.Millisecond)); err != nil {
+		t.Fatalf("set read deadline: %v", err)
+	}
+	if _, err := wire.ReadFrame(serverConn, 0); err == nil {
+		t.Fatal("expected no response for non-RPC command")
+	}
+
+	// MasterConn must still be alive — a follow-up PING should work.
+	pingPayload, _ := serialize.SerializeToBytes(&wire.PingRequest{
+		ID:              []byte("master"),
+		FullShardIDList: []uint32{0x00010001},
+	})
+	writeMasterFrame(t, serverConn, &wire.Frame{
+		Meta:    wire.ClusterMetadata{Branch: 0x00010001},
+		Opcode:  byte(wire.ClusterOpPing),
+		RPCID:   1,
+		Payload: pingPayload,
+	})
+	resp := readMasterFrame(t, serverConn)
+	if resp.Opcode != byte(wire.ClusterOpPong) {
+		t.Fatalf("expected PONG after non-RPC command, got opcode 0x%x", resp.Opcode)
+	}
+}
+
+// TestPeerConn_CloseStopsReadLoop verifies that closing a PeerConn causes its
+// read loop to exit (no goroutine leak). After Close(), the read loop's
+// deferred Close should be a no-op and the closed channel should be signaled.
+func TestPeerConn_CloseStopsReadLoop(t *testing.T) {
+	client, _, cleanup := newMasterConnWithDispatcher(t)
+	defer cleanup()
+
+	const clusterPeerID uint64 = 43
+	const branch uint32 = 0x00010001
+
+	client.dispatcher.CreatePeerConns(clusterPeerID, []uint32{branch}, client, log.New())
+	pc := client.dispatcher.peers[clusterPeerID][branch]
+
+	// Verify the PeerConn is active and its read loop is running.
+	if !pc.IsActive() {
+		t.Fatal("expected PeerConn to be active after Start")
+	}
+
+	// Close the PeerConn.
+	pc.Close()
+
+	// The closed channel should be signaled promptly.
+	select {
+	case <-pc.WaitUntilClosed():
+		// OK
+	case <-time.After(2 * time.Second):
+		t.Fatal("PeerConn read loop did not exit after Close")
+	}
+
+	// HandleFrame after close should return an error.
+	if err := pc.HandleFrame(&wire.Frame{}); err == nil {
+		t.Fatal("expected error from HandleFrame after Close")
+	}
+}
+
+// TestDispatcher_DestroyNonexistentPeer verifies that destroying a non-existent
+// cluster_peer_id is a no-op and does not affect MasterConn.
+// Python: slave.py#L321-L327 (pop with default None)
+func TestDispatcher_DestroyNonexistentPeer(t *testing.T) {
+	client, serverConn, cleanup := newMasterConnWithDispatcher(t)
+	defer cleanup()
+
+	// Destroy a cluster_peer_id that was never created.
+	client.dispatcher.DestroyPeerConns(9999)
+
+	// MasterConn must still be alive.
+	pingPayload, _ := serialize.SerializeToBytes(&wire.PingRequest{
+		ID:              []byte("master"),
+		FullShardIDList: []uint32{0x00010001},
+	})
+	writeMasterFrame(t, serverConn, &wire.Frame{
+		Meta:    wire.ClusterMetadata{Branch: 0x00010001},
+		Opcode:  byte(wire.ClusterOpPing),
+		RPCID:   1,
+		Payload: pingPayload,
+	})
+	resp := readMasterFrame(t, serverConn)
+	if resp.Opcode != byte(wire.ClusterOpPong) {
+		t.Fatalf("expected PONG after destroying nonexistent peer, got opcode 0x%x", resp.Opcode)
+	}
+
+	// Destroy the same ID again — should still be idempotent.
+	client.dispatcher.DestroyPeerConns(9999)
+
+	writeMasterFrame(t, serverConn, &wire.Frame{
+		Meta:    wire.ClusterMetadata{Branch: 0x00010001},
+		Opcode:  byte(wire.ClusterOpPing),
+		RPCID:   2,
+		Payload: pingPayload,
+	})
+	resp = readMasterFrame(t, serverConn)
+	if resp.Opcode != byte(wire.ClusterOpPong) {
+		t.Fatalf("expected PONG after second destroy, got opcode 0x%x", resp.Opcode)
+	}
+}
diff --git a/qkc/cluster/slave/testdata/pyproto/peer_master.py b/qkc/cluster/slave/testdata/pyproto/peer_master.py
new file mode 100644
index 000000000000..55eacd58668b
--- /dev/null
+++ b/qkc/cluster/slave/testdata/pyproto/peer_master.py
@@ -0,0 +1,245 @@
+#!/usr/bin/env python3
+"""Python Master protocol peer for PeerConn interoperability tests.
+
+This peer simulates a Python Master that:
+  1. Accepts a Go Slave connection (MasterConn)
+  2. Performs PING/PONG handshake
+  3. Sends CreateClusterPeerConnectionRequest to create a PeerConn on the Go side
+  4. Sends peer traffic (CommandOp frames with cluster_peer_id != 0) through
+     the MasterConn transport, simulating forwarded external peer traffic
+  5. Validates that the Go PeerConn correctly handles the traffic
+  6. Sends DestroyClusterPeerConnectionCommand to tear down the PeerConn
+  7. Verifies the connection is still alive after destroy
+
+Wire format for master frames:
+  [4B payload_len][4B branch][8B cluster_peer_id][1B opcode][8B rpc_id][payload]
+
+Usage:
+    python3 peer_master.py --port 0 --id "py-master" --shards "1,2" --cluster-peer-id 42
+
+Output:
+    PORT:<port>
+    PONG_OK id=<hex>
+    CREATE_OK error_code=<n>
+    PEER_RPC_OK opcode=0x0a rpc_id=<n>
+    PEER_NONRPC_OK
+    DESTROY_OK
+    POST_DESTROY_PONG_OK id=<hex>
+    DISCONNECTED
+"""
+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 opcodes (master <-> slave)
+CLUSTER_OP_PING = 1 + CLUSTER_OP_BASE
+CLUSTER_OP_PONG = 2 + CLUSTER_OP_BASE
+CLUSTER_OP_CREATE_CLUSTER_PEER_CONNECTION_REQUEST = 25 + CLUSTER_OP_BASE
+CLUSTER_OP_CREATE_CLUSTER_PEER_CONNECTION_RESPONSE = 26 + CLUSTER_OP_BASE
+CLUSTER_OP_DESTROY_CLUSTER_PEER_CONNECTION_COMMAND = 27 + CLUSTER_OP_BASE
+
+# Command opcodes (peer <-> peer, tunneled through master)
+COMMAND_OP_GET_MINOR_BLOCK_LIST_REQUEST = 0x09
+COMMAND_OP_GET_MINOR_BLOCK_LIST_RESPONSE = 0x0A
+COMMAND_OP_NEW_MINOR_BLOCK_HEADER_LIST = 0x01
+
+
+def serialize_create_peer_connection_request(cluster_peer_id):
+    """Serialize CreateClusterPeerConnectionRequest.
+
+    Fields:
+      ClusterPeerID: uint64 (8 bytes BE)
+    """
+    return struct.pack('>Q', cluster_peer_id)
+
+
+def serialize_destroy_peer_connection_command(cluster_peer_id):
+    """Serialize DestroyClusterPeerConnectionCommand.
+
+    Fields:
+      ClusterPeerID: uint64 (8 bytes BE)
+    """
+    return struct.pack('>Q', cluster_peer_id)
+
+
+def serialize_get_minor_block_list_request():
+    """Serialize GetMinorBlockListRequest.
+
+    Fields:
+      MinorBlockHashList: [][32]byte (4B count + hashes)
+    """
+    # Empty hash list
+    return struct.pack('>I', 0)
+
+
+def serialize_new_minor_block_header_list_command():
+    """Serialize NewMinorBlockHeaderListCommand.
+
+    Fields:
+      RootBlockHeader: *RawBytes (nil marker 0x00)
+      MinorBlockHeaderList: []*RawBytes (4B count + items)
+    """
+    data = b'\x00'  # RootBlockHeader: nil
+    data += struct.pack('>I', 0)  # empty list
+    return data
+
+
+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)
+    parser.add_argument('--cluster-peer-id', type=int, required=True)
+    args = parser.parse_args()
+
+    master_id = args.id.encode('utf-8')
+    shard_list = [int(s) for s in args.shards.split(',')]
+    cluster_peer_id = args.cluster_peer_id
+    branch = 0x00010001  # shard_id=1, chain_size=1
+
+    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 (verify MasterConn is alive)
+        _send_ping(conn, master_id, shard_list)
+
+        # 2. CreateClusterPeerConnectionRequest -> Response
+        create_payload = serialize_create_peer_connection_request(cluster_peer_id)
+        write_master_frame(
+            conn,
+            CLUSTER_OP_CREATE_CLUSTER_PEER_CONNECTION_REQUEST,
+            2,  # rpc_id
+            create_payload,
+            branch=branch,
+        )
+
+        frame = read_master_frame(conn)
+        if frame is None:
+            print("ERROR: no response for create peer connection", flush=True)
+            sys.exit(1)
+
+        if frame['opcode'] != CLUSTER_OP_CREATE_CLUSTER_PEER_CONNECTION_RESPONSE:
+            print(f"ERROR: expected CREATE_RESPONSE(0x{CLUSTER_OP_CREATE_CLUSTER_PEER_CONNECTION_RESPONSE:02x}), "
+                  f"got 0x{frame['opcode']:02x}", flush=True)
+            sys.exit(1)
+
+        if frame['rpc_id'] != 2:
+            print(f"ERROR: expected rpc_id 2, got {frame['rpc_id']}", flush=True)
+            sys.exit(1)
+
+        error_code = struct.unpack('>I', frame['payload'][0:4])[0]
+        print(f"CREATE_OK error_code={error_code}", flush=True)
+
+        # 3. Send peer RPC traffic (cluster_peer_id != 0)
+        #    GetMinorBlockListRequest -> GetMinorBlockListResponse
+        peer_rpc_payload = serialize_get_minor_block_list_request()
+        write_master_frame(
+            conn,
+            COMMAND_OP_GET_MINOR_BLOCK_LIST_REQUEST,
+            100,  # peer rpc_id
+            peer_rpc_payload,
+            branch=branch,
+            cluster_peer_id=cluster_peer_id,
+        )
+
+        frame = read_master_frame(conn)
+        if frame is None:
+            print("ERROR: no response for peer RPC", flush=True)
+            sys.exit(1)
+
+        if frame['opcode'] != COMMAND_OP_GET_MINOR_BLOCK_LIST_RESPONSE:
+            print(f"ERROR: expected peer response 0x{COMMAND_OP_GET_MINOR_BLOCK_LIST_RESPONSE:02x}, "
+                  f"got 0x{frame['opcode']:02x}", flush=True)
+            sys.exit(1)
+
+        if frame['rpc_id'] != 100:
+            print(f"ERROR: expected peer rpc_id 100, got {frame['rpc_id']}", flush=True)
+            sys.exit(1)
+
+        # Verify response metadata has correct cluster_peer_id
+        if frame['cluster_peer_id'] != cluster_peer_id:
+            print(f"ERROR: expected response cluster_peer_id {cluster_peer_id}, "
+                  f"got {frame['cluster_peer_id']}", flush=True)
+            sys.exit(1)
+
+        print(f"PEER_RPC_OK opcode=0x{frame['opcode']:02x} rpc_id={frame['rpc_id']}", flush=True)
+
+        # 4. Send peer non-RPC traffic (fire-and-forget)
+        nonrpc_payload = serialize_new_minor_block_header_list_command()
+        write_master_frame(
+            conn,
+            COMMAND_OP_NEW_MINOR_BLOCK_HEADER_LIST,
+            0,  # non-RPC: rpc_id must be 0
+            nonrpc_payload,
+            branch=branch,
+            cluster_peer_id=cluster_peer_id,
+        )
+
+        # Non-RPC produces no response. Verify by sending a follow-up PING.
+        _send_ping(conn, master_id, shard_list, rpc_id=3)
+        print("PEER_NONRPC_OK", flush=True)
+
+        # 5. DestroyClusterPeerConnectionCommand (fire-and-forget)
+        destroy_payload = serialize_destroy_peer_connection_command(cluster_peer_id)
+        write_master_frame(
+            conn,
+            CLUSTER_OP_DESTROY_CLUSTER_PEER_CONNECTION_COMMAND,
+            0,  # non-RPC
+            destroy_payload,
+            branch=branch,
+        )
+        print("DESTROY_OK", flush=True)
+
+        # 6. Verify MasterConn is still alive after destroy
+        _send_ping(conn, master_id, shard_list, rpc_id=4)
+
+    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, rpc_id=1):
+    """Send PING, wait for PONG, validate and print result."""
+    ping_payload = serialize_ping_request(master_id, shard_list)
+    write_master_frame(conn, CLUSTER_OP_PING, rpc_id, ping_payload)
+
+    frame = read_master_frame(conn)
+    if frame is None:
+        print(f"ERROR: no pong received for rpc_id={rpc_id}", 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'] != rpc_id:
+        print(f"ERROR: expected rpc_id {rpc_id}, got {frame['rpc_id']}", flush=True)
+        sys.exit(1)
+
+    peer_id_recv, _ = parse_pong_response(frame['payload'])
+    if rpc_id == 1:
+        print(f"PONG_OK id={peer_id_recv.hex()}", flush=True)
+    else:
+        print(f"POST_DESTROY_PONG_OK id={peer_id_recv.hex()}", flush=True)
+
+
+if __name__ == '__main__':
+    main()

From d38744d783de89792614a497551a1399c6d1fca8 Mon Sep 17 00:00:00 2001
From: iteye <iteyelmp@gmail.com>
Date: Thu, 16 Jul 2026 17:58:28 +0800
Subject: [PATCH 2/4] fix review bug

---
 qkc/cluster/slave/master_conn.go    |  8 +++++---
 qkc/cluster/slave/peer_conn.go      |  9 +++++++--
 qkc/cluster/slave/peer_conn_test.go | 15 ++++++++-------
 3 files changed, 20 insertions(+), 12 deletions(-)

diff --git a/qkc/cluster/slave/master_conn.go b/qkc/cluster/slave/master_conn.go
index 1fe77c148704..097cacb3bbf0 100644
--- a/qkc/cluster/slave/master_conn.go
+++ b/qkc/cluster/slave/master_conn.go
@@ -273,13 +273,15 @@ func (mc *MasterConn) handlePing(req any) (any, error) {
 	}, nil
 }
 
-// handleCreateClusterPeerConnection creates virtual peer connections for all
-// shards. Until PR7's Shard Registry is available, localFullShardIDList is used
-// as the shard list.
+// handleCreateClusterPeerConnection creates virtual peer connections for all shards.
 // Python: returns CreateClusterPeerConnectionResponse(error_code=0) on success.
 func (mc *MasterConn) handleCreateClusterPeerConnection(req any) (any, error) {
 	r := req.(*wire.CreateClusterPeerConnectionRequest)
 
+	// TODO: localFullShardIDList is a temporary stand-in for the real shard
+	// registry. Once the shard registry is available, replace with the actual
+	// per-shard branch list from the registry.
+
 	mc.dispatcherMu.RLock()
 	d := mc.dispatcher
 	mc.dispatcherMu.RUnlock()
diff --git a/qkc/cluster/slave/peer_conn.go b/qkc/cluster/slave/peer_conn.go
index c224ceb9bff9..5173e783179b 100644
--- a/qkc/cluster/slave/peer_conn.go
+++ b/qkc/cluster/slave/peer_conn.go
@@ -141,10 +141,15 @@ func (pc *PeerConn) registerOpSerializers() {
 	})
 }
 
-// registerHandlers registers the shard-level peer handlers. These are stubs
-// because PR6 does not implement shard runtime / block processing.
+// registerHandlers registers the shard-level peer handlers. These are stubs;
+// real implementations require the shard runtime to be ported.
 func (pc *PeerConn) registerHandlers() {
 	pc.rpcConn.RegisterTypedHandlers(map[byte]TypedHandler{
+		// ── Migration stubs ─────────────────────────────────────────────
+		// These handlers exist only to preserve protocol compatibility.
+		// Real implementations must be added outside the connection layer.
+		// After migration, remove these stub registrations and handlers.
+
 		// Non-RPC commands (fire-and-forget).
 		byte(wire.CommandOpNewMinorBlockHeaderList): pc.handleNewMinorBlockHeaderList,
 		byte(wire.CommandOpNewTransactionList):      pc.handleNewTransactionList,
diff --git a/qkc/cluster/slave/peer_conn_test.go b/qkc/cluster/slave/peer_conn_test.go
index d2224e9034f2..73c118f28253 100644
--- a/qkc/cluster/slave/peer_conn_test.go
+++ b/qkc/cluster/slave/peer_conn_test.go
@@ -382,14 +382,15 @@ func TestMasterConn_CreateDestroyPeerConnection(t *testing.T) {
 		Payload: destroyPayload,
 	})
 
-	// Give the close goroutines a moment to finish.
-	time.Sleep(50 * time.Millisecond)
-
-	for _, branch := range client.localFullShardIDList {
-		if pc := client.dispatcher.peers[clusterPeerID][branch]; pc != nil && !pc.IsClosed() {
-			t.Fatalf("peer conn for branch 0x%x was not closed after destroy", branch)
+	// Wait for peer connections to be closed (async since handler runs in goroutine).
+	waitForCondition(t, 2*time.Second, func() bool {
+		for _, branch := range client.localFullShardIDList {
+			if pc := client.dispatcher.peers[clusterPeerID][branch]; pc != nil && !pc.IsClosed() {
+				return false
+			}
 		}
-	}
+		return true
+	})
 
 	// MasterConn must still be alive for a follow-up PING.
 	pingPayload, _ := serialize.SerializeToBytes(&wire.PingRequest{

From 21424374ab9069c34c7081df3e6243535e07fd04 Mon Sep 17 00:00:00 2001
From: iteye <iteyelmp@gmail.com>
Date: Mon, 20 Jul 2026 10:24:44 +0800
Subject: [PATCH 3/4] fix bugs

---
 qkc/cluster/slave/master_conn.go | 32 ++++++++++++++++++++++++++++----
 1 file changed, 28 insertions(+), 4 deletions(-)

diff --git a/qkc/cluster/slave/master_conn.go b/qkc/cluster/slave/master_conn.go
index 097cacb3bbf0..307093c53e0e 100644
--- a/qkc/cluster/slave/master_conn.go
+++ b/qkc/cluster/slave/master_conn.go
@@ -7,6 +7,7 @@ import (
 	"fmt"
 	"io"
 	"net"
+	"slices"
 	"sync"
 
 	"github.com/ethereum/go-ethereum/log"
@@ -264,8 +265,11 @@ func (mc *MasterConn) LocalFullShardIDList() []uint32 {
 // 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)
+	ping := req.(*wire.PingRequest)
+
+	if ping.RootTip != nil {
+		// TODO: create/update shard runtime from root tip. when core.RootBlock is ported, use ping.root_tip to drive shard creation.
+	}
 
 	return &wire.PongResponse{
 		ID:              append([]byte(nil), mc.localID...),
@@ -590,12 +594,32 @@ func (mc *MasterConn) ForwardFrame(f *wire.Frame) error {
 }
 
 // SetDispatcher wires the dispatcher that routes peer traffic. It also installs
-// the dispatcher as the raw-frame forwarder on this MasterConn.
+// a forwarder that validates the branch before routing.
+// Python: MasterConnection.get_connection_to_forward() closes the connection if
+// the branch is not in the configured full_shard_id_list.
 func (mc *MasterConn) SetDispatcher(d *Dispatcher) {
 	mc.dispatcherMu.Lock()
 	mc.dispatcher = d
 	mc.dispatcherMu.Unlock()
-	mc.SetForwarder(d.RouteFrame)
+
+	// Create a wrapper forwarder that validates branch before routing.
+	// Python: MasterConnection.get_connection_to_forward() only validates
+	// branch when cluster_peer_id != 0 (i.e., on the forwarding path).
+	// Master-local RPCs (cluster_peer_id == 0) use branch=0 and must not
+	// be validated.
+	mc.SetForwarder(func(frame *wire.Frame) bool {
+		if frame.Meta.ClusterPeerID != 0 && !mc.isValidBranch(frame.Meta.Branch) {
+			mc.log.Error("incorrect forwarding branch", "branch", frame.Meta.Branch)
+			mc.Close()
+			return true
+		}
+		return d.RouteFrame(frame)
+	})
+}
+
+// isValidBranch checks if the given branch is in the local full shard ID list.
+func (mc *MasterConn) isValidBranch(branch uint32) bool {
+	return slices.Contains(mc.localFullShardIDList, branch)
 }
 
 // Close closes the master connection and all associated peer connections.

From 6c432889e938d6fb592f5316ad078e7799055a55 Mon Sep 17 00:00:00 2001
From: iteye <iteyelmp@gmail.com>
Date: Tue, 21 Jul 2026 16:54:06 +0800
Subject: [PATCH 4/4] fix comment

---
 qkc/cluster/slave/master_conn.go              |  31 +--
 qkc/cluster/slave/peer_conn.go                |  10 +-
 qkc/cluster/slave/peer_conn_test.go           |  15 +-
 .../slave/testdata/pyproto/peer_master.py     | 245 ------------------
 4 files changed, 26 insertions(+), 275 deletions(-)
 delete mode 100644 qkc/cluster/slave/testdata/pyproto/peer_master.py

diff --git a/qkc/cluster/slave/master_conn.go b/qkc/cluster/slave/master_conn.go
index 5380f1bf203a..c5bdb2fdd742 100644
--- a/qkc/cluster/slave/master_conn.go
+++ b/qkc/cluster/slave/master_conn.go
@@ -8,7 +8,6 @@ import (
 	"io"
 	"net"
 	"slices"
-	"sync"
 
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/qkc/cluster/wire"
@@ -33,16 +32,16 @@ type MasterConn struct {
 	localFullShardIDList []uint32
 
 	// dispatcher routes peer traffic (cluster_peer_id != 0) to virtual PeerConns.
-	// It is nil until wired by SetDispatcher.
-	dispatcher   *Dispatcher
-	dispatcherMu sync.RWMutex
+	// It is nil until wired by SetDispatcher. Once Start() is called, it is
+	// immutable — handlers and Close() read it without synchronization.
+	dispatcher *Dispatcher
 
 	// peerRPCIDs tracks the most recent inbound RPC ID per cluster_peer_id.
 	// cluster_peer_id == 0 is the master itself; each non-zero peer has its
 	// own independent monotonic sequence so PeerConns sharing this MasterConn
 	// do not collide on rpc_id.
-	peerRPCIDs   map[uint64]int64
-	peerRPCIDsMu sync.Mutex
+	// Only accessed by readLoop; no lock needed.
+	peerRPCIDs map[uint64]int64
 }
 
 // NewMasterConn dials the master at addr and returns a MasterConn.
@@ -73,7 +72,7 @@ func newMasterConn(conn net.Conn, maxPayloadSize uint32, localID []byte, localFu
 		peerRPCIDs:           make(map[uint64]int64),
 	}
 
-	mc.rpcConn.validateRPCID = mc.validatePeerRPCID
+	mc.baseConn.validateRPCID = mc.validatePeerRPCID
 
 	mc.registerOpSerializers()
 	mc.registerHandlers()
@@ -238,9 +237,6 @@ func emptyRawBytes() *wire.RawBytes {
 // cluster_peer_id. This lets multiple PeerConns share one MasterConn without
 // colliding on rpc_id.
 func (mc *MasterConn) validatePeerRPCID(clusterPeerID uint64, rpcID uint64) bool {
-	mc.peerRPCIDsMu.Lock()
-	defer mc.peerRPCIDsMu.Unlock()
-
 	last, ok := mc.peerRPCIDs[clusterPeerID]
 	if !ok {
 		last = -1
@@ -286,11 +282,9 @@ func (mc *MasterConn) handleCreateClusterPeerConnection(req any) (any, error) {
 	// registry. Once the shard registry is available, replace with the actual
 	// per-shard branch list from the registry.
 
-	mc.dispatcherMu.RLock()
 	d := mc.dispatcher
-	mc.dispatcherMu.RUnlock()
 	if d != nil {
-		d.CreatePeerConns(r.ClusterPeerID, mc.localFullShardIDList, mc, mc.rpcConn.log)
+		d.CreatePeerConns(r.ClusterPeerID, mc.localFullShardIDList, mc, mc.baseConn.log)
 	}
 
 	return &wire.CreateClusterPeerConnectionResponse{ErrorCode: 0}, nil
@@ -301,9 +295,7 @@ func (mc *MasterConn) handleCreateClusterPeerConnection(req any) (any, error) {
 func (mc *MasterConn) handleDestroyClusterPeerConnection(req any) (any, error) {
 	r := req.(*wire.DestroyClusterPeerConnectionCommand)
 
-	mc.dispatcherMu.RLock()
 	d := mc.dispatcher
-	mc.dispatcherMu.RUnlock()
 	if d != nil {
 		d.DestroyPeerConns(r.ClusterPeerID)
 	}
@@ -595,12 +587,13 @@ func (mc *MasterConn) ForwardFrame(f *wire.Frame) error {
 
 // SetDispatcher wires the dispatcher that routes peer traffic. It also installs
 // a forwarder that validates the branch before routing.
+//
+// Must be called before Start(). dispatcher and forwarder are immutable after Start().
+//
 // Python: MasterConnection.get_connection_to_forward() closes the connection if
 // the branch is not in the configured full_shard_id_list.
 func (mc *MasterConn) SetDispatcher(d *Dispatcher) {
-	mc.dispatcherMu.Lock()
 	mc.dispatcher = d
-	mc.dispatcherMu.Unlock()
 
 	// Create a wrapper forwarder that validates branch before routing.
 	// Python: MasterConnection.get_connection_to_forward() only validates
@@ -624,13 +617,11 @@ func (mc *MasterConn) isValidBranch(branch uint32) bool {
 
 // Close closes the master connection and all associated peer connections.
 func (mc *MasterConn) Close() error {
-	mc.dispatcherMu.RLock()
 	d := mc.dispatcher
-	mc.dispatcherMu.RUnlock()
 	if d != nil {
 		d.Close()
 	}
-	return mc.rpcConn.Close()
+	return mc.baseConn.Close()
 }
 
 // SendRPCMeta sends a request with ClusterMetadata and waits for the response.
diff --git a/qkc/cluster/slave/peer_conn.go b/qkc/cluster/slave/peer_conn.go
index 5173e783179b..5dcd80302ebe 100644
--- a/qkc/cluster/slave/peer_conn.go
+++ b/qkc/cluster/slave/peer_conn.go
@@ -82,7 +82,7 @@ func (vt *virtualTransport) receive(frame *wire.Frame) bool {
 // responsibilities: independent RPC ID namespace, CommandOp handler dispatch,
 // and lifecycle tied to master commands.
 type PeerConn struct {
-	*rpcConn
+	*baseConn
 
 	clusterPeerID uint64
 	branch        uint32
@@ -94,7 +94,7 @@ type PeerConn struct {
 func NewPeerConn(clusterPeerID uint64, branch uint32, masterConn *MasterConn, logger log.Logger) *PeerConn {
 	vt := newVirtualTransport(clusterPeerID, branch, masterConn)
 	pc := &PeerConn{
-		rpcConn:       newRPCConn(vt, logger),
+		baseConn:      newBaseConn(vt, logger),
 		clusterPeerID: clusterPeerID,
 		branch:        branch,
 		vt:            vt,
@@ -111,7 +111,7 @@ const ReservedClusterPeerID = 0
 // registerOpSerializers registers serializers for every CommandOp so that both
 // inbound requests and outbound responses can be (de)serialized.
 func (pc *PeerConn) registerOpSerializers() {
-	pc.rpcConn.RegisterOpSerializers(map[byte]*OpSerializer{
+	pc.baseConn.RegisterOpSerializers(map[byte]*OpSerializer{
 		// §1 Hello / master-only
 		byte(wire.CommandOpHello):                          OpSerializerFor[wire.HelloCommand, wire.HelloCommand](),
 		byte(wire.CommandOpNewMinorBlockHeaderList):        OpSerializerFor[wire.NewMinorBlockHeaderListCommand, wire.NewMinorBlockHeaderListCommand](),
@@ -144,7 +144,7 @@ func (pc *PeerConn) registerOpSerializers() {
 // registerHandlers registers the shard-level peer handlers. These are stubs;
 // real implementations require the shard runtime to be ported.
 func (pc *PeerConn) registerHandlers() {
-	pc.rpcConn.RegisterTypedHandlers(map[byte]TypedHandler{
+	pc.baseConn.RegisterTypedHandlers(map[byte]TypedHandler{
 		// ── Migration stubs ─────────────────────────────────────────────
 		// These handlers exist only to preserve protocol compatibility.
 		// Real implementations must be added outside the connection layer.
@@ -161,7 +161,7 @@ func (pc *PeerConn) registerHandlers() {
 		byte(wire.CommandOpGetMinorBlockHeaderListWithSkipRequest): pc.handleGetMinorBlockHeaderListWithSkipRequest,
 	})
 
-	pc.rpcConn.RegisterNonRPCOps([]byte{
+	pc.baseConn.RegisterNonRPCOps([]byte{
 		byte(wire.CommandOpNewMinorBlockHeaderList),
 		byte(wire.CommandOpNewTransactionList),
 		byte(wire.CommandOpNewBlockMinor),
diff --git a/qkc/cluster/slave/peer_conn_test.go b/qkc/cluster/slave/peer_conn_test.go
index 73c118f28253..980584321cb6 100644
--- a/qkc/cluster/slave/peer_conn_test.go
+++ b/qkc/cluster/slave/peer_conn_test.go
@@ -356,18 +356,22 @@ func TestMasterConn_CreateDestroyPeerConnection(t *testing.T) {
 		t.Fatalf("expected error_code 0, got %d", createResp.ErrorCode)
 	}
 
-	// The dispatcher should now hold one PeerConn per local shard.
+	// Capture PeerConn pointers before destroy to avoid racing with dispatcher's
+	// internal map mutation during DestroyPeerConns.
 	branchMap := client.dispatcher.peers[clusterPeerID]
 	if len(branchMap) != len(client.localFullShardIDList) {
 		t.Fatalf("expected %d peer conns, got %d", len(client.localFullShardIDList), len(branchMap))
 	}
+	peerConns := make([]*PeerConn, 0, len(client.localFullShardIDList))
 	for _, branch := range client.localFullShardIDList {
-		if branchMap[branch] == nil {
+		pc := branchMap[branch]
+		if pc == nil {
 			t.Fatalf("missing peer conn for branch 0x%x", branch)
 		}
-		if branchMap[branch].IsClosed() {
+		if pc.IsClosed() {
 			t.Fatalf("peer conn for branch 0x%x is already closed", branch)
 		}
+		peerConns = append(peerConns, pc)
 	}
 
 	// Destroy the peer connections.
@@ -383,9 +387,10 @@ func TestMasterConn_CreateDestroyPeerConnection(t *testing.T) {
 	})
 
 	// Wait for peer connections to be closed (async since handler runs in goroutine).
+	// Check captured pointers instead of reading dispatcher.peers to avoid data race.
 	waitForCondition(t, 2*time.Second, func() bool {
-		for _, branch := range client.localFullShardIDList {
-			if pc := client.dispatcher.peers[clusterPeerID][branch]; pc != nil && !pc.IsClosed() {
+		for _, pc := range peerConns {
+			if !pc.IsClosed() {
 				return false
 			}
 		}
diff --git a/qkc/cluster/slave/testdata/pyproto/peer_master.py b/qkc/cluster/slave/testdata/pyproto/peer_master.py
deleted file mode 100644
index 55eacd58668b..000000000000
--- a/qkc/cluster/slave/testdata/pyproto/peer_master.py
+++ /dev/null
@@ -1,245 +0,0 @@
-#!/usr/bin/env python3
-"""Python Master protocol peer for PeerConn interoperability tests.
-
-This peer simulates a Python Master that:
-  1. Accepts a Go Slave connection (MasterConn)
-  2. Performs PING/PONG handshake
-  3. Sends CreateClusterPeerConnectionRequest to create a PeerConn on the Go side
-  4. Sends peer traffic (CommandOp frames with cluster_peer_id != 0) through
-     the MasterConn transport, simulating forwarded external peer traffic
-  5. Validates that the Go PeerConn correctly handles the traffic
-  6. Sends DestroyClusterPeerConnectionCommand to tear down the PeerConn
-  7. Verifies the connection is still alive after destroy
-
-Wire format for master frames:
-  [4B payload_len][4B branch][8B cluster_peer_id][1B opcode][8B rpc_id][payload]
-
-Usage:
-    python3 peer_master.py --port 0 --id "py-master" --shards "1,2" --cluster-peer-id 42
-
-Output:
-    PORT:<port>
-    PONG_OK id=<hex>
-    CREATE_OK error_code=<n>
-    PEER_RPC_OK opcode=0x0a rpc_id=<n>
-    PEER_NONRPC_OK
-    DESTROY_OK
-    POST_DESTROY_PONG_OK id=<hex>
-    DISCONNECTED
-"""
-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 opcodes (master <-> slave)
-CLUSTER_OP_PING = 1 + CLUSTER_OP_BASE
-CLUSTER_OP_PONG = 2 + CLUSTER_OP_BASE
-CLUSTER_OP_CREATE_CLUSTER_PEER_CONNECTION_REQUEST = 25 + CLUSTER_OP_BASE
-CLUSTER_OP_CREATE_CLUSTER_PEER_CONNECTION_RESPONSE = 26 + CLUSTER_OP_BASE
-CLUSTER_OP_DESTROY_CLUSTER_PEER_CONNECTION_COMMAND = 27 + CLUSTER_OP_BASE
-
-# Command opcodes (peer <-> peer, tunneled through master)
-COMMAND_OP_GET_MINOR_BLOCK_LIST_REQUEST = 0x09
-COMMAND_OP_GET_MINOR_BLOCK_LIST_RESPONSE = 0x0A
-COMMAND_OP_NEW_MINOR_BLOCK_HEADER_LIST = 0x01
-
-
-def serialize_create_peer_connection_request(cluster_peer_id):
-    """Serialize CreateClusterPeerConnectionRequest.
-
-    Fields:
-      ClusterPeerID: uint64 (8 bytes BE)
-    """
-    return struct.pack('>Q', cluster_peer_id)
-
-
-def serialize_destroy_peer_connection_command(cluster_peer_id):
-    """Serialize DestroyClusterPeerConnectionCommand.
-
-    Fields:
-      ClusterPeerID: uint64 (8 bytes BE)
-    """
-    return struct.pack('>Q', cluster_peer_id)
-
-
-def serialize_get_minor_block_list_request():
-    """Serialize GetMinorBlockListRequest.
-
-    Fields:
-      MinorBlockHashList: [][32]byte (4B count + hashes)
-    """
-    # Empty hash list
-    return struct.pack('>I', 0)
-
-
-def serialize_new_minor_block_header_list_command():
-    """Serialize NewMinorBlockHeaderListCommand.
-
-    Fields:
-      RootBlockHeader: *RawBytes (nil marker 0x00)
-      MinorBlockHeaderList: []*RawBytes (4B count + items)
-    """
-    data = b'\x00'  # RootBlockHeader: nil
-    data += struct.pack('>I', 0)  # empty list
-    return data
-
-
-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)
-    parser.add_argument('--cluster-peer-id', type=int, required=True)
-    args = parser.parse_args()
-
-    master_id = args.id.encode('utf-8')
-    shard_list = [int(s) for s in args.shards.split(',')]
-    cluster_peer_id = args.cluster_peer_id
-    branch = 0x00010001  # shard_id=1, chain_size=1
-
-    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 (verify MasterConn is alive)
-        _send_ping(conn, master_id, shard_list)
-
-        # 2. CreateClusterPeerConnectionRequest -> Response
-        create_payload = serialize_create_peer_connection_request(cluster_peer_id)
-        write_master_frame(
-            conn,
-            CLUSTER_OP_CREATE_CLUSTER_PEER_CONNECTION_REQUEST,
-            2,  # rpc_id
-            create_payload,
-            branch=branch,
-        )
-
-        frame = read_master_frame(conn)
-        if frame is None:
-            print("ERROR: no response for create peer connection", flush=True)
-            sys.exit(1)
-
-        if frame['opcode'] != CLUSTER_OP_CREATE_CLUSTER_PEER_CONNECTION_RESPONSE:
-            print(f"ERROR: expected CREATE_RESPONSE(0x{CLUSTER_OP_CREATE_CLUSTER_PEER_CONNECTION_RESPONSE:02x}), "
-                  f"got 0x{frame['opcode']:02x}", flush=True)
-            sys.exit(1)
-
-        if frame['rpc_id'] != 2:
-            print(f"ERROR: expected rpc_id 2, got {frame['rpc_id']}", flush=True)
-            sys.exit(1)
-
-        error_code = struct.unpack('>I', frame['payload'][0:4])[0]
-        print(f"CREATE_OK error_code={error_code}", flush=True)
-
-        # 3. Send peer RPC traffic (cluster_peer_id != 0)
-        #    GetMinorBlockListRequest -> GetMinorBlockListResponse
-        peer_rpc_payload = serialize_get_minor_block_list_request()
-        write_master_frame(
-            conn,
-            COMMAND_OP_GET_MINOR_BLOCK_LIST_REQUEST,
-            100,  # peer rpc_id
-            peer_rpc_payload,
-            branch=branch,
-            cluster_peer_id=cluster_peer_id,
-        )
-
-        frame = read_master_frame(conn)
-        if frame is None:
-            print("ERROR: no response for peer RPC", flush=True)
-            sys.exit(1)
-
-        if frame['opcode'] != COMMAND_OP_GET_MINOR_BLOCK_LIST_RESPONSE:
-            print(f"ERROR: expected peer response 0x{COMMAND_OP_GET_MINOR_BLOCK_LIST_RESPONSE:02x}, "
-                  f"got 0x{frame['opcode']:02x}", flush=True)
-            sys.exit(1)
-
-        if frame['rpc_id'] != 100:
-            print(f"ERROR: expected peer rpc_id 100, got {frame['rpc_id']}", flush=True)
-            sys.exit(1)
-
-        # Verify response metadata has correct cluster_peer_id
-        if frame['cluster_peer_id'] != cluster_peer_id:
-            print(f"ERROR: expected response cluster_peer_id {cluster_peer_id}, "
-                  f"got {frame['cluster_peer_id']}", flush=True)
-            sys.exit(1)
-
-        print(f"PEER_RPC_OK opcode=0x{frame['opcode']:02x} rpc_id={frame['rpc_id']}", flush=True)
-
-        # 4. Send peer non-RPC traffic (fire-and-forget)
-        nonrpc_payload = serialize_new_minor_block_header_list_command()
-        write_master_frame(
-            conn,
-            COMMAND_OP_NEW_MINOR_BLOCK_HEADER_LIST,
-            0,  # non-RPC: rpc_id must be 0
-            nonrpc_payload,
-            branch=branch,
-            cluster_peer_id=cluster_peer_id,
-        )
-
-        # Non-RPC produces no response. Verify by sending a follow-up PING.
-        _send_ping(conn, master_id, shard_list, rpc_id=3)
-        print("PEER_NONRPC_OK", flush=True)
-
-        # 5. DestroyClusterPeerConnectionCommand (fire-and-forget)
-        destroy_payload = serialize_destroy_peer_connection_command(cluster_peer_id)
-        write_master_frame(
-            conn,
-            CLUSTER_OP_DESTROY_CLUSTER_PEER_CONNECTION_COMMAND,
-            0,  # non-RPC
-            destroy_payload,
-            branch=branch,
-        )
-        print("DESTROY_OK", flush=True)
-
-        # 6. Verify MasterConn is still alive after destroy
-        _send_ping(conn, master_id, shard_list, rpc_id=4)
-
-    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, rpc_id=1):
-    """Send PING, wait for PONG, validate and print result."""
-    ping_payload = serialize_ping_request(master_id, shard_list)
-    write_master_frame(conn, CLUSTER_OP_PING, rpc_id, ping_payload)
-
-    frame = read_master_frame(conn)
-    if frame is None:
-        print(f"ERROR: no pong received for rpc_id={rpc_id}", 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'] != rpc_id:
-        print(f"ERROR: expected rpc_id {rpc_id}, got {frame['rpc_id']}", flush=True)
-        sys.exit(1)
-
-    peer_id_recv, _ = parse_pong_response(frame['payload'])
-    if rpc_id == 1:
-        print(f"PONG_OK id={peer_id_recv.hex()}", flush=True)
-    else:
-        print(f"POST_DESTROY_PONG_OK id={peer_id_recv.hex()}", flush=True)
-
-
-if __name__ == '__main__':
-    main()