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

159 statements  

« prev     ^ index     » next       coverage.py v7.16.0, created at 2026-09-25 00:49 +0000

1"""Shared ticket HTTP/PM/env helpers for per-tool ticket modules.""" 

2from __future__ import annotations 

3 

4import json 

5import os 

6from typing import Any, Callable, TypeVar 

7from urllib.parse import urlparse 

8 

9 

10_T = TypeVar("_T") 

11 

12 

13def _tool_content( 

14 name: str, 

15 *, 

16 ok: bool, 

17 output: Any = None, 

18 error: str | None = None, 

19) -> str: 

20 """Serialize the JSON content envelope of a per-tool ticket module.""" 

21 payload: dict[str, Any] = {"name": name, "ok": ok} 

22 if output is not None: 

23 payload["output"] = output 

24 if error is not None: 

25 payload["error"] = error 

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

27 

28 

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

30 """Build a code-bearing hard failure for a ticket builtin.""" 

31 try: 

32 from runner import RunHardFailure 

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

34 from langgraph.runner import RunHardFailure 

35 return RunHardFailure(code, message, details) 

36 

37 

38def _require_env(name: str) -> str: 

39 value = os.environ.get(name) 

40 if not value: 

41 raise _hard_failure( 

42 "VALIDATION_ERROR", 

43 f"ticket-publish: required env var {name!r} is not set", 

44 ) 

45 return value 

46 

47 

48def _repo_full_name(git_repo_url: str) -> str: 

49 """Derive owner/repo from the code-repo URL (repo fact). 

50 

51 The API sets REPO_FULL_NAME only in the PR profile; the TICKET profile 

52 derives it deterministically from GIT_REPO_URL (path without .git). 

53 """ 

54 path = urlparse(git_repo_url).path.strip("/") 

55 if path.endswith(".git"): 

56 path = path[:-len(".git")] 

57 if not path or "/" not in path: 

58 raise _hard_failure( 

59 "VALIDATION_ERROR", 

60 f"ticket-publish: cannot derive repo full name from GIT_REPO_URL {git_repo_url!r}", 

61 ) 

62 return path 

63 

64 

65def _raise_for_http_status(resp: httpx.Response, error_label: str) -> None: 

66 """Convert a non-2xx external response into a RESOURCE_ERROR.""" 

67 if resp.status_code >= 400: 

68 raise _hard_failure( 

69 "RESOURCE_ERROR", 

70 f"{error_label} failed ({resp.status_code}): {resp.text[:500]}", 

71 ) 

72 

73 

74def _response_json(resp: httpx.Response, error_label: str) -> Any: 

75 """Parse an external response body; a malformed body is a RESOURCE_ERROR.""" 

76 try: 

77 return resp.json() 

78 except (ValueError, KeyError, TypeError, AttributeError): 

79 raise _hard_failure( 

80 "RESOURCE_ERROR", 

81 f"{error_label} returned invalid JSON: {resp.text[:500]}", 

82 ) from None 

83 

84 

85def _json_list_response(resp: httpx.Response, error_label: str) -> list: 

86 """Parse a 2xx external response body and require a JSON array.""" 

87 data = _response_json(resp, error_label) 

88 if not isinstance(data, list): 

89 raise _hard_failure( 

90 "RESOURCE_ERROR", 

91 f"{error_label} returned non-list JSON: {resp.text[:500]}", 

92 ) 

93 return data 

94 

95 

96def _typed_response(label: str, process: Callable[[], _T]) -> _T: 

97 """Run external-response parsing/dereferencing; malformed shapes fail typed. 

98 

99 Closes the whole untyped-exception class: any ValueError/KeyError/ 

100 TypeError/AttributeError raised while reading a response body converts to 

101 a RESOURCE_ERROR RunHardFailure (never a raw exception). 

102 """ 

103 try: 

104 return process() 

105 except (ValueError, KeyError, TypeError, AttributeError) as exc: 

