Coverage for core/langgraph/tools/ticket_approval.py: 97%
61 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-25 05:04 +0000
1"""Per-tool module for the `ticket-approval` deterministic builtin.
3Owns the tool's model-facing definition, its instruction surface (none) and the
4deterministic approval-marker detection itself. The definition and the execution
5here are the single source; the legacy builtin site delegates to the pure
6`ticket_approval` handler.
7"""
8from __future__ import annotations
10import os
11import re
12from typing import Any
14from .contract import ToolContext, ToolResult
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
21TICKET_APPROVAL_TOOL_NAME = "ticket-approval"
22DEFAULT_STEP_ID = "ticket_approval"
24TICKET_APPROVAL_DEFINITION: dict[str, Any] = {
25 "type": "function",
26 "function": {
27 "name": TICKET_APPROVAL_TOOL_NAME,
28 "description": (
29 "Detect the human approval marker with `ticket-approval`: pass the "
30 "ticket context as `description` and `comments`; a human comment "
31 "whose entire content is `APPROVED` (case-insensitive) passes, and "
32 "the result reports `passed` and the matching `evidence`."
33 ),
34 "parameters": {
35 "type": "object",
36 "properties": {
37 "description": {
38 "type": "string",
39 "description": (
40 "The `description` field of the `ticket_context` object."
41 ),
42 },
43 "comments": {
44 "type": "array",
45 "items": {"type": "object"},
46 "description": (
47 "The `comments` list of the `ticket_context` object; each "
48 "comment has string `author` and `text` fields."
49 ),
50 },
51 },
52 "required": ["description"],
53 "additionalProperties": False,
54 },
55 },
56}
59def _hard_failure_type() -> type[Exception]:
60 """The runner's hard-failure type, resolved without an import cycle."""
61 try:
62 from runner import RunHardFailure
63 except ImportError: # pragma: no cover - main.py run (qualified imports)
64 from langgraph.runner import RunHardFailure
65 return RunHardFailure
68_APPROVAL_MARKER = re.compile(r"^\s*APPROVED\s*$", re.IGNORECASE)
71def _approval_marker_text(text: str) -> str | None:
72 """Return the trimmed comment when its whole content is the APPROVED marker."""
73 stripped = text.strip()
74 if _APPROVAL_MARKER.fullmatch(stripped):
75 return stripped
76 return None
79def ticket_approval(step: dict, state: dict) -> dict:
80 """Deterministic Stage A approval-marker detection (no LLM call).
82 Scans the resolved step.input (the ticket_context object {"description",
83 "comments", "blocked_by"}; the runner resolves the input before the builtin
84 call) for human comments (author not "platform"/"unknown") and recognises
85 approval only from a comment whose entire content, whitespace-trimmed, is
86 APPROVED (case-insensitive) - a comment containing any other text (prose,
87 questions, justifications, multi-line comments) is not approval. The result
88 in the step result is {"passed": bool, "evidence": {"author",
89 "comment_index", "snippet"} | null} - evidence from the first match (snippet
90 = the whole trimmed comment text), null when there is no match.
92 An invalid/changed input shape -> hard failure (fail-fast, ADR-0012) - the
93 error names the missing part. Wrong RUN_PROFILE -> hard failure.
94 """
95 run_profile = os.environ.get("RUN_PROFILE")
96 if run_profile != "TICKET":
97 raise _hard_failure(
98 "VALIDATION_ERROR",
99 f"ticket-approval requires RUN_PROFILE=TICKET (got {run_profile!r})",
100 )
102 ticket_context = step.get("input")
103 if not isinstance(ticket_context, dict):
104 raise _hard_failure(
105 "VALIDATION_ERROR",
106 "ticket-approval: step input must be a ticket_context object "
107 f"with 'description' and 'comments', got {type(ticket_context).__name__}",
108 )
109 if not isinstance(ticket_context.get("description"), str):
110 raise _hard_failure(
111 "VALIDATION_ERROR",
112 "ticket-approval: ticket_context['description'] must be a string",
113 )
114 comments = ticket_context.get("comments")
115 if not isinstance(comments, list) or not all(
116 isinstance(c, dict)
117 and isinstance(c.get("author"), str)
118 and isinstance(c.get("text"), str)
119 for c in comments
120 ):
121 raise _hard_failure(
122 "VALIDATION_ERROR",
123 "ticket-approval: ticket_context['comments'] must be a list of "
124 "objects with string 'author' and 'text' fields",
125 )
127 for index, comment in enumerate(comments):
128 if comment["author"] in ("platform", "unknown"):
129 continue
130 snippet = _approval_marker_text(comment["text"])
131 if snippet is not None:
132 step_id = step.get("id", "ticket_approval")
133 evidence = {"author": comment["author"], "comment_index": index, "snippet": snippet}
134 return {**state, step_id: {"passed": True, "evidence": evidence}}
136 step_id = step.get("id", "ticket_approval")
137 return {**state, step_id: {"passed": False, "evidence": None}}
140class _TicketApprovalTool:
141 name = TICKET_APPROVAL_TOOL_NAME
143 def definition(self) -> dict[str, Any]:
144 return TICKET_APPROVAL_DEFINITION
146 def instruction(self) -> str | None:
147 return None
149 def params(self) -> list[dict[str, Any]] | None:
150 return None
152 def execute(self, context: ToolContext) -> ToolResult:
153 step_id = (context.step or {}).get("id") or DEFAULT_STEP_ID
154 args = context.args or {}
155 synthetic_step = {
156 "id": step_id,
157 "input": {
158 "description": args.get("description"),
159 "comments": args.get("comments"),
160 },
161 }
162 try:
163 new_state = ticket_approval(synthetic_step, context.state or {})
164 except _hard_failure_type() as exc:
165 return ToolResult(
166 ok=False,
167 content=_tool_content(TICKET_APPROVAL_TOOL_NAME, ok=False, error=exc.message),
168 error=exc.message,
169 )
170 output = new_state[step_id]
171 return ToolResult(
172 ok=True,
173 content=_tool_content(TICKET_APPROVAL_TOOL_NAME, ok=True, output=output),
174 state_patch={step_id: output},
175 )
178TOOL = _TicketApprovalTool()