Coverage for core/langgraph/_builtins.py: 100%

1""" 

2_builtins.py — Domain builtins for agent-platform pipelines (v1). 

3 

4Toteuttaa ne builtin-komponentit, joita agent-platform-pipelinet käyttävät 

5ja jotka porttaavat ei-LLM-logiikan generic-muotoon: 

6 

7- list-dir-entries deterministinen yhden tason hakemistolistaus 

8- aggregate geneerinen fold-mekanismi (13 operaattoria, dedupe, on_error) 

9- ticket-publisher kirjoittaa hyväksytyn ticketin staging → published 

10- run-finalizer koostaa stats + flushaa Langfusen 

11- git-checkout kloonaa repon ja valmistelee työpuun (PR/BRANCH/TICKET) 

12- files-write kirjoittaa {path, content} -entryt workspaceen 

13- files-delete poistaa workspace-tiedostot (tracked → git rm, untracked → unlink) 

14 

15Kaikki builtinit noudattavat sopimusta: 

16 fn(step: dict, state: dict) -> dict # palauttaa uuden state-snapshotin 

17ja kirjoittavat tuloksensa kenttään `state[step["id"]]`. 

18""" 

19from __future__ import annotations 

20 

21from pathlib import Path 

22from typing import Any 

23 

24# Files-output-helper (resolve_workspace_root) 

25# — sama dual-import-kuvio kuin runner.py:ssa: flat sys.path (testit) vs. 

26# langgraph.* (main.py). 

27try: 

28 from _files_output import ( # noqa: F401 

29 resolve_workspace_root, 

30 ) 

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

32 from langgraph._files_output import ( # noqa: F401 

33 resolve_workspace_root, 

34 ) 

35 

36 

37# --------------------------------------------------------------------------- 

38# ticket-publisher (core/langgraph/tools/ticket_publisher.py) 

39# --------------------------------------------------------------------------- 

40 

41# The publish execution, its origin/date/slug helpers and its definition live 

42# in the per-tool module (tools/ticket_publisher.py); the handler below is a 

43# thin delegating adapter and a module hard failure propagates unchanged. 

44 

45 

46def ticket_publisher(step: dict, state: dict) -> dict: 

47 """Thin adapter over `tools/ticket_publisher.py` (single source). 

48 

49 The publish execution, its helpers and its definition live in the per-tool 

50 module (ADR-0012); a module hard failure propagates unchanged. 

51 """ 

52 try: 

53 from tools.ticket_publisher import ticket_publisher as _exec 

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

55 from langgraph.tools.ticket_publisher import ticket_publisher as _exec 

56 

57 return _exec(step, state) 

58 

59 

60# --------------------------------------------------------------------------- 

61# run-finalizer (core/langgraph/tools/run_finalizer.py) 

62# --------------------------------------------------------------------------- 

63 

64# The finalize execution and its definition live in the per-tool module 

65# (tools/run_finalizer.py); the handler below is a thin delegating adapter and 

66# a module hard failure propagates unchanged. 

67 

68 

69def run_finalizer(step: dict, state: dict) -> dict: 

70 """Thin adapter over `tools/run_finalizer.py` (single source). 

71 

72 The finalize execution and its definition live in the per-tool module 

73 (ADR-0012); a module hard failure propagates unchanged. 

74 """ 

75 try: 

76 from tools.run_finalizer import run_finalizer as _exec 

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

78 from langgraph.tools.run_finalizer import run_finalizer as _exec 

79 

80 return _exec(step, state) 

81 

82 

83# --------------------------------------------------------------------------- 

84# git-checkout (core/langgraph/tools/git_checkout.py) 

85# --------------------------------------------------------------------------- 

86 

87# The checkout, its git helpers/constants and the TICKET branch derivation 

88# live in the per-tool module (tools/git_checkout.py). The test-pinned names 

89# are re-exported because `test_builtins_git_checkout.py` and 

90# `test_builtins_ticket.py` import them from `_builtins`. 

91try: 

92 from tools.git_checkout import ( # noqa: F401 

93 _PR_DIFF_MAX_CHARS, 

94 _PR_DIFF_TRUNCATION_SUFFIX, 

95 _resolve_default_branch, 

96 _ticket_branch, 

97 ) 

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

99 from langgraph.tools.git_checkout import ( # noqa: F401 

100 _PR_DIFF_MAX_CHARS, 

101 _PR_DIFF_TRUNCATION_SUFFIX, 

102 _resolve_default_branch, 

103 _ticket_branch, 

104 ) 

105 

106 

107def _git_checkout(step: dict, state: dict) -> dict: 

108 """Thin adapter over `tools/git_checkout.py` (single source). 

109 

110 The checkout, its git helpers and the TICKET branch derivation live in the 

111 per-tool module (ADR-0012/ADR-0018); a module hard failure propagates 

112 unchanged. 

113 """ 

114 try: 

115 from tools.git_checkout import _git_checkout as _exec 

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

117 from langgraph.tools.git_checkout import _git_checkout as _exec 

118 

119 return _exec(step, state) 

120 

121 

122# --------------------------------------------------------------------------- 

123# files-write 

124# --------------------------------------------------------------------------- 

125 

126 

127def files_write(step: dict, state: dict) -> dict: 

128 """Thin adapter over `tools/files_write.py` (single source). 

129 

130 Writes the template-resolved {path, content} entries into the workspace 

131 root (state.project_root → PROJECT_ROOT-env → /project, ADR-0024). 

132 Validation and writing live in the per-tool module (1 MiB/entry, 

133 10 MiB/step, traversal/.git/symlink escape rejection, fail-fast 

134 ADR-0012); a module error becomes a hard failure with the module's 

135 message and `details` verbatim. 

136 """ 

137 try: 

138 from tools.files_write import files_write as _exec 

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

140 from langgraph.tools.files_write import files_write as _exec 

141 

142 params = step.get("params") or {} 

143 res = _exec(params.get("entries"), resolve_workspace_root(state), step["id"]) 

144 if res.error is not None: 

145 raise _hard_failure("EXECUTION_ERROR", res.error, res.details) 

146 return {**state, step["id"]: {"written": res.written, "count": res.count}} 

147 

148 

149# --------------------------------------------------------------------------- 

150# files-delete 

151# --------------------------------------------------------------------------- 

152 

153 

154def _git_rm(workspace_root: Path, relative: str) -> None: 

155 """Thin adapter over `tools/files_delete.py` (single source). 

156 

157 Removes a tracked file and stages the deletion via the per-tool module; 

158 a module error becomes a hard failure with the module's message and 

159 `details` verbatim. 

160 """ 

161 try: 

162 from tools.files_delete import git_rm as _exec 

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

164 from langgraph.tools.files_delete import git_rm as _exec 

165 

166 res = _exec(workspace_root, relative) 

167 if res.error is not None: 

168 raise _hard_failure("EXECUTION_ERROR", res.error, res.details) 

169 

170 

171def files_delete(step: dict, state: dict) -> dict: 

172 """Thin adapter over `tools/files_delete.py` (single source). 

173 

174 Deletes the template-resolved workspace-relative `paths` (tracked → git 

175 rm, untracked → unlink). Validation, atomicity and deletion live in the 

176 per-tool module (fail-fast ADR-0012); a module error becomes a hard 

177 failure with the module's message and `details` verbatim. 

178 """ 

179 try: 

180 from tools.files_delete import files_delete as _exec 

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

182 from langgraph.tools.files_delete import files_delete as _exec 

183 

184 params = step.get("params") or {} 

185 res = _exec(params.get("paths"), resolve_workspace_root(state), step["id"]) 

186 if res.error is not None: 

187 raise _hard_failure("EXECUTION_ERROR", res.error, res.details) 

188 return {**state, step["id"]: {"deleted": res.deleted, "count": res.count}} 

189 

190 

191# --------------------------------------------------------------------------- 

192# list-dir-entries 

193# --------------------------------------------------------------------------- 

194 

195def _hard_failure(code: str, message: str, details: dict | None = None) -> "RunHardFailure": 

196 try: 

197 from runner import RunHardFailure 

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

199 from langgraph.runner import RunHardFailure 

200 return RunHardFailure(code, message, details) 

201 

202 

203def list_dir_entries(step: dict, state: dict) -> dict: 

204 """Thin adapter over `tools/list_dir_entries.py` (single source). 

205 

206 Lists exactly one level of the resolved `path` (structural skip only, 

207 ADR-0012). Delegates execution to the per-tool module; a module error 

208 becomes a hard failure with the module's message and `details` verbatim. 

209 """ 

210 try: 

211 from tools.list_dir_entries import list_dir_entries as _exec 

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

213 from langgraph.tools.list_dir_entries import list_dir_entries as _exec 

214 

215 step_id = step.get("id", "list_dir_entries") 

216 params = step.get("params") or {} 

217 res = _exec(params.get("path"), state.get("project_root"), step_id) 

218 if res.error is not None: 

219 raise _hard_failure("EXECUTION_ERROR", res.error, res.details) 

220 return {**state, step_id: {"entries": res.entries, "stats": res.stats}} 

221 

222 

223# --------------------------------------------------------------------------- 

224# aggregate (core/docs/builtins/aggregate.md) 

225# --------------------------------------------------------------------------- 

226 

227# The aggregate fold and its dot-path helpers live in the per-tool module 

228# (tools/aggregate.py). `_resolve_dot_path`/`_MISSING_PATH` are re-exported 

229# because `test_builtins_aggregate.py` imports them from `_builtins`. 

230try: 

231 from tools.aggregate import _MISSING_PATH, _resolve_dot_path # noqa: F401 

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

233 from langgraph.tools.aggregate import _MISSING_PATH, _resolve_dot_path # noqa: F401 

234 

235 

236def aggregate(step: dict, state: dict) -> dict: 

237 """Thin adapter over `tools/aggregate.py` (single source). 

238 

239 The aggregate fold, its 13 operators and its dot-path/error-policy helpers 

240 live in the per-tool module (ADR-0012); the definition and the execution 

241 there are the single source. A module hard failure propagates unchanged. 

242 """ 

243 try: 

244 from tools.aggregate import aggregate as _exec 

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

246 from langgraph.tools.aggregate import aggregate as _exec 

247 

248 return _exec(step, state) 

249 

250 

251# --------------------------------------------------------------------------- 

252# Registry helper 

253# --------------------------------------------------------------------------- 

254 

255 

256def register_all(executor: Any) -> None: 

257 """Rekisteröi kaikki builtinit BuiltinExecutoriin (sekä '-' että '_' nimillä).""" 

258 executor.register("list-dir-entries", list_dir_entries) 

259 executor.register("aggregate", aggregate) 

260 executor.register("ticket-publisher", ticket_publisher) 

261 executor.register("run-finalizer", run_finalizer) 

262 executor.register("git-checkout", _git_checkout) 

263 executor.register("files-write", files_write) 

264 executor.register("files-delete", files_delete)