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

Merge branch 'master' into fix/database-parallelism

Browse Source
This commit is contained in:
Brent Edwards
2026-02-24 00:45:01 +00:00
parent f2b9ccfde4 88f1a1264b
commit dd963daeaa
25 changed files with 4400 additions and 210 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+3
View File
@@ -164,3 +164,6 @@ hive-mind-prompt-*.txt
.cleveragents/
.pabotsuitenames
PIPE
# Git worktrees for parallel task branches
worktrees/
benchmarks/decision_model_bench.py
+192
View File
@@ -0,0 +1,192 @@
"""ASV benchmarks for decision domain model operations.
Measures the performance of:
- Decision model construction (Pydantic validation)
- DecisionType enum access
- ContextSnapshot construction
- Decision serialization (model_dump / model_validate)
"""
from __future__ import annotations
import sys
from pathlib import Path
try:
from cleveragents.domain.models.core.decision import (
ArtifactRef,
ContextSnapshot,
Decision,
DecisionType,
ResourceRef,
)
except ModuleNotFoundError:
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
from cleveragents.domain.models.core.decision import (
ArtifactRef,
ContextSnapshot,
Decision,
DecisionType,
ResourceRef,
)
from ulid import ULID
_PLAN_ID = str(ULID())
_PARENT_ID = str(ULID())
_RESOURCE_ID = str(ULID())
class DecisionConstructionSuite:
"""Benchmark Decision model construction."""
def time_minimal_root_decision(self) -> None:
"""Benchmark minimal prompt_definition creation."""
Decision(
plan_id=_PLAN_ID,
sequence_number=0,
decision_type=DecisionType.PROMPT_DEFINITION,
question="What should we build?",
chosen_option="A REST API",
)
def time_child_decision_with_confidence(self) -> None:
"""Benchmark child decision with confidence score."""
Decision(
plan_id=_PLAN_ID,
parent_decision_id=_PARENT_ID,
sequence_number=1,
decision_type=DecisionType.STRATEGY_CHOICE,
question="Which framework?",
chosen_option="FastAPI",
alternatives_considered=["Flask", "Django", "Starlette"],
confidence_score=0.85,
)
def time_decision_with_snapshot(self) -> None:
"""Benchmark decision with full context snapshot."""
Decision(
plan_id=_PLAN_ID,
parent_decision_id=_PARENT_ID,
sequence_number=2,
decision_type=DecisionType.IMPLEMENTATION_CHOICE,
question="How to implement auth?",
chosen_option="JWT tokens",
context_snapshot=ContextSnapshot(
hot_context_hash="sha256:bench123",
hot_context_ref="store://snapshots/bench123",
relevant_resources=[
ResourceRef(resource_id=_RESOURCE_ID, path="src/main.py"),
ResourceRef(resource_id=str(ULID())),
],
actor_state_ref="checkpoint://actor/bench",
),
)
def time_correction_decision(self) -> None:
"""Benchmark correction decision creation."""
Decision(
plan_id=_PLAN_ID,
parent_decision_id=_PARENT_ID,
sequence_number=3,
decision_type=DecisionType.STRATEGY_CHOICE,
question="Which framework?",
chosen_option="Django instead",
is_correction=True,
corrects_decision_id=str(ULID()),
correction_reason="Changed requirements",
)
def time_decision_with_artifacts(self) -> None:
"""Benchmark decision with artifacts list."""
Decision(
plan_id=_PLAN_ID,
parent_decision_id=_PARENT_ID,
sequence_number=4,
decision_type=DecisionType.TOOL_INVOCATION,
question="Which tool?",
chosen_option="code_editor",
artifacts_produced=[
ArtifactRef(artifact_path=f"src/file_{i}.py") for i in range(10)
],
)
class DecisionSerializationSuite:
"""Benchmark Decision serialization round-trips."""
def setup(self) -> None:
self.decision = Decision(
plan_id=_PLAN_ID,
parent_decision_id=_PARENT_ID,
sequence_number=5,
decision_type=DecisionType.STRATEGY_CHOICE,
question="Architecture choice?",
chosen_option="Microservices",
alternatives_considered=["Monolith", "Serverless"],
confidence_score=0.78,
context_snapshot=ContextSnapshot(
hot_context_hash="sha256:serial",
hot_context_ref="store://serial",
relevant_resources=[
ResourceRef(resource_id=_RESOURCE_ID, path="src/app.py"),
],
actor_state_ref="checkpoint://serial",
),
artifacts_produced=[
ArtifactRef(artifact_path="src/service.py"),
],
)
self.dump = self.decision.model_dump()
def time_model_dump(self) -> None:
"""Benchmark model_dump serialization."""
self.decision.model_dump()
def time_model_dump_json(self) -> None:
"""Benchmark model_dump_json serialization."""
self.decision.model_dump_json()
def time_model_validate(self) -> None:
"""Benchmark model_validate deserialization."""
Decision.model_validate(self.dump)
def time_as_cli_dict(self) -> None:
"""Benchmark as_cli_dict rendering."""
self.decision.as_cli_dict()
class ContextSnapshotSuite:
"""Benchmark ContextSnapshot construction."""
def time_empty_snapshot(self) -> None:
"""Benchmark empty snapshot creation."""
ContextSnapshot()
def time_snapshot_with_resources(self) -> None:
"""Benchmark snapshot with multiple resource refs."""
ContextSnapshot(
hot_context_hash="sha256:bench",
hot_context_ref="store://bench",
relevant_resources=[
ResourceRef(resource_id=str(ULID()), path=f"src/f{i}.py")
for i in range(20)
],
actor_state_ref="checkpoint://bench",
)
class DecisionTypeEnumSuite:
"""Benchmark DecisionType enum operations."""
def time_enum_access(self) -> None:
"""Benchmark accessing an enum member."""
_ = DecisionType.PROMPT_DEFINITION
def time_enum_from_value(self) -> None:
"""Benchmark creating enum from string value."""
DecisionType("strategy_choice")
def time_enum_iteration(self) -> None:
"""Benchmark iterating all enum members."""
list(DecisionType)
benchmarks/resource_handler_bench.py
+214
View File
@@ -0,0 +1,214 @@
"""ASV benchmarks for resource handler resolution overhead.
Measures the performance of:
- Handler resolver (import + cache lookup)
- GitCheckoutHandler.resolve() with mock sandbox
- FsDirectoryHandler.resolve() with mock sandbox
- ResourceHandlerService.resolve_binding() end-to-end
"""
from __future__ import annotations
import sys
from datetime import datetime
from pathlib import Path
from unittest.mock import MagicMock, PropertyMock
try:
from cleveragents.domain.models.core.resource import (
PhysVirt,
Resource,
ResourceCapabilities,
)
from cleveragents.domain.models.core.resource_slot import BindingResult
from cleveragents.domain.models.core.resource_type import (
ResourceKind,
ResourceTypeSpec,
)
from cleveragents.domain.models.core.resource_type import (
SandboxStrategy as TypeSandboxStrategy,
)
from cleveragents.infrastructure.sandbox.factory import SandboxFactory
from cleveragents.infrastructure.sandbox.manager import SandboxManager
from cleveragents.infrastructure.sandbox.protocol import (
SandboxContext,
SandboxStatus,
)
from cleveragents.resource.handlers.fs_directory import FsDirectoryHandler
from cleveragents.resource.handlers.git_checkout import GitCheckoutHandler
from cleveragents.resource.handlers.resolver import (
clear_handler_cache,
resolve_handler,
)
except ModuleNotFoundError:
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
from cleveragents.domain.models.core.resource import (
PhysVirt,
Resource,
ResourceCapabilities,
)
from cleveragents.domain.models.core.resource_slot import BindingResult
from cleveragents.domain.models.core.resource_type import (
ResourceKind,
ResourceTypeSpec,
)
from cleveragents.domain.models.core.resource_type import (
SandboxStrategy as TypeSandboxStrategy,
)
from cleveragents.infrastructure.sandbox.factory import SandboxFactory
from cleveragents.infrastructure.sandbox.manager import SandboxManager
from cleveragents.infrastructure.sandbox.protocol import (
SandboxContext,
SandboxStatus,
)
from cleveragents.resource.handlers.fs_directory import FsDirectoryHandler
from cleveragents.resource.handlers.git_checkout import GitCheckoutHandler
from cleveragents.resource.handlers.resolver import (
clear_handler_cache,
resolve_handler,
)
def _make_resource(rtype: str, location: str) -> Resource:
return Resource(
resource_id="01KJ5C5TPMP8GGX3QC83E2MAQS",
resource_type_name=rtype,
classification=PhysVirt.PHYSICAL,
location=location,
capabilities=ResourceCapabilities(
readable=True, writable=True, sandboxable=True
),
)
def _make_mock_manager() -> SandboxManager:
mock_factory = MagicMock(spec=SandboxFactory)
mock_sandbox = MagicMock()
mock_sandbox.sandbox_id = "sb-bench-001"
type(mock_sandbox).status = PropertyMock(return_value=SandboxStatus.CREATED)
mock_sandbox.context = SandboxContext(
sandbox_id="sb-bench-001",
sandbox_path="/tmp/sandbox/sb-bench-001",
original_path="/tmp/original",
resource_id="res-bench",
plan_id="plan-bench",
created_at=datetime.now(),
)
mock_sandbox.create.return_value = mock_sandbox.context
mock_factory.create_sandbox.return_value = mock_sandbox
return SandboxManager(factory=mock_factory, cleanup_on_exit=False)
class HandlerResolverSuite:
"""Benchmark handler resolver import and cache performance."""
def time_resolve_git_handler_cold(self) -> None:
"""Benchmark cold resolve (no cache)."""
clear_handler_cache()
resolve_handler(
"cleveragents.resource.handlers.git_checkout:GitCheckoutHandler"
)
def time_resolve_git_handler_cached(self) -> None:
"""Benchmark cached resolve."""
resolve_handler(
"cleveragents.resource.handlers.git_checkout:GitCheckoutHandler"
)
def time_resolve_fs_handler_cold(self) -> None:
"""Benchmark cold resolve for fs-directory."""
clear_handler_cache()
resolve_handler(
"cleveragents.resource.handlers.fs_directory:FsDirectoryHandler"
)
class GitCheckoutHandlerSuite:
"""Benchmark GitCheckoutHandler.resolve() overhead."""
def setup(self) -> None:
self.resource = _make_resource("git-checkout", "/tmp/bench-repo")
self.manager = _make_mock_manager()
self.handler = GitCheckoutHandler()
def time_resolve_git_checkout(self) -> None:
"""Benchmark single git-checkout resolution."""
self.handler.resolve(
resource=self.resource,
plan_id="PLAN_BENCH",
slot_name="repo",
sandbox_manager=self.manager,
)
class FsDirectoryHandlerSuite:
"""Benchmark FsDirectoryHandler.resolve() overhead."""
def setup(self) -> None:
self.resource = _make_resource("fs-directory", "/tmp/bench-dir")
self.manager = _make_mock_manager()
self.handler = FsDirectoryHandler()
def time_resolve_fs_directory(self) -> None:
"""Benchmark single fs-directory resolution."""
self.handler.resolve(
resource=self.resource,
plan_id="PLAN_BENCH",
slot_name="workdir",
sandbox_manager=self.manager,
)
class ResourceHandlerServiceSuite:
"""Benchmark ResourceHandlerService end-to-end resolution."""
def setup(self) -> None:
from cleveragents.application.services.resource_handler_service import (
ResourceHandlerService,
)
self.resource = _make_resource("git-checkout", "/tmp/bench-repo")
self.type_spec = ResourceTypeSpec(
name="git-checkout",
description="Test git-checkout type",
resource_kind=ResourceKind.PHYSICAL,
sandbox_strategy=TypeSandboxStrategy.GIT_WORKTREE,
user_addable=True,
built_in=True,
handler="cleveragents.resource.handlers.git_checkout:GitCheckoutHandler",
)
self.manager = _make_mock_manager()
self.service = ResourceHandlerService(
sandbox_manager=self.manager,
resource_lookup=lambda _: self.resource,
type_lookup=lambda _: self.type_spec,
)
self.binding = BindingResult(
slot_name="repo",
resource_id="01KJ5C5TPMP8GGX3QC83E2MAQS",
resource_name="test-repo",
binding_mode="contextual",
)
def time_resolve_binding(self) -> None:
"""Benchmark single binding resolution."""
self.service.resolve_binding(
binding=self.binding,
plan_id="PLAN_BENCH",
)
def time_resolve_bindings_batch(self) -> None:
"""Benchmark batch binding resolution (5 bindings)."""
bindings = [
BindingResult(
slot_name=f"slot-{i}",
resource_id="01KJ5C5TPMP8GGX3QC83E2MAQS",
resource_name=f"res-{i}",
binding_mode="contextual",
)
for i in range(5)
]
self.service.resolve_bindings(
bindings=bindings,
plan_id="PLAN_BENCH",
)
benchmarks/resource_registry_bench.py
+188
View File
@@ -0,0 +1,188 @@
"""ASV benchmarks for resource registry lookup operations.
Measures the performance of:
- ResourceTypeSpec construction from config dict
- ResourceRegistryService.show_type() lookup
- ResourceRegistryService.show_resource() lookup by name and ULID
- ResourceRegistryService.list_types() enumeration
- ResourceRegistryService.list_resources() enumeration
- ResourceRegistryService.register_resource() creation
"""
from __future__ import annotations
import sys
from pathlib import Path
from typing import Any
try:
from cleveragents.application.services.resource_registry_service import (
ResourceRegistryService,
)
from cleveragents.domain.models.core.resource_type import (
ResourceTypeSpec,
)
except ModuleNotFoundError:
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
from cleveragents.application.services.resource_registry_service import (
ResourceRegistryService,
)
from cleveragents.domain.models.core.resource_type import (
ResourceTypeSpec,
)
def _setup_db() -> Any:
"""Create in-memory database and return session factory."""
from sqlalchemy import create_engine, event
from sqlalchemy.orm import sessionmaker
from sqlalchemy.pool import StaticPool
from cleveragents.infrastructure.database.models import Base
engine = create_engine(
"sqlite:///:memory:",
connect_args={"check_same_thread": False},
poolclass=StaticPool,
)
@event.listens_for(engine, "connect")
def _fk(conn: Any, _rec: Any) -> None:
conn.cursor().execute("PRAGMA foreign_keys=ON")
Base.metadata.create_all(engine)
return sessionmaker(bind=engine)
class TypeSpecConstructionSuite:
"""Benchmark ResourceTypeSpec construction from config dicts."""
def time_from_config_minimal(self) -> None:
"""Benchmark minimal type spec construction."""
ResourceTypeSpec.from_config(
{
"name": "bench/minimal",
"description": "Minimal benchmark type",
"resource_kind": "physical",
"sandbox_strategy": "copy_on_write",
}
)
def time_from_config_full(self) -> None:
"""Benchmark full type spec construction with all fields."""
ResourceTypeSpec.from_config(
{
"name": "bench/full",
"description": "Full benchmark type",
"resource_kind": "physical",
"sandbox_strategy": "git_worktree",
"user_addable": True,
"built_in": False,
"cli_args": [
{
"name": "path",
"type": "path",
"required": True,
"description": "Path to resource",
},
{
"name": "branch",
"type": "string",
"required": False,
"description": "Branch name",
"default": "main",
},
],
"parent_types": ["git-checkout"],
"child_types": ["fs-directory", "fs-file"],
"handler": "bench.handler:BenchHandler",
"capabilities": {
"read": True,
"write": True,
"sandbox": True,
"checkpoint": False,
},
"auto_discovery": {
"enabled": True,
"rules": [
{"type": "fs-directory", "pattern": "*/"},
],
},
}
)
class RegistryLookupSuite:
"""Benchmark registry service lookup operations."""
def setup(self) -> None:
self._sf = _setup_db()
self._svc = ResourceRegistryService(session_factory=self._sf)
self._svc.bootstrap_builtin_types()
# Register some resources for lookup benchmarks
self._resources = []
for i in range(20):
r = self._svc.register_resource(
type_name="git-checkout",
name=f"bench/repo-{i}",
location=f"/tmp/bench/repo-{i}",
description=f"Benchmark repo {i}",
)
self._resources.append(r)
def time_show_type_builtin(self) -> None:
"""Benchmark looking up a built-in type by name."""
self._svc.show_type("git-checkout")
def time_show_type_second_builtin(self) -> None:
"""Benchmark looking up the second built-in type."""
self._svc.show_type("fs-directory")
def time_list_types(self) -> None:
"""Benchmark listing all registered types."""
self._svc.list_types()
def time_list_types_filtered(self) -> None:
"""Benchmark listing types filtered by namespace."""
self._svc.list_types(namespace="builtin")
def time_show_resource_by_name(self) -> None:
"""Benchmark looking up a resource by namespaced name."""
self._svc.show_resource("bench/repo-0")
def time_show_resource_by_ulid(self) -> None:
"""Benchmark looking up a resource by ULID."""
self._svc.show_resource(self._resources[0].resource_id)
def time_list_resources_all(self) -> None:
"""Benchmark listing all resources."""
self._svc.list_resources()
def time_list_resources_filtered(self) -> None:
"""Benchmark listing resources filtered by type."""
self._svc.list_resources(type_name="git-checkout")
class RegistryRegistrationSuite:
"""Benchmark resource registration throughput."""
_reg_ctr: int = 0
def setup(self) -> None:
self._sf = _setup_db()
self._svc = ResourceRegistryService(session_factory=self._sf)
self._svc.bootstrap_builtin_types()
def time_register_resource(self) -> None:
"""Benchmark registering a single resource."""
RegistryRegistrationSuite._reg_ctr += 1
self._svc.register_resource(
type_name="git-checkout",
name=f"bench/reg-{RegistryRegistrationSuite._reg_ctr}",
location=f"/tmp/bench/reg-{RegistryRegistrationSuite._reg_ctr}",
)
def time_bootstrap_builtin_types_idempotent(self) -> None:
"""Benchmark idempotent bootstrap (types already exist)."""
self._svc.bootstrap_builtin_types()
docs/reference/decision_model.md
+122
View File
@@ -0,0 +1,122 @@
# Decision Domain Model
The decision subsystem records every choice point in a plan's lifecycle
as a **Decision** node in a persistent tree. Decisions are created
during the Strategize and Execute phases and form the basis for
targeted correction and replay.
## Decision Types
| Type | Phase | Description |
|--------------------------|------------|------------------------------------------|
| `prompt_definition` | Strategize | Root decision — the plan prompt |
| `invariant_enforced` | Strategize | An invariant constraint was applied |
| `strategy_choice` | Strategize | High-level approach chosen |
| `implementation_choice` | Execute | How to implement a specific task |
| `resource_selection` | Execute | Which resources to read / modify |
| `subplan_spawn` | Strategize | Decision to create a child plan |
| `subplan_parallel_spawn` | Strategize | Spawn a group of child plans in parallel |
| `tool_invocation` | Execute | Which skill / tool to use |
| `error_recovery` | Execute | How to handle a failure |
| `validation_response` | Execute | Response to a validation failure |
| `user_intervention` | Any | User-provided guidance / correction |
## Decision Fields
### Identity
- **decision_id** — ULID, auto-generated
- **plan_id** — ULID of the parent plan (required)
- **parent_decision_id** — ULID of the parent decision, or `None` for the root
- **sequence_number** — monotonic order within the plan (0-indexed, never reused)
### Classification
- **decision_type** — one of the 11 `DecisionType` enum values
### Content
- **question** — what question was being answered (required, non-empty)
- **chosen_option** — the option that was selected (required, non-empty)
- **alternatives_considered** — list of other options evaluated
- **confidence_score** — float 0.0–1.0, or `None` if not applicable
### Context Snapshot
Every decision captures a `ContextSnapshot` for replay:
- **hot_context_hash** — cryptographic hash of the context window
- **hot_context_ref** — storage pointer to the full serialised context
- **relevant_resources** — list of `ResourceRef` entries (resource_id + optional path)
- **actor_state_ref** — LangGraph actor checkpoint reference
### Rationale
- **rationale** — human-readable explanation
- **actor_reasoning** — raw LLM reasoning trace, if available
### Downstream Impact
- **downstream_decision_ids** — ULIDs of decisions that depend on this one
- **downstream_plan_ids** — ULIDs of child plans spawned from this decision
- **artifacts_produced** — list of `ArtifactRef` (artifact_path + artifact_type)
### Timestamps
- **created_at** — UTC datetime, auto-set on creation
### Correction Metadata
- **is_correction** — boolean, True if this decision corrects another
- **corrects_decision_id** — ULID of the original decision
- **correction_reason** — why the correction was made
- **superseded_by** — ULID of the decision that replaced this one
## Tree Structure
Decisions form a tree via `parent_decision_id`:
```
prompt_definition (root, parent=None)
├── invariant_enforced
├── strategy_choice
│ ├── implementation_choice
│ │ ├── resource_selection
│ │ └── tool_invocation
│ └── subplan_spawn
└── strategy_choice
```
The `prompt_definition` type is always the root and must have
`parent_decision_id = None`. This is enforced by a model validator.
## Correction Lifecycle
Corrections never mutate existing decisions. Instead:
1. A new `Decision` is created with `is_correction=True` and
`corrects_decision_id` pointing to the original.
2. The original decision has its `superseded_by` field set to the
new decision's ID.
3. All downstream decisions of the original are also superseded.
The **current tree** consists of all decisions where
`superseded_by IS NULL`.
## Validation Rules
- `plan_id` must be a valid ULID
- `parent_decision_id`, `corrects_decision_id`, `superseded_by` must
be valid ULIDs or None
- `confidence_score` must be in [0.0, 1.0] or None
- `prompt_definition` decisions must have `parent_decision_id = None`
- If `is_correction` is True, `corrects_decision_id` must be set
- If `corrects_decision_id` is set, `is_correction` must be True
## Source
- Domain model: `src/cleveragents/domain/models/core/decision.py`
- Specification: `docs/specification.md` L18390–L18521
- ADR-007: Decision tree and correction
- ADR-033: Decision recording protocol
- ADR-034: Decision tree versioning and history
docs/reference/resource_handlers.md
+172
View File
@@ -0,0 +1,172 @@
# Resource Handlers
Resource handlers bridge resource types to sandbox provisioning. When a plan
executes, each tool's resource slot is resolved to a physical sandbox path
through the handler pipeline.
## Architecture
```
BindingResolutionService ResourceHandlerService
resolve(tool, project) resolve_binding(binding, plan_id)
│ │
▼ ▼
BindingResult(resource_id) ┌── Resource (location)
│ ResourceTypeSpec (handler, strategy)
│ │
│ ▼
│ resolve_handler("module:Class")
│ │
│ ▼
│ handler.resolve(resource, sandbox_manager)
│ │
│ ▼
└── BoundResource(sandbox_path="/tmp/sandbox/...")
```
## Handler Protocol
All handlers implement the `ResourceHandler` protocol:
```python
class ResourceHandler(Protocol):
def resolve(
self,
*,
resource: Resource,
plan_id: str,
slot_name: str,
sandbox_manager: SandboxManager,
access: str = "read_only",
) -> BoundResource: ...
```
## Built-in Handlers
### GitCheckoutHandler
Resolves `git-checkout` resources using the `git_worktree` sandbox strategy.
| Property | Value |
|----------|-------|
| Module | `cleveragents.resource.handlers.git_checkout` |
| Class | `GitCheckoutHandler` |
| Default strategy | `git_worktree` |
| Fallback strategy | `copy_on_write` |
| Required fields | `resource.location` (path to git repo root) |
### FsDirectoryHandler
Resolves `fs-directory` resources using the `copy_on_write` sandbox strategy.
| Property | Value |
|----------|-------|
| Module | `cleveragents.resource.handlers.fs_directory` |
| Class | `FsDirectoryHandler` |
| Default strategy | `copy_on_write` |
| Required fields | `resource.location` (path to directory) |
## Handler Resolution
Handler strings use the `module.path:ClassName` format and are stored on
`ResourceTypeSpec.handler`. Resolution is dynamic via `importlib`:
```python
from cleveragents.resource.handlers import resolve_handler
handler = resolve_handler(
"cleveragents.resource.handlers.git_checkout:GitCheckoutHandler"
)
```
Resolved handlers are cached for the process lifetime. Call
`clear_handler_cache()` to reset.
### Fallback Behavior
If no handler string is set (or resolution fails), the
`ResourceHandlerService` uses a default handler that delegates directly
to `SandboxManager` using the type's `sandbox_strategy`.
## Strategy Precedence
The sandbox strategy is determined in this order:
1. **Resource-level override** (`resource.sandbox_strategy`) — per-resource
2. **Type default** (`ResourceTypeSpec.sandbox_strategy`) — per-type
3. **Fallback** (`none`) — no sandboxing
## ResourceHandlerService
The orchestration service that chains the resolution pipeline:
```python
from cleveragents.application.services.resource_handler_service import (
ResourceHandlerService,
)
service = ResourceHandlerService(
sandbox_manager=sandbox_manager,
resource_lookup=registry_service.show_resource,
type_lookup=registry_service.show_type,
)
# Resolve a single binding
bound = service.resolve_binding(binding, plan_id="01ARZ3...")
# Resolve all bindings for a tool
bindings_map = service.resolve_bindings(bindings, plan_id="01ARZ3...")
# -> {"repo": BoundResource(sandbox_path="/tmp/sandbox/...")}
```
## Sandbox Outputs
After resolution, each `BoundResource` carries:
| Field | Description |
|-------|-------------|
| `slot_name` | Tool slot this binding fills |
| `resource_id` | ULID of the resolved resource |
| `resource_type` | Type name (e.g. `git-checkout`) |
| `sandbox_path` | Root path of the provisioned sandbox |
| `access` | `read_only` or `read_write` |
The `sandbox_path` points to an isolated directory managed by
`SandboxManager`. Changes are committed or rolled back via
`SandboxManager.commit_all()` / `rollback_all()`.
## Writing Custom Handlers
To add a handler for a new resource type:
1. Create a module under `cleveragents/resource/handlers/`.
2. Implement a class satisfying `ResourceHandler`.
3. Register it on the `ResourceTypeSpec` via the `handler` field:
`"cleveragents.resource.handlers.my_handler:MyHandler"`.
```python
class MyHandler:
def resolve(
self,
*,
resource: Resource,
plan_id: str,
slot_name: str,
sandbox_manager: SandboxManager,
access: str = "read_only",
) -> BoundResource:
# Custom validation / setup
sandbox = sandbox_manager.get_or_create_sandbox(
plan_id=plan_id,
resource_id=resource.resource_id,
original_path=resource.location,
sandbox_strategy="copy_on_write",
)
return BoundResource(
slot_name=slot_name,
resource_id=resource.resource_id,
resource_type=resource.resource_type_name,
sandbox_path=sandbox.context.sandbox_path if sandbox.context else "",
access=access,
)
```
docs/reference/resources.md
+211
View File
@@ -0,0 +1,211 @@
# Resource Registry
The Resource Registry manages **resource types** and **resource
instances** in CleverAgents. Resource types define schemas, sandbox
strategies, and handler implementations. Resource instances are
registered entries that plans operate on during execution.
## Built-in Resource Types
CleverAgents ships with two built-in types, available without
registration:
| Type | Kind | Sandbox Strategy | Handler |
|------------------|----------|------------------|--------------------------------|
| `git-checkout` | physical | `git_worktree` | `GitCheckoutHandler` |
| `fs-directory` | physical | `copy_on_write` | `FsDirectoryHandler` |
Built-in types are registered idempotently at startup via
`ResourceRegistryService.bootstrap_builtin_types()`.
## Resource Type Fields
Each `ResourceTypeSpec` defines a resource type schema:
| Field | Type | Description |
|--------------------|---------------------|------------------------------------------------------|
| `name` | `str` | Unique name (built-in: bare, custom: `namespace/name`) |
| `description` | `str` | Human-readable description |
| `resource_kind` | `ResourceKind` | `physical` or `virtual` |
| `sandbox_strategy` | `SandboxStrategy` | Default sandbox isolation strategy |
| `user_addable` | `bool` | Whether users can register instances of this type |
| `built_in` | `bool` | Whether this is a built-in type |
| `cli_args` | `list[ResourceTypeArgument]` | CLI argument definitions for registration |
| `parent_types` | `list[str]` | Allowed parent type names in the DAG |
| `child_types` | `list[str]` | Allowed child type names in the DAG |
| `handler` | `str \| None` | Handler ref in `module:Class` format |
| `capabilities` | `dict[str, bool]` | `read`, `write`, `sandbox`, `checkpoint` flags |
| `auto_discovery` | `dict \| None` | Auto-discovery rules (see below) |
| `equivalence` | `dict \| None` | Virtual resource deduplication config |
### Custom Types
Custom types are defined in YAML and registered via CLI or API:
```yaml
name: myteam/python-package
description: A Python package directory
resource_kind: physical
sandbox_strategy: copy_on_write
user_addable: true
cli_args:
- name: path
type: path
required: true
description: Path to the package root
parent_types:
- git-checkout
child_types:
- fs-directory
```
Register with: `agents resource type add path/to/type.yaml`
## Resource Instance Fields
Each `Resource` represents a registered asset:
| Field | Type | Description |
|----------------------|-----------------------|----------------------------------------------|
| `resource_id` | `str` (ULID) | Unique identifier |
| `name` | `str \| None` | Optional namespaced name |
| `resource_type_name` | `str` | Type this resource belongs to |
| `classification` | `PhysVirt` | `physical` or `virtual` |
| `description` | `str \| None` | Human-readable description |
| `location` | `str \| None` | Filesystem path (physical resources) |
| `sandbox_strategy` | `SandboxStrategy \| None` | Per-resource sandbox override |
| `content_hash` | `str \| None` | Hash of resource contents for change detection |
| `parents` | `list[str]` | Parent resource IDs in the DAG |
| `children` | `list[str]` | Child resource IDs in the DAG |
| `properties` | `dict` | Type-specific key-value properties |
| `capabilities` | `ResourceCapabilities`| `readable`, `writable`, `sandboxable`, `checkpointable` |
| `created_at` | `datetime` | Registration timestamp |
| `updated_at` | `datetime` | Last modification timestamp |
## Registry Service API
### Type Operations
| Method | Description |
|-------------------------|------------------------------------------|
| `bootstrap_builtin_types()` | Register built-in types (idempotent) |
| `register_type(config_path)` | Register a custom type from YAML |
| `list_types(namespace=)` | List types, optionally filtered |
| `show_type(name)` | Get a type by name |
### Resource Operations
| Method | Description |
|-------------------------------|--------------------------------------|
| `register_resource(type_name, name=, location=, ...)` | Create and register an instance |
| `list_resources(type_name=)` | List resources, optionally by type |
| `show_resource(name_or_id)` | Get by namespaced name or ULID |
### DAG Operations
| Method | Description |
|---------------------------------|------------------------------------|
| `link_child(parent, child)` | Link a child to a parent |
| `unlink_child(parent, child)` | Remove a parent-child link |
| `get_children(name_or_id)` | Get direct children |
| `get_resource_tree(name_or_id, depth=, type_filter=)` | Recursive tree traversal |
For DAG rules, cycle detection, type compatibility, and auto-discovery
details, see [Resource DAG](resource_dag.md).
## Discovery Behavior
### Auto-Discovery
Resource types can define auto-discovery rules that materialize child
resources automatically:
```yaml
auto_discovery:
enabled: true
rules:
- type: fs-directory
pattern: "*/"
- type: fs-file
pattern: "*"
```
When `auto_discover_children(resource_id)` is called:
1. The resource and its type are looked up.
2. Each discovery rule is evaluated:
- The child type must exist in the registry.
- The child type must be compatible with the parent type (per
`child_types`).
3. New child resources are created with `auto_discovered = true`.
4. Links are created in the `resource_links` table.
Auto-discovery is triggered:
- When a resource is first registered
- When resource contents change
- On demand via CLI (`agents resource discover`)
### Local-Mode Discovery
In local mode (the default), discovery operates against the local
filesystem. The registry creates placeholder resource entries based
on discovery rules; actual file enumeration is delegated to resource
handlers at sandbox creation time.
## Database Schema
### `resource_types` Table
| Column | Type | Description |
|---------------------------|---------------|--------------------------------|
| `name` | `String(128)` | Primary key |
| `namespace` | `String(64)` | Namespace (or `builtin`) |
| `description` | `Text` | Human-readable description |
| `resource_kind` | `String(16)` | `physical` or `virtual` |
| `sandbox_strategy` | `String(32)` | Default sandbox strategy |
| `user_addable` | `Boolean` | User registration allowed |
| `built_in` | `Boolean` | Built-in type flag |
| `handler` | `String(256)` | Handler `module:Class` ref |
| `args_schema_json` | `Text` | CLI args JSON |
| `allowed_parent_types_json` | `Text` | Allowed parent types JSON |
| `allowed_child_types_json` | `Text` | Allowed child types JSON |
| `auto_discover_json` | `Text` | Auto-discovery config JSON |
| `capabilities_json` | `Text` | Capabilities JSON |
| `equivalence_json` | `Text` | Equivalence config JSON |
| `source` | `String(256)` | Registration source |
| `created_at` | `String(30)` | ISO-8601 timestamp |
| `updated_at` | `String(30)` | ISO-8601 timestamp |
### `resources` Table
| Column | Type | Description |
|--------------------|---------------|-------------------------------------|
| `resource_id` | `String(26)` | ULID primary key |
| `namespaced_name` | `String(256)` | Optional namespaced name (unique) |
| `namespace` | `String(64)` | Namespace extracted from name |
| `type_name` | `String(128)` | FK to `resource_types.name` |
| `resource_kind` | `String(16)` | `physical` or `virtual` |
| `location` | `Text` | Filesystem path |
| `description` | `Text` | Human-readable description |
| `read_only` | `Boolean` | Read-only flag |
| `auto_discovered` | `Boolean` | Created by auto-discovery |
| `sandbox_strategy` | `String(32)` | Per-resource sandbox override |
| `content_hash` | `String(128)` | Content hash for change detection |
| `properties_json` | `Text` | Type-specific properties JSON |
| `metadata_json` | `Text` | Additional metadata JSON |
| `created_at` | `String(30)` | ISO-8601 timestamp |
| `updated_at` | `String(30)` | ISO-8601 timestamp |
### `resource_links` Table
See [Resource DAG](resource_dag.md#database-schema) for link table
schema.
## Source
- Domain models: `src/cleveragents/domain/models/core/resource.py`,
`resource_type.py`
- Service: `src/cleveragents/application/services/resource_registry_service.py`
- Migration: `alembic/versions/b1_001_resource_registry_tables.py`
- CLI: `src/cleveragents/cli/commands/resource.py`
- Specification: `docs/specification.md` — Resource Registry sections
docs/specification.md
+3 -3
View File
@@ -45250,11 +45250,11 @@ This section provides a high-level overview of the CleverAgents implementation r
### Current Status Summary
As of Day 14 (2026-02-22), the **Day 14 rebaseline plan** is authoritative and **88 of 182 COMMIT checklist items are complete** (~48.35%). The foundational layers — domain models, persistence, YAML schemas, CLI commands, quality automation, and CI/CD — are substantially built out. The primary blocker for the first end-to-end milestone remains the **tool-aware execute/apply pipeline**: the actor runtime that drives LLM-based plan execution, ChangeSet capture, and the sandbox merge path that commits changes to real resources with validation gating.
As of Day 15 (2026-02-23), the **Day 15 rebaseline plan** is authoritative and **1 of 53 rebaseline COMMIT checklist items are complete** (~1.89%). Foundational layers remain substantially built out (domain models, persistence, YAML schemas, CLI commands, quality automation, CI/CD, tool runtime, and diff review artifacts). The primary blocker for the first end-to-end milestone is now the **sandbox merge/apply path + resource handler bindings**: resource handlers must resolve git-checkout/fs-directory roots and the apply pipeline must merge sandbox changes into real resources with validation gating.
!!! warning "Critical Priority: v2 Feature Restoration"
!!! warning "Critical Priority: M1 Merge Path + Resource Handlers"
During the transition from the earlier v2 architecture to the current v3 spec-aligned design, approximately 4,000 lines of LangGraph agent infrastructure were stubbed or removed. **Restoring these v2 capabilities** — specifically the LangGraph-based execution actors, tool-calling runtime, and actor graph compilation — is the highest-priority blocker for reaching a minimal working version. Until the actor runtime can invoke tools through the LLM, persist decisions, and capture changes into a ChangeSet, the plan execute/apply lifecycle cannot function end-to-end.
The actor runtime and tool-aware execute flow are in place, but **apply is still blocked** by missing sandbox merge logic, validation gating, and resource handler bindings. Until git-checkout/fs-directory handlers can resolve sandbox roots and the apply pipeline can merge ChangeSets into real project resources safely, the plan lifecycle cannot complete end-to-end.
### Parallel Workstreams
features/decision_model.feature
+246
View File
@@ -0,0 +1,246 @@
Feature: Decision domain model
Validates the Decision, DecisionType, ContextSnapshot, ResourceRef,
and ArtifactRef domain models including ULID validation, enum coverage,
tree structure helpers, correction constraints, and serialization.
# ------------------------------------------------------------------
# DecisionType enum
# ------------------------------------------------------------------
Scenario: All 11 decision types are defined
Then the DecisionType enum should have exactly 11 members
Scenario: Strategize-phase types are correctly classified
Then STRATEGIZE_TYPES should contain "prompt_definition"
And STRATEGIZE_TYPES should contain "invariant_enforced"
And STRATEGIZE_TYPES should contain "strategy_choice"
And STRATEGIZE_TYPES should contain "subplan_spawn"
And STRATEGIZE_TYPES should contain "subplan_parallel_spawn"
And STRATEGIZE_TYPES should have exactly 5 members
Scenario: Execute-phase types are correctly classified
Then EXECUTE_TYPES should contain "implementation_choice"
And EXECUTE_TYPES should contain "resource_selection"
And EXECUTE_TYPES should contain "tool_invocation"
And EXECUTE_TYPES should contain "error_recovery"
And EXECUTE_TYPES should contain "validation_response"
And EXECUTE_TYPES should have exactly 5 members
Scenario: user_intervention is not in either phase set
Then STRATEGIZE_TYPES should not contain "user_intervention"
And EXECUTE_TYPES should not contain "user_intervention"
# ------------------------------------------------------------------
# Decision creation and ULID validation
# ------------------------------------------------------------------
Scenario: Create a minimal root decision
Given a valid plan ULID
When I create a prompt_definition decision with sequence 0
Then the decision should be created successfully
And the decision should be a root decision
And the decision should have a valid ULID as decision_id
And the decision type should be "prompt_definition"
And the decision is_strategize_type should be true
Scenario: Create a non-root decision with parent
Given a valid plan ULID
And a valid parent decision ULID
When I create a strategy_choice decision with sequence 1 and a parent
Then the decision should be created successfully
And the decision should not be a root decision
And the decision is_strategize_type should be true
Scenario: Create an execute-phase decision
Given a valid plan ULID
And a valid parent decision ULID
When I create an implementation_choice decision with sequence 2 and a parent
Then the decision should be created successfully
And the decision is_execute_type should be true
And the decision is_strategize_type should be false
Scenario: user_intervention is valid in any phase
Given a valid plan ULID
When I create a user_intervention decision with sequence 3
Then the decision should be created successfully
And the decision is_any_phase_type should be true
Scenario: Invalid plan_id is rejected
When I try to create a decision with plan_id "not-a-ulid"
Then a decision validation error should be raised
And the decision error should mention "plan_id"
Scenario: Invalid parent_decision_id is rejected
Given a valid plan ULID
When I try to create a decision with parent_decision_id "bad-id"
Then a decision validation error should be raised
And the decision error should mention "parent_decision_id"
# ------------------------------------------------------------------
# prompt_definition root constraint
# ------------------------------------------------------------------
Scenario: prompt_definition with a parent is rejected
Given a valid plan ULID
And a valid parent decision ULID
When I try to create a prompt_definition with a parent
Then a decision validation error should be raised
And the decision error should mention "root"
# ------------------------------------------------------------------
# Confidence score validation
# ------------------------------------------------------------------
Scenario: Confidence score within valid range
Given a valid plan ULID
When I create a decision with confidence score 0.85
Then the decision should be created successfully
And the decision confidence score should be 0.85
Scenario: Confidence score of 0.0 is valid
Given a valid plan ULID
When I create a decision with confidence score 0.0
Then the decision should be created successfully
Scenario: Confidence score of 1.0 is valid
Given a valid plan ULID
When I create a decision with confidence score 1.0
Then the decision should be created successfully
Scenario: Confidence score above 1.0 is rejected
Given a valid plan ULID
When I try to create a decision with confidence score 1.5
Then a decision validation error should be raised
Scenario: Confidence score below 0.0 is rejected
Given a valid plan ULID
When I try to create a decision with confidence score -0.1
Then a decision validation error should be raised
Scenario: Confidence score of None is valid
Given a valid plan ULID
When I create a decision with confidence score None
Then the decision should be created successfully
And the decision confidence score should be None
# ------------------------------------------------------------------
# Correction metadata
# ------------------------------------------------------------------
Scenario: Create a correction decision
Given a valid plan ULID
And a valid corrected decision ULID
When I create a correction decision
Then the decision should be created successfully
And the decision is_correction should be true
Scenario: is_correction without corrects_decision_id is rejected
Given a valid plan ULID
When I try to create a decision with is_correction true but no corrects_decision_id
Then a decision validation error should be raised
And the decision error should mention "corrects_decision_id"
Scenario: corrects_decision_id without is_correction is rejected
Given a valid plan ULID
And a valid corrected decision ULID
When I try to create a decision with corrects_decision_id but is_correction false
Then a decision validation error should be raised
And the decision error should mention "is_correction"
Scenario: superseded_by marks decision as superseded
Given a valid plan ULID
When I create a decision that is superseded
Then the decision is_superseded should be true
Scenario: Invalid superseded_by ULID is rejected
Given a valid plan ULID
When I try to create a decision with superseded_by "invalid"
Then a decision validation error should be raised
And the decision error should mention "superseded_by"
Scenario: Invalid corrects_decision_id ULID is rejected
Given a valid plan ULID
When I try to create a decision with corrects_decision_id ULID "not-valid"
Then a decision validation error should be raised
And the decision error should mention "corrects_decision_id"
Scenario: with_superseded_by returns a new superseded copy
Given a valid plan ULID
When I create a prompt_definition decision with sequence 0
And I call with_superseded_by on the decision
Then the original decision should not be superseded
And the superseded copy should be superseded
# ------------------------------------------------------------------
# ContextSnapshot
# ------------------------------------------------------------------
Scenario: Create a decision with a populated context snapshot
Given a valid plan ULID
And a context snapshot with hash and resources
When I create a decision with the context snapshot
Then the decision should be created successfully
And the decision context snapshot hash should not be empty
And the decision context snapshot should have resources
Scenario: Default context snapshot is empty
Given a valid plan ULID
When I create a prompt_definition decision with sequence 0
Then the decision context snapshot hash should be empty
# ------------------------------------------------------------------
# ResourceRef and ArtifactRef
# ------------------------------------------------------------------
Scenario: ResourceRef requires non-empty resource_id
When I try to create a ResourceRef with empty resource_id
Then a decision validation error should be raised
Scenario: ArtifactRef requires non-empty artifact_path
When I try to create an ArtifactRef with empty artifact_path
Then a decision validation error should be raised
Scenario: Decision with artifacts_produced
Given a valid plan ULID
When I create a decision with artifacts
Then the decision should have 2 artifacts produced
# ------------------------------------------------------------------
# Serialization
# ------------------------------------------------------------------
Scenario: Decision round-trips through dict serialization
Given a valid plan ULID
When I create a fully populated decision
Then the decision should round-trip through model_dump and model_validate
Scenario: as_cli_dict returns expected keys
Given a valid plan ULID
When I create a prompt_definition decision with sequence 0
Then as_cli_dict should contain key "decision_id"
And as_cli_dict should contain key "type"
And as_cli_dict should contain key "confidence"
And as_cli_dict should contain key "parent"
# ------------------------------------------------------------------
# All 11 decision types can be instantiated
# ------------------------------------------------------------------
Scenario Outline: Each decision type can be instantiated
Given a valid plan ULID
When I create a decision of type "<dtype>"
Then the decision should be created successfully
Examples:
| dtype |
| prompt_definition |
| invariant_enforced |
| strategy_choice |
| implementation_choice |
| resource_selection |
| subplan_spawn |
| subplan_parallel_spawn |
| tool_invocation |
| error_recovery |
| validation_response |
| user_intervention |
features/resource_handlers.feature
+124
View File
@@ -0,0 +1,124 @@
Feature: Resource handler runtime
As a developer
I want resource handlers that resolve resources into sandbox-backed paths
So that tools receive isolated working directories during plan execution
# === Handler Protocol ===
Scenario: GitCheckoutHandler satisfies ResourceHandler protocol
Then GitCheckoutHandler should satisfy the ResourceHandler protocol
Scenario: FsDirectoryHandler satisfies ResourceHandler protocol
Then FsDirectoryHandler should satisfy the ResourceHandler protocol
# === GitCheckoutHandler ===
Scenario: GitCheckoutHandler resolves a git-checkout resource
Given a git-checkout resource with location "/tmp/test-repo"
And a sandbox manager with a mock factory
When I resolve the resource with GitCheckoutHandler for plan "PLAN001"
Then the bound resource should have slot name "repo"
And the bound resource should have resource type "git-checkout"
And the bound resource sandbox path should not be empty
Scenario: GitCheckoutHandler uses resource-level strategy override
Given a git-checkout resource at "/tmp/test-repo" using strategy "copy_on_write"
And a sandbox manager with a mock factory
When I resolve the resource with GitCheckoutHandler for plan "PLAN002"
Then the sandbox was created with strategy "copy_on_write"
Scenario: GitCheckoutHandler rejects resource without location
Given a git-checkout resource without location
And a sandbox manager with a mock factory
When I try to resolve the resource with GitCheckoutHandler
Then a handler ValueError should be raised
And the handler error should mention "no location"
# === FsDirectoryHandler ===
Scenario: FsDirectoryHandler resolves an fs-directory resource
Given an fs-directory resource with location "/tmp/test-dir"
And a sandbox manager with a mock factory
When I resolve the resource with FsDirectoryHandler for plan "PLAN003"
Then the bound resource should have slot name "workdir"
And the bound resource should have resource type "fs-directory"
And the bound resource sandbox path should not be empty
Scenario: FsDirectoryHandler uses copy_on_write by default
Given an fs-directory resource with location "/tmp/test-dir"
And a sandbox manager with a mock factory
When I resolve the resource with FsDirectoryHandler for plan "PLAN004"
Then the sandbox was created with strategy "copy_on_write"
Scenario: FsDirectoryHandler rejects resource without location
Given an fs-directory resource without location
And a sandbox manager with a mock factory
When I try to resolve the resource with FsDirectoryHandler
Then a handler ValueError should be raised
# === Handler Resolver ===
Scenario: Resolve handler from valid module:class string
When I resolve handler "cleveragents.resource.handlers.git_checkout:GitCheckoutHandler"
Then the resolved handler should be a GitCheckoutHandler instance
Scenario: Resolve handler from fs-directory reference
When I resolve handler "cleveragents.resource.handlers.fs_directory:FsDirectoryHandler"
Then the resolved handler should be a FsDirectoryHandler instance
Scenario: Resolve handler caches instances
When I resolve handler "cleveragents.resource.handlers.git_checkout:GitCheckoutHandler"
And I resolve the same handler reference again
Then both resolved handlers should be the same object
Scenario: Resolve handler rejects empty string
When I try to resolve an empty handler reference
Then a HandlerResolutionError should be raised
And the resolution error should mention "empty"
Scenario: Resolve handler rejects missing colon separator
When I try to resolve handler "cleveragents.resource.handlers.git_checkout"
Then a HandlerResolutionError should be raised
And the resolution error should mention "module.path:ClassName"
Scenario: Resolve handler rejects nonexistent module
When I try to resolve handler "nonexistent.module:SomeClass"
Then a HandlerResolutionError should be raised
And the resolution error should mention "Cannot import"
Scenario: Resolve handler rejects nonexistent class in valid module
When I try to resolve handler "cleveragents.resource.handlers.git_checkout:NonExistentClass"
Then a HandlerResolutionError should be raised
And the resolution error should mention "not found"
# === ResourceHandlerService ===
Scenario: ResourceHandlerService resolves a binding to BoundResource
Given a resource handler service with mock lookups
And a binding result for slot "repo" with resource id "RES001"
When I resolve the binding via resource handler service for plan "PLAN010"
Then the handler service should return a BoundResource
And the handler service BoundResource slot should be "repo"
And the handler service BoundResource sandbox path should not be empty
Scenario: ResourceHandlerService skips deferred bindings
Given a resource handler service with mock lookups
And a deferred binding result for slot "extra"
And a binding result for slot "repo" with resource id "RES001"
When I resolve all bindings via resource handler service for plan "PLAN011"
Then the handler service should return 1 bound resource
And the handler service should have resolved slot "repo"
Scenario: ResourceHandlerService rejects deferred single binding
Given a resource handler service with mock lookups
And a deferred binding result for slot "extra"
When I try to resolve the single deferred binding via resource handler service
Then a handler ValueError should be raised
And the handler error should mention "deferred"
Scenario: ResourceHandlerService uses fallback handler when type has no handler string
Given a resource handler service with mock lookups and no handler string
And a binding result for slot "data" with resource id "RES002"
When I resolve the binding via resource handler service for plan "PLAN012"
Then the handler service should return a BoundResource
And the handler service BoundResource sandbox path should not be empty
features/steps/decision_model_steps.py
+556
View File
@@ -0,0 +1,556 @@
"""Step definitions for decision_model.feature.
Tests the Decision, DecisionType, ContextSnapshot, ResourceRef,
and ArtifactRef domain models.
"""
from __future__ import annotations
from typing import Any
from behave import given, then, when # type: ignore[import-untyped]
from behave.runner import Context
from pydantic import ValidationError
from ulid import ULID
from cleveragents.domain.models.core.decision import (
EXECUTE_TYPES,
STRATEGIZE_TYPES,
ArtifactRef,
ContextSnapshot,
Decision,
DecisionType,
ResourceRef,
)
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
_VALID_PLAN_ULID = str(ULID())
_VALID_PARENT_ULID = str(ULID())
_VALID_CORRECTED_ULID = str(ULID())
_VALID_SUPERSEDED_ULID = str(ULID())
def _make_decision(
plan_id: str = _VALID_PLAN_ULID,
parent_decision_id: str | None = None,
sequence_number: int = 0,
decision_type: DecisionType = DecisionType.PROMPT_DEFINITION,
question: str = "What approach should we take?",
chosen_option: str = "Build a REST API",
confidence_score: float | None = None,
context_snapshot: ContextSnapshot | None = None,
is_correction: bool = False,
corrects_decision_id: str | None = None,
correction_reason: str | None = None,
superseded_by: str | None = None,
artifacts_produced: list[ArtifactRef] | None = None,
) -> Decision:
kwargs: dict[str, Any] = {
"plan_id": plan_id,
"sequence_number": sequence_number,
"decision_type": decision_type,
"question": question,
"chosen_option": chosen_option,
"is_correction": is_correction,
}
if parent_decision_id is not None:
kwargs["parent_decision_id"] = parent_decision_id
if confidence_score is not None:
kwargs["confidence_score"] = confidence_score
if context_snapshot is not None:
kwargs["context_snapshot"] = context_snapshot
if corrects_decision_id is not None:
kwargs["corrects_decision_id"] = corrects_decision_id
if correction_reason is not None:
kwargs["correction_reason"] = correction_reason
if superseded_by is not None:
kwargs["superseded_by"] = superseded_by
if artifacts_produced is not None:
kwargs["artifacts_produced"] = artifacts_produced
return Decision(**kwargs)
# ---------------------------------------------------------------------------
# DecisionType enum
# ---------------------------------------------------------------------------
@then("the DecisionType enum should have exactly {count:d} members")
def step_decision_type_count(context: Context, count: int) -> None:
assert len(DecisionType) == count, (
f"Expected {count} members, got {len(DecisionType)}"
)
@then('STRATEGIZE_TYPES should contain "{dtype}"')
def step_strategize_contains(context: Context, dtype: str) -> None:
assert DecisionType(dtype) in STRATEGIZE_TYPES, f"{dtype} not in STRATEGIZE_TYPES"
@then('STRATEGIZE_TYPES should not contain "{dtype}"')
def step_strategize_not_contains(context: Context, dtype: str) -> None:
assert DecisionType(dtype) not in STRATEGIZE_TYPES
@then("STRATEGIZE_TYPES should have exactly {count:d} members")
def step_strategize_count(context: Context, count: int) -> None:
assert len(STRATEGIZE_TYPES) == count
@then('EXECUTE_TYPES should contain "{dtype}"')
def step_execute_contains(context: Context, dtype: str) -> None:
assert DecisionType(dtype) in EXECUTE_TYPES, f"{dtype} not in EXECUTE_TYPES"
@then('EXECUTE_TYPES should not contain "{dtype}"')
def step_execute_not_contains(context: Context, dtype: str) -> None:
assert DecisionType(dtype) not in EXECUTE_TYPES
@then("EXECUTE_TYPES should have exactly {count:d} members")
def step_execute_count(context: Context, count: int) -> None:
assert len(EXECUTE_TYPES) == count
# ---------------------------------------------------------------------------
# Given steps
# ---------------------------------------------------------------------------
@given("a valid plan ULID")
def step_valid_plan_ulid(context: Context) -> None:
context.decision_plan_id = str(ULID())
@given("a valid parent decision ULID")
def step_valid_parent_ulid(context: Context) -> None:
context.decision_parent_id = str(ULID())
@given("a valid corrected decision ULID")
def step_valid_corrected_ulid(context: Context) -> None:
context.decision_corrected_id = str(ULID())
@given("a context snapshot with hash and resources")
def step_context_snapshot(context: Context) -> None:
context.decision_snapshot = ContextSnapshot(
hot_context_hash="sha256:abc123",
hot_context_ref="store://snapshots/abc123",
relevant_resources=[
ResourceRef(resource_id=str(ULID()), path="src/main.py"),
ResourceRef(resource_id=str(ULID())),
],
actor_state_ref="checkpoint://actor/001",
)
# ---------------------------------------------------------------------------
# When steps — successful creation
# ---------------------------------------------------------------------------
@when("I create a prompt_definition decision with sequence {seq:d}")
def step_create_root(context: Context, seq: int) -> None:
context.decision_result = _make_decision(
plan_id=context.decision_plan_id,
sequence_number=seq,
decision_type=DecisionType.PROMPT_DEFINITION,
)
@when("I create a strategy_choice decision with sequence {seq:d} and a parent")
def step_create_strategy(context: Context, seq: int) -> None:
context.decision_result = _make_decision(
plan_id=context.decision_plan_id,
parent_decision_id=context.decision_parent_id,
sequence_number=seq,
decision_type=DecisionType.STRATEGY_CHOICE,
)
@when("I create an implementation_choice decision with sequence {seq:d} and a parent")
def step_create_impl(context: Context, seq: int) -> None:
context.decision_result = _make_decision(
plan_id=context.decision_plan_id,
parent_decision_id=context.decision_parent_id,
sequence_number=seq,
decision_type=DecisionType.IMPLEMENTATION_CHOICE,
)
@when("I create a user_intervention decision with sequence {seq:d}")
def step_create_user_intervention(context: Context, seq: int) -> None:
context.decision_result = _make_decision(
plan_id=context.decision_plan_id,
sequence_number=seq,
decision_type=DecisionType.USER_INTERVENTION,
)
@when("I create a decision with confidence score {score}")
def step_create_with_confidence(context: Context, score: str) -> None:
conf = None if score == "None" else float(score)
context.decision_result = _make_decision(
plan_id=context.decision_plan_id,
confidence_score=conf,
decision_type=DecisionType.STRATEGY_CHOICE,
)
@when("I create a correction decision")
def step_create_correction(context: Context) -> None:
context.decision_result = _make_decision(
plan_id=context.decision_plan_id,
decision_type=DecisionType.STRATEGY_CHOICE,
is_correction=True,
corrects_decision_id=context.decision_corrected_id,
correction_reason="Original approach was suboptimal",
)
@when("I create a decision that is superseded")
def step_create_superseded(context: Context) -> None:
context.decision_result = _make_decision(
plan_id=context.decision_plan_id,
decision_type=DecisionType.STRATEGY_CHOICE,
superseded_by=str(ULID()),
)
@when("I create a decision with the context snapshot")
def step_create_with_snapshot(context: Context) -> None:
context.decision_result = _make_decision(
plan_id=context.decision_plan_id,
decision_type=DecisionType.STRATEGY_CHOICE,
context_snapshot=context.decision_snapshot,
)
@when("I create a decision with artifacts")
def step_create_with_artifacts(context: Context) -> None:
context.decision_result = _make_decision(
plan_id=context.decision_plan_id,
decision_type=DecisionType.STRATEGY_CHOICE,
artifacts_produced=[
ArtifactRef(artifact_path="src/api.py", artifact_type="file"),
ArtifactRef(artifact_path="patches/fix.patch", artifact_type="patch"),
],
)
@when("I create a fully populated decision")
def step_create_fully_populated(context: Context) -> None:
context.decision_result = _make_decision(
plan_id=context.decision_plan_id,
parent_decision_id=str(ULID()),
sequence_number=5,
decision_type=DecisionType.STRATEGY_CHOICE,
question="Which framework?",
chosen_option="FastAPI",
confidence_score=0.92,
context_snapshot=ContextSnapshot(
hot_context_hash="sha256:def456",
hot_context_ref="store://snapshots/def456",
relevant_resources=[ResourceRef(resource_id=str(ULID()))],
actor_state_ref="checkpoint://actor/005",
),
artifacts_produced=[ArtifactRef(artifact_path="src/app.py")],
)
@when('I create a decision of type "{dtype}"')
def step_create_any_type(context: Context, dtype: str) -> None:
dt = DecisionType(dtype)
parent = None if dt == DecisionType.PROMPT_DEFINITION else str(ULID())
context.decision_result = _make_decision(
plan_id=context.decision_plan_id,
parent_decision_id=parent,
decision_type=dt,
)
# ---------------------------------------------------------------------------
# When steps — expected failures
# ---------------------------------------------------------------------------
@when('I try to create a decision with plan_id "{plan_id}"')
def step_try_invalid_plan_id(context: Context, plan_id: str) -> None:
try:
_make_decision(plan_id=plan_id)
context.decision_error = None
except ValidationError as exc:
context.decision_error = exc
@when('I try to create a decision with parent_decision_id "{parent_id}"')
def step_try_invalid_parent(context: Context, parent_id: str) -> None:
try:
_make_decision(
plan_id=context.decision_plan_id,
parent_decision_id=parent_id,
decision_type=DecisionType.STRATEGY_CHOICE,
)
context.decision_error = None
except ValidationError as exc:
context.decision_error = exc
@when("I try to create a prompt_definition with a parent")
def step_try_root_with_parent(context: Context) -> None:
try:
_make_decision(
plan_id=context.decision_plan_id,
parent_decision_id=context.decision_parent_id,
decision_type=DecisionType.PROMPT_DEFINITION,
)
context.decision_error = None
except ValidationError as exc:
context.decision_error = exc
@when("I try to create a decision with confidence score {score}")
def step_try_bad_confidence(context: Context, score: str) -> None:
try:
_make_decision(
plan_id=context.decision_plan_id,
confidence_score=float(score),
decision_type=DecisionType.STRATEGY_CHOICE,
)
context.decision_error = None
except ValidationError as exc:
context.decision_error = exc
@when("I try to create a decision with is_correction true but no corrects_decision_id")
def step_try_correction_no_target(context: Context) -> None:
try:
_make_decision(
plan_id=context.decision_plan_id,
decision_type=DecisionType.STRATEGY_CHOICE,
is_correction=True,
)
context.decision_error = None
except ValidationError as exc:
context.decision_error = exc
@when("I try to create a decision with corrects_decision_id but is_correction false")
def step_try_correction_flag_mismatch(context: Context) -> None:
try:
_make_decision(
plan_id=context.decision_plan_id,
decision_type=DecisionType.STRATEGY_CHOICE,
is_correction=False,
corrects_decision_id=context.decision_corrected_id,
)
context.decision_error = None
except ValidationError as exc:
context.decision_error = exc
@when('I try to create a decision with superseded_by "{value}"')
def step_try_bad_superseded(context: Context, value: str) -> None:
try:
_make_decision(
plan_id=context.decision_plan_id,
decision_type=DecisionType.STRATEGY_CHOICE,
superseded_by=value,
)
context.decision_error = None
except ValidationError as exc:
context.decision_error = exc
@when('I try to create a decision with corrects_decision_id ULID "{value}"')
def step_try_bad_corrects_id(context: Context, value: str) -> None:
try:
_make_decision(
plan_id=context.decision_plan_id,
decision_type=DecisionType.STRATEGY_CHOICE,
is_correction=True,
corrects_decision_id=value,
)
context.decision_error = None
except ValidationError as exc:
context.decision_error = exc
@when("I call with_superseded_by on the decision")
def step_call_with_superseded_by(context: Context) -> None:
context.decision_superseded_copy = context.decision_result.with_superseded_by(
str(ULID())
)
@when("I try to create a ResourceRef with empty resource_id")
def step_try_empty_resource_ref(context: Context) -> None:
try:
ResourceRef(resource_id="")
context.decision_error = None
except ValidationError as exc:
context.decision_error = exc
@when("I try to create an ArtifactRef with empty artifact_path")
def step_try_empty_artifact_ref(context: Context) -> None:
try:
ArtifactRef(artifact_path="")
context.decision_error = None
except ValidationError as exc:
context.decision_error = exc
# ---------------------------------------------------------------------------
# Then steps
# ---------------------------------------------------------------------------
@then("the decision should be created successfully")
def step_decision_created(context: Context) -> None:
assert context.decision_result is not None
assert isinstance(context.decision_result, Decision)
@then("the decision should be a root decision")
def step_is_root(context: Context) -> None:
assert context.decision_result.is_root
@then("the decision should not be a root decision")
def step_not_root(context: Context) -> None:
assert not context.decision_result.is_root
@then("the decision should have a valid ULID as decision_id")
def step_valid_decision_id(context: Context) -> None:
import re
pattern = r"^[0-9A-HJKMNP-TV-Z]{26}$"
assert re.match(pattern, context.decision_result.decision_id), (
f"Invalid ULID: {context.decision_result.decision_id}"
)
@then('the decision type should be "{dtype}"')
def step_decision_type_check(context: Context, dtype: str) -> None:
assert context.decision_result.decision_type == DecisionType(dtype)
@then("the decision is_strategize_type should be true")
def step_is_strategize(context: Context) -> None:
assert context.decision_result.is_strategize_type
@then("the decision is_strategize_type should be false")
def step_not_strategize(context: Context) -> None:
assert not context.decision_result.is_strategize_type
@then("the decision is_execute_type should be true")
def step_is_execute(context: Context) -> None:
assert context.decision_result.is_execute_type
@then("the decision is_execute_type should be false")
def step_not_execute(context: Context) -> None:
assert not context.decision_result.is_execute_type
@then("the decision is_any_phase_type should be true")
def step_is_any_phase(context: Context) -> None:
assert context.decision_result.is_any_phase_type
@then("the decision confidence score should be {score}")
def step_confidence_value(context: Context, score: str) -> None:
if score == "None":
assert context.decision_result.confidence_score is None
else:
assert context.decision_result.confidence_score == float(score)
@then("the decision is_correction should be true")
def step_is_correction(context: Context) -> None:
assert context.decision_result.is_correction
@then("the decision is_superseded should be true")
def step_is_superseded(context: Context) -> None:
assert context.decision_result.is_superseded
@then("the original decision should not be superseded")
def step_original_not_superseded(context: Context) -> None:
assert not context.decision_result.is_superseded
@then("the superseded copy should be superseded")
def step_copy_is_superseded(context: Context) -> None:
assert context.decision_superseded_copy.is_superseded
assert context.decision_superseded_copy.decision_id == (
context.decision_result.decision_id
)
@then("a decision validation error should be raised")
def step_validation_error(context: Context) -> None:
assert context.decision_error is not None, (
"Expected ValidationError but none raised"
)
assert isinstance(context.decision_error, ValidationError)
@then('the decision error should mention "{text}"')
def step_error_mentions(context: Context, text: str) -> None:
assert text.lower() in str(context.decision_error).lower(), (
f"Expected '{text}' in error: {context.decision_error}"
)
@then("the decision context snapshot hash should not be empty")
def step_snapshot_hash_not_empty(context: Context) -> None:
assert context.decision_result.context_snapshot.hot_context_hash
@then("the decision context snapshot should have resources")
def step_snapshot_has_resources(context: Context) -> None:
assert len(context.decision_result.context_snapshot.relevant_resources) > 0
@then("the decision context snapshot hash should be empty")
def step_snapshot_hash_empty(context: Context) -> None:
assert context.decision_result.context_snapshot.hot_context_hash == ""
@then("the decision should have {count:d} artifacts produced")
def step_artifact_count(context: Context, count: int) -> None:
assert len(context.decision_result.artifacts_produced) == count
@then("the decision should round-trip through model_dump and model_validate")
def step_roundtrip(context: Context) -> None:
data = context.decision_result.model_dump()
restored = Decision.model_validate(data)
assert restored.decision_id == context.decision_result.decision_id
assert restored.decision_type == context.decision_result.decision_type
assert restored.question == context.decision_result.question
assert restored.chosen_option == context.decision_result.chosen_option
assert restored.confidence_score == context.decision_result.confidence_score
assert (
restored.context_snapshot.hot_context_hash
== context.decision_result.context_snapshot.hot_context_hash
)
assert len(restored.artifacts_produced) == len(
context.decision_result.artifacts_produced
)
@then('as_cli_dict should contain key "{key}"')
def step_cli_dict_key(context: Context, key: str) -> None:
cli = context.decision_result.as_cli_dict()
assert key in cli, f"Key '{key}' not in as_cli_dict: {list(cli.keys())}"
features/steps/resource_handlers_steps.py
+462
View File
@@ -0,0 +1,462 @@
"""Step definitions for resource_handlers.feature.
Tests the resource handler protocol, GitCheckoutHandler, FsDirectoryHandler,
handler resolver, and ResourceHandlerService orchestration bridge.
"""
from __future__ import annotations
from datetime import datetime
from unittest.mock import MagicMock, PropertyMock
from behave import given, then, when # type: ignore[import-untyped]
from behave.runner import Context
from cleveragents.domain.models.core.resource import (
PhysVirt,
Resource,
ResourceCapabilities,
SandboxStrategy,
)
from cleveragents.domain.models.core.resource_slot import BindingResult
from cleveragents.domain.models.core.resource_type import (
ResourceKind,
ResourceTypeSpec,
)
from cleveragents.domain.models.core.resource_type import (
SandboxStrategy as TypeSandboxStrategy,
)
from cleveragents.infrastructure.sandbox.factory import SandboxFactory
from cleveragents.infrastructure.sandbox.manager import SandboxManager
from cleveragents.infrastructure.sandbox.protocol import (
SandboxContext,
SandboxStatus,
)
from cleveragents.resource.handlers.fs_directory import FsDirectoryHandler
from cleveragents.resource.handlers.git_checkout import GitCheckoutHandler
from cleveragents.resource.handlers.protocol import ResourceHandler
from cleveragents.resource.handlers.resolver import (
HandlerResolutionError,
clear_handler_cache,
resolve_handler,
)
from cleveragents.tool.context import BoundResource
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _make_resource(
resource_type_name: str,
location: str | None = None,
sandbox_strategy: SandboxStrategy | None = None,
resource_id: str = "01KJ5C5TPMP8GGX3QC83E2MAQS",
) -> Resource:
"""Create a test Resource domain object."""
return Resource(
resource_id=resource_id,
resource_type_name=resource_type_name,
classification=PhysVirt.PHYSICAL,
location=location,
sandbox_strategy=sandbox_strategy,
capabilities=ResourceCapabilities(
readable=True, writable=True, sandboxable=True
),
)
def _make_mock_sandbox_manager() -> tuple[SandboxManager, MagicMock]:
"""Create a SandboxManager with a mock factory that tracks calls."""
mock_factory = MagicMock(spec=SandboxFactory)
mock_sandbox = MagicMock()
mock_sandbox.sandbox_id = "sb-test-001"
type(mock_sandbox).status = PropertyMock(return_value=SandboxStatus.CREATED)
mock_sandbox.context = SandboxContext(
sandbox_id="sb-test-001",
sandbox_path="/tmp/sandbox/sb-test-001",
original_path="/tmp/original",
resource_id="res-test",
plan_id="plan-test",
created_at=datetime.now(),
)
mock_sandbox.create.return_value = mock_sandbox.context
mock_factory.create_sandbox.return_value = mock_sandbox
manager = SandboxManager(factory=mock_factory, cleanup_on_exit=False)
return manager, mock_factory
def _make_type_spec(
name: str = "git-checkout",
sandbox_strategy: str = "git_worktree",
handler: str
| None = "cleveragents.resource.handlers.git_checkout:GitCheckoutHandler",
) -> ResourceTypeSpec:
"""Create a minimal ResourceTypeSpec for testing."""
return ResourceTypeSpec(
name=name,
description=f"Test {name} type",
resource_kind=ResourceKind.PHYSICAL,
sandbox_strategy=TypeSandboxStrategy(sandbox_strategy),
user_addable=True,
built_in=True,
handler=handler,
)
# ---------------------------------------------------------------------------
# Protocol conformance
# ---------------------------------------------------------------------------
@then("GitCheckoutHandler should satisfy the ResourceHandler protocol")
def step_git_handler_protocol(context: Context) -> None:
handler = GitCheckoutHandler()
assert isinstance(handler, ResourceHandler), (
"GitCheckoutHandler does not satisfy ResourceHandler protocol"
)
@then("FsDirectoryHandler should satisfy the ResourceHandler protocol")
def step_fs_handler_protocol(context: Context) -> None:
handler = FsDirectoryHandler()
assert isinstance(handler, ResourceHandler), (
"FsDirectoryHandler does not satisfy ResourceHandler protocol"
)
# ---------------------------------------------------------------------------
# GitCheckoutHandler
# ---------------------------------------------------------------------------
@given('a git-checkout resource with location "{location}"')
def step_git_resource(context: Context, location: str) -> None:
context.handler_resource = _make_resource("git-checkout", location=location)
@given('a git-checkout resource at "{location}" using strategy "{strategy}"')
def step_git_resource_with_strategy(
context: Context, location: str, strategy: str
) -> None:
context.handler_resource = _make_resource(
"git-checkout",
location=location,
sandbox_strategy=SandboxStrategy(strategy),
)
@given("a git-checkout resource without location")
def step_git_resource_no_location(context: Context) -> None:
context.handler_resource = _make_resource("git-checkout", location=None)
@given("a sandbox manager with a mock factory")
def step_mock_sandbox_manager(context: Context) -> None:
context.handler_sandbox_manager, context.handler_mock_factory = (
_make_mock_sandbox_manager()
)
@when('I resolve the resource with GitCheckoutHandler for plan "{plan_id}"')
def step_resolve_git(context: Context, plan_id: str) -> None:
handler = GitCheckoutHandler()
context.handler_bound = handler.resolve(
resource=context.handler_resource,
plan_id=plan_id,
slot_name="repo",
sandbox_manager=context.handler_sandbox_manager,
)
@when("I try to resolve the resource with GitCheckoutHandler")
def step_try_resolve_git(context: Context) -> None:
handler = GitCheckoutHandler()
try:
handler.resolve(
resource=context.handler_resource,
plan_id="PLAN_ERR",
slot_name="repo",
sandbox_manager=context.handler_sandbox_manager,
)
context.handler_error = None
except ValueError as exc:
context.handler_error = exc
# ---------------------------------------------------------------------------
# FsDirectoryHandler
# ---------------------------------------------------------------------------
@given('an fs-directory resource with location "{location}"')
def step_fs_resource(context: Context, location: str) -> None:
context.handler_resource = _make_resource("fs-directory", location=location)
@given("an fs-directory resource without location")
def step_fs_resource_no_location(context: Context) -> None:
context.handler_resource = _make_resource("fs-directory", location=None)
@when('I resolve the resource with FsDirectoryHandler for plan "{plan_id}"')
def step_resolve_fs(context: Context, plan_id: str) -> None:
handler = FsDirectoryHandler()
context.handler_bound = handler.resolve(
resource=context.handler_resource,
plan_id=plan_id,
slot_name="workdir",
sandbox_manager=context.handler_sandbox_manager,
)
@when("I try to resolve the resource with FsDirectoryHandler")
def step_try_resolve_fs(context: Context) -> None:
handler = FsDirectoryHandler()
try:
handler.resolve(
resource=context.handler_resource,
plan_id="PLAN_ERR",
slot_name="workdir",
sandbox_manager=context.handler_sandbox_manager,
)
context.handler_error = None
except ValueError as exc:
context.handler_error = exc
# ---------------------------------------------------------------------------
# Shared handler assertions
# ---------------------------------------------------------------------------
@then('the bound resource should have slot name "{name}"')
def step_bound_slot(context: Context, name: str) -> None:
assert context.handler_bound.slot_name == name
@then('the bound resource should have resource type "{rtype}"')
def step_bound_type(context: Context, rtype: str) -> None:
assert context.handler_bound.resource_type == rtype
@then("the bound resource sandbox path should not be empty")
def step_bound_sandbox_not_empty(context: Context) -> None:
assert context.handler_bound.sandbox_path, (
f"sandbox_path is empty: {context.handler_bound.sandbox_path!r}"
)
@then('the sandbox was created with strategy "{strategy}"')
def step_sandbox_strategy(context: Context, strategy: str) -> None:
call_args = context.handler_mock_factory.create_sandbox.call_args
assert call_args is not None, "create_sandbox was not called"
assert call_args.kwargs.get("sandbox_strategy") == strategy or (
len(call_args.args) >= 3 and call_args.args[2] == strategy
), f"Expected strategy {strategy}, got {call_args}"
@then("a handler ValueError should be raised")
def step_handler_value_error(context: Context) -> None:
assert context.handler_error is not None
assert isinstance(context.handler_error, ValueError)
@then('the handler error should mention "{text}"')
def step_handler_error_text(context: Context, text: str) -> None:
assert text.lower() in str(context.handler_error).lower(), (
f"Expected '{text}' in error: {context.handler_error}"
)
# ---------------------------------------------------------------------------
# Handler Resolver
# ---------------------------------------------------------------------------
@when('I resolve handler "{ref}"')
def step_resolve_handler(context: Context, ref: str) -> None:
clear_handler_cache()
context.handler_resolved = resolve_handler(ref)
context.handler_ref_used = ref
@when("I resolve the same handler reference again")
def step_resolve_handler_again(context: Context) -> None:
context.handler_resolved_second = resolve_handler(context.handler_ref_used)
@then("the resolved handler should be a GitCheckoutHandler instance")
def step_resolved_is_git(context: Context) -> None:
assert isinstance(context.handler_resolved, GitCheckoutHandler)
@then("the resolved handler should be a FsDirectoryHandler instance")
def step_resolved_is_fs(context: Context) -> None:
assert isinstance(context.handler_resolved, FsDirectoryHandler)
@then("both resolved handlers should be the same object")
def step_same_object(context: Context) -> None:
assert context.handler_resolved is context.handler_resolved_second
@when("I try to resolve an empty handler reference")
def step_try_resolve_empty_handler(context: Context) -> None:
clear_handler_cache()
try:
resolve_handler("")
context.handler_resolution_error = None
except HandlerResolutionError as exc:
context.handler_resolution_error = exc
@when('I try to resolve handler "{ref}"')
def step_try_resolve_handler(context: Context, ref: str) -> None:
clear_handler_cache()
try:
resolve_handler(ref)
context.handler_resolution_error = None
except HandlerResolutionError as exc:
context.handler_resolution_error = exc
@then("a HandlerResolutionError should be raised")
def step_resolution_error(context: Context) -> None:
assert context.handler_resolution_error is not None
assert isinstance(context.handler_resolution_error, HandlerResolutionError)
@then('the resolution error should mention "{text}"')
def step_resolution_error_text(context: Context, text: str) -> None:
assert text.lower() in str(context.handler_resolution_error).lower(), (
f"Expected '{text}' in error: {context.handler_resolution_error}"
)
# ---------------------------------------------------------------------------
# ResourceHandlerService
# ---------------------------------------------------------------------------
@given("a resource handler service with mock lookups")
def step_handler_service(context: Context) -> None:
from cleveragents.application.services.resource_handler_service import (
ResourceHandlerService,
)
manager, _mock_factory = _make_mock_sandbox_manager()
resource = _make_resource("git-checkout", location="/tmp/test-repo")
type_spec = _make_type_spec()
context.handler_svc = ResourceHandlerService(
sandbox_manager=manager,
resource_lookup=lambda _name_or_id: resource,
type_lookup=lambda _name: type_spec,
)
context.handler_svc_manager = manager
@given("a resource handler service with mock lookups and no handler string")
def step_handler_service_no_handler(context: Context) -> None:
from cleveragents.application.services.resource_handler_service import (
ResourceHandlerService,
)
manager, _mock_factory = _make_mock_sandbox_manager()
resource = _make_resource("fs-directory", location="/tmp/test-dir")
type_spec = _make_type_spec(
name="fs-directory",
sandbox_strategy="copy_on_write",
handler=None,
)
context.handler_svc = ResourceHandlerService(
sandbox_manager=manager,
resource_lookup=lambda _name_or_id: resource,
type_lookup=lambda _name: type_spec,
)
@given('a binding result for slot "{slot}" with resource id "{res_id}"')
def step_binding_result(context: Context, slot: str, res_id: str) -> None:
if not hasattr(context, "handler_bindings"):
context.handler_bindings = []
context.handler_binding = BindingResult(
slot_name=slot,
resource_id=res_id,
resource_name=f"test-{res_id}",
binding_mode="contextual",
)
context.handler_bindings.append(context.handler_binding)
@given('a deferred binding result for slot "{slot}"')
def step_deferred_binding(context: Context, slot: str) -> None:
if not hasattr(context, "handler_bindings"):
context.handler_bindings = []
context.handler_deferred_binding = BindingResult(
slot_name=slot,
resource_id=None,
binding_mode="parameter",
deferred=True,
)
context.handler_bindings.append(context.handler_deferred_binding)
@when('I resolve the binding via resource handler service for plan "{plan_id}"')
def step_svc_resolve_binding(context: Context, plan_id: str) -> None:
context.handler_svc_bound = context.handler_svc.resolve_binding(
binding=context.handler_binding,
plan_id=plan_id,
)
@when('I resolve all bindings via resource handler service for plan "{plan_id}"')
def step_svc_resolve_all(context: Context, plan_id: str) -> None:
context.handler_svc_all = context.handler_svc.resolve_bindings(
bindings=context.handler_bindings,
plan_id=plan_id,
)
@when("I try to resolve the single deferred binding via resource handler service")
def step_svc_try_resolve_deferred(context: Context) -> None:
try:
context.handler_svc.resolve_binding(
binding=context.handler_deferred_binding,
plan_id="PLAN_ERR",
)
context.handler_error = None
except ValueError as exc:
context.handler_error = exc
@then("the handler service should return a BoundResource")
def step_svc_is_bound(context: Context) -> None:
assert isinstance(context.handler_svc_bound, BoundResource)
@then('the handler service BoundResource slot should be "{slot}"')
def step_svc_bound_slot(context: Context, slot: str) -> None:
assert context.handler_svc_bound.slot_name == slot
@then("the handler service BoundResource sandbox path should not be empty")
def step_svc_bound_path(context: Context) -> None:
assert context.handler_svc_bound.sandbox_path, (
f"sandbox_path is empty: {context.handler_svc_bound.sandbox_path!r}"
)
@then("the handler service should return {count:d} bound resource")
def step_svc_count(context: Context, count: int) -> None:
assert len(context.handler_svc_all) == count
@then('the handler service should have resolved slot "{slot}"')
def step_svc_has_slot(context: Context, slot: str) -> None:
assert slot in context.handler_svc_all
implementation_plan.md
+259 -207
View File
File diff suppressed because it is too large Load Diff
robot/decision_model.robot
+51
View File
@@ -0,0 +1,51 @@
*** Settings ***
Documentation Smoke tests for decision domain model persistence contract
Resource ${CURDIR}/common.resource
Suite Setup Setup Test Environment
Suite Teardown Cleanup Test Environment
*** Variables ***
${HELPER_SCRIPT} robot/helper_decision_model.py
*** Test Cases ***
Create Root Decision
[Documentation] Create a prompt_definition root decision
${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} create_root cwd=${WORKSPACE}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} decision-create-root-ok
Create Child Decision
[Documentation] Create a strategy_choice child decision
${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} create_child cwd=${WORKSPACE}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} decision-create-child-ok
Decision Type Enum Values
[Documentation] Verify all 11 decision type enum values
${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} enum_values cwd=${WORKSPACE}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} decision-enums-ok
Context Snapshot
[Documentation] Create a decision with a populated context snapshot
${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} context_snapshot cwd=${WORKSPACE}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} decision-snapshot-ok
Correction Metadata
[Documentation] Create a correction decision and verify metadata
${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} correction cwd=${WORKSPACE}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} decision-correction-ok
Round Trip Serialization
[Documentation] Verify model_dump / model_validate round-trip
${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} roundtrip cwd=${WORKSPACE}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} decision-roundtrip-ok
CLI Dict Output
[Documentation] Verify as_cli_dict produces expected keys
${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} cli_dict cwd=${WORKSPACE}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} decision-cli-dict-ok
robot/helper_decision_model.py
+202
View File
@@ -0,0 +1,202 @@
"""Helper script for Robot Framework decision model smoke tests."""
from __future__ import annotations
import sys
from pathlib import Path
# Ensure src is importable when run from workspace root
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
from ulid import ULID
from cleveragents.domain.models.core.decision import (
ArtifactRef,
ContextSnapshot,
Decision,
DecisionType,
ResourceRef,
)
def _test_create_root() -> None:
"""Create a root prompt_definition decision."""
d = Decision(
plan_id=str(ULID()),
sequence_number=0,
decision_type=DecisionType.PROMPT_DEFINITION,
question="What should we build?",
chosen_option="A REST API for user management",
)
assert d.is_root
assert d.decision_type == DecisionType.PROMPT_DEFINITION
assert d.parent_decision_id is None
print("decision-create-root-ok")
def _test_create_child() -> None:
"""Create a strategy_choice child decision."""
parent_id = str(ULID())
d = Decision(
plan_id=str(ULID()),
parent_decision_id=parent_id,
sequence_number=1,
decision_type=DecisionType.STRATEGY_CHOICE,
question="Which framework?",
chosen_option="FastAPI",
alternatives_considered=["Flask", "Django"],
confidence_score=0.9,
)
assert not d.is_root
assert d.parent_decision_id == parent_id
assert d.confidence_score == 0.9
assert len(d.alternatives_considered) == 2
print("decision-create-child-ok")
def _test_enum_values() -> None:
"""Verify all 11 decision type enum values."""
expected = {
"prompt_definition",
"invariant_enforced",
"strategy_choice",
"implementation_choice",
"resource_selection",
"subplan_spawn",
"subplan_parallel_spawn",
"tool_invocation",
"error_recovery",
"validation_response",
"user_intervention",
}
actual = {dt.value for dt in DecisionType}
assert actual == expected, f"Mismatch: {actual.symmetric_difference(expected)}"
assert len(DecisionType) == 11
print("decision-enums-ok")
def _test_context_snapshot() -> None:
"""Create a decision with a populated context snapshot."""
snap = ContextSnapshot(
hot_context_hash="sha256:abc123def",
hot_context_ref="store://snapshots/abc123def",
relevant_resources=[
ResourceRef(resource_id=str(ULID()), path="src/main.py"),
ResourceRef(resource_id=str(ULID())),
],
actor_state_ref="checkpoint://actor/001",
)
d = Decision(
plan_id=str(ULID()),
sequence_number=0,
decision_type=DecisionType.PROMPT_DEFINITION,
question="What approach?",
chosen_option="Microservices",
context_snapshot=snap,
)
assert d.context_snapshot.hot_context_hash == "sha256:abc123def"
assert len(d.context_snapshot.relevant_resources) == 2
assert d.context_snapshot.actor_state_ref == "checkpoint://actor/001"
print("decision-snapshot-ok")
def _test_correction() -> None:
"""Create a correction decision and verify metadata."""
original_id = str(ULID())
d = Decision(
plan_id=str(ULID()),
parent_decision_id=str(ULID()),
sequence_number=3,
decision_type=DecisionType.STRATEGY_CHOICE,
question="Which framework?",
chosen_option="Django instead",
is_correction=True,
corrects_decision_id=original_id,
correction_reason="FastAPI lacks admin panel",
)
assert d.is_correction
assert d.corrects_decision_id == original_id
assert d.correction_reason == "FastAPI lacks admin panel"
assert not d.is_superseded
print("decision-correction-ok")
def _test_roundtrip() -> None:
"""Verify model_dump / model_validate round-trip."""
d = Decision(
plan_id=str(ULID()),
parent_decision_id=str(ULID()),
sequence_number=5,
decision_type=DecisionType.IMPLEMENTATION_CHOICE,
question="How to implement auth?",
chosen_option="JWT tokens",
confidence_score=0.85,
context_snapshot=ContextSnapshot(
hot_context_hash="sha256:roundtrip",
relevant_resources=[ResourceRef(resource_id=str(ULID()))],
),
artifacts_produced=[
ArtifactRef(artifact_path="src/auth.py", artifact_type="file"),
],
)
data = d.model_dump()
restored = Decision.model_validate(data)
assert restored.decision_id == d.decision_id
assert restored.decision_type == d.decision_type
assert restored.confidence_score == d.confidence_score
assert restored.context_snapshot.hot_context_hash == "sha256:roundtrip"
assert len(restored.artifacts_produced) == 1
print("decision-roundtrip-ok")
def _test_cli_dict() -> None:
"""Verify as_cli_dict produces expected keys."""
d = Decision(
plan_id=str(ULID()),
sequence_number=0,
decision_type=DecisionType.PROMPT_DEFINITION,
question="Build what?",
chosen_option="API",
)
cli = d.as_cli_dict()
required_keys = {
"decision_id",
"plan_id",
"type",
"sequence",
"question",
"chosen",
"confidence",
"parent",
"is_correction",
"superseded",
}
assert required_keys.issubset(cli.keys()), (
f"Missing keys: {required_keys - cli.keys()}"
)
print("decision-cli-dict-ok")
_TESTS = {
"create_root": _test_create_root,
"create_child": _test_create_child,
"enum_values": _test_enum_values,
"context_snapshot": _test_context_snapshot,
"correction": _test_correction,
"roundtrip": _test_roundtrip,
"cli_dict": _test_cli_dict,
}
if __name__ == "__main__":
if len(sys.argv) < 2:
print(f"Usage: {sys.argv[0]} <test_name>")
print(f"Available tests: {', '.join(sorted(_TESTS))}")
sys.exit(1)
test_name = sys.argv[1]
if test_name not in _TESTS:
print(f"Unknown test: {test_name}")
print(f"Available: {', '.join(sorted(_TESTS))}")
sys.exit(1)
_TESTS[test_name]()
robot/helper_resource_handlers.py
+147
View File
@@ -0,0 +1,147 @@
"""Helper utilities for resource handler Robot smoke tests.
Each command prints a single ``<command>-ok`` token on success so the
calling Robot test can assert on ``stdout``.
"""
from __future__ import annotations
import sys
from datetime import datetime
from unittest.mock import MagicMock, PropertyMock
from cleveragents.domain.models.core.resource import (
PhysVirt,
Resource,
ResourceCapabilities,
)
from cleveragents.infrastructure.sandbox.factory import SandboxFactory
from cleveragents.infrastructure.sandbox.manager import SandboxManager
from cleveragents.infrastructure.sandbox.protocol import (
SandboxContext,
SandboxStatus,
)
from cleveragents.resource.handlers.fs_directory import FsDirectoryHandler
from cleveragents.resource.handlers.git_checkout import GitCheckoutHandler
from cleveragents.resource.handlers.protocol import ResourceHandler
from cleveragents.resource.handlers.resolver import (
HandlerResolutionError,
clear_handler_cache,
resolve_handler,
)
from cleveragents.tool.context import BoundResource
def _make_resource(rtype: str, location: str) -> Resource:
"""Create a test resource with a valid ULID."""
return Resource(
resource_id="01KJ5C5TPMP8GGX3QC83E2MAQS",
resource_type_name=rtype,
classification=PhysVirt.PHYSICAL,
location=location,
capabilities=ResourceCapabilities(
readable=True, writable=True, sandboxable=True
),
)
def _make_mock_manager() -> SandboxManager:
"""Create a SandboxManager with a mock factory."""
mock_factory = MagicMock(spec=SandboxFactory)
mock_sandbox = MagicMock()
mock_sandbox.sandbox_id = "sb-robot-001"
type(mock_sandbox).status = PropertyMock(return_value=SandboxStatus.CREATED)
mock_sandbox.context = SandboxContext(
sandbox_id="sb-robot-001",
sandbox_path="/tmp/sandbox/sb-robot-001",
original_path="/tmp/original",
resource_id="res-robot",
plan_id="plan-robot",
created_at=datetime.now(),
)
mock_sandbox.create.return_value = mock_sandbox.context
mock_factory.create_sandbox.return_value = mock_sandbox
return SandboxManager(factory=mock_factory, cleanup_on_exit=False)
def _protocol_check() -> None:
"""Verify both handlers satisfy the ResourceHandler protocol."""
git = GitCheckoutHandler()
fs = FsDirectoryHandler()
assert isinstance(git, ResourceHandler), "GitCheckoutHandler not a ResourceHandler"
assert isinstance(fs, ResourceHandler), "FsDirectoryHandler not a ResourceHandler"
print("protocol-check-ok")
def _git_resolve() -> None:
"""Verify GitCheckoutHandler resolves a resource."""
resource = _make_resource("git-checkout", "/tmp/test-repo")
manager = _make_mock_manager()
handler = GitCheckoutHandler()
bound = handler.resolve(
resource=resource,
plan_id="PLAN_ROBOT",
slot_name="repo",
sandbox_manager=manager,
)
assert isinstance(bound, BoundResource)
assert bound.sandbox_path, "sandbox_path empty"
assert bound.resource_type == "git-checkout"
print("git-resolve-ok")
def _fs_resolve() -> None:
"""Verify FsDirectoryHandler resolves a resource."""
resource = _make_resource("fs-directory", "/tmp/test-dir")
manager = _make_mock_manager()
handler = FsDirectoryHandler()
bound = handler.resolve(
resource=resource,
plan_id="PLAN_ROBOT",
slot_name="workdir",
sandbox_manager=manager,
)
assert isinstance(bound, BoundResource)
assert bound.sandbox_path, "sandbox_path empty"
assert bound.resource_type == "fs-directory"
print("fs-resolve-ok")
def _resolver_import() -> None:
"""Verify handler resolver loads handlers from module:class strings."""
clear_handler_cache()
git = resolve_handler(
"cleveragents.resource.handlers.git_checkout:GitCheckoutHandler"
)
assert isinstance(git, GitCheckoutHandler)
fs = resolve_handler(
"cleveragents.resource.handlers.fs_directory:FsDirectoryHandler"
)
assert isinstance(fs, FsDirectoryHandler)
print("resolver-import-ok")
def _resolver_error() -> None:
"""Verify handler resolver raises on bad references."""
clear_handler_cache()
try:
resolve_handler("nonexistent.module:FakeClass")
print("resolver-error-FAIL")
except HandlerResolutionError:
print("resolver-error-ok")
_COMMANDS = {
"protocol-check": _protocol_check,
"git-resolve": _git_resolve,
"fs-resolve": _fs_resolve,
"resolver-import": _resolver_import,
"resolver-error": _resolver_error,
}
if __name__ == "__main__":
if len(sys.argv) < 2 or sys.argv[1] not in _COMMANDS:
print(f"Usage: {sys.argv[0]} <{'|'.join(_COMMANDS)}>", file=sys.stderr)
sys.exit(1)
_COMMANDS[sys.argv[1]]()
robot/resource_handlers.robot
+49
View File
@@ -0,0 +1,49 @@
*** Settings ***
Documentation Smoke tests for resource handler runtime
Resource ${CURDIR}/common.resource
Suite Setup Setup Test Environment
Suite Teardown Cleanup Test Environment
*** Variables ***
${HELPER} ${CURDIR}/helper_resource_handlers.py
*** Test Cases ***
Handler Protocol Conformance
[Documentation] Verify GitCheckoutHandler and FsDirectoryHandler satisfy ResourceHandler
${result}= Run Process ${PYTHON} ${HELPER} protocol-check cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} protocol-check-ok
GitCheckout Handler Resolution
[Documentation] Verify GitCheckoutHandler resolves a git-checkout resource to a sandbox
${result}= Run Process ${PYTHON} ${HELPER} git-resolve cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} git-resolve-ok
FsDirectory Handler Resolution
[Documentation] Verify FsDirectoryHandler resolves an fs-directory resource to a sandbox
${result}= Run Process ${PYTHON} ${HELPER} fs-resolve cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} fs-resolve-ok
Handler Resolver Dynamic Import
[Documentation] Verify resolve_handler loads handlers from module:class strings
${result}= Run Process ${PYTHON} ${HELPER} resolver-import cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} resolver-import-ok
Handler Resolver Error On Bad Reference
[Documentation] Verify resolve_handler raises HandlerResolutionError on bad references
${result}= Run Process ${PYTHON} ${HELPER} resolver-error cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} resolver-error-ok
src/cleveragents/application/services/resource_handler_service.py
+298
View File
@@ -0,0 +1,298 @@
"""Resource handler orchestration service.
The :class:`ResourceHandlerService` is the **orchestration bridge** that
connects the existing infrastructure pieces into a complete resource
resolution pipeline:
1. :class:`BindingResolutionService` resolves tool slots to
:class:`BindingResult` (resource_id).
2. :class:`ResourceRegistryService` looks up the :class:`Resource` to
get ``location`` and ``resource_type_name``.
3. :class:`ResourceRegistryService` looks up the :class:`ResourceTypeSpec`
to get ``sandbox_strategy`` and ``handler`` reference.
4. :func:`resolve_handler` dynamically loads the handler class.
5. The handler calls :meth:`SandboxManager.get_or_create_sandbox` and
returns a :class:`BoundResource` with ``sandbox_path`` populated.
The service also provides a convenience method that resolves ALL tool
bindings in one call, producing a ``dict[str, BoundResource]`` ready
for :class:`PlanExecutionContext`.
Based on:
- implementation_plan.md group M1.resource-handlers (L2254-L2271)
"""
from __future__ import annotations
import logging
from collections.abc import Callable
from typing import cast
from cleveragents.core.exceptions import NotFoundError
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.infrastructure.sandbox.factory import SandboxStrategyStr
from cleveragents.infrastructure.sandbox.manager import SandboxManager
from cleveragents.resource.handlers.protocol import ResourceHandler
from cleveragents.resource.handlers.resolver import (
HandlerResolutionError,
resolve_handler,
)
from cleveragents.tool.context import BoundResource
logger = logging.getLogger(__name__)
class ResourceHandlerService:
"""Orchestrates resource resolution from binding results to bound resources.
Bridges the gap between :class:`BindingResolutionService` output
(``BindingResult`` with ``resource_id``) and the execution context's
``BoundResource`` (with ``sandbox_path``).
Parameters
----------
sandbox_manager:
The sandbox lifecycle manager for creating/reusing sandboxes.
resource_lookup:
Callable that takes a resource name-or-id and returns a
:class:`Resource`. Typically
``ResourceRegistryService.show_resource``.
type_lookup:
Callable that takes a type name and returns a
:class:`ResourceTypeSpec`. Typically
``ResourceRegistryService.show_type``.
"""
def __init__(
self,
*,
sandbox_manager: SandboxManager,
resource_lookup: Callable[[str], Resource],
type_lookup: Callable[[str], ResourceTypeSpec],
) -> None:
self._sandbox_manager = sandbox_manager
self._resource_lookup = resource_lookup
self._type_lookup = type_lookup
self._logger = logger
def resolve_binding(
self,
binding: BindingResult,
plan_id: str,
access: str = "read_only",
) -> BoundResource:
"""Resolve a single binding result into a BoundResource.
Steps:
1. Look up the Resource by ``binding.resource_id``.
2. Look up the ResourceTypeSpec by ``resource.resource_type_name``.
3. Resolve or fallback the handler.
4. Call ``handler.resolve()`` to provision a sandbox and
produce a BoundResource.
Args:
binding: A resolved binding with a ``resource_id``.
plan_id: The plan requesting the resolution.
access: Access mode for the bound resource.
Returns:
A :class:`BoundResource` with ``sandbox_path`` populated.
Raises:
ValueError: If the binding has no resource_id (deferred).
NotFoundError: If the resource or type is not found.
HandlerResolutionError: If the handler cannot be loaded.
"""
if binding.deferred or not binding.resource_id:
raise ValueError(
f"Cannot resolve deferred binding for slot '{binding.slot_name}'"
)
# Step 1: Look up resource
resource: Resource = self._resource_lookup(binding.resource_id)
# Step 2: Look up type spec
type_spec: ResourceTypeSpec = self._type_lookup(resource.resource_type_name)
# Step 3: Resolve handler
handler = self._resolve_handler_for_type(type_spec, resource)
# Step 4: Delegate to handler
return handler.resolve(
resource=resource,
plan_id=plan_id,
slot_name=binding.slot_name,
sandbox_manager=self._sandbox_manager,
access=access,
)
def resolve_bindings(
self,
bindings: list[BindingResult],
plan_id: str,
access: str = "read_only",
) -> dict[str, BoundResource]:
"""Resolve all non-deferred bindings into BoundResources.
Deferred bindings (parameter mode) are silently skipped.
Args:
bindings: List of binding results from
:class:`BindingResolutionService`.
plan_id: The plan requesting the resolution.
access: Default access mode for all bindings.
Returns:
A dict mapping slot names to :class:`BoundResource` objects.
"""
result: dict[str, BoundResource] = {}
for binding in bindings:
if binding.deferred or not binding.resource_id:
self._logger.debug(
"Skipping deferred binding for slot '%s'",
binding.slot_name,
)
continue
try:
bound = self.resolve_binding(
binding=binding,
plan_id=plan_id,
access=access,
)
result[binding.slot_name] = bound
self._logger.info(
"Resolved binding slot='%s' -> resource='%s' sandbox='%s'",
binding.slot_name,
binding.resource_id,
bound.sandbox_path,
)
except (NotFoundError, HandlerResolutionError, ValueError) as exc:
self._logger.warning(
"Failed to resolve binding for slot '%s': %s",
binding.slot_name,
exc,
)
raise
return result
def resolve_resource(
self,
resource: Resource,
plan_id: str,
slot_name: str,
access: str = "read_only",
) -> BoundResource:
"""Resolve a resource directly (without going through BindingResult).
Convenience method when you already have the Resource object.
Args:
resource: The resource to resolve.
plan_id: The plan requesting the resolution.
slot_name: Name for the binding slot.
access: Access mode.
Returns:
A :class:`BoundResource` with ``sandbox_path`` populated.
"""
type_spec = self._type_lookup(resource.resource_type_name)
handler = self._resolve_handler_for_type(type_spec, resource)
return handler.resolve(
resource=resource,
plan_id=plan_id,
slot_name=slot_name,
sandbox_manager=self._sandbox_manager,
access=access,
)
def _resolve_handler_for_type(
self,
type_spec: ResourceTypeSpec,
resource: Resource,
) -> ResourceHandler:
"""Resolve the handler for a resource type spec.
If the type spec has a handler reference string, resolves it
dynamically. Otherwise falls back to a default handler based
on the sandbox strategy.
Args:
type_spec: The resource type specification.
resource: The resource being resolved.
Returns:
A handler instance satisfying :class:`ResourceHandler`.
"""
if type_spec.handler:
try:
return resolve_handler(type_spec.handler)
except HandlerResolutionError as exc:
self._logger.debug(
"Handler resolution error for '%s': %s",
type_spec.handler,
exc,
)
self._logger.warning(
"Failed to resolve handler '%s' for type '%s', "
"falling back to default handler",
type_spec.handler,
type_spec.name,
)
# Fallback: use DefaultHandler which delegates to SandboxManager
return _DefaultHandler(type_spec=type_spec)
class _DefaultHandler:
"""Fallback handler when no handler class is specified or resolution fails.
Delegates directly to :class:`SandboxManager` using the type's
default sandbox strategy (or resource override).
"""
def __init__(self, type_spec: ResourceTypeSpec) -> None:
self._type_spec = type_spec
def resolve(
self,
*,
resource: Resource,
plan_id: str,
slot_name: str,
sandbox_manager: SandboxManager,
access: str = "read_only",
) -> BoundResource:
"""Resolve using the default sandbox strategy."""
if not resource.location:
raise ValueError(f"Resource '{resource.resource_id}' has no location")
# resource.sandbox_strategy may be a StrEnum or plain str (Pydantic coercion)
if resource.sandbox_strategy:
strategy_str = str(resource.sandbox_strategy)
else:
strategy_str = str(self._type_spec.sandbox_strategy)
sandbox = sandbox_manager.get_or_create_sandbox(
plan_id=plan_id,
resource_id=resource.resource_id,
original_path=resource.location,
sandbox_strategy=cast(SandboxStrategyStr, strategy_str),
)
if sandbox.context is None:
raise RuntimeError(
f"Sandbox for resource '{resource.resource_id}' "
f"(plan={plan_id}) was created but has no context"
)
return BoundResource(
slot_name=slot_name,
resource_id=resource.resource_id,
resource_type=resource.resource_type_name,
sandbox_path=sandbox.context.sandbox_path,
access=access,
)
src/cleveragents/domain/models/core/decision.py
+490
View File
@@ -0,0 +1,490 @@
"""Decision domain models for the decision tree subsystem.
A :class:`Decision` represents a persisted choice point in a plan's
decision tree, created during the Strategize or Execute phases. Each
decision records the question being answered, the chosen option,
alternatives considered, a confidence score, rationale, a
:class:`ContextSnapshot` for replay, and downstream dependency links.
Decision types
--------------
.. list-table::
:header-rows: 1
* - Type
- Phase
- Description
* - ``prompt_definition``
- Strategize
- Root decision — the plan prompt
* - ``invariant_enforced``
- Strategize
- An invariant constraint was applied
* - ``strategy_choice``
- Strategize
- High-level approach chosen
* - ``implementation_choice``
- Execute
- How to implement a specific task
* - ``resource_selection``
- Execute
- Which resources to read / modify
* - ``subplan_spawn``
- Strategize
- Decision to create a child plan
* - ``subplan_parallel_spawn``
- Strategize
- Spawn a group of child plans in parallel
* - ``tool_invocation``
- Execute
- Which skill / tool to use
* - ``error_recovery``
- Execute
- How to handle a failure
* - ``validation_response``
- Execute
- Response to a validation failure
* - ``user_intervention``
- Any
- User-provided guidance / correction
Context snapshots
-----------------
Every decision captures a :class:`ContextSnapshot` sufficient to
replay the decision from the same informational starting point. The
snapshot includes a hash of the hot context window, a storage
reference for the full context, a list of :class:`ResourceRef`
entries, and the LangGraph actor checkpoint reference.
Based on:
- docs/specification.md L18390-L18521
- docs/adr/ADR-007-decision-tree-and-correction.md
- docs/adr/ADR-033-decision-recording-protocol.md
- docs/adr/ADR-034-decision-tree-versioning-and-history.md
"""
from __future__ import annotations
import re
from datetime import UTC, datetime
from enum import StrEnum
from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
from ulid import ULID
# ---------------------------------------------------------------------------
# Constants
# ---------------------------------------------------------------------------
ULID_PATTERN = r"^[0-9A-HJKMNP-TV-Z]{26}$"
_ULID_RE = re.compile(ULID_PATTERN)
# ---------------------------------------------------------------------------
# Enums
# ---------------------------------------------------------------------------
class DecisionType(StrEnum):
"""Classification of a decision node in the plan decision tree.
Each type is associated with one or more plan phases and constrains
what downstream decisions are permitted.
"""
PROMPT_DEFINITION = "prompt_definition"
INVARIANT_ENFORCED = "invariant_enforced"
STRATEGY_CHOICE = "strategy_choice"
IMPLEMENTATION_CHOICE = "implementation_choice"
RESOURCE_SELECTION = "resource_selection"
SUBPLAN_SPAWN = "subplan_spawn"
SUBPLAN_PARALLEL_SPAWN = "subplan_parallel_spawn"
TOOL_INVOCATION = "tool_invocation"
ERROR_RECOVERY = "error_recovery"
VALIDATION_RESPONSE = "validation_response"
USER_INTERVENTION = "user_intervention"
#: Decision types that may only be created during the Strategize phase.
STRATEGIZE_TYPES: frozenset[DecisionType] = frozenset(
{
DecisionType.PROMPT_DEFINITION,
DecisionType.INVARIANT_ENFORCED,
DecisionType.STRATEGY_CHOICE,
DecisionType.SUBPLAN_SPAWN,
DecisionType.SUBPLAN_PARALLEL_SPAWN,
}
)
#: Decision types that may only be created during the Execute phase.
EXECUTE_TYPES: frozenset[DecisionType] = frozenset(
{
DecisionType.IMPLEMENTATION_CHOICE,
DecisionType.RESOURCE_SELECTION,
DecisionType.TOOL_INVOCATION,
DecisionType.ERROR_RECOVERY,
DecisionType.VALIDATION_RESPONSE,
}
)
# ---------------------------------------------------------------------------
# Embedded value objects
# ---------------------------------------------------------------------------
class ResourceRef(BaseModel):
"""Reference to a resource that influenced a decision.
Captures the resource identifier and an optional path / symbol
within that resource for fine-grained provenance tracking.
"""
model_config = ConfigDict(frozen=True, str_strip_whitespace=True)
resource_id: str = Field(
...,
description="ULID of the referenced resource.",
)
path: str = Field(
default="",
description="Optional sub-path or symbol within the resource.",
)
@field_validator("resource_id")
@classmethod
def _resource_id_not_empty(cls, v: str) -> str:
if not v or not v.strip():
raise ValueError("resource_id must not be empty")
return v
class ArtifactRef(BaseModel):
"""Reference to an artifact produced by a decision.
Artifacts are files, patches, or outputs created as a side-effect
of executing a decision.
"""
model_config = ConfigDict(frozen=True, str_strip_whitespace=True)
artifact_path: str = Field(
...,
description="Path to the artifact (relative to workspace root).",
)
artifact_type: str = Field(
default="file",
description="Kind of artifact: file, patch, log, etc.",
)
@field_validator("artifact_path")
@classmethod
def _artifact_path_not_empty(cls, v: str) -> str:
if not v or not v.strip():
raise ValueError("artifact_path must not be empty")
return v
class ContextSnapshot(BaseModel):
"""Snapshot of the informational context at decision time.
Captures enough state to replay the decision from the same
starting point. The ``hot_context_hash`` is a cryptographic hash
of the exact context window; ``hot_context_ref`` is a storage
pointer to the full serialised context.
See ADR-033 §Context Snapshot Auto-Capture for the recording
protocol.
"""
model_config = ConfigDict(frozen=True, str_strip_whitespace=True)
hot_context_hash: str = Field(
default="",
description="Cryptographic hash of the hot context window.",
)
hot_context_ref: str = Field(
default="",
description="Storage pointer to the full context snapshot.",
)
relevant_resources: list[ResourceRef] = Field(
default_factory=list,
description="Resources that influenced this decision.",
)
actor_state_ref: str = Field(
default="",
description="LangGraph actor checkpoint reference.",
)
# ---------------------------------------------------------------------------
# Decision model
# ---------------------------------------------------------------------------
class Decision(BaseModel):
"""A persisted choice point in a plan's decision tree.
Decisions form a tree (via ``parent_decision_id``) and an influence
DAG (via ``downstream_decision_ids``). The ``prompt_definition``
type is always the root of the tree and must have no parent.
Immutable once created — corrections create *new* decisions and
set ``superseded_by`` on the original.
.. note::
Because the model is frozen, ``superseded_by`` cannot be set
via attribute assignment. Use :meth:`with_superseded_by` to
obtain a copy with the field populated. The persistence layer
is responsible for writing the updated copy back to storage.
"""
model_config = ConfigDict(
frozen=True,
str_strip_whitespace=True,
use_enum_values=False,
)
# --- Identity ---
decision_id: str = Field(
default_factory=lambda: str(ULID()),
description="Unique decision identifier (ULID).",
pattern=ULID_PATTERN,
)
plan_id: str = Field(
...,
description="ULID of the parent plan.",
pattern=ULID_PATTERN,
)
parent_decision_id: str | None = Field(
default=None,
description="ULID of the parent decision (None for root).",
)
sequence_number: int = Field(
...,
ge=0,
description=(
"Monotonic order within the plan's decisions. "
"Uniqueness within a plan is enforced at the persistence "
"layer, not the domain model."
),
)
# --- Classification ---
decision_type: DecisionType = Field(
...,
description="The kind of decision being recorded.",
)
# --- The decision itself ---
question: str = Field(
...,
min_length=1,
description="What question was being answered.",
)
chosen_option: str = Field(
...,
min_length=1,
description="The option that was chosen.",
)
alternatives_considered: list[str] = Field(
default_factory=list,
description="Other options that were evaluated.",
)
confidence_score: float | None = Field(
default=None,
ge=0.0,
le=1.0,
description="Confidence in the decision (0.0-1.0).",
)
# --- Context snapshot ---
context_snapshot: ContextSnapshot = Field(
default_factory=ContextSnapshot,
description="Informational context at decision time.",
)
# --- Rationale ---
rationale: str = Field(
default="",
description="Human-readable rationale for the decision.",
)
actor_reasoning: str | None = Field(
default=None,
description="Raw LLM reasoning trace, if available.",
)
# --- Downstream impact ---
downstream_decision_ids: list[str] = Field(
default_factory=list,
description="ULIDs of decisions that depend on this one.",
)
downstream_plan_ids: list[str] = Field(
default_factory=list,
description="ULIDs of child plans spawned from this decision.",
)
artifacts_produced: list[ArtifactRef] = Field(
default_factory=list,
description="Artifacts created as side-effects.",
)
# --- Timestamps ---
created_at: datetime = Field(
default_factory=lambda: datetime.now(UTC),
description="When the decision was recorded.",
)
# --- Correction metadata ---
is_correction: bool = Field(
default=False,
description="Whether this decision is a correction of another.",
)
corrects_decision_id: str | None = Field(
default=None,
description="ULID of the decision being corrected.",
)
correction_reason: str | None = Field(
default=None,
description="Why the correction was made.",
)
superseded_by: str | None = Field(
default=None,
description="ULID of the decision that supersedes this one.",
)
# --- Validators ---
@field_validator("plan_id")
@classmethod
def _plan_id_valid(cls, v: str) -> str:
if not _ULID_RE.match(v):
raise ValueError(f"plan_id must be a valid ULID, got '{v}'")
return v
@field_validator("parent_decision_id")
@classmethod
def _parent_decision_id_valid(cls, v: str | None) -> str | None:
if v is not None and not _ULID_RE.match(v):
raise ValueError(
f"parent_decision_id must be a valid ULID or None, got '{v}'"
)
return v
@field_validator("corrects_decision_id")
@classmethod
def _corrects_decision_id_valid(cls, v: str | None) -> str | None:
if v is not None and not _ULID_RE.match(v):
raise ValueError(
f"corrects_decision_id must be a valid ULID or None, got '{v}'"
)
return v
@field_validator("superseded_by")
@classmethod
def _superseded_by_valid(cls, v: str | None) -> str | None:
if v is not None and not _ULID_RE.match(v):
raise ValueError(f"superseded_by must be a valid ULID or None, got '{v}'")
return v
@model_validator(mode="after")
def _root_decision_constraints(self) -> Decision:
"""Ensure prompt_definition decisions have no parent."""
if (
self.decision_type == DecisionType.PROMPT_DEFINITION
and self.parent_decision_id is not None
):
raise ValueError(
"prompt_definition decisions must be the tree root "
"(parent_decision_id must be None)"
)
return self
@model_validator(mode="after")
def _correction_fields_consistent(self) -> Decision:
"""Ensure correction metadata is internally consistent."""
if self.is_correction and not self.corrects_decision_id:
raise ValueError(
"is_correction is True but corrects_decision_id is not set"
)
if self.corrects_decision_id and not self.is_correction:
raise ValueError("corrects_decision_id is set but is_correction is False")
return self
# --- Domain helpers ---
@property
def is_root(self) -> bool:
"""Return True if this is the root of the decision tree."""
return self.parent_decision_id is None
@property
def is_superseded(self) -> bool:
"""Return True if this decision has been superseded by a correction."""
return self.superseded_by is not None
@property
def is_strategize_type(self) -> bool:
"""Return True if this is a Strategize-phase decision type."""
return self.decision_type in STRATEGIZE_TYPES
@property
def is_execute_type(self) -> bool:
"""Return True if this is an Execute-phase decision type."""
return self.decision_type in EXECUTE_TYPES
@property
def is_any_phase_type(self) -> bool:
"""Return True if this decision type is valid in any phase."""
return self.decision_type == DecisionType.USER_INTERVENTION
def as_cli_dict(self) -> dict[str, object]:
"""Return a stable dict for CLI table rendering."""
return {
"decision_id": self.decision_id,
"plan_id": self.plan_id,
"type": str(self.decision_type),
"sequence": self.sequence_number,
"question": self.question,
"chosen": self.chosen_option,
"confidence": self.confidence_score,
"parent": self.parent_decision_id or "(root)",
"is_correction": self.is_correction,
"superseded": self.is_superseded,
}
def with_superseded_by(self, new_decision_id: str) -> Decision:
"""Return a copy with ``superseded_by`` set.
Because the model is frozen, this is the canonical way to mark
a decision as superseded during the correction lifecycle.
Args:
new_decision_id: ULID of the decision that supersedes this one.
Returns:
A new :class:`Decision` instance identical to this one
except with ``superseded_by`` populated.
"""
return self.model_copy(update={"superseded_by": new_decision_id})
__all__ = [
"EXECUTE_TYPES",
"STRATEGIZE_TYPES",
"ArtifactRef",
"ContextSnapshot",
"Decision",
"DecisionType",
"ResourceRef",
]
src/cleveragents/resource/handlers/__init__.py
+43
View File
@@ -0,0 +1,43 @@
"""Resource handler implementations for CleverAgents.
This package provides the handler runtime that bridges resource types
to sandbox provisioning. Each handler resolves a :class:`Resource`
into a sandboxed working path via the :class:`SandboxManager`.
## Handler Protocol
All handlers implement
:class:`~cleveragents.resource.handlers.protocol.ResourceHandler`,
which defines a single ``resolve`` method returning a
:class:`~cleveragents.tool.context.BoundResource` with a populated
``sandbox_path``.
## Built-in Handlers
| Handler | Resource Type | Sandbox Strategy |
|-----------------------|------------------|------------------|
| ``GitCheckoutHandler``| ``git-checkout`` | ``git_worktree`` |
| ``FsDirectoryHandler``| ``fs-directory`` | ``copy_on_write``|
## Handler Resolution
Handler strings stored on :class:`ResourceTypeSpec` use the format
``module.path:ClassName``. The :func:`resolve_handler` function
dynamically imports the module and returns an instance.
"""
from cleveragents.resource.handlers.fs_directory import FsDirectoryHandler
from cleveragents.resource.handlers.git_checkout import GitCheckoutHandler
from cleveragents.resource.handlers.protocol import ResourceHandler
from cleveragents.resource.handlers.resolver import (
HandlerResolutionError,
resolve_handler,
)
__all__ = [
"FsDirectoryHandler",
"GitCheckoutHandler",
"HandlerResolutionError",
"ResourceHandler",
"resolve_handler",
]
src/cleveragents/resource/handlers/_base.py
+99
View File
@@ -0,0 +1,99 @@
"""Shared base class for resource handlers.
Provides the common ``resolve`` logic used by both
:class:`GitCheckoutHandler` and :class:`FsDirectoryHandler`.
Subclasses set ``_default_strategy`` and ``_type_label`` to
customise behaviour.
"""
from __future__ import annotations
import logging
from typing import ClassVar, cast
from cleveragents.domain.models.core.resource import Resource, SandboxStrategy
from cleveragents.infrastructure.sandbox.factory import SandboxStrategyStr
from cleveragents.infrastructure.sandbox.manager import SandboxManager
from cleveragents.tool.context import BoundResource
logger = logging.getLogger(__name__)
class BaseResourceHandler:
"""Base handler with shared resolve logic.
Subclasses **must** set:
* ``_default_strategy`` — the :class:`SandboxStrategy` to use when
the resource has no explicit override.
* ``_type_label`` — a human-readable label for error messages
(e.g. ``"git-checkout"``).
"""
_default_strategy: ClassVar[SandboxStrategy]
_type_label: ClassVar[str]
def resolve(
self,
*,
resource: Resource,
plan_id: str,
slot_name: str,
sandbox_manager: SandboxManager,
access: str = "read_only",
) -> BoundResource:
"""Resolve a resource into a sandboxed BoundResource.
Args:
resource: A resource with ``location`` pointing to the
original path.
plan_id: The plan requesting the sandbox.
slot_name: Name of the tool resource slot being filled.
sandbox_manager: The sandbox lifecycle manager.
access: Access mode (``read_only`` or ``read_write``).
Returns:
A :class:`BoundResource` with ``sandbox_path`` populated.
Raises:
ValueError: If the resource has no location.
RuntimeError: If the sandbox was created but has no context.
"""
if not resource.location:
raise ValueError(
f"{self._type_label} resource '{resource.resource_id}' has no location"
)
strategy_raw = resource.sandbox_strategy or self._default_strategy
strategy_str = (
strategy_raw.value if hasattr(strategy_raw, "value") else str(strategy_raw)
)
logger.debug(
"Resolving %s resource %s (location=%s, strategy=%s)",
self._type_label,
resource.resource_id,
resource.location,
strategy_str,
)
sandbox = sandbox_manager.get_or_create_sandbox(
plan_id=plan_id,
resource_id=resource.resource_id,
original_path=resource.location,
sandbox_strategy=cast(SandboxStrategyStr, strategy_str),
)
if sandbox.context is None:
raise RuntimeError(
f"Sandbox for resource '{resource.resource_id}' "
f"(plan={plan_id}) was created but has no context"
)
return BoundResource(
slot_name=slot_name,
resource_id=resource.resource_id,
resource_type=resource.resource_type_name,
sandbox_path=sandbox.context.sandbox_path,
access=access,
)
src/cleveragents/resource/handlers/fs_directory.py
+35
View File
@@ -0,0 +1,35 @@
"""Filesystem-directory resource handler.
Resolves ``fs-directory`` resources into sandbox-backed
:class:`BoundResource` instances using the ``copy_on_write`` sandbox
strategy.
The handler:
1. Validates that the resource has a non-empty ``location``.
2. Determines the sandbox strategy: resource-level override takes
precedence over the default ``copy_on_write``.
3. Calls :meth:`SandboxManager.get_or_create_sandbox` to provision
(or reuse) an isolated copy-on-write directory.
4. Returns a :class:`BoundResource` with ``sandbox_path`` set to the
sandbox root.
Based on:
- implementation_plan.md group M1.resource-handlers (L2254-L2271)
- Built-in type definition in resource_registry_service.py L99-123
"""
from __future__ import annotations
from cleveragents.domain.models.core.resource import SandboxStrategy
from cleveragents.resource.handlers._base import BaseResourceHandler
class FsDirectoryHandler(BaseResourceHandler):
"""Handler for ``fs-directory`` resource types.
Provisions a copy-on-write sandbox for a local filesystem directory.
"""
_default_strategy = SandboxStrategy.COPY_ON_WRITE
_type_label = "fs-directory"
src/cleveragents/resource/handlers/git_checkout.py
+36
View File
@@ -0,0 +1,36 @@
"""Git-checkout resource handler.
Resolves ``git-checkout`` resources into sandbox-backed
:class:`BoundResource` instances using the ``git_worktree`` sandbox
strategy (with fallback to ``copy_on_write``).
The handler:
1. Validates that the resource has a non-empty ``location``.
2. Determines the sandbox strategy: resource-level override takes
precedence over the default ``git_worktree``.
3. Calls :meth:`SandboxManager.get_or_create_sandbox` to provision
(or reuse) an isolated git worktree.
4. Returns a :class:`BoundResource` with ``sandbox_path`` set to the
sandbox root.
Based on:
- implementation_plan.md group M1.resource-handlers (L2254-L2271)
- Built-in type definition in resource_registry_service.py L62-98
"""
from __future__ import annotations
from cleveragents.domain.models.core.resource import SandboxStrategy
from cleveragents.resource.handlers._base import BaseResourceHandler
class GitCheckoutHandler(BaseResourceHandler):
"""Handler for ``git-checkout`` resource types.
Provisions a git-worktree sandbox (or copy-on-write fallback)
for a git repository checkout.
"""
_default_strategy = SandboxStrategy.GIT_WORKTREE
_type_label = "git-checkout"
src/cleveragents/resource/handlers/protocol.py
+78
View File
@@ -0,0 +1,78 @@
"""Resource handler protocol for CleverAgents.
Defines the :class:`ResourceHandler` protocol that all resource type
handlers must satisfy. A handler bridges a :class:`Resource` domain
object to sandbox provisioning by:
1. Reading the resource's ``location`` (the original filesystem path).
2. Determining the sandbox strategy (from resource override or type default).
3. Calling :meth:`SandboxManager.get_or_create_sandbox` to provision an
isolated working directory.
4. Returning a :class:`BoundResource` with ``sandbox_path`` populated.
Based on:
- implementation_plan.md group M1.resource-handlers (L2254-L2271)
- docs/specification.md Resource Handler architecture
"""
from __future__ import annotations
from typing import Protocol, runtime_checkable
from cleveragents.domain.models.core.resource import Resource
from cleveragents.infrastructure.sandbox.manager import SandboxManager
from cleveragents.tool.context import BoundResource
@runtime_checkable
class ResourceHandler(Protocol):
"""Protocol for resource type handlers.
Each handler knows how to resolve a specific resource type into a
sandboxed working path. Implementations are registered via the
``handler`` field on :class:`ResourceTypeSpec` using the
``module:ClassName`` string format.
Lifecycle::
handler = resolve_handler(
"cleveragents.resource.handlers.git_checkout:GitCheckoutHandler"
)
bound = handler.resolve(
resource=resource,
plan_id="01ARZ3...",
slot_name="repo",
sandbox_manager=sandbox_manager,
access="read_write",
)
# bound.sandbox_path is now populated
"""
def resolve(
self,
*,
resource: Resource,
plan_id: str,
slot_name: str,
sandbox_manager: SandboxManager,
access: str = "read_only",
) -> BoundResource:
"""Resolve a resource into a sandbox-backed BoundResource.
Args:
resource: The resource domain object to resolve.
plan_id: The plan requesting the sandbox.
slot_name: Name of the tool resource slot being filled.
sandbox_manager: The sandbox lifecycle manager.
access: Access mode (``read_only`` or ``read_write``).
Returns:
A :class:`BoundResource` with ``sandbox_path`` populated
from the provisioned sandbox.
Raises:
ValueError: If the resource lacks a location or has an
incompatible type.
SandboxError: If sandbox creation fails.
"""
...
src/cleveragents/resource/handlers/resolver.py
+120
View File
@@ -0,0 +1,120 @@
"""Dynamic handler resolution from ``module:ClassName`` strings.
The :func:`resolve_handler` function takes a handler reference string
(as stored on :class:`ResourceTypeSpec`) and returns an instantiated
handler object that satisfies the :class:`ResourceHandler` protocol.
Format::
"cleveragents.resource.handlers.git_checkout:GitCheckoutHandler"
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ ^^^^^^^^^^^^^^^^^^^^^
module path class name
The function uses :func:`importlib.import_module` for dynamic loading,
with a cache to avoid repeated imports.
Based on:
- implementation_plan.md group M1.resource-handlers (L2254-L2271)
"""
from __future__ import annotations
import importlib
import logging
import threading
from typing import Any
from cleveragents.resource.handlers.protocol import ResourceHandler
logger = logging.getLogger(__name__)
# Cache of already-resolved handler instances keyed by reference string
_handler_cache: dict[str, ResourceHandler] = {}
_cache_lock: threading.Lock = threading.Lock()
class HandlerResolutionError(Exception):
"""Raised when a handler reference string cannot be resolved."""
def resolve_handler(handler_ref: str) -> ResourceHandler:
"""Resolve a handler reference string to an instance.
Args:
handler_ref: A string in ``module.path:ClassName`` format.
Returns:
An instantiated handler that satisfies :class:`ResourceHandler`.
Raises:
HandlerResolutionError: If the reference is malformed, the
module cannot be imported, or the class does not exist.
"""
if not handler_ref or not handler_ref.strip():
raise HandlerResolutionError("Handler reference must not be empty")
handler_ref = handler_ref.strip()
# Return cached instance if available
with _cache_lock:
if handler_ref in _handler_cache:
return _handler_cache[handler_ref]
# Parse module:class format
if ":" not in handler_ref:
raise HandlerResolutionError(
f"Invalid handler reference format '{handler_ref}': "
"expected 'module.path:ClassName'"
)
module_path, class_name = handler_ref.rsplit(":", 1)
if not module_path or not class_name:
raise HandlerResolutionError(
f"Invalid handler reference '{handler_ref}': "
"both module path and class name are required"
)
# Import module
try:
module = importlib.import_module(module_path)
except ModuleNotFoundError as exc:
raise HandlerResolutionError(
f"Cannot import handler module '{module_path}': {exc}"
) from exc
except Exception as exc:
raise HandlerResolutionError(
f"Error importing handler module '{module_path}': {exc}"
) from exc
# Get class
handler_cls: Any = getattr(module, class_name, None)
if handler_cls is None:
raise HandlerResolutionError(
f"Handler class '{class_name}' not found in module '{module_path}'"
)
# Instantiate
try:
instance = handler_cls()
except Exception as exc:
raise HandlerResolutionError(
f"Cannot instantiate handler '{handler_ref}': {exc}"
) from exc
# Validate protocol conformance
if not isinstance(instance, ResourceHandler):
raise HandlerResolutionError(
f"Handler '{handler_ref}' does not satisfy ResourceHandler protocol"
)
# Cache and return
with _cache_lock:
_handler_cache[handler_ref] = instance
logger.debug("Resolved handler: %s", handler_ref)
return instance
def clear_handler_cache() -> None:
"""Clear the handler instance cache (useful for testing)."""
with _cache_lock:
_handler_cache.clear()