x: wire MetadataFeature (relay v0.7.0) for tunnel recording mode negotiation

Tunnel connectors send record.mode via MetadataFeature in relay requests.
The handler parses MetadataFeature, stores on tunnel during bind, and
propagates to entrypoint services where recorder context applies it
(per-request Gost-Record header still takes precedence).

- connector/tunnel: send record.mode MetadataFeature in Connect/Bind
- connector/tunnel: parse record.mode from metadata config
- handler/tunnel: parse MetadataFeature from relay request features
- handler/tunnel: pass connMetadata through handleBind → pool.SetMetadata
- handler/tunnel: SetMetadata/GetMetadata on ConnectorPool + Tunnel
- handler/tunnel/entrypointsvc: apply pool metadata to recorder context
- deps: bump relay v0.6.2 → v0.7.0
- sniffer_sse.go: whitespace gofmt fixes

connector/tunnel/bind.go

@@ -78,6 +78,11 @@ func (c *tunnelConnector) initTunnel(conn net.Conn, network, address string) (ad
 	req.Features = append(req.Features, &relay.TunnelFeature{
 		ID: c.md.tunnelID,
 	})
+	if c.md.recordMode != "" {
+		req.Features = append(req.Features, &relay.MetadataFeature{
+			KVs: map[string]string{"record.mode": c.md.recordMode},
+		})
+	}
 	if _, err = req.WriteTo(conn); err != nil {
 		return
 	}

connector/tunnel/connector.go

@@ -89,6 +89,11 @@ func (c *tunnelConnector) Connect(ctx context.Context, conn net.Conn, network, a
 	req.Features = append(req.Features, &relay.TunnelFeature{
 		ID: c.md.tunnelID,
 	})
+	if c.md.recordMode != "" {
+		req.Features = append(req.Features, &relay.MetadataFeature{
+			KVs: map[string]string{"record.mode": c.md.recordMode},
+		})
+	}
 
 	if _, err := req.WriteTo(conn); err != nil {
 		return nil, err

connector/tunnel/metadata.go

@@ -19,6 +19,7 @@ type metadata struct {
 	connectTimeout time.Duration
 	tunnelID       relay.TunnelID
 	muxCfg         *mux.Config
+	recordMode     string
 }
 
 func (c *tunnelConnector) parseMetadata(md mdata.Metadata) (err error) {
@@ -44,6 +45,8 @@ func (c *tunnelConnector) parseMetadata(md mdata.Metadata) (err error) {
 		c.md.tunnelID = c.md.tunnelID.SetWeight(uint8(weight))
 	}
 
+	c.md.recordMode = mdutil.GetString(md, "record.mode")
+
 	c.md.muxCfg = &mux.Config{
 		Version:           mdutil.GetInt(md, "mux.version"),
 		KeepAliveInterval: mdutil.GetDuration(md, "mux.keepaliveInterval"),

go.mod

@@ -13,7 +13,7 @@ require (
 	github.com/go-gost/gosocks4 v0.1.0
 	github.com/go-gost/gosocks5 v0.5.0
 	github.com/go-gost/plugin v0.5.0
-	github.com/go-gost/relay v0.6.2
+	github.com/go-gost/relay v0.7.0
 	github.com/go-gost/tls-dissector v0.2.0
 	github.com/go-redis/redis/v8 v8.11.5
 	github.com/gobwas/glob v0.2.3

go.sum

@@ -61,8 +61,8 @@ github.com/go-gost/gosocks5 v0.5.0 h1:YE37l1MJwde8diIQdynStqogMotG5enoTdborhA5yi
 github.com/go-gost/gosocks5 v0.5.0/go.mod h1:1G6I7HP7VFVxveGkoK8mnprnJqSqJjdcASKsdUn4Pp4=
 github.com/go-gost/plugin v0.5.0 h1:zK3Ezcv+z6bz6aUhaFZcZiJc42w44ohybzrj5tnWzcM=
 github.com/go-gost/plugin v0.5.0/go.mod h1:oN23l+yGDCIP9G3KnDl/I/0zVGOobZUDCB2Z5yYYXts=
-github.com/go-gost/relay v0.6.2 h1:0Rgm1TOZvrilAKj6zn+UMvAGWtS0H08e2wBX03Y6g+I=
-github.com/go-gost/relay v0.6.2/go.mod h1:Dku0f5sfjOClrZFiDmQUrYYJ4uof7rnkCUBfsl0PSAI=
+github.com/go-gost/relay v0.7.0 h1:J8e3Sba6DtBJQotXY5j5EaZNWwtv6Z9CoCFQsnrHNcE=
+github.com/go-gost/relay v0.7.0/go.mod h1:Dku0f5sfjOClrZFiDmQUrYYJ4uof7rnkCUBfsl0PSAI=
 github.com/go-gost/tls-dissector v0.2.0 h1:9tE6WOzzpurATTBWn60DU4R8gibpGNY8/qVcc1SicVg=
 github.com/go-gost/tls-dissector v0.2.0/go.mod h1:/9QfdewqmHdaE362Hv5nDaSWLx3pCmtD870d6GaquXs=
 github.com/go-logr/logr v1.4.3 h1:CjnDlHq8ikf6E492q6eKboGOC0T8CDaOvkHCIg8idEI=

handler/tunnel/bind.go

@@ -43,7 +43,7 @@ import (
 // The mux session ownership is transferred to the Connector — conn is NOT
 // closed after this function returns (no defer conn.Close()). The Connector's
 // waitClose goroutine handles session lifecycle.
-func (h *tunnelHandler) handleBind(ctx context.Context, conn net.Conn, network, address string, tunnelID relay.TunnelID, log logger.Logger) (err error) {
+func (h *tunnelHandler) handleBind(ctx context.Context, conn net.Conn, network, address string, tunnelID relay.TunnelID, metadata map[string]string, log logger.Logger) (err error) {
 	resp := relay.Response{
 		Version: relay.Version1,
 		Status:  relay.StatusOK,
@@ -109,6 +109,9 @@ func (h *tunnelHandler) handleBind(ctx context.Context, conn net.Conn, network,
 	})
 
 	h.pool.Add(tunnelID, c, h.md.tunnelTTL)
+	if len(metadata) > 0 {
+		h.pool.SetMetadata(tunnelID, metadata)
+	}
 	if h.md.ingress != nil {
 		h.md.ingress.SetRule(ctx, &ingress.Rule{
 			Hostname: endpoint,

handler/tunnel/bind_test.go

@@ -39,7 +39,7 @@ func TestHandleBind_ResponseWritten(t *testing.T) {
 
 	errCh := make(chan error, 1)
 	go func() {
-		errCh <- h.handleBind(context.Background(), server, "tcp", "0.0.0.0:0", tid, testLogger())
+		errCh <- h.handleBind(context.Background(), server, "tcp", "0.0.0.0:0", tid, nil, testLogger())
 	}()
 
 	// Read the relay response from the client side of the pipe.
@@ -110,7 +110,7 @@ func TestHandleBind_WithIngress(t *testing.T) {
 
 	errCh := make(chan error, 1)
 	go func() {
-		errCh <- h.handleBind(context.Background(), server, "tcp", "0.0.0.0:0", tid, testLogger())
+		errCh <- h.handleBind(context.Background(), server, "tcp", "0.0.0.0:0", tid, nil, testLogger())
 	}()
 
 	_, _ = (&relay.Response{}).ReadFrom(client)
@@ -164,7 +164,7 @@ func TestHandleBind_WithSD(t *testing.T) {
 
 	errCh := make(chan error, 1)
 	go func() {
-		errCh <- h.handleBind(context.Background(), server, "tcp", "0.0.0.0:0", tid, testLogger())
+		errCh <- h.handleBind(context.Background(), server, "tcp", "0.0.0.0:0", tid, nil, testLogger())
 	}()
 
 	// Read the response from client side.
@@ -223,7 +223,7 @@ func TestHandleBind_UDPConnector(t *testing.T) {
 
 	errCh := make(chan error, 1)
 	go func() {
-		errCh <- h.handleBind(context.Background(), server, "udp", "0.0.0.0:0", tid, testLogger())
+		errCh <- h.handleBind(context.Background(), server, "udp", "0.0.0.0:0", tid, nil, testLogger())
 	}()
 
 	resp := &relay.Response{}
@@ -362,7 +362,7 @@ func TestConnectWithBindEndToEnd(t *testing.T) {
 
 	bindResult := make(chan error, 1)
 	go func() {
-		bindResult <- h.handleBind(context.Background(), bindServer, "tcp", "0.0.0.0:0", tid, testLogger())
+		bindResult <- h.handleBind(context.Background(), bindServer, "tcp", "0.0.0.0:0", tid, nil, testLogger())
 	}()
 
 	// Read the bind response from client side.
@@ -423,7 +423,7 @@ func TestHandleBind_WithHostEndpoint(t *testing.T) {
 
 	errCh := make(chan error, 1)
 	go func() {
-		errCh <- h.handleBind(context.Background(), server, "tcp", "myapp.example.com:8080", tid, testLogger())
+		errCh <- h.handleBind(context.Background(), server, "tcp", "myapp.example.com:8080", tid, nil, testLogger())
 	}()
 
 	_, _ = (&relay.Response{}).ReadFrom(client)
@@ -467,7 +467,7 @@ func TestHandleBind_CustomHost(t *testing.T) {
 
 	errCh := make(chan error, 1)
 	go func() {
-		errCh <- h.handleBind(context.Background(), server, "tcp", "dash:8081", tid, testLogger())
+		errCh <- h.handleBind(context.Background(), server, "tcp", "dash:8081", tid, nil, testLogger())
 	}()
 
 	resp := &relay.Response{}
@@ -533,7 +533,7 @@ func TestHandleBind_CustomHostConflict(t *testing.T) {
 
 	errCh := make(chan error, 1)
 	go func() {
-		errCh <- h.handleBind(context.Background(), server, "tcp", "dash:8081", tid, testLogger())
+		errCh <- h.handleBind(context.Background(), server, "tcp", "dash:8081", tid, nil, testLogger())
 	}()
 
 	resp := &relay.Response{}

handler/tunnel/connector.go

@@ -227,6 +227,23 @@ func (p *ConnectorPool) Close() error {
 	return nil
 }
 
+func (p *ConnectorPool) SetMetadata(tid relay.TunnelID, md map[string]string) {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	if t := p.tunnels[tid.String()]; t != nil {
+		t.SetMetadata(md)
+	}
+}
+
+func (p *ConnectorPool) GetMetadata(tid string) map[string]string {
+	p.mu.RLock()
+	defer p.mu.RUnlock()
+	if t := p.tunnels[tid]; t != nil {
+		return t.Metadata()
+	}
+	return nil
+}
+
 func (p *ConnectorPool) closeIdles(ctx context.Context) {
 	ticker := time.NewTicker(15 * time.Minute)
 	defer ticker.Stop()

handler/tunnel/entrypointsvc.go

@@ -10,6 +10,7 @@ import (
 	"github.com/go-gost/core/listener"
 	"github.com/go-gost/core/recorder"
 	"github.com/go-gost/core/service"
+	ictx "github.com/go-gost/x/internal/ctx"
 	xnet "github.com/go-gost/x/internal/net"
 	xrecorder "github.com/go-gost/x/recorder"
 	xservice "github.com/go-gost/x/service"
@@ -76,6 +77,18 @@ func (h *tunnelHandler) createEntrypointService(addr string, ing ingress.Ingress
 	}
 
 	dialFn := func(ctx epkg.DialContext, network, tid string) (net.Conn, string, string, error) {
+		realCtx := ctx.(context.Context)
+
+		// Apply connector-level recording preferences to the recorder object.
+		// Per-request Gost-Record header takes precedence — only set when empty.
+		if md := h.pool.GetMetadata(tid); len(md) > 0 {
+			if ro := ictx.RecorderObjectFromContext(realCtx); ro != nil && ro.RecordMode == "" {
+				if mode, ok := md["record.mode"]; ok {
+					ro.RecordMode = mode
+				}
+			}
+		}
+
 		d := &Dialer{
 			Node:    h.id,
 			Pool:    h.pool,

handler/tunnel/handler.go

@@ -250,6 +250,7 @@ func (h *tunnelHandler) Handle(ctx context.Context, conn net.Conn, opts ...handl
 	var srcAddr, dstAddr string
 	network := "tcp"
 	var tunnelID relay.TunnelID
+	var connMetadata map[string]string
 	for _, f := range req.Features {
 		switch f.Type() {
 		case relay.FeatureUserAuth:
@@ -273,6 +274,10 @@ func (h *tunnelHandler) Handle(ctx context.Context, conn net.Conn, opts ...handl
 			if feature, _ := f.(*relay.NetworkFeature); feature != nil {
 				network = feature.Network.String()
 			}
+		case relay.FeatureMetadata:
+			if feature, _ := f.(*relay.MetadataFeature); feature != nil {
+				connMetadata = feature.KVs
+			}
 		}
 	}
 
@@ -305,7 +310,7 @@ func (h *tunnelHandler) Handle(ctx context.Context, conn net.Conn, opts ...handl
 
 	case relay.CmdBind:
 		log.Debugf("bind: %s >> %s/%s", srcAddr, dstAddr, network)
-		return h.handleBind(ctx, conn, network, dstAddr, tunnelID, log)
+		return h.handleBind(ctx, conn, network, dstAddr, tunnelID, connMetadata, log)
 	default:
 		resp.Status = relay.StatusBadRequest
 		resp.WriteTo(conn) // write error ignored — conn is about to be closed

handler/tunnel/tunnel.go

@@ -46,6 +46,8 @@ type Tunnel struct {
 	mu    sync.RWMutex
 	// ttl is the interval between clean() ticks. Defaults to defaultTTL.
 	ttl time.Duration
+	// metadata carries key-value pairs set during tunnel bind (e.g. recording prefs).
+	metadata map[string]string
 }
 
 func NewTunnel(node string, tid relay.TunnelID, ttl time.Duration) *Tunnel {
@@ -149,6 +151,18 @@ func (t *Tunnel) CloseOnIdle() bool {
 	return false
 }
 
+func (t *Tunnel) SetMetadata(md map[string]string) {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+	t.metadata = md
+}
+
+func (t *Tunnel) Metadata() map[string]string {
+	t.mu.RLock()
+	defer t.mu.RUnlock()
+	return t.metadata
+}
+
 func (t *Tunnel) clean() {
 	ticker := time.NewTicker(t.ttl)
 	defer ticker.Stop()

internal/util/forwarder/sniffer_sse.go

@@ -46,17 +46,17 @@ func scanSSEEvents(data []byte, atEOF bool) (advance int, token []byte, err erro
 //
 // Three processing modes:
 //
-// 1. Non-streaming (plain): eagerly buffer the entire body, rewrite once,
-//    expose the final ContentLength.
+//  1. Non-streaming (plain): eagerly buffer the entire body, rewrite once,
+//     expose the final ContentLength.
 //
-// 2. Non-streaming (compressed): decompress, rewrite, recompress, expose
-//    the final ContentLength.
+//  2. Non-streaming (compressed): decompress, rewrite, recompress, expose
+//     the final ContentLength.
 //
-// 3. SSE streaming: split on \n\n via bufio.Scanner, rewrite each event
-//    independently. The first event carries sse_phase:"start", subsequent
-//    events carry sse_phase:"event", and when the scanner hits EOF an
-//    empty-body apply(nil, sse_phase:"end") is emitted so Rewriter plugins
-//    can append a trailing event (e.g. message_stop for LLM API conversion).
+//  3. SSE streaming: split on \n\n via bufio.Scanner, rewrite each event
+//     independently. The first event carries sse_phase:"start", subsequent
+//     events carry sse_phase:"event", and when the scanner hits EOF an
+//     empty-body apply(nil, sse_phase:"end") is emitted so Rewriter plugins
+//     can append a trailing event (e.g. message_stop for LLM API conversion).
 type rewriteBody struct {
 	ctx         context.Context
 	src         io.ReadCloser
@@ -69,8 +69,8 @@ type rewriteBody struct {
 	contentLength int64 // >= 0 for non-streaming bodies; -1 for streaming
 
 	// SSE stream lifecycle state.
-	eventIndex int  // incremented per SSE event for the "event" phase
-	ended      bool // true after the stream-end phase has been emitted
+	eventIndex int   // incremented per SSE event for the "event" phase
+	ended      bool  // true after the stream-end phase has been emitted
 	scannerErr error // non-nil when scanner terminated with an error
 
 	sid string // session ID, cached from context at construction time
@@ -143,10 +143,10 @@ func newRewriteBody(ctx context.Context, src io.ReadCloser, rewrites []chain.HTT
 		}
 		// Fits: rewrite and set ContentLength.
 		rb := &rewriteBody{
-			ctx:         ctx,
-			rewrites:    rewrites,
-			contentType: ct,
-			streaming:   false,
+			ctx:           ctx,
+			rewrites:      rewrites,
+			contentType:   ct,
+			streaming:     false,
 			contentLength: -1,
 		}
 		rb.sid = xctx.SidFromContext(ctx).String()
@@ -260,8 +260,8 @@ func newRewriteBody(ctx context.Context, src io.ReadCloser, rewrites []chain.HTT
 		rb.contentLength = int64(len(rewritten))
 	} else {
 		// SSE streaming: scan events split on \n\n and rewrite each one
-	// independently on Read. Lifecycle phases (first event = start,
-	// subsequent = event, EOF = end) are assigned in Read().
+		// independently on Read. Lifecycle phases (first event = start,
+		// subsequent = event, EOF = end) are assigned in Read().
 		scanner := bufio.NewScanner(src)
 		scanner.Split(scanSSEEvents)
 		scanner.Buffer(make([]byte, 64*1024), 64*1024*1024)
@@ -298,8 +298,8 @@ func (b *rewriteBody) Read(p []byte) (n int, err error) {
 				b.ended = true
 				md := map[string]any{
 					"sid":         b.sid,
-			"direction": b.direction,
-			"uri":       b.uri,
+					"direction":   b.direction,
+					"uri":         b.uri,
 					"sse_phase":   "end",
 					"event_index": b.eventIndex,
 				}
@@ -322,8 +322,8 @@ func (b *rewriteBody) Read(p []byte) (n int, err error) {
 				b.scannerErr = err
 				md := map[string]any{
 					"sid":          b.sid,
-			"direction": b.direction,
-			"uri":       b.uri,
+					"direction":    b.direction,
+					"uri":          b.uri,
 					"sse_phase":    "end",
 					"event_index":  b.eventIndex,
 					"stream_error": err.Error(),
@@ -351,11 +351,11 @@ func (b *rewriteBody) Read(p []byte) (n int, err error) {
 			phase = "start"
 		}
 		md := map[string]any{
-			"sid":          b.sid,
-			"direction": b.direction,
-			"uri":       b.uri,
-			"sse_phase":    phase,
-			"event_index":  b.eventIndex,
+			"sid":         b.sid,
+			"direction":   b.direction,
+			"uri":         b.uri,
+			"sse_phase":   phase,
+			"event_index": b.eventIndex,
 		}
 		b.eventIndex++
 
@@ -484,9 +484,9 @@ func (b *rewriteBody) apply(body []byte, opts ...rewriter.RewriteOption) ([]byte
 		return body, nil
 	}
 	for _, rw := range b.rewrites {
-			if !shouldApply(rw.Type, b.contentType, rw.Rewriter != nil) {
-				continue
-			}
+		if !shouldApply(rw.Type, b.contentType, rw.Rewriter != nil) {
+			continue
+		}
 
 		if rw.Rewriter != nil {
 			if body == nil || rw.Pattern == nil || rw.Pattern.Match(body) {