Coverage for core/langgraph/_instructions.py: 100%
50 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-20 02:14 +0000
1"""
2_instructions.py — Core-owned prompt-text registry for whitelisted steps.
4The deterministic-step whitelist (`allow`) is implicit: Core injects a usage
5instruction per whitelisted step instead of the agent declaring the whitelist
6itself (core/docs/agent-io-envelope.md § Usage instructions). This module owns
7the per-step text for Core `builtin` steps; consumer `StepFunction` steps own
8their text via `spec.agent_instructions`.
10`CORE_COMMON_INSTRUCTION` is prepended to the per-step texts and deliberately
11does not enumerate step names. Each registry text is written in the imperative,
12agent-facing voice and describes what the builtin does and which params it
13accepts, so the decorated instruction is self-contained.
15The registry is the single source for builtin instruction texts — runtime
16injection (a later phase) and the build-time `AGENT_INSTRUCTION_MISSING`
17validator both resolve through `builtin_instruction`.
18"""
19from __future__ import annotations
21import json
22from typing import Any
24# Reserved resolved-step key carrying a step's resolved usage-instruction
25# entries (one per `allow` entry, in declaration order). Distinct from the
26# COMMUNICATION context key in _data_structure.py.
27USAGE_INSTRUCTIONS_KEY = "_usage_instructions"
29CORE_COMMON_INSTRUCTION = (
30 "Use the provided deterministic tools to carry out this step instead of "
31 "guessing their results or reproducing their effect yourself. Invoke only "
32 "the tools listed below, pass exactly the parameters they accept, and "
33 "continue from their actual output."
34)
36_BUILTIN_INSTRUCTIONS: dict[str, str] = {
37 "list-dir-entries": (
38 "List one level of a directory with `list-dir-entries` instead of "
39 "inspecting files one by one. Pass `params.path` (a path; defaults to "
40 "the project root) and read the returned `entries` and `stats` to "
41 "decide what to inspect next."
42 ),
43 "aggregate": (
44 "Fold the collected per-iteration results deterministically with "
45 "`aggregate` instead of recomputing them yourself. Provide "
46 "`input.results` (or `input.sources`), a `fields` list of "
47 "`{target, source, op, dedupe_key/dedupe_keys, on_error}` entries, and "
48 "optionally the `op`/`on_error` defaults; each `target` is written to "
49 "the step result."
50 ),
51 "ticket-publisher": (
52 "Publish an approved ticket by passing its markdown as the step "
53 "`input` to `ticket-publisher`. Set `params.status_from`/`status_to` to "
54 "flip the status and `params.publish_dir` (default `docs/tickets`) as "
55 "the destination; the result reports the published `path` and `slug`."
56 ),
57 "run-finalizer": (
58 "Finalize the run with `run-finalizer` after all work steps. It "
59 "composes the run status and stats from state and appends a runlog "
60 "line; pass `params.pipeline_id` (and optionally `params.runlog`) so "
61 "the record is attributed correctly."
62 ),
63 "git-checkout": (
64 "Prepare the working tree with `git-checkout` before any step that "
65 "reads or writes repository files. It clones the target repository for "
66 "the active run profile and exposes the checkout location and branch "
67 "facts in state; do not clone the repository yourself."
68 ),
69 "files-write": (
70 "Write files with `files-write` instead of ad-hoc shell writes. Supply "
71 "`params.entries` as one `{path, content}` object or a list of them; "
72 "paths are workspace-relative and the result reports the written paths."
73 ),
74 "files-delete": (
75 "Delete workspace files with `files-delete` instead of a shell `rm`. "
76 "Supply `params.paths` as one workspace-relative path or a list of "
77 "them; every path is validated up front and the result reports the "
78 "`deleted` paths."
79 ),
80 "question-block-form": (
81 "Form the human-interface question block with `question-block-form` "
82 "when the agentic effort is exhausted. Pass the exhaustion sources as "
83 "the step `input` (`coverage`, `decompose`, `research`, `refine`, "
84 "`close_readers`, `close_synthesis` — all optional); the result "
85 "reports `status` and a `question_block` carrying the problems and "
86 "open questions for the human."
87 ),
88 "github-pr-comment": (
89 "Post the pull-request review with `github-pr-comment` instead of "
90 "calling the forge API yourself. Set `params.findings_key` (default "
91 "`review_output`) to the state dot-path holding the findings and "
92 "`params.fail_on_severity` (default `Critical`) to the severity that "
93 "forces `REQUEST_CHANGES`; the result reports `findings_count` and the "
94 "WIP title sync status."
95 ),
96 "ticket-publish": (
97 "Publish the TICKET run's output with `ticket-publish`: it commits the "
98 "workspace, pushes the feature branch, and ensures the pull request. "
99 "It reads `project_dir`, `ticket_branch` and `ticket_base_branch` "
100 "(set by `git-checkout`); the result reports `branch`, `base`, "
101 "`pushed`, `pr_action` and, once a PR is ensured, `pr_url`."
102 ),
103 "ticket-context": (
104 "Load the ticket context with `ticket-context`: it reads the "
105 "`TICKET_CONTEXT` environment JSON (description plus author-tagged "
106 "comments) and stores it under the step id and the top-level "
107 "`ticket_context` key for later steps to read."
108 ),
109 "ticket-lane": (
110 "Convert the agent's lane decision with `ticket-lane` by passing the "
111 "lane string (or an object `{lane, reason, questions}`) as the step "
112 "`input`; it continues on `pr_open` and otherwise stops the run with "
113 "the matching outcome code."
114 ),
115 "ticket-dor": (
116 "Gate the ticket with `ticket-dor` before implementation: it reads "
117 "`TICKET_CONTEXT` and asks the model whether the ticket inputs are "
118 "sufficient, stopping the run with outcome 10 and the open questions "
119 "when they are not."
120 ),
121 "ticket-review": (
122 "Wait for and classify the pull-request review with `ticket-review` "
123 "after `ticket-publish`: it polls until the PR head commit carries a "
124 "review and maps it to an outcome (approved continues; fixable / "
125 "too_bad / cannot_fix stop the run). Set `params.poll_interval_seconds` "
126 "(default 15) and `params.timeout_seconds` (default 900)."
127 ),
128 "ticket-search": (
129 "List the open work items with `ticket-search` instead of querying the "
130 "project-management system yourself; the result reports `tickets` "
131 "(`id`, `title`, `done`, `description`) resolved through the run's PM "
132 "configuration."
133 ),
134 "ticket-create": (
135 "Create a single work item with `ticket-create` (idempotent): set "
136 "`params.title` (required), optional `params.description` and optional "
137 "`params.parent_task_id`; a matching open title is skipped and the "
138 "result reports `created` and `skipped`."
139 ),
140 "ticket-approval": (
141 "Detect the human approval marker with `ticket-approval`: pass the "
142 "`ticket_context` object as the step `input`; a human comment whose "
143 "entire content is `APPROVED` (case-insensitive) passes, and the "
144 "result reports `passed` and the matching `evidence`."
145 ),
146}
149def builtin_instruction(name: str) -> str | None:
150 """The registry text for a whitelistable builtin, or None.
152 Accepts the hyphenated canonical name and its underscore-normalized alias,
153 mirroring `BuiltinExecutor`'s hyphen→underscore name normalization.
154 """
155 return _BUILTIN_INSTRUCTIONS.get(name) or _BUILTIN_INSTRUCTIONS.get(
156 name.replace("_", "-")
157 )
160def instruction_text(name: str, builtins: Any, step_functions: Any) -> str | None:
161 """Resolve one whitelist entry's usage instruction, builtin-first.
163 Mirrors the runtime dispatch order (ADR-0037 D2): a name in the builtin
164 register resolves to the Core-owned registry text, otherwise to the
165 StepFunction's ``agent_instructions``.
166 """
167 if builtins.has(name):
168 return builtin_instruction(name)
169 if step_functions is not None:
170 return step_functions.instruction_for(name)
171 return None
174def resolve_usage_instructions(
175 allow: list[str] | None, builtins: Any, step_functions: Any
176) -> list[dict]:
177 """Resolve a step's `allow` whitelist into ordered instruction entries.
179 Each resolvable entry carries the whitelisted name, its per-step text and
180 its declared params (StepFunctions only). Declaration order is preserved.
181 An empty or absent whitelist yields an empty list.
182 """
183 entries: list[dict] = []
184 for name in allow or []:
185 text = instruction_text(name, builtins, step_functions)
186 if not text:
187 continue
188 entries.append(
189 {
190 "name": name,
191 "text": text,
192 "params": _declared_params(name, builtins, step_functions),
193 }
194 )
195 return entries
198def render_usage_instructions_block(entries: list[dict] | None) -> str:
199 """Render the usage-instruction block, or "" when nothing is whitelisted.
201 The Core common instruction is prepended once; each whitelisted step
202 follows as a sub-heading with its params and per-step text, in the order
203 the entries were resolved.
204 """
205 if not entries:
206 return ""
207 blocks = ["## Usage instructions", CORE_COMMON_INSTRUCTION]
208 blocks.extend(_render_entry(entry) for entry in entries)
209 return "\n\n".join(blocks)
212def _declared_params(
213 name: str, builtins: Any, step_functions: Any
214) -> list[dict] | None:
215 """A whitelisted StepFunction's declared `interface.params`, else None.
217 Builtins carry no formally declared param interface — their registry text
218 conveys the accepted params.
219 """
220 if builtins.has(name) or step_functions is None:
221 return None
222 return step_functions.params_for(name)
225def _render_entry(entry: dict) -> str:
226 sections = [f"### {entry['name']}"]
227 params_text = _render_params(entry.get("params"))
228 if params_text:
229 sections.append(params_text)
230 if entry.get("text"):
231 sections.append(entry["text"])
232 return "\n\n".join(sections)
235def _render_params(params: list[dict] | None) -> str:
236 if not params:
237 return ""
238 return "\n".join(["Params:"] + [_render_param_line(param) for param in params])
241def _render_param_line(param: dict) -> str:
242 label = "required" if param.get("required") else "optional"
243 if "default" in param:
244 default = json.dumps(param["default"], ensure_ascii=False)
245 return f"- `{param['name']}` ({label}, default: {default})"
246 return f"- `{param['name']}` ({label})"