rollback

Instructions

Implementation Reference

• vmware_pilot/executor.py:101-147 (handler)

  WorkflowExecutor.rollback() — the core rollback logic: sets state to ROLLING_BACK, iterates completed steps in reverse, dispatches each step's rollback_tool with rollback_params, collects results, and sets state to FAILED.

  def rollback(self, wf: Workflow) -> dict[str, Any]:
      """Rollback completed steps in reverse order."""
      wf.state = WorkflowState.ROLLING_BACK
      wf.log("rollback_started")
      self._store.save(wf)
  
      rollback_results = []
      for step in reversed(wf.completed_steps()):
          if not step.rollback_tool:
              rollback_results.append({
                  "step": step.index,
                  "tool": step.tool,
                  "status": "skipped",
                  "reason": "no rollback defined",
              })
              continue
  
          try:
              resolved_rb = self._resolve_step_refs(step.rollback_params, wf.steps)
              result = self._dispatch(step.skill, step.rollback_tool, resolved_rb)
              step.status = "rolled_back"
              rollback_results.append({
                  "step": step.index,
                  "tool": step.rollback_tool,
                  "status": "success",
                  "result": result,
              })
              wf.log("rollback_step", f"Step {step.index}: {step.rollback_tool} → success")
          except Exception as exc:
              rollback_results.append({
                  "step": step.index,
                  "tool": step.rollback_tool,
                  "status": "failed",
                  "error": str(exc),
              })
              wf.log("rollback_failed", f"Step {step.index}: {step.rollback_tool} → {exc}")
              # Continue rolling back other steps even if one fails
  
          self._store.save(wf)
  
      wf.state = WorkflowState.FAILED
      wf.log("rollback_completed")
      self._store.save(wf)
  
      result = wf.to_dict()
      result["rollback_results"] = rollback_results
      return result

• mcp_server/server.py:214-238 (handler)

  The MCP tool handler function 'rollback' — decorated with @mcp.tool and @vmware_tool, loads the workflow from store, validates it's not completed, then delegates to WorkflowExecutor.rollback().

  @mcp.tool(annotations={"readOnlyHint": False, "destructiveHint": True, "idempotentHint": False, "openWorldHint": True})
  @vmware_tool(risk_level="high")
  def rollback(workflow_id: str) -> dict:
      """[WRITE] Abort a workflow and rollback completed steps in reverse order.
  
      Works in any state except 'completed'. Irreversible steps are skipped.
      The workflow state is set to 'failed' after rollback.
  
      Args:
          workflow_id: The workflow ID to rollback.
  
      Returns:
          Rollback results for each step.
      """
      try:
          wf = _get_store().load(workflow_id)
          if not wf:
              return {"error": f"Workflow '{workflow_id}' not found"}
  
          if wf.state == WorkflowState.COMPLETED:
              return {"error": f"Workflow '{workflow_id}' is already completed, cannot rollback"}
  
          return _get_executor().rollback(wf)
      except Exception as e:
          return {"error": str(e), "hint": f"Rollback failed for '{workflow_id}'. Use get_workflow_status() to check state."}

• vmware_pilot/models.py:30-44 (schema)

  WorkflowStep dataclass — defines rollback_tool (str) and rollback_params (dict) fields used by the rollback logic.

  @dataclass
  class WorkflowStep:
      index: int
      action: str
      skill: str
      tool: str
      params: dict[str, Any]
      status: str = "pending"  # pending | running | success | failed | skipped | rolled_back
      result: Any = None
      started_at: str = ""
      completed_at: str = ""
      rollback_tool: str = ""
      rollback_params: dict[str, Any] = field(default_factory=dict)
      group_id: str = ""  # non-empty = parallel-group sibling; agent may dispatch concurrently with peers

• vmware_pilot/models.py:17-28 (schema)

  WorkflowState enum — includes ROLLING_BACK state used during rollback execution.

  class WorkflowState(str, Enum):
      DRAFT = "draft"
      PENDING = "pending"
      RUNNING = "running"
      MONITORING = "monitoring"
      AWAITING_APPROVAL = "awaiting_approval"
      COMMITTING = "committing"
      ROLLING_BACK = "rolling_back"
      COMPLETED = "completed"
      FAILED = "failed"
      BLOCKED_BY_POLICY = "blocked_by_policy"

• vmware_pilot/models.py:74-76 (helper)

  Workflow.completed_steps() — helper method returning list of steps with status 'success', used by executor.rollback() to determine which steps to roll back.

  def completed_steps(self) -> list[WorkflowStep]:
      return [s for s in self.steps if s.status == "success"]