hamza.khyari
bdd1ea4f3a
CI / push-validation (pull_request) Successful in 10s
CI / build (pull_request) Successful in 16s
CI / helm (pull_request) Successful in 16s
CI / typecheck (pull_request) Successful in 30s
CI / lint (pull_request) Successful in 35s
CI / quality (pull_request) Successful in 35s
CI / security (pull_request) Successful in 50s
CI / integration_tests (pull_request) Successful in 4m3s
CI / e2e_tests (pull_request) Successful in 4m14s
CI / unit_tests (pull_request) Successful in 5m13s
CI / docker (pull_request) Successful in 8s
CI / coverage (pull_request) Successful in 11m2s
CI / status-check (pull_request) Successful in 1s
CheckpointManager was never wired into PlanExecutor — the CLI factory constructed PlanExecutor without a checkpoint_manager (defaulted to None), silently skipping all checkpoint hooks. Fix: - Register CheckpointManager as Singleton in DI container - Resolve container singleton in _get_plan_executor() and pass to PlanExecutor constructor - Bridge infra→domain: _try_create_checkpoint() now persists last_checkpoint_id on the plan via _commit_plan(), raises PlanError if persistence fails - Default checkpointable=True for writable+sandboxable resources and write-capable tools (model_validators on ResourceCapabilities and ToolCapability) - Validate that non-writable/non-sandboxable resources cannot be checkpointable (ValueError guard) - Add post-execute A2A facade notification using plan.status to avoid duplicate execute→execute transition errors Tests: - 10 Behave scenarios covering DI wiring, singleton identity, checkpoint creation, plan metadata update, rollback, graceful fallback, no-arg constructor, capability defaults (positive + 2 negative) - Updated consolidated_resource, consolidated_skill, and Robot helper_skill_flatten for new checkpointable defaults ISSUES CLOSED: #1253
298 lines
10 KiB
Python
298 lines
10 KiB
Python
"""Step definitions for checkpoint_wiring.feature."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
from behave import given, then, when
|
|
|
|
from cleveragents.infrastructure.sandbox.checkpoint import (
|
|
CheckpointManager,
|
|
SandboxCheckpoint,
|
|
)
|
|
|
|
# ── CLI factory ──────────────────────────────────────────
|
|
|
|
|
|
@when("I build a PlanExecutor via _get_plan_executor for ckpt_wire")
|
|
def step_build_executor(context: object) -> None:
|
|
from cleveragents.cli.commands.plan import _get_plan_executor
|
|
|
|
executor = _get_plan_executor()
|
|
context.ckpt_executor = executor
|
|
|
|
|
|
@then("the executor checkpoint_manager should not be None for ckpt_wire")
|
|
def step_executor_has_mgr(context: object) -> None:
|
|
executor = context.ckpt_executor
|
|
assert executor._checkpoint_manager is not None, (
|
|
"PlanExecutor._checkpoint_manager is None -- not wired"
|
|
)
|
|
|
|
|
|
# ── DI container ─────────────────────────────────────────
|
|
|
|
|
|
@when("I get CheckpointManager from the DI container for ckpt_wire")
|
|
def step_container_mgr(context: object) -> None:
|
|
from cleveragents.application.container import get_container
|
|
|
|
container = get_container()
|
|
context.ckpt_container_mgr = container.checkpoint_manager()
|
|
|
|
|
|
@then("the container checkpoint_manager should not be None for ckpt_wire")
|
|
def step_container_has_mgr(context: object) -> None:
|
|
mgr = context.ckpt_container_mgr
|
|
assert mgr is not None
|
|
assert isinstance(mgr, CheckpointManager)
|
|
|
|
|
|
@then("the executor checkpoint_manager should be the container singleton for ckpt_wire")
|
|
def step_executor_matches_container(context: object) -> None:
|
|
executor = context.ckpt_executor
|
|
container_mgr = context.ckpt_container_mgr
|
|
assert executor._checkpoint_manager is container_mgr, (
|
|
"PlanExecutor did not reuse container CheckpointManager singleton"
|
|
)
|
|
|
|
|
|
# ── _try_create_checkpoint with real manager ─────────────
|
|
|
|
|
|
@given("a PlanExecutor with a real CheckpointManager for ckpt_wire")
|
|
def step_executor_with_mgr(context: object) -> None:
|
|
from cleveragents.application.services.plan_executor import PlanExecutor
|
|
from cleveragents.application.services.plan_lifecycle_service import (
|
|
PlanLifecycleService,
|
|
)
|
|
from cleveragents.config.settings import Settings
|
|
|
|
lifecycle = PlanLifecycleService(settings=Settings())
|
|
mgr = CheckpointManager()
|
|
executor = PlanExecutor(
|
|
lifecycle_service=lifecycle,
|
|
checkpoint_manager=mgr,
|
|
)
|
|
context.ckpt_executor = executor
|
|
context.ckpt_lifecycle = lifecycle
|
|
context.ckpt_manager = mgr
|
|
context.ckpt_plan_id = "plan-ckpt-test-01"
|
|
|
|
|
|
@given("a sandbox exists for the test plan for ckpt_wire")
|
|
def step_sandbox_exists(context: object) -> None:
|
|
from cleveragents.infrastructure.sandbox.factory import SandboxFactory
|
|
from cleveragents.infrastructure.sandbox.manager import SandboxManager
|
|
|
|
plan_id = context.ckpt_plan_id
|
|
tmpdir = tempfile.mkdtemp(prefix="ckpt-wire-")
|
|
source_file = Path(tmpdir, "test.txt")
|
|
initial_content = "checkpoint test"
|
|
source_file.write_text(initial_content)
|
|
|
|
factory = SandboxFactory()
|
|
mgr = SandboxManager(factory=factory, cleanup_on_exit=False)
|
|
sandbox = mgr.get_or_create_sandbox(
|
|
plan_id=plan_id,
|
|
resource_id="res-ckpt-test",
|
|
original_path=tmpdir,
|
|
sandbox_strategy="copy_on_write",
|
|
)
|
|
context.ckpt_sandbox = sandbox
|
|
|
|
sandbox_path = Path(sandbox.context.sandbox_path)
|
|
sandbox_path.mkdir(parents=True, exist_ok=True)
|
|
sandbox_file = sandbox_path / "test.txt"
|
|
if not sandbox_file.exists():
|
|
sandbox_file.write_text(initial_content)
|
|
context.ckpt_sandbox_file = sandbox_file
|
|
context.ckpt_original_content = sandbox_file.read_text()
|
|
|
|
# Patch the executor's sandbox resolver to return our sandbox
|
|
executor = context.ckpt_executor
|
|
executor._resolve_sandbox_for_checkpoint = lambda pid: sandbox
|
|
|
|
|
|
@given("a plan exists in the lifecycle service for ckpt_wire")
|
|
def step_plan_exists(context: object) -> None:
|
|
from cleveragents.domain.models.core.plan import ProjectLink
|
|
|
|
lifecycle = context.ckpt_lifecycle
|
|
lifecycle.create_action(
|
|
name="local/ckpt-wire-action",
|
|
description="Checkpoint wiring test",
|
|
definition_of_done="Checkpoint created",
|
|
strategy_actor="local/stub",
|
|
execution_actor="local/stub",
|
|
)
|
|
plan = lifecycle.use_action(
|
|
action_name="local/ckpt-wire-action",
|
|
project_links=[ProjectLink(project_name="test")],
|
|
)
|
|
context.ckpt_plan_id = plan.identity.plan_id
|
|
|
|
# Re-patch resolver with correct plan_id
|
|
executor = context.ckpt_executor
|
|
sandbox = getattr(context, "ckpt_sandbox", None)
|
|
if sandbox is not None:
|
|
executor._resolve_sandbox_for_checkpoint = lambda pid: sandbox
|
|
|
|
|
|
@when("I call _try_create_checkpoint for ckpt_wire")
|
|
def step_try_checkpoint(context: object) -> None:
|
|
executor = context.ckpt_executor
|
|
plan_id = context.ckpt_plan_id
|
|
result = executor._try_create_checkpoint(plan_id, "test_phase")
|
|
context.ckpt_result = result
|
|
|
|
|
|
@then("a SandboxCheckpoint should be returned for ckpt_wire")
|
|
def step_checkpoint_returned(context: object) -> None:
|
|
result = context.ckpt_result
|
|
assert result is not None, "Expected SandboxCheckpoint, got None"
|
|
assert isinstance(result, SandboxCheckpoint)
|
|
|
|
|
|
@then("None should be returned for ckpt_wire")
|
|
def step_none_returned(context: object) -> None:
|
|
result = context.ckpt_result
|
|
assert result is None
|
|
|
|
|
|
@then("the plan last_checkpoint_id should not be None for ckpt_wire")
|
|
def step_plan_has_checkpoint(context: object) -> None:
|
|
lifecycle = context.ckpt_lifecycle
|
|
plan_id = context.ckpt_plan_id
|
|
plan = lifecycle.get_plan(plan_id)
|
|
assert plan.last_checkpoint_id is not None, (
|
|
f"plan.last_checkpoint_id is None for plan {plan_id}"
|
|
)
|
|
|
|
|
|
@when("I mutate the sandbox file for ckpt_wire")
|
|
def step_mutate_sandbox(context: object) -> None:
|
|
file_path = context.ckpt_sandbox_file
|
|
mutated_content = "checkpoint test mutated"
|
|
file_path.write_text(mutated_content)
|
|
context.ckpt_mutated_content = mutated_content
|
|
|
|
|
|
@when("I rollback to the last checkpoint via the executor for ckpt_wire")
|
|
def step_rollback_to_checkpoint(context: object) -> None:
|
|
executor = context.ckpt_executor
|
|
plan_id = context.ckpt_plan_id
|
|
success = executor._try_rollback_to_last_checkpoint(plan_id)
|
|
context.ckpt_rollback_success = success
|
|
|
|
|
|
@then("rollback should succeed for ckpt_wire")
|
|
def step_verify_rollback_success(context: object) -> None:
|
|
assert context.ckpt_rollback_success is True, "Rollback did not succeed"
|
|
|
|
|
|
@then("the sandbox file should match the original contents for ckpt_wire")
|
|
def step_verify_sandbox_restored(context: object) -> None:
|
|
file_path = context.ckpt_sandbox_file
|
|
assert file_path.read_text() == context.ckpt_original_content, (
|
|
"Sandbox contents were not restored from checkpoint"
|
|
)
|
|
|
|
|
|
# ── PlanExecutor with no manager ─────────────────────────
|
|
|
|
|
|
@given("a PlanExecutor with no CheckpointManager for ckpt_wire")
|
|
def step_executor_no_mgr(context: object) -> None:
|
|
from cleveragents.application.services.plan_executor import PlanExecutor
|
|
from cleveragents.application.services.plan_lifecycle_service import (
|
|
PlanLifecycleService,
|
|
)
|
|
from cleveragents.config.settings import Settings
|
|
|
|
executor = PlanExecutor(
|
|
lifecycle_service=PlanLifecycleService(settings=Settings()),
|
|
)
|
|
context.ckpt_executor = executor
|
|
context.ckpt_plan_id = "plan-no-mgr"
|
|
|
|
|
|
# ── CheckpointManager no-arg construction ────────────────
|
|
|
|
|
|
@when("I instantiate CheckpointManager with no args for ckpt_wire")
|
|
def step_instantiate_mgr(context: object) -> None:
|
|
context.ckpt_new_mgr = CheckpointManager()
|
|
|
|
|
|
@then("CheckpointManager should be created successfully for ckpt_wire")
|
|
def step_mgr_created(context: object) -> None:
|
|
mgr = context.ckpt_new_mgr
|
|
assert mgr is not None
|
|
assert isinstance(mgr, CheckpointManager)
|
|
|
|
|
|
# ── Default capability behaviour ─────────────────────────
|
|
|
|
|
|
@when("I create default capabilities for a writable resource for ckpt_wire")
|
|
def step_create_resource_capabilities(context: object) -> None:
|
|
from cleveragents.domain.models.core.resource import ResourceCapabilities
|
|
|
|
# Explicit flags to exercise the model_validator that derives
|
|
# checkpointable from writable AND sandboxable.
|
|
context.ckpt_resource_caps = ResourceCapabilities(
|
|
writable=True,
|
|
sandboxable=True,
|
|
)
|
|
|
|
|
|
@then("the resource capability should be checkpointable for ckpt_wire")
|
|
def step_resource_checkpointable(context: object) -> None:
|
|
capabilities = context.ckpt_resource_caps
|
|
assert capabilities.checkpointable is True
|
|
|
|
|
|
@when("I create default capabilities for a writable tool for ckpt_wire")
|
|
def step_create_tool_capabilities(context: object) -> None:
|
|
from cleveragents.domain.models.core.tool import ToolCapability
|
|
|
|
context.ckpt_tool_cap = ToolCapability(writes=True)
|
|
|
|
|
|
@then("the tool capability should be checkpointable for ckpt_wire")
|
|
def step_tool_checkpointable(context: object) -> None:
|
|
capability = context.ckpt_tool_cap
|
|
assert capability.checkpointable is True
|
|
|
|
|
|
@when("I create capabilities for a non-writable resource for ckpt_wire")
|
|
def step_create_non_writable_resource(context: object) -> None:
|
|
from cleveragents.domain.models.core.resource import ResourceCapabilities
|
|
|
|
context.ckpt_resource_caps = ResourceCapabilities(writable=False)
|
|
|
|
|
|
@then("the resource capability should not be checkpointable for ckpt_wire")
|
|
def step_resource_not_checkpointable(context: object) -> None:
|
|
capabilities = context.ckpt_resource_caps
|
|
assert capabilities.checkpointable is False, (
|
|
f"Expected checkpointable=False for non-writable resource, "
|
|
f"got {capabilities.checkpointable}"
|
|
)
|
|
|
|
|
|
@then(
|
|
"creating a non-writable resource with checkpointable True "
|
|
"should raise ValueError for ckpt_wire"
|
|
)
|
|
def step_non_writable_checkpointable_raises(context: object) -> None:
|
|
from cleveragents.domain.models.core.resource import ResourceCapabilities
|
|
|
|
try:
|
|
ResourceCapabilities(writable=False, checkpointable=True)
|
|
raise AssertionError("Expected ValueError but no exception was raised")
|
|
except ValueError:
|
|
pass # Expected
|