diff --git a/CHANGELOG.md b/CHANGELOG.md index 1d2b812af..595123305 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -54,6 +54,15 @@ The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/). message listing available built-in profiles. The resolved profile name is also logged at debug level for observability. +- **CheckpointManager rollback_to always returned False** (#7488): Fixed a data + integrity bug in `CheckpointManager.create_checkpoint()` where `sandbox_path` + was computed from `sandbox.context.sandbox_path` but never stored in the + checkpoint metadata. As a result, `rollback_to()` always found + `checkpoint.metadata.get("sandbox_path")` returning `None` and silently + skipped the rollback, returning `False`. The fix adds `sandbox_path` to the + metadata dict before constructing the `SandboxCheckpoint`, enabling + `rollback_to()` to correctly restore the sandbox filesystem state. + ### Added - **TDD Issue-Capture Test Activation** (#7025): Replaced 234 bare `@skip` tags diff --git a/CONTRIBUTORS.md b/CONTRIBUTORS.md index 0bda08851..67cfaa955 100644 --- a/CONTRIBUTORS.md +++ b/CONTRIBUTORS.md @@ -20,5 +20,6 @@ Below are some of the specific details of various contributions. * HAL 9000 has contributed the plan concurrency race-condition fix (#7989): wired `LockService` into the plan lifecycle, guarding `execute_plan()` and `apply_plan()` with plan-level advisory locks and unique per-invocation owner identities to prevent silent concurrent state corruption. * HAL 9000 has contributed the plugin entry point security hardening fix (#7476): enforced entry point allowlist validation before importing plugin modules to prevent malicious plugin loading. * Rui Hu has contributed the v3 actor YAML schema validation fix (#5869): added `ActorConfigSchema` validation to the `agents actor add --config` CLI command, covering cycle detection, required field validation, and enum validation for v3 YAML actor definitions. +* HAL 9000 has contributed automated bug fixes, including fix #7488 (store sandbox_path in checkpoint metadata to enable rollback). * This project was made possible thanks to considerable donation of time, money, and resources by CleverThis, Inc. * HAL 9000 has contributed automated bug fixes, CLI output formatting improvements, and ongoing maintenance as part of the CleverAgents automation system. diff --git a/features/checkpoint_manager_coverage.feature b/features/checkpoint_manager_coverage.feature index d4334cc58..ee185ae37 100644 --- a/features/checkpoint_manager_coverage.feature +++ b/features/checkpoint_manager_coverage.feature @@ -71,6 +71,23 @@ Feature: CheckpointManager full coverage When I invoke create_checkpoint with plan "plan-s1" phase "pre_execute" and no metadata Then the snapshot directory should contain copies of the sample files + Scenario: Creating a checkpoint with a real sandbox path stores sandbox_path in metadata + Given a freshly created CheckpointManager + And a temporary sandbox directory with sample files + And a mock sandbox whose context points to the temporary directory + When I invoke create_checkpoint with plan "plan-s2" phase "pre_execute" and no metadata + Then the returned checkpoint metadata should contain the sandbox path + + Scenario: Rollback succeeds without manually supplying sandbox_path in metadata + Given a freshly created CheckpointManager + And a temporary sandbox directory with sample files + And a mock sandbox whose context points to the temporary directory + And a checkpoint was created for the temporary sandbox without explicit sandbox_path metadata + And the temporary sandbox files are then modified + When I invoke rollback_to on the created checkpoint + Then the rollback result should be true + And the temporary sandbox should contain the original files + # --------------------------------------------------------------------------- # _snapshot_directory – branches # --------------------------------------------------------------------------- diff --git a/features/steps/checkpoint_manager_bug7488_steps.py b/features/steps/checkpoint_manager_bug7488_steps.py new file mode 100644 index 000000000..48be605dd --- /dev/null +++ b/features/steps/checkpoint_manager_bug7488_steps.py @@ -0,0 +1,59 @@ +"""Step definitions for Bug #7488 — sandbox_path auto-stored in checkpoint metadata. + +These steps verify that create_checkpoint() automatically stores sandbox_path +in metadata from sandbox.context.sandbox_path, enabling rollback_to() to work +without the caller manually supplying sandbox_path in metadata. + +Split from checkpoint_manager_coverage_steps.py to keep files under 500 lines. +""" + +from __future__ import annotations + +import os + +from behave import given, then +from behave.runner import Context + + +def _register_cleanup(context: Context, path: str) -> None: + """Schedule a directory for cleanup after the scenario.""" + if not hasattr(context, "_cleanup_handlers"): + context._cleanup_handlers = [] + context._cleanup_handlers.append( + lambda p=path: __import__("shutil").rmtree(p, ignore_errors=True) + ) + + +@then("the returned checkpoint metadata should contain the sandbox path") +def step_assert_metadata_has_sandbox_path(context: Context) -> None: + """Verify that create_checkpoint() automatically stores sandbox_path in metadata.""" + cp = context.cp_result + _register_cleanup(context, os.path.dirname(cp.snapshot_path)) + assert "sandbox_path" in cp.metadata, ( + "Expected 'sandbox_path' key in checkpoint metadata, but it was not found. " + "Bug #7488: create_checkpoint() must store sandbox_path in metadata." + ) + assert cp.metadata["sandbox_path"] == context.temp_sandbox, ( + f"Expected sandbox_path={context.temp_sandbox!r}, " + f"got {cp.metadata['sandbox_path']!r}" + ) + + +@given( + "a checkpoint was created for the temporary sandbox without explicit sandbox_path metadata" +) +def step_create_cp_for_temp_sandbox_no_explicit_path(context: Context) -> None: + """Create a checkpoint without manually supplying sandbox_path in metadata. + + After the fix for bug #7488, create_checkpoint() should automatically + store sandbox_path in metadata from sandbox.context.sandbox_path. + """ + context.cp_result = context.mgr.create_checkpoint( + sandbox=context.mock_sb, + plan_id="plan-rb-autopath", + phase="pre_execute", + # Intentionally NOT passing sandbox_path in metadata — the fix should + # auto-populate it from sandbox.context.sandbox_path. + ) + _register_cleanup(context, os.path.dirname(context.cp_result.snapshot_path)) + context.crafted_cp = context.cp_result diff --git a/features/steps/checkpoint_manager_coverage_steps.py b/features/steps/checkpoint_manager_coverage_steps.py index 038204418..d35803461 100644 --- a/features/steps/checkpoint_manager_coverage_steps.py +++ b/features/steps/checkpoint_manager_coverage_steps.py @@ -396,257 +396,13 @@ def step_cleanup_nonexistent(context: Context) -> None: # =========================================================================== -# rollback_to - all branches +# rollback_to, list_checkpoints, delete_checkpoint branches +# (step definitions moved to checkpoint_manager_rollback_steps.py to keep +# this file under the 500-line limit) # =========================================================================== - -@given("a SandboxCheckpoint with empty snapshot_path") -def step_cp_empty_snapshot(context: Context) -> None: - context.crafted_cp = SandboxCheckpoint( - checkpoint_id="cp-empty-snap", - sandbox_id="sb-rb1", - plan_id="plan-rb1", - phase="pre_execute", - created_at=datetime.now(), - metadata={}, - snapshot_path="", - ) - - -@when("I invoke rollback_to on the crafted checkpoint") -def step_rollback_crafted(context: Context) -> None: - if getattr(context, "rollback_should_patch_oserror", False): - # Patch os.listdir to raise on the second call (copy phase) - orig_listdir = os.listdir - call_count = {"n": 0} - - def _failing_listdir(path: str) -> list[str]: - call_count["n"] += 1 - if call_count["n"] >= 2: - raise OSError("simulated listdir failure") - return orig_listdir(path) - - with patch("os.listdir", side_effect=_failing_listdir): - context.rollback_result = context.mgr.rollback_to(context.crafted_cp) - else: - context.rollback_result = context.mgr.rollback_to(context.crafted_cp) - - -@then("the rollback result should be false") -def step_assert_rollback_false(context: Context) -> None: - assert context.rollback_result is False, "Expected rollback to return False" - - -@given("a SandboxCheckpoint with a non-existent snapshot_path") -def step_cp_nonexistent_snapshot(context: Context) -> None: - context.crafted_cp = SandboxCheckpoint( - checkpoint_id="cp-nosnap", - sandbox_id="sb-rb2", - plan_id="plan-rb2", - phase="pre_execute", - created_at=datetime.now(), - metadata={}, - snapshot_path="/nonexistent/snapshot/dir", - ) - - -@given("a SandboxCheckpoint whose snapshot exists but metadata has no sandbox_path") -def step_cp_no_sandbox_path_meta(context: Context) -> None: - # Create a real snapshot dir so the first guard passes - snap_parent = tempfile.mkdtemp(prefix="ca-rb-test-") - snap = os.path.join(snap_parent, "snapshot") - os.makedirs(snap, exist_ok=True) - _register_cleanup(context, snap_parent) - context.crafted_cp = SandboxCheckpoint( - checkpoint_id="cp-nometa", - sandbox_id="sb-rb3", - plan_id="plan-rb3", - phase="pre_execute", - created_at=datetime.now(), - metadata={}, # no sandbox_path key - snapshot_path=snap, - ) - - -@given("a SandboxCheckpoint whose snapshot exists but sandbox_path points nowhere") -def step_cp_sandbox_path_missing(context: Context) -> None: - snap_parent = tempfile.mkdtemp(prefix="ca-rb-test2-") - snap = os.path.join(snap_parent, "snapshot") - os.makedirs(snap, exist_ok=True) - _register_cleanup(context, snap_parent) - context.crafted_cp = SandboxCheckpoint( - checkpoint_id="cp-badpath", - sandbox_id="sb-rb4", - plan_id="plan-rb4", - phase="pre_execute", - created_at=datetime.now(), - metadata={"sandbox_path": "/nonexistent/sandbox/dir"}, - snapshot_path=snap, - ) - - -@given("a checkpoint was created for the temporary sandbox") -def step_create_cp_for_temp_sandbox(context: Context) -> None: - context.cp_result = context.mgr.create_checkpoint( - sandbox=context.mock_sb, - plan_id="plan-rb-ok", - phase="pre_execute", - metadata={"sandbox_path": context.temp_sandbox}, - ) - _register_cleanup(context, os.path.dirname(context.cp_result.snapshot_path)) - context.crafted_cp = context.cp_result - - -@given("the temporary sandbox files are then modified") -def step_modify_temp_sandbox(context: Context) -> None: - # Overwrite existing file - with open(os.path.join(context.temp_sandbox, "file.txt"), "w") as f: - f.write("modified!\n") - # Add a new file - with open(os.path.join(context.temp_sandbox, "extra.txt"), "w") as f: - f.write("extra\n") - # Add a new subdirectory with file - new_sub = os.path.join(context.temp_sandbox, "newdir") - os.makedirs(new_sub, exist_ok=True) - with open(os.path.join(new_sub, "added.txt"), "w") as f: - f.write("added\n") - - -@when("I invoke rollback_to on the created checkpoint") -def step_rollback_created_cp(context: Context) -> None: - context.rollback_result = context.mgr.rollback_to(context.cp_result) - - -@then("the rollback result should be true") -def step_assert_rollback_true(context: Context) -> None: - assert context.rollback_result is True, "Expected rollback to return True" - - -@then("the temporary sandbox should contain the original files") -def step_assert_temp_sandbox_restored(context: Context) -> None: - with open(os.path.join(context.temp_sandbox, "file.txt")) as f: - assert f.read() == "original\n", "file.txt should be restored" - with open(os.path.join(context.temp_sandbox, "subdir", "nested.txt")) as f: - assert f.read() == "nested-original\n", "nested.txt should be restored" - assert not os.path.exists(os.path.join(context.temp_sandbox, "extra.txt")), ( - "extra.txt should not exist after rollback" - ) - assert not os.path.exists(os.path.join(context.temp_sandbox, "newdir")), ( - "newdir should not exist after rollback" - ) - - -@given("a SandboxCheckpoint whose snapshot and sandbox exist but restore will fail") -def step_cp_rollback_oserror(context: Context) -> None: - # Create a real snapshot dir with a file - snap_parent = tempfile.mkdtemp(prefix="ca-rb-err-") - snap = os.path.join(snap_parent, "snapshot") - os.makedirs(snap, exist_ok=True) - with open(os.path.join(snap, "ok.txt"), "w") as f: - f.write("ok\n") - - # Create a real sandbox dir - sandbox_dir = tempfile.mkdtemp(prefix="ca-rb-err-sb-") - with open(os.path.join(sandbox_dir, "existing.txt"), "w") as f: - f.write("existing\n") - - _register_cleanup(context, snap_parent) - _register_cleanup(context, sandbox_dir) - - context.crafted_cp = SandboxCheckpoint( - checkpoint_id="cp-oserr", - sandbox_id="sb-rb-err", - plan_id="plan-rb-err", - phase="pre_execute", - created_at=datetime.now(), - metadata={"sandbox_path": sandbox_dir}, - snapshot_path=snap, - ) - # Flag the OSError scenario so the when step knows to patch - context.rollback_should_patch_oserror = True - - # =========================================================================== -# list_checkpoints - branches +# Bug fix #7488 — sandbox_path auto-stored in metadata +# (step definitions moved to checkpoint_manager_bug7488_steps.py to keep +# this file under the 500-line limit) # =========================================================================== - - -@when('I invoke list_checkpoints for sandbox "{sandbox_id}"') -def step_list_checkpoints_by_id(context: Context, sandbox_id: str) -> None: - context.cp_list = context.mgr.list_checkpoints(sandbox_id) - - -@then("the checkpoint list should be empty") -def step_assert_list_empty(context: Context) -> None: - assert context.cp_list == [], f"Expected empty list, got {context.cp_list}" - - -@given("I create two checkpoints for the mock sandbox") -def step_create_two_checkpoints(context: Context) -> None: - context.mgr.create_checkpoint( - sandbox=context.mock_sb, - plan_id="plan-list", - phase="pre_execute", - ) - context.mgr.create_checkpoint( - sandbox=context.mock_sb, - plan_id="plan-list", - phase="post_execute", - ) - - -@when("I invoke list_checkpoints for the mock sandbox") -def step_list_checkpoints_mock_sb(context: Context) -> None: - context.cp_list = context.mgr.list_checkpoints(context.mock_sb.sandbox_id) - - -@then("the checkpoint list should have {count:d} entries in creation order") -def step_assert_list_count_ordered(context: Context, count: int) -> None: - assert len(context.cp_list) == count, ( - f"Expected {count} entries, got {len(context.cp_list)}" - ) - for i in range(1, len(context.cp_list)): - assert context.cp_list[i].created_at >= context.cp_list[i - 1].created_at - - -# =========================================================================== -# delete_checkpoint - branches -# =========================================================================== - - -@given("a single checkpoint is created for the mock sandbox") -def step_create_single_cp(context: Context) -> None: - context.cp_result = context.mgr.create_checkpoint( - sandbox=context.mock_sb, - plan_id="plan-del", - phase="pre_execute", - ) - _register_cleanup(context, os.path.dirname(context.cp_result.snapshot_path)) - - -@when("I invoke delete_checkpoint with the created checkpoint id") -def step_delete_created_cp(context: Context) -> None: - context.delete_result = context.mgr.delete_checkpoint( - context.cp_result.checkpoint_id - ) - - -@then("the checkpoint delete result should be true") -def step_assert_delete_true(context: Context) -> None: - assert context.delete_result is True, "Expected delete to return True" - - -@then("listing checkpoints for the mock sandbox should return {count:d} entries") -def step_assert_list_after_delete(context: Context, count: int) -> None: - result = context.mgr.list_checkpoints(context.mock_sb.sandbox_id) - assert len(result) == count, f"Expected {count} entries, got {len(result)}" - - -@when('I invoke delete_checkpoint with id "{cp_id}"') -def step_delete_unknown_cp(context: Context, cp_id: str) -> None: - context.delete_result = context.mgr.delete_checkpoint(cp_id) - - -@then("the checkpoint delete result should be false") -def step_assert_delete_false(context: Context) -> None: - assert context.delete_result is False, "Expected delete to return False" diff --git a/features/steps/checkpoint_manager_rollback_steps.py b/features/steps/checkpoint_manager_rollback_steps.py new file mode 100644 index 000000000..fa8d5da2b --- /dev/null +++ b/features/steps/checkpoint_manager_rollback_steps.py @@ -0,0 +1,283 @@ +"""Step definitions for rollback_to, list_checkpoints, and delete_checkpoint branches. + +Split from checkpoint_manager_coverage_steps.py to keep files under 500 lines. +""" + +from __future__ import annotations + +import os +import shutil +import tempfile +from datetime import datetime +from unittest.mock import patch + +from behave import given, then, when +from behave.runner import Context + +from cleveragents.infrastructure.sandbox.checkpoint import SandboxCheckpoint + + +def _register_cleanup(context: Context, path: str) -> None: + """Schedule a directory for cleanup after the scenario.""" + if not hasattr(context, "_cleanup_handlers"): + context._cleanup_handlers = [] + context._cleanup_handlers.append( + lambda p=path: shutil.rmtree(p, ignore_errors=True) + ) + + +# =========================================================================== +# rollback_to - all branches +# =========================================================================== + + +@given("a SandboxCheckpoint with empty snapshot_path") +def step_cp_empty_snapshot(context: Context) -> None: + context.crafted_cp = SandboxCheckpoint( + checkpoint_id="cp-empty-snap", + sandbox_id="sb-rb1", + plan_id="plan-rb1", + phase="pre_execute", + created_at=datetime.now(), + metadata={}, + snapshot_path="", + ) + + +@when("I invoke rollback_to on the crafted checkpoint") +def step_rollback_crafted(context: Context) -> None: + if getattr(context, "rollback_should_patch_oserror", False): + # Patch os.listdir to raise on the second call (copy phase) + orig_listdir = os.listdir + call_count = {"n": 0} + + def _failing_listdir(path: str) -> list[str]: + call_count["n"] += 1 + if call_count["n"] >= 2: + raise OSError("simulated listdir failure") + return orig_listdir(path) + + with patch("os.listdir", side_effect=_failing_listdir): + context.rollback_result = context.mgr.rollback_to(context.crafted_cp) + else: + context.rollback_result = context.mgr.rollback_to(context.crafted_cp) + + +@then("the rollback result should be false") +def step_assert_rollback_false(context: Context) -> None: + assert context.rollback_result is False, "Expected rollback to return False" + + +@given("a SandboxCheckpoint with a non-existent snapshot_path") +def step_cp_nonexistent_snapshot(context: Context) -> None: + context.crafted_cp = SandboxCheckpoint( + checkpoint_id="cp-nosnap", + sandbox_id="sb-rb2", + plan_id="plan-rb2", + phase="pre_execute", + created_at=datetime.now(), + metadata={}, + snapshot_path="/nonexistent/snapshot/dir", + ) + + +@given("a SandboxCheckpoint whose snapshot exists but metadata has no sandbox_path") +def step_cp_no_sandbox_path_meta(context: Context) -> None: + # Create a real snapshot dir so the first guard passes + snap_parent = tempfile.mkdtemp(prefix="ca-rb-test-") + snap = os.path.join(snap_parent, "snapshot") + os.makedirs(snap, exist_ok=True) + _register_cleanup(context, snap_parent) + context.crafted_cp = SandboxCheckpoint( + checkpoint_id="cp-nometa", + sandbox_id="sb-rb3", + plan_id="plan-rb3", + phase="pre_execute", + created_at=datetime.now(), + metadata={}, # no sandbox_path key + snapshot_path=snap, + ) + + +@given("a SandboxCheckpoint whose snapshot exists but sandbox_path points nowhere") +def step_cp_sandbox_path_missing(context: Context) -> None: + snap_parent = tempfile.mkdtemp(prefix="ca-rb-test2-") + snap = os.path.join(snap_parent, "snapshot") + os.makedirs(snap, exist_ok=True) + _register_cleanup(context, snap_parent) + context.crafted_cp = SandboxCheckpoint( + checkpoint_id="cp-badpath", + sandbox_id="sb-rb4", + plan_id="plan-rb4", + phase="pre_execute", + created_at=datetime.now(), + metadata={"sandbox_path": "/nonexistent/sandbox/dir"}, + snapshot_path=snap, + ) + + +@given("a checkpoint was created for the temporary sandbox") +def step_create_cp_for_temp_sandbox(context: Context) -> None: + context.cp_result = context.mgr.create_checkpoint( + sandbox=context.mock_sb, + plan_id="plan-rb-ok", + phase="pre_execute", + metadata={"sandbox_path": context.temp_sandbox}, + ) + _register_cleanup(context, os.path.dirname(context.cp_result.snapshot_path)) + context.crafted_cp = context.cp_result + + +@given("the temporary sandbox files are then modified") +def step_modify_temp_sandbox(context: Context) -> None: + # Overwrite existing file + with open(os.path.join(context.temp_sandbox, "file.txt"), "w") as f: + f.write("modified!\n") + # Add a new file + with open(os.path.join(context.temp_sandbox, "extra.txt"), "w") as f: + f.write("extra\n") + # Add a new subdirectory with file + new_sub = os.path.join(context.temp_sandbox, "newdir") + os.makedirs(new_sub, exist_ok=True) + with open(os.path.join(new_sub, "added.txt"), "w") as f: + f.write("added\n") + + +@when("I invoke rollback_to on the created checkpoint") +def step_rollback_created_cp(context: Context) -> None: + context.rollback_result = context.mgr.rollback_to(context.cp_result) + + +@then("the rollback result should be true") +def step_assert_rollback_true(context: Context) -> None: + assert context.rollback_result is True, "Expected rollback to return True" + + +@then("the temporary sandbox should contain the original files") +def step_assert_temp_sandbox_restored(context: Context) -> None: + with open(os.path.join(context.temp_sandbox, "file.txt")) as f: + assert f.read() == "original\n", "file.txt should be restored" + with open(os.path.join(context.temp_sandbox, "subdir", "nested.txt")) as f: + assert f.read() == "nested-original\n", "nested.txt should be restored" + assert not os.path.exists(os.path.join(context.temp_sandbox, "extra.txt")), ( + "extra.txt should not exist after rollback" + ) + assert not os.path.exists(os.path.join(context.temp_sandbox, "newdir")), ( + "newdir should not exist after rollback" + ) + + +@given("a SandboxCheckpoint whose snapshot and sandbox exist but restore will fail") +def step_cp_rollback_oserror(context: Context) -> None: + # Create a real snapshot dir with a file + snap_parent = tempfile.mkdtemp(prefix="ca-rb-err-") + snap = os.path.join(snap_parent, "snapshot") + os.makedirs(snap, exist_ok=True) + with open(os.path.join(snap, "ok.txt"), "w") as f: + f.write("ok\n") + + # Create a real sandbox dir + sandbox_dir = tempfile.mkdtemp(prefix="ca-rb-err-sb-") + with open(os.path.join(sandbox_dir, "existing.txt"), "w") as f: + f.write("existing\n") + + _register_cleanup(context, snap_parent) + _register_cleanup(context, sandbox_dir) + + context.crafted_cp = SandboxCheckpoint( + checkpoint_id="cp-oserr", + sandbox_id="sb-rb-err", + plan_id="plan-rb-err", + phase="pre_execute", + created_at=datetime.now(), + metadata={"sandbox_path": sandbox_dir}, + snapshot_path=snap, + ) + # Flag the OSError scenario so the when step knows to patch + context.rollback_should_patch_oserror = True + + +# =========================================================================== +# list_checkpoints - branches +# =========================================================================== + + +@when('I invoke list_checkpoints for sandbox "{sandbox_id}"') +def step_list_checkpoints_by_id(context: Context, sandbox_id: str) -> None: + context.cp_list = context.mgr.list_checkpoints(sandbox_id) + + +@then("the checkpoint list should be empty") +def step_assert_list_empty(context: Context) -> None: + assert context.cp_list == [], f"Expected empty list, got {context.cp_list}" + + +@given("I create two checkpoints for the mock sandbox") +def step_create_two_checkpoints(context: Context) -> None: + context.mgr.create_checkpoint( + sandbox=context.mock_sb, + plan_id="plan-list", + phase="pre_execute", + ) + context.mgr.create_checkpoint( + sandbox=context.mock_sb, + plan_id="plan-list", + phase="post_execute", + ) + + +@when("I invoke list_checkpoints for the mock sandbox") +def step_list_checkpoints_mock_sb(context: Context) -> None: + context.cp_list = context.mgr.list_checkpoints(context.mock_sb.sandbox_id) + + +@then("the checkpoint list should have {count:d} entries in creation order") +def step_assert_list_count_ordered(context: Context, count: int) -> None: + assert len(context.cp_list) == count, ( + f"Expected {count} entries, got {len(context.cp_list)}" + ) + for i in range(1, len(context.cp_list)): + assert context.cp_list[i].created_at >= context.cp_list[i - 1].created_at + + +# =========================================================================== +# delete_checkpoint - branches +# =========================================================================== + + +@given("a single checkpoint is created for the mock sandbox") +def step_create_single_cp(context: Context) -> None: + context.cp_result = context.mgr.create_checkpoint( + sandbox=context.mock_sb, + plan_id="plan-del", + phase="pre_execute", + ) + _register_cleanup(context, os.path.dirname(context.cp_result.snapshot_path)) + + +@when("I invoke delete_checkpoint with the created checkpoint id") +def step_delete_created_cp(context: Context) -> None: + context.delete_result = context.mgr.delete_checkpoint( + context.cp_result.checkpoint_id + ) + + +@then("the checkpoint delete result should be true") +def step_assert_delete_true(context: Context) -> None: + assert context.delete_result is True, "Expected delete to return True" + + +@then("listing checkpoints for the mock sandbox should return {count:d} entries") +def step_assert_list_after_delete(context: Context, count: int) -> None: + result = context.mgr.list_checkpoints(context.mock_sb.sandbox_id) + assert len(result) == count, f"Expected {count} entries, got {len(result)}" + + +@when('I invoke delete_checkpoint with id "{cp_id}"') +def step_delete_unknown_cp(context: Context, cp_id: str) -> None: + context.delete_result = context.mgr.delete_checkpoint(cp_id) + + +@then("the checkpoint delete result should be false") +def step_assert_delete_false(context: Context) -> None: + assert context.delete_result is False, "Expected delete to return False" diff --git a/src/cleveragents/infrastructure/sandbox/checkpoint.py b/src/cleveragents/infrastructure/sandbox/checkpoint.py index cd65b346d..65ffc4ef9 100644 --- a/src/cleveragents/infrastructure/sandbox/checkpoint.py +++ b/src/cleveragents/infrastructure/sandbox/checkpoint.py @@ -148,6 +148,9 @@ class CheckpointManager: if sandbox.context is not None: sandbox_path = sandbox.context.sandbox_path + if sandbox_path is not None: + meta["sandbox_path"] = str(sandbox_path) + snapshot_path = self._snapshot_directory(sandbox_path, checkpoint_id) checkpoint = SandboxCheckpoint(