feat: [EPIC-GW-05] 会话路由与流式透传中继 (Gateway 核心能力补全)

internal/cli/gateway_commands.go

@@ -124,23 +124,29 @@ func defaultGatewayCommandRunner(ctx context.Context, options gatewayCommandOpti
 
 	signalContext, stop := signal.NotifyContext(ctx, os.Interrupt, syscall.SIGTERM)
 	defer stop()
+	relay := gateway.NewStreamRelay(gateway.StreamRelayOptions{
+		Logger: logger,
+	})
 
 	ipcServer, err := newGatewayServer(gateway.ServerOptions{
 		ListenAddress: options.ListenAddress,
 		Logger:        logger,
+		Relay:         relay,
 	})
 	if err != nil {
 		return err
 	}
 	networkServer, err := newGatewayNetwork(gateway.NetworkServerOptions{
 		ListenAddress: options.HTTPAddress,
 		Logger:        logger,
+		Relay:         relay,
 	})
 	if err != nil {
 		_ = ipcServer.Close(context.Background())
 		return err
 	}
 	defer func() {
+		relay.Stop()
 		_ = networkServer.Close(context.Background())
 		_ = ipcServer.Close(context.Background())
 	}()

internal/gateway/bootstrap.go

@@ -10,26 +10,27 @@ import (
 	"neo-code/internal/gateway/protocol"
 )
 
-type requestFrameHandler func(frame MessageFrame) MessageFrame
+type requestFrameHandler func(ctx context.Context, frame MessageFrame) MessageFrame
 
 var wakeOpenURLHandler = handlers.NewWakeOpenURLHandler()
 
 var requestFrameHandlers = map[FrameAction]requestFrameHandler{
 	FrameActionPing:        handlePingFrame,
+	FrameActionBindStream:  handleBindStreamFrame,
 	FrameActionWakeOpenURL: handleWakeOpenURLFrame,
 }
 
 // dispatchRequestFrame 统一分发 request 帧到对应动作处理器。
-func dispatchRequestFrame(_ context.Context, frame MessageFrame, _ RuntimePort) MessageFrame {
+func dispatchRequestFrame(ctx context.Context, frame MessageFrame, _ RuntimePort) MessageFrame {
 	handler, ok := requestFrameHandlers[frame.Action]
 	if !ok {
 		return errorFrame(frame, NewFrameError(ErrorCodeUnsupportedAction, "action is not implemented in gateway step 2"))
 	}
-	return handler(frame)
+	return handler(ctx, frame)
 }
 
 // handlePingFrame 处理 ping 探活请求并返回 pong 响应。
-func handlePingFrame(frame MessageFrame) MessageFrame {
+func handlePingFrame(_ context.Context, frame MessageFrame) MessageFrame {
 	return MessageFrame{
 		Type:      FrameTypeAck,
 		Action:    FrameActionPing,
@@ -40,8 +41,43 @@ func handlePingFrame(frame MessageFrame) MessageFrame {
 	}
 }
 
+// handleBindStreamFrame 处理 gateway.bindStream 请求并登记连接到会话路由表。
+func handleBindStreamFrame(ctx context.Context, frame MessageFrame) MessageFrame {
+	params, err := decodeBindStreamParams(frame.Payload)
+	if err != nil {
+		return errorFrame(frame, err)
+	}
+
+	relay, relayExists := StreamRelayFromContext(ctx)
+	connectionID, connectionExists := ConnectionIDFromContext(ctx)
+	if !relayExists || !connectionExists {
+		return errorFrame(frame, NewFrameError(ErrorCodeInternalError, "stream relay context is unavailable"))
+	}
+
+	if bindErr := relay.BindConnection(connectionID, StreamBinding{
+		SessionID: params.SessionID,
+		RunID:     params.RunID,
+		Channel:   params.Channel,
+		Explicit:  true,
+	}); bindErr != nil {
+		return errorFrame(frame, bindErr)
+	}
+
+	return MessageFrame{
+		Type:      FrameTypeAck,
+		Action:    FrameActionBindStream,
+		RequestID: frame.RequestID,
+		SessionID: params.SessionID,
+		RunID:     params.RunID,
+		Payload: map[string]any{
+			"message": "stream binding updated",
+			"channel": params.Channel,
+		},
+	}
+}
+
 // handleWakeOpenURLFrame 解析并处理 wake.openUrl 请求。
-func handleWakeOpenURLFrame(frame MessageFrame) MessageFrame {
+func handleWakeOpenURLFrame(_ context.Context, frame MessageFrame) MessageFrame {
 	intent, err := decodeWakeIntent(frame.Payload)
 	if err != nil {
 		return errorFrame(frame, NewFrameError(ErrorCodeInvalidFrame, "invalid wake payload"))
@@ -51,16 +87,92 @@ func handleWakeOpenURLFrame(frame MessageFrame) MessageFrame {
 	if wakeErr != nil {
 		return errorFrame(frame, toFrameError(wakeErr))
 	}
+	sessionID := intent.SessionID
+	if strings.TrimSpace(sessionID) == "" {
+		sessionID = strings.TrimSpace(frame.SessionID)
+	}
 
 	return MessageFrame{
 		Type:      FrameTypeAck,
 		Action:    FrameActionWakeOpenURL,
 		RequestID: frame.RequestID,
-		SessionID: intent.SessionID,
+		SessionID: sessionID,
 		Payload:   result,
 	}
 }
 
+type bindStreamParams struct {
+	SessionID string
+	RunID     string
+	Channel   StreamChannel
+}
+
+// decodeBindStreamParams 将 payload 解析为 bind_stream 所需参数。
+func decodeBindStreamParams(payload any) (bindStreamParams, *FrameError) {
+	switch typed := payload.(type) {
+	case protocol.BindStreamParams:
+		return normalizeBindStreamParams(typed)
+	case *protocol.BindStreamParams:
+		if typed == nil {
+			return bindStreamParams{}, NewFrameError(ErrorCodeInvalidFrame, "invalid bind_stream payload")
+		}
+		return normalizeBindStreamParams(*typed)
+	case map[string]any:
+		return normalizeBindStreamParams(protocol.BindStreamParams{
+			SessionID: readStringValue(typed, "session_id"),
+			RunID:     readStringValue(typed, "run_id"),
+			Channel:   readStringValue(typed, "channel"),
+		})
+	default:
+		raw, marshalErr := json.Marshal(payload)
+		if marshalErr != nil {
+			return bindStreamParams{}, NewFrameError(ErrorCodeInvalidFrame, "invalid bind_stream payload")
+		}
+		var decoded protocol.BindStreamParams
+		if unmarshalErr := json.Unmarshal(raw, &decoded); unmarshalErr != nil {
+			return bindStreamParams{}, NewFrameError(ErrorCodeInvalidFrame, "invalid bind_stream payload")
+		}
+		return normalizeBindStreamParams(decoded)
+	}
+}
+
+// normalizeBindStreamParams 对 bind_stream 参数执行归一化与有效性校验。
+func normalizeBindStreamParams(params protocol.BindStreamParams) (bindStreamParams, *FrameError) {
+	sessionID := strings.TrimSpace(params.SessionID)
+	if sessionID == "" {
+		return bindStreamParams{}, NewMissingRequiredFieldError("payload.session_id")
+	}
+
+	runID := strings.TrimSpace(params.RunID)
+	channel := strings.ToLower(strings.TrimSpace(params.Channel))
+	if channel == "" {
+		channel = string(StreamChannelAll)
+	}
+	parsedChannel, validChannel := ParseStreamChannel(channel)
+	if !validChannel {
+		return bindStreamParams{}, NewFrameError(ErrorCodeInvalidAction, "invalid bind_stream channel")
+	}
+
+	return bindStreamParams{
+		SessionID: sessionID,
+		RunID:     runID,
+		Channel:   parsedChannel,
+	}, nil
+}
+
+// readStringValue 从 map 负载中读取并归一化字符串字段。
+func readStringValue(payload map[string]any, key string) string {
+	rawValue, exists := payload[key]
+	if !exists {
+		return ""
+	}
+	stringValue, ok := rawValue.(string)
+	if !ok {
+		return ""
+	}
+	return strings.TrimSpace(stringValue)
+}
+
 // decodeWakeIntent 将任意 payload 解码为 WakeIntent 结构。
 func decodeWakeIntent(payload any) (protocol.WakeIntent, error) {
 	if payload == nil {

internal/gateway/bootstrap_test.go

@@ -95,6 +95,108 @@ func TestDispatchRequestFrameUnsupportedAction(t *testing.T) {
 	}
 }
 
+func TestDispatchRequestFrameBindStream(t *testing.T) {
+	relay := NewStreamRelay(StreamRelayOptions{})
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	connectionID := NewConnectionID()
+	connectionCtx := WithConnectionID(ctx, connectionID)
+	connectionCtx = WithStreamRelay(connectionCtx, relay)
+
+	if err := relay.RegisterConnection(ConnectionRegistration{
+		ConnectionID: connectionID,
+		Channel:      StreamChannelIPC,
+		Context:      connectionCtx,
+		Cancel:       cancel,
+		Write: func(message RelayMessage) error {
+			_ = message
+			return nil
+		},
+		Close: func() {},
+	}); err != nil {
+		t.Fatalf("register connection: %v", err)
+	}
+	defer relay.dropConnection(connectionID)
+
+	response := dispatchRequestFrame(connectionCtx, MessageFrame{
+		Type:      FrameTypeRequest,
+		Action:    FrameActionBindStream,
+		RequestID: "bind-1",
+		Payload: protocol.BindStreamParams{
+			SessionID: "session-1",
+			RunID:     "run-1",
+			Channel:   "ipc",
+		},
+	}, nil)
+	if response.Type != FrameTypeAck {
+		t.Fatalf("response type = %q, want %q", response.Type, FrameTypeAck)
+	}
+	if response.Action != FrameActionBindStream {
+		t.Fatalf("response action = %q, want %q", response.Action, FrameActionBindStream)
+	}
+	if response.SessionID != "session-1" {
+		t.Fatalf("session_id = %q, want %q", response.SessionID, "session-1")
+	}
+}
+
+func TestHandleBindStreamFrameErrors(t *testing.T) {
+	t.Run("missing relay context", func(t *testing.T) {
+		response := handleBindStreamFrame(context.Background(), MessageFrame{
+			Type:   FrameTypeRequest,
+			Action: FrameActionBindStream,
+			Payload: protocol.BindStreamParams{
+				SessionID: "session-1",
+			},
+		})
+		if response.Type != FrameTypeError {
+			t.Fatalf("response type = %q, want %q", response.Type, FrameTypeError)
+		}
+		if response.Error == nil || response.Error.Code != ErrorCodeInternalError.String() {
+			t.Fatalf("response error = %#v, want %q", response.Error, ErrorCodeInternalError.String())
+		}
+	})
+
+	t.Run("channel mismatch", func(t *testing.T) {
+		relay := NewStreamRelay(StreamRelayOptions{})
+		ctx, cancel := context.WithCancel(context.Background())
+		defer cancel()
+
+		connectionID := NewConnectionID()
+		connectionCtx := WithConnectionID(ctx, connectionID)
+		connectionCtx = WithStreamRelay(connectionCtx, relay)
+		if err := relay.RegisterConnection(ConnectionRegistration{
+			ConnectionID: connectionID,
+			Channel:      StreamChannelWS,
+			Context:      connectionCtx,
+			Cancel:       cancel,
+			Write: func(message RelayMessage) error {
+				_ = message
+				return nil
+			},
+			Close: func() {},
+		}); err != nil {
+			t.Fatalf("register connection: %v", err)
+		}
+		defer relay.dropConnection(connectionID)
+
+		response := handleBindStreamFrame(connectionCtx, MessageFrame{
+			Type:   FrameTypeRequest,
+			Action: FrameActionBindStream,
+			Payload: protocol.BindStreamParams{
+				SessionID: "session-1",
+				Channel:   "ipc",
+			},
+		})
+		if response.Type != FrameTypeError {
+			t.Fatalf("response type = %q, want %q", response.Type, FrameTypeError)
+		}
+		if response.Error == nil || response.Error.Code != ErrorCodeInvalidAction.String() {
+			t.Fatalf("response error = %#v, want %q", response.Error, ErrorCodeInvalidAction.String())
+		}
+	})
+}
+
 func TestDecodeWakeIntentAdditionalBranches(t *testing.T) {
 	t.Run("nil payload", func(t *testing.T) {
 		_, err := decodeWakeIntent(nil)