Coverage for core/langgraph/tools/_ticket_common.py: 97%
159 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-25 03:42 +0000
1"""Shared ticket HTTP/PM/env helpers for per-tool ticket modules."""
2from __future__ import annotations
4import json
5import os
6from typing import Any, Callable, TypeVar
7from urllib.parse import urlparse
10_T = TypeVar("_T")
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)
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)
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
48def _repo_full_name(git_repo_url: str) -> str:
49 """Derive owner/repo from the code-repo URL (repo fact).
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
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 )
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
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
96def _typed_response(label: str, process: Callable[[], _T]) -> _T:
97 """Run external-response parsing/dereferencing; malformed shapes fail typed.
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
112def _http_get(url: str, *, headers: dict, error_label: str) -> httpx.Response:
113 import httpx
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
124def _http_post(url: str, *, headers: dict, payload: dict, error_label: str) -> httpx.Response:
125 import httpx
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
136def _http_patch(url: str, *, headers: dict, payload: dict, error_label: str) -> httpx.Response:
137 import httpx
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
148def _require_pm_env(builtin: str) -> tuple[str, str, str]:
149 """Guard RUN_PROFILE=TICKET and require the PM API envs (fail-fast).
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 )
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
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).
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
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
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.
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
241class TicketOutcome(Exception):
242 """Outcome-coded stop for TICKET-profile runs (exit-code lanes).
244 Raised by the decision builtins (ticket-dor / ticket-review) to end
245 the run with a defined outcome instead of a generic failure:
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 """
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
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
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 )
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 )
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.
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
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
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
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