Coverage for open_webui/utils/chat_fork.py: 90%

1from copy import deepcopy 

2 

3 

4def build_fork_history(messages_map: dict, source_message_id: str) -> tuple[dict, list[dict]]: 

5 if not messages_map: 

6 raise ValueError('chat has no messages to fork') 

7 

8 branch: list[tuple[str, dict]] = [] 

9 seen: set[str] = set() 

10 message_id = source_message_id 

11 

12 while message_id: 

13 if message_id in seen: 13 ↛ 14line 13 didn't jump to line 14 because the condition on line 13 was never true

14 raise ValueError('message branch contains a cycle') 

15 seen.add(message_id) 

16 

17 message = messages_map.get(message_id) 

18 if not isinstance(message, dict): 

19 raise ValueError('message not found') 

20 

21 branch.append((message_id, message)) 

22 message_id = message.get('parentId') 

23 

24 fork_messages: dict[str, dict] = {} 

25 ordered_messages: list[dict] = [] 

26 parent_id = None 

27 

28 for message_id, message in reversed(branch): 

29 copied = deepcopy(message) 

30 copied['id'] = message_id 

31 copied['parentId'] = parent_id 

32 copied['childrenIds'] = [] 

33 

34 if parent_id: 34 ↛ 35line 34 didn't jump to line 35 because the condition on line 34 was never true

35 fork_messages[parent_id]['childrenIds'] = [message_id] 

36 

37 fork_messages[message_id] = copied 

38 ordered_messages.append(copied) 

39 parent_id = message_id 

40 

41 return {'messages': fork_messages, 'currentId': source_message_id}, ordered_messages