Coverage for core/langgraph/tools/ticket_lane.py: 97%
67 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-25 03:42 +0000
1"""Per-tool module for the `ticket-lane` deterministic builtin.
3Owns the tool's model-facing definition, its instruction surface (none) and the
4agent-decision to lane-outcome mapping itself. The definition and the execution
5here are the single source; the legacy builtin site delegates to the pure
6`ticket_lane` handler.
7"""
8from __future__ import annotations
10import os
11from typing import Any
13from .contract import ToolContext, ToolResult
15try:
16 from tools._ticket_common import (
17 _as_questions, _hard_failure, _tool_content, TicketOutcome,
18 )
19except ImportError: # pragma: no cover - main.py run (qualified imports)
20 from langgraph.tools._ticket_common import (
21 _as_questions, _hard_failure, _tool_content, TicketOutcome,
22 )
24TICKET_LANE_TOOL_NAME = "ticket-lane"
25DEFAULT_STEP_ID = "ticket_lane"
27TICKET_LANE_DEFINITION: dict[str, Any] = {
28 "type": "function",
29 "function": {
30 "name": TICKET_LANE_TOOL_NAME,
31 "description": (
32 "Convert the agent's lane decision with `ticket-lane` via `lane` "
33 "(with optional `reason` and `questions`); it continues on "
34 "`pr_open` and otherwise stops the run with the matching outcome code."
35 ),
36 "parameters": {
37 "type": "object",
38 "properties": {
39 "lane": {
40 "type": "string",
41 "description": (
42 "The agent's lane decision: `pr_open`, `open_questions`, "
43 "`blocked`, `needs_rework` or `pr_rejected`."
44 ),
45 },
46 "reason": {
47 "type": "string",
48 "description": (
49 "Required only for the `blocked` lane — the reason is "
50 "carried verbatim as the outcome message."
51 ),
52 },
53 "questions": {
54 "type": "array",
55 "items": {"type": "string"},
56 "description": (
57 "Open questions for the human; used only for the "
58 "`open_questions` lane."
59 ),
60 },
61 },
62 "required": ["lane"],
63 "additionalProperties": False,
64 },
65 },
66}
69def _hard_failure_type() -> type[Exception]:
70 """The runner's hard-failure type, resolved without an import cycle."""
71 try:
72 from runner import RunHardFailure
73 except ImportError: # pragma: no cover - main.py run (qualified imports)
74 from langgraph.runner import RunHardFailure
75 return RunHardFailure
78def ticket_lane(step: dict, state: dict) -> dict:
79 """Map the agent's decision (step.input) to a lane-outcome code.
81 Input contract (locked):
82 - String input allowed: pr_open / open_questions / needs_rework /
83 pr_rejected (reason empty, questions empty).
84 - Object input {"lane": str, "reason": str, "questions": [str, ...]?}
85 allowed for all; questions is an optional list of strings and is passed
86 through only to the open_questions outcome.
87 - blocked REQUIRES the reason from an object input - a string input, or an
88 object without a reason field (or a non-string reason) -> hard failure
89 (fail-fast, ADR-0012): a blocked comment needs the reason. The runner
90 resolves step.input before the builtin call.
92 Lane -> outcome:
93 pr_open -> returns state (run continues -> exit 0 -> pr_open lane)
94 open_questions-> TicketOutcome(22) + questions (22, NOT 10 - 10 is
95 DoR-specific; the D3 fail-fast at the entrypoint requires
96 a non-empty questions)
97 blocked -> TicketOutcome(11), message = the bare reason -
98 the "Blocked: <reason>" comment format is owned by the API
99 needs_rework -> TicketOutcome(20)
100 pr_rejected -> TicketOutcome(21)
101 done / in_progress / failed / unknown -> hard failure (RunHardFailure;
102 done only via merge-webhook, in_progress a mid-run state, failed
103 reserved for technical errors - the contract).
105 Lanes other than open_questions: questions are ignored (not set).
106 Wrong RUN_PROFILE -> hard failure (ADR-0012).
107 """
108 run_profile = os.environ.get("RUN_PROFILE")
109 if run_profile != "TICKET":
110 raise _hard_failure(
111 "VALIDATION_ERROR",
112 f"ticket-lane requires RUN_PROFILE=TICKET (got {run_profile!r})",
113 )
115 decision = step.get("input")
116 if isinstance(decision, str):
117 lane = decision
118 reason = ""
119 questions = None
120 elif isinstance(decision, dict):
121 lane = decision.get("lane")
122 if not isinstance(lane, str):
123 raise _hard_failure(
124 "AGENT_ERROR",
125 "ticket-lane: object input must have a string 'lane' "
126 f"field, got {decision!r}",
127 )
128 reason = decision.get("reason", "")
129 if not isinstance(reason, str):
130 raise _hard_failure(
131 "AGENT_ERROR",
132 f"ticket-lane: 'reason' must be a string, got {reason!r}",
133 )
134 questions_raw = decision.get("questions")
135 questions = (
136 None
137 if questions_raw is None
138 else _as_questions(questions_raw, "ticket-lane")
139 )
140 else:
141 raise _hard_failure(
142 "AGENT_ERROR",
143 "ticket-lane: agent decision must be a lane string or "
144 f"an object, got {type(decision).__name__}",
145 )
147 if lane == "pr_open":
148 step_id = step.get("id", "ticket_lane")
149 return {**state, step_id: {"lane": "pr_open"}}
151 if lane == "blocked":
152 if not reason:
153 raise _hard_failure(
154 "AGENT_ERROR",
155 "ticket-lane: lane 'blocked' requires an object input with "
156 "a non-empty 'reason' — the blocked comment needs the reason",
157 )
158 raise TicketOutcome(code=11, message=reason)
160 if lane == "open_questions":
161 raise TicketOutcome(
162 code=22,
163 message=(
164 "ticket-lane: agent decided open_questions — run stopped, "
165 "human input needed (outcome 22)"
166 ),
167 questions=questions,
168 )
169 if lane == "needs_rework":
170 raise TicketOutcome(
171 code=20,
172 message=(
173 "ticket-lane: agent decided needs_rework — rework round "
174 "required (outcome 20)"
175 ),
176 )
177 if lane == "pr_rejected":
178 raise TicketOutcome(
179 code=21,
180 message="ticket-lane: agent decided pr_rejected (outcome 21)",
181 )
182 if lane in ("done", "in_progress", "failed"):
183 raise _hard_failure(
184 "AGENT_ERROR",
185 f"ticket-lane: lane {lane!r} is not a valid agent decision — "
186 "done (merge-webhook), in_progress (mid-run state) and failed "
187 "(technical errors) are API-owned",
188 )
189 raise _hard_failure("AGENT_ERROR", f"ticket-lane: unknown lane {lane!r}")
192class _TicketLaneTool:
193 name = TICKET_LANE_TOOL_NAME
195 def definition(self) -> dict[str, Any]:
196 return TICKET_LANE_DEFINITION
198 def instruction(self) -> str | None:
199 return None
201 def params(self) -> list[dict[str, Any]] | None:
202 return None
204 def execute(self, context: ToolContext) -> ToolResult:
205 step_id = (context.step or {}).get("id") or DEFAULT_STEP_ID
206 args = context.args or {}
207 synthetic_step = {
208 "id": step_id,
209 "input": {
210 "lane": args.get("lane"),
211 "reason": args.get("reason", ""),
212 "questions": args.get("questions"),
213 },
214 }
215 try:
216 new_state = ticket_lane(synthetic_step, context.state or {})
217 except _hard_failure_type() as exc:
218 return ToolResult(
219 ok=False,
220 content=_tool_content(TICKET_LANE_TOOL_NAME, ok=False, error=exc.message),
221 error=exc.message,
222 )
223 output = new_state[step_id]
224 return ToolResult(
225 ok=True,
226 content=_tool_content(TICKET_LANE_TOOL_NAME, ok=True, output=output),
227 state_patch={step_id: output},
228 )
231TOOL = _TicketLaneTool()