Coverage for core/langgraph/tools/ticket_context.py: 96%

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

2 

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

4TICKET_CONTEXT env → state loading itself. The definition and the execution here 

5are the single source; the legacy builtin site delegates to the pure 

6`ticket_context` handler. 

7""" 

8from __future__ import annotations 

9 

10import json 

11import os 

12from typing import Any 

13 

14from .contract import ToolContext, ToolResult 

15 

16try: 

17 from tools._ticket_common import _hard_failure, _tool_content 

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

19 from langgraph.tools._ticket_common import _hard_failure, _tool_content 

20 

21TICKET_CONTEXT_TOOL_NAME = "ticket-context" 

22DEFAULT_STEP_ID = "ticket_context" 

23 

24TICKET_CONTEXT_DEFINITION: dict[str, Any] = { 

25 "type": "function", 

26 "function": { 

27 "name": TICKET_CONTEXT_TOOL_NAME, 

28 "description": ( 

29 "Load the ticket context with `ticket-context`: it reads the " 

30 "`TICKET_CONTEXT` environment JSON (description plus author-tagged " 

31 "comments) and stores it under the step id and the top-level " 

32 "`ticket_context` key for later steps to read." 

33 ), 

34 "parameters": { 

35 "type": "object", 

36 "properties": {}, 

37 "additionalProperties": False, 

38 }, 

39 }, 

40} 

41 

42 

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

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

45 try: 

46 from runner import RunHardFailure 

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

48 from langgraph.runner import RunHardFailure 

49 return RunHardFailure 

50 

51 

52def ticket_context(step: dict, state: dict) -> dict: 

53 """Store the TICKET_CONTEXT env in state as pipeline-agent input. 

54 

55 Env: TICKET_CONTEXT — JSON {"description": str, "comments": 

56 [{"author": str, "text": str}, ...]} (ticket description + full 

57 comment history with author tags). The result is stored under the 

58 step-id key (default "ticket_context") AND the top-level key 

59 "ticket_context" (state dot-path, e.g. ticket_context.comments). 

60 

61 Missing env or invalid JSON/structure → hard failure (fail-fast, 

62 ADR-0012): the API hard-fails on the env setting (Phase 2), so this 

63 is the last line of defense. The legacy comments form (list of 

64 strings) is a hard failure — comments is a list of author-tagged 

65 objects. Wrong RUN_PROFILE → hard failure. 

66 """ 

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

68 if run_profile != "TICKET": 

69 raise _hard_failure( 

70 "VALIDATION_ERROR", 

71 f"ticket-context requires RUN_PROFILE=TICKET (got {run_profile!r})", 

72 ) 

73 

74 raw_context = os.environ.get("TICKET_CONTEXT") 

75 if not raw_context: 

76 raise _hard_failure( 

77 "VALIDATION_ERROR", 

78 "ticket-context: required env var 'TICKET_CONTEXT' is not set — " 

79 "the API could not fetch the ticket context (Vikunja-driven runs only)", 

80 ) 

81 try: 

82 ticket_context = json.loads(raw_context) 

83 except json.JSONDecodeError as exc: 

84 raise _hard_failure( 

85 "VALIDATION_ERROR", 

86 f"ticket-context: TICKET_CONTEXT is not valid JSON: {exc}", 

87 ) from exc 

88 if not isinstance(ticket_context, dict): 

89 raise _hard_failure( 

90 "VALIDATION_ERROR", 

91 "ticket-context: TICKET_CONTEXT must be a JSON object, " 

92 f"got {type(ticket_context).__name__}", 

93 ) 

94 if not isinstance(ticket_context.get("description"), str): 

95 raise _hard_failure( 

96 "VALIDATION_ERROR", 

97 "ticket-context: TICKET_CONTEXT['description'] must be a string", 

98 ) 

99 comments = ticket_context.get("comments") 

100 if not isinstance(comments, list) or not all( 

101 isinstance(c, dict) 

102 and isinstance(c.get("author"), str) 

103 and isinstance(c.get("text"), str) 

104 for c in comments 

105 ): 

106 raise _hard_failure( 

107 "VALIDATION_ERROR", 

108 "ticket-context: TICKET_CONTEXT['comments'] must be a list of " 

109 "objects with string 'author' and 'text' fields", 

110 ) 

111 

112 step_id = step.get("id", "ticket_context") 

113 return {**state, step_id: ticket_context, "ticket_context": ticket_context} 

114 

115 

116class _TicketContextTool: 

117 name = TICKET_CONTEXT_TOOL_NAME 

118 

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

120 return TICKET_CONTEXT_DEFINITION 

121 

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

123 return None 

124 

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

126 return None 

127 

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

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

130 synthetic_step = {"id": step_id} 

131 try: 

132 new_state = ticket_context(synthetic_step, context.state or {}) 

133 except _hard_failure_type() as exc: 

134 return ToolResult( 

135 ok=False, 

136 content=_tool_content(TICKET_CONTEXT_TOOL_NAME, ok=False, error=exc.message), 

137 error=exc.message, 

138 ) 

139 output = new_state[step_id] 

140 return ToolResult( 

141 ok=True, 

142 content=_tool_content(TICKET_CONTEXT_TOOL_NAME, ok=True, output=output), 

143 state_patch={step_id: output, "ticket_context": new_state["ticket_context"]}, 

144 ) 

145 

146 

147TOOL = _TicketContextTool()