Coverage for core/langgraph/_channel_reducers.py: 94%

1""" 

2_channel_reducers.py — Core-owned channel reducer catalogue. 

3 

4The six manifest-selectable ``BinaryOperatorAggregate`` reducers, the 

5name-keyed registry the manifest ``operator`` field selects from, and the 

6selector the engine uses. Every reducer is a pure two-argument callable 

7``reducer(accumulated, write) -> merged``; first-write seeding is the engine's 

8job, so ``accumulated`` is always an already-seeded value. 

9 

10Stdlib only. This module must not import the native LangGraph channel module 

11or materialize any native LangGraph channel. 

12""" 

13from __future__ import annotations 

14 

15from typing import Any, Callable 

16 

17try: 

18 from _reducers import ( 

19 ReducerTypeError, 

20 _dedupe_first_occurrence, 

21 _require_list, 

22 _require_mapping, 

23 ) 

24except ImportError: # pragma: no cover — main.py-ajossa (qualified importit) 

25 from langgraph._reducers import ( 

26 ReducerTypeError, 

27 _dedupe_first_occurrence, 

28 _require_list, 

29 _require_mapping, 

30 ) 

31 

32_MAX_DEPTH = 50 

33 

34_SENSITIVE_KEYS = frozenset({ 

35 "password", 

36 "passwd", 

37 "secret", 

38 "token", 

39 "access_token", 

40 "refresh_token", 

41 "api_key", 

42 "apikey", 

43 "authorization", 

44 "credential", 

45 "credentials", 

46 "private_key", 

47 "client_secret", 

48}) 

49 

50 

51def shallow_merge(accumulated: dict, write: Any) -> dict: 

52 """One-level dict merge; the incoming write wins, no recursion.""" 

53 _require_mapping(accumulated) 

54 _require_mapping(write) 

55 return {**accumulated, **write} 

56 

57 

58def list_dedup(accumulated: list, write: Any) -> list: 

59 """Union of lists without duplicates, first-occurrence order preserved.""" 

60 _require_list(accumulated) 

61 incoming = write if isinstance(write, list) else [write] 

62 return _dedupe_first_occurrence(accumulated + incoming) 

63 

64 

65def json_merge_patch(accumulated: Any, write: Any) -> Any: 

66 """RFC 7386 merge patch applied to ``accumulated`` with ``write``.""" 

67 return _merge_patch(accumulated, write, 0) 

68 

69 

70def json_merge_append(accumulated: Any, write: Any) -> Any: 

71 """RFC 7386 merge patch, except arrays at a shared key are appended.""" 

72 return _merge_append(accumulated, write, 0) 

73 

74 

75def json_merge_dedup(accumulated: Any, write: Any) -> Any: 

76 """RFC 7386 merge patch, except combined arrays are de-duplicated.""" 

77 return _merge_dedup(accumulated, write, 0) 

78 

79 

80def prune_sensitive(accumulated: dict, write: Any) -> dict: 

81 """Shallow-merge with the write, then drop sensitive keys recursively.""" 

82 _require_mapping(accumulated) 

83 _require_mapping(write) 

84 return _prune_sensitive({**accumulated, **write}, 0) 

85 

86 

87def _merge_patch(target: Any, patch: Any, depth: int) -> Any: 

88 if not isinstance(patch, dict): 

89 return patch 

90 if depth >= _MAX_DEPTH: 

91 return dict(patch) 

92 merged = dict(target) if isinstance(target, dict) else {} 

93 for key, value in patch.items(): 

94 if value is None: 

95 merged.pop(key, None) 

96 continue 

97 merged[key] = _merge_patch(merged.get(key), value, depth + 1) 

98 return merged 

99 

100 

101def _merge_append(target: Any, patch: Any, depth: int) -> Any: 

102 if not isinstance(patch, dict): 

103 return patch 

104 if depth >= _MAX_DEPTH: 

105 return dict(patch) 

106 merged = dict(target) if isinstance(target, dict) else {} 

107 for key, value in patch.items(): 

108 if value is None: 

109 merged.pop(key, None) 

110 continue 

111 target_value = merged.get(key) 

112 if isinstance(target_value, list) and isinstance(value, list): 

113 merged[key] = target_value + value 

114 continue 

115 merged[key] = _merge_append(target_value, value, depth + 1) 

116 return merged 

117 

118 

119def _merge_dedup(target: Any, patch: Any, depth: int) -> Any: 

120 if not isinstance(patch, dict): 

121 return patch 

122 if depth >= _MAX_DEPTH: 

123 return dict(patch) 

124 merged = dict(target) if isinstance(target, dict) else {} 

125 for key, value in patch.items(): 

126 if value is None: 

127 merged.pop(key, None) 

128 continue 

129 target_value = merged.get(key) 

130 if isinstance(target_value, list) and isinstance(value, list): 

131 merged[key] = _dedupe_first_occurrence(target_value + value) 

132 continue 

133 merged[key] = _merge_dedup(target_value, value, depth + 1) 

134 return merged 

135 

136 

137def _prune_sensitive(value: Any, depth: int) -> Any: 

138 if depth >= _MAX_DEPTH: 

139 return value 

140 if isinstance(value, dict): 

141 return { 

142 key: _prune_sensitive(item, depth + 1) 

143 for key, item in value.items() 

144 if not _is_sensitive_key(key) 

145 } 

146 if isinstance(value, list): 

147 return [_prune_sensitive(item, depth + 1) for item in value] 

148 return value 

149 

150 

151def _is_sensitive_key(key: Any) -> bool: 

152 return isinstance(key, str) and key.lower() in _SENSITIVE_KEYS 

153 

154 

155CHANNEL_REDUCERS: dict[str, Callable[[Any, Any], Any]] = { 

156 "shallow_merge": shallow_merge, 

157 "prune_sensitive": prune_sensitive, 

158 "list_dedup": list_dedup, 

159 "json_merge_patch": json_merge_patch, 

160 "json_merge_append": json_merge_append, 

161 "json_merge_dedup": json_merge_dedup, 

162} 

163 

164 

165def select_channel_reducer(name: str) -> Callable[[Any, Any], Any]: 

166 """Return the registered reducer for ``name`` or raise ``ValueError``.""" 

167 if not isinstance(name, str): 

168 raise ReducerTypeError( 

169 "channel reducer name must be a string, received " 

170 f"{type(name).__name__}" 

171 ) 

172 try: 

173 return CHANNEL_REDUCERS[name] 

174 except KeyError: 

175 raise ValueError(f"unknown channel reducer: {name!r}") from None