freemo/cleveragents-core
0
0
Fork 0
forked from cleveragents/cleveragents-core
Code Pull Requests Activity

feat(tool): add resource binding resolution

Browse Source
This commit is contained in:
Jeffrey Phillips Freeman
2026-02-17 11:08:14 +00:00
parent efc48864e1
commit 07ec9f04d8
9 changed files with 1536 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files
benchmarks/binding_resolution_bench.py
+225
View File
@@ -0,0 +1,225 @@
"""ASV benchmarks for BindingResolutionService.
Measures resolve latency for contextual, static, and parameter
bindings across varying numbers of linked resources.
"""
from __future__ import annotations
from unittest.mock import MagicMock
from cleveragents.application.services.binding_resolution_service import (
BindingResolutionService,
)
from cleveragents.core.exceptions import NotFoundError
from cleveragents.domain.models.core.project import (
LinkedResource,
NamespacedProject,
)
from cleveragents.domain.models.core.resource import PhysVirt, Resource
from cleveragents.domain.models.core.resource_type import (
ResourceKind,
ResourceTypeSpec,
SandboxStrategy,
)
from cleveragents.domain.models.core.tool import (
BindingMode,
ResourceAccessMode,
ResourceSlot,
Tool,
ToolSource,
ToolType,
)
_CB32 = "0123456789ABCDEFGHJKMNPQRSTVWXYZ"
_bench_counter = 0
def _bench_ulid() -> str:
global _bench_counter
_bench_counter += 1
n = _bench_counter
suffix = ""
for _ in range(8):
suffix = _CB32[n % 32] + suffix
n //= 32
return f"01HGZ6FE0AQDYTR4BX{suffix}"
def _make_registry(
resources: dict[str, Resource],
types: dict[str, ResourceTypeSpec],
) -> MagicMock:
registry = MagicMock()
def show_resource(name_or_id: str) -> Resource:
res = resources.get(name_or_id)
if res is None:
raise NotFoundError(
resource_type="resource",
resource_id=name_or_id,
)
return res
def show_type(name: str) -> ResourceTypeSpec:
spec = types.get(name)
if spec is None:
raise NotFoundError(
resource_type="resource_type",
resource_id=name,
)
return spec
registry.show_resource = show_resource
registry.show_type = show_type
return registry
class BindingResolutionContextualBench:
"""Benchmark contextual binding resolution latency."""
timeout = 60
params: list[int] = [1, 10, 50]
param_names: list[str] = ["num_resources"]
def setup(self, num_resources: int) -> None:
global _bench_counter
_bench_counter = 0
resources: dict[str, Resource] = {}
links: list[LinkedResource] = []
for i in range(num_resources):
rid = _bench_ulid()
res = Resource(
resource_id=rid,
name=None,
resource_type_name="git-checkout",
classification=PhysVirt.PHYSICAL,
)
resources[rid] = res
alias = "repo" if i == 0 else f"res-{i}"
links.append(
LinkedResource(
resource_id=rid, alias=alias
)
)
types: dict[str, ResourceTypeSpec] = {
"git-checkout": ResourceTypeSpec(
name="git-checkout",
resource_kind=ResourceKind.PHYSICAL,
sandbox_strategy=SandboxStrategy.NONE,
built_in=True,
),
}
self.registry = _make_registry(resources, types)
self.project = NamespacedProject(
name="bench",
namespace="local",
linked_resources=links,
)
self.tool = Tool(
name="bench/reader",
description="Benchmark tool",
source=ToolSource.BUILTIN,
tool_type=ToolType.TOOL,
resource_slots=[
ResourceSlot(
name="repo",
resource_type="git-checkout",
access=ResourceAccessMode.READ_WRITE,
binding=BindingMode.CONTEXTUAL,
),
],
)
self.service = BindingResolutionService(self.registry)
def time_resolve_contextual(
self, num_resources: int
) -> None:
self.service.resolve(self.tool, self.project)
class BindingResolutionStaticBench:
"""Benchmark static binding resolution latency."""
timeout = 60
def setup(self) -> None:
global _bench_counter
_bench_counter = 0
rid = _bench_ulid()
res = Resource(
resource_id=rid,
name="bench/config",
resource_type_name="fs-directory",
classification=PhysVirt.PHYSICAL,
)
resources = {
"bench/config": res,
rid: res,
}
types: dict[str, ResourceTypeSpec] = {
"fs-directory": ResourceTypeSpec(
name="fs-directory",
resource_kind=ResourceKind.PHYSICAL,
sandbox_strategy=SandboxStrategy.NONE,
built_in=True,
),
}
self.registry = _make_registry(resources, types)
self.project = NamespacedProject(
name="bench", namespace="local"
)
self.tool = Tool(
name="bench/static",
description="Benchmark static tool",
source=ToolSource.BUILTIN,
tool_type=ToolType.TOOL,
resource_slots=[
ResourceSlot(
name="config",
resource_type="fs-directory",
access=ResourceAccessMode.READ_ONLY,
binding=BindingMode.STATIC,
static_resource="bench/config",
),
],
)
self.service = BindingResolutionService(self.registry)
def time_resolve_static(self) -> None:
self.service.resolve(self.tool, self.project)
class BindingResolutionParameterBench:
"""Benchmark parameter binding (deferred) resolution."""
timeout = 60
def setup(self) -> None:
registry = MagicMock()
self.project = NamespacedProject(
name="bench", namespace="local"
)
self.tool = Tool(
name="bench/param",
description="Benchmark param tool",
source=ToolSource.BUILTIN,
tool_type=ToolType.TOOL,
resource_slots=[
ResourceSlot(
name="target",
resource_type="fs-directory",
access=ResourceAccessMode.READ_WRITE,
binding=BindingMode.PARAMETER,
),
],
)
self.service = BindingResolutionService(registry)
def time_resolve_parameter_deferred(self) -> None:
self.service.resolve(self.tool, self.project)
docs/reference/tool_bindings.md
+132
View File
@@ -0,0 +1,132 @@
# Tool Resource Bindings
Tools in CleverAgents declare **resource slots** that describe which
resources they operate on. At activation time the binding resolution
service maps each slot to a concrete resource from the registry.
## Binding Modes
### Contextual Binding (default)
The slot declares only a `resource_type` requirement. The system
searches the plan's project for linked resources that match at
activation time.
**Resolution rules:**
1. Collect all linked resources whose type matches the slot's
`resource_type` (including sub-types via `parent_types`).
2. If exactly **one** resource matches, it is automatically bound.
3. If **multiple** resources match, the slot `name` is used as a
hint: the resource whose alias (or short name) equals the slot
name is selected. If no alias matches, a `ValidationError` is
raised.
4. If **no** resource matches and the slot is `required`, a
`ValidationError` is raised.
```yaml
resource_slots:
- name: repo
resource_type: git-checkout
access: read_write
binding: contextual # default -- can be omitted
```
### Static Binding
The slot is hardcoded to a specific registered resource by name via
the `static_resource` field. Validated at registration time.
```yaml
resource_slots:
- name: config_dir
resource_type: fs-directory
access: read_only
binding: static
static_resource: local/shared-config
```
### Parameter Binding
The resource reference is passed as a tool argument at invocation
time. Resolution is deferred until the tool is actually called.
```yaml
resource_slots:
- name: target
resource_type: fs-directory
access: read_write
binding: parameter
```
At invocation the caller supplies `target=<resource-name-or-id>`.
## Resolution Order
For each resource slot on the tool:
1. Determine the binding mode (`static`, `parameter`, or
`contextual`).
2. **Static** -- look up the named resource in the registry;
verify type compatibility.
3. **Parameter** -- if invocation params include the slot name,
resolve eagerly; otherwise mark as *deferred*.
4. **Contextual** -- search the project's linked resources for
type-compatible matches and apply disambiguation rules.
## Type Compatibility
A resource is *type-compatible* with a slot when:
- The resource's type **exactly matches** the slot's
`resource_type`, **or**
- The resource's type declares the slot's `resource_type` in its
`parent_types` list (sub-type relationship).
For example, `fs-directory` lists `git-checkout` in its
`parent_types`, so an `fs-directory` resource satisfies a slot
requiring `git-checkout`.
## Examples
### Single Contextual Resource
```python
from cleveragents.application.services.binding_resolution_service import (
BindingResolutionService,
)
service = BindingResolutionService(resource_registry)
results = service.resolve(tool, project)
# results[0].binding_mode == "contextual"
# results[0].resource_id == "01HGZ..."
```
### Mixed Bindings
A tool may combine all three modes:
```yaml
resource_slots:
- name: repo
resource_type: git-checkout
access: read_write
binding: contextual
- name: config
resource_type: fs-directory
access: read_only
binding: static
static_resource: local/shared-config
- name: output
resource_type: fs-directory
access: write_only
binding: parameter
```
## API Reference
See:
- [`BindingResolutionService`](../../src/cleveragents/application/services/binding_resolution_service.py)
- [`BindingResult`](../../src/cleveragents/domain/models/core/resource_slot.py)
- [`ResourceSlot`](../../src/cleveragents/domain/models/core/tool.py)
features/binding_resolution.feature
+125
View File
@@ -0,0 +1,125 @@
@phase1 @domain @tool_binding
Feature: Tool Resource Binding Resolution
As a system activating a tool within a plan
I want resource slots to be resolved to concrete resources
So that the tool has access to the correct resources at runtime
Background:
Given a mock resource registry
And a binding resolution service using the registry
# ---------------------------------------------------------------------------
# Contextual binding -- single match (auto-bind)
# ---------------------------------------------------------------------------
@contextual @single_match
Scenario: Contextual binding with single matching resource
Given a project "local/my-project" with a linked resource:
| resource_id | type_name | alias |
| 01HGZ6FE0AQDYTR4BX00000001 | git-checkout | repo |
And a tool "local/reader" with a contextual slot "repo" of type "git-checkout"
When the bindings are resolved
Then the binding for slot "repo" should have mode "contextual"
And the binding for slot "repo" should have resource_id "01HGZ6FE0AQDYTR4BX00000001"
And the binding for slot "repo" should not be deferred
# ---------------------------------------------------------------------------
# Contextual binding -- multiple matches with alias hint
# ---------------------------------------------------------------------------
@contextual @alias_hint
Scenario: Contextual binding with multiple matches uses alias hint
Given a project "local/multi" with linked resources:
| resource_id | type_name | alias |
| 01HGZ6FE0AQDYTR4BX00000010 | git-checkout | repo |
| 01HGZ6FE0AQDYTR4BX00000011 | git-checkout | backup |
And a tool "local/reader" with a contextual slot "repo" of type "git-checkout"
When the bindings are resolved
Then the binding for slot "repo" should have resource_id "01HGZ6FE0AQDYTR4BX00000010"
# ---------------------------------------------------------------------------
# Contextual binding -- no match (error)
# ---------------------------------------------------------------------------
@contextual @no_match @error_handling
Scenario: Contextual binding with no matching resource raises error
Given a project "local/empty" with no linked resources
And a tool "local/reader" with a contextual slot "repo" of type "git-checkout"
When the binding resolution is attempted
Then a binding ValidationError should be raised mentioning "No resource of type"
# ---------------------------------------------------------------------------
# Static binding resolution
# ---------------------------------------------------------------------------
@static
Scenario: Static binding resolves named resource
Given a resource "local/shared-config" of type "fs-directory" in the registry
And a tool "local/deployer" with a static slot "config_dir" bound to "local/shared-config" of type "fs-directory"
And a project "local/proj" with no linked resources
When the bindings are resolved
Then the binding for slot "config_dir" should have mode "static"
And the binding for slot "config_dir" should have resource_id "01HGZ6FE0AQDYTR4BX00000020"
And the binding for slot "config_dir" should not be deferred
# ---------------------------------------------------------------------------
# Parameter binding (deferred)
# ---------------------------------------------------------------------------
@parameter @deferred
Scenario: Parameter binding is deferred when no invocation params given
Given a tool "local/builder" with a parameter slot "target" of type "fs-directory"
And a project "local/proj" with no linked resources
When the bindings are resolved
Then the binding for slot "target" should have mode "parameter"
And the binding for slot "target" should be deferred
And the binding for slot "target" should have no resource_id
# ---------------------------------------------------------------------------
# Type compatibility -- pass (sub-type)
# ---------------------------------------------------------------------------
@type_compat @pass
Scenario: Type compatibility passes for sub-type via parent_types
Given a project "local/typed" with a linked resource:
| resource_id | type_name | alias |
| 01HGZ6FE0AQDYTR4BX00000030 | fs-directory | repo |
And the resource type "fs-directory" has parent_types including "git-checkout"
And a tool "local/compat" with a contextual slot "repo" of type "git-checkout"
When the bindings are resolved
Then the binding for slot "repo" should have resource_id "01HGZ6FE0AQDYTR4BX00000030"
# ---------------------------------------------------------------------------
# Type compatibility -- fail
# ---------------------------------------------------------------------------
@type_compat @fail @error_handling
Scenario: Type compatibility fails for incompatible types
Given a resource "local/wrong-type" of type "fs-file" in the registry
And the resource type "fs-file" has no parent_types
And a tool "local/strict" with a static slot "data" bound to "local/wrong-type" of type "git-checkout"
And a project "local/proj" with no linked resources
When the binding resolution is attempted
Then a binding ValidationError should be raised mentioning "Type mismatch"
# ---------------------------------------------------------------------------
# Mixed bindings on same tool
# ---------------------------------------------------------------------------
@mixed
Scenario: Mixed bindings on same tool resolve correctly
Given a project "local/mixed" with a linked resource:
| resource_id | type_name | alias |
| 01HGZ6FE0AQDYTR4BX00000040 | git-checkout | repo |
And a resource "local/shared-config" of type "fs-directory" in the registry
And a tool "local/multi-bind" with mixed slots:
| name | resource_type | binding | static_resource |
| repo | git-checkout | contextual | |
| config | fs-directory | static | local/shared-config |
| output | fs-directory | parameter | |
When the bindings are resolved
Then there should be 3 binding results
And the binding for slot "repo" should have mode "contextual"
And the binding for slot "config" should have mode "static"
And the binding for slot "output" should have mode "parameter"
And the binding for slot "output" should be deferred
features/steps/binding_resolution_steps.py
+482
View File
@@ -0,0 +1,482 @@
"""Step definitions for binding_resolution.feature."""
from __future__ import annotations
from unittest.mock import MagicMock
from behave import given, then, when
from behave.runner import Context
from cleveragents.application.services.binding_resolution_service import (
BindingResolutionService,
)
from cleveragents.core.exceptions import NotFoundError, ValidationError
from cleveragents.domain.models.core.project import (
LinkedResource,
NamespacedProject,
)
from cleveragents.domain.models.core.resource import (
PhysVirt,
Resource,
)
from cleveragents.domain.models.core.resource_slot import BindingResult
from cleveragents.domain.models.core.resource_type import (
ResourceKind,
ResourceTypeSpec,
SandboxStrategy,
)
from cleveragents.domain.models.core.tool import (
BindingMode,
ResourceAccessMode,
ResourceSlot,
Tool,
ToolSource,
ToolType,
)
# ------------------------------------------------------------------
# Helpers
# ------------------------------------------------------------------
def _make_resource(
resource_id: str,
type_name: str,
name: str | None = None,
) -> Resource:
return Resource(
resource_id=resource_id,
name=name,
resource_type_name=type_name,
classification=PhysVirt.PHYSICAL,
description=f"Test resource {resource_id}",
)
def _make_type_spec(
name: str,
parent_types: list[str] | None = None,
) -> ResourceTypeSpec:
is_builtin = "/" not in name
return ResourceTypeSpec(
name=name,
description=f"Type {name}",
resource_kind=ResourceKind.PHYSICAL,
sandbox_strategy=SandboxStrategy.NONE,
parent_types=parent_types or [],
built_in=is_builtin,
)
# ------------------------------------------------------------------
# Background
# ------------------------------------------------------------------
@given("a mock resource registry")
def step_mock_registry(context: Context) -> None:
registry = MagicMock()
registry._resources: dict[str, Resource] = {}
registry._types: dict[str, ResourceTypeSpec] = {
"git-checkout": _make_type_spec(
"git-checkout", parent_types=[]
),
"fs-directory": _make_type_spec(
"fs-directory",
parent_types=["git-checkout", "fs-directory"],
),
"fs-file": _make_type_spec("fs-file", parent_types=[]),
}
def show_resource(name_or_id: str) -> Resource:
res = registry._resources.get(name_or_id)
if res is None:
raise NotFoundError(
resource_type="resource",
resource_id=name_or_id,
)
return res
def show_type(name: str) -> ResourceTypeSpec:
spec = registry._types.get(name)
if spec is None:
raise NotFoundError(
resource_type="resource_type",
resource_id=name,
)
return spec
registry.show_resource = show_resource
registry.show_type = show_type
context.mock_registry = registry
@given("a binding resolution service using the registry")
def step_service(context: Context) -> None:
context.binding_service = BindingResolutionService(
resource_registry=context.mock_registry,
)
# ------------------------------------------------------------------
# Given: projects
# ------------------------------------------------------------------
@given('a project "{name}" with a linked resource:')
def step_project_one_resource(context: Context, name: str) -> None:
parts = name.split("/", 1)
namespace = parts[0] if len(parts) == 2 else "local"
proj_name = parts[1] if len(parts) == 2 else parts[0]
links: list[LinkedResource] = []
for row in context.table:
rid = row["resource_id"]
tname = row["type_name"]
alias = row["alias"] if row["alias"] else None
resource = _make_resource(rid, tname, name=None)
context.mock_registry._resources[rid] = resource
links.append(
LinkedResource(
resource_id=rid,
alias=alias,
)
)
context.project = NamespacedProject(
name=proj_name,
namespace=namespace,
linked_resources=links,
)
@given('a project "{name}" with linked resources:')
def step_project_multi_resources(
context: Context, name: str
) -> None:
parts = name.split("/", 1)
namespace = parts[0] if len(parts) == 2 else "local"
proj_name = parts[1] if len(parts) == 2 else parts[0]
links: list[LinkedResource] = []
for row in context.table:
rid = row["resource_id"]
tname = row["type_name"]
alias = row["alias"] if row["alias"] else None
resource = _make_resource(rid, tname, name=None)
context.mock_registry._resources[rid] = resource
links.append(
LinkedResource(
resource_id=rid,
alias=alias,
)
)
context.project = NamespacedProject(
name=proj_name,
namespace=namespace,
linked_resources=links,
)
@given('a project "{name}" with no linked resources')
def step_project_empty(context: Context, name: str) -> None:
parts = name.split("/", 1)
namespace = parts[0] if len(parts) == 2 else "local"
proj_name = parts[1] if len(parts) == 2 else parts[0]
context.project = NamespacedProject(
name=proj_name,
namespace=namespace,
linked_resources=[],
)
# ------------------------------------------------------------------
# Given: resources in registry
# ------------------------------------------------------------------
@given(
'a resource "{res_name}" of type "{type_name}" in the registry'
)
def step_resource_in_registry(
context: Context,
res_name: str,
type_name: str,
) -> None:
rid = "01HGZ6FE0AQDYTR4BX00000020"
if res_name == "local/wrong-type":
rid = "01HGZ6FE0AQDYTR4BX00000050"
resource = _make_resource(rid, type_name, name=res_name)
context.mock_registry._resources[res_name] = resource
context.mock_registry._resources[rid] = resource
# ------------------------------------------------------------------
# Given: resource type overrides
# ------------------------------------------------------------------
@given(
'the resource type "{tname}" has parent_types including'
' "{parent}"'
)
def step_type_parent(
context: Context, tname: str, parent: str
) -> None:
existing = context.mock_registry._types.get(tname)
parents = list(existing.parent_types) if existing else []
if parent not in parents:
parents.append(parent)
context.mock_registry._types[tname] = _make_type_spec(
tname, parent_types=parents
)
@given('the resource type "{tname}" has no parent_types')
def step_type_no_parents(context: Context, tname: str) -> None:
context.mock_registry._types[tname] = _make_type_spec(
tname, parent_types=[]
)
# ------------------------------------------------------------------
# Given: tools
# ------------------------------------------------------------------
@given(
'a tool "{tool_name}" with a contextual slot "{slot}" of type'
' "{rtype}"'
)
def step_tool_contextual(
context: Context,
tool_name: str,
slot: str,
rtype: str,
) -> None:
context.tool = Tool(
name=tool_name,
description=f"Tool {tool_name}",
source=ToolSource.BUILTIN,
tool_type=ToolType.TOOL,
resource_slots=[
ResourceSlot(
name=slot,
resource_type=rtype,
access=ResourceAccessMode.READ_WRITE,
binding=BindingMode.CONTEXTUAL,
),
],
)
@given(
'a tool "{tool_name}" with a static slot "{slot}" bound to'
' "{res}" of type "{rtype}"'
)
def step_tool_static(
context: Context,
tool_name: str,
slot: str,
res: str,
rtype: str,
) -> None:
context.tool = Tool(
name=tool_name,
description=f"Tool {tool_name}",
source=ToolSource.BUILTIN,
tool_type=ToolType.TOOL,
resource_slots=[
ResourceSlot(
name=slot,
resource_type=rtype,
access=ResourceAccessMode.READ_ONLY,
binding=BindingMode.STATIC,
static_resource=res,
),
],
)
@given(
'a tool "{tool_name}" with a parameter slot "{slot}" of type'
' "{rtype}"'
)
def step_tool_parameter(
context: Context,
tool_name: str,
slot: str,
rtype: str,
) -> None:
context.tool = Tool(
name=tool_name,
description=f"Tool {tool_name}",
source=ToolSource.BUILTIN,
tool_type=ToolType.TOOL,
resource_slots=[
ResourceSlot(
name=slot,
resource_type=rtype,
access=ResourceAccessMode.READ_WRITE,
binding=BindingMode.PARAMETER,
),
],
)
@given('a tool "{tool_name}" with mixed slots:')
def step_tool_mixed(context: Context, tool_name: str) -> None:
slots: list[ResourceSlot] = []
for row in context.table:
binding = BindingMode(row["binding"])
static_res: str | None = (
row["static_resource"]
if row["static_resource"]
else None
)
slots.append(
ResourceSlot(
name=row["name"],
resource_type=row["resource_type"],
access=ResourceAccessMode.READ_WRITE,
binding=binding,
static_resource=static_res,
)
)
context.tool = Tool(
name=tool_name,
description=f"Tool {tool_name}",
source=ToolSource.BUILTIN,
tool_type=ToolType.TOOL,
resource_slots=slots,
)
# ------------------------------------------------------------------
# When
# ------------------------------------------------------------------
@when("the bindings are resolved")
def step_resolve(context: Context) -> None:
context.binding_results = context.binding_service.resolve(
context.tool,
context.project,
)
@when("the binding resolution is attempted")
def step_resolve_attempt(context: Context) -> None:
try:
context.binding_results = context.binding_service.resolve(
context.tool,
context.project,
)
context.binding_error = None
except ValidationError as exc:
context.binding_error = exc
context.binding_results = None
# ------------------------------------------------------------------
# Then
# ------------------------------------------------------------------
@then(
'the binding for slot "{slot}" should have mode "{mode}"'
)
def step_check_mode(
context: Context, slot: str, mode: str
) -> None:
result = _find_result(context, slot)
assert result.binding_mode == mode, (
f"Expected mode '{mode}', "
f"got '{result.binding_mode}'"
)
@then(
'the binding for slot "{slot}" should have resource_id'
' "{rid}"'
)
def step_check_rid(
context: Context, slot: str, rid: str
) -> None:
result = _find_result(context, slot)
assert result.resource_id == rid, (
f"Expected resource_id '{rid}', "
f"got '{result.resource_id}'"
)
@then('the binding for slot "{slot}" should not be deferred')
def step_not_deferred(context: Context, slot: str) -> None:
result = _find_result(context, slot)
assert not result.deferred, (
f"Slot '{slot}' should not be deferred"
)
@then('the binding for slot "{slot}" should be deferred')
def step_is_deferred(context: Context, slot: str) -> None:
result = _find_result(context, slot)
assert result.deferred, (
f"Slot '{slot}' should be deferred"
)
@then('the binding for slot "{slot}" should have no resource_id')
def step_no_rid(context: Context, slot: str) -> None:
result = _find_result(context, slot)
assert result.resource_id is None, (
f"Expected None resource_id, got '{result.resource_id}'"
)
@then(
'a binding ValidationError should be raised mentioning'
' "{text}"'
)
def step_binding_error(context: Context, text: str) -> None:
assert context.binding_error is not None, (
"Expected ValidationError but none was raised"
)
assert isinstance(context.binding_error, ValidationError), (
f"Expected ValidationError, "
f"got {type(context.binding_error).__name__}"
)
assert text in str(context.binding_error), (
f"Expected '{text}' in error, "
f"got '{context.binding_error}'"
)
@then("there should be {count:d} binding results")
def step_count_results(context: Context, count: int) -> None:
assert len(context.binding_results) == count, (
f"Expected {count} results, "
f"got {len(context.binding_results)}"
)
# ------------------------------------------------------------------
# Internal helper
# ------------------------------------------------------------------
def _find_result(
context: Context, slot_name: str
) -> BindingResult:
for result in context.binding_results:
if result.slot_name == slot_name:
return result
raise AssertionError(
f"No binding result for slot '{slot_name}'"
)
implementation_plan.md
+13 -13
View File
@@ -2623,19 +2623,19 @@ No standalone Q0-Advanced commits planned. Advanced QA enhancements are bundled
- [ ] Git [Jeff]: `git branch -d feature/m3-tool-registry`
- [ ] Quality [Jeff]: Verify coverage >=97% via `nox -s coverage_report`. If coverage is <97% then review the current unit test coverage report at `build/coverage.xml` and use it to write new Behave based unit tests to improve code coverage. Specifically, write Behave style unit tests that are descriptively named and specifically improves coverage on whichever file has the most uncovered lines by writing tests that will target the uncovered lines in the report. Once that is done rerun `nox -s coverage_report` to verify all tests pass and coverage is above >=97%. Only mark this as complete once coverage is >=97%, if not repeat this task as many times as is needed until coverage reaches >=97%.
- [ ] **COMMIT (Owner: Jeff | Group: C1.tool.binding | Branch: feature/m3-tool-binding | Planned: Day 15 | Expected: Day 21) - Commit message: "feat(tool): add resource binding resolution"**
- [ ] Git [Jeff]: `git checkout master`
- [ ] Git [Jeff]: `git pull origin master`
- [ ] Git [Jeff]: `git checkout -b feature/m3-tool-binding`
- [ ] Git [Jeff]: `git fetch origin && git merge origin/master` (run before final tests and before commit)
- [ ] Code [Jeff]: Implement binding resolution for contextual, static, and parameter bindings with type compatibility checks.
- [ ] Docs [Jeff]: Add `docs/reference/tool_bindings.md` with resolution order and examples.
- [ ] Tests (Behave) [Jeff]: Add binding resolution scenarios (context vs static vs parameter).
- [ ] Tests (Robot) [Jeff]: Add Robot test resolving a bound resource by name.
- [ ] Tests (ASV) [Jeff]: Add `benchmarks/binding_resolution_bench.py` for resolution latency.
- [ ] Quality [Jeff]: Run `nox` (all default sessions, including benchmark), fix any errors if needed ensuring nox passes.
- [ ] Git [Jeff]: `git add .`
- [ ] Git [Jeff]: `git commit -m "feat(tool): add resource binding resolution"`
- [x] **COMMIT (Owner: Jeff | Group: C1.tool.binding | Branch: feature/m3-tool-binding | Planned: Day 15 | Expected: Day 21) - Commit message: "feat(tool): add resource binding resolution"**
- [x] Git [Jeff]: `git checkout master`
- [x] Git [Jeff]: `git pull origin master`
- [x] Git [Jeff]: `git checkout -b feature/m3-tool-binding`
- [x] Git [Jeff]: `git fetch origin && git merge origin/master` (run before final tests and before commit)
- [x] Code [Jeff]: Implement binding resolution for contextual, static, and parameter bindings with type compatibility checks.
- [x] Docs [Jeff]: Add `docs/reference/tool_bindings.md` with resolution order and examples.
- [x] Tests (Behave) [Jeff]: Add binding resolution scenarios (context vs static vs parameter).
- [x] Tests (Robot) [Jeff]: Add Robot test resolving a bound resource by name.
- [x] Tests (ASV) [Jeff]: Add `benchmarks/binding_resolution_bench.py` for resolution latency.
- [x] Quality [Jeff]: Run `nox` (all default sessions, including benchmark), fix any errors if needed ensuring nox passes.
- [x] Git [Jeff]: `git add .`
- [x] Git [Jeff]: `git commit -m "feat(tool): add resource binding resolution"`
- [ ] Forgejo PR [Jeff]: Open PR from `feature/m3-tool-binding` to `master` with description "Add tool resource binding resolution and docs/tests.".
- [ ] Git [Jeff]: `git checkout master`
- [ ] Git [Jeff]: `git branch -d feature/m3-tool-binding`
robot/binding_resolution.robot
+85
View File
@@ -0,0 +1,85 @@
*** Settings ***
Documentation Binding Resolution Service Smoke Tests
Library Process
Library OperatingSystem
*** Variables ***
${PYTHON} python
*** Test Cases ***
Contextual Binding Resolves Single Resource
[Documentation] Resolve a contextual slot with one matching linked resource
${script}= Catenate SEPARATOR=\n
... from unittest.mock import MagicMock
... from cleveragents.core.exceptions import NotFoundError
... from cleveragents.domain.models.core.resource import Resource, PhysVirt
... from cleveragents.domain.models.core.resource_type import ResourceTypeSpec, ResourceKind, SandboxStrategy
... from cleveragents.domain.models.core.project import NamespacedProject, LinkedResource
... from cleveragents.domain.models.core.tool import Tool, ToolSource, ToolType, ResourceSlot, ResourceAccessMode, BindingMode
... from cleveragents.application.services.binding_resolution_service import BindingResolutionService
... registry = MagicMock()
... res = Resource(resource_id="01HGZ6FE0AQDYTR4BX00000001", name=None, resource_type_name="git-checkout", classification=PhysVirt.PHYSICAL)
... registry.show_resource = lambda x: res
... registry.show_type = lambda x: ResourceTypeSpec(name="git-checkout", resource_kind=ResourceKind.PHYSICAL, sandbox_strategy=SandboxStrategy.NONE, built_in=True)
... project = NamespacedProject(name="proj", namespace="local", linked_resources=[LinkedResource(resource_id="01HGZ6FE0AQDYTR4BX00000001", alias="repo")])
... tool = Tool(name="local/reader", description="Read", source=ToolSource.BUILTIN, tool_type=ToolType.TOOL, resource_slots=[ResourceSlot(name="repo", resource_type="git-checkout", access=ResourceAccessMode.READ_WRITE, binding=BindingMode.CONTEXTUAL)])
... svc = BindingResolutionService(registry)
... results = svc.resolve(tool, project)
... assert len(results) == 1, f"Expected 1 result, got {len(results)}"
... assert results[0].binding_mode == "contextual"
... assert results[0].resource_id == "01HGZ6FE0AQDYTR4BX00000001"
... assert not results[0].deferred
... print("Contextual binding resolution passed")
${result}= Run Process ${PYTHON} -c ${script}
Should Be Equal As Integers ${result.rc} 0 Contextual binding test failed: ${result.stderr}
Should Contain ${result.stdout} Contextual binding resolution passed
Static Binding Resolves Named Resource
[Documentation] Resolve a static slot bound to a named resource
${script}= Catenate SEPARATOR=\n
... from unittest.mock import MagicMock
... from cleveragents.core.exceptions import NotFoundError
... from cleveragents.domain.models.core.resource import Resource, PhysVirt
... from cleveragents.domain.models.core.resource_type import ResourceTypeSpec, ResourceKind, SandboxStrategy
... from cleveragents.domain.models.core.project import NamespacedProject
... from cleveragents.domain.models.core.tool import Tool, ToolSource, ToolType, ResourceSlot, ResourceAccessMode, BindingMode
... from cleveragents.application.services.binding_resolution_service import BindingResolutionService
... registry = MagicMock()
... res = Resource(resource_id="01HGZ6FE0AQDYTR4BX00000020", name="local/shared-config", resource_type_name="fs-directory", classification=PhysVirt.PHYSICAL)
... registry.show_resource = lambda x: res
... registry.show_type = lambda x: ResourceTypeSpec(name="fs-directory", resource_kind=ResourceKind.PHYSICAL, sandbox_strategy=SandboxStrategy.NONE, built_in=True)
... project = NamespacedProject(name="proj", namespace="local")
... tool = Tool(name="local/deployer", description="Deploy", source=ToolSource.BUILTIN, tool_type=ToolType.TOOL, resource_slots=[ResourceSlot(name="config_dir", resource_type="fs-directory", access=ResourceAccessMode.READ_ONLY, binding=BindingMode.STATIC, static_resource="local/shared-config")])
... svc = BindingResolutionService(registry)
... results = svc.resolve(tool, project)
... assert len(results) == 1
... assert results[0].binding_mode == "static"
... assert results[0].resource_id == "01HGZ6FE0AQDYTR4BX00000020"
... assert not results[0].deferred
... print("Static binding resolution passed")
${result}= Run Process ${PYTHON} -c ${script}
Should Be Equal As Integers ${result.rc} 0 Static binding test failed: ${result.stderr}
Should Contain ${result.stdout} Static binding resolution passed
Parameter Binding Is Deferred
[Documentation] Parameter binding returns a deferred result
${script}= Catenate SEPARATOR=\n
... from cleveragents.domain.models.core.project import NamespacedProject
... from cleveragents.domain.models.core.tool import Tool, ToolSource, ToolType, ResourceSlot, ResourceAccessMode, BindingMode
... from cleveragents.application.services.binding_resolution_service import BindingResolutionService
... from unittest.mock import MagicMock
... registry = MagicMock()
... project = NamespacedProject(name="proj", namespace="local")
... tool = Tool(name="local/builder", description="Build", source=ToolSource.BUILTIN, tool_type=ToolType.TOOL, resource_slots=[ResourceSlot(name="target", resource_type="fs-directory", access=ResourceAccessMode.READ_WRITE, binding=BindingMode.PARAMETER)])
... svc = BindingResolutionService(registry)
... results = svc.resolve(tool, project)
... assert len(results) == 1
... assert results[0].binding_mode == "parameter"
... assert results[0].deferred is True
... assert results[0].resource_id is None
... print("Parameter binding deferred passed")
${result}= Run Process ${PYTHON} -c ${script}
Should Be Equal As Integers ${result.rc} 0 Parameter binding test failed: ${result.stderr}
Should Contain ${result.stdout} Parameter binding deferred passed
*** Keywords ***
src/cleveragents/application/services/binding_resolution_service.py
+397
View File
@@ -0,0 +1,397 @@
"""Binding resolution service for tool resource slots.
Resolves each :class:`~cleveragents.domain.models.core.tool.ResourceSlot`
on a tool to a concrete resource (or marks it as *deferred* for parameter
bindings). Three binding modes are supported:
Contextual (default)
The slot declares a ``resource_type`` requirement. The service
searches the plan's project for linked resources whose type matches
(or is a compatible sub-type). Exactly one match is auto-bound;
multiple matches use the slot ``name`` as a hint against the
linked-resource alias. No match raises an error.
Static
The slot's ``static_resource`` field names a registered resource
directly. Resolved by looking it up in the resource registry.
Parameter
The slot declares ``binding=parameter``. Resolution is deferred
to invocation time -- the caller passes the resource reference as a
tool argument.
Based on:
- ``docs/specification.md`` -- Binding Resolution (lines 7291-7422)
- ``implementation_plan.md`` -- Task C1.tool.binding
- ADR-011: Tool System
See Also:
- :class:`~cleveragents.domain.models.core.tool.ResourceSlot`
- :class:`~cleveragents.domain.models.core.resource_slot.BindingResult`
"""
from __future__ import annotations
import logging
from typing import Any
from cleveragents.core.exceptions import NotFoundError, ValidationError
from cleveragents.domain.models.core.project import (
LinkedResource,
NamespacedProject,
)
from cleveragents.domain.models.core.resource import Resource
from cleveragents.domain.models.core.resource_slot import BindingResult
from cleveragents.domain.models.core.resource_type import ResourceTypeSpec
from cleveragents.domain.models.core.tool import (
BindingMode,
ResourceSlot,
Tool,
)
logger = logging.getLogger(__name__)
class BindingResolutionService:
"""Resolve resource-slot bindings for a tool activation.
Requires access to a *resource registry* (for resource and type
look-ups) and the *project* that owns the plan being activated.
The registry is injected as a protocol-compatible object exposing
``show_resource`` and ``show_type`` methods (matching
:class:`~cleveragents.application.services.resource_registry_service.ResourceRegistryService`).
"""
def __init__(self, resource_registry: Any) -> None:
"""Initialise with a resource registry dependency.
Args:
resource_registry: Object with ``show_resource(name_or_id)``
and ``show_type(name)`` methods.
"""
self._registry = resource_registry
# ------------------------------------------------------------------
# Public API
# ------------------------------------------------------------------
def resolve(
self,
tool: Tool,
project: NamespacedProject,
*,
invocation_params: dict[str, str] | None = None,
) -> list[BindingResult]:
"""Resolve all resource slots on *tool* within *project*.
Args:
tool: The tool whose slots need resolution.
project: The project providing linked resources for
contextual binding.
invocation_params: Optional mapping of parameter names to
resource references (used by ``parameter`` bindings).
Returns:
A :class:`BindingResult` for each slot.
Raises:
ValidationError: When a required slot cannot be resolved.
"""
results: list[BindingResult] = []
for slot in tool.resource_slots:
result = self._resolve_slot(
slot,
project,
invocation_params=invocation_params,
)
results.append(result)
return results
# ------------------------------------------------------------------
# Internal resolution dispatch
# ------------------------------------------------------------------
def _resolve_slot(
self,
slot: ResourceSlot,
project: NamespacedProject,
*,
invocation_params: dict[str, str] | None = None,
) -> BindingResult:
"""Dispatch to the appropriate resolver based on binding mode."""
if slot.binding == BindingMode.STATIC:
return self._resolve_static(slot)
if slot.binding == BindingMode.PARAMETER:
return self._resolve_parameter(
slot,
invocation_params=invocation_params,
)
return self._resolve_contextual(slot, project)
# ------------------------------------------------------------------
# Static binding
# ------------------------------------------------------------------
def _resolve_static(self, slot: ResourceSlot) -> BindingResult:
"""Resolve a statically-bound slot.
The resource is identified by the ``static_resource`` field.
"""
assert slot.static_resource is not None # enforced by model
try:
resource: Resource = self._registry.show_resource(
slot.static_resource,
)
except NotFoundError as exc:
raise ValidationError(
message=(
f"Static binding for slot '{slot.name}' "
f"references unknown resource "
f"'{slot.static_resource}'"
),
details={
"slot": slot.name,
"resource": slot.static_resource,
},
) from exc
self._check_type_compatibility(
slot,
resource.resource_type_name,
)
return BindingResult(
slot_name=slot.name,
resource_id=resource.resource_id,
resource_name=resource.name,
binding_mode="static",
deferred=False,
)
# ------------------------------------------------------------------
# Parameter binding
# ------------------------------------------------------------------
def _resolve_parameter(
self,
slot: ResourceSlot,
*,
invocation_params: dict[str, str] | None = None,
) -> BindingResult:
"""Resolve a parameter-bound slot.
If *invocation_params* contains the slot name, the resource is
eagerly resolved. Otherwise the result is marked as deferred.
"""
ref = (invocation_params or {}).get(slot.name)
if ref is None:
return BindingResult(
slot_name=slot.name,
resource_id=None,
resource_name=None,
binding_mode="parameter",
deferred=True,
)
try:
resource: Resource = self._registry.show_resource(ref)
except NotFoundError as exc:
raise ValidationError(
message=(
f"Parameter binding for slot '{slot.name}' "
f"references unknown resource '{ref}'"
),
details={"slot": slot.name, "resource": ref},
) from exc
self._check_type_compatibility(
slot,
resource.resource_type_name,
)
return BindingResult(
slot_name=slot.name,
resource_id=resource.resource_id,
resource_name=resource.name,
binding_mode="parameter",
deferred=False,
)
# ------------------------------------------------------------------
# Contextual binding
# ------------------------------------------------------------------
def _resolve_contextual(
self,
slot: ResourceSlot,
project: NamespacedProject,
) -> BindingResult:
"""Resolve a contextual slot from project linked resources.
Resolution rules:
1. Collect all linked resources whose type matches the slot's
``resource_type`` (including sub-types).
2. If exactly one match: auto-bind.
3. If multiple matches: use ``slot.name`` as alias hint.
4. If no match and ``slot.required``: raise an error.
"""
candidates = self._find_matching_resources(
slot,
project.linked_resources,
)
if len(candidates) == 1:
resource, _link = candidates[0]
return BindingResult(
slot_name=slot.name,
resource_id=resource.resource_id,
resource_name=resource.name,
binding_mode="contextual",
deferred=False,
)
if len(candidates) > 1:
return self._disambiguate(slot, candidates)
# No matches
if slot.required:
raise ValidationError(
message=(
f"No resource of type '{slot.resource_type}' "
f"found in project "
f"'{project.namespaced_name}' for "
f"required slot '{slot.name}'"
),
details={
"slot": slot.name,
"resource_type": slot.resource_type,
"project": project.namespaced_name,
},
)
return BindingResult(
slot_name=slot.name,
resource_id=None,
resource_name=None,
binding_mode="contextual",
deferred=False,
)
# ------------------------------------------------------------------
# Helpers
# ------------------------------------------------------------------
def _find_matching_resources(
self,
slot: ResourceSlot,
linked_resources: list[LinkedResource],
) -> list[tuple[Resource, LinkedResource]]:
"""Return linked resources whose type is compatible."""
matches: list[tuple[Resource, LinkedResource]] = []
for link in linked_resources:
try:
resource = self._registry.show_resource(
link.resource_id,
)
except NotFoundError:
continue
if self._is_type_compatible(
slot.resource_type,
resource.resource_type_name,
):
matches.append((resource, link))
return matches
def _disambiguate(
self,
slot: ResourceSlot,
candidates: list[tuple[Resource, LinkedResource]],
) -> BindingResult:
"""Pick the best candidate using the slot name as alias hint."""
for resource, link in candidates:
if link.alias == slot.name:
return BindingResult(
slot_name=slot.name,
resource_id=resource.resource_id,
resource_name=resource.name,
binding_mode="contextual",
deferred=False,
)
# Also try matching slot name against resource name suffix
for resource, _link in candidates:
if resource.name is not None:
short = resource.name.rsplit("/", 1)[-1]
if short == slot.name:
return BindingResult(
slot_name=slot.name,
resource_id=resource.resource_id,
resource_name=resource.name,
binding_mode="contextual",
deferred=False,
)
names = [
r.name or r.resource_id for r, _ in candidates
]
raise ValidationError(
message=(
f"Ambiguous contextual binding for slot "
f"'{slot.name}': multiple resources of type "
f"'{slot.resource_type}' found and no alias "
f"hint matched. Candidates: {names}"
),
details={
"slot": slot.name,
"candidates": names,
},
)
def _is_type_compatible(
self,
required_type: str,
actual_type: str,
) -> bool:
"""Check if *actual_type* satisfies *required_type*.
Exact match always passes. If *actual_type* declares
*required_type* in its ``parent_types`` hierarchy, the
check also passes (sub-type relationship).
"""
if actual_type == required_type:
return True
try:
spec: ResourceTypeSpec = self._registry.show_type(
actual_type,
)
except NotFoundError:
return False
return required_type in spec.parent_types
def _check_type_compatibility(
self,
slot: ResourceSlot,
actual_type: str,
) -> None:
"""Raise ``ValidationError`` if types are incompatible."""
if not self._is_type_compatible(
slot.resource_type,
actual_type,
):
raise ValidationError(
message=(
f"Type mismatch for slot '{slot.name}': "
f"expected '{slot.resource_type}', "
f"got '{actual_type}'"
),
details={
"slot": slot.name,
"expected_type": slot.resource_type,
"actual_type": actual_type,
},
)
src/cleveragents/domain/models/core/__init__.py
+4
View File
@@ -86,6 +86,9 @@ from cleveragents.domain.models.core.resource import (
SandboxStrategy,
)
# Binding result model
from cleveragents.domain.models.core.resource_slot import BindingResult
# Resource type domain model
from cleveragents.domain.models.core.resource_type import (
ResourceKind,
@@ -143,6 +146,7 @@ __all__ = [
"Actor",
"AutomationProfile",
"BindingMode",
"BindingResult",
"Change",
"ChangeEntry",
"ChangeOperation",
src/cleveragents/domain/models/core/resource_slot.py
+73
View File
@@ -0,0 +1,73 @@
"""Binding result model for resolved resource slot bindings.
When a tool is activated within a plan context, each of its
:class:`~cleveragents.domain.models.core.tool.ResourceSlot` declarations
must be resolved to a concrete resource. The resolution produces a
:class:`BindingResult` for every slot, capturing the outcome (or
deferral) of the resolution process.
Three binding modes are supported:
| Mode | Determination |
|---------------|----------------------------------------------|
| ``contextual``| Resolved from the plan's project at runtime |
| ``static`` | Hardcoded at registration via ``static_resource`` |
| ``parameter`` | Passed as a tool argument at invocation time |
See Also:
- :class:`~cleveragents.domain.models.core.tool.ResourceSlot`
- :mod:`~cleveragents.application.services.binding_resolution_service`
"""
from __future__ import annotations
from pydantic import BaseModel, ConfigDict, Field
class BindingResult(BaseModel):
"""Outcome of resolving a single resource slot.
Produced by
:meth:`~cleveragents.application.services.binding_resolution_service.BindingResolutionService.resolve`
for each :class:`~cleveragents.domain.models.core.tool.ResourceSlot`
on a tool.
When ``deferred`` is ``True`` the slot uses *parameter* binding and
will only be fully resolved at invocation time.
"""
slot_name: str = Field(
...,
min_length=1,
description="Name of the resource slot being resolved",
)
resource_id: str | None = Field(
default=None,
description=(
"ULID of the resolved resource "
"(None for deferred parameter bindings)"
),
)
resource_name: str | None = Field(
default=None,
description="Namespaced name of the resolved resource",
)
binding_mode: str = Field(
...,
description=(
"How the binding was resolved: "
"'contextual', 'static', or 'parameter'"
),
)
deferred: bool = Field(
default=False,
description=(
"True when the binding is deferred to invocation "
"time (parameter binding)"
),
)
model_config = ConfigDict(
frozen=True,
str_strip_whitespace=True,
)