fix(cli): restore plan rollback checkpoint listing
CI / load-versions (pull_request) Has been cancelled
CI / lint (pull_request) Has been cancelled
CI / typecheck (pull_request) Has been cancelled
CI / security (pull_request) Has been cancelled
CI / quality (pull_request) Has been cancelled
CI / unit_tests (pull_request) Has been cancelled
CI / integration_tests (pull_request) Has been cancelled
CI / coverage (pull_request) Has been cancelled
CI / build (pull_request) Has been cancelled
CI / docker (pull_request) Has been cancelled
CI / helm (pull_request) Has been cancelled
CI / push-validation (pull_request) Has been cancelled
CI / status-check (pull_request) Has been cancelled

This commit is contained in:
2026-06-16 17:54:35 -04:00
parent 93997632ed
commit 0fc697c61e
2 changed files with 99 additions and 9 deletions
+13
View File
@@ -73,6 +73,9 @@ def _build_mock_container(context: Context) -> MagicMock:
mock_lifecycle_svc.rollback_plan.side_effect = BusinessRuleViolation(
"plan is already applied"
)
mock_cp_svc.selective_rollback.side_effect = BusinessRuleViolation(
"plan is already applied"
)
else:
mock_result = MagicMock()
mock_result.from_checkpoint_id = checkpoint_id or ""
@@ -82,6 +85,7 @@ def _build_mock_container(context: Context) -> MagicMock:
]
mock_result.changes_reverted = None
mock_lifecycle_svc.rollback_plan.return_value = mock_result
mock_cp_svc.selective_rollback.return_value = mock_result
if checkpoint_id:
matching = [
@@ -93,8 +97,14 @@ def _build_mock_container(context: Context) -> MagicMock:
resource_type="checkpoint", resource_id=checkpoint_id
)
)
mock_cp_svc.selective_rollback.side_effect = ResourceNotFoundError(
resource_type="checkpoint", resource_id=checkpoint_id
)
elif plan_id == "INVALID_PLAN_ID":
mock_cp_svc.list_checkpoints.side_effect = CleverAgentsError("invalid plan ID")
mock_cp_svc.selective_rollback.side_effect = CleverAgentsError(
"invalid plan ID"
)
mock_lifecycle_svc.rollback_plan.side_effect = CleverAgentsError(
"invalid plan ID"
)
@@ -102,6 +112,9 @@ def _build_mock_container(context: Context) -> MagicMock:
mock_cp_svc.list_checkpoints.side_effect = ResourceNotFoundError(
resource_type="plan", resource_id=plan_id or "unknown"
)
mock_cp_svc.selective_rollback.side_effect = ResourceNotFoundError(
resource_type="plan", resource_id=plan_id or "unknown"
)
mock_lifecycle_svc.rollback_plan.side_effect = ResourceNotFoundError(
resource_type="plan", resource_id=plan_id or "unknown"
)
+86 -9
View File
@@ -4049,18 +4049,95 @@ def rollback_plan(
ResourceNotFoundError as RNF,
)
# Resolve checkpoint from positional arg or --to-checkpoint option
resolved_checkpoint_id = checkpoint_id or to_checkpoint
if not resolved_checkpoint_id:
console.print(
"[red]Error:[/red] checkpoint ID required. "
"Provide as positional argument or via --to-checkpoint."
)
raise typer.Abort()
container = get_container()
svc = container.checkpoint_service()
# Resolve checkpoint from positional arg or --to-checkpoint option. When
# neither is supplied, keep the historical list mode:
# ``agents plan rollback <plan_id>`` shows available checkpoints.
resolved_checkpoint_id = checkpoint_id or to_checkpoint
if not resolved_checkpoint_id:
try:
checkpoints = sorted(
svc.list_checkpoints(plan_id), key=lambda cp: cp.created_at
)
checkpoint_rows: list[dict[str, object]] = [
{
"checkpoint_id": cp.checkpoint_id,
"plan_id": cp.plan_id,
"checkpoint_type": cp.checkpoint_type,
"sandbox_ref": cp.sandbox_ref,
"created_at": cp.created_at.isoformat(),
"reason": cp.metadata.reason,
"phase": cp.metadata.phase,
"decision_id": cp.decision_id,
}
for cp in checkpoints
]
list_data = {
"plan_id": plan_id,
"checkpoints": checkpoint_rows,
"count": len(checkpoint_rows),
}
if fmt != OutputFormat.RICH.value:
console.print(
format_output(
list_data,
fmt,
command="plan rollback",
messages=[
{
"level": "ok",
"text": f"{len(checkpoint_rows)} checkpoints listed",
}
],
)
)
return
if not checkpoint_rows:
console.print(f"[dim]No checkpoints found for plan {plan_id}.[/dim]")
return
table = Table(title=f"Checkpoints for Plan {plan_id}", show_header=True)
table.add_column("Checkpoint ID", style="cyan", max_width=26)
table.add_column("Checkpoint Type", style="yellow")
table.add_column("Created", style="green")
table.add_column("Reason")
table.add_column("Phase")
table.add_column("Decision ID", style="dim", max_width=26)
for cp in checkpoints:
table.add_row(
cp.checkpoint_id,
cp.checkpoint_type,
_format_relative_time(cp.created_at),
cp.metadata.reason or "(none)",
cp.metadata.phase or "(none)",
cp.decision_id or "(none)",
)
console.print(table)
console.print(
"[dim]Checkpoint IDs: "
+ ", ".join(str(row["checkpoint_id"]) for row in checkpoint_rows)
+ "[/dim]"
)
console.print(
"[dim]Fields: checkpoint_id, checkpoint_type, created_at, reason, "
"phase, decision_id[/dim]"
)
cp_word = "checkpoint" if len(checkpoint_rows) == 1 else "checkpoints"
console.print(
f"[green bold]✓ OK[/green bold] {len(checkpoint_rows)} {cp_word} listed"
)
return
except RNF as e:
console.print(f"[red]Not found:[/red] {e.message}")
raise typer.Abort() from e
except CleverAgentsError as e:
console.print(f"[red]Error:[/red] {e.message}")
raise typer.Abort() from e
if not yes:
# Default prompt (used as fallback if metadata fetch fails)
prompt_text = (