Coverage for core/langgraph/_instructions.py: 97%
66 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-20 03:43 +0000
1"""
2_instructions.py — Core-owned prompt-text and tool registry.
4The calling agent declares `tools`; Core resolves them through this registry
5into native function-tool entries. This module owns the per-step how-to text
6for Core `builtin` steps; consumer `StepFunction` steps own their text via
7`spec.agent_instructions`. Each resolved tool carries its own how-to text as
8the native `function.description` — there is no generic common instruction and
9no per-tool decoration.
11The registry is the single source for builtin instruction texts — runtime
12injection and the build-time `AGENT_INSTRUCTION_MISSING` validator both resolve
13through `builtin_instruction`.
14"""
15from __future__ import annotations
17from typing import Any
19# Reserved resolved-step key carrying a step's resolved tool entries
20# (one per entry in the calling agent's `tools`, in declaration order).
21# Distinct from the COMMUNICATION context key in _data_structure.py.
22USAGE_INSTRUCTIONS_KEY = "_usage_instructions"
24_BUILTIN_INSTRUCTIONS: dict[str, str] = {
25 "list-dir-entries": (
26 "List one level of a directory with `list-dir-entries` instead of "
27 "inspecting files one by one. Pass `params.path` (a path; defaults to "
28 "the project root) and read the returned `entries` and `stats` to "
29 "decide what to inspect next."
30 ),
31 "aggregate": (
32 "Fold the collected per-iteration results deterministically with "
33 "`aggregate` instead of recomputing them yourself. Provide "
34 "`input.results` (or `input.sources`), a `fields` list of "
35 "`{target, source, op, dedupe_key/dedupe_keys, on_error}` entries, and "
36 "optionally the `op`/`on_error` defaults; each `target` is written to "
37 "the step result."
38 ),
39 "ticket-publisher": (
40 "Publish an approved ticket by passing its markdown as the step "
41 "`input` to `ticket-publisher`. Set `params.status_from`/`status_to` to "
42 "flip the status and `params.publish_dir` (default `docs/tickets`) as "
43 "the destination; the result reports the published `path` and `slug`."
44 ),
45 "run-finalizer": (
46 "Finalize the run with `run-finalizer` after all work steps. It "
47 "composes the run status and stats from state and appends a runlog "
48 "line; pass `params.pipeline_id` (and optionally `params.runlog`) so "
49 "the record is attributed correctly."
50 ),
51 "git-checkout": (
52 "Prepare the working tree with `git-checkout` before any step that "
53 "reads or writes repository files. It clones the target repository for "
54 "the active run profile and exposes the checkout location and branch "
55 "facts in state; do not clone the repository yourself."
56 ),
57 "files-write": (
58 "Write files with `files-write` instead of ad-hoc shell writes. Supply "
59 "`params.entries` as one `{path, content}` object or a list of them; "
60 "paths are workspace-relative and the result reports the written paths."
61 ),
62 "files-delete": (
63 "Delete workspace files with `files-delete` instead of a shell `rm`. "
64 "Supply `params.paths` as one workspace-relative path or a list of "
65 "them; every path is validated up front and the result reports the "
66 "`deleted` paths."
67 ),
68 "question-block-form": (
69 "Form the human-interface question block with `question-block-form` "
70 "when the agentic effort is exhausted. Pass the exhaustion sources as "
71 "the step `input` (`coverage`, `decompose`, `research`, `refine`, "
72 "`close_readers`, `close_synthesis` — all optional); the result "
73 "reports `status` and a `question_block` carrying the problems and "
74 "open questions for the human."
75 ),
76 "github-pr-comment": (
77 "Post the pull-request review with `github-pr-comment` instead of "
78 "calling the forge API yourself. Set `params.findings_key` (default "
79 "`review_output`) to the state dot-path holding the findings and "
80 "`params.fail_on_severity` (default `Critical`) to the severity that "
81 "forces `REQUEST_CHANGES`; the result reports `findings_count` and the "
82 "WIP title sync status."
83 ),
84 "ticket-publish": (
85 "Publish the TICKET run's output with `ticket-publish`: it commits the "
86 "workspace, pushes the feature branch, and ensures the pull request. "
87 "It reads `project_dir`, `ticket_branch` and `ticket_base_branch` "
88 "(set by `git-checkout`); the result reports `branch`, `base`, "
89 "`pushed`, `pr_action` and, once a PR is ensured, `pr_url`."
90 ),
91 "ticket-context": (
92 "Load the ticket context with `ticket-context`: it reads the "
93 "`TICKET_CONTEXT` environment JSON (description plus author-tagged "
94 "comments) and stores it under the step id and the top-level "
95 "`ticket_context` key for later steps to read."
96 ),
97 "ticket-lane": (
98 "Convert the agent's lane decision with `ticket-lane` by passing the "
99 "lane string (or an object `{lane, reason, questions}`) as the step "
100 "`input`; it continues on `pr_open` and otherwise stops the run with "
101 "the matching outcome code."
102 ),
103 "ticket-dor": (
104 "Gate the ticket with `ticket-dor` before implementation: it reads "
105 "`TICKET_CONTEXT` and asks the model whether the ticket inputs are "
106 "sufficient, stopping the run with outcome 10 and the open questions "
107 "when they are not."
108 ),
109 "ticket-review": (
110 "Wait for and classify the pull-request review with `ticket-review` "
111 "after `ticket-publish`: it polls until the PR head commit carries a "
112 "review and maps it to an outcome (approved continues; fixable / "
113 "too_bad / cannot_fix stop the run). Set `params.poll_interval_seconds` "
114 "(default 15) and `params.timeout_seconds` (default 900)."
115 ),
116 "ticket-search": (
117 "List the open work items with `ticket-search` instead of querying the "
118 "project-management system yourself; the result reports `tickets` "
119 "(`id`, `title`, `done`, `description`) resolved through the run's PM "
120 "configuration."
121 ),
122 "ticket-create": (
123 "Create a single work item with `ticket-create` (idempotent): set "
124 "`params.title` (required), optional `params.description` and optional "
125 "`params.parent_task_id`; a matching open title is skipped and the "
126 "result reports `created` and `skipped`."
127 ),
128 "ticket-approval": (
129 "Detect the human approval marker with `ticket-approval`: pass the "
130 "`ticket_context` object as the step `input`; a human comment whose "
131 "entire content is `APPROVED` (case-insensitive) passes, and the "
132 "result reports `passed` and the matching `evidence`."
133 ),
134}
137def builtin_instruction(name: str) -> str | None:
138 """The registry text for a supported builtin, or None.
140 Accepts the hyphenated canonical name and its underscore-normalized alias,
141 mirroring `BuiltinExecutor`'s hyphen→underscore name normalization.
142 """
143 return _BUILTIN_INSTRUCTIONS.get(name) or _BUILTIN_INSTRUCTIONS.get(
144 name.replace("_", "-")
145 )
148def instruction_text(name: str, builtins: Any, step_functions: Any) -> str | None:
149 """Resolve one whitelist entry's usage instruction, builtin-first.
151 Mirrors the runtime dispatch order (ADR-0037 D2): a name in the builtin
152 register resolves to the Core-owned registry text, otherwise to the
153 StepFunction's ``agent_instructions``.
154 """
155 if builtins.has(name):
156 return builtin_instruction(name)
157 if step_functions is not None:
158 return step_functions.instruction_for(name)
159 return None
162def resolve_usage_instructions(
163 allow: list[str] | None, builtins: Any, step_functions: Any
164) -> list[dict]:
165 """Resolve a step's `allow` whitelist into ordered instruction entries.
167 Retained legacy/test-only helper: it is not the source of runtime usage
168 instructions, which now derive from the calling agent's resolved `tools`
169 via `resolve_tool_entries`. Each resolvable entry carries the whitelisted
170 name, its per-step text and its declared params (StepFunctions only).
171 Declaration order is preserved. An empty or absent whitelist yields an
172 empty list.
173 """
174 entries: list[dict] = []
175 for name in allow or []:
176 text = instruction_text(name, builtins, step_functions)
177 if not text:
178 continue
179 entries.append(
180 {
181 "name": name,
182 "text": text,
183 "params": _declared_params(name, builtins, step_functions),
184 }
185 )
186 return entries
189def _declared_params(
190 name: str, builtins: Any, step_functions: Any
191) -> list[dict] | None:
192 """A resolved StepFunction tool's declared `interface.params`, else None.
194 Builtins carry no formally declared param interface — their registry text
195 conveys the accepted params.
196 """
197 if builtins.has(name) or step_functions is None:
198 return None
199 return step_functions.params_for(name)
202# --- Single Core-owned tool registry -------------------------------------
203# One registry = file-access tools ∪ deterministic step tools (platform
204# `builtin` ∪ consumer `StepFunction`). An agent declares `tools: [<name>]`;
205# `resolve_tool_entries` turns that list into ordered, native-function-calling
206# ready entries. The two file-access definitions live here so the registry owns
207# every definition.
209AGENT_INSTRUCTION_MISSING = "AGENT_INSTRUCTION_MISSING"
211READ_FILE_TOOL_NAME = "read_file"
212LIST_DIR_TOOL_NAME = "list_dir"
214READ_FILE_TOOL: dict[str, Any] = {
215 "type": "function",
216 "function": {
217 "name": READ_FILE_TOOL_NAME,
218 "description": (
219 "Read a text file from the project root. The path is resolved "
220 "against the project root; any read that resolves outside the "
221 "project root is rejected. Content is truncated and flagged when "
222 "it exceeds the caps, and can be windowed by offset/limit (lines)."
223 ),
224 "parameters": {
225 "type": "object",
226 "properties": {
227 "path": {
228 "type": "string",
229 "description": "Path to the file, relative to the project root.",
230 },
231 "offset": {
232 "type": "integer",
233 "description": "Number of lines to skip (0-based). Optional.",
234 },
235 "limit": {
236 "type": "integer",
237 "description": "Max number of lines to return. Optional.",
238 },
239 },
240 "required": ["path"],
241 "additionalProperties": False,
242 },
243 },
244}
246LIST_DIR_TOOL: dict[str, Any] = {
247 "type": "function",
248 "function": {
249 "name": LIST_DIR_TOOL_NAME,
250 "description": (
251 "List ONE level of a directory inside the project root. The path is "
252 "resolved against the project root; any listing that resolves "
253 "outside the project root is rejected. Non-recursive: nested "
254 "subdirectories are not expanded. Only structural skip dirs are "
255 "omitted. The listing is capped and flagged when it exceeds the cap."
256 ),
257 "parameters": {
258 "type": "object",
259 "properties": {
260 "path": {
261 "type": "string",
262 "description": "Directory path, relative to the project root.",
263 },
264 },
265 "required": ["path"],
266 "additionalProperties": False,
267 },
268 },
269}
271_FILE_ACCESS_TOOLS: dict[str, dict[str, Any]] = {
272 READ_FILE_TOOL_NAME: READ_FILE_TOOL,
273 LIST_DIR_TOOL_NAME: LIST_DIR_TOOL,
274}
277class ToolRegistryError(ValueError):
278 """A declared agent tool cannot be resolved to a registry definition."""
280 def __init__(self, code: str, name: str) -> None:
281 super().__init__(f"{code}: tool '{name}' has no registry definition")
282 self.code = code
283 self.name = name
286def resolve_tool_entries(
287 tools: list[str] | None, builtins: Any, step_functions: Any
288) -> list[dict]:
289 """Resolve an agent's declared `tools` list into ordered tool entries.
291 Each entry carries the tool's `name`, its OpenAI `definition`, its
292 instruction `text` (None for file-access tools) and its declared `params`
293 (StepFunctions only). Declaration order is preserved; an empty or absent
294 list yields no entries. An unresolvable name, or a deterministic step
295 without instruction text, is a hard `ToolRegistryError`.
296 """
297 entries: list[dict] = []
298 for name in tools or []:
299 entries.append(_resolve_tool_entry(name, builtins, step_functions))
300 return entries
303def _resolve_tool_entry(name: str, builtins: Any, step_functions: Any) -> dict:
304 if name in _FILE_ACCESS_TOOLS:
305 return {
306 "name": name,
307 "definition": _FILE_ACCESS_TOOLS[name],
308 "text": None,
309 "params": None,
310 }
311 if builtins is not None and builtins.has(name):
312 text = _required_instruction(builtin_instruction(name), name)
313 return {
314 "name": name,
315 "definition": _generic_function_tool(name, text),
316 "text": text,
317 "params": None,
318 }
319 if step_functions is not None and step_functions.has(name):
320 text = _required_instruction(step_functions.instruction_for(name), name)
321 params = step_functions.params_for(name)
322 return {
323 "name": name,
324 "definition": _step_function_tool(name, text, params),
325 "text": text,
326 "params": params,
327 }
328 raise ToolRegistryError(AGENT_INSTRUCTION_MISSING, name)
331def _required_instruction(text: str | None, name: str) -> str:
332 if not text:
333 raise ToolRegistryError(AGENT_INSTRUCTION_MISSING, name)
334 return text
337def _generic_function_tool(name: str, description: str) -> dict:
338 return {
339 "type": "function",
340 "function": {
341 "name": name,
342 "description": description,
343 "parameters": {
344 "type": "object",
345 "properties": {},
346 "additionalProperties": True,
347 },
348 },
349 }
352def _step_function_tool(
353 name: str, description: str, params: list[dict] | None
354) -> dict:
355 return {
356 "type": "function",
357 "function": {
358 "name": name,
359 "description": description,
360 "parameters": _tool_parameters(params),
361 },
362 }
365def _tool_parameters(params: list[dict] | None) -> dict:
366 declared = params or []
367 schema: dict[str, Any] = {
368 "type": "object",
369 "properties": {param["name"]: {} for param in declared},
370 "additionalProperties": False,
371 }
372 required = [param["name"] for param in declared if param.get("required")]
373 if required:
374 schema["required"] = required
375 return schema