106 raise _hard_failure( 

107 "RESOURCE_ERROR", 

108 f"{label}: malformed response ({type(exc).__name__}: {exc})", 

109 ) from exc 

110 

111 

112def _http_get(url: str, *, headers: dict, error_label: str) -> httpx.Response: 

113 import httpx 

114 

115 try: 

116 return httpx.get(url, headers=headers) 

117 except httpx.HTTPError as exc: 

118 raise _hard_failure( 

119 "RESOURCE_ERROR", 

120 f"{error_label} request failed: {exc}", 

121 ) from exc 

122 

123 

124def _http_post(url: str, *, headers: dict, payload: dict, error_label: str) -> httpx.Response: 

125 import httpx 

126 

127 try: 

128 return httpx.post(url, json=payload, headers=headers) 

129 except httpx.HTTPError as exc: 

130 raise _hard_failure( 

131 "RESOURCE_ERROR", 

132 f"{error_label} request failed: {exc}", 

133 ) from exc 

134 

135 

136def _http_patch(url: str, *, headers: dict, payload: dict, error_label: str) -> httpx.Response: 

137 import httpx 

138 

139 try: 

140 return httpx.patch(url, json=payload, headers=headers) 

141 except httpx.HTTPError as exc: 

142 raise _hard_failure( 

143 "RESOURCE_ERROR", 

144 f"{error_label} request failed: {exc}", 

145 ) from exc 

146 

147 

148def _require_pm_env(builtin: str) -> tuple[str, str, str]: 

149 """Guard RUN_PROFILE=TICKET and require the PM API envs (fail-fast). 

150 

151 Returns (api_url, api_key, run_id); API_URL is joined without a 

152 trailing slash so the caller can append the internal API path. 

153 """ 

154 run_profile = os.environ.get("RUN_PROFILE") 

155 if run_profile != "TICKET": 

156 raise _hard_failure( 

157 "VALIDATION_ERROR", 

158 f"{builtin} requires RUN_PROFILE=TICKET (got {run_profile!r})", 

159 ) 

160 missing = [name for name in ("API_URL", "API_KEY", "RUN_ID") if not os.environ.get(name)] 

161 if missing: 

162 raise _hard_failure( 

163 "VALIDATION_ERROR", 

164 f"{builtin}: required env var(s) {', '.join(repr(n) for n in missing)} not set", 

165 ) 

166 return ( 

167 os.environ["API_URL"].rstrip("/"), 

168 os.environ["API_KEY"], 

169 os.environ["RUN_ID"], 

170 ) 

171 

172 

173def _valid_tickets(tickets: Any) -> bool: 

174 """Validate the pinned tickets-list shape: list of typed objects.""" 

175 if not isinstance(tickets, list): 

176 return False 

177 for ticket in tickets: 

178 if not ( 

179 isinstance(ticket, dict) 

180 and isinstance(ticket.get("id"), int) 

181 and isinstance(ticket.get("title"), str) 

182 and isinstance(ticket.get("done"), bool) 

183 and isinstance(ticket.get("description"), str) 

184 ): 

185 return False 

186 return True 

187 

188 

189def _fetch_open_pm_tickets(api_url: str, api_key: str, run_id: str, builtin: str) -> tuple[Any, str]: 

190 """GET the open PM tickets via the internal endpoint; fail-fast (ADR-0012). 

191 

192 Returns (payload, raw_text) so callers can include the response body in 

193 shape-validation errors. Non-2xx response, invalid JSON, or an httpx 

194 transport error → RunHardFailure("RESOURCE_ERROR"). 

195 """ 

196 label = f"{builtin}: PM tickets GET" 

197 resp = _http_get( 

198 f"{api_url}/api/v1/internal/pm/tickets", 

199 headers={"X-API-Key": api_key, "X-Run-ID": run_id}, 

200 error_label=label, 

201 ) 

202 _raise_for_http_status(resp, label) 

203 payload = _response_json(resp, label) 

204 return payload, resp.text 

205 

206 

207def _valid_dedupe_tickets(tickets: Any) -> bool: 

208 """Validate the dedupe tickets shape: open items keyed by title.""" 

209 if not isinstance(tickets, list): 

210 return False 

211 for ticket in tickets: 

212 if not isinstance(ticket, dict): 

213 return False 

214 parent = ticket.get("parent_task_id") 

215 if parent is not None and not isinstance(parent, int): 

216 return False 

217 if not ( 

218 isinstance(ticket.get("id"), int) 

219 and isinstance(ticket.get("title"), str) 

220 and isinstance(ticket.get("done"), bool) 

221 and isinstance(ticket.get("description"), str) 

222 ): 

223 return False 

224 return True 

225 

226 

227def _dedupe_skip(tickets: list, title: str, parent_task_id: int | None) -> dict | None: 

228 """Return the would-be created item when an open ticket matches the title. 

229 

230 Match = ticket title equals the params title (both trimmed). The 

231 skipped item carries the would-be created values, not the matched 

232 ticket's. 

233 """ 

234 normalized = title.strip() 

235 for ticket in tickets: 

236 if ticket["title"].strip() == normalized: 

237 return {"title": normalized, "parent_task_id": parent_task_id} 

238 return None 

239 

240 

241class TicketOutcome(Exception): 

242 """Outcome-coded stop for TICKET-profile runs (exit-code lanes). 

243 

244 Raised by the decision builtins (ticket-dor / ticket-review) to end 

245 the run with a defined outcome instead of a generic failure: 

246 

247 code → exit_code in the final status-callback PATCH (the API 

248 owns the exit-code → lane mapping, e.g. 10/22 → 

249 open_questions, 20 → needs_rework, 21 → pr_rejected) 

250 message → error.message in the final PATCH 

251 questions → optional list of open questions; in the TICKET profile 

252 the entrypoint includes them as UpdateRunRequest.questions 

253 (non-empty only) so the API can post them as a ticket 

254 comment instead of the generic result comment 

255 """ 

256 

257 def __init__( 

258 self, 

259 code: int, 

260 message: str, 

261 questions: list[str] | None = None, 

262 ) -> None: 

263 super().__init__(message) 

264 self.code = code 

265 self.message = message 

266 self.questions = questions 

267 

268 

269def _submission_tool() -> Any: 

270 """The synthetic `submit_decision` tool object (lazy import, no cycle).""" 

271 try: 

272 from tools.submit_decision import TOOL as _SUBMIT_DECISION_TOOL 

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

274 from langgraph.tools.submit_decision import TOOL as _SUBMIT_DECISION_TOOL 

275 return _SUBMIT_DECISION_TOOL 

276 

277 

278def _decision_failure(message: str, step_id: str) -> "RunHardFailure": 

279 """Build a code-bearing hard failure for a decision step's agent error.""" 

280 return _hard_failure( 

281 "AGENT_ERROR", 

282 message, 

283 {"error_code": "AGENT_ERROR", "step_id": step_id}, 

284 ) 

285 

286 

287def _submitted_decision(agent_name: str, tool_meta: Any) -> dict: 

288 """Read the decision from the inner call's recorded dispatched tool calls.""" 

289 tool_name = _submission_tool().name 

290 dispatched = ( 

291 tool_meta.get("dispatched_calls") if isinstance(tool_meta, dict) else None 

292 ) 

293 for call in dispatched or []: 

294 if isinstance(call, dict) and call.get("name") == tool_name: 

295 arguments = call.get("arguments") 

296 if not isinstance(arguments, dict): 

297 raise ValueError( 

298 f"{agent_name}: {tool_name} arguments must be a " 

299 f"JSON object (got {type(arguments).__name__})" 

300 ) 

301 return arguments 

302 raise ValueError( 

303 f"{agent_name}: LLM did not deliver a decision via the " 

304 f"{tool_name} tool call" 

305 ) 

306 

307 

308def _llm_json_decision( 

309 invoke_agent: Any, 

310 step: dict, 

311 state: dict, 

312 *, 

313 agent_name: str, 

314 system_prompt: str, 

315 context: Any, 

316) -> dict: 

317 """Fixed-prompt LLM decision: invoke_agent(spec, step, state) → dict. 

318 

319 The decision builtins have no manifest agent spec — they decide with 

320 a fixed system prompt through the pipeline's invoke_agent_fn 

321 (GraphBuilder wires it via BuiltinExecutor.invoke_agent_fn). The 

322 synthetic inner agent is given exactly one native tool, 

323 `submit_decision`, and the decision object is the arguments of that 

324 tool call (recorded by the engine in the parallel core key 

325 ``step_tool_meta[step_id]["dispatched_calls"]``); the envelope 

326 ``content`` is opaque and may be absent. A non-envelope 

327 step result, an error envelope, a missing submission, or non-object 

328 submission arguments is a hard error (ADR-0012 — no silent fallback). 

329 """ 

330 try: 

331 from _envelope import is_error_envelope 

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

333 from langgraph._envelope import is_error_envelope 

334 

335 try: 

336 from tools.contract import RESOLVED_TOOLS_KEY 

337 from tools.resolver import ToolResolver, build_platform_registry 

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

339 from langgraph.tools.contract import RESOLVED_TOOLS_KEY 

340 from langgraph.tools.resolver import ToolResolver, build_platform_registry 

341 

342 step_id = step.get("id", agent_name) 

343 if invoke_agent is None: 

344 raise _decision_failure( 

345 f"{agent_name}: LLM invoker is not wired — " 

346 "GraphBuilder must set BuiltinExecutor.invoke_agent_fn", 

347 step_id, 

348 ) 

349 agent_spec: dict[str, Any] = { 

350 "metadata": {"name": agent_name}, 

351 "spec": {"model": {}, "system_prompt": system_prompt}, 

352 } 

353 resolved_step: dict[str, Any] = { 

354 "id": step_id, 

355 "input": context, 

356 "params": {}, 

357 RESOLVED_TOOLS_KEY: ToolResolver(build_platform_registry()).resolve( 

358 [_submission_tool().name] 

359 ), 

360 } 

361 result = invoke_agent(agent_spec, resolved_step, state) 

362 out = result.get(step_id) if isinstance(result, dict) else None 

363 if not isinstance(out, dict): 

364 raise _decision_failure( 

365 f"{agent_name}: LLM call did not return a JSON object " 

366 f"(got {type(out).__name__})", 

367 step_id, 

368 ) 

369 if is_error_envelope(out) or out.get("error"): 

370 error = out.get("error") 

371 if isinstance(error, str): 

372 message = f"{agent_name}: LLM call failed: {error}" 

373 else: 

374 message = f"{agent_name}: LLM call failed (error envelope without message)" 

375 raise _decision_failure(message, step_id) 

376 step_tool_meta = result.get("step_tool_meta") 

377 tool_meta = ( 

378 step_tool_meta.get(step_id) if isinstance(step_tool_meta, dict) else None 

379 ) 

380 try: 

381 return _submitted_decision(agent_name, tool_meta) 

382 except ValueError as exc: 

383 raise _decision_failure(str(exc), step_id) from exc 

384 

385 

386def _as_questions(raw: Any, builtin: str) -> list[str]: 

387 """Validate the LLM 'questions' field: list of strings (None → []).""" 

388 if raw is None: 

389 return [] 

390 if not isinstance(raw, list) or not all(isinstance(q, str) for q in raw): 

391 raise _hard_failure( 

392 "AGENT_ERROR", 

393 f"{builtin}: 'questions' must be a list of strings, got {raw!r}", 

394 ) 

395 return raw