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 and the TICKET branch derivation live in the 

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

89# 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 _resolve_default_branch, 

94 _ticket_branch, 

95 ) 

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

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

98 _resolve_default_branch, 

99 _ticket_branch, 

100 ) 

101 

102 

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

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

105 

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

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

108 unchanged. 

109 """ 

110 try: 

111 from tools.git_checkout import _git_checkout as _exec 

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

113 from langgraph.tools.git_checkout import _git_checkout as _exec 

114 

115 return _exec(step, state) 

116 

117 

118# --------------------------------------------------------------------------- 

119# files-write 

120# --------------------------------------------------------------------------- 

121 

122 

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

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

125 

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

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

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

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

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

131 message and `details` verbatim. 

132 """ 

133 try: 

134 from tools.files_write import files_write as _exec 

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

136 from langgraph.tools.files_write import files_write as _exec 

137 

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

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

140 if res.error is not None: 

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

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

143 

144 

145# --------------------------------------------------------------------------- 

146# files-delete 

147# --------------------------------------------------------------------------- 

148 

149 

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

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

152 

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

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

155 `details` verbatim. 

156 """ 

157 try: 

158 from tools.files_delete import git_rm as _exec 

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

160 from langgraph.tools.files_delete import git_rm as _exec 

161 

162 res = _exec(workspace_root, relative) 

163 if res.error is not None: 

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

165 

166 

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

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

169 

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

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

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

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

174 """ 

175 try: 

176 from tools.files_delete import files_delete as _exec 

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

178 from langgraph.tools.files_delete import files_delete as _exec 

179 

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

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

182 if res.error is not None: 

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

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

185 

186 

187# --------------------------------------------------------------------------- 

188# list-dir-entries 

189# --------------------------------------------------------------------------- 

190 

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

192 try: 

193 from runner import RunHardFailure 

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

195 from langgraph.runner import RunHardFailure 

196 return RunHardFailure(code, message, details) 

197 

198 

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

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

201 

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

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

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

205 """ 

206 try: 

207 from tools.list_dir_entries import list_dir_entries as _exec 

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

209 from langgraph.tools.list_dir_entries import list_dir_entries as _exec 

210 

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

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

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

214 if res.error is not None: 

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

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

217 

218 

219# --------------------------------------------------------------------------- 

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

221# --------------------------------------------------------------------------- 

222 

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

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

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

226try: 

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

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

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

230 

231 

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

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

234 

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

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

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

238 """ 

239 try: 

240 from tools.aggregate import aggregate as _exec 

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

242 from langgraph.tools.aggregate import aggregate as _exec 

243 

244 return _exec(step, state) 

245 

246 

247# --------------------------------------------------------------------------- 

248# Registry helper 

249# --------------------------------------------------------------------------- 

250 

251 

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

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

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

255 executor.register("aggregate", aggregate) 

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

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

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

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

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