Coverage for core/langgraph/tools/files_write.py: 97%

1"""Per-tool module for the `files-write` deterministic builtin. 

2 

3Owns the tool's model-facing definition, its instruction surface (none), and 

4its atomic workspace write of `{path, content}` entries (ADR-0012/ADR-0024). 

5The definition and the execution are the single source; the legacy builtin 

6site delegates here. 

7""" 

8from __future__ import annotations 

9 

10import json 

11from dataclasses import dataclass 

12from pathlib import Path 

13from typing import Any 

14 

15from .contract import ToolContext, ToolResult 

16 

17try: 

18 from _files_output import ( 

19 FilesOutputError, 

20 resolve_workspace_root, 

21 validate_and_write_files, 

22 ) 

23except ImportError: # pragma: no cover — main.py run (qualified imports) 

24 from langgraph._files_output import ( 

25 FilesOutputError, 

26 resolve_workspace_root, 

27 validate_and_write_files, 

28 ) 

29 

30FILES_WRITE_TOOL_NAME = "files-write" 

31 

32FILES_WRITE_DEFINITION: dict[str, Any] = { 

33 "type": "function", 

34 "function": { 

35 "name": FILES_WRITE_TOOL_NAME, 

36 "description": ( 

37 "Write one or more {path, content} entries into the workspace. " 

38 "Accepts a single {path, content} object or a list of them; paths " 

39 "are workspace-relative. Validation is atomic: one invalid entry " 

40 "(traversal, .git target, per-file/step cap, non-string fields) " 

41 "rejects the whole call and nothing is written (ADR-0012)." 

42 ), 

43 "parameters": { 

44 "type": "object", 

45 "properties": { 

46 "entries": { 

47 "description": ( 

48 "A single {path, content} object or a list of them." 

49 ), 

50 "anyOf": [ 

51 { 

52 "type": "object", 

53 "properties": { 

54 "path": { 

55 "type": "string", 

56 "description": ( 

57 "Workspace-relative target path." 

58 ), 

59 }, 

60 "content": { 

61 "type": "string", 

62 "description": ( 

63 "UTF-8 file content to write." 

64 ), 

65 }, 

66 }, 

67 "required": ["path", "content"], 

68 "additionalProperties": False, 

69 }, 

70 { 

71 "type": "array", 

72 "items": { 

73 "type": "object", 

74 "properties": { 

75 "path": { 

76 "type": "string", 

77 "description": ( 

78 "Workspace-relative target path." 

79 ), 

80 }, 

81 "content": { 

82 "type": "string", 

83 "description": ( 

84 "UTF-8 file content to write." 

85 ), 

86 }, 

87 }, 

88 "required": ["path", "content"], 

89 "additionalProperties": False, 

90 }, 

91 }, 

92 ], 

93 }, 

94 }, 

95 "required": ["entries"], 

96 "additionalProperties": False, 

97 }, 

98 }, 

99} 

100 

101 

102@dataclass 

103class FilesWriteResult: 

104 written: list[str] 

105 count: int 

106 error: str | None 

107 details: dict[str, Any] | None 

108 

109 def as_tool_message(self) -> str: 

110 payload: dict[str, Any] = { 

111 "name": FILES_WRITE_TOOL_NAME, 

112 "ok": self.error is None, 

113 "written": self.written, 

114 "count": self.count, 

115 } 

116 if self.error is not None: 

117 payload["error"] = self.error 

118 return json.dumps(payload, ensure_ascii=False) 

119 

120 

121def _files_write_error(step_id: str, message: str) -> FilesWriteResult: 

122 return FilesWriteResult( 

123 [], 0, f"files-write step '{step_id}': {message}", None 

124 ) 

125 

126 

127def _normalize_entries( 

128 step_id: str, entries: Any 

129) -> tuple[list | None, FilesWriteResult | None]: 

130 if entries is None or entries == []: 

131 return None, _files_write_error( 

132 step_id, 

133 "'params.entries' is required — single " 

134 "{path, content} object or a list", 

135 ) 

136 if isinstance(entries, dict): 

137 return [entries], None 

138 if not isinstance(entries, list): 

139 return None, _files_write_error( 

140 step_id, 

141 f"'params.entries' must be a single " 

142 f"{{path, content}} object or a list, got {type(entries).__name__}", 

143 ) 

144 return entries, None 

145 

146 

147def _write_validated_entries( 

148 step_id: str, entries: list, workspace_root: Path 

149) -> FilesWriteResult | None: 

150 try: 

151 validate_and_write_files(entries, workspace_root) 

152 except FilesOutputError as exc: 

153 return _files_write_error(step_id, str(exc)) 

154 return None 

155 

156 

157def files_write( 

158 entries: Any, workspace_root: Path, step_id: str 

159) -> FilesWriteResult: 

160 """Atomically write `{path, content}` entries into the workspace 

161 (ADR-0012/ADR-0024). Every rejected write returns an explicit error 

162 carrying its `details`; it never raises. 

163 """ 

164 normalized, failure = _normalize_entries(step_id, entries) 

165 if failure is not None: 

166 return failure 

167 failure = _write_validated_entries(step_id, normalized, workspace_root) 

168 if failure is not None: 

169 return failure 

170 written = [entry["path"] for entry in normalized] 

171 return FilesWriteResult(written, len(written), None, None) 

172 

173 

174class _FilesWriteTool: 

175 name = FILES_WRITE_TOOL_NAME 

176 

177 def definition(self) -> dict[str, Any]: 

178 return FILES_WRITE_DEFINITION 

179 

180 def instruction(self) -> str | None: 

181 return None 

182 

183 def params(self) -> list[dict[str, Any]] | None: 

184 return None 

185 

186 def execute(self, context: ToolContext) -> ToolResult: 

187 step = context.step or {} 

188 step_id = step["id"] 

189 entries = (context.args or {}).get("entries") 

190 workspace_root = resolve_workspace_root(context.state or {}) 

191 res = files_write(entries, workspace_root, step_id) 

192 if res.error is not None: 

193 return ToolResult(ok=False, content=res.as_tool_message(), error=res.error) 

194 return ToolResult( 

195 ok=True, 

196 content=res.as_tool_message(), 

197 state_patch={step_id: {"written": res.written, "count": res.count}}, 

198 ) 

199 

200 

201TOOL = _FilesWriteTool()