Coverage for core/langgraph/tools/resolver.py: 100%
60 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-25 05:49 +0000
1"""Resolution-composition layer across the three tool sources.
3An agent declares `tools: [<name>]`; this module turns that declaration into
4ordered `Tool` instances by consulting the single Core-owned `ToolRegistry`
5(file-access tools ∪ platform builtins) first and the StepFunction register
6second (ADR-0039 D2). A name present in both sources is a hard collision, never
7a silent precedence choice (ADR-0041 D1).
8"""
9from __future__ import annotations
11import json
12from collections.abc import Sequence
13from typing import Any
15from .contract import Tool, ToolContext, ToolResult
16from .registry import ToolRegistry, ToolRegistryError
18_TOOL_MODULES: dict[str, str] = {
19 "read_file": "langgraph.tools.read_file",
20 "list_dir": "langgraph.tools.list_dir",
21 "list-dir-entries": "langgraph.tools.list_dir_entries",
22 "aggregate": "langgraph.tools.aggregate",
23 "ticket-publisher": "langgraph.tools.ticket_publisher",
24 "run-finalizer": "langgraph.tools.run_finalizer",
25 "git-checkout": "langgraph.tools.git_checkout",
26 "files-write": "langgraph.tools.files_write",
27 "files-delete": "langgraph.tools.files_delete",
28 "question-block-form": "langgraph.tools.question_block_form",
29 "github-pr-comment": "langgraph.tools.github_pr_comment",
30 "ticket-publish": "langgraph.tools.ticket_publish",
31 "ticket-context": "langgraph.tools.ticket_context",
32 "ticket-lane": "langgraph.tools.ticket_lane",
33 "ticket-dor": "langgraph.tools.ticket_dor",
34 "ticket-review": "langgraph.tools.ticket_review",
35 "ticket-search": "langgraph.tools.ticket_search",
36 "ticket-create": "langgraph.tools.ticket_create",
37 "ticket-approval": "langgraph.tools.ticket_approval",
38 "submit_decision": "langgraph.tools.submit_decision",
39}
42class ToolCollisionError(ToolRegistryError):
43 """A declared name is registered in both platform and StepFunction sources."""
45 code = "BUILTIN_NAME_COLLISION"
47 def __init__(self, name: str) -> None:
48 self.name = name
49 super().__init__(
50 f"tool {name!r} exists in both the platform registry and the "
51 f"StepFunction register"
52 )
55def build_platform_registry() -> ToolRegistry:
56 """A fresh registry holding every platform tool as a lazy locator."""
57 registry = ToolRegistry()
58 for name, module_path in _TOOL_MODULES.items():
59 registry.register_lazy(name, module_path, attr="TOOL")
60 return registry
63class ToolResolver:
64 """Composes the registry and the StepFunction register into one surface."""
66 def __init__(
67 self, registry: ToolRegistry, step_functions: object | None = None
68 ) -> None:
69 self._registry = registry
70 self._step_functions = step_functions
72 def resolve(self, names: Sequence[str]) -> list[Tool]:
73 """Resolve *names* in declaration order; duplicates are preserved."""
74 return [self._resolve_one(name) for name in names]
76 def _resolve_one(self, name: str) -> Tool:
77 if self._is_collision(name):
78 raise ToolCollisionError(name)
79 if self._registry.has(name):
80 return self._registry.resolve([name])[0]
81 if self._has_step_function(name):
82 return _StepFunctionTool(name, self._step_functions)
83 raise ToolRegistryError(f"tool {name!r} is not registered")
85 def _is_collision(self, name: str) -> bool:
86 return self._registry.has(name) and self._has_step_function(name)
88 def _has_step_function(self, name: str) -> bool:
89 return self._step_functions is not None and self._step_functions.has(name)
92def _hard_failure(code: str, message: str, details: dict | None = None) -> Any:
93 """Build the engine's RunHardFailure without a module-level circular import."""
94 try:
95 from runner import RunHardFailure
96 except ImportError: # pragma: no cover — main.py qualified imports
97 from langgraph.runner import RunHardFailure
98 return RunHardFailure(code, message, details)
101class _StepFunctionTool:
102 """Adapts an injected StepFunction executor to the `Tool` protocol."""
104 def __init__(self, name: str, step_functions: object) -> None:
105 self.name = name
106 self._step_functions = step_functions
108 def instruction(self) -> str | None:
109 return self._step_functions.instruction_for(self.name)
111 def params(self) -> list[dict[str, Any]] | None:
112 return self._step_functions.params_for(self.name)
114 def definition(self) -> dict[str, Any]:
115 return {
116 "type": "function",
117 "function": {
118 "name": self.name,
119 "description": self.instruction() or "",
120 "parameters": _params_schema(self.params()),
121 },
122 }
124 def execute(self, context: ToolContext) -> ToolResult:
125 step_id = (context.step or {}).get("id", self.name)
126 step = {"id": step_id, "params": dict(context.args or {})}
127 state = self._step_functions.execute(self.name, step, context.state or {})
128 fragment = state.get(step_id)
129 if fragment is None:
130 raise _hard_failure(
131 "UNRESOLVED_OUTPUT",
132 f"StepFunction '{self.name}': executor returned no fragment for "
133 f"step '{step_id}'",
134 {"step_function": self.name, "step_id": step_id},
135 )
136 return ToolResult(ok=True, content=json.dumps(fragment), state_patch=None)
139def _params_schema(params: list[dict[str, Any]] | None) -> dict[str, Any]:
140 """Emit an untyped JSON schema for a StepFunction's declared params.
142 Every property is emitted as `{}` because the manifest descriptor
143 (`StepFunctionInterfaceSpec.params` in `manifest.schema.json`) carries only
144 `name`, `required`, and `default` — no type descriptor (its items are
145 closed by `additionalProperties: false`). The absence of types, and the
146 non-propagation of `default` into the schema, are a deliberate response to
147 that manifest limitation, not an oversight: inferring a type from `default`
148 would be a silent heuristic, forbidden by ADR-0012.
149 """
150 declared = params or []
151 return {
152 "type": "object",
153 "properties": {param["name"]: {} for param in declared},
154 "required": [param["name"] for param in declared if param.get("required")],
155 "additionalProperties": False,
156 }