Coverage for core/langgraph/tools/ticket_search.py: 95%

1"""Per-tool module for the `ticket-search` deterministic builtin. 

2 

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

4open-PM-tickets GET execution itself (ADR-0012). The definition and the 

5execution here are the single source; the legacy builtin site delegates to the 

6pure `ticket_search` handler. 

7""" 

8from __future__ import annotations 

9 

10from typing import Any 

11 

12from .contract import ToolContext, ToolResult 

13 

14try: 

15 from tools._ticket_common import ( 

16 _fetch_open_pm_tickets, _hard_failure, _require_pm_env, _tool_content, 

17 _valid_tickets, 

18 ) 

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

20 from langgraph.tools._ticket_common import ( 

21 _fetch_open_pm_tickets, _hard_failure, _require_pm_env, _tool_content, 

22 _valid_tickets, 

23 ) 

24 

25TICKET_SEARCH_TOOL_NAME = "ticket-search" 

26DEFAULT_STEP_ID = "ticket_search" 

27 

28TICKET_SEARCH_DEFINITION: dict[str, Any] = { 

29 "type": "function", 

30 "function": { 

31 "name": TICKET_SEARCH_TOOL_NAME, 

32 "description": ( 

33 "List the open work items with `ticket-search` instead of querying the " 

34 "project-management system yourself; the result reports `tickets` " 

35 "(`id`, `title`, `done`, `description`) resolved through the run's PM " 

36 "configuration." 

37 ), 

38 "parameters": { 

39 "type": "object", 

40 "properties": {}, 

41 "additionalProperties": False, 

42 }, 

43 }, 

44} 

45 

46 

47def _hard_failure_type() -> type[Exception]: 

48 """The runner's hard-failure type, resolved without an import cycle.""" 

49 try: 

50 from runner import RunHardFailure 

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

52 from langgraph.runner import RunHardFailure 

53 return RunHardFailure 

54 

55 

56def ticket_search(step: dict, state: dict) -> dict: 

57 """Fetch the PM system's open work items via the API internal endpoint. 

58 

59 GET {API_URL}/api/v1/internal/pm/tickets with X-API-Key and X-Run-ID 

60 headers; the API resolves the run's pipeline configuration into the PM 

61 settings (the engine is PM-agnostic — no product vocabulary). Result into 

62 the step result {"tickets": [{id, title, done, description}]} — an empty 

63 list is allowed. Non-2xx response or unexpected payload shape → hard 

64 failure (fail-fast, ADR-0012). 

65 

66 Env vars used: 

67 RUN_PROFILE — must be "TICKET" (guard; anything else raises) 

68 API_URL, API_KEY, RUN_ID — credentials for the API internal port 

69 (X-Run-ID is passed as the X-Run-ID header) 

70 """ 

71 api_url, api_key, run_id = _require_pm_env("ticket-search") 

72 payload, raw = _fetch_open_pm_tickets(api_url, api_key, run_id, "ticket-search") 

73 tickets = payload.get("tickets") if isinstance(payload, dict) else None 

74 if not _valid_tickets(tickets): 

75 raise _hard_failure( 

76 "RESOURCE_ERROR", 

77 "ticket-search: PM tickets GET response 'tickets' must be a list of " 

78 f"{{id: int, title: str, done: bool, description: str}}: {raw[:500]}", 

79 ) 

80 return {**state, step.get("id", "ticket_search"): {"tickets": tickets}} 

81 

82 

83class _TicketSearchTool: 

84 name = TICKET_SEARCH_TOOL_NAME 

85 

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

87 return TICKET_SEARCH_DEFINITION 

88 

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

90 return None 

91 

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

93 return None 

94 

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

96 step_id = (context.step or {}).get("id") or DEFAULT_STEP_ID 

97 synthetic_step = {"id": step_id} 

98 try: 

99 new_state = ticket_search(synthetic_step, context.state or {}) 

100 except _hard_failure_type() as exc: 

101 return ToolResult( 

102 ok=False, 

103 content=_tool_content(TICKET_SEARCH_TOOL_NAME, ok=False, error=exc.message), 

104 error=exc.message, 

105 ) 

106 output = new_state[step_id] 

107 return ToolResult( 

108 ok=True, 

109 content=_tool_content(TICKET_SEARCH_TOOL_NAME, ok=True, output=output), 

110 state_patch={step_id: output}, 

111 ) 

112 

113 

114TOOL = _TicketSearchTool()