Coverage for core/langgraph/tools/ticket_create.py: 97%
65 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-25 03:05 +0000
1"""Per-tool module for the `ticket-create` deterministic builtin.
3Owns the tool's model-facing definition, its instruction surface (none) and the
4idempotent open-PM-ticket POST execution itself (ADR-0012). The definition and
5the execution here are the single source; the legacy builtin site delegates to
6the pure `ticket_create` handler.
7"""
8from __future__ import annotations
10from typing import Any
12from .contract import ToolContext, ToolResult
14try:
15 from tools._ticket_common import (
16 _dedupe_skip, _fetch_open_pm_tickets, _hard_failure, _http_post,
17 _raise_for_http_status, _require_pm_env, _response_json, _tool_content,
18 _valid_dedupe_tickets,
19 )
20except ImportError: # pragma: no cover — main.py run (qualified imports)
21 from langgraph.tools._ticket_common import (
22 _dedupe_skip, _fetch_open_pm_tickets, _hard_failure, _http_post,
23 _raise_for_http_status, _require_pm_env, _response_json, _tool_content,
24 _valid_dedupe_tickets,
25 )
27TICKET_CREATE_TOOL_NAME = "ticket-create"
28DEFAULT_STEP_ID = "ticket_create"
30TICKET_CREATE_DEFINITION: dict[str, Any] = {
31 "type": "function",
32 "function": {
33 "name": TICKET_CREATE_TOOL_NAME,
34 "description": (
35 "Create a single work item with `ticket-create` (idempotent): set "
36 "`title` (required), optional `description` and optional "
37 "`parent_task_id`; a matching open title is skipped and the "
38 "result reports `created` and `skipped`."
39 ),
40 "parameters": {
41 "type": "object",
42 "properties": {
43 "title": {
44 "type": "string",
45 "description": (
46 "The work item title; required and must be a non-empty string."
47 ),
48 },
49 "description": {
50 "type": "string",
51 "description": (
52 "Optional work item description; defaults to an empty string."
53 ),
54 },
55 "parent_task_id": {
56 "type": "integer",
57 "description": (
58 "Optional id of the parent work item; omit to create a "
59 "top-level item."
60 ),
61 },
62 },
63 "required": ["title"],
64 "additionalProperties": False,
65 },
66 },
67}
70def _hard_failure_type() -> type[Exception]:
71 """The runner's hard-failure type, resolved without an import cycle."""
72 try:
73 from runner import RunHardFailure
74 except ImportError: # pragma: no cover — main.py run (qualified imports)
75 from langgraph.runner import RunHardFailure
76 return RunHardFailure
79def ticket_create(step: dict, state: dict) -> dict:
80 """Create one PM work item via the API internal endpoint (one per call).
82 POST {API_URL}/api/v1/internal/pm/tickets with X-API-Key and X-Run-ID
83 headers; the API resolves the run's PM configuration and attaches the
84 X-Run-ID to the traceability record. Body: title (required), description
85 (optional, default ""), parent_task_id (optional int — absent
86 → no parent relation). 201 → step result
87 {"created": [{id, url, title}]}. Non-2xx response or unexpected
88 payload shape → hard failure (fail-fast, ADR-0012).
90 Creation is preceded by a deterministic dedupe check: open
91 PM work items (ticket-search mechanics) are searched for the same title
92 (trimmed); a match → no creation (idempotent re-run) and
93 step result {"created": [], "skipped": [{title, parent_task_id}]}.
94 GET fails or payload is unexpected → hard failure (no creation).
96 Env vars used:
97 RUN_PROFILE — must be "TICKET" (guard; anything else raises)
98 API_URL, API_KEY, RUN_ID — credentials for the API internal port
99 (X-Run-ID is passed as the X-Run-ID header)
100 Step params:
101 title — required non-empty string
102 description — optional string (default "")
103 parent_task_id — optional int
104 """
105 api_url, api_key, run_id = _require_pm_env("ticket-create")
107 params = step.get("params") or {}
108 title = params.get("title")
109 if not isinstance(title, str) or not title.strip():
110 raise _hard_failure(
111 "VALIDATION_ERROR",
112 "ticket-create: params['title'] is required (non-empty string)",
113 )
114 description = params.get("description")
115 if description is None:
116 description = ""
117 elif not isinstance(description, str):
118 raise _hard_failure(
119 "VALIDATION_ERROR",
120 f"ticket-create: params['description'] must be a string, got {description!r}",
121 )
122 parent_task_id = params.get("parent_task_id")
123 if parent_task_id is not None:
124 if not isinstance(parent_task_id, int):
125 raise _hard_failure(
126 "VALIDATION_ERROR",
127 "ticket-create: params['parent_task_id'] must be an int, "
128 f"got {parent_task_id!r}",
129 )
131 payload, raw = _fetch_open_pm_tickets(api_url, api_key, run_id, "ticket-create")
132 tickets = payload.get("tickets") if isinstance(payload, dict) else None
133 if not _valid_dedupe_tickets(tickets):
134 raise _hard_failure(
135 "RESOURCE_ERROR",
136 "ticket-create: PM tickets GET response 'tickets' must be a list of "
137 f"{{id: int, title: str, done: bool, description: str}}: {raw[:500]}",
138 )
139 skipped = _dedupe_skip(tickets, title, parent_task_id)
140 if skipped is not None:
141 return {**state, step.get("id", "ticket_create"): {"created": [], "skipped": [skipped]}}
143 body: dict[str, Any] = {"title": title, "description": description}
144 if parent_task_id is not None:
145 body["parent_task_id"] = parent_task_id
147 label = "ticket-create: PM tickets POST"
148 resp = _http_post(
149 f"{api_url}/api/v1/internal/pm/tickets",
150 headers={"X-API-Key": api_key, "X-Run-ID": run_id},
151 payload=body,
152 error_label=label,
153 )
154 _raise_for_http_status(resp, label)
155 payload = _response_json(resp, label)
156 if not (
157 isinstance(payload, dict)
158 and isinstance(payload.get("id"), int)
159 and isinstance(payload.get("url"), str)
160 and isinstance(payload.get("title"), str)
161 ):
162 raise _hard_failure(
163 "RESOURCE_ERROR",
164 "ticket-create: PM tickets POST response must be "
165 f"{{id: int, url: str, title: str}}: {resp.text[:500]}",
166 )
167 created = {"id": payload["id"], "url": payload["url"], "title": payload["title"]}
168 return {**state, step.get("id", "ticket_create"): {"created": [created], "skipped": []}}
171class _TicketCreateTool:
172 name = TICKET_CREATE_TOOL_NAME
174 def definition(self) -> dict[str, Any]:
175 return TICKET_CREATE_DEFINITION
177 def instruction(self) -> str | None:
178 return None
180 def params(self) -> list[dict[str, Any]] | None:
181 return None
183 def execute(self, context: ToolContext) -> ToolResult:
184 step = context.step or {}
185 step_id = step.get("id", DEFAULT_STEP_ID)
186 args = context.args or {}
187 synthetic_step = {
188 "id": step_id,
189 "params": {
190 key: args.get(key)
191 for key in ("title", "description", "parent_task_id")
192 },
193 }
194 try:
195 new_state = ticket_create(synthetic_step, context.state or {})
196 except _hard_failure_type() as exc:
197 return ToolResult(
198 ok=False,
199 content=_tool_content(TICKET_CREATE_TOOL_NAME, ok=False, error=exc.message),
200 error=exc.message,
201 )
202 output = new_state[step_id]
203 return ToolResult(
204 ok=True,
205 content=_tool_content(TICKET_CREATE_TOOL_NAME, ok=True, output=output),
206 state_patch={step_id: output},
207 )
210TOOL = _TicketCreateTool